Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d5dc3c0f31 | ||
|
|
a3df11b693 |
@@ -1,748 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Server-local governed immutable Host seed executor.
|
||||
|
||||
This executor is deliberately not a general-purpose privileged command runner.
|
||||
It polls one fixed Git ref/path, verifies a high-risk governed-task binding as
|
||||
the durable authorization evidence, materializes one exact candidate from the
|
||||
authoritative Git object database plus a declared patch/manifest, and atomically
|
||||
publishes the immutable release tree.
|
||||
|
||||
It never starts/restarts services, never creates the continuation Task/Job,
|
||||
never writes Git refs, and never switches Production.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
import argparse
|
||||
import fcntl
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import pwd
|
||||
from pathlib import Path, PurePosixPath
|
||||
import re
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import tempfile
|
||||
from typing import Any, Mapping
|
||||
|
||||
REQUEST_SCHEMA = "aios.control-plane-host-seed-request.v1"
|
||||
MANIFEST_SCHEMAS = {
|
||||
"aios.control-plane-immutable-candidate.v5",
|
||||
"aios.control-plane-immutable-candidate.v6",
|
||||
}
|
||||
OPERATION = "HOST_SEED_IMMUTABLE_CANDIDATE"
|
||||
REPOSITORY = "aios/aios"
|
||||
REQUEST_REF = "refs/heads/agent/host-seed-authorization-requests"
|
||||
REQUEST_PATH = "control-plane/host-seed/request.json"
|
||||
OID_RE = re.compile(r"^[0-9a-f]{40}$")
|
||||
SHA_RE = re.compile(r"^[0-9a-f]{64}$")
|
||||
TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
|
||||
RELEASE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{3,191}$")
|
||||
CANDIDATE_RE = re.compile(r"^V[0-9]+$")
|
||||
MAX_REQUEST_BYTES = 128 * 1024
|
||||
MAX_MANIFEST_BYTES = 4 * 1024 * 1024
|
||||
MAX_PATCH_BYTES = 2 * 1024 * 1024
|
||||
MAX_MACHINE_GOAL_BYTES = 512 * 1024
|
||||
MAX_BOOTSTRAP_BYTES = 512 * 1024
|
||||
|
||||
|
||||
class HostSeedError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def canonical(value: object) -> bytes:
|
||||
return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")
|
||||
|
||||
|
||||
def sha256(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def canonical_compact(value: object) -> bytes:
|
||||
return json.dumps(
|
||||
value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def _git(git_dir: Path, *args: str, input_bytes: bytes | None = None) -> bytes:
|
||||
env = {"PATH": os.environ.get("PATH", "")}
|
||||
proc = subprocess.run(
|
||||
["git", f"--git-dir={git_dir}", *args],
|
||||
input=input_bytes,
|
||||
capture_output=True,
|
||||
timeout=120,
|
||||
env=env,
|
||||
)
|
||||
if proc.returncode:
|
||||
raise HostSeedError("GIT_" + (args[0] if args else "COMMAND").upper() + "_FAILED")
|
||||
return proc.stdout
|
||||
|
||||
|
||||
def _git_text(git_dir: Path, *args: str) -> str:
|
||||
return _git(git_dir, *args).decode("utf-8", "strict").strip()
|
||||
|
||||
|
||||
def _safe_repo_path(value: object, *, field: str) -> str:
|
||||
if not isinstance(value, str) or not value or len(value) > 512:
|
||||
raise HostSeedError(field + "_INVALID")
|
||||
p = PurePosixPath(value)
|
||||
if p.is_absolute() or any(part in {"", ".", ".."} for part in p.parts):
|
||||
raise HostSeedError(field + "_INVALID")
|
||||
normalized = p.as_posix()
|
||||
if normalized.startswith(".git/") or normalized == ".git":
|
||||
raise HostSeedError(field + "_INVALID")
|
||||
return normalized
|
||||
|
||||
|
||||
def _read_git_blob(git_dir: Path, oid: str, path: str, *, maximum: int) -> bytes:
|
||||
if not OID_RE.fullmatch(oid):
|
||||
raise HostSeedError("PAYLOAD_COMMIT_INVALID")
|
||||
raw = _git(git_dir, "show", f"{oid}:{path}")
|
||||
if len(raw) <= 0 or len(raw) > maximum:
|
||||
raise HostSeedError("GIT_BLOB_SIZE_INVALID")
|
||||
return raw
|
||||
|
||||
|
||||
def _parse_json(raw: bytes, *, code: str) -> dict[str, Any]:
|
||||
try:
|
||||
value = json.loads(raw.decode("utf-8"))
|
||||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise HostSeedError(code) from exc
|
||||
if not isinstance(value, dict):
|
||||
raise HostSeedError(code)
|
||||
return value
|
||||
|
||||
|
||||
def _parse_time(value: object, *, field: str) -> datetime:
|
||||
if not isinstance(value, str) or not value:
|
||||
raise HostSeedError(field + "_INVALID")
|
||||
text = value[:-1] + "+00:00" if value.endswith("Z") else value
|
||||
try:
|
||||
parsed = datetime.fromisoformat(text)
|
||||
except ValueError as exc:
|
||||
raise HostSeedError(field + "_INVALID") from exc
|
||||
if parsed.tzinfo is None:
|
||||
raise HostSeedError(field + "_INVALID")
|
||||
return parsed.astimezone(timezone.utc)
|
||||
|
||||
|
||||
def _regular_file_bytes(
|
||||
path: Path, *, maximum: int, expected_uid: int | None = None
|
||||
) -> bytes:
|
||||
if path.parent.is_symlink():
|
||||
raise HostSeedError("AUTHORIZATION_BINDING_PARENT_SYMLINK_REJECTED")
|
||||
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
|
||||
try:
|
||||
fd = os.open(path, flags)
|
||||
except OSError as exc:
|
||||
raise HostSeedError("AUTHORIZATION_BINDING_UNAVAILABLE") from exc
|
||||
try:
|
||||
info = os.fstat(fd)
|
||||
if not stat.S_ISREG(info.st_mode) or info.st_size <= 0 or info.st_size > maximum:
|
||||
raise HostSeedError("AUTHORIZATION_BINDING_INVALID")
|
||||
if info.st_mode & 0o022:
|
||||
raise HostSeedError("AUTHORIZATION_BINDING_WRITABLE_BY_GROUP_OR_OTHER")
|
||||
if expected_uid is not None and info.st_uid != expected_uid:
|
||||
raise HostSeedError("AUTHORIZATION_BINDING_OWNER_MISMATCH")
|
||||
raw = os.read(fd, maximum + 1)
|
||||
if len(raw) > maximum:
|
||||
raise HostSeedError("AUTHORIZATION_BINDING_INVALID")
|
||||
return raw
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
|
||||
def _atomic_json(path: Path, value: Mapping[str, Any], *, mode: int = 0o600) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||||
if path.parent.is_symlink() or path.is_symlink():
|
||||
raise HostSeedError("STATE_SYMLINK_REJECTED")
|
||||
fd, temporary = tempfile.mkstemp(prefix=".tmp-", dir=path.parent)
|
||||
try:
|
||||
os.fchmod(fd, mode)
|
||||
with os.fdopen(fd, "wb", closefd=True) as stream:
|
||||
stream.write(canonical(value))
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(temporary, path)
|
||||
os.chmod(path, mode)
|
||||
parent_fd = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
|
||||
try:
|
||||
os.fsync(parent_fd)
|
||||
finally:
|
||||
os.close(parent_fd)
|
||||
finally:
|
||||
if os.path.exists(temporary):
|
||||
os.unlink(temporary)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Intent:
|
||||
candidate: str
|
||||
release_id: str
|
||||
base_oid: str
|
||||
payload_commit: str
|
||||
patch_path: str
|
||||
manifest_path: str
|
||||
machine_goal_path: str
|
||||
seed_bootstrap_path: str
|
||||
artifact_sha256: str
|
||||
manifest_sha256: str
|
||||
machine_goal_sha256: str
|
||||
release_tree_sha256: str
|
||||
seed_bootstrap_sha256: str
|
||||
target_host: str
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"candidate": self.candidate,
|
||||
"release_id": self.release_id,
|
||||
"base_oid": self.base_oid,
|
||||
"payload_commit": self.payload_commit,
|
||||
"patch_path": self.patch_path,
|
||||
"manifest_path": self.manifest_path,
|
||||
"machine_goal_path": self.machine_goal_path,
|
||||
"seed_bootstrap_path": self.seed_bootstrap_path,
|
||||
"artifact_sha256": self.artifact_sha256,
|
||||
"manifest_sha256": self.manifest_sha256,
|
||||
"machine_goal_sha256": self.machine_goal_sha256,
|
||||
"release_tree_sha256": self.release_tree_sha256,
|
||||
"seed_bootstrap_sha256": self.seed_bootstrap_sha256,
|
||||
"target_host": self.target_host,
|
||||
}
|
||||
|
||||
@property
|
||||
def digest(self) -> str:
|
||||
return sha256(canonical(self.as_dict()))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Request:
|
||||
intent: Intent
|
||||
authorization_task_id: str
|
||||
authorization_binding_sha256: str
|
||||
authorization_sha256: str
|
||||
expires_at: datetime
|
||||
|
||||
@classmethod
|
||||
def load(cls, raw: Mapping[str, Any], *, host_identity: str, now: datetime) -> "Request":
|
||||
if set(raw) != {"schema", "operation", "repository", "intent", "authorization"}:
|
||||
raise HostSeedError("REQUEST_FIELDS_INVALID")
|
||||
if raw.get("schema") != REQUEST_SCHEMA or raw.get("operation") != OPERATION:
|
||||
raise HostSeedError("REQUEST_CONTRACT_INVALID")
|
||||
if raw.get("repository") != REPOSITORY:
|
||||
raise HostSeedError("REQUEST_REPOSITORY_MISMATCH")
|
||||
intent_raw = raw.get("intent")
|
||||
if not isinstance(intent_raw, Mapping) or set(intent_raw) != {
|
||||
"candidate", "release_id", "base_oid", "payload_commit", "patch_path",
|
||||
"manifest_path", "machine_goal_path", "seed_bootstrap_path",
|
||||
"artifact_sha256", "manifest_sha256", "machine_goal_sha256",
|
||||
"release_tree_sha256", "seed_bootstrap_sha256", "target_host",
|
||||
}:
|
||||
raise HostSeedError("INTENT_FIELDS_INVALID")
|
||||
candidate = str(intent_raw.get("candidate") or "")
|
||||
release_id = str(intent_raw.get("release_id") or "")
|
||||
base_oid = str(intent_raw.get("base_oid") or "")
|
||||
payload_commit = str(intent_raw.get("payload_commit") or "")
|
||||
artifact_sha = str(intent_raw.get("artifact_sha256") or "")
|
||||
manifest_sha = str(intent_raw.get("manifest_sha256") or "")
|
||||
machine_goal_sha = str(intent_raw.get("machine_goal_sha256") or "")
|
||||
release_tree_sha = str(intent_raw.get("release_tree_sha256") or "")
|
||||
seed_bootstrap_sha = str(intent_raw.get("seed_bootstrap_sha256") or "")
|
||||
target_host = str(intent_raw.get("target_host") or "")
|
||||
if not CANDIDATE_RE.fullmatch(candidate):
|
||||
raise HostSeedError("CANDIDATE_INVALID")
|
||||
if not RELEASE_RE.fullmatch(release_id):
|
||||
raise HostSeedError("RELEASE_ID_INVALID")
|
||||
if not OID_RE.fullmatch(base_oid) or not OID_RE.fullmatch(payload_commit):
|
||||
raise HostSeedError("INTENT_OID_INVALID")
|
||||
if not all(SHA_RE.fullmatch(x) for x in (
|
||||
artifact_sha, manifest_sha, machine_goal_sha, release_tree_sha, seed_bootstrap_sha
|
||||
)):
|
||||
raise HostSeedError("INTENT_SHA_INVALID")
|
||||
if target_host != host_identity:
|
||||
raise HostSeedError("TARGET_HOST_MISMATCH")
|
||||
intent = Intent(
|
||||
candidate=candidate,
|
||||
release_id=release_id,
|
||||
base_oid=base_oid,
|
||||
payload_commit=payload_commit,
|
||||
patch_path=_safe_repo_path(intent_raw.get("patch_path"), field="PATCH_PATH"),
|
||||
manifest_path=_safe_repo_path(intent_raw.get("manifest_path"), field="MANIFEST_PATH"),
|
||||
machine_goal_path=_safe_repo_path(intent_raw.get("machine_goal_path"), field="MACHINE_GOAL_PATH"),
|
||||
seed_bootstrap_path=_safe_repo_path(intent_raw.get("seed_bootstrap_path"), field="SEED_BOOTSTRAP_PATH"),
|
||||
artifact_sha256=artifact_sha,
|
||||
manifest_sha256=manifest_sha,
|
||||
machine_goal_sha256=machine_goal_sha,
|
||||
release_tree_sha256=release_tree_sha,
|
||||
seed_bootstrap_sha256=seed_bootstrap_sha,
|
||||
target_host=target_host,
|
||||
)
|
||||
auth = raw.get("authorization")
|
||||
if not isinstance(auth, Mapping) or set(auth) != {
|
||||
"task_id", "binding_sha256", "expires_at", "intent_sha256",
|
||||
"authorization_sha256",
|
||||
}:
|
||||
raise HostSeedError("AUTHORIZATION_FIELDS_INVALID")
|
||||
task_id = str(auth.get("task_id") or "")
|
||||
binding_sha = str(auth.get("binding_sha256") or "")
|
||||
authorization_sha = str(auth.get("authorization_sha256") or "")
|
||||
if (
|
||||
not TASK_RE.fullmatch(task_id)
|
||||
or not SHA_RE.fullmatch(binding_sha)
|
||||
or not SHA_RE.fullmatch(authorization_sha)
|
||||
):
|
||||
raise HostSeedError("AUTHORIZATION_IDENTITY_INVALID")
|
||||
if auth.get("intent_sha256") != intent.digest:
|
||||
raise HostSeedError("AUTHORIZATION_INTENT_MISMATCH")
|
||||
expires_text = str(auth.get("expires_at") or "")
|
||||
expires_at = _parse_time(expires_text, field="AUTHORIZATION_EXPIRES_AT")
|
||||
expected_authorization_sha = sha256(canonical({
|
||||
"task_id": task_id,
|
||||
"intent_sha256": intent.digest,
|
||||
"expires_at": expires_text,
|
||||
}))
|
||||
if authorization_sha != expected_authorization_sha:
|
||||
raise HostSeedError("AUTHORIZATION_DIGEST_MISMATCH")
|
||||
if now >= expires_at:
|
||||
raise HostSeedError("AUTHORIZATION_EXPIRED")
|
||||
if (expires_at - now).total_seconds() > 3600:
|
||||
raise HostSeedError("AUTHORIZATION_WINDOW_TOO_LARGE")
|
||||
return cls(
|
||||
intent=intent,
|
||||
authorization_task_id=task_id,
|
||||
authorization_binding_sha256=binding_sha,
|
||||
authorization_sha256=authorization_sha,
|
||||
expires_at=expires_at,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Config:
|
||||
git_dir: Path
|
||||
task_binding_root: Path
|
||||
state_root: Path
|
||||
release_root: Path
|
||||
host_identity: str
|
||||
task_binding_owner: str | None = "aios-bridge-staging"
|
||||
request_ref: str = REQUEST_REF
|
||||
request_path: str = REQUEST_PATH
|
||||
|
||||
|
||||
def _load_task_binding(config: Config, request: Request) -> dict[str, Any]:
|
||||
safe = re.sub(r"[^A-Za-z0-9_.-]+", "-", request.authorization_task_id).strip("-")[:120] or "task"
|
||||
path = config.task_binding_root / f"{safe}.json"
|
||||
expected_uid = None
|
||||
if config.task_binding_owner:
|
||||
try:
|
||||
expected_uid = pwd.getpwnam(config.task_binding_owner).pw_uid
|
||||
except KeyError as exc:
|
||||
raise HostSeedError("AUTHORIZATION_BINDING_OWNER_UNKNOWN") from exc
|
||||
raw = _regular_file_bytes(path, maximum=256 * 1024, expected_uid=expected_uid)
|
||||
if sha256(raw) != request.authorization_binding_sha256:
|
||||
raise HostSeedError("AUTHORIZATION_BINDING_HASH_MISMATCH")
|
||||
binding = _parse_json(raw, code="AUTHORIZATION_BINDING_JSON_INVALID")
|
||||
required = {
|
||||
"task_id", "repository", "base_oid", "target_branch", "risk",
|
||||
"description", "idempotency_key", "expected_branch_oid",
|
||||
"branch_mode", "generation", "job_id",
|
||||
}
|
||||
if set(binding) != required:
|
||||
raise HostSeedError("AUTHORIZATION_BINDING_FIELDS_INVALID")
|
||||
if binding.get("task_id") != request.authorization_task_id:
|
||||
raise HostSeedError("AUTHORIZATION_TASK_MISMATCH")
|
||||
if binding.get("repository") != REPOSITORY:
|
||||
raise HostSeedError("AUTHORIZATION_REPOSITORY_MISMATCH")
|
||||
if binding.get("risk") != "high":
|
||||
raise HostSeedError("AUTHORIZATION_RISK_MISMATCH")
|
||||
if binding.get("description") != "HOST_SEED_AUTHORIZATION:" + request.authorization_sha256:
|
||||
raise HostSeedError("AUTHORIZATION_DESCRIPTION_MISMATCH")
|
||||
if not str(binding.get("target_branch") or "").startswith("agent/host-seed-authorization-"):
|
||||
raise HostSeedError("AUTHORIZATION_BRANCH_MISMATCH")
|
||||
if not isinstance(binding.get("generation"), int) or binding["generation"] < 1:
|
||||
raise HostSeedError("AUTHORIZATION_GENERATION_INVALID")
|
||||
return binding
|
||||
|
||||
|
||||
def _load_manifest(config: Config, request: Request) -> dict[str, Any]:
|
||||
raw = _read_git_blob(
|
||||
config.git_dir, request.intent.payload_commit, request.intent.manifest_path,
|
||||
maximum=MAX_MANIFEST_BYTES,
|
||||
)
|
||||
if sha256(raw) != request.intent.manifest_sha256:
|
||||
raise HostSeedError("MANIFEST_HASH_MISMATCH")
|
||||
manifest = _parse_json(raw, code="MANIFEST_JSON_INVALID")
|
||||
if manifest.get("schema") not in MANIFEST_SCHEMAS:
|
||||
raise HostSeedError("MANIFEST_SCHEMA_REJECTED")
|
||||
if manifest.get("candidate") != request.intent.candidate:
|
||||
raise HostSeedError("MANIFEST_CANDIDATE_MISMATCH")
|
||||
if manifest.get("repository") != REPOSITORY:
|
||||
raise HostSeedError("MANIFEST_REPOSITORY_MISMATCH")
|
||||
if manifest.get("base_oid") != request.intent.base_oid:
|
||||
raise HostSeedError("MANIFEST_BASE_MISMATCH")
|
||||
if manifest.get("release_id") != request.intent.release_id:
|
||||
raise HostSeedError("MANIFEST_RELEASE_ID_MISMATCH")
|
||||
members = manifest.get("package_members")
|
||||
changes = manifest.get("changed_files")
|
||||
if not isinstance(members, Mapping) or not isinstance(changes, list):
|
||||
raise HostSeedError("MANIFEST_MEMBERS_INVALID")
|
||||
return manifest
|
||||
|
||||
|
||||
def _load_identity_evidence(config: Config, request: Request) -> dict[str, Any]:
|
||||
goal_raw = _read_git_blob(
|
||||
config.git_dir, request.intent.payload_commit, request.intent.machine_goal_path,
|
||||
maximum=MAX_MACHINE_GOAL_BYTES,
|
||||
)
|
||||
if sha256(goal_raw) != request.intent.machine_goal_sha256:
|
||||
raise HostSeedError("MACHINE_GOAL_HASH_MISMATCH")
|
||||
goal = _parse_json(goal_raw, code="MACHINE_GOAL_JSON_INVALID")
|
||||
required_pairs = {
|
||||
"candidate": request.intent.candidate,
|
||||
"repository": REPOSITORY,
|
||||
"base_oid": request.intent.base_oid,
|
||||
"release_id": request.intent.release_id,
|
||||
"artifact_sha256": request.intent.artifact_sha256,
|
||||
"manifest_sha256": request.intent.manifest_sha256,
|
||||
"release_tree_sha256": request.intent.release_tree_sha256,
|
||||
"seed_bootstrap_sha256": request.intent.seed_bootstrap_sha256,
|
||||
}
|
||||
for key, expected in required_pairs.items():
|
||||
if goal.get(key) != expected:
|
||||
raise HostSeedError("MACHINE_GOAL_BINDING_MISMATCH:" + key)
|
||||
if goal.get("host_seed_authorized") is not False:
|
||||
raise HostSeedError("MACHINE_GOAL_HOST_SEED_FLAG_INVALID")
|
||||
bootstrap = _read_git_blob(
|
||||
config.git_dir, request.intent.payload_commit, request.intent.seed_bootstrap_path,
|
||||
maximum=MAX_BOOTSTRAP_BYTES,
|
||||
)
|
||||
if sha256(bootstrap) != request.intent.seed_bootstrap_sha256:
|
||||
raise HostSeedError("SEED_BOOTSTRAP_HASH_MISMATCH")
|
||||
return goal
|
||||
|
||||
|
||||
def _load_patch(config: Config, request: Request) -> bytes:
|
||||
patch = _read_git_blob(
|
||||
config.git_dir, request.intent.payload_commit, request.intent.patch_path,
|
||||
maximum=MAX_PATCH_BYTES,
|
||||
)
|
||||
if b"\x00" in patch:
|
||||
raise HostSeedError("PATCH_BINARY_NUL_REJECTED")
|
||||
return patch
|
||||
|
||||
|
||||
def _manifest_release_map(manifest: Mapping[str, Any]) -> dict[str, str]:
|
||||
members = manifest["package_members"]
|
||||
result: dict[str, str] = {}
|
||||
prefix = "payload/repo/after/"
|
||||
for name, digest in members.items():
|
||||
if not isinstance(name, str) or not name.startswith(prefix) or not isinstance(digest, str) or not SHA_RE.fullmatch(digest):
|
||||
continue
|
||||
rel = _safe_repo_path(name[len(prefix):], field="RELEASE_MEMBER")
|
||||
if rel in result:
|
||||
raise HostSeedError("DUPLICATE_RELEASE_MEMBER")
|
||||
result[rel] = digest
|
||||
declared_count = manifest.get("release_member_count")
|
||||
if not isinstance(declared_count, int) or declared_count != len(result):
|
||||
raise HostSeedError("RELEASE_MEMBER_COUNT_MISMATCH")
|
||||
return result
|
||||
|
||||
|
||||
def _expected_changes(manifest: Mapping[str, Any]) -> dict[str, tuple[str, str]]:
|
||||
result: dict[str, tuple[str, str]] = {}
|
||||
for entry in manifest["changed_files"]:
|
||||
if not isinstance(entry, Mapping):
|
||||
raise HostSeedError("CHANGED_FILE_ENTRY_INVALID")
|
||||
rel = _safe_repo_path(entry.get("path"), field="CHANGED_PATH")
|
||||
digest = str(entry.get("after_sha256") or "")
|
||||
status = str(entry.get("status") or "")
|
||||
if not SHA_RE.fullmatch(digest) or status not in {"modified", "added"} or rel in result:
|
||||
raise HostSeedError("CHANGED_FILE_ENTRY_INVALID")
|
||||
result[rel] = (digest, status)
|
||||
if manifest.get("changed_file_count") != len(result):
|
||||
raise HostSeedError("CHANGED_FILE_COUNT_MISMATCH")
|
||||
return result
|
||||
|
||||
|
||||
def _regular_member(root: Path, relative: str) -> Path:
|
||||
current = root
|
||||
for part in PurePosixPath(relative).parts:
|
||||
current = current / part
|
||||
try:
|
||||
info = current.lstat()
|
||||
except FileNotFoundError as exc:
|
||||
raise HostSeedError("RELEASE_MEMBER_MISSING:" + relative) from exc
|
||||
if stat.S_ISLNK(info.st_mode):
|
||||
raise HostSeedError("RELEASE_MEMBER_SYMLINK_REJECTED:" + relative)
|
||||
if not current.is_file():
|
||||
raise HostSeedError("RELEASE_MEMBER_NOT_REGULAR:" + relative)
|
||||
return current
|
||||
|
||||
|
||||
def _materialize_and_verify(config: Config, request: Request, manifest: Mapping[str, Any], patch: bytes, workspace: Path) -> dict[str, str]:
|
||||
if _git_text(config.git_dir, "cat-file", "-t", request.intent.base_oid) != "commit":
|
||||
raise HostSeedError("BASE_COMMIT_NOT_FOUND")
|
||||
if _git_text(config.git_dir, "cat-file", "-t", request.intent.payload_commit) != "commit":
|
||||
raise HostSeedError("PAYLOAD_COMMIT_NOT_FOUND")
|
||||
workspace.parent.mkdir(parents=True, exist_ok=True)
|
||||
clone = subprocess.run(
|
||||
["git", "clone", "--no-checkout", "--shared", str(config.git_dir), str(workspace)],
|
||||
capture_output=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
|
||||
)
|
||||
if clone.returncode:
|
||||
raise HostSeedError("WORKSPACE_CLONE_FAILED")
|
||||
checkout = subprocess.run(
|
||||
["git", "-C", str(workspace), "checkout", "--detach", request.intent.base_oid],
|
||||
capture_output=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
|
||||
)
|
||||
if checkout.returncode:
|
||||
raise HostSeedError("WORKSPACE_CHECKOUT_FAILED")
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
["git", "-C", str(workspace), "apply", "--whitespace=nowarn", "-"],
|
||||
input=patch, capture_output=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
|
||||
)
|
||||
if proc.returncode:
|
||||
raise HostSeedError("PATCH_APPLY_FAILED")
|
||||
changes = _expected_changes(manifest)
|
||||
status_raw = subprocess.run(
|
||||
["git", "-C", str(workspace), "status", "--porcelain", "-z"],
|
||||
check=True, capture_output=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
|
||||
).stdout
|
||||
actual: set[str] = set()
|
||||
chunks = status_raw.split(b"\0")
|
||||
for chunk in chunks:
|
||||
if not chunk:
|
||||
continue
|
||||
text = chunk.decode("utf-8", "surrogateescape")
|
||||
if len(text) < 4:
|
||||
raise HostSeedError("WORKTREE_STATUS_INVALID")
|
||||
actual.add(text[3:].replace("\\", "/"))
|
||||
if actual != set(changes):
|
||||
raise HostSeedError("PATCH_CHANGESET_MISMATCH")
|
||||
base_paths = {
|
||||
_safe_repo_path(path, field="TRACKED_PATH")
|
||||
for path in _git_text(config.git_dir, "ls-tree", "-r", "--name-only", request.intent.base_oid).splitlines()
|
||||
if path
|
||||
}
|
||||
for rel, (digest, status) in changes.items():
|
||||
if status == "modified" and rel not in base_paths:
|
||||
raise HostSeedError("MODIFIED_PATH_NOT_IN_BASE:" + rel)
|
||||
if status == "added" and rel in base_paths:
|
||||
raise HostSeedError("ADDED_PATH_ALREADY_IN_BASE:" + rel)
|
||||
p = _regular_member(workspace, rel)
|
||||
if sha256(p.read_bytes()) != digest:
|
||||
raise HostSeedError("CHANGED_FILE_HASH_MISMATCH:" + rel)
|
||||
release_map = _manifest_release_map(manifest)
|
||||
expected_release_paths = base_paths | {
|
||||
rel for rel, (_digest, status) in changes.items() if status == "added"
|
||||
}
|
||||
if set(release_map) != expected_release_paths:
|
||||
raise HostSeedError("RELEASE_MEMBER_SET_MISMATCH")
|
||||
for rel, digest in release_map.items():
|
||||
p = _regular_member(workspace, rel)
|
||||
if sha256(p.read_bytes()) != digest:
|
||||
raise HostSeedError("RELEASE_MEMBER_HASH_MISMATCH:" + rel)
|
||||
tree_digest = sha256(canonical_compact({f"payload/repo/after/{k}": release_map[k] for k in sorted(release_map)}))
|
||||
if tree_digest != request.intent.release_tree_sha256:
|
||||
raise HostSeedError("RELEASE_TREE_HASH_MISMATCH")
|
||||
if manifest.get("release_tree_sha256") not in {None, request.intent.release_tree_sha256}:
|
||||
raise HostSeedError("MANIFEST_RELEASE_TREE_HASH_MISMATCH")
|
||||
return release_map
|
||||
except Exception:
|
||||
raise
|
||||
|
||||
|
||||
def _publish_release(config: Config, request: Request, workspace: Path, release_map: Mapping[str, str]) -> Path:
|
||||
if config.release_root.exists() and config.release_root.is_symlink():
|
||||
raise HostSeedError("RELEASE_ROOT_SYMLINK_REJECTED")
|
||||
final = config.release_root / request.intent.release_id
|
||||
if final.exists():
|
||||
if final.is_symlink() or not final.is_dir():
|
||||
raise HostSeedError("RELEASE_DESTINATION_CONFLICT")
|
||||
for rel, digest in release_map.items():
|
||||
p = final / PurePosixPath(rel)
|
||||
if p.is_symlink() or not p.is_file() or sha256(p.read_bytes()) != digest:
|
||||
raise HostSeedError("EXISTING_RELEASE_HASH_MISMATCH")
|
||||
return final
|
||||
|
||||
config.release_root.mkdir(parents=True, exist_ok=True)
|
||||
os.chmod(config.release_root, 0o755)
|
||||
staging = Path(tempfile.mkdtemp(prefix=".host-seed-", dir=config.release_root))
|
||||
os.chmod(staging, 0o755)
|
||||
try:
|
||||
for rel, digest in release_map.items():
|
||||
source = _regular_member(workspace, rel)
|
||||
target = staging / PurePosixPath(rel)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
current = target.parent
|
||||
while current != staging:
|
||||
os.chmod(current, 0o755)
|
||||
current = current.parent
|
||||
data = source.read_bytes()
|
||||
if sha256(data) != digest:
|
||||
raise HostSeedError("STAGING_SOURCE_DRIFT")
|
||||
fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600)
|
||||
try:
|
||||
with os.fdopen(fd, "wb", closefd=True) as out:
|
||||
out.write(data)
|
||||
out.flush()
|
||||
os.fsync(out.fileno())
|
||||
except Exception:
|
||||
raise
|
||||
source_mode = stat.S_IMODE(source.stat(follow_symlinks=False).st_mode)
|
||||
os.chmod(target, source_mode & 0o755 or 0o644)
|
||||
for rel, digest in release_map.items():
|
||||
p = staging / PurePosixPath(rel)
|
||||
if p.is_symlink() or sha256(p.read_bytes()) != digest:
|
||||
raise HostSeedError("STAGING_VERIFY_FAILED")
|
||||
os.replace(staging, final)
|
||||
parent_fd = os.open(config.release_root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
|
||||
try:
|
||||
os.fsync(parent_fd)
|
||||
finally:
|
||||
os.close(parent_fd)
|
||||
return final
|
||||
finally:
|
||||
if staging.exists():
|
||||
shutil.rmtree(staging)
|
||||
|
||||
|
||||
def execute(config: Config, *, now: datetime | None = None, require_root: bool = True) -> dict[str, Any]:
|
||||
if require_root and os.geteuid() != 0:
|
||||
raise HostSeedError("ROOT_REQUIRED")
|
||||
current = now or datetime.now(timezone.utc)
|
||||
config.state_root.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||||
lock_path = config.state_root / "executor.lock"
|
||||
lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
|
||||
try:
|
||||
fcntl.flock(lock_fd, fcntl.LOCK_EX)
|
||||
ref_probe = subprocess.run(
|
||||
["git", f"--git-dir={config.git_dir}", "rev-parse", "--verify", config.request_ref],
|
||||
capture_output=True, timeout=30, env={"PATH": os.environ.get("PATH", "")},
|
||||
)
|
||||
if ref_probe.returncode:
|
||||
return {
|
||||
"schema": "aios.control-plane-host-seed-poll.v1",
|
||||
"state": "NO_REQUEST",
|
||||
"request_ref": config.request_ref,
|
||||
"service_mutation": False,
|
||||
"task_creation": False,
|
||||
}
|
||||
request_oid = ref_probe.stdout.decode("utf-8", "strict").strip()
|
||||
if not OID_RE.fullmatch(request_oid):
|
||||
raise HostSeedError("REQUEST_REF_INVALID")
|
||||
raw = _read_git_blob(config.git_dir, request_oid, config.request_path, maximum=MAX_REQUEST_BYTES)
|
||||
request_value = _parse_json(raw, code="REQUEST_JSON_INVALID")
|
||||
request = Request.load(request_value, host_identity=config.host_identity, now=current)
|
||||
binding = _load_task_binding(config, request)
|
||||
manifest = _load_manifest(config, request)
|
||||
_load_identity_evidence(config, request)
|
||||
patch = _load_patch(config, request)
|
||||
|
||||
op_id = request.intent.digest
|
||||
receipt_path = config.state_root / "receipts" / f"{op_id}.json"
|
||||
if receipt_path.exists():
|
||||
receipt = _parse_json(_regular_file_bytes(receipt_path, maximum=256 * 1024), code="RECEIPT_INVALID")
|
||||
if receipt.get("state") == "EXECUTED" and receipt.get("intent_sha256") == op_id:
|
||||
return receipt
|
||||
raise HostSeedError("OPERATION_STATE_CONFLICT")
|
||||
|
||||
_atomic_json(config.state_root / "operations" / f"{op_id}.json", {
|
||||
"schema": "aios.control-plane-host-seed-operation.v1",
|
||||
"state": "PENDING",
|
||||
"intent_sha256": op_id,
|
||||
"request_commit": request_oid,
|
||||
"authorization_task_id": request.authorization_task_id,
|
||||
"authorization_binding_sha256": request.authorization_binding_sha256,
|
||||
"started_at": current.isoformat().replace("+00:00", "Z"),
|
||||
})
|
||||
|
||||
workspace_root = config.state_root / "workspaces"
|
||||
workspace_root.mkdir(parents=True, exist_ok=True)
|
||||
workspace = workspace_root / f"seed-{op_id[:20]}"
|
||||
if workspace.exists():
|
||||
raise HostSeedError("WORKSPACE_ALREADY_EXISTS")
|
||||
release_map: dict[str, str] | None = None
|
||||
try:
|
||||
release_map = _materialize_and_verify(config, request, manifest, patch, workspace)
|
||||
destination = _publish_release(config, request, workspace, release_map)
|
||||
finally:
|
||||
if workspace.exists():
|
||||
shutil.rmtree(workspace, ignore_errors=True)
|
||||
|
||||
receipt = {
|
||||
"schema": "aios.control-plane-host-seed-receipt.v1",
|
||||
"state": "EXECUTED",
|
||||
"operation": OPERATION,
|
||||
"intent_sha256": op_id,
|
||||
"request_commit": request_oid,
|
||||
"repository": REPOSITORY,
|
||||
"target_host": request.intent.target_host,
|
||||
"candidate": request.intent.candidate,
|
||||
"release_id": request.intent.release_id,
|
||||
"base_oid": request.intent.base_oid,
|
||||
"payload_commit": request.intent.payload_commit,
|
||||
"validation_artifact_sha256": request.intent.artifact_sha256,
|
||||
"artifact_delivery_used": False,
|
||||
"manifest_sha256": request.intent.manifest_sha256,
|
||||
"machine_goal_sha256": request.intent.machine_goal_sha256,
|
||||
"release_tree_sha256": request.intent.release_tree_sha256,
|
||||
"seed_bootstrap_sha256": request.intent.seed_bootstrap_sha256,
|
||||
"authorization_task_id": request.authorization_task_id,
|
||||
"authorization_binding_sha256": request.authorization_binding_sha256,
|
||||
"release_path": str(destination),
|
||||
"release_member_count": len(release_map or {}),
|
||||
"service_mutation": False,
|
||||
"service_restart": False,
|
||||
"projector_started": False,
|
||||
"task_creation": False,
|
||||
"production_write": False,
|
||||
"main_write": False,
|
||||
"tag_write": False,
|
||||
"baseline_write": False,
|
||||
"executed_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
|
||||
}
|
||||
_atomic_json(receipt_path, receipt)
|
||||
_atomic_json(config.state_root / "operations" / f"{op_id}.json", receipt)
|
||||
return receipt
|
||||
finally:
|
||||
fcntl.flock(lock_fd, fcntl.LOCK_UN)
|
||||
os.close(lock_fd)
|
||||
|
||||
|
||||
def config_from_environment() -> Config:
|
||||
return Config(
|
||||
git_dir=Path(os.environ.get("AIOS_HOST_SEED_GIT_DIR", "/data/aios-gitops/repo.git")),
|
||||
task_binding_root=Path(os.environ.get(
|
||||
"AIOS_HOST_SEED_TASK_BINDING_ROOT",
|
||||
"/var/lib/aios-gitops/bridge-v2-staging/task-bindings",
|
||||
)),
|
||||
state_root=Path(os.environ.get(
|
||||
"AIOS_HOST_SEED_STATE_ROOT",
|
||||
"/var/lib/aios-control-plane-host-seed-executor",
|
||||
)),
|
||||
release_root=Path(os.environ.get("AIOS_HOST_SEED_RELEASE_ROOT", "/opt/aios/releases")),
|
||||
host_identity=os.environ.get("AIOS_HOST_IDENTITY", "cn-network-mkt"),
|
||||
task_binding_owner=os.environ.get(
|
||||
"AIOS_HOST_SEED_TASK_BINDING_OWNER", "aios-bridge-staging"
|
||||
) or None,
|
||||
request_ref=os.environ.get("AIOS_HOST_SEED_REQUEST_REF", REQUEST_REF),
|
||||
request_path=os.environ.get("AIOS_HOST_SEED_REQUEST_PATH", REQUEST_PATH),
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("command", choices=("execute",))
|
||||
args = parser.parse_args()
|
||||
if args.command == "execute":
|
||||
result = execute(config_from_environment())
|
||||
print(json.dumps(result, sort_keys=True, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,52 +0,0 @@
|
||||
#!/bin/sh
|
||||
set -eu
|
||||
|
||||
if [ "$(id -u)" -ne 0 ]; then
|
||||
echo "ROOT_REQUIRED" >&2
|
||||
exit 20
|
||||
fi
|
||||
|
||||
SOURCE_ROOT=${1:-}
|
||||
if [ -z "$SOURCE_ROOT" ]; then
|
||||
echo "SOURCE_ROOT_REQUIRED" >&2
|
||||
exit 21
|
||||
fi
|
||||
|
||||
EXECUTOR="$SOURCE_ROOT/active-runtime/app/aios_control_plane_host_seed_executor.py"
|
||||
SERVICE="$SOURCE_ROOT/deploy/systemd/aios-control-plane-host-seed-executor.service"
|
||||
TIMER="$SOURCE_ROOT/deploy/systemd/aios-control-plane-host-seed-executor.timer"
|
||||
|
||||
for file in "$EXECUTOR" "$SERVICE" "$TIMER"; do
|
||||
if [ ! -f "$file" ] || [ -L "$file" ]; then
|
||||
echo "SOURCE_FILE_INVALID:$file" >&2
|
||||
exit 22
|
||||
fi
|
||||
done
|
||||
|
||||
AUTHORITY_GIT=/data/aios-gitops/repo.git
|
||||
REQUEST_REF=refs/heads/agent/host-seed-authorization-requests
|
||||
if [ ! -d "$AUTHORITY_GIT" ]; then
|
||||
echo "AUTHORITATIVE_GIT_UNAVAILABLE" >&2
|
||||
exit 23
|
||||
fi
|
||||
if ! id -u aios-bridge-staging >/dev/null 2>&1; then
|
||||
echo "BRIDGE_SERVICE_ACCOUNT_UNAVAILABLE" >&2
|
||||
exit 24
|
||||
fi
|
||||
if git --git-dir="$AUTHORITY_GIT" show-ref --verify --quiet "$REQUEST_REF"; then
|
||||
echo "HOST_SEED_REQUEST_REF_ALREADY_PRESENT" >&2
|
||||
exit 25
|
||||
fi
|
||||
|
||||
install -d -m 0755 /usr/local/libexec
|
||||
install -m 0755 "$EXECUTOR" /usr/local/libexec/aios-control-plane-host-seed-executor
|
||||
install -m 0644 "$SERVICE" /etc/systemd/system/aios-control-plane-host-seed-executor.service
|
||||
install -m 0644 "$TIMER" /etc/systemd/system/aios-control-plane-host-seed-executor.timer
|
||||
install -d -m 0700 /var/lib/aios-control-plane-host-seed-executor
|
||||
|
||||
systemctl daemon-reload
|
||||
systemctl enable --now aios-control-plane-host-seed-executor.timer
|
||||
systemctl is-enabled --quiet aios-control-plane-host-seed-executor.timer
|
||||
systemctl is-active --quiet aios-control-plane-host-seed-executor.timer
|
||||
|
||||
echo "HOST_SEED_EXECUTOR_INSTALL=PASS"
|
||||
@@ -1,15 +0,0 @@
|
||||
#!/bin/sh
|
||||
set -eu
|
||||
|
||||
if [ "$(id -u)" -ne 0 ]; then
|
||||
echo "ROOT_REQUIRED" >&2
|
||||
exit 20
|
||||
fi
|
||||
|
||||
systemctl disable --now aios-control-plane-host-seed-executor.timer 2>/dev/null || true
|
||||
rm -f /etc/systemd/system/aios-control-plane-host-seed-executor.timer
|
||||
rm -f /etc/systemd/system/aios-control-plane-host-seed-executor.service
|
||||
rm -f /usr/local/libexec/aios-control-plane-host-seed-executor
|
||||
systemctl daemon-reload
|
||||
|
||||
echo "HOST_SEED_EXECUTOR_ROLLBACK=PASS"
|
||||
@@ -1,41 +0,0 @@
|
||||
[Unit]
|
||||
Description=AIOS governed immutable Host seed executor
|
||||
After=local-fs.target
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
User=root
|
||||
Group=root
|
||||
UMask=0077
|
||||
Environment=PYTHONDONTWRITEBYTECODE=1
|
||||
Environment=AIOS_HOST_IDENTITY=cn-network-mkt
|
||||
Environment=AIOS_HOST_SEED_GIT_DIR=/data/aios-gitops/repo.git
|
||||
Environment=AIOS_HOST_SEED_TASK_BINDING_ROOT=/var/lib/aios-gitops/bridge-v2-staging/task-bindings
|
||||
Environment=AIOS_HOST_SEED_TASK_BINDING_OWNER=aios-bridge-staging
|
||||
Environment=AIOS_HOST_SEED_STATE_ROOT=/var/lib/aios-control-plane-host-seed-executor
|
||||
Environment=AIOS_HOST_SEED_RELEASE_ROOT=/opt/aios/releases
|
||||
Environment=AIOS_HOST_SEED_REQUEST_REF=refs/heads/agent/host-seed-authorization-requests
|
||||
Environment=AIOS_HOST_SEED_REQUEST_PATH=control-plane/host-seed/request.json
|
||||
ExecStart=/usr/bin/python3 -B /usr/local/libexec/aios-control-plane-host-seed-executor execute
|
||||
NoNewPrivileges=true
|
||||
PrivateTmp=true
|
||||
PrivateDevices=true
|
||||
ProtectSystem=strict
|
||||
ProtectHome=true
|
||||
ProtectKernelTunables=true
|
||||
ProtectKernelModules=true
|
||||
ProtectControlGroups=true
|
||||
ProtectKernelLogs=true
|
||||
LockPersonality=true
|
||||
RestrictSUIDSGID=true
|
||||
RestrictRealtime=true
|
||||
RestrictNamespaces=true
|
||||
MemoryDenyWriteExecute=true
|
||||
RestrictAddressFamilies=AF_UNIX
|
||||
IPAddressDeny=any
|
||||
CapabilityBoundingSet=CAP_DAC_OVERRIDE CAP_FOWNER CAP_CHOWN
|
||||
ReadOnlyPaths=/usr/local/libexec/aios-control-plane-host-seed-executor
|
||||
ReadOnlyPaths=/data/aios-gitops/repo.git
|
||||
ReadOnlyPaths=/var/lib/aios-gitops/bridge-v2-staging/task-bindings
|
||||
ReadWritePaths=/var/lib/aios-control-plane-host-seed-executor
|
||||
ReadWritePaths=/opt/aios/releases
|
||||
@@ -1,12 +0,0 @@
|
||||
[Unit]
|
||||
Description=Poll governed immutable Host seed requests
|
||||
|
||||
[Timer]
|
||||
OnBootSec=2min
|
||||
OnUnitActiveSec=30s
|
||||
AccuracySec=1s
|
||||
Persistent=true
|
||||
Unit=aios-control-plane-host-seed-executor.service
|
||||
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
@@ -0,0 +1,46 @@
|
||||
[Unit]
|
||||
Description=AIOS generic governed Online Agent Worker (2026-08-15)
|
||||
After=aios-agent-task-api.service aios-online-agent-worker-register-20260815.service
|
||||
Wants=aios-online-agent-worker-register-20260815.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=aios-engineering-worker
|
||||
Group=aios-engineering-worker
|
||||
SupplementaryGroups=gitops
|
||||
LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token
|
||||
Environment=PYTHONUNBUFFERED=1
|
||||
Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
|
||||
Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
|
||||
Environment=AIOS_AGENT_ID=online-engineering-worker-cn-network-mkt-generic-20260815
|
||||
Environment=AIOS_AGENT_TYPE=trusted-online-engineering-worker
|
||||
Environment=AIOS_AGENT_VERSION=2026-08-15.3-generic
|
||||
Environment=AIOS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
|
||||
Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-worker-20260815
|
||||
Environment=AIOS_WORKER_WORKSPACE_ROOT=/var/lib/aios-online-agent-worker-20260815/workspaces
|
||||
Environment=AIOS_WORKER_POLL_SECONDS=5
|
||||
Environment=AIOS_WORKER_HEARTBEAT_SECONDS=10
|
||||
Environment=AIOS_WORKER_LEASE_SECONDS=300
|
||||
ExecStart=/usr/bin/python3 -B /usr/local/lib/aios-online-agent-worker-20260815/online_agent_worker.py
|
||||
Restart=always
|
||||
RestartSec=5
|
||||
TimeoutStopSec=15
|
||||
NoNewPrivileges=true
|
||||
PrivateTmp=true
|
||||
ProtectSystem=strict
|
||||
ProtectHome=true
|
||||
ProtectKernelTunables=true
|
||||
ProtectKernelModules=true
|
||||
ProtectControlGroups=true
|
||||
LockPersonality=true
|
||||
RestrictSUIDSGID=true
|
||||
RestrictRealtime=true
|
||||
RestrictNamespaces=true
|
||||
MemoryDenyWriteExecute=true
|
||||
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
|
||||
ReadOnlyPaths=/data/aios-gitops/repo.git
|
||||
ReadWritePaths=/var/lib/aios-online-agent-worker-20260815
|
||||
UMask=0027
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -0,0 +1,36 @@
|
||||
[Unit]
|
||||
Description=AIOS generic Online Agent register-or-recover (2026-08-15)
|
||||
After=network-online.target aios-gitops-model-api.service aios-agent-task-api.service
|
||||
Wants=network-online.target
|
||||
Before=aios-online-agent-worker-20260815.service
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
RemainAfterExit=yes
|
||||
User=root
|
||||
Group=root
|
||||
LoadCredential=bridge_control_token:/run/credentials/aios-gitops-model-api.service/agent_task_control_token
|
||||
Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
|
||||
Environment=AIOS_AGENT_ID=online-engineering-worker-cn-network-mkt-generic-20260815
|
||||
Environment=AIOS_AGENT_TYPE=trusted-online-engineering-worker
|
||||
Environment=AIOS_AGENT_VERSION=2026-08-15.3-generic
|
||||
Environment=AIOS_WORKER_ENCRYPTED_CREDENTIAL=/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token
|
||||
Environment=AIOS_WORKER_REGISTRATION_STATE=/var/lib/aios-online-agent-worker-20260815/registration
|
||||
ExecStart=/usr/bin/python3 -B /usr/local/lib/aios-online-agent-worker-20260815/register.py
|
||||
NoNewPrivileges=true
|
||||
PrivateTmp=true
|
||||
ProtectSystem=strict
|
||||
ProtectHome=true
|
||||
ProtectKernelTunables=true
|
||||
ProtectKernelModules=true
|
||||
ProtectControlGroups=true
|
||||
LockPersonality=true
|
||||
RestrictSUIDSGID=true
|
||||
RestrictRealtime=true
|
||||
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
|
||||
ReadWritePaths=/etc/credstore.encrypted /var/lib/aios-online-agent-worker-20260815
|
||||
UMask=0077
|
||||
TimeoutStartSec=120
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -1,117 +0,0 @@
|
||||
# Server-local governed immutable Host seed executor
|
||||
|
||||
Date: 2026-08-22
|
||||
|
||||
## Goal
|
||||
|
||||
Provide a server-local, fail-closed way to publish an already validated
|
||||
immutable control-plane candidate without depending on Windows, an external
|
||||
Agent worker, SSH, or an arbitrary root command channel.
|
||||
|
||||
The executor is deliberately narrower than deployment:
|
||||
|
||||
- it reads only the authoritative local Git object database;
|
||||
- it consumes one fixed request ref and path;
|
||||
- it accepts only `HOST_SEED_IMMUTABLE_CANDIDATE`;
|
||||
- it requires a high-risk governed-task binding whose description binds the
|
||||
exact candidate intent *and* authorization expiry;
|
||||
- it materializes `base_oid + candidate.patch`, verifies every release-member
|
||||
SHA against the candidate manifest, and verifies the candidate Machine Goal;
|
||||
- it atomically publishes one immutable directory under `/opt/aios/releases`;
|
||||
- it never starts or restarts services, starts the Projector, creates the fixed
|
||||
continuation Task/Job, switches Production, or writes `main`, tags, or the
|
||||
baseline.
|
||||
|
||||
## Canonical server path
|
||||
|
||||
```text
|
||||
explicit Owner HOST_SEED authorization
|
||||
-> high-risk governed authorization task binding
|
||||
-> governed request commit on refs/heads/agent/host-seed-authorization-requests
|
||||
-> root timer polls local authoritative Git
|
||||
-> validate exact intent, expiry, binding SHA, candidate evidence
|
||||
-> read payload commit from local Git
|
||||
-> shared disposable clone at exact base
|
||||
-> apply declared patch
|
||||
-> verify exact changed set + all release-member bytes
|
||||
-> atomic immutable publish
|
||||
-> durable sanitized receipt
|
||||
```
|
||||
|
||||
The Host seed request is not the fixed continuation Task. The fixed
|
||||
continuation Task remains a separate later gate.
|
||||
|
||||
## Trust and authorization binding
|
||||
|
||||
The request contains an immutable `intent` with:
|
||||
|
||||
- candidate / release id;
|
||||
- base OID and payload commit;
|
||||
- fixed patch, manifest, Machine Goal and bootstrap paths;
|
||||
- validation artifact SHA;
|
||||
- Manifest, Machine Goal, release-tree and bootstrap SHAs;
|
||||
- exact target Host identity.
|
||||
|
||||
`intent_sha256` is the canonical SHA of that object. Authorization adds
|
||||
`task_id` and `expires_at`; those are hashed into `authorization_sha256`.
|
||||
The server-side governed-task binding must contain:
|
||||
|
||||
```text
|
||||
risk=high
|
||||
description=HOST_SEED_AUTHORIZATION:<authorization_sha256>
|
||||
target_branch=agent/host-seed-authorization-...
|
||||
```
|
||||
|
||||
The request also binds the exact byte SHA of that server-side binding. This
|
||||
prevents extending the authorization window or changing the candidate without
|
||||
creating a new governed authorization binding.
|
||||
|
||||
The executor additionally requires the binding file to be a non-symlink
|
||||
regular file, not group/world writable, and owned by the configured Bridge
|
||||
service account in production.
|
||||
|
||||
## Delivery model
|
||||
|
||||
The live delivery payload is server-native Git data, not the external ZIP.
|
||||
The payload commit carries:
|
||||
|
||||
```text
|
||||
control-plane/candidates/<candidate>/candidate.patch
|
||||
control-plane/candidates/<candidate>/implementation_manifest_*.json
|
||||
control-plane/candidates/<candidate>/machine_goal_*.json
|
||||
control-plane/candidates/<candidate>/aios_control_plane_immutable_seed_*.py
|
||||
```
|
||||
|
||||
The external ZIP SHA remains bound as validation evidence through Machine
|
||||
Goal, but the executor records `artifact_delivery_used=false`: it does not
|
||||
claim that ZIP bytes were copied to the Host.
|
||||
|
||||
The release tree itself is reconstructed and verified from authoritative Git
|
||||
objects plus the declared patch.
|
||||
|
||||
## Idempotency and recovery
|
||||
|
||||
Before publishing, the executor writes a durable `PENDING` operation record
|
||||
keyed by the exact intent SHA. The final receipt is `EXECUTED`.
|
||||
|
||||
If the immutable release directory already exists, every member must match the
|
||||
manifest before the operation is accepted as idempotent. Any drift fails
|
||||
closed.
|
||||
|
||||
A missing request ref is a normal `NO_REQUEST` poll and has no side effects.
|
||||
|
||||
## Systemd confinement
|
||||
|
||||
The service:
|
||||
|
||||
- runs as root only because immutable release publication requires it;
|
||||
- has no network access;
|
||||
- reads authoritative Git and Bridge task bindings read-only;
|
||||
- can write only its private state root and `/opt/aios/releases`;
|
||||
- has no service-control, Projector, Task, Git-ref, Production, tag or baseline
|
||||
operation in its code path.
|
||||
|
||||
The timer polls every 30 seconds. Installing/enabling this executor is itself a
|
||||
separate governed control-plane deployment. Installing it does **not** grant a
|
||||
Host seed authorization and does not seed any candidate while the fixed
|
||||
request ref is absent.
|
||||
@@ -0,0 +1,54 @@
|
||||
# Generic Governed Online Agent Worker
|
||||
|
||||
This worker is a pull-based, restartable consumer for `aios.agent-task.v1`.
|
||||
The service configuration contains only stable runtime identity, capability
|
||||
names, API endpoints, heartbeat timing, and the credential projection. It does
|
||||
not contain a task id, repository base OID, target branch, or plan-file path.
|
||||
|
||||
## Dynamic task contract
|
||||
|
||||
Engineering tasks carry `metadata.execution` with schema
|
||||
`aios.online-agent.execution.v2`. The payload supplies the repository, exact
|
||||
base OID, target branch, detached-worktree policy, adapter name, explicit file
|
||||
change set, verification commands, and governed commit idempotency keys.
|
||||
Task prose is descriptive only and is never passed to a shell.
|
||||
|
||||
The worker accepts only structured operations:
|
||||
|
||||
- `deterministic`: allow-listed file changes and test commands;
|
||||
- `codex`: fail-closed until a task supplies an explicit Codex file/command
|
||||
policy;
|
||||
- `hermes`: fail-closed because the current Hermes runtime is a user-chat
|
||||
service, not an engineering executor.
|
||||
|
||||
All command execution uses `shell=False`, fixed program allowlists, relative
|
||||
paths beneath an ephemeral detached worktree, and output redaction. Direct
|
||||
`git branch`, `update-ref`, `push`, `commit`, `reset`, `clean`, main,
|
||||
production, tag, and baseline mutations are rejected by the worker.
|
||||
|
||||
## Governed write path
|
||||
|
||||
The worker calls the Bridge in this order:
|
||||
|
||||
1. `POST /v2/governed-task`;
|
||||
2. `POST /v2/writer-lease`;
|
||||
3. `POST /v2/commit-files`;
|
||||
4. read-only authority-ref visibility verification.
|
||||
|
||||
CREATE_NEW is represented by the task payload and is normalized by the Bridge
|
||||
lease authority. The worker never creates an authoritative ref itself.
|
||||
|
||||
## Lifecycle and recovery
|
||||
|
||||
The runtime registers with `register_or_recover`, receives its agent token
|
||||
once, and stores only the encrypted systemd credential. The consumer loops over
|
||||
poll, atomically claims one task, sends the first heartbeat, emits
|
||||
`TASK_STARTED`, and continues heartbeats while executing.
|
||||
|
||||
State is atomically stored per task without agent or lease token material. If
|
||||
the process stops after a governed commit but before result submission, the
|
||||
next lease attempt reuses the stable governed idempotency key and submits the
|
||||
stored commit evidence without issuing a second commit.
|
||||
|
||||
The control plane must independently verify submitted results; worker-local
|
||||
success is not production or mainline authorization.
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Generic governed Online Agent Worker runtime."""
|
||||
|
||||
__all__ = ["worker"]
|
||||
@@ -0,0 +1,937 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generic, restartable Online Agent Worker.
|
||||
|
||||
The worker is deliberately task-agnostic. A task's signed/durable payload is
|
||||
the only source of repository, branch, change, verification, and governance
|
||||
parameters. There is no task id, base OID, branch name, or plan file in the
|
||||
runtime configuration.
|
||||
|
||||
The worker performs local work only in an ephemeral detached worktree. All
|
||||
authoritative ref updates go through the governed Bridge API; direct branch,
|
||||
update-ref, push, main, production, tag, and baseline mutations are rejected.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from abc import ABC, abstractmethod
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from typing import Any, Mapping
|
||||
from urllib.parse import unquote
|
||||
|
||||
|
||||
API_BASE = os.environ.get("AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940").rstrip("/")
|
||||
BRIDGE_BASE = os.environ.get("AIOS_GOVERNED_WRITE_API_URL", "http://127.0.0.1:18892").rstrip("/")
|
||||
AGENT_ID = os.environ.get(
|
||||
"AIOS_AGENT_ID", "online-engineering-worker-cn-network-mkt-generic-20260815"
|
||||
)
|
||||
AGENT_TYPE = os.environ.get("AIOS_AGENT_TYPE", "trusted-online-engineering-worker")
|
||||
AGENT_VERSION = os.environ.get("AIOS_AGENT_VERSION", "2026-08-15.3-generic")
|
||||
AUTHORITY_GIT_DIR = Path(
|
||||
os.environ.get("AIOS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git")
|
||||
)
|
||||
STATE_ROOT = Path(
|
||||
os.environ.get("AIOS_WORKER_STATE_ROOT", "/var/lib/aios-online-agent-worker-20260815")
|
||||
)
|
||||
WORKSPACE_ROOT = Path(
|
||||
os.environ.get(
|
||||
"AIOS_WORKER_WORKSPACE_ROOT", "/var/lib/aios-online-agent-worker-20260815/workspaces"
|
||||
)
|
||||
)
|
||||
GIT_CONFIG_GLOBAL = STATE_ROOT / "gitconfig"
|
||||
POLL_SECONDS = int(os.environ.get("AIOS_WORKER_POLL_SECONDS", "5"))
|
||||
HEARTBEAT_SECONDS = int(os.environ.get("AIOS_WORKER_HEARTBEAT_SECONDS", "20"))
|
||||
LEASE_SECONDS = int(os.environ.get("AIOS_WORKER_LEASE_SECONDS", "300"))
|
||||
MAX_TASK_BYTES = 512 * 1024
|
||||
MAX_FILE_BYTES = 2 * 1024 * 1024
|
||||
OID_RE = re.compile(r"^[0-9a-f]{40}$")
|
||||
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
|
||||
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
|
||||
BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
|
||||
PATH_RE = re.compile(r"^[^\\\x00-\x1f\x7f]+$")
|
||||
SENSITIVE_RE = re.compile(
|
||||
r"(?i)(authorization\s*:\s*bearer\s+\S+|(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*\S+)"
|
||||
)
|
||||
|
||||
|
||||
class WorkerError(RuntimeError):
|
||||
def __init__(self, code: str, *, phase: str, details: Mapping[str, Any] | None = None) -> None:
|
||||
super().__init__(code)
|
||||
self.code = code
|
||||
self.phase = phase
|
||||
self.details = dict(details or {})
|
||||
|
||||
|
||||
def now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def digest_bytes(value: bytes) -> str:
|
||||
return hashlib.sha256(value).hexdigest()
|
||||
|
||||
|
||||
def digest_json(value: Any) -> str:
|
||||
return digest_bytes(
|
||||
json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||||
)
|
||||
|
||||
|
||||
def sanitize(value: Any) -> Any:
|
||||
if isinstance(value, Mapping):
|
||||
cleaned: dict[str, Any] = {}
|
||||
for key, inner in value.items():
|
||||
lowered = str(key).lower()
|
||||
if any(marker in lowered for marker in ("token", "secret", "credential", "password", "authorization")):
|
||||
cleaned[str(key)] = "[REDACTED]"
|
||||
else:
|
||||
cleaned[str(key)] = sanitize(inner)
|
||||
return cleaned
|
||||
if isinstance(value, list):
|
||||
return [sanitize(item) for item in value]
|
||||
if isinstance(value, tuple):
|
||||
return [sanitize(item) for item in value]
|
||||
if isinstance(value, str):
|
||||
return SENSITIVE_RE.sub("[REDACTED]", value)
|
||||
return value
|
||||
|
||||
|
||||
def excerpt(value: str, limit: int = 800) -> str:
|
||||
value = str(sanitize(value))
|
||||
return value if len(value) <= limit else value[: limit // 2] + "...[TRUNCATED]..." + value[-limit // 2 :]
|
||||
|
||||
|
||||
def audit(event_type: str, **fields: Any) -> None:
|
||||
STATE_ROOT.mkdir(mode=0o750, parents=True, exist_ok=True)
|
||||
path = STATE_ROOT / "events.jsonl"
|
||||
record = {
|
||||
"event_type": event_type,
|
||||
"created_at": now(),
|
||||
"agent_id": AGENT_ID,
|
||||
"credential_material_exposed": False,
|
||||
"lease_material_exposed": False,
|
||||
**sanitize(fields),
|
||||
}
|
||||
with path.open("a", encoding="utf-8") as handle:
|
||||
handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
|
||||
os.chmod(path, 0o640)
|
||||
print(json.dumps(record, ensure_ascii=False, sort_keys=True), flush=True)
|
||||
|
||||
|
||||
def atomic_json(path: Path, value: Mapping[str, Any], mode: int = 0o640) -> None:
|
||||
path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
|
||||
if path.parent.is_symlink() or path.is_symlink():
|
||||
raise WorkerError("STATE_PATH_SYMLINK_REJECTED", phase="state")
|
||||
fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
|
||||
try:
|
||||
os.fchmod(fd, mode)
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
json.dump(sanitize(dict(value)), handle, ensure_ascii=False, sort_keys=True)
|
||||
handle.write("\n")
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
os.replace(temporary, path)
|
||||
os.chmod(path, mode)
|
||||
finally:
|
||||
if os.path.exists(temporary):
|
||||
os.unlink(temporary)
|
||||
|
||||
|
||||
def safe_relative_path(value: Any, *, field: str = "path") -> str:
|
||||
if not isinstance(value, str) or not value or len(value) > 240:
|
||||
raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
|
||||
if unquote(value) != value or not PATH_RE.fullmatch(value):
|
||||
raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
|
||||
if value.startswith(("/", "~")) or re.match(r"^[A-Za-z]:", value):
|
||||
raise WorkerError(f"{field.upper()}_ABSOLUTE", phase="validation")
|
||||
parts = value.replace("\\", "/").split("/")
|
||||
if any(part in {"", ".", ".."} for part in parts):
|
||||
raise WorkerError(f"{field.upper()}_TRAVERSAL", phase="validation")
|
||||
return "/".join(parts)
|
||||
|
||||
|
||||
def resolve_under(root: Path, relative: str, *, write: bool = False) -> Path:
|
||||
relative = safe_relative_path(relative)
|
||||
root = root.resolve()
|
||||
candidate = (root / relative).resolve(strict=False)
|
||||
if candidate != root and root not in candidate.parents:
|
||||
raise WorkerError("WORKSPACE_ESCAPE", phase="validation")
|
||||
current = root
|
||||
for part in Path(relative).parts:
|
||||
current /= part
|
||||
if current.exists() and current.is_symlink():
|
||||
raise WorkerError("WORKSPACE_SYMLINK_REJECTED", phase="validation")
|
||||
if write and candidate.exists() and candidate.is_symlink():
|
||||
raise WorkerError("WRITE_SYMLINK_REJECTED", phase="validation")
|
||||
return candidate
|
||||
|
||||
|
||||
def validate_oid(value: Any, field: str = "oid") -> str:
|
||||
if not isinstance(value, str) or not OID_RE.fullmatch(value):
|
||||
raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
|
||||
return value
|
||||
|
||||
|
||||
def validate_sha(value: Any, field: str = "sha256") -> str:
|
||||
if not isinstance(value, str) or not SHA256_RE.fullmatch(value):
|
||||
raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
|
||||
return value
|
||||
|
||||
|
||||
def validate_branch(value: Any) -> str:
|
||||
if not isinstance(value, str) or not 3 <= len(value) <= 180:
|
||||
raise WorkerError("TARGET_BRANCH_INVALID", phase="validation")
|
||||
if value in {"main", "master", "HEAD"} or value.startswith(
|
||||
("main/", "master/", "refs/", "tags/", "tag/", "release/", "releases/", "baseline/", "production/")
|
||||
):
|
||||
raise WorkerError("PROTECTED_BRANCH_REJECTED", phase="validation")
|
||||
if (
|
||||
value.startswith("/")
|
||||
or value.endswith(("/", ".", ".lock"))
|
||||
or ".." in value
|
||||
or "//" in value
|
||||
or "@{" in value
|
||||
or any(ch in value for ch in " ~^:?*[\\")
|
||||
or any(part in {"", ".", ".."} for part in value.split("/"))
|
||||
or not BRANCH_RE.fullmatch(value)
|
||||
):
|
||||
raise WorkerError("TARGET_BRANCH_INVALID", phase="validation")
|
||||
return value
|
||||
|
||||
|
||||
def validate_program(argv: Any, *, internal_git: bool = False) -> list[str]:
|
||||
if not isinstance(argv, list) or not argv or not all(isinstance(item, str) and item for item in argv):
|
||||
raise WorkerError("STRUCTURED_ARGV_REQUIRED", phase="validation")
|
||||
command = argv[0]
|
||||
if command not in {"/usr/bin/git", "/usr/bin/python3"}:
|
||||
raise WorkerError("PROGRAM_NOT_ALLOWLISTED", phase="validation", details={"program": Path(command).name})
|
||||
if command.endswith("python3"):
|
||||
if any(item in {"-c", "-", "--command"} for item in argv):
|
||||
raise WorkerError("PYTHON_INLINE_CODE_REJECTED", phase="validation")
|
||||
if "-m" in argv:
|
||||
index = argv.index("-m")
|
||||
if index + 1 >= len(argv) or argv[index + 1] not in {"pytest", "py_compile"}:
|
||||
raise WorkerError("PYTHON_MODULE_NOT_ALLOWLISTED", phase="validation")
|
||||
if command.endswith("git") and not internal_git:
|
||||
rejected = {"push", "update-ref", "receive-pack", "send-pack", "reset", "clean", "branch", "commit", "merge", "rebase"}
|
||||
if any(item in rejected for item in argv[1:]):
|
||||
raise WorkerError("GIT_MUTATION_NOT_ALLOWLISTED", phase="validation")
|
||||
return list(argv)
|
||||
|
||||
|
||||
def run_command(
|
||||
label: str,
|
||||
argv: list[str],
|
||||
*,
|
||||
cwd: Path,
|
||||
timeout: int,
|
||||
internal_git: bool = False,
|
||||
allowed_returncodes: tuple[int, ...] = (0,),
|
||||
) -> dict[str, Any]:
|
||||
validate_program(argv, internal_git=internal_git)
|
||||
if not 1 <= timeout <= 3600:
|
||||
raise WorkerError("COMMAND_TIMEOUT_INVALID", phase=label)
|
||||
started = time.monotonic()
|
||||
try:
|
||||
completed = subprocess.run(
|
||||
argv,
|
||||
cwd=str(cwd),
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
shell=False,
|
||||
timeout=timeout,
|
||||
env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8", "GIT_CONFIG_GLOBAL": str(GIT_CONFIG_GLOBAL)},
|
||||
)
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
raise WorkerError("COMMAND_TIMEOUT", phase=label, details={"timeout_seconds": timeout}) from exc
|
||||
result = {
|
||||
"label": label,
|
||||
"argv": [Path(item).name if index == 0 else item for index, item in enumerate(argv)],
|
||||
"exit_code": completed.returncode,
|
||||
"stdout_sha256": digest_bytes(completed.stdout.encode("utf-8", "replace")),
|
||||
"stderr_sha256": digest_bytes(completed.stderr.encode("utf-8", "replace")),
|
||||
"stdout_excerpt": excerpt(completed.stdout),
|
||||
"stderr_excerpt": excerpt(completed.stderr),
|
||||
"duration_ms": int((time.monotonic() - started) * 1000),
|
||||
}
|
||||
audit("COMMAND_RESULT", **result)
|
||||
if completed.returncode not in allowed_returncodes:
|
||||
raise WorkerError("COMMAND_FAILED", phase=label, details={"exit_code": completed.returncode, "stderr": excerpt(completed.stderr, 400)})
|
||||
return result
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExecutionEnvelope:
|
||||
task_id: str
|
||||
repository: str
|
||||
base_oid: str
|
||||
target_branch: str
|
||||
adapter: str
|
||||
change_set: tuple[dict[str, Any], ...]
|
||||
verification: tuple[dict[str, Any], ...]
|
||||
governance: dict[str, Any]
|
||||
workspace: dict[str, Any]
|
||||
digest: str
|
||||
|
||||
@classmethod
|
||||
def from_task(cls, task: Mapping[str, Any]) -> "ExecutionEnvelope":
|
||||
task_id = str(task.get("task_id") or "")
|
||||
if not ID_RE.fullmatch(task_id):
|
||||
raise WorkerError("TASK_ID_INVALID", phase="validation")
|
||||
metadata = task.get("metadata")
|
||||
if not isinstance(metadata, Mapping):
|
||||
raise WorkerError("TASK_METADATA_REQUIRED", phase="validation")
|
||||
execution = metadata.get("execution")
|
||||
if not isinstance(execution, Mapping):
|
||||
raise WorkerError("DYNAMIC_EXECUTION_PAYLOAD_REQUIRED", phase="validation")
|
||||
if execution.get("schema") != "aios.online-agent.execution.v2":
|
||||
raise WorkerError("EXECUTION_SCHEMA_INVALID", phase="validation")
|
||||
repository = execution.get("repository")
|
||||
if not isinstance(repository, str) or not repository:
|
||||
raise WorkerError("REPOSITORY_INVALID", phase="validation")
|
||||
base_oid = validate_oid(execution.get("base_oid"), "base_oid")
|
||||
target_branch = validate_branch(execution.get("target_branch"))
|
||||
workspace = execution.get("workspace")
|
||||
if not isinstance(workspace, Mapping):
|
||||
raise WorkerError("WORKSPACE_POLICY_REQUIRED", phase="validation")
|
||||
workspace_value = dict(workspace)
|
||||
if workspace_value.get("authority_git_dir") != str(AUTHORITY_GIT_DIR):
|
||||
raise WorkerError("AUTHORITY_ROOT_INVALID", phase="validation")
|
||||
if workspace_value.get("read_only_repository") is not True:
|
||||
raise WorkerError("REPOSITORY_WRITE_POLICY_INVALID", phase="validation")
|
||||
if workspace_value.get("cleanup_policy") not in {"always", "cleanup_on_success"}:
|
||||
raise WorkerError("WORKSPACE_CLEANUP_POLICY_INVALID", phase="validation")
|
||||
adapter_value = execution.get("adapter")
|
||||
if isinstance(adapter_value, Mapping):
|
||||
adapter = str(adapter_value.get("name") or "")
|
||||
else:
|
||||
adapter = str(adapter_value or "")
|
||||
if adapter not in {"deterministic", "codex", "hermes"}:
|
||||
raise WorkerError("RUNTIME_ADAPTER_UNSUPPORTED", phase="validation")
|
||||
change_set = execution.get("change_set")
|
||||
if not isinstance(change_set, list) or not change_set or len(change_set) > 128:
|
||||
raise WorkerError("CHANGE_SET_INVALID", phase="validation")
|
||||
normalized_changes: list[dict[str, Any]] = []
|
||||
for item in change_set:
|
||||
if not isinstance(item, Mapping):
|
||||
raise WorkerError("CHANGE_OPERATION_INVALID", phase="validation")
|
||||
operation = str(item.get("operation") or "")
|
||||
relative = safe_relative_path(item.get("path"))
|
||||
normalized = dict(item)
|
||||
normalized["operation"] = operation
|
||||
normalized["path"] = relative
|
||||
if operation == "write_file":
|
||||
content = item.get("content")
|
||||
if not isinstance(content, str) or len(content.encode("utf-8")) > MAX_FILE_BYTES:
|
||||
raise WorkerError("WRITE_CONTENT_INVALID", phase="validation")
|
||||
elif operation == "replace_text":
|
||||
if not isinstance(item.get("old"), str) or not item.get("old") or not isinstance(item.get("new"), str):
|
||||
raise WorkerError("REPLACE_TEXT_INVALID", phase="validation")
|
||||
if len(str(item["old"]).encode("utf-8")) > MAX_FILE_BYTES:
|
||||
raise WorkerError("REPLACE_TEXT_INVALID", phase="validation")
|
||||
elif operation == "delete_file":
|
||||
pass
|
||||
else:
|
||||
raise WorkerError("CHANGE_OPERATION_UNSUPPORTED", phase="validation", details={"operation": operation})
|
||||
expected = item.get("expected_sha256")
|
||||
if expected is not None:
|
||||
validate_sha(expected, "expected_sha256")
|
||||
normalized_changes.append(normalized)
|
||||
verification = execution.get("verification") or []
|
||||
if not isinstance(verification, list) or len(verification) > 128:
|
||||
raise WorkerError("VERIFICATION_INVALID", phase="validation")
|
||||
normalized_verification: list[dict[str, Any]] = []
|
||||
for item in verification:
|
||||
if not isinstance(item, Mapping):
|
||||
raise WorkerError("VERIFICATION_COMMAND_INVALID", phase="validation")
|
||||
label = str(item.get("label") or "verification")
|
||||
argv = validate_program(item.get("argv"))
|
||||
cwd_value = item.get("cwd")
|
||||
cwd = "." if cwd_value in {None, "", "."} else safe_relative_path(cwd_value, field="cwd")
|
||||
timeout = item.get("timeout_seconds", 120)
|
||||
if not isinstance(timeout, int) or isinstance(timeout, bool) or not 1 <= timeout <= 3600:
|
||||
raise WorkerError("VERIFICATION_TIMEOUT_INVALID", phase="validation")
|
||||
normalized_verification.append({"label": label, "argv": argv, "cwd": cwd, "timeout_seconds": timeout})
|
||||
governance = execution.get("governance")
|
||||
if not isinstance(governance, Mapping):
|
||||
raise WorkerError("GOVERNANCE_POLICY_REQUIRED", phase="validation")
|
||||
governance_value = dict(governance)
|
||||
governed_task_id = str(governance_value.get("task_id") or f"{task_id}-writer")
|
||||
if not ID_RE.fullmatch(governed_task_id):
|
||||
raise WorkerError("GOVERNED_TASK_ID_INVALID", phase="validation")
|
||||
idempotency_key = str(governance_value.get("idempotency_key") or f"{task_id}-governed-v1")
|
||||
commit_key = str(governance_value.get("commit_idempotency_key") or f"{task_id}-commit-v1")
|
||||
for field, value in (("idempotency_key", idempotency_key), ("commit_idempotency_key", commit_key)):
|
||||
if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,159}", value):
|
||||
raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
|
||||
governance_value.update(
|
||||
{
|
||||
"task_id": governed_task_id,
|
||||
"idempotency_key": idempotency_key,
|
||||
"commit_idempotency_key": commit_key,
|
||||
"message": str(governance_value.get("message") or f"chore(gitops): execute task {task_id}"),
|
||||
}
|
||||
)
|
||||
for field in ("main_write", "force_push", "production_write", "tag_write", "baseline_write", "protected_ref_bypass"):
|
||||
if governance_value.get(field, False) is not False:
|
||||
raise WorkerError("GOVERNANCE_POLICY_REJECTED", phase="validation", details={"field": field})
|
||||
allowed_files = governance_value.get("allowed_files")
|
||||
if allowed_files is not None:
|
||||
if not isinstance(allowed_files, list):
|
||||
raise WorkerError("ALLOWED_FILES_INVALID", phase="validation")
|
||||
allowed = {safe_relative_path(item, field="allowed_file") for item in allowed_files}
|
||||
if not {item["path"] for item in normalized_changes}.issubset(allowed):
|
||||
raise WorkerError("CHANGE_OUTSIDE_ALLOWED_FILES", phase="validation")
|
||||
governance_value["allowed_files"] = sorted(allowed)
|
||||
return cls(
|
||||
task_id=task_id,
|
||||
repository=repository,
|
||||
base_oid=base_oid,
|
||||
target_branch=target_branch,
|
||||
adapter=adapter,
|
||||
change_set=tuple(normalized_changes),
|
||||
verification=tuple(normalized_verification),
|
||||
governance=governance_value,
|
||||
workspace=workspace_value,
|
||||
digest=digest_json(execution),
|
||||
)
|
||||
|
||||
|
||||
class RuntimeAdapter(ABC):
|
||||
name: str
|
||||
|
||||
@abstractmethod
|
||||
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class DeterministicExecutor(RuntimeAdapter):
|
||||
name = "deterministic"
|
||||
|
||||
@staticmethod
|
||||
def _write_atomic(path: Path, content: str) -> None:
|
||||
path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
|
||||
fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
handle.write(content)
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
os.replace(temporary, path)
|
||||
finally:
|
||||
if os.path.exists(temporary):
|
||||
os.unlink(temporary)
|
||||
|
||||
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
|
||||
changed: set[str] = set()
|
||||
for change in envelope.change_set:
|
||||
relative = str(change["path"])
|
||||
path = resolve_under(worktree, relative, write=True)
|
||||
expected = change.get("expected_sha256")
|
||||
current = path.read_bytes() if path.is_file() else None
|
||||
if expected is not None and (current is None or digest_bytes(current) != expected):
|
||||
raise WorkerError("WORKTREE_SOURCE_DRIFT", phase="implementing", details={"path": relative})
|
||||
operation = change["operation"]
|
||||
if operation == "write_file":
|
||||
self._write_atomic(path, str(change["content"]))
|
||||
elif operation == "replace_text":
|
||||
if current is None:
|
||||
raise WorkerError("REPLACE_SOURCE_MISSING", phase="implementing", details={"path": relative})
|
||||
original = current.decode("utf-8")
|
||||
old = str(change["old"])
|
||||
count = original.count(old)
|
||||
expected_count = int(change.get("count", 1))
|
||||
if count != expected_count:
|
||||
raise WorkerError("REPLACE_TEXT_MATCH_COUNT_INVALID", phase="implementing", details={"path": relative, "count": count})
|
||||
self._write_atomic(path, original.replace(old, str(change["new"])))
|
||||
elif operation == "delete_file":
|
||||
if path.exists():
|
||||
path.unlink()
|
||||
changed.add(relative)
|
||||
audit("REPOSITORY_CHANGE_APPLIED", operation=operation, path=relative, sha256=digest_bytes(path.read_bytes()) if path.exists() else None)
|
||||
commands: list[dict[str, Any]] = []
|
||||
if changed:
|
||||
commands.append(run_command("git_intent_to_add", ["/usr/bin/git", "-C", str(worktree), "add", "-N", "--", *sorted(changed)], cwd=worktree, timeout=120, internal_git=True))
|
||||
commands.append(run_command("git_diff_check", ["/usr/bin/git", "-C", str(worktree), "diff", "--check"], cwd=worktree, timeout=120))
|
||||
commands.append(run_command("git_diff_stat", ["/usr/bin/git", "-C", str(worktree), "diff", "--no-ext-diff", "--stat"], cwd=worktree, timeout=120))
|
||||
tests: list[dict[str, Any]] = []
|
||||
for item in envelope.verification:
|
||||
cwd = worktree if item["cwd"] == "." else resolve_under(worktree, item["cwd"])
|
||||
tests.append(run_command(item["label"], item["argv"], cwd=cwd, timeout=item["timeout_seconds"]))
|
||||
return {
|
||||
"changed_files": sorted(changed),
|
||||
"commands": commands,
|
||||
"tests": tests,
|
||||
"adapter": self.name,
|
||||
}
|
||||
|
||||
|
||||
class CodexExecutor(RuntimeAdapter):
|
||||
name = "codex"
|
||||
|
||||
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
|
||||
raise WorkerError(
|
||||
"CODEX_ADAPTER_REQUIRES_EXPLICIT_POLICY",
|
||||
phase="adapter",
|
||||
details={"reason": "Codex is not allowed to mutate a governed worktree without an explicit file/command policy"},
|
||||
)
|
||||
|
||||
|
||||
class HermesExecutor(RuntimeAdapter):
|
||||
name = "hermes"
|
||||
|
||||
def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
|
||||
raise WorkerError(
|
||||
"HERMES_ENGINEERING_ADAPTER_UNAVAILABLE",
|
||||
phase="adapter",
|
||||
details={"reason": "The live Hermes service is a user-chat runtime, not an engineering executor"},
|
||||
)
|
||||
|
||||
|
||||
def adapter_for(name: str) -> RuntimeAdapter:
|
||||
return {"deterministic": DeterministicExecutor, "codex": CodexExecutor, "hermes": HermesExecutor}[name]()
|
||||
|
||||
|
||||
class HttpClient:
|
||||
@staticmethod
|
||||
def call(base: str, method: str, route: str, token: str | None, body: Mapping[str, Any] | None = None, timeout: int = 30) -> tuple[int, dict[str, Any]]:
|
||||
payload = None if body is None else json.dumps(body, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||||
headers = {"Accept": "application/json"}
|
||||
if token:
|
||||
headers["Authorization"] = "Bearer " + token
|
||||
if payload is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
request = urllib.request.Request(base + route, data=payload, headers=headers, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=timeout) as response:
|
||||
value = json.loads(response.read().decode("utf-8"))
|
||||
return response.status, value if isinstance(value, dict) else {}
|
||||
except urllib.error.HTTPError as exc:
|
||||
raw = exc.read(8192).decode("utf-8", "replace")
|
||||
try:
|
||||
value = json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
value = {"error": "HTTP_ERROR"}
|
||||
return exc.code, value if isinstance(value, dict) else {"error": "HTTP_ERROR"}
|
||||
except (urllib.error.URLError, TimeoutError):
|
||||
return 599, {"error": "SERVICE_UNAVAILABLE"}
|
||||
|
||||
|
||||
class AgentTaskClient:
|
||||
def __init__(self, token: str) -> None:
|
||||
self.token = token
|
||||
|
||||
def poll(self) -> list[dict[str, Any]]:
|
||||
status, response = HttpClient.call(API_BASE, "POST", "/v1/tasks/poll", self.token, {"limit": 100})
|
||||
if status != 200:
|
||||
raise WorkerError("TASK_POLL_FAILED", phase="poll", details={"api_status": status, "error": response.get("error")})
|
||||
tasks = response.get("tasks")
|
||||
return [dict(item) for item in tasks if isinstance(item, Mapping)] if isinstance(tasks, list) else []
|
||||
|
||||
def claim(self, task_id: str) -> tuple[int, dict[str, Any]]:
|
||||
return HttpClient.call(API_BASE, "POST", f"/v1/tasks/{task_id}/claim", self.token, {"lease_seconds": LEASE_SECONDS})
|
||||
|
||||
def heartbeat(self, task_id: str, lease_token: str, stage: str) -> tuple[int, dict[str, Any]]:
|
||||
return HttpClient.call(
|
||||
API_BASE,
|
||||
"POST",
|
||||
f"/v1/tasks/{task_id}/heartbeat",
|
||||
self.token,
|
||||
{"lease_token": lease_token, "lease_seconds": LEASE_SECONDS, "progress": {"stage": stage, "worker_version": AGENT_VERSION}},
|
||||
)
|
||||
|
||||
def result(self, task_id: str, lease_token: str, idempotency_key: str, result: Mapping[str, Any]) -> tuple[int, dict[str, Any]]:
|
||||
return HttpClient.call(
|
||||
API_BASE,
|
||||
"POST",
|
||||
f"/v1/tasks/{task_id}/result",
|
||||
self.token,
|
||||
{"lease_token": lease_token, "idempotency_key": idempotency_key, "result": dict(result)},
|
||||
)
|
||||
|
||||
|
||||
class BridgeClient:
|
||||
def call(self, route: str, body: Mapping[str, Any]) -> tuple[int, dict[str, Any]]:
|
||||
return HttpClient.call(BRIDGE_BASE, "POST", route, None, body, timeout=60)
|
||||
|
||||
|
||||
class LeaseHeartbeat:
|
||||
def __init__(self, client: AgentTaskClient, task_id: str, lease_token: str) -> None:
|
||||
self.client = client
|
||||
self.task_id = task_id
|
||||
self.lease_token = lease_token
|
||||
self.stage = "claimed"
|
||||
self.failed: str | None = None
|
||||
self.stop_event = threading.Event()
|
||||
self.lock = threading.Lock()
|
||||
self.thread = threading.Thread(target=self._run, name="online-agent-heartbeat", daemon=True)
|
||||
|
||||
def first(self) -> None:
|
||||
status, response = self.client.heartbeat(self.task_id, self.lease_token, self.stage)
|
||||
if not 200 <= status < 300:
|
||||
self.failed = str(response.get("error") or status)
|
||||
raise WorkerError("TASK_HEARTBEAT_FAILED", phase="heartbeat", details={"api_status": status, "error": self.failed})
|
||||
audit("TASK_HEARTBEAT", task_id=self.task_id, stage=self.stage, api_status=status)
|
||||
|
||||
def start(self) -> None:
|
||||
self.thread.start()
|
||||
|
||||
def set_stage(self, stage: str) -> None:
|
||||
with self.lock:
|
||||
self.stage = stage
|
||||
|
||||
def _run(self) -> None:
|
||||
while not self.stop_event.wait(HEARTBEAT_SECONDS):
|
||||
with self.lock:
|
||||
stage = self.stage
|
||||
try:
|
||||
status, response = self.client.heartbeat(self.task_id, self.lease_token, stage)
|
||||
if not 200 <= status < 300:
|
||||
self.failed = str(response.get("error") or status)
|
||||
audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, stage=stage, api_status=status, error=self.failed)
|
||||
else:
|
||||
audit("TASK_HEARTBEAT", task_id=self.task_id, stage=stage, api_status=status)
|
||||
except Exception as exc: # pragma: no cover - defensive runtime guard
|
||||
self.failed = type(exc).__name__
|
||||
audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, stage=stage, error=self.failed)
|
||||
|
||||
def ensure(self) -> None:
|
||||
if self.failed:
|
||||
raise WorkerError("TASK_LEASE_LOST", phase="heartbeat", details={"error": self.failed})
|
||||
|
||||
def close(self) -> None:
|
||||
self.stop_event.set()
|
||||
if self.thread.ident is not None:
|
||||
self.thread.join(timeout=2)
|
||||
|
||||
|
||||
def load_agent_token() -> str:
|
||||
directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
|
||||
path = directory / "agent_token"
|
||||
if not path.is_file() or path.is_symlink():
|
||||
raise WorkerError("AGENT_RUNTIME_CREDENTIAL_UNAVAILABLE", phase="startup")
|
||||
value = path.read_text(encoding="utf-8").strip()
|
||||
if len(value) < 24:
|
||||
raise WorkerError("AGENT_RUNTIME_CREDENTIAL_INVALID", phase="startup")
|
||||
return value
|
||||
|
||||
|
||||
def ensure_roots() -> None:
|
||||
if AUTHORITY_GIT_DIR.is_symlink() or STATE_ROOT.is_symlink() or WORKSPACE_ROOT.is_symlink():
|
||||
raise WorkerError("ROOT_SYMLINK_REJECTED", phase="startup")
|
||||
if not AUTHORITY_GIT_DIR.is_dir():
|
||||
raise WorkerError("AUTHORITY_GIT_DIR_UNAVAILABLE", phase="startup")
|
||||
STATE_ROOT.mkdir(mode=0o750, parents=True, exist_ok=True)
|
||||
WORKSPACE_ROOT.mkdir(mode=0o750, parents=True, exist_ok=True)
|
||||
if GIT_CONFIG_GLOBAL.is_symlink():
|
||||
raise WorkerError("GIT_CONFIG_SYMLINK_REJECTED", phase="startup")
|
||||
expected = f"[safe]\n\tdirectory = {AUTHORITY_GIT_DIR}\n"
|
||||
if GIT_CONFIG_GLOBAL.exists() and GIT_CONFIG_GLOBAL.read_text(encoding="utf-8") != expected:
|
||||
raise WorkerError("GIT_CONFIG_BINDING_CONFLICT", phase="startup")
|
||||
if not GIT_CONFIG_GLOBAL.exists():
|
||||
GIT_CONFIG_GLOBAL.write_text(expected, encoding="utf-8")
|
||||
os.chmod(GIT_CONFIG_GLOBAL, 0o640)
|
||||
|
||||
|
||||
def task_state_path(task_id: str) -> Path:
|
||||
if not ID_RE.fullmatch(task_id):
|
||||
raise WorkerError("TASK_ID_INVALID", phase="state")
|
||||
return STATE_ROOT / "tasks" / f"{task_id}.json"
|
||||
|
||||
|
||||
def load_task_state(task_id: str, execution_digest: str) -> dict[str, Any] | None:
|
||||
path = task_state_path(task_id)
|
||||
if not path.is_file() or path.is_symlink():
|
||||
return None
|
||||
try:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
raise WorkerError("TASK_STATE_INVALID", phase="state")
|
||||
if not isinstance(value, dict):
|
||||
raise WorkerError("TASK_STATE_INVALID", phase="state")
|
||||
if value.get("execution_digest") not in {None, execution_digest}:
|
||||
raise WorkerError("TASK_PAYLOAD_DRIFT", phase="recovery")
|
||||
return value
|
||||
|
||||
|
||||
def save_task_state(task_id: str, value: Mapping[str, Any]) -> None:
|
||||
atomic_json(task_state_path(task_id), value)
|
||||
|
||||
|
||||
def new_run_root() -> Path:
|
||||
ensure_roots()
|
||||
root = Path(tempfile.mkdtemp(prefix="task-", dir=str(WORKSPACE_ROOT)))
|
||||
os.chmod(root, 0o750)
|
||||
return root
|
||||
|
||||
|
||||
def cleanup_run_root(path: Path) -> None:
|
||||
root = WORKSPACE_ROOT.resolve()
|
||||
candidate = path.resolve(strict=False)
|
||||
if candidate == root or root not in candidate.parents:
|
||||
raise WorkerError("CLEANUP_SCOPE_INVALID", phase="cleanup")
|
||||
if candidate.exists() or candidate.is_symlink():
|
||||
shutil.rmtree(candidate)
|
||||
|
||||
|
||||
def prepare_worktree(run_root: Path, base_oid: str) -> Path:
|
||||
clone = run_root / "repository"
|
||||
worktree = run_root / "worktree"
|
||||
run_command("workspace_clone", ["/usr/bin/git", "clone", "--no-local", "--no-hardlinks", str(AUTHORITY_GIT_DIR), str(clone)], cwd=run_root, timeout=300, internal_git=True)
|
||||
run_command("workspace_base_probe", ["/usr/bin/git", "-C", str(clone), "rev-parse", "--verify", base_oid + "^{commit}"], cwd=run_root, timeout=60)
|
||||
run_command("workspace_worktree_create", ["/usr/bin/git", "-C", str(clone), "worktree", "add", "--detach", str(worktree), base_oid], cwd=run_root, timeout=300, internal_git=True)
|
||||
return worktree
|
||||
|
||||
|
||||
def governed_commit(envelope: ExecutionEnvelope, worktree: Path, bridge: BridgeClient, heartbeat: LeaseHeartbeat) -> dict[str, Any]:
|
||||
governance = envelope.governance
|
||||
create_body = {
|
||||
"task_id": governance["task_id"],
|
||||
"repository": envelope.repository,
|
||||
"base_oid": envelope.base_oid,
|
||||
"target_branch": envelope.target_branch,
|
||||
"risk": str(governance.get("risk") or "engineering"),
|
||||
"description": str(governance.get("description") or f"Governed execution for {envelope.task_id}"),
|
||||
"idempotency_key": governance["idempotency_key"],
|
||||
}
|
||||
status, response = bridge.call("/v2/governed-task", create_body)
|
||||
if not 200 <= status < 300:
|
||||
raise WorkerError("GOVERNED_TASK_CREATE_FAILED", phase="committing", details={"api_status": status, "error": response.get("error")})
|
||||
generation = int(response.get("generation") or 0)
|
||||
expected_branch_oid = str(response.get("expected_branch_oid") or envelope.base_oid)
|
||||
branch_mode = str(response.get("branch_mode") or "CREATE_NEW")
|
||||
lease_body = {
|
||||
"schema": "aios.task-writer-lease.v2",
|
||||
"task_id": governance["task_id"],
|
||||
"repository": envelope.repository,
|
||||
"branch_mode": branch_mode,
|
||||
"base_oid": envelope.base_oid,
|
||||
"expected_branch_oid": expected_branch_oid,
|
||||
"target_branch": envelope.target_branch,
|
||||
"generation": generation,
|
||||
"lease_seconds": int(governance.get("lease_seconds") or 900),
|
||||
}
|
||||
status, lease_response = bridge.call("/v2/writer-lease", lease_body)
|
||||
if not 200 <= status < 300:
|
||||
raise WorkerError("WRITER_LEASE_ACQUIRE_FAILED", phase="committing", details={"api_status": status, "error": lease_response.get("error")})
|
||||
heartbeat.ensure()
|
||||
files: dict[str, str] = {}
|
||||
paths = governance.get("allowed_files") or [item["path"] for item in envelope.change_set]
|
||||
for relative in sorted({safe_relative_path(item, field="allowed_file") for item in paths}):
|
||||
path = resolve_under(worktree, relative)
|
||||
if not path.is_file():
|
||||
raise WorkerError("COMMIT_FILE_MISSING", phase="committing", details={"path": relative})
|
||||
files[relative] = base64.b64encode(path.read_bytes()).decode("ascii")
|
||||
commit_body = {
|
||||
"task_id": governance["task_id"],
|
||||
"repository": envelope.repository,
|
||||
"expected_branch_oid": expected_branch_oid,
|
||||
"target_branch": envelope.target_branch,
|
||||
"generation": generation,
|
||||
"idempotency_key": governance["commit_idempotency_key"],
|
||||
"files": files,
|
||||
"message": governance["message"],
|
||||
}
|
||||
status, commit_response = bridge.call("/v2/commit-files", commit_body)
|
||||
if not 200 <= status < 300:
|
||||
raise WorkerError("GOVERNED_COMMIT_FAILED", phase="committing", details={"api_status": status, "error": commit_response.get("error")})
|
||||
commit_oid = commit_response.get("commit_oid")
|
||||
if not isinstance(commit_oid, str) or not OID_RE.fullmatch(commit_oid):
|
||||
raise WorkerError("GOVERNED_COMMIT_OID_INVALID", phase="committing")
|
||||
visible_oid: str | None = None
|
||||
for _ in range(5):
|
||||
probe = subprocess.run(
|
||||
["/usr/bin/git", "--git-dir", str(AUTHORITY_GIT_DIR), "rev-parse", "--verify", f"refs/heads/{envelope.target_branch}"],
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8", "GIT_CONFIG_GLOBAL": str(GIT_CONFIG_GLOBAL)},
|
||||
timeout=30,
|
||||
)
|
||||
if probe.returncode == 0:
|
||||
visible_oid = probe.stdout.strip()
|
||||
if visible_oid == commit_oid:
|
||||
break
|
||||
time.sleep(1)
|
||||
if visible_oid != commit_oid:
|
||||
raise WorkerError("GIT_VISIBILITY_MISMATCH", phase="verification", details={"commit_oid": commit_oid, "visible_oid": visible_oid})
|
||||
return {
|
||||
"governed_task_id": governance["task_id"],
|
||||
"branch_mode": branch_mode,
|
||||
"generation": generation,
|
||||
"target_branch": envelope.target_branch,
|
||||
"commit_oid": commit_oid,
|
||||
"visible_oid": visible_oid,
|
||||
"commit_idempotency_key": governance["commit_idempotency_key"],
|
||||
"api_commit_visible": commit_response.get("commit_visible"),
|
||||
"direct_main_write": False,
|
||||
"force_push": False,
|
||||
"production_write": False,
|
||||
"protected_ref_bypass": False,
|
||||
}
|
||||
|
||||
|
||||
def build_result(task_id: str, envelope: ExecutionEnvelope, execution: Mapping[str, Any], governed: Mapping[str, Any]) -> dict[str, Any]:
|
||||
return {
|
||||
"status": "SUCCEEDED",
|
||||
"verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW",
|
||||
"task_id": task_id,
|
||||
"execution_backend": "generic-online-agent-worker-v2",
|
||||
"adapter": envelope.adapter,
|
||||
"execution_digest": envelope.digest,
|
||||
"repository": envelope.repository,
|
||||
"base_oid": envelope.base_oid,
|
||||
"target_branch": envelope.target_branch,
|
||||
"changed_files": execution.get("changed_files", []),
|
||||
"commands": [{k: item.get(k) for k in ("label", "exit_code", "stdout_sha256", "stderr_sha256", "duration_ms")} for item in execution.get("commands", [])],
|
||||
"tests": [{k: item.get(k) for k in ("label", "exit_code", "stdout_sha256", "stderr_sha256", "duration_ms")} for item in execution.get("tests", [])],
|
||||
"governed": dict(governed),
|
||||
"workspace_cleanup": "PASS",
|
||||
"secret_material_exposed": False,
|
||||
"direct_main_write": False,
|
||||
"force_push": False,
|
||||
"production_write": False,
|
||||
"protected_ref_bypass": False,
|
||||
}
|
||||
|
||||
|
||||
class WorkerSupervisor:
|
||||
def __init__(self, token: str) -> None:
|
||||
self.agent = AgentTaskClient(token)
|
||||
self.bridge = BridgeClient()
|
||||
|
||||
def process_claim(self, task: Mapping[str, Any], claim: Mapping[str, Any]) -> None:
|
||||
task_id = str(claim.get("task_id") or task.get("task_id") or "")
|
||||
lease_token = claim.get("lease_token")
|
||||
if not isinstance(lease_token, str) or len(lease_token) < 24:
|
||||
raise WorkerError("LEASE_TOKEN_NOT_RETURNED", phase="claim")
|
||||
audit("TASK_CLAIMED", task_id=task_id, attempt=claim.get("attempt"), revision=claim.get("revision"))
|
||||
heartbeat = LeaseHeartbeat(self.agent, task_id, lease_token)
|
||||
try:
|
||||
heartbeat.first()
|
||||
heartbeat.set_stage("started")
|
||||
heartbeat.start()
|
||||
audit("TASK_STARTED", task_id=task_id, attempt=claim.get("attempt"))
|
||||
envelope = ExecutionEnvelope.from_task(task)
|
||||
state = load_task_state(task_id, envelope.digest) or {"task_id": task_id, "execution_digest": envelope.digest, "phase": "claimed"}
|
||||
governed = state.get("governed") if isinstance(state.get("governed"), Mapping) else None
|
||||
if isinstance(governed, Mapping) and isinstance(governed.get("commit_oid"), str) and OID_RE.fullmatch(governed["commit_oid"]):
|
||||
result = state.get("result")
|
||||
if not isinstance(result, Mapping):
|
||||
result = {"status": "SUCCEEDED", "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW", "task_id": task_id, "execution_digest": envelope.digest, "governed": dict(governed), "recovered_without_duplicate_commit": True, "direct_main_write": False, "production_write": False, "force_push": False, "protected_ref_bypass": False}
|
||||
audit("TASK_RECOVERED_COMMIT", task_id=task_id, commit_oid=governed["commit_oid"], duplicate_commit=False)
|
||||
else:
|
||||
run_root = new_run_root()
|
||||
try:
|
||||
worktree = prepare_worktree(run_root, envelope.base_oid)
|
||||
state.update({"phase": "workspace_prepared", "execution_digest": envelope.digest, "attempt": claim.get("attempt")})
|
||||
save_task_state(task_id, state)
|
||||
adapter = adapter_for(envelope.adapter)
|
||||
execution = adapter.execute(envelope, worktree)
|
||||
state.update({"phase": "verification_passed", "execution": execution})
|
||||
save_task_state(task_id, state)
|
||||
heartbeat.ensure()
|
||||
heartbeat.set_stage("committing")
|
||||
governed = governed_commit(envelope, worktree, self.bridge, heartbeat)
|
||||
result = build_result(task_id, envelope, execution, governed)
|
||||
state.update({"phase": "commit_done", "governed": governed, "result": result})
|
||||
save_task_state(task_id, state)
|
||||
finally:
|
||||
cleanup_run_root(run_root)
|
||||
audit("WORKSPACE_CLEANED", task_id=task_id, cleanup="PASS")
|
||||
heartbeat.ensure()
|
||||
heartbeat.set_stage("result")
|
||||
result_key = str(result.get("governed", {}).get("commit_idempotency_key") or f"{task_id}-result-{envelope.digest[:16]}")
|
||||
status, response = self.agent.result(task_id, lease_token, result_key, result)
|
||||
if not 200 <= status < 300:
|
||||
raise WorkerError("TASK_RESULT_SUBMIT_FAILED", phase="result", details={"api_status": status, "error": response.get("error")})
|
||||
audit("TASK_RESULT_SUBMITTED", task_id=task_id, api_status=status, result_idempotency_key=result_key, commit_oid=result.get("governed", {}).get("commit_oid"))
|
||||
except WorkerError as exc:
|
||||
audit("TASK_FAILED", task_id=task_id, error=exc.code, phase=exc.phase, details=exc.details)
|
||||
failure_key = f"{task_id}-attempt-{claim.get('attempt', 0)}-failure"
|
||||
failure = {
|
||||
"status": "FAILED",
|
||||
"verification_state": "UNVERIFIED",
|
||||
"task_id": task_id,
|
||||
"execution_backend": "generic-online-agent-worker-v2",
|
||||
"error": exc.code,
|
||||
"phase": exc.phase,
|
||||
"details": exc.details,
|
||||
"direct_main_write": False,
|
||||
"force_push": False,
|
||||
"production_write": False,
|
||||
"protected_ref_bypass": False,
|
||||
"secret_material_exposed": False,
|
||||
}
|
||||
try:
|
||||
status, response = self.agent.result(task_id, lease_token, failure_key, failure)
|
||||
audit("TASK_FAILURE_RESULT_SUBMITTED", task_id=task_id, api_status=status, error=response.get("error"))
|
||||
except Exception as submit_exc: # pragma: no cover - defensive runtime guard
|
||||
audit("TASK_FAILURE_RESULT_SUBMIT_FAILED", task_id=task_id, error=type(submit_exc).__name__)
|
||||
finally:
|
||||
heartbeat.close()
|
||||
|
||||
def once(self) -> int:
|
||||
try:
|
||||
tasks = self.agent.poll()
|
||||
audit("TASK_POLL", visible_count=len(tasks))
|
||||
except WorkerError as exc:
|
||||
audit("TASK_POLL_FAILED", error=exc.code, phase=exc.phase, details=exc.details)
|
||||
return 10
|
||||
dynamic_tasks: list[dict[str, Any]] = []
|
||||
for task in tasks:
|
||||
metadata = task.get("metadata")
|
||||
execution = metadata.get("execution") if isinstance(metadata, Mapping) else None
|
||||
if not isinstance(execution, Mapping) or execution.get("schema") != "aios.online-agent.execution.v2":
|
||||
audit("TASK_SKIPPED_NON_DYNAMIC", task_id=task.get("task_id"), reason="DYNAMIC_EXECUTION_PAYLOAD_REQUIRED")
|
||||
continue
|
||||
dynamic_tasks.append(task)
|
||||
for task in dynamic_tasks:
|
||||
task_id = str(task.get("task_id") or "")
|
||||
status, claim = self.agent.claim(task_id)
|
||||
if status == 409:
|
||||
audit("TASK_CLAIM_SKIPPED", task_id=task_id, api_status=status, error=claim.get("error"))
|
||||
continue
|
||||
if status != 200:
|
||||
audit("TASK_CLAIM_FAILED", task_id=task_id, api_status=status, error=claim.get("error"))
|
||||
continue
|
||||
self.process_claim(task, claim)
|
||||
return 0
|
||||
return 0
|
||||
|
||||
def run_forever(self) -> int:
|
||||
ensure_roots()
|
||||
audit("WORKER_STARTED", task_agnostic=True, api_base=API_BASE, bridge_base=BRIDGE_BASE, adapter_names=["deterministic", "codex", "hermes"], runtime_only_credentials=True)
|
||||
while True:
|
||||
self.once()
|
||||
time.sleep(POLL_SECONDS)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
token = load_agent_token()
|
||||
supervisor = WorkerSupervisor(token)
|
||||
if os.environ.get("AIOS_WORKER_ONCE") == "1":
|
||||
ensure_roots()
|
||||
return supervisor.once()
|
||||
return supervisor.run_forever()
|
||||
except WorkerError as exc:
|
||||
audit("WORKER_START_FAILED", error=exc.code, phase=exc.phase, details=exc.details)
|
||||
return 20
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,157 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Register or recover the generic Online Agent using runtime-only secrets."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
|
||||
BASE = os.environ.get("AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940").rstrip("/")
|
||||
AGENT_ID = os.environ.get("AIOS_AGENT_ID", "online-engineering-worker-cn-network-mkt-generic-20260815")
|
||||
AGENT_TYPE = os.environ.get("AIOS_AGENT_TYPE", "trusted-online-engineering-worker")
|
||||
AGENT_VERSION = os.environ.get("AIOS_AGENT_VERSION", "2026-08-15.3-generic")
|
||||
PROJECTION_PROVIDER = "BridgeV2RuntimeCredentialProjection"
|
||||
STATE_DIR = Path(os.environ.get("AIOS_WORKER_REGISTRATION_STATE", "/var/lib/aios-online-agent-worker-20260815/registration"))
|
||||
CREDENTIAL_PATH = Path(os.environ.get("AIOS_WORKER_ENCRYPTED_CREDENTIAL", "/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token"))
|
||||
IDENTITY_PATH = STATE_DIR / "identity.json"
|
||||
AUDIT_PATH = STATE_DIR / "events.jsonl"
|
||||
CAPABILITIES = [
|
||||
"online_agent_task_api", "task.poll", "task.claim", "task.heartbeat", "task.result",
|
||||
"engineering.consumer", "engineering.execute", "workspace.worktree", "git.read", "git.diff",
|
||||
"git.test", "governed.write.branch", "governed.write.file",
|
||||
]
|
||||
|
||||
|
||||
def now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def audit(event_type: str, **fields: Any) -> None:
|
||||
STATE_DIR.mkdir(mode=0o750, parents=True, exist_ok=True)
|
||||
record = {"event_type": event_type, "created_at": now(), "agent_id": AGENT_ID, "credential_material_exposed": False, **fields}
|
||||
with AUDIT_PATH.open("a", encoding="utf-8") as handle:
|
||||
handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
|
||||
os.chmod(AUDIT_PATH, 0o640)
|
||||
print(json.dumps(record, ensure_ascii=False, sort_keys=True), flush=True)
|
||||
|
||||
|
||||
def api_call(method: str, route: str, token: str, body: dict[str, Any] | None = None) -> tuple[int, dict[str, Any]]:
|
||||
payload = None if body is None else json.dumps(body, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||||
headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
if payload is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
request = urllib.request.Request(BASE + route, data=payload, headers=headers, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=15) as response:
|
||||
value = json.loads(response.read().decode("utf-8"))
|
||||
return response.status, value if isinstance(value, dict) else {}
|
||||
except urllib.error.HTTPError as exc:
|
||||
try:
|
||||
value = json.loads(exc.read(2048).decode("utf-8"))
|
||||
except Exception:
|
||||
value = {"error": "HTTP_ERROR"}
|
||||
return exc.code, value if isinstance(value, dict) else {"error": "HTTP_ERROR"}
|
||||
except (urllib.error.URLError, TimeoutError):
|
||||
return 599, {"error": "AGENT_TASK_API_UNAVAILABLE"}
|
||||
|
||||
|
||||
def read_identity() -> dict[str, Any] | None:
|
||||
if not IDENTITY_PATH.is_file() or IDENTITY_PATH.is_symlink():
|
||||
return None
|
||||
try:
|
||||
value = json.loads(IDENTITY_PATH.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
return None
|
||||
return value if isinstance(value, dict) else None
|
||||
|
||||
|
||||
def identity_matches(value: dict[str, Any] | None) -> bool:
|
||||
return bool(value and value.get("agent_id") == AGENT_ID and value.get("agent_type") == AGENT_TYPE and value.get("agent_version") == AGENT_VERSION and value.get("credential_projection_provider") == PROJECTION_PROVIDER and value.get("task_scope") == "dynamic")
|
||||
|
||||
|
||||
def encrypted_credential_is_valid() -> bool:
|
||||
if not CREDENTIAL_PATH.is_file() or CREDENTIAL_PATH.is_symlink():
|
||||
return False
|
||||
result = subprocess.run(["systemd-creds", "decrypt", "--with-key=host", str(CREDENTIAL_PATH), "/dev/null"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, timeout=15)
|
||||
return result.returncode == 0
|
||||
|
||||
|
||||
def write_identity() -> None:
|
||||
value = {"agent_id": AGENT_ID, "agent_type": AGENT_TYPE, "agent_version": AGENT_VERSION, "task_scope": "dynamic", "credential_projection_provider": PROJECTION_PROVIDER, "credential_delivery": "systemd-LoadCredentialEncrypted", "task_api_url": BASE, "task_api_schema": "aios.agent-task.v1", "execution_backend": "generic-online-agent-worker-v2", "runtime_only_secret_delivery": True, "production_write": False, "main_write": False, "force_push": False, "tag_write": False, "baseline_write": False, "writer_lease": True, "created_at": now()}
|
||||
STATE_DIR.mkdir(mode=0o750, parents=True, exist_ok=True)
|
||||
fd, temporary = tempfile.mkstemp(prefix="identity.", dir=str(STATE_DIR))
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
json.dump(value, handle, ensure_ascii=False, sort_keys=True)
|
||||
handle.write("\n")
|
||||
os.chmod(temporary, 0o640)
|
||||
os.replace(temporary, IDENTITY_PATH)
|
||||
finally:
|
||||
if os.path.exists(temporary):
|
||||
os.unlink(temporary)
|
||||
|
||||
|
||||
def encrypt_agent_token(agent_token: str) -> None:
|
||||
CREDENTIAL_PATH.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
fd, temporary = tempfile.mkstemp(prefix=".online-agent-token.", dir=str(CREDENTIAL_PATH.parent))
|
||||
os.close(fd)
|
||||
os.unlink(temporary)
|
||||
try:
|
||||
result = subprocess.run(["systemd-creds", "encrypt", "--with-key=host", "--name=agent_token", "/dev/stdin", temporary], input=(agent_token + "\n").encode("utf-8"), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, timeout=20)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError("AGENT_CREDENTIAL_ENCRYPT_FAILED")
|
||||
os.chmod(temporary, 0o600)
|
||||
os.replace(temporary, CREDENTIAL_PATH)
|
||||
os.chmod(CREDENTIAL_PATH, 0o600)
|
||||
finally:
|
||||
if os.path.exists(temporary):
|
||||
os.unlink(temporary)
|
||||
|
||||
|
||||
def register_or_recover() -> str:
|
||||
STATE_DIR.mkdir(mode=0o750, parents=True, exist_ok=True)
|
||||
existing = read_identity()
|
||||
if CREDENTIAL_PATH.exists():
|
||||
if not identity_matches(existing):
|
||||
raise RuntimeError("AGENT_IDENTITY_BINDING_CONFLICT")
|
||||
if not encrypted_credential_is_valid():
|
||||
raise RuntimeError("AGENT_ENCRYPTED_CREDENTIAL_INVALID")
|
||||
audit("AGENT_RECOVERED", credential_delivery="systemd-LoadCredentialEncrypted", credential_projection_provider=PROJECTION_PROVIDER, register_or_recover=True)
|
||||
return "RECOVERED"
|
||||
if existing is not None:
|
||||
raise RuntimeError("AGENT_CREDENTIAL_MISSING_AFTER_PRIOR_REGISTRATION")
|
||||
credential_directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
|
||||
control_path = credential_directory / "bridge_control_token"
|
||||
if not control_path.is_file():
|
||||
raise RuntimeError("BRIDGE_RUNTIME_CONTROL_CREDENTIAL_UNAVAILABLE")
|
||||
control_token = control_path.read_text(encoding="utf-8").strip()
|
||||
if len(control_token) < 32:
|
||||
raise RuntimeError("BRIDGE_RUNTIME_CONTROL_CREDENTIAL_INVALID")
|
||||
status, response = api_call("POST", "/v1/agents/register", control_token, {"agent_id": AGENT_ID, "agent_type": AGENT_TYPE, "agent_version": AGENT_VERSION, "capabilities": CAPABILITIES, "metadata": {"task_scope": "dynamic", "credential_projection_provider": PROJECTION_PROVIDER, "runtime_only_secret_delivery": True, "task_api_schema": "aios.agent-task.v1", "execution_backend": "generic-online-agent-worker-v2", "production_write": False, "main_write": False, "force_push": False, "tag_write": False, "baseline_write": False, "writer_lease": True}})
|
||||
if status == 409 and response.get("error") == "AGENT_ALREADY_EXISTS":
|
||||
raise RuntimeError("AGENT_ALREADY_EXISTS_RECOVERY_CREDENTIAL_UNAVAILABLE")
|
||||
if status != 201:
|
||||
raise RuntimeError("AGENT_REGISTER_FAILED_" + str(response.get("error") or status))
|
||||
agent_token = response.get("agent_token")
|
||||
if not isinstance(agent_token, str) or len(agent_token) < 24:
|
||||
raise RuntimeError("AGENT_TOKEN_NOT_RETURNED")
|
||||
encrypt_agent_token(agent_token)
|
||||
write_identity()
|
||||
audit("AGENT_REGISTERED", credential_delivery="systemd-LoadCredentialEncrypted", credential_projection_provider=PROJECTION_PROVIDER, register_or_recover=True)
|
||||
return "REGISTERED"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
raise SystemExit(0 if register_or_recover() in {"REGISTERED", "RECOVERED"} else 20)
|
||||
except Exception as exc:
|
||||
audit("AGENT_REGISTRATION_FAILED", error=type(exc).__name__)
|
||||
raise SystemExit(20)
|
||||
@@ -0,0 +1,79 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from aios_online_agent_worker.worker import ExecutionEnvelope, WorkerError, sanitize # noqa: E402
|
||||
|
||||
|
||||
BASE = "a" * 40
|
||||
|
||||
|
||||
def payload(task_id: str, branch: str, *, base_oid: str = BASE) -> dict[str, object]:
|
||||
return {
|
||||
"task_id": task_id,
|
||||
"metadata": {
|
||||
"execution": {
|
||||
"schema": "aios.online-agent.execution.v2",
|
||||
"repository": "aios/aios",
|
||||
"base_oid": base_oid,
|
||||
"target_branch": branch,
|
||||
"adapter": "deterministic",
|
||||
"workspace": {
|
||||
"authority_git_dir": str(Path(os.environ.get("AIOS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git"))),
|
||||
"read_only_repository": True,
|
||||
"cleanup_policy": "always",
|
||||
},
|
||||
"change_set": [
|
||||
{
|
||||
"operation": "write_file",
|
||||
"path": "tests/generated_marker.py",
|
||||
"content": "def test_marker():\n assert True\n",
|
||||
"expected_sha256": None,
|
||||
}
|
||||
],
|
||||
"verification": [
|
||||
{"label": "compile", "argv": ["/usr/bin/python3", "-m", "py_compile", "tests/generated_marker.py"], "cwd": ".", "timeout_seconds": 60}
|
||||
],
|
||||
"governance": {
|
||||
"task_id": f"{task_id}-writer",
|
||||
"idempotency_key": f"{task_id}-governed-v1",
|
||||
"commit_idempotency_key": f"{task_id}-commit-v1",
|
||||
"allowed_files": ["tests/generated_marker.py"],
|
||||
"main_write": False,
|
||||
"force_push": False,
|
||||
"production_write": False,
|
||||
"tag_write": False,
|
||||
"baseline_write": False,
|
||||
"protected_ref_bypass": False,
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_dynamic_payload_is_not_bound_to_task_or_base() -> None:
|
||||
first = ExecutionEnvelope.from_task(payload("task-alpha-20260815", "agent/alpha-20260815"))
|
||||
second = ExecutionEnvelope.from_task(payload("task-beta-20260815", "agent/beta-20260815", base_oid="b" * 40))
|
||||
assert first.task_id != second.task_id
|
||||
assert first.base_oid != second.base_oid
|
||||
assert first.target_branch != second.target_branch
|
||||
|
||||
|
||||
def test_protected_branch_is_rejected() -> None:
|
||||
try:
|
||||
ExecutionEnvelope.from_task(payload("task-main-20260815", "main"))
|
||||
except WorkerError as exc:
|
||||
assert exc.code == "PROTECTED_BRANCH_REJECTED"
|
||||
else:
|
||||
raise AssertionError("protected branch accepted")
|
||||
|
||||
|
||||
def test_task_payload_does_not_persist_secret_values() -> None:
|
||||
value = sanitize({"agent_token": "agt-secret", "nested": {"authorization": "Bearer secret"}})
|
||||
assert value == {"agent_token": "[REDACTED]", "nested": {"authorization": "[REDACTED]"}}
|
||||
|
||||
|
||||
def test_execution_digest_is_stable() -> None:
|
||||
value = payload("task-digest-20260815", "agent/digest-20260815")
|
||||
assert ExecutionEnvelope.from_task(value).digest == ExecutionEnvelope.from_task(value).digest
|
||||
@@ -1,328 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
MODULE_PATH = Path(__file__).parents[1] / "active-runtime" / "app" / "aios_control_plane_host_seed_executor.py"
|
||||
spec = importlib.util.spec_from_file_location("host_seed_executor", MODULE_PATH)
|
||||
m = importlib.util.module_from_spec(spec)
|
||||
assert spec and spec.loader
|
||||
import sys
|
||||
sys.modules[spec.name] = m
|
||||
spec.loader.exec_module(m)
|
||||
|
||||
|
||||
def run(*args: str, cwd: Path) -> str:
|
||||
return subprocess.check_output(args, cwd=cwd, text=True).strip()
|
||||
|
||||
|
||||
def sha(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def canonical(value: object) -> bytes:
|
||||
return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
|
||||
|
||||
|
||||
def setup_case(tmp_path: Path, *, risk: str = "high", tamper_bootstrap: bool = False):
|
||||
repo = tmp_path / "src"
|
||||
bare = tmp_path / "repo.git"
|
||||
repo.mkdir()
|
||||
run("git", "init", "-b", "main", cwd=repo)
|
||||
run("git", "config", "user.email", "test@example.invalid", cwd=repo)
|
||||
run("git", "config", "user.name", "test", cwd=repo)
|
||||
(repo / "a.txt").write_text("base\n", encoding="utf-8")
|
||||
(repo / "bin.sh").write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
||||
(repo / "bin.sh").chmod(0o755)
|
||||
run("git", "add", "-A", cwd=repo)
|
||||
run("git", "commit", "-m", "base", cwd=repo)
|
||||
base = run("git", "rev-parse", "HEAD", cwd=repo)
|
||||
run("git", "clone", "--bare", str(repo), str(bare), cwd=tmp_path)
|
||||
|
||||
(repo / "a.txt").write_text("changed\n", encoding="utf-8")
|
||||
(repo / "c.txt").write_text("added\n", encoding="utf-8")
|
||||
run("git", "add", "-N", "c.txt", cwd=repo)
|
||||
patch = subprocess.check_output(["git", "diff", "--binary", base], cwd=repo)
|
||||
after = {
|
||||
"a.txt": sha((repo / "a.txt").read_bytes()),
|
||||
"bin.sh": sha((repo / "bin.sh").read_bytes()),
|
||||
"c.txt": sha((repo / "c.txt").read_bytes()),
|
||||
}
|
||||
package_members = {f"payload/repo/after/{k}": v for k, v in after.items()}
|
||||
release_tree = sha(json.dumps({k: package_members[k] for k in sorted(package_members)}, sort_keys=True, separators=(",", ":")).encode())
|
||||
manifest = {
|
||||
"schema": "aios.control-plane-immutable-candidate.v6",
|
||||
"candidate": "V9",
|
||||
"repository": "aios/aios",
|
||||
"base_oid": base,
|
||||
"release_id": "control-plane-test-v9",
|
||||
"changed_file_count": 2,
|
||||
"changed_files": [
|
||||
{"path": "a.txt", "status": "modified", "after_sha256": after["a.txt"]},
|
||||
{"path": "c.txt", "status": "added", "after_sha256": after["c.txt"]},
|
||||
],
|
||||
"release_member_count": 3,
|
||||
"package_members": package_members,
|
||||
"release_tree_sha256": release_tree,
|
||||
}
|
||||
manifest_raw = canonical(manifest)
|
||||
bootstrap = b"#!/usr/bin/env python3\nprint('seed helper evidence only')\n"
|
||||
bootstrap_sha = sha(bootstrap)
|
||||
artifact_sha = "a" * 64
|
||||
goal = {
|
||||
"schema": "aios.control-plane-immutable-candidate-goal.v6",
|
||||
"candidate": "V9",
|
||||
"repository": "aios/aios",
|
||||
"base_oid": base,
|
||||
"release_id": "control-plane-test-v9",
|
||||
"artifact_sha256": artifact_sha,
|
||||
"manifest_sha256": sha(manifest_raw),
|
||||
"release_tree_sha256": release_tree,
|
||||
"seed_bootstrap_sha256": bootstrap_sha,
|
||||
"host_seed_authorized": False,
|
||||
}
|
||||
goal_raw = canonical(goal)
|
||||
|
||||
# Reset the source worktree to base; delivery files live in a separate commit.
|
||||
run("git", "reset", "--hard", base, cwd=repo)
|
||||
delivery = repo / "control-plane" / "candidates" / "v9"
|
||||
delivery.mkdir(parents=True)
|
||||
(delivery / "candidate.patch").write_bytes(patch)
|
||||
(delivery / "implementation_manifest_v9.json").write_bytes(manifest_raw)
|
||||
(delivery / "machine_goal_v9.json").write_bytes(goal_raw)
|
||||
(delivery / "aios_control_plane_immutable_seed_v9.py").write_bytes(
|
||||
b"tampered\n" if tamper_bootstrap else bootstrap
|
||||
)
|
||||
run("git", "add", "-A", cwd=repo)
|
||||
run("git", "commit", "-m", "candidate payload", cwd=repo)
|
||||
payload_commit = run("git", "rev-parse", "HEAD", cwd=repo)
|
||||
run("git", "push", str(bare), f"{payload_commit}:refs/heads/agent/candidate-payload", cwd=repo)
|
||||
|
||||
intent = {
|
||||
"candidate": "V9",
|
||||
"release_id": "control-plane-test-v9",
|
||||
"base_oid": base,
|
||||
"payload_commit": payload_commit,
|
||||
"patch_path": "control-plane/candidates/v9/candidate.patch",
|
||||
"manifest_path": "control-plane/candidates/v9/implementation_manifest_v9.json",
|
||||
"machine_goal_path": "control-plane/candidates/v9/machine_goal_v9.json",
|
||||
"seed_bootstrap_path": "control-plane/candidates/v9/aios_control_plane_immutable_seed_v9.py",
|
||||
"artifact_sha256": artifact_sha,
|
||||
"manifest_sha256": sha(manifest_raw),
|
||||
"machine_goal_sha256": sha(goal_raw),
|
||||
"release_tree_sha256": release_tree,
|
||||
"seed_bootstrap_sha256": bootstrap_sha,
|
||||
"target_host": "cn-network-mkt",
|
||||
}
|
||||
intent_digest = sha(canonical(intent))
|
||||
task_id = "host-seed-authz-v9-cn-network-mkt"
|
||||
now = datetime(2026, 8, 22, 12, 0, tzinfo=timezone.utc)
|
||||
expires_text = (now + timedelta(minutes=30)).isoformat().replace("+00:00", "Z")
|
||||
authorization_digest = sha(canonical({
|
||||
"task_id": task_id,
|
||||
"intent_sha256": intent_digest,
|
||||
"expires_at": expires_text,
|
||||
}))
|
||||
binding = {
|
||||
"task_id": task_id,
|
||||
"repository": "aios/aios",
|
||||
"base_oid": payload_commit,
|
||||
"target_branch": "agent/host-seed-authorization-v9-cn-network-mkt",
|
||||
"risk": risk,
|
||||
"description": "HOST_SEED_AUTHORIZATION:" + authorization_digest,
|
||||
"idempotency_key": "host-seed-authz-v9-1",
|
||||
"expected_branch_oid": payload_commit,
|
||||
"branch_mode": "CREATE_NEW",
|
||||
"generation": 1,
|
||||
"job_id": "job-" + "1" * 32,
|
||||
}
|
||||
binding_root = tmp_path / "bindings"
|
||||
binding_root.mkdir()
|
||||
binding_raw = json.dumps(binding, sort_keys=True).encode()
|
||||
(binding_root / f"{task_id}.json").write_bytes(binding_raw)
|
||||
(binding_root / f"{task_id}.json").chmod(0o600)
|
||||
|
||||
request = {
|
||||
"schema": "aios.control-plane-host-seed-request.v1",
|
||||
"operation": "HOST_SEED_IMMUTABLE_CANDIDATE",
|
||||
"repository": "aios/aios",
|
||||
"intent": intent,
|
||||
"authorization": {
|
||||
"task_id": task_id,
|
||||
"binding_sha256": sha(binding_raw),
|
||||
"expires_at": expires_text,
|
||||
"intent_sha256": intent_digest,
|
||||
"authorization_sha256": authorization_digest,
|
||||
},
|
||||
}
|
||||
request_path = repo / "control-plane" / "host-seed" / "request.json"
|
||||
request_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
request_path.write_bytes(canonical(request))
|
||||
run("git", "add", str(request_path.relative_to(repo)), cwd=repo)
|
||||
run("git", "commit", "-m", "host seed request", cwd=repo)
|
||||
request_commit = run("git", "rev-parse", "HEAD", cwd=repo)
|
||||
run("git", "push", str(bare), f"{request_commit}:{m.REQUEST_REF}", cwd=repo)
|
||||
|
||||
config = m.Config(
|
||||
git_dir=bare,
|
||||
task_binding_root=binding_root,
|
||||
state_root=tmp_path / "state",
|
||||
release_root=tmp_path / "releases",
|
||||
host_identity="cn-network-mkt",
|
||||
task_binding_owner=None,
|
||||
)
|
||||
return config, now, after, request
|
||||
|
||||
|
||||
def test_execute_publishes_exact_release_and_is_idempotent(tmp_path: Path):
|
||||
config, now, after, request = setup_case(tmp_path)
|
||||
receipt = m.execute(config, now=now, require_root=False)
|
||||
assert receipt["state"] == "EXECUTED"
|
||||
assert receipt["service_mutation"] is False
|
||||
assert receipt["service_restart"] is False
|
||||
assert receipt["projector_started"] is False
|
||||
assert receipt["task_creation"] is False
|
||||
assert receipt["production_write"] is False
|
||||
final = config.release_root / "control-plane-test-v9"
|
||||
assert final.is_dir() and not final.is_symlink()
|
||||
assert final.stat().st_mode & 0o777 == 0o755
|
||||
assert sha((final / "a.txt").read_bytes()) == after["a.txt"]
|
||||
assert sha((final / "bin.sh").read_bytes()) == after["bin.sh"]
|
||||
assert sha((final / "c.txt").read_bytes()) == after["c.txt"]
|
||||
assert (final / "bin.sh").stat().st_mode & 0o111
|
||||
again = m.execute(config, now=now, require_root=False)
|
||||
assert again == receipt
|
||||
|
||||
|
||||
def test_execute_rejects_non_high_risk_authorization(tmp_path: Path):
|
||||
config, now, _, _ = setup_case(tmp_path, risk="engineering")
|
||||
with pytest.raises(m.HostSeedError, match="AUTHORIZATION_RISK_MISMATCH"):
|
||||
m.execute(config, now=now, require_root=False)
|
||||
assert not config.release_root.exists()
|
||||
|
||||
|
||||
def test_execute_rejects_tampered_seed_bootstrap(tmp_path: Path):
|
||||
config, now, _, _ = setup_case(tmp_path, tamper_bootstrap=True)
|
||||
with pytest.raises(m.HostSeedError, match="SEED_BOOTSTRAP_HASH_MISMATCH"):
|
||||
m.execute(config, now=now, require_root=False)
|
||||
assert not config.release_root.exists()
|
||||
|
||||
|
||||
def test_request_rejects_wrong_host(tmp_path: Path):
|
||||
config, now, _, _ = setup_case(tmp_path)
|
||||
config = m.Config(
|
||||
git_dir=config.git_dir,
|
||||
task_binding_root=config.task_binding_root,
|
||||
state_root=config.state_root,
|
||||
release_root=config.release_root,
|
||||
host_identity="other-host",
|
||||
task_binding_owner=None,
|
||||
)
|
||||
with pytest.raises(m.HostSeedError, match="TARGET_HOST_MISMATCH"):
|
||||
m.execute(config, now=now, require_root=False)
|
||||
|
||||
|
||||
def test_existing_release_drift_is_fail_closed(tmp_path: Path):
|
||||
config, now, _, _ = setup_case(tmp_path)
|
||||
first = m.execute(config, now=now, require_root=False)
|
||||
assert first["state"] == "EXECUTED"
|
||||
(config.release_root / "control-plane-test-v9" / "a.txt").write_text("drift\n")
|
||||
# remove durable receipt so existing-release verification path is exercised.
|
||||
for p in (config.state_root / "receipts").glob("*.json"):
|
||||
p.unlink()
|
||||
with pytest.raises(m.HostSeedError, match="EXISTING_RELEASE_HASH_MISMATCH"):
|
||||
m.execute(config, now=now, require_root=False)
|
||||
|
||||
|
||||
def test_no_request_ref_is_safe_noop(tmp_path: Path):
|
||||
repo = tmp_path / "src"
|
||||
bare = tmp_path / "repo.git"
|
||||
repo.mkdir()
|
||||
run("git", "init", "-b", "main", cwd=repo)
|
||||
run("git", "config", "user.email", "test@example.invalid", cwd=repo)
|
||||
run("git", "config", "user.name", "test", cwd=repo)
|
||||
(repo / "a").write_text("x\n")
|
||||
run("git", "add", "a", cwd=repo)
|
||||
run("git", "commit", "-m", "base", cwd=repo)
|
||||
run("git", "clone", "--bare", str(repo), str(bare), cwd=tmp_path)
|
||||
cfg = m.Config(
|
||||
git_dir=bare,
|
||||
task_binding_root=tmp_path / "bindings",
|
||||
state_root=tmp_path / "state",
|
||||
release_root=tmp_path / "releases",
|
||||
host_identity="cn-network-mkt",
|
||||
task_binding_owner=None,
|
||||
)
|
||||
result = m.execute(cfg, now=datetime(2026, 8, 22, tzinfo=timezone.utc), require_root=False)
|
||||
assert result["state"] == "NO_REQUEST"
|
||||
assert result["service_mutation"] is False
|
||||
assert result["task_creation"] is False
|
||||
assert not cfg.release_root.exists()
|
||||
|
||||
|
||||
def test_authorization_expiry_cannot_be_extended_without_new_binding(tmp_path: Path):
|
||||
config, now, _, _ = setup_case(tmp_path)
|
||||
# Rewrite only the request expiry and recompute the request commit, but keep
|
||||
# the original governed-task binding description. The authorization digest
|
||||
# therefore no longer matches the binding and must fail closed.
|
||||
repo = tmp_path / "src"
|
||||
req_path = repo / "control-plane" / "host-seed" / "request.json"
|
||||
value = json.loads(req_path.read_text())
|
||||
value["authorization"]["expires_at"] = (now + timedelta(minutes=45)).isoformat().replace("+00:00", "Z")
|
||||
value["authorization"]["authorization_sha256"] = sha(canonical({
|
||||
"task_id": value["authorization"]["task_id"],
|
||||
"intent_sha256": value["authorization"]["intent_sha256"],
|
||||
"expires_at": value["authorization"]["expires_at"],
|
||||
}))
|
||||
req_path.write_bytes(canonical(value))
|
||||
run("git", "add", str(req_path.relative_to(repo)), cwd=repo)
|
||||
run("git", "commit", "-m", "tamper expiry", cwd=repo)
|
||||
new_commit = run("git", "rev-parse", "HEAD", cwd=repo)
|
||||
run("git", "push", "--force", str(config.git_dir), f"{new_commit}:{m.REQUEST_REF}", cwd=repo)
|
||||
with pytest.raises(m.HostSeedError, match="AUTHORIZATION_DESCRIPTION_MISMATCH"):
|
||||
m.execute(config, now=now, require_root=False)
|
||||
|
||||
|
||||
def test_systemd_contract_has_no_network_or_implicit_activation():
|
||||
root = Path(__file__).parents[1]
|
||||
service = (root / "deploy/systemd/aios-control-plane-host-seed-executor.service").read_text()
|
||||
timer = (root / "deploy/systemd/aios-control-plane-host-seed-executor.timer").read_text()
|
||||
assert "User=root" in service
|
||||
assert "IPAddressDeny=any" in service
|
||||
assert "RestrictAddressFamilies=AF_UNIX" in service
|
||||
assert "ReadOnlyPaths=/data/aios-gitops/repo.git" in service
|
||||
assert "ReadWritePaths=/opt/aios/releases" in service
|
||||
assert "systemctl start" not in service
|
||||
assert "projector" not in service.lower()
|
||||
assert "OnUnitActiveSec=30s" in timer
|
||||
|
||||
|
||||
def test_release_root_symlink_is_rejected(tmp_path: Path):
|
||||
config, now, _, _ = setup_case(tmp_path)
|
||||
real = tmp_path / "real-releases"
|
||||
real.mkdir()
|
||||
config.release_root.symlink_to(real, target_is_directory=True)
|
||||
with pytest.raises(m.HostSeedError, match="RELEASE_ROOT_SYMLINK_REJECTED"):
|
||||
m.execute(config, now=now, require_root=False)
|
||||
|
||||
def test_install_and_rollback_scripts_have_valid_bash_syntax():
|
||||
root = Path(__file__).parents[1]
|
||||
for relative in (
|
||||
"deploy/install/install_control_plane_host_seed_executor.sh",
|
||||
"deploy/install/rollback_control_plane_host_seed_executor.sh",
|
||||
):
|
||||
subprocess.run(
|
||||
["/usr/bin/bash", "-n", str(root / relative)],
|
||||
check=True,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user