Compare commits

...
24 changed files with 6765 additions and 2 deletions
@@ -0,0 +1,470 @@
#!/usr/bin/env python3
"""Canonical AIOS all-modified-files evidence packager.
The bundle contract is intentionally compatible with the historical
aios_package_all_modified_files_v1.sh output while keeping secret scanning in
version-controlled, tested code.
"""
from __future__ import annotations
from datetime import datetime, timezone
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import tempfile
import zipfile
from typing import Iterable
OID_RE = re.compile(r"^[0-9a-f]{40}$")
TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
# A PEM private-key finding requires the complete PEM header. Source code,
# tests, docs, or error text containing only "BEGIN PRIVATE KEY" are not
# credentials and must not be rejected.
PEM_PRIVATE_KEY_RE = re.compile(
rb"-----BEGIN (?:RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----",
re.IGNORECASE,
)
BEARER_RE = re.compile(
rb"authorization\s*:\s*bearer\s+[A-Za-z0-9._~+/=-]{20,}",
re.IGNORECASE,
)
KEY_VALUE_SECRET_RE = re.compile(
rb"(?:password|passwd|secret|api[_-]?key)\s*[:=]\s*[^\s,;]{12,}",
re.IGNORECASE,
)
DOCUMENT_SUFFIXES = {".md", ".rst", ".adoc", ".txt"}
DOCUMENT_PREFIXES = (
"README",
"CHANGELOG",
"RUNBOOK",
"ADR",
"ARCHITECTURE",
"DEPLOYMENT",
"TROUBLESHOOTING",
)
class PackagerError(RuntimeError):
pass
def sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def secret_finding(data: bytes) -> str | None:
if PEM_PRIVATE_KEY_RE.search(data):
return "PEM_PRIVATE_KEY"
if BEARER_RE.search(data):
return "AUTHORIZATION_BEARER"
if KEY_VALUE_SECRET_RE.search(data):
return "KEY_VALUE_SECRET"
return None
def scan_secret(data: bytes, label: str) -> None:
finding = secret_finding(data)
if finding is not None:
raise PackagerError(
f"MODIFIED_FILES_SECRET_SCAN_MATCH:{finding}:{label}"
)
def git_bytes(git_dir: Path, oid: str, path: str) -> bytes:
try:
return subprocess.check_output(
[
"/usr/bin/git",
f"--git-dir={git_dir}",
"show",
f"{oid}:{path}",
],
timeout=30,
)
except subprocess.CalledProcessError as exc:
raise PackagerError(
"MODIFIED_FILES_GIT_SHOW_FAILED:" + path
) from exc
def git_text(git_dir: Path, *args: str) -> str:
try:
return subprocess.check_output(
["/usr/bin/git", f"--git-dir={git_dir}", *args],
text=True,
timeout=30,
)
except subprocess.CalledProcessError as exc:
raise PackagerError(
"MODIFIED_FILES_GIT_COMMAND_FAILED:" + " ".join(args)
) from exc
def validate_lineage(git_dir: Path, base: str, candidate: str) -> None:
if OID_RE.fullmatch(base) is None or OID_RE.fullmatch(candidate) is None:
raise PackagerError("MODIFIED_FILES_OID_INVALID")
for oid in (base, candidate):
try:
subprocess.run(
[
"/usr/bin/git",
f"--git-dir={git_dir}",
"cat-file",
"-e",
f"{oid}^{{commit}}",
],
check=True,
timeout=30,
)
except subprocess.CalledProcessError as exc:
raise PackagerError("MODIFIED_FILES_COMMIT_MISSING:" + oid) from exc
if subprocess.run(
[
"/usr/bin/git",
f"--git-dir={git_dir}",
"merge-base",
"--is-ancestor",
base,
candidate,
],
check=False,
timeout=30,
).returncode != 0:
raise PackagerError("MODIFIED_FILES_LINEAGE_INVALID")
def safe_destination(root: Path, relative: str) -> Path:
path = Path(relative)
if path.is_absolute() or ".." in path.parts:
raise PackagerError("MODIFIED_FILE_PATH_INVALID:" + relative)
destination = root / path
destination.parent.mkdir(parents=True, exist_ok=True)
return destination
def is_document_path(path: str) -> bool:
value = Path(path)
if path.startswith("docs/"):
return True
if value.suffix.lower() in DOCUMENT_SUFFIXES:
return True
upper = value.name.upper()
return any(upper.startswith(prefix) for prefix in DOCUMENT_PREFIXES)
def diff_entries(
git_dir: Path,
base: str,
candidate: str,
before_root: Path,
after_root: Path,
) -> list[dict[str, object]]:
raw = git_text(
git_dir,
"diff",
"--name-status",
"-M",
"-C",
base,
candidate,
)
entries: list[dict[str, object]] = []
for line in raw.splitlines():
if not line.strip():
continue
parts = line.split("\t")
status = parts[0]
code = status[0]
old_path: str | None = None
new_path: str | None = None
if code in {"R", "C"}:
if len(parts) != 3:
raise PackagerError(
"MODIFIED_FILES_DIFF_PARSE_INVALID:" + line
)
old_path, new_path = parts[1], parts[2]
else:
if len(parts) != 2:
raise PackagerError(
"MODIFIED_FILES_DIFF_PARSE_INVALID:" + line
)
new_path = parts[1] if code != "D" else None
old_path = parts[1] if code != "A" else None
entry: dict[str, object] = {"status": status}
if old_path is not None:
entry["previous_path"] = old_path
before = git_bytes(git_dir, base, old_path)
scan_secret(before, "before:" + old_path)
safe_destination(before_root, old_path).write_bytes(before)
entry["before_sha256"] = sha256(before)
entry["before_bytes"] = len(before)
if new_path is not None:
entry["path"] = new_path
after = git_bytes(git_dir, candidate, new_path)
scan_secret(after, "after:" + new_path)
safe_destination(after_root, new_path).write_bytes(after)
entry["after_sha256"] = sha256(after)
entry["after_bytes"] = len(after)
entry["deleted"] = code == "D"
entry["added"] = code == "A"
entry["renamed"] = code == "R"
entry["copied"] = code == "C"
entries.append(entry)
return entries
def execution_entries(
extra_root: Path | None,
execution_root: Path,
) -> list[dict[str, object]]:
if extra_root is None or not extra_root.exists():
return []
if extra_root.is_symlink() or not extra_root.is_dir():
raise PackagerError("EXTRA_FILES_DIR_INVALID")
entries: list[dict[str, object]] = []
for path in sorted(extra_root.rglob("*")):
if path.is_symlink():
raise PackagerError(
"EXTRA_FILE_SYMLINK_REJECTED:" + str(path)
)
if not path.is_file():
continue
relative = path.relative_to(extra_root).as_posix()
data = path.read_bytes()
scan_secret(data, "execution:" + relative)
safe_destination(execution_root, relative).write_bytes(data)
entries.append(
{
"path": relative,
"sha256": sha256(data),
"bytes": len(data),
}
)
return entries
def write_checksums(bundle: Path) -> None:
checks: list[tuple[str, str]] = []
for path in sorted(bundle.rglob("*")):
if path.is_file() and path.name != "checksums.sha256":
checks.append(
(
sha256(path.read_bytes()),
path.relative_to(bundle).as_posix(),
)
)
(bundle / "checksums.sha256").write_text(
"".join(f"{digest} {name}\n" for digest, name in checks),
encoding="utf-8",
)
def verify_checksums(bundle: Path) -> None:
checksum_path = bundle / "checksums.sha256"
for line in checksum_path.read_text(encoding="utf-8").splitlines():
digest, relative = line.split(" ", 1)
path = bundle / relative
if not path.is_file() or sha256(path.read_bytes()) != digest:
raise PackagerError(
"MODIFIED_FILES_CHECKSUM_CLOSURE_INVALID:" + relative
)
def build_bundle(
*,
git_dir: Path,
base: str,
candidate: str,
task_id: str,
extra_dir: Path | None = None,
output_root: Path | None = None,
now: datetime | None = None,
) -> dict[str, object]:
if TASK_RE.fullmatch(task_id) is None:
raise PackagerError("TASK_ID_INVALID")
validate_lineage(git_dir, base, candidate)
timestamp = (now or datetime.now(timezone.utc)).strftime(
"%Y%m%dT%H%M%SZ"
)
safe_task = re.sub(r"[^A-Za-z0-9._-]+", "-", task_id).strip("-")
if output_root is None:
output_root = Path(
tempfile.mkdtemp(
prefix=f"aios-modified-files-bundle-{safe_task}-{timestamp}-",
dir="/var/tmp",
)
)
else:
output_root.mkdir(parents=True, exist_ok=True)
bundle = output_root / "bundle"
before_root = bundle / "repo" / "before"
after_root = bundle / "repo" / "after"
execution_root = bundle / "execution"
before_root.mkdir(parents=True)
after_root.mkdir(parents=True)
execution_root.mkdir(parents=True)
changes = diff_entries(
git_dir,
base,
candidate,
before_root,
after_root,
)
execution_files = execution_entries(extra_dir, execution_root)
document_changes = [
entry
for entry in changes
if is_document_path(
str(entry.get("path") or entry.get("previous_path") or "")
)
]
execution_documents = [
entry
for entry in execution_files
if is_document_path(str(entry["path"]))
]
documents_index = {
"schema": "aios.modified-documents.index.v1",
"task_id": task_id,
"base_oid": base,
"candidate_oid": candidate,
"repository_documents": document_changes,
"execution_documents": execution_documents,
"repository_document_count": len(document_changes),
"execution_document_count": len(execution_documents),
}
(bundle / "documents-index.json").write_text(
json.dumps(documents_index, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
manifest = {
"schema": "aios.all-modified-files.bundle.v1",
"task_id": task_id,
"base_oid": base,
"candidate_oid": candidate,
"generated_at": (
now or datetime.now(timezone.utc)
).isoformat(),
"changed_file_count": len(changes),
"execution_file_count": len(execution_files),
"repository_document_count": len(document_changes),
"execution_document_count": len(execution_documents),
"changes": changes,
"execution_files": execution_files,
}
(bundle / "all-modified-files-manifest.json").write_text(
json.dumps(manifest, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
write_checksums(bundle)
verify_checksums(bundle)
zip_name = (
f"AIOS_{task_id}_ALL_MODIFIED_FILES_{candidate[:12]}.zip"
.replace("/", "-")
)
zip_path = output_root / zip_name
with zipfile.ZipFile(
zip_path,
"w",
compression=zipfile.ZIP_DEFLATED,
) as archive:
for path in sorted(bundle.rglob("*")):
if path.is_symlink():
raise PackagerError(
"BUNDLE_SYMLINK_REJECTED:" + str(path)
)
if path.is_file():
archive.write(
path,
arcname=path.relative_to(bundle).as_posix(),
)
zip_sha = sha256(zip_path.read_bytes())
summary = {
"schema": "aios.all-modified-files.bundle-summary.v1",
"task_id": task_id,
"base_oid": base,
"candidate_oid": candidate,
"bundle_path": str(zip_path),
"bundle_sha256": zip_sha,
"changed_file_count": len(changes),
"execution_file_count": len(execution_files),
"repository_document_count": len(document_changes),
"execution_document_count": len(execution_documents),
}
(output_root / "ALL_MODIFIED_FILES_BUNDLE.json").write_text(
json.dumps(summary, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
return summary
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("base")
parser.add_argument("candidate")
parser.add_argument("task_id")
parser.add_argument("extra_dir", nargs="?")
parser.add_argument(
"--authority-git-dir",
default=os.environ.get(
"AIOS_AUTHORITY_GIT_DIR",
"/data/aios-gitops/repo.git",
),
)
parser.add_argument("--output-root")
args = parser.parse_args()
try:
summary = build_bundle(
git_dir=Path(args.authority_git_dir),
base=args.base,
candidate=args.candidate,
task_id=args.task_id,
extra_dir=Path(args.extra_dir) if args.extra_dir else None,
output_root=(
Path(args.output_root) if args.output_root else None
),
)
except PackagerError as exc:
print("FATAL:" + str(exc))
return 20
print("ALL_MODIFIED_FILES_BUNDLE=PASS")
print("ALL_MODIFIED_FILES_BUNDLE_PATH=" + str(summary["bundle_path"]))
print("ALL_MODIFIED_FILES_BUNDLE_SHA256=" + str(summary["bundle_sha256"]))
print("CHANGED_FILE_COUNT=" + str(summary["changed_file_count"]))
print("EXECUTION_FILE_COUNT=" + str(summary["execution_file_count"]))
print(
"REPOSITORY_DOCUMENT_COUNT="
+ str(summary["repository_document_count"])
)
print(
"EXECUTION_DOCUMENT_COUNT="
+ str(summary["execution_document_count"])
)
print(
"ALL_MODIFIED_FILES_EVIDENCE="
+ str(Path(str(summary["bundle_path"])).parent)
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,288 @@
"""Unified autonomous control-plane primitives.
This is deliberately a library, not another daemon. Task API, workers and host
recovery code consume the same Admission -> Invariant -> Repair semantics.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass
from enum import Enum
import re
from typing import Callable, Iterable, Mapping, Sequence
from aios_capability_contract import (
capability_contract,
has_capabilities,
normalize_required_capabilities,
normalize_worker_capabilities,
registered_capability,
validate_capability_contract_registry,
)
class FailureClass(str, Enum):
CAPABILITY_CONTRACT_ERROR = "CAPABILITY_CONTRACT_ERROR"
CREDENTIAL_PROJECTION_ERROR = "CREDENTIAL_PROJECTION_ERROR"
DEPENDENCY_UNAVAILABLE = "DEPENDENCY_UNAVAILABLE"
POLICY_DENIED = "POLICY_DENIED"
RESOURCE_UNAVAILABLE = "RESOURCE_UNAVAILABLE"
WORKER_CRASH = "WORKER_CRASH"
LEASE_LOST = "LEASE_LOST"
BUSINESS_PERMANENT = "BUSINESS_PERMANENT"
UNKNOWN = "UNKNOWN"
class RepairLevel(int, Enum):
AUTO_LOCAL = 0
AUTO_REPAIR_TASK = 1
OWNER_APPROVAL = 2
MANUAL = 3
@dataclass(frozen=True)
class InvariantResult:
name: str
passed: bool
detail: str = ""
def as_dict(self) -> dict[str, object]:
return asdict(self)
@dataclass(frozen=True)
class AdmissionDecision:
admitted: bool
state: str
invariants: tuple[InvariantResult, ...]
failure_class: FailureClass | None = None
repair_level: RepairLevel | None = None
repair_action: str | None = None
def as_dict(self) -> dict[str, object]:
return {
"schema": "aios.runtime-admission.decision.v1",
"admitted": self.admitted,
"state": self.state,
"invariants": [item.as_dict() for item in self.invariants],
"failure_class": None if self.failure_class is None else self.failure_class.value,
"repair_level": None if self.repair_level is None else int(self.repair_level),
"repair_action": self.repair_action,
}
@dataclass(frozen=True)
class RepairDecision:
failure_class: FailureClass
level: RepairLevel
action: str
consumes_business_attempt: bool
def as_dict(self) -> dict[str, object]:
return {
"failure_class": self.failure_class.value,
"level": int(self.level),
"action": self.action,
"consumes_business_attempt": self.consumes_business_attempt,
}
def classify_failure(error: object, *, phase: str = "") -> FailureClass:
text = (str(error) + " " + str(phase)).lower()
if any(token in text for token in (
"credential", "permission denied", "token registry", "secret", "unauthorized"
)):
return FailureClass.CREDENTIAL_PROJECTION_ERROR
if any(token in text for token in (
"transport", "unavailable", "timeout", "connection", "bad gateway", "http_5"
)):
return FailureClass.DEPENDENCY_UNAVAILABLE
if any(token in text for token in (
"capability", "eligible", "matcher", "contract"
)):
return FailureClass.CAPABILITY_CONTRACT_ERROR
if any(token in text for token in (
"policy", "forbidden", "approval required", "not authorized"
)):
return FailureClass.POLICY_DENIED
if any(token in text for token in ("lease lost", "lease_expired", "heartbeat_failed")):
return FailureClass.LEASE_LOST
if any(token in text for token in ("out of memory", "disk full", "resource")):
return FailureClass.RESOURCE_UNAVAILABLE
if any(token in text for token in ("worker crash", "process exited", "signal")):
return FailureClass.WORKER_CRASH
if any(token in text for token in ("invalid release", "business permanent")):
return FailureClass.BUSINESS_PERMANENT
return FailureClass.UNKNOWN
def repair_for_failure(failure: FailureClass) -> RepairDecision:
table = {
FailureClass.CAPABILITY_CONTRACT_ERROR: RepairDecision(
failure, RepairLevel.AUTO_REPAIR_TASK,
"create_same_lineage_capability_contract_repair", False,
),
FailureClass.CREDENTIAL_PROJECTION_ERROR: RepairDecision(
failure, RepairLevel.AUTO_REPAIR_TASK,
"restore_service_private_credential_projection", False,
),
FailureClass.DEPENDENCY_UNAVAILABLE: RepairDecision(
failure, RepairLevel.AUTO_LOCAL,
"retry_dependency_health_and_resume", False,
),
FailureClass.RESOURCE_UNAVAILABLE: RepairDecision(
failure, RepairLevel.AUTO_LOCAL,
"heal_runtime_resource_and_resume", False,
),
FailureClass.WORKER_CRASH: RepairDecision(
failure, RepairLevel.AUTO_LOCAL,
"restart_worker_and_resume_same_lineage", False,
),
FailureClass.LEASE_LOST: RepairDecision(
failure, RepairLevel.AUTO_LOCAL,
"recover_lease_and_resume_same_lineage", False,
),
FailureClass.POLICY_DENIED: RepairDecision(
failure, RepairLevel.OWNER_APPROVAL,
"request_bound_owner_authorization", False,
),
FailureClass.BUSINESS_PERMANENT: RepairDecision(
failure, RepairLevel.MANUAL,
"terminal_business_failure", True,
),
FailureClass.UNKNOWN: RepairDecision(
failure, RepairLevel.AUTO_REPAIR_TASK,
"create_diagnostic_repair_task", False,
),
}
return table[failure]
def _probe(
name: str,
probes: Mapping[str, Callable[[], bool]],
) -> InvariantResult:
fn = probes.get(name)
if fn is None:
return InvariantResult(name, False, "PROBE_MISSING")
try:
passed = bool(fn())
except Exception as exc:
return InvariantResult(name, False, "PROBE_ERROR:" + type(exc).__name__)
return InvariantResult(name, passed, "PASS" if passed else "NOT_READY")
def runtime_admission(
*,
worker_capabilities: Iterable[object],
required_capabilities: Iterable[object],
requested_operations: Sequence[str] = (),
probes: Mapping[str, Callable[[], bool]] | None = None,
) -> AdmissionDecision:
"""Admission is evaluated before claim; failures do not consume attempts."""
worker_values = tuple(worker_capabilities)
required_values = tuple(required_capabilities)
probe_map = probes or {}
invariants: list[InvariantResult] = []
registry = validate_capability_contract_registry()
invariants.append(InvariantResult(
"capability_contract_registry",
bool(registry.get("valid")),
"PASS" if registry.get("valid") else "REGISTRY_INVALID",
))
task_known = all(registered_capability(item) for item in required_values)
worker_known = all(registered_capability(item, worker=True) for item in worker_values)
invariants.extend((
InvariantResult("task_capabilities_registered", task_known, "PASS" if task_known else "UNKNOWN_TASK_CAPABILITY"),
InvariantResult("worker_capabilities_registered", worker_known, "PASS" if worker_known else "UNKNOWN_WORKER_CAPABILITY"),
))
capability_match = (
task_known
and worker_known
and has_capabilities(worker_values, required_values)
)
invariants.append(InvariantResult(
"capability_match",
capability_match,
"PASS" if capability_match else "WORKER_NOT_ELIGIBLE",
))
normalized = normalize_required_capabilities(required_values)
contracts = [
capability_contract(item)
for item in normalized
if not str(item).startswith("exact:")
]
contracts = [item for item in contracts if item is not None]
forbidden: set[str] = set()
dependencies: set[str] = set()
health: set[str] = set()
for contract in contracts:
forbidden.update(contract.forbidden)
dependencies.update(contract.dependencies)
health.update(contract.health_requirements)
requested = {str(item).strip().lower() for item in requested_operations if str(item).strip()}
forbidden_hit = sorted(requested & {item.lower() for item in forbidden})
invariants.append(InvariantResult(
"forbidden_operations_absent",
not forbidden_hit,
"PASS" if not forbidden_hit else "FORBIDDEN:" + ",".join(forbidden_hit),
))
for name in sorted(dependencies | health):
invariants.append(_probe(name, probe_map))
failed = [item for item in invariants if not item.passed]
if not failed:
return AdmissionDecision(
True,
"EXECUTION_READY",
tuple(invariants),
)
details = " ".join(item.name + ":" + item.detail for item in failed)
failure = classify_failure(details, phase="admission")
if any(item.name == "forbidden_operations_absent" and not item.passed for item in failed):
failure = FailureClass.POLICY_DENIED
elif any("credential" in item.name for item in failed):
failure = FailureClass.CREDENTIAL_PROJECTION_ERROR
elif any(item.name in {"capability_contract_registry", "task_capabilities_registered", "worker_capabilities_registered", "capability_match"} for item in failed):
failure = FailureClass.CAPABILITY_CONTRACT_ERROR
elif any(item.name in dependencies or item.name in health for item in failed):
failure = FailureClass.DEPENDENCY_UNAVAILABLE
repair = repair_for_failure(failure)
return AdmissionDecision(
False,
"EXECUTION_READY_FAILED",
tuple(invariants),
failure,
repair.level,
repair.action,
)
def system_invariant_snapshot(
*,
worker_capabilities: Iterable[object],
required_capabilities: Iterable[object],
requested_operations: Sequence[str] = (),
probes: Mapping[str, Callable[[], bool]] | None = None,
) -> dict[str, object]:
decision = runtime_admission(
worker_capabilities=worker_capabilities,
required_capabilities=required_capabilities,
requested_operations=requested_operations,
probes=probes,
)
return {
"schema": "aios.system-invariants.v1",
"status": "HEALTHY" if decision.admitted else "DEGRADED",
"execution_ready": decision.admitted,
"admission": decision.as_dict(),
}
+173 -2
View File
@@ -3,6 +3,8 @@
from __future__ import annotations
from collections.abc import Iterable
from dataclasses import asdict, dataclass
from typing import Any
_TASK_ALIASES: dict[str, frozenset[str]] = {
@@ -37,6 +39,7 @@ _TASK_ALIASES: dict[str, frozenset[str]] = {
"gpu.train": frozenset({"legacy.gpu.train"}),
"agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
"agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
"release:candidate": frozenset({"release:candidate"}),
}
_WORKER_ALIASES: dict[str, frozenset[str]] = {
@@ -71,8 +74,172 @@ _WORKER_ALIASES: dict[str, frozenset[str]] = {
"gpu.train": frozenset({"legacy.gpu.train"}),
"agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
"agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
"release:candidate": frozenset({"release:candidate"}),
}
@dataclass(frozen=True)
class CapabilityContract:
"""Versioned runtime contract attached to one canonical semantic capability."""
name: str
min_version: int = 1
max_version: int = 1
permissions: tuple[str, ...] = ()
forbidden: tuple[str, ...] = ()
dependencies: tuple[str, ...] = ()
health_requirements: tuple[str, ...] = ()
def as_dict(self) -> dict[str, Any]:
return asdict(self)
# This registry is the semantic source of truth. Alias tables below are retained
# as compatibility adapters; startup/tests validate every alias target against
# this registry so a name cannot be accepted in one layer and be unknown in
# another.
_CAPABILITY_CONTRACTS: dict[str, CapabilityContract] = {
"repo.read": CapabilityContract("repo.read", permissions=("repository_read",)),
"test.execute": CapabilityContract("test.execute", permissions=("test_execution",)),
"governed.git.write.branch": CapabilityContract(
"governed.git.write.branch",
permissions=("governed_branch_write",),
dependencies=("writer_lease", "governed_gitops"),
),
"governed.git.write.file": CapabilityContract(
"governed.git.write.file",
permissions=("governed_file_write",),
dependencies=("writer_lease", "governed_gitops"),
),
"governed.git.write": CapabilityContract(
"governed.git.write",
permissions=("governed_git_write",),
dependencies=("writer_lease", "governed_gitops"),
),
"runtime.shell": CapabilityContract("runtime.shell", permissions=("runtime_shell",)),
"runtime.git": CapabilityContract("runtime.git", permissions=("runtime_git",)),
"runtime.python": CapabilityContract("runtime.python", permissions=("runtime_python",)),
"legacy.code.read": CapabilityContract("legacy.code.read"),
"legacy.code.review": CapabilityContract("legacy.code.review"),
"legacy.code.write": CapabilityContract("legacy.code.write"),
"legacy.code.write.patch": CapabilityContract("legacy.code.write.patch"),
"legacy.test.python": CapabilityContract("legacy.test.python"),
"legacy.test.shell.allowlisted": CapabilityContract("legacy.test.shell.allowlisted"),
"legacy.git.commit": CapabilityContract("legacy.git.commit"),
"legacy.git.diff.read": CapabilityContract("legacy.git.diff.read"),
"legacy.git.force_push": CapabilityContract(
"legacy.git.force_push",
forbidden=("protected_ref_bypass",),
),
"legacy.deploy.production": CapabilityContract(
"legacy.deploy.production",
permissions=("production_deploy",),
dependencies=("owner_approval", "deployment_controller"),
health_requirements=("approval_envelope_verified", "runtime_acceptance_available"),
),
"legacy.secret.transform": CapabilityContract("legacy.secret.transform"),
"legacy.secure_file_transformer": CapabilityContract("legacy.secure_file_transformer"),
"legacy.runtime.shadow": CapabilityContract("legacy.runtime.shadow"),
"legacy.review.security": CapabilityContract("legacy.review.security"),
"legacy.review.api-contract": CapabilityContract("legacy.review.api-contract"),
"legacy.gpu.train": CapabilityContract("legacy.gpu.train"),
"agent.runtime.hermes": CapabilityContract(
"agent.runtime.hermes",
dependencies=("hermes_runtime",),
health_requirements=("worker_liveness",),
),
"agent.runtime.openclaw": CapabilityContract(
"agent.runtime.openclaw",
dependencies=("openclaw_runtime",),
health_requirements=("worker_liveness",),
),
"release:candidate": CapabilityContract(
"release:candidate",
min_version=1,
max_version=1,
permissions=("prepare_release",),
forbidden=(
"production_deploy",
"submit_owner_approval",
"tag_publish",
"baseline_registry_finalize",
"runtime_switch",
),
dependencies=(
"credential_projection",
"trusted_builder",
),
health_requirements=(
"authenticated_probe",
"release_api_health",
"worker_liveness",
),
),
}
def capability_contract(name: object) -> CapabilityContract | None:
"""Return the contract for one canonical or registered capability name."""
text = _clean(name)
direct = _CAPABILITY_CONTRACTS.get(text)
if direct is not None:
return direct
mapped = _TASK_ALIASES.get(text) or _WORKER_ALIASES.get(text)
if mapped is None or len(mapped) != 1:
return None
return _CAPABILITY_CONTRACTS.get(next(iter(mapped)))
def registered_capability(name: object, *, worker: bool = False) -> bool:
"""Fail-closed validation used by registration/task admission."""
text = _clean(name)
if "*" in text:
return False
aliases = _WORKER_ALIASES if worker else _TASK_ALIASES
mapped = aliases.get(text)
if mapped is None:
return text in _CAPABILITY_CONTRACTS
return all(item in _CAPABILITY_CONTRACTS for item in mapped)
def validate_capability_contract_registry() -> dict[str, object]:
"""Prove alias/contract closure and contract safety invariants."""
alias_targets: set[str] = set()
for mapping in (_TASK_ALIASES, _WORKER_ALIASES):
for values in mapping.values():
alias_targets.update(values)
missing = sorted(alias_targets - set(_CAPABILITY_CONTRACTS))
invalid_versions = sorted(
name
for name, contract in _CAPABILITY_CONTRACTS.items()
if contract.min_version < 1 or contract.max_version < contract.min_version
)
permission_forbidden_overlap = sorted(
name
for name, contract in _CAPABILITY_CONTRACTS.items()
if set(contract.permissions) & set(contract.forbidden)
)
return {
"schema": "aios.capability-contract-registry.v1",
"contract_count": len(_CAPABILITY_CONTRACTS),
"missing_alias_targets": missing,
"invalid_versions": invalid_versions,
"permission_forbidden_overlap": permission_forbidden_overlap,
"valid": not missing and not invalid_versions and not permission_forbidden_overlap,
}
def capability_contract_snapshot() -> dict[str, dict[str, Any]]:
return {
name: contract.as_dict()
for name, contract in sorted(_CAPABILITY_CONTRACTS.items())
}
_UNKNOWN_PREFIX = "exact:"
@@ -98,7 +265,11 @@ def normalize_required_capabilities(values: Iterable[object]) -> frozenset[str]:
if "*" in text:
result.add(_UNKNOWN_PREFIX + text)
else:
result.update(_TASK_ALIASES.get(text, frozenset({_UNKNOWN_PREFIX + text})))
mapped = _TASK_ALIASES.get(text)
if mapped is None or any(item not in _CAPABILITY_CONTRACTS for item in mapped):
result.add(_UNKNOWN_PREFIX + text)
else:
result.update(mapped)
return frozenset(result)
@@ -111,7 +282,7 @@ def normalize_worker_capabilities(values: Iterable[object]) -> frozenset[str]:
if "*" in text:
continue
mapped = _WORKER_ALIASES.get(text)
if mapped is not None:
if mapped is not None and all(item in _CAPABILITY_CONTRACTS for item in mapped):
result.update(mapped)
if {
"governed.git.write.branch",
File diff suppressed because it is too large Load Diff
+456
View File
@@ -0,0 +1,456 @@
#!/usr/bin/env python3
"""Disabled-by-default TLS edge for the read-only GPT Online Task ingress.
The internal AIOS ingress remains loopback-only. This service is a separate edge
that may bind non-loopback only when the owner explicitly approves it. It
forwards only the tiny read-only ingress API and never proxies the canonical
Bridge or Task API.
"""
from __future__ import annotations
from collections import defaultdict, deque
from dataclasses import dataclass
import hashlib
import hmac
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from pathlib import Path
import re
import secrets
import ssl
import sys
import threading
import time
import urllib.error
import urllib.request
from typing import Any, Mapping
from urllib.parse import urlsplit
UPSTREAM = "http://127.0.0.1:18945"
MAX_BODY = 64 * 1024
MAX_RESPONSE = 256 * 1024
MAX_CONCURRENT_REQUESTS = 32
REQUEST_TIMEOUT_SECONDS = 20
RATE_WINDOW_SECONDS = 60
RATE_LIMIT_REQUESTS = 60
RATE_BURST_REQUESTS = 15
RATE_MAX_CLIENTS = 4096
_AUDIT_SALT = secrets.token_bytes(32)
PUBLIC_APPROVAL_ENV = "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED"
PUBLIC_DOMAIN_ENV = "AIOS_GPT_EDGE_PUBLIC_DOMAIN"
TASK_PATH_RE = re.compile(r"^/v1/read-only-tasks(?:/[A-Za-z0-9][A-Za-z0-9._:-]{7,127})?$")
def _client_pseudonym(value: str) -> str:
return hashlib.sha256(_AUDIT_SALT + value.encode("utf-8", "replace")).hexdigest()[:16]
def audit(event: str, **fields: object) -> None:
safe = {
"event": event,
"ts_unix_ms": int(time.time() * 1000),
}
for key, value in fields.items():
if key in {"authorization", "body", "token", "cookie", "client_ip"}:
continue
safe[key] = value
print(
json.dumps(safe, sort_keys=True, separators=(",", ":")),
file=sys.stderr,
flush=True,
)
class EdgeError(RuntimeError):
def __init__(self, code: str, status: int = 400) -> None:
super().__init__(code)
self.code = code
self.status = status
def canonical(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def public_bind_allowed(host: str, approval: str) -> str:
value = host.strip()
if value in {"127.0.0.1", "::1", "localhost"}:
return value
if approval != "1":
raise RuntimeError("GPT_EDGE_PUBLIC_BIND_OWNER_APPROVAL_REQUIRED")
if value not in {"0.0.0.0", "::"}:
raise RuntimeError("GPT_EDGE_BIND_ADDRESS_INVALID")
return value
def validated_public_domain(value: str) -> str:
domain = value.strip().lower()
if not domain:
return ""
if len(domain) > 253 or re.fullmatch(
r"(?=.{1,253}\Z)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?",
domain,
) is None:
raise RuntimeError("GPT_EDGE_PUBLIC_DOMAIN_INVALID")
return domain
def validated_edge_identity(host: str, approval: str, domain: str) -> tuple[str, str]:
bind_host = public_bind_allowed(host, approval)
public_domain = validated_public_domain(domain)
if bind_host not in {"127.0.0.1", "::1", "localhost"} and not public_domain:
raise RuntimeError("GPT_EDGE_PUBLIC_DOMAIN_REQUIRED")
return bind_host, public_domain
def validate_tls_file(path: Path, *, key: bool) -> Path:
if path.is_symlink() or not path.is_file():
raise RuntimeError("GPT_EDGE_TLS_FILE_INVALID")
size = path.stat().st_size
if size < 16 or size > 256 * 1024:
raise RuntimeError("GPT_EDGE_TLS_FILE_INVALID")
if key and (path.stat().st_mode & 0o077):
raise RuntimeError("GPT_EDGE_TLS_KEY_MODE_INVALID")
return path
def build_tls_context(cert_path: Path, key_path: Path) -> ssl.SSLContext:
cert = validate_tls_file(cert_path, key=False)
key = validate_tls_file(key_path, key=True)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.options |= ssl.OP_NO_COMPRESSION
context.load_cert_chain(certfile=str(cert), keyfile=str(key))
return context
def allowed_route(method: str, raw_path: str) -> str:
parsed = urlsplit(raw_path)
if parsed.query or parsed.fragment:
raise EdgeError("QUERY_OR_FRAGMENT_DENY", 400)
path = parsed.path
if method == "GET" and path in {"/health", "/openapi.json"}:
return path
if method == "POST" and path == "/v1/read-only-tasks":
return path
if method == "GET" and TASK_PATH_RE.fullmatch(path):
return path
raise EdgeError("ROUTE_DENY", 404)
def route_label(method: str, path: str) -> str:
if method == "GET" and path == "/health":
return "health"
if method == "GET" and path == "/openapi.json":
return "openapi"
if method == "POST" and path == "/v1/read-only-tasks":
return "task_create"
if method == "GET" and path.startswith("/v1/read-only-tasks/"):
return "task_status"
return "unknown"
def forwarded_headers(headers: Mapping[str, str], *, body: bool) -> dict[str, str]:
result = {"Accept": "application/json"}
authorization = headers.get("Authorization")
if not isinstance(authorization, str) or not authorization.startswith("Bearer "):
raise EdgeError("AUTHORIZATION_REQUIRED", 401)
token = authorization[7:]
if len(token) < 32 or len(token) > 4096 or any(ch.isspace() for ch in token):
raise EdgeError("AUTHORIZATION_INVALID", 401)
result["Authorization"] = authorization
if body:
content_type = headers.get("Content-Type", "")
if content_type.split(";", 1)[0].strip().lower() != "application/json":
raise EdgeError("CONTENT_TYPE_INVALID", 415)
result["Content-Type"] = "application/json"
return result
class RateLimiter:
def __init__(
self,
*,
window_seconds: int = RATE_WINDOW_SECONDS,
limit: int = RATE_LIMIT_REQUESTS,
burst: int = RATE_BURST_REQUESTS,
max_clients: int = RATE_MAX_CLIENTS,
now_fn=time.monotonic,
) -> None:
self.window_seconds = window_seconds
self.limit = limit
self.burst = burst
self.max_clients = max_clients
self.now_fn = now_fn
self._lock = threading.Lock()
self._events: dict[str, deque[float]] = defaultdict(deque)
def _prune_stale(self, cutoff: float) -> None:
stale = [
key
for key, events in self._events.items()
if not events or events[-1] <= cutoff
]
for key in stale:
self._events.pop(key, None)
def allowed(self, key: str) -> bool:
now = self.now_fn()
cutoff = now - self.window_seconds
with self._lock:
if key not in self._events and len(self._events) >= self.max_clients:
self._prune_stale(cutoff)
if len(self._events) >= self.max_clients:
return False
events = self._events[key]
while events and events[0] <= cutoff:
events.popleft()
if len(events) >= self.limit + self.burst:
return False
events.append(now)
return True
@dataclass(frozen=True)
class Proxy:
upstream: str = UPSTREAM
timeout: int = REQUEST_TIMEOUT_SECONDS
def request(
self,
method: str,
path: str,
headers: Mapping[str, str],
body: bytes | None,
) -> tuple[int, bytes, str]:
if self.upstream != UPSTREAM:
raise EdgeError("UPSTREAM_DRIFT_DENY", 503)
req = urllib.request.Request(
self.upstream + path,
method=method,
data=body,
headers=dict(headers),
)
try:
with urllib.request.urlopen(req, timeout=self.timeout) as response:
raw = response.read(MAX_RESPONSE + 1)
status = response.status
content_type = response.headers.get("Content-Type", "application/json")
except urllib.error.HTTPError as exc:
raw = exc.read(MAX_RESPONSE + 1)
status = exc.code
content_type = exc.headers.get("Content-Type", "application/json")
except Exception as exc:
raise EdgeError("UPSTREAM_UNAVAILABLE", 502) from exc
if len(raw) > MAX_RESPONSE:
raise EdgeError("UPSTREAM_RESPONSE_TOO_LARGE", 502)
if content_type.split(";", 1)[0].strip().lower() != "application/json":
raise EdgeError("UPSTREAM_CONTENT_TYPE_INVALID", 502)
return status, raw, "application/json"
class EdgeServer(ThreadingHTTPServer):
daemon_threads = True
request_queue_size = 64
allow_reuse_address = True
class Handler(BaseHTTPRequestHandler):
proxy = Proxy()
limiter = RateLimiter()
concurrency = threading.BoundedSemaphore(MAX_CONCURRENT_REQUESTS)
expected_domain = ""
server_version = "AIOSGPTIngressEdge/1.0"
sys_version = ""
def log_message(self, *_args: Any) -> None:
return
def respond_json(self, value: Mapping[str, Any], status: int) -> None:
raw = canonical(value)
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Cache-Control", "no-store")
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("Content-Security-Policy", "default-src 'none'")
self.send_header("Strict-Transport-Security", "max-age=31536000")
self.send_header("Content-Length", str(len(raw)))
self.end_headers()
self.wfile.write(raw)
def send_proxy_response(self, status: int, raw: bytes, content_type: str) -> None:
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Cache-Control", "no-store")
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("Content-Security-Policy", "default-src 'none'")
self.send_header("Strict-Transport-Security", "max-age=31536000")
self.send_header("Content-Length", str(len(raw)))
self.end_headers()
self.wfile.write(raw)
def client_key(self) -> str:
return str(self.client_address[0])
def validate_host(self) -> None:
if not self.expected_domain:
return
host = self.headers.get("Host", "").split(":", 1)[0].lower()
if not hmac.compare_digest(host, self.expected_domain):
raise EdgeError("HOST_HEADER_DENY", 421)
def body(self) -> bytes:
try:
length = int(self.headers.get("Content-Length", "-1"))
except ValueError as exc:
raise EdgeError("CONTENT_LENGTH_INVALID") from exc
if length < 1 or length > MAX_BODY:
raise EdgeError("REQUEST_SIZE_INVALID", 413)
raw = self.rfile.read(length)
if len(raw) != length:
raise EdgeError("REQUEST_BODY_TRUNCATED", 400)
# Validate JSON at the edge so malformed bodies do not consume upstream work.
try:
value = json.loads(raw.decode("utf-8"))
except Exception as exc:
raise EdgeError("REQUEST_JSON_INVALID", 400) from exc
if not isinstance(value, Mapping):
raise EdgeError("REQUEST_JSON_OBJECT_REQUIRED", 400)
return canonical(value)
def handle_proxy(self, method: str) -> None:
client_id = _client_pseudonym(self.client_key())
started = time.monotonic()
acquired = self.concurrency.acquire(blocking=False)
if not acquired:
audit("EDGE_BUSY", client_id=client_id, method=method)
self.respond_json({"error": "EDGE_BUSY"}, 503)
return
label = "unknown"
try:
self.validate_host()
path = allowed_route(method, self.path)
label = route_label(method, path)
if path == "/health" and method == "GET":
self.respond_json(
{
"status": "ok",
"service": "aios-gpt-https-edge",
"upstream": "loopback-readonly-ingress",
},
200,
)
audit(
"EDGE_REQUEST",
client_id=client_id,
method=method,
route=label,
status=200,
duration_ms=int((time.monotonic() - started) * 1000),
)
return
if not self.limiter.allowed(self.client_key()):
audit(
"EDGE_RATE_LIMIT",
client_id=client_id,
method=method,
route=label,
)
self.respond_json({"error": "RATE_LIMITED"}, 429)
return
has_body = method == "POST"
headers = forwarded_headers(self.headers, body=has_body)
body = self.body() if has_body else None
status, raw, content_type = self.proxy.request(
method,
path,
headers,
body,
)
self.send_proxy_response(status, raw, content_type)
audit(
"EDGE_REQUEST",
client_id=client_id,
method=method,
route=label,
status=status,
duration_ms=int((time.monotonic() - started) * 1000),
)
except EdgeError as exc:
audit(
"EDGE_REJECT",
client_id=client_id,
method=method,
route=label,
status=exc.status,
error=exc.code,
)
self.respond_json({"error": exc.code}, exc.status)
finally:
self.concurrency.release()
def do_GET(self) -> None:
self.handle_proxy("GET")
def do_POST(self) -> None:
self.handle_proxy("POST")
def do_PUT(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def do_PATCH(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def do_DELETE(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def do_CONNECT(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def do_OPTIONS(self) -> None:
self.respond_json({"error": "METHOD_DENY"}, 405)
def serve() -> None:
host, public_domain = validated_edge_identity(
os.environ.get("AIOS_GPT_EDGE_HOST", "127.0.0.1"),
os.environ.get(PUBLIC_APPROVAL_ENV, "0"),
os.environ.get(PUBLIC_DOMAIN_ENV, ""),
)
port = int(os.environ.get("AIOS_GPT_EDGE_PORT", "18443"))
if not 1024 <= port <= 65535:
raise RuntimeError("GPT_EDGE_PORT_INVALID")
Handler.expected_domain = public_domain
credentials = Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
cert_path = Path(
os.environ.get(
"AIOS_GPT_EDGE_TLS_CERT_FILE",
str(credentials / "tls_cert"),
)
)
key_path = Path(
os.environ.get(
"AIOS_GPT_EDGE_TLS_KEY_FILE",
str(credentials / "tls_key"),
)
)
context = build_tls_context(cert_path, key_path)
server = EdgeServer((host, port), Handler)
server.socket = context.wrap_socket(server.socket, server_side=True)
server.serve_forever()
if __name__ == "__main__":
serve()
@@ -0,0 +1,561 @@
#!/usr/bin/env python3
"""Owner-approved activation/rollback controller for the AIOS GPT HTTPS edge.
This controller never creates approval, TLS, or ingress credentials. It consumes
owner-managed inputs, applies one exact public-bind overlay, verifies the live
edge, and provides an idempotent rollback path.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import socket
import ssl
import stat
import subprocess
import tempfile
import time
import urllib.request
from typing import Any, Mapping
APPROVAL_SCHEMA = "aios.gpt-https-edge.public-approval.v1"
STATE_SCHEMA = "aios.gpt-https-edge.activation-state.v1"
OID_RE = re.compile(r"^[0-9a-f]{40}$")
SHA_RE = re.compile(r"^[0-9a-f]{64}$")
APPROVAL_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
DOMAIN_RE = re.compile(
r"(?=.{1,253}\Z)"
r"(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+"
r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?"
)
EDGE_UNIT = "aios-gpt-https-edge.service"
INGRESS_UNIT = "aios-gpt-online-task-ingress.service"
PUBLIC_HOST = "0.0.0.0"
PUBLIC_PORT = 18443
INGRESS_HEALTH = "http://127.0.0.1:18945/health"
DROPIN_NAME = "zzzzzzzzzz-owner-approved-public.conf"
class ActivationError(RuntimeError):
pass
def canonical(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def sha256_bytes(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def sha256_file(path: Path) -> str:
return sha256_bytes(path.read_bytes())
def parse_time(value: object, *, field: str) -> datetime:
if not isinstance(value, str) or not value.strip():
raise ActivationError("APPROVAL_" + field.upper() + "_INVALID")
text = value.strip()
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(text)
except ValueError as exc:
raise ActivationError("APPROVAL_" + field.upper() + "_INVALID") from exc
if parsed.tzinfo is None:
raise ActivationError("APPROVAL_" + field.upper() + "_INVALID")
return parsed.astimezone(timezone.utc)
def validate_domain(value: object) -> str:
if not isinstance(value, str):
raise ActivationError("APPROVAL_DOMAIN_INVALID")
domain = value.strip().lower()
if DOMAIN_RE.fullmatch(domain) is None:
raise ActivationError("APPROVAL_DOMAIN_INVALID")
return domain
def validate_root_file(
path: Path,
*,
exact_mode: int | None = None,
max_size: int = 256 * 1024,
) -> Path:
if path.is_symlink() or not path.is_file():
raise ActivationError("OWNER_FILE_INVALID")
info = path.stat()
if info.st_uid != 0 or info.st_gid != 0:
raise ActivationError("OWNER_FILE_OWNERSHIP_INVALID")
mode = stat.S_IMODE(info.st_mode)
if exact_mode is not None and mode != exact_mode:
raise ActivationError("OWNER_FILE_MODE_INVALID")
if info.st_size < 16 or info.st_size > max_size:
raise ActivationError("OWNER_FILE_SIZE_INVALID")
return path
def certificate_dns_names(cert_path: Path) -> tuple[str, ...]:
try:
decoded = ssl._ssl._test_decode_cert(str(cert_path))
except Exception as exc:
raise ActivationError("TLS_CERT_DECODE_INVALID") from exc
names = [
value
for kind, value in decoded.get("subjectAltName", ())
if kind == "DNS"
]
if not names:
for rdn in decoded.get("subject", ()):
for key, value in rdn:
if key == "commonName":
names.append(value)
return tuple(names)
def cert_matches_domain(cert_path: Path, domain: str) -> bool:
names = certificate_dns_names(cert_path)
try:
return bool(names) and any(ssl._dnsname_match(name, domain) for name in names)
except ssl.CertificateError:
return False
def validate_tls_pair(
cert_path: Path,
key_path: Path,
domain: str,
*,
now_fn=time.time,
) -> dict[str, Any]:
validate_root_file(cert_path)
validate_root_file(key_path, exact_mode=0o600)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.options |= ssl.OP_NO_COMPRESSION
try:
context.load_cert_chain(str(cert_path), str(key_path))
except Exception as exc:
raise ActivationError("TLS_CERT_KEY_PAIR_INVALID") from exc
if not cert_matches_domain(cert_path, domain):
raise ActivationError("TLS_CERT_DOMAIN_MISMATCH")
decoded = ssl._ssl._test_decode_cert(str(cert_path))
not_after = decoded.get("notAfter")
if not not_after:
raise ActivationError("TLS_CERT_EXPIRY_MISSING")
remaining = ssl.cert_time_to_seconds(not_after) - now_fn()
if remaining < 7 * 86400:
raise ActivationError("TLS_CERT_EXPIRY_TOO_CLOSE")
return {
"cert_sha256": sha256_file(cert_path),
"remaining_days": int(remaining // 86400),
}
@dataclass(frozen=True)
class Approval:
approval_id: str
candidate_oid: str
domain: str
bind_host: str
port: int
tls_cert_sha256: str
owner: str
issued_at: datetime
not_before: datetime
expires_at: datetime
raw_sha256: str
@classmethod
def load(
cls,
path: Path,
*,
candidate_oid: str,
cert_path: Path,
now: datetime | None = None,
) -> "Approval":
validate_root_file(path, exact_mode=0o600, max_size=64 * 1024)
try:
value = json.loads(path.read_text(encoding="utf-8"))
except Exception as exc:
raise ActivationError("APPROVAL_JSON_INVALID") from exc
if not isinstance(value, Mapping):
raise ActivationError("APPROVAL_JSON_INVALID")
allowed = {
"schema",
"approval_id",
"candidate_oid",
"domain",
"public_bind",
"bind_host",
"port",
"tls_cert_sha256",
"issued_at",
"not_before",
"expires_at",
"owner",
}
if set(value) != allowed or value.get("schema") != APPROVAL_SCHEMA:
raise ActivationError("APPROVAL_FIELDS_INVALID")
approval_id = value.get("approval_id")
if not isinstance(approval_id, str) or APPROVAL_ID_RE.fullmatch(approval_id) is None:
raise ActivationError("APPROVAL_ID_INVALID")
if not OID_RE.fullmatch(str(value.get("candidate_oid") or "")):
raise ActivationError("APPROVAL_CANDIDATE_INVALID")
if value["candidate_oid"] != candidate_oid:
raise ActivationError("APPROVAL_CANDIDATE_MISMATCH")
if value.get("public_bind") is not True:
raise ActivationError("APPROVAL_PUBLIC_BIND_REQUIRED")
if value.get("bind_host") != PUBLIC_HOST or value.get("port") != PUBLIC_PORT:
raise ActivationError("APPROVAL_BINDING_INVALID")
cert_sha = str(value.get("tls_cert_sha256") or "")
if SHA_RE.fullmatch(cert_sha) is None:
raise ActivationError("APPROVAL_CERT_SHA_INVALID")
if not cert_path.is_file() or sha256_file(cert_path) != cert_sha:
raise ActivationError("APPROVAL_CERT_SHA_MISMATCH")
domain = validate_domain(value.get("domain"))
owner = value.get("owner")
if not isinstance(owner, str) or not owner.strip() or len(owner) > 255:
raise ActivationError("APPROVAL_OWNER_INVALID")
issued_at = parse_time(value.get("issued_at"), field="issued_at")
not_before = parse_time(value.get("not_before"), field="not_before")
expires_at = parse_time(value.get("expires_at"), field="expires_at")
current = now or datetime.now(timezone.utc)
if not (issued_at <= not_before <= expires_at):
raise ActivationError("APPROVAL_TIME_ORDER_INVALID")
if current < not_before:
raise ActivationError("APPROVAL_NOT_YET_VALID")
if current >= expires_at:
raise ActivationError("APPROVAL_EXPIRED")
if expires_at - issued_at > __import__("datetime").timedelta(days=7):
raise ActivationError("APPROVAL_VALIDITY_TOO_LONG")
return cls(
approval_id=approval_id,
candidate_oid=candidate_oid,
domain=domain,
bind_host=PUBLIC_HOST,
port=PUBLIC_PORT,
tls_cert_sha256=cert_sha,
owner=owner.strip(),
issued_at=issued_at,
not_before=not_before,
expires_at=expires_at,
raw_sha256=sha256_file(path),
)
def validate_release(release_root: Path) -> dict[str, Path]:
if release_root.is_symlink() or not release_root.is_dir():
raise ActivationError("IMMUTABLE_RELEASE_INVALID")
edge = release_root / "active-runtime/app/aios_gpt_https_edge.py"
controller = release_root / "active-runtime/app/aios_gpt_https_edge_activation.py"
for path in (edge, controller):
if path.is_symlink() or not path.is_file():
raise ActivationError("IMMUTABLE_RELEASE_INVALID")
return {"edge": edge, "controller": controller}
def render_dropin(approval: Approval, release_root: Path) -> str:
app = release_root / "active-runtime/app/aios_gpt_https_edge.py"
return (
"[Service]\n"
f"Environment=AIOS_GPT_EDGE_HOST={approval.bind_host}\n"
f"Environment=AIOS_GPT_EDGE_PORT={approval.port}\n"
"Environment=AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1\n"
f"Environment=AIOS_GPT_EDGE_PUBLIC_DOMAIN={approval.domain}\n"
"ExecStart=\n"
f"ExecStart=/usr/bin/python3 -B {app}\n"
f"ReadOnlyPaths={release_root}\n"
)
def atomic_write(path: Path, data: str, mode: int = 0o600) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fd, name = tempfile.mkstemp(prefix="." + path.name + ".", dir=path.parent)
try:
os.fchmod(fd, mode)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(data)
handle.flush()
os.fsync(handle.fileno())
os.replace(name, path)
finally:
if os.path.exists(name):
os.unlink(name)
class Systemd:
def run(self, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["/usr/bin/systemctl", *args],
check=check,
text=True,
capture_output=True,
timeout=30,
)
def active(self, unit: str) -> bool:
return self.run("is-active", "--quiet", unit, check=False).returncode == 0
def enabled(self, unit: str) -> bool:
return self.run("is-enabled", "--quiet", unit, check=False).returncode == 0
def show(self, unit: str, prop: str) -> str:
return self.run("show", unit, f"--property={prop}", "--value").stdout.strip()
def require_ingress_ready(systemd: Systemd) -> None:
if not systemd.active(INGRESS_UNIT):
raise ActivationError("INGRESS_SERVICE_NOT_ACTIVE")
try:
with urllib.request.urlopen(INGRESS_HEALTH, timeout=5) as response:
if response.status != 200:
raise ActivationError("INGRESS_HEALTH_FAILED")
value = json.loads(response.read().decode())
except ActivationError:
raise
except Exception as exc:
raise ActivationError("INGRESS_HEALTH_FAILED") from exc
if value.get("status") != "ok":
raise ActivationError("INGRESS_HEALTH_FAILED")
def tls_probe(domain: str, cert_path: Path, *, port: int = PUBLIC_PORT) -> None:
context = ssl.create_default_context(cafile=str(cert_path))
context.check_hostname = True
context.verify_mode = ssl.CERT_REQUIRED
try:
with socket.create_connection(("127.0.0.1", port), timeout=5) as sock:
with context.wrap_socket(sock, server_hostname=domain) as tls_sock:
if tls_sock.version() not in {"TLSv1.2", "TLSv1.3"}:
raise ActivationError("EDGE_TLS_VERSION_INVALID")
except ActivationError:
raise
except Exception as exc:
raise ActivationError("EDGE_TLS_PROBE_FAILED") from exc
def receipt_for(
approval: Approval,
release_root: Path,
*,
state: str,
) -> dict[str, Any]:
return {
"schema": STATE_SCHEMA,
"state": state,
"approval_id": approval.approval_id,
"approval_sha256": approval.raw_sha256,
"candidate_oid": approval.candidate_oid,
"domain": approval.domain,
"bind_host": approval.bind_host,
"port": approval.port,
"tls_cert_sha256": approval.tls_cert_sha256,
"release_root": str(release_root),
"updated_at": datetime.now(timezone.utc).isoformat(),
}
def activate(
*,
approval: Approval,
release_root: Path,
cert_path: Path,
key_path: Path,
dropin_path: Path,
state_path: Path,
systemd: Systemd,
) -> dict[str, Any]:
if os.geteuid() != 0:
raise ActivationError("ROOT_REQUIRED")
release = validate_release(release_root)
app = release["edge"]
require_ingress_ready(systemd)
validate_tls_pair(cert_path, key_path, approval.domain)
if state_path.is_file():
try:
existing = json.loads(state_path.read_text(encoding="utf-8"))
except Exception:
existing = {}
if (
existing.get("schema") == STATE_SCHEMA
and existing.get("state") == "ACTIVE"
and existing.get("approval_sha256") == approval.raw_sha256
and existing.get("candidate_oid") == approval.candidate_oid
and systemd.active(EDGE_UNIT)
):
return existing
dropin = render_dropin(approval, release_root)
modified = False
try:
atomic_write(dropin_path, dropin, mode=0o644)
modified = True
systemd.run("daemon-reload")
systemd.run("enable", EDGE_UNIT)
systemd.run("restart", EDGE_UNIT)
if not systemd.active(EDGE_UNIT):
raise ActivationError("EDGE_SERVICE_NOT_ACTIVE")
exec_start = systemd.show(EDGE_UNIT, "ExecStart")
environment = systemd.show(EDGE_UNIT, "Environment")
if str(app) not in exec_start:
raise ActivationError("EDGE_EFFECTIVE_EXECSTART_INVALID")
for expected in (
f"AIOS_GPT_EDGE_HOST={approval.bind_host}",
f"AIOS_GPT_EDGE_PORT={approval.port}",
"AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1",
f"AIOS_GPT_EDGE_PUBLIC_DOMAIN={approval.domain}",
):
if expected not in environment:
raise ActivationError("EDGE_EFFECTIVE_ENVIRONMENT_INVALID")
tls_probe(approval.domain, cert_path, port=approval.port)
receipt = receipt_for(approval, release_root, state="ACTIVE")
atomic_write(state_path, json.dumps(receipt, indent=2, sort_keys=True) + "\n")
return receipt
except Exception:
if modified:
rollback(
dropin_path=dropin_path,
state_path=state_path,
systemd=systemd,
reason="activation_failed",
)
raise
def rollback(
*,
dropin_path: Path,
state_path: Path,
systemd: Systemd,
reason: str,
) -> dict[str, Any]:
if os.geteuid() != 0:
raise ActivationError("ROOT_REQUIRED")
systemd.run("stop", EDGE_UNIT, check=False)
systemd.run("disable", EDGE_UNIT, check=False)
dropin_path.unlink(missing_ok=True)
systemd.run("daemon-reload")
if systemd.active(EDGE_UNIT):
raise ActivationError("EDGE_ROLLBACK_SERVICE_STILL_ACTIVE")
previous: dict[str, Any] = {}
if state_path.is_file():
try:
previous = json.loads(state_path.read_text(encoding="utf-8"))
except Exception:
previous = {}
receipt = {
"schema": STATE_SCHEMA,
"state": "ROLLED_BACK",
"reason": reason[:255],
"candidate_oid": previous.get("candidate_oid"),
"approval_id": previous.get("approval_id"),
"domain": previous.get("domain"),
"updated_at": datetime.now(timezone.utc).isoformat(),
}
atomic_write(state_path, json.dumps(receipt, indent=2, sort_keys=True) + "\n")
return receipt
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=("check", "activate", "rollback"))
parser.add_argument("--candidate")
parser.add_argument("--approval", default="/etc/aios-gitops/gpt-https-edge.approved")
parser.add_argument("--tls-cert", default="/etc/aios-gitops/gpt-https-edge-tls.crt")
parser.add_argument("--tls-key", default="/etc/aios-gitops/gpt-https-edge-tls.key")
parser.add_argument("--release-root")
parser.add_argument(
"--dropin",
default=f"/etc/systemd/system/{EDGE_UNIT}.d/{DROPIN_NAME}",
)
parser.add_argument(
"--state",
default="/var/lib/aios-gpt-https-edge-activation/state.json",
)
parser.add_argument("--reason", default="owner_requested")
args = parser.parse_args()
systemd = Systemd()
if args.command == "rollback":
value = rollback(
dropin_path=Path(args.dropin),
state_path=Path(args.state),
systemd=systemd,
reason=args.reason,
)
print(json.dumps(value, indent=2, sort_keys=True))
return 0
candidate = str(args.candidate or "")
if OID_RE.fullmatch(candidate) is None:
raise SystemExit("CANDIDATE_REQUIRED")
release_root = (
Path(args.release_root)
if args.release_root
else Path(f"/opt/aios/releases/{candidate}-gpt-https-edge-public")
)
cert_path = Path(args.tls_cert)
key_path = Path(args.tls_key)
approval = Approval.load(
Path(args.approval),
candidate_oid=candidate,
cert_path=cert_path,
)
tls = validate_tls_pair(cert_path, key_path, approval.domain)
release = validate_release(release_root)
require_ingress_ready(systemd)
if args.command == "check":
value = {
"status": "READY",
"candidate_oid": candidate,
"approval_id": approval.approval_id,
"approval_sha256": approval.raw_sha256,
"domain": approval.domain,
"bind_host": approval.bind_host,
"port": approval.port,
"edge_sha256": sha256_file(release["edge"]),
"controller_sha256": sha256_file(release["controller"]),
**tls,
"public_network_mutation": False,
}
print(json.dumps(value, indent=2, sort_keys=True))
return 0
value = activate(
approval=approval,
release_root=release_root,
cert_path=cert_path,
key_path=key_path,
dropin_path=Path(args.dropin),
state_path=Path(args.state),
systemd=systemd,
)
print(json.dumps(value, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,714 @@
#!/usr/bin/env python3
"""Persistent release-preparation Online Agent worker.
The worker is intentionally narrower than the generic engineering worker:
it can only consume tasks whose required capability set is exactly
{"release:candidate"} and whose payload asks for immutable release-candidate
creation. It never receives Owner signing, deployment, tag, baseline or Git
write credentials.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import socket
import threading
import time
import urllib.error
import urllib.request
from typing import Any, Mapping
from aios_autonomous_control_plane import (
classify_failure,
repair_for_failure,
runtime_admission,
)
AGENT_ID = os.environ.get(
"AIOS_RELEASE_WORKER_AGENT_ID",
"online-release-worker-cn-network-mkt-20260819",
).strip()
AGENT_VERSION = os.environ.get(
"AIOS_RELEASE_WORKER_AGENT_VERSION",
"2026-08-19.1-release-candidate",
).strip()
TASK_API = os.environ.get(
"AIOS_AGENT_TASK_API_URL",
"http://127.0.0.1:18940",
).rstrip("/")
RELEASE_API = os.environ.get("AIOS_RELEASE_API_URL", "").rstrip("/")
POLL_SECONDS = int(os.environ.get("AIOS_RELEASE_WORKER_POLL_SECONDS", "3"))
LEASE_SECONDS = int(os.environ.get("AIOS_RELEASE_WORKER_LEASE_SECONDS", "60"))
HEARTBEAT_SECONDS = int(
os.environ.get("AIOS_RELEASE_WORKER_HEARTBEAT_SECONDS", "15")
)
STATE_ROOT = Path(
os.environ.get(
"AIOS_RELEASE_WORKER_STATE_ROOT",
"/var/lib/aios-online-agent-release",
)
)
EXPECTED_CAPABILITIES = frozenset({"release:candidate"})
LIVENESS_PATH = Path(
os.environ.get(
"AIOS_WORKER_LIVENESS_PATH",
str(STATE_ROOT / "worker-liveness.json"),
)
)
CANDIDATE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,255}$")
OID_RE = re.compile(r"^[0-9a-f]{40}$")
VERSION_RE = re.compile(r"^V[0-9]+R[0-9]+C[0-9]+$")
TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
SECRET_MARKERS = (
"authorization",
"token",
"secret",
"credential",
"password",
"private_key",
"signature",
)
class ReleaseWorkerError(RuntimeError):
def __init__(self, code: str, *, phase: str) -> None:
super().__init__(code)
self.code = code
self.phase = phase
def now() -> str:
return datetime.now(timezone.utc).isoformat()
def canonical(value: object) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def sha256(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
def sanitized(value: Any) -> Any:
if isinstance(value, Mapping):
result: dict[str, Any] = {}
for key, inner in value.items():
lowered = str(key).lower()
if any(marker in lowered for marker in SECRET_MARKERS):
result[str(key)] = "[REDACTED]"
else:
result[str(key)] = sanitized(inner)
return result
if isinstance(value, list):
return [sanitized(item) for item in value]
return value
def audit(event: str, **fields: Any) -> None:
STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
path = STATE_ROOT / "events.jsonl"
record = {
"schema": "aios.release-worker.event.v1",
"event": event,
"created_at": now(),
"agent_id": AGENT_ID,
**sanitized(fields),
}
with path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(record, sort_keys=True) + "\n")
os.chmod(path, 0o640)
def sd_notify(message: str) -> None:
target = os.environ.get("NOTIFY_SOCKET", "")
if not target:
return
if target.startswith("@"):
address = "\0" + target[1:]
else:
address = target
sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
try:
sock.connect(address)
sock.sendall(message.encode("utf-8"))
except OSError:
# Liveness JSON remains the diagnostic source; systemd will enforce
# WatchdogSec if notify transport is persistently broken.
pass
finally:
sock.close()
def write_liveness(
state: str,
*,
task_id: str | None = None,
stage: str | None = None,
error: str | None = None,
visible_count: int | None = None,
) -> None:
STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
value = {
"schema": "aios.online-agent.worker-liveness.v1",
"agent_id": AGENT_ID,
"agent_type": "release-preparation-worker",
"agent_version": AGENT_VERSION,
"pid": os.getpid(),
"state": state,
"last_seen_epoch": time.time(),
"last_seen_at": now(),
"poll_seconds": POLL_SECONDS,
"heartbeat_seconds": HEARTBEAT_SECONDS,
"api_base": TASK_API,
"capabilities": sorted(EXPECTED_CAPABILITIES),
"task_id": task_id,
"stage": stage,
"error": error,
"visible_count": visible_count,
}
LIVENESS_PATH.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
temporary = LIVENESS_PATH.with_name(
"." + LIVENESS_PATH.name + f".{os.getpid()}.tmp"
)
temporary.write_text(
json.dumps(value, ensure_ascii=False, sort_keys=True) + "\n",
encoding="utf-8",
)
os.chmod(temporary, 0o640)
os.replace(temporary, LIVENESS_PATH)
def load_credential(name: str) -> str:
directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
path = directory / name
if path.is_symlink() or not path.is_file():
raise ReleaseWorkerError(
"CREDENTIAL_UNAVAILABLE:" + name,
phase="startup",
)
value = path.read_text(encoding="utf-8").strip()
if len(value) < 24 or any(ch.isspace() for ch in value):
raise ReleaseWorkerError(
"CREDENTIAL_INVALID:" + name,
phase="startup",
)
return value
def http_json(
base: str,
method: str,
route: str,
token: str,
body: Mapping[str, Any] | None = None,
*,
timeout: int = 30,
) -> tuple[int, dict[str, Any]]:
data = None if body is None else canonical(body)
headers = {"Accept": "application/json"}
if token:
headers["Authorization"] = "Bearer " + token
if data is not None:
headers["Content-Type"] = "application/json"
request = urllib.request.Request(
base + route,
method=method,
data=data,
headers=headers,
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
raw, status = response.read(), response.status
except urllib.error.HTTPError as exc:
raw, status = exc.read(), exc.code
except (urllib.error.URLError, TimeoutError) as exc:
raise ReleaseWorkerError(
"HTTP_TRANSPORT_UNAVAILABLE",
phase="network",
) from exc
try:
value = json.loads(raw.decode())
except Exception as exc:
raise ReleaseWorkerError(
"HTTP_RESPONSE_INVALID",
phase="network",
) from exc
return status, value if isinstance(value, dict) else {}
@dataclass(frozen=True)
class ReleaseRequest:
candidate_id: str
source_commit: str
version: str
projection_id: str | None
@classmethod
def from_task(cls, task: Mapping[str, Any]) -> "ReleaseRequest":
if not TASK_RE.fullmatch(str(task.get("task_id") or "")):
raise ReleaseWorkerError("TASK_ID_INVALID", phase="validation")
caps = task.get("required_capabilities")
if (
not isinstance(caps, list)
or frozenset(caps) != EXPECTED_CAPABILITIES
or len(caps) != 1
):
raise ReleaseWorkerError(
"TASK_CAPABILITY_SCOPE_INVALID",
phase="validation",
)
metadata = task.get("metadata")
if not isinstance(metadata, Mapping):
metadata = {}
constraints = task.get("constraints")
if not isinstance(constraints, Mapping):
constraints = {}
release = constraints.get("release_candidate")
if not isinstance(release, Mapping):
release = metadata.get("release_candidate")
if not isinstance(release, Mapping):
release = task.get("release_candidate")
if not isinstance(release, Mapping):
raise ReleaseWorkerError(
"RELEASE_CANDIDATE_PAYLOAD_MISSING",
phase="validation",
)
allowed = {
"candidate_id",
"source_commit",
"version",
"projection_id",
}
if set(release) - allowed:
raise ReleaseWorkerError(
"RELEASE_CANDIDATE_PAYLOAD_FIELDS_INVALID",
phase="validation",
)
candidate_id = str(release.get("candidate_id") or "")
source_commit = str(release.get("source_commit") or "")
version = str(release.get("version") or "")
projection_id = release.get("projection_id")
if CANDIDATE_RE.fullmatch(candidate_id) is None:
raise ReleaseWorkerError(
"RELEASE_CANDIDATE_ID_INVALID",
phase="validation",
)
if OID_RE.fullmatch(source_commit) is None:
raise ReleaseWorkerError(
"RELEASE_SOURCE_COMMIT_INVALID",
phase="validation",
)
if VERSION_RE.fullmatch(version) is None:
raise ReleaseWorkerError(
"RELEASE_VERSION_INVALID",
phase="validation",
)
if projection_id is not None:
projection_id = str(projection_id)
if not re.fullmatch(r"projection-[0-9a-f]{32}", projection_id):
raise ReleaseWorkerError(
"RELEASE_PROJECTION_ID_INVALID",
phase="validation",
)
return cls(
candidate_id=candidate_id,
source_commit=source_commit,
version=version,
projection_id=projection_id,
)
def body(self) -> dict[str, str]:
return {
"candidate_id": self.candidate_id,
"source_commit": self.source_commit,
"version": self.version,
}
class Heartbeat:
def __init__(
self,
agent_token: str,
task_id: str,
lease_token: str,
) -> None:
self.agent_token = agent_token
self.task_id = task_id
self.lease_token = lease_token
self.stage = "claimed"
self.failure: str | None = None
self.stop = threading.Event()
self.thread = threading.Thread(
target=self._run,
name="release-worker-heartbeat",
daemon=True,
)
def first(self) -> None:
status, value = http_json(
TASK_API,
"POST",
f"/v1/tasks/{self.task_id}/heartbeat",
self.agent_token,
{
"lease_token": self.lease_token,
"lease_seconds": LEASE_SECONDS,
"progress": {
"stage": self.stage,
"worker_version": AGENT_VERSION,
},
},
)
if status != 200:
raise ReleaseWorkerError(
"TASK_HEARTBEAT_FAILED",
phase="heartbeat",
)
audit(
"TASK_HEARTBEAT",
task_id=self.task_id,
stage=self.stage,
api_status=status,
)
write_liveness(
"BUSY",
task_id=self.task_id,
stage=self.stage,
)
sd_notify("WATCHDOG=1")
def _run(self) -> None:
while not self.stop.wait(HEARTBEAT_SECONDS):
try:
status, value = http_json(
TASK_API,
"POST",
f"/v1/tasks/{self.task_id}/heartbeat",
self.agent_token,
{
"lease_token": self.lease_token,
"lease_seconds": LEASE_SECONDS,
"progress": {
"stage": self.stage,
"worker_version": AGENT_VERSION,
},
},
)
if status != 200:
self.failure = str(value.get("error") or status)
return
audit(
"TASK_HEARTBEAT",
task_id=self.task_id,
stage=self.stage,
api_status=status,
)
write_liveness(
"BUSY",
task_id=self.task_id,
stage=self.stage,
)
sd_notify("WATCHDOG=1")
except Exception as exc:
self.failure = type(exc).__name__
return
def ensure(self) -> None:
if self.failure:
raise ReleaseWorkerError(
"TASK_LEASE_LOST",
phase="heartbeat",
)
def close(self) -> None:
self.stop.set()
if self.thread.ident is not None:
self.thread.join(timeout=2)
def release_admission(
task: Mapping[str, Any],
*,
release_token: str,
):
required = task.get("required_capabilities")
if not isinstance(required, list):
required = []
def release_api_health() -> bool:
status, value = http_json(
RELEASE_API,
"GET",
"/health",
"",
timeout=10,
)
return status == 200 and str(value.get("status") or "").lower() in {"ok", "healthy"}
def authenticated_probe() -> bool:
status, _ = http_json(
RELEASE_API,
"POST",
"/createReleaseCandidate",
release_token,
{},
timeout=10,
)
# Authentication + credential projection succeeded; the empty release
# request is intentionally rejected by request validation.
return status == 400
probes = {
"credential_projection": authenticated_probe,
"trusted_builder": release_api_health,
"authenticated_probe": authenticated_probe,
"release_api_health": release_api_health,
"worker_liveness": lambda: True,
}
return runtime_admission(
worker_capabilities=EXPECTED_CAPABILITIES,
required_capabilities=required,
requested_operations=("prepare_release",),
probes=probes,
)
def result_payload(
task_id: str,
release: ReleaseRequest,
builder: Mapping[str, Any],
) -> dict[str, Any]:
builder_candidate = builder.get("candidate_id")
if builder_candidate not in {None, release.candidate_id}:
raise ReleaseWorkerError(
"BUILDER_CANDIDATE_BINDING_INVALID",
phase="release",
)
immutable = {
"candidate_id": release.candidate_id,
"source_commit": release.source_commit,
"version": release.version,
}
return {
"status": "SUCCEEDED",
"classification": "RELEASE_CANDIDATE_CREATED",
"execution_mode": "release_candidate",
"adapter": "release_candidate",
"task_id": task_id,
"agent_id": AGENT_ID,
"release_candidate": immutable,
"release_candidate_request_sha256": sha256(canonical(immutable)),
"builder_response": sanitized(dict(builder)),
"projection_id": release.projection_id,
"writer_lease": "NONE",
"changed_files": [],
"direct_main_write": False,
"production_write": False,
"tag_write": False,
"baseline_write": False,
"runtime_switch": False,
"owner_approval_submitted": False,
"credential_material_exposed": False,
}
def execute_task(
task: Mapping[str, Any],
*,
agent_token: str,
release_token: str,
) -> bool:
task_id = str(task.get("task_id") or "")
release = ReleaseRequest.from_task(task)
admission = release_admission(
task,
release_token=release_token,
)
if not admission.admitted:
audit(
"TASK_ADMISSION_BLOCKED",
task_id=task_id,
admission=admission.as_dict(),
)
write_liveness(
"BLOCKED",
task_id=task_id,
stage="execution_ready",
error=admission.failure_class.value if admission.failure_class else "ADMISSION_FAILED",
)
return False
audit(
"TASK_ADMISSION_PASSED",
task_id=task_id,
admission=admission.as_dict(),
)
status, claimed = http_json(
TASK_API,
"POST",
f"/v1/tasks/{task_id}/claim",
agent_token,
{"lease_seconds": LEASE_SECONDS},
)
if status in {409, 423}:
return False
if status != 200:
raise ReleaseWorkerError(
"TASK_CLAIM_FAILED",
phase="claim",
)
lease_token = str(claimed.get("lease_token") or "")
if len(lease_token) < 16:
raise ReleaseWorkerError(
"TASK_LEASE_TOKEN_INVALID",
phase="claim",
)
attempt = int(claimed.get("attempt") or 0)
audit(
"TASK_CLAIMED",
task_id=task_id,
attempt=attempt,
)
heartbeat = Heartbeat(agent_token, task_id, lease_token)
heartbeat.first()
heartbeat.thread.start()
try:
heartbeat.stage = "release_candidate"
heartbeat.ensure()
status, builder = http_json(
RELEASE_API,
"POST",
"/createReleaseCandidate",
release_token,
release.body(),
timeout=120,
)
if status not in {200, 201, 202}:
raise ReleaseWorkerError(
"RELEASE_CANDIDATE_CREATE_FAILED:"
+ str(builder.get("error") or status),
phase="release",
)
heartbeat.ensure()
result = result_payload(task_id, release, builder)
idempotency_key = (
f"{task_id}-attempt-{attempt}-release-result-v1"
)
status, submitted = http_json(
TASK_API,
"POST",
f"/v1/tasks/{task_id}/result",
agent_token,
{
"lease_token": lease_token,
"idempotency_key": idempotency_key,
"result": result,
},
)
if status not in {200, 201, 202}:
raise ReleaseWorkerError(
"TASK_RESULT_SUBMIT_FAILED",
phase="result",
)
audit(
"RESULT_SUBMITTED",
task_id=task_id,
attempt=attempt,
release_candidate_id=release.candidate_id,
)
return True
finally:
heartbeat.close()
def main() -> int:
if not RELEASE_API:
raise SystemExit("FATAL:AIOS_RELEASE_API_URL_REQUIRED")
agent_token = load_credential("agent_token")
release_token = load_credential("release_candidate_token")
STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
audit(
"WORKER_STARTED",
capabilities=sorted(EXPECTED_CAPABILITIES),
release_api=RELEASE_API,
)
write_liveness("STARTING", stage="startup")
sd_notify("READY=1\nSTATUS=release worker ready")
while True:
try:
status, response = http_json(
TASK_API,
"POST",
"/v1/tasks/poll",
agent_token,
{"limit": 100},
)
if status != 200:
raise ReleaseWorkerError(
"TASK_POLL_FAILED",
phase="poll",
)
tasks = response.get("tasks")
if not isinstance(tasks, list):
tasks = []
write_liveness(
"IDLE",
stage="poll",
visible_count=len(tasks),
)
sd_notify("WATCHDOG=1\nSTATUS=release worker poll ok")
handled = False
for raw in tasks:
if not isinstance(raw, Mapping):
continue
caps = raw.get("required_capabilities")
if (
not isinstance(caps, list)
or frozenset(caps) != EXPECTED_CAPABILITIES
):
continue
try:
handled = execute_task(
raw,
agent_token=agent_token,
release_token=release_token,
) or handled
except ReleaseWorkerError as exc:
failure = classify_failure(exc.code, phase=exc.phase)
repair = repair_for_failure(failure)
audit(
"TASK_EXECUTION_FAILED",
task_id=raw.get("task_id"),
phase=exc.phase,
error=exc.code,
failure_class=failure.value,
repair=repair.as_dict(),
)
time.sleep(0 if handled else POLL_SECONDS)
except KeyboardInterrupt:
return 0
except Exception as exc:
audit(
"WORKER_LOOP_ERROR",
error=type(exc).__name__,
)
write_liveness(
"DEGRADED",
stage="loop",
error=type(exc).__name__,
)
time.sleep(POLL_SECONDS)
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,41 @@
[Unit]
Description=AIOS disabled-by-default HTTPS edge for GPT read-only Online Task ingress
After=network-online.target aios-gpt-online-task-ingress.service
Wants=network-online.target aios-gpt-online-task-ingress.service
ConditionPathExists=/etc/aios-gitops/gpt-https-edge.approved
[Service]
Type=simple
DynamicUser=yes
LoadCredential=tls_cert:/etc/aios-gitops/gpt-https-edge-tls.crt
LoadCredential=tls_key:/etc/aios-gitops/gpt-https-edge-tls.key
Environment=AIOS_GPT_EDGE_TLS_CERT_FILE=%d/tls_cert
Environment=AIOS_GPT_EDGE_TLS_KEY_FILE=%d/tls_key
Environment=AIOS_GPT_EDGE_HOST=127.0.0.1
Environment=AIOS_GPT_EDGE_PORT=18443
Environment=AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=0
Environment=AIOS_GPT_EDGE_PUBLIC_DOMAIN=
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_gpt_https_edge.py
Restart=on-failure
RestartSec=2
NoNewPrivileges=yes
PrivateTmp=yes
PrivateDevices=yes
ProtectSystem=strict
ProtectHome=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectControlGroups=yes
ProtectKernelLogs=yes
ProtectClock=yes
LockPersonality=yes
RestrictSUIDSGID=yes
RestrictRealtime=yes
RestrictNamespaces=yes
MemoryDenyWriteExecute=yes
RestrictAddressFamilies=AF_INET AF_INET6
InaccessiblePaths=/data/aios-gitops /etc/aios-gitops /home /var/lib/aios-agent-task
UMask=0077
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,47 @@
[Unit]
Description=AIOS durable Owner authorization broker
After=network.target aios-chatgpt-bridge-v2-staging.service
Wants=aios-chatgpt-bridge-v2-staging.service
[Service]
Type=simple
User=root
Group=aios-executor-exchange
RuntimeDirectory=aios-owner-authorization-broker
RuntimeDirectoryMode=0750
StateDirectory=aios-owner-authorization-broker
StateDirectoryMode=0750
LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_POLICY=/etc/aios-gitops/owner-durable-authorization.json
Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_REVOCATIONS=/etc/aios-gitops/owner-durable-authorization-revocations.json
Environment=AIOS_OWNER_AUTH_BROKER_ACCOUNT=chatgpt-governed-implementer-singlehost
Environment=AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_durable_owner_authorization.py
Restart=on-failure
RestartSec=2
NoNewPrivileges=yes
PrivateTmp=yes
PrivateDevices=yes
ProtectSystem=strict
ProtectHome=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectControlGroups=yes
ProtectKernelLogs=yes
ProtectClock=yes
LockPersonality=yes
RestrictSUIDSGID=yes
RestrictRealtime=yes
RestrictNamespaces=yes
MemoryDenyWriteExecute=yes
RestrictAddressFamilies=AF_UNIX AF_INET
IPAddressDeny=any
IPAddressAllow=127.0.0.0/8
ReadOnlyPaths=/etc/aios-gitops/owner-durable-authorization.json
ReadOnlyPaths=/etc/aios-gitops/owner-durable-authorization-revocations.json
InaccessiblePaths=/data/aios-gitops /home
UMask=0007
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,50 @@
[Unit]
Description=AIOS release-candidate Online Agent worker
After=network.target aios-agent-task-api.service
Wants=aios-agent-task-api.service
[Service]
Type=notify
NotifyAccess=main
WatchdogSec=90
User=root
Group=root
StateDirectory=aios-online-agent-release
StateDirectoryMode=0750
LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-release-online-agent-worker/agent_token
LoadCredentialEncrypted=release_candidate_token:/etc/credstore.encrypted/aios-release-online-agent-worker/release_candidate_token
Environment=AIOS_RELEASE_WORKER_AGENT_ID=online-release-worker-cn-network-mkt-20260819
Environment=AIOS_RELEASE_WORKER_AGENT_VERSION=2026-08-19.1-release-candidate
Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
Environment=AIOS_RELEASE_WORKER_STATE_ROOT=/var/lib/aios-online-agent-release
Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-release
Environment=AIOS_RELEASE_WORKER_POLL_SECONDS=3
Environment=AIOS_RELEASE_WORKER_LEASE_SECONDS=60
Environment=AIOS_RELEASE_WORKER_HEARTBEAT_SECONDS=15
Environment=AIOS_RELEASE_API_URL=http://127.0.0.1:18890
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_release_online_agent_worker.py
Restart=always
RestartSec=2
NoNewPrivileges=yes
PrivateTmp=yes
PrivateDevices=yes
ProtectSystem=strict
ProtectHome=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectControlGroups=yes
ProtectKernelLogs=yes
ProtectClock=yes
LockPersonality=yes
RestrictSUIDSGID=yes
RestrictRealtime=yes
RestrictNamespaces=yes
MemoryDenyWriteExecute=yes
RestrictAddressFamilies=AF_UNIX AF_INET
IPAddressDeny=any
IPAddressAllow=127.0.0.0/8
InaccessiblePaths=/data/aios-gitops /home
UMask=0077
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,96 @@
# Canonical all-modified-files packager and secret scanning
## Problem
The historical all-modified-files evidence packager lived only as a copied handoff script.
Its private-key scanner matched the fragment:
```text
BEGIN PRIVATE KEY
```
instead of a complete PEM header. A regression test containing that literal therefore caused a
false secret finding even though the candidate contained no credential.
A task had to use a temporary evidence-only compatibility packager to complete its bundle. That is
acceptable as an emergency workaround but not as a durable unattended path.
## Repository source of truth
The canonical implementation is now:
```text
active-runtime/app/aios_all_modified_files_packager.py
```
It preserves the historical output/bundle schemas:
```text
aios.all-modified-files.bundle.v1
aios.all-modified-files.bundle-summary.v1
aios.modified-documents.index.v1
```
and the host-visible status lines:
```text
ALL_MODIFIED_FILES_BUNDLE=PASS
ALL_MODIFIED_FILES_BUNDLE_PATH=...
ALL_MODIFIED_FILES_BUNDLE_SHA256=...
```
Future governed tasks should use this repository version instead of copying an unversioned scanner.
## Secret scanning invariant
Private-key detection requires a complete PEM header:
```text
-----BEGIN [RSA|EC|DSA|OPENSSH ]PRIVATE KEY-----
```
The following source/test/document text is not secret material and must remain packageable:
```text
BEGIN PRIVATE KEY
PRIVATE KEY
PEM private-key header
```
Bearer and key/value credential detection remains fail-closed:
```text
Authorization: Bearer <20+ token chars>
password=<12+ non-space chars>
passwd=<12+ non-space chars>
secret=<12+ non-space chars>
api_key=<12+ non-space chars>
api-key=<12+ non-space chars>
```
Tests construct the real PEM header dynamically so the scanner's own test source does not contain a
literal credential-looking header.
## Bundle integrity
The packager continues to enforce:
- exact base→candidate ancestry;
- before/after file capture;
- rename/copy/delete handling;
- path traversal denial;
- execution evidence symlink denial;
- secret scan over repository before/after and execution evidence;
- document index;
- SHA-256 closure;
- ZIP without symlinks.
No secret finding is downgraded to a warning.
## Migration
The old handoff script may remain for historical reproducibility, but new handoffs should call the
repository packager from the verified candidate lineage.
A temporary compatibility scanner is no longer needed once this candidate is in the lineage.
@@ -0,0 +1,210 @@
# AIOS Autonomous Control Plane — Admission + Invariant + Repair
## Why this exists
Recent Agent registration, Task claim, release-worker, credential and retry failures shared one
failure mode:
> The control plane accepted an intent before the runtime had proved it could execute the intent.
This design does not add another fleet of daemons. It adds one shared library contract that Task API,
workers and governed host recovery consume.
## Three layers
```text
Intent
↓
Runtime Admission
↓
System Invariants
↓
Repair Policy
↓
Execution
```
### Runtime Admission
Admission occurs before Task claim.
A capability such as `release:candidate` is not considered executable merely because its name
matches. The capability contract also binds:
- supported contract version;
- allowed permission;
- forbidden operations;
- runtime dependencies;
- health requirements.
For `release:candidate`:
```text
permission:
prepare_release
forbidden:
production_deploy
submit_owner_approval
tag_publish
baseline_registry_finalize
runtime_switch
dependencies:
credential_projection
trusted_builder
health:
authenticated_probe
release_api_health
worker_liveness
```
Only a fully passing decision reaches:
```text
EXECUTION_READY
```
A failed decision remains:
```text
EXECUTION_READY_FAILED
```
and the worker does not call `/claim`.
Therefore an infrastructure defect discovered by admission does not consume a business attempt.
### System Invariants
The same admission checks are exposed as a machine-readable invariant snapshot.
Examples:
```text
capability_contract_registry
task_capabilities_registered
worker_capabilities_registered
capability_match
forbidden_operations_absent
credential_projection
trusted_builder
authenticated_probe
release_api_health
worker_liveness
```
A runtime can expose:
```text
HEALTHY
DEGRADED
```
without relying on a human reconstructing state from logs.
### Repair Policy
Failure classification maps to bounded recovery:
```text
DEPENDENCY_UNAVAILABLE → level 0 / auto local
RESOURCE_UNAVAILABLE → level 0 / auto local
WORKER_CRASH → level 0 / auto local
LEASE_LOST → level 0 / auto local
CAPABILITY_CONTRACT_ERROR → level 1 / same-lineage repair task
CREDENTIAL_PROJECTION_ERROR → level 1 / same-lineage repair task
UNKNOWN → level 1 / diagnostic repair task
POLICY_DENIED → level 2 / Owner authorization
BUSINESS_PERMANENT → level 3 / terminal/manual decision
```
Only business-permanent failure consumes the business retry budget in this first contract.
The current Task API does not yet have a first-class `EXECUTION_READY_FAILED` durable state. Until it
does, the Release Worker enforces the most important invariant by refusing to claim a Task whose
admission fails.
## Capability Contract Registry
`aios_capability_contract.py` remains backward-compatible with existing aliases but now validates
every alias target against a versioned semantic contract registry.
Unknown capabilities and wildcard capabilities remain fail-closed.
Agent registration and Task creation should next be changed to call `registered_capability()` so a
capability cannot be accepted at registration and rejected only later by the matcher.
## Credential rule
Admission probes credential usability, not merely credential existence.
For release candidate creation the authenticated probe intentionally sends an invalid empty
`/createReleaseCandidate` request.
Expected:
```text
HTTP 400
```
This proves:
```text
credential readable
→ authentication succeeded
→ scope accepted
→ request validation rejected empty business input
```
A permission failure/401/403/5xx is `EXECUTION_READY_FAILED`.
## Lineage
Repair actions are descriptive, not permission grants. Any repair Task must retain the existing
root/parent/attempt/authority chain and use governed same-lineage recovery.
No repair action may silently create a replacement business Task.
## Authority separation
Capability answers:
```text
can this runtime perform the operation?
```
Authority answers:
```text
is this operation currently authorized?
```
`release:candidate` never implies Owner approval, deployment, tag publication, baseline finalization
or runtime switch.
## Deployment sequence
Phase 1 (this candidate):
- Capability Contract Registry;
- shared Autonomous Control Plane library;
- Release Worker pre-claim admission;
- failure classification + repair evidence.
Phase 2:
- Task API uses the same library before returning Tasks from poll and again during claim CAS;
- Task schema gains durable `ADMITTED/EXECUTION_READY/EXECUTION_READY_FAILED` evidence without
counting pre-side-effect infrastructure failures against business attempts.
Phase 3:
- Agent registration validates against the same Capability Contract Registry;
- external connector writes only durable Intent/Command Inbox records and leaves execution to the
local control plane.
This sequencing avoids a second controller stack while converging every decision onto one contract.
@@ -0,0 +1,228 @@
# Durable Owner Authorization and capability projection
## Problem
AIOS already supports unattended worker execution after a Task is executable, but Owner intent was
not durable at the capability layer. The same Owner repeatedly had to authorize or repair:
- `governed-task:project`;
- `release:candidate`.
This caused valid one-time Owner intent to be lost between conversations and forced repeated human
interaction.
## Design
Durable authorization is represented by a root-owned mode `0600` policy outside Git. The policy is
persistent until explicitly revoked.
The first persistable capabilities are deliberately narrow:
```text
governed-task:project
release:candidate
```
They cover preparation work only.
The following capabilities can never be inherited from this durable grant:
```text
submitOwnerApproval
startDeployment
tag_publish
baseline_registry_finalize
runtime_switch
main_write
production_deploy
force_push
credential_admin
```
Those remain separate high-risk release/Owner-signature gates.
## Policy
Example schema:
```json
{
"schema": "aios.owner-durable-authorization.v1",
"grant_id": "owner-permanent-preparation-20260819",
"owner_identity": "owner@example",
"account": "chatgpt-governed-implementer-singlehost",
"capabilities": [
"governed-task:project",
"release:candidate"
],
"issued_at": "2026-08-19T05:00:00Z",
"not_before": "2026-08-19T05:00:00Z",
"persistent_until_revoked": true,
"constraints": {
"repository": "aios/aios",
"allowed_target_prefix": "agent/",
"release_version_prefix": "V2R2C",
"max_task_risk": "engineering"
}
}
```
The repository never stores the live Owner policy or release credential.
## Revocation
A root-owned mode `0600` revocation file lists `revoked_grant_ids`.
Revocation is checked on every operation, not only at service startup. A running broker therefore
fails closed immediately after revocation without needing a new conversation.
## Broker boundary
The broker exposes only an AF_UNIX socket:
```text
/run/aios-owner-authorization-broker/broker.sock
```
The socket is mode `0660` and owned by the service's `aios-executor-exchange` group. This avoids introducing another long-lived bearer secret solely for local broker access.
It exposes only:
```text
GET /v1/capabilities
POST /v1/project-governed-task
POST /v1/authorize-release-candidate
```
`project-governed-task` calls only the canonical Bridge.
`authorize-release-candidate` persists an Owner authorization decision only. It deliberately does
**not** invent a local trusted-builder endpoint. The external release connector remains responsible
for projecting the account's real `release:candidate` scope. Until that external projection exists,
the decision state is `AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION`.
The broker holds only the existing host control credential needed for canonical governed-task
projection. It never holds an Owner signing key or a release/deployment credential. The caller also supplies the expected account identifier over the Unix transport; the OS socket permission is the local access boundary.
## Canonical Online Task projection
A durable `governed-task:project` grant may create a complete canonical Online Task rather than a
bare Task. It requires:
```text
require_online_task_projection=true
generation=1
branch_mode=CREATE_NEW
max_attempts<=3
risk<=engineering
initial action=source_capture
```
Projected task capabilities are limited to:
```text
git.read
git.test
governed.write.branch
governed.write.file
```
A write-capable task still begins as `source_capture`. Existing placeholder-deferral prevents the
generic worker from claiming it until a deterministic plan is registered. Repository mutation
therefore still requires the normal deterministic plan, WriterLease and governed commit path.
This fixes the previous split-brain behavior where a bare Task could exist without the canonical
governed execution projection.
## External capability projection outbox
A `release:candidate` authorization now creates a durable machine-readable outbox record:
```text
schema=aios.external-capability-projection.request.v1
state=PENDING_EXTERNAL_CONNECTOR
capability=release:candidate
```
The projection identifier is deterministic from:
```text
grant_id
account
capability
request_sha256
```
The outbox stores no credentials or Owner signing material. It retains only the already-authorized
release-candidate request, policy/request hashes, account, grant id and timestamps.
The broker exposes read-only Unix-socket endpoints:
```text
GET /v1/projections
GET /v1/projections/{projection_id}
```
These endpoints re-check the current Owner policy on every read. If the durable grant has been
revoked, enumeration fails closed with `OWNER_GRANT_REVOKED`. The historical outbox record is
retained for audit and is not silently deleted.
The repository intentionally does **not** provide a connector acknowledgement/write endpoint yet.
No in-repository code can truthfully assert that the external ChatGPT connector account has gained
`release:candidate`. Until the platform consumes the outbox and performs its own identity-scope
projection, the only valid state is:
```text
AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION
PENDING_EXTERNAL_CONNECTOR
```
A future external projector must preserve:
```text
release:candidate
!= submitOwnerApproval
!= startDeployment
!= tag_publish
!= baseline_registry_finalize
!= runtime_switch
```
## Idempotency and audit
Every broker operation requires an idempotency key. A receipt SQLite database stores only:
- idempotency key;
- capability;
- request SHA-256;
- response SHA-256;
- sanitized upstream response;
- timestamp.
An exact retry returns the same receipt. Reusing an idempotency key with different content fails.
## Unattended invariant
Once the Owner policy and narrow service credentials are provisioned:
```text
Owner durable grant
→ broker decision
→ canonical governed task projection / release candidate preparation
→ worker / trusted builder
```
No repeated Owner interaction is required for those preparation capabilities.
Production execution remains:
```text
release candidate
→ immutable approval envelope
→ Owner signature
→ deployment
```
The durable grant never substitutes for the Owner signature.
@@ -0,0 +1,114 @@
# Owner-approved GPT HTTPS edge activation and rollback
## Scope
This controller is the final host-side control mechanism before the disabled-by-default GPT HTTPS
edge may bind publicly.
Repository implementation and public activation remain separate states:
```text
IMPLEMENTED
≠
OWNER_APPROVED
≠
ACTIVATED
≠
RUNTIME_VERIFIED
```
The controller never creates owner approval, TLS assets, DNS, firewall rules or ingress bearer
credentials.
## Approval binding
The owner approval file is root-owned mode `0600` and has schema:
```json
{
"schema": "aios.gpt-https-edge.public-approval.v1",
"approval_id": "owner-public-edge-...",
"candidate_oid": "<exact governed edge candidate>",
"domain": "tasks.example.com",
"public_bind": true,
"bind_host": "0.0.0.0",
"port": 18443,
"tls_cert_sha256": "<exact certificate sha256>",
"issued_at": "2026-08-19T00:00:00Z",
"not_before": "2026-08-19T00:00:00Z",
"expires_at": "2026-08-20T00:00:00Z",
"owner": "<owner identity>"
}
```
Approval lifetime is at most seven days. Candidate/domain/bind/port/certificate are immutable within
one approval.
Changing any of those requires a new owner approval.
## Check
```text
aios_gpt_https_edge_activation.py check --candidate <GE>
```
Checks:
- approval ownership/mode/schema/time window;
- exact candidate binding;
- exact certificate hash;
- certificate/private-key pair;
- certificate hostname;
- certificate remaining lifetime;
- loopback GPT ingress service health.
It performs no public network mutation.
## Activate
`activate` additionally:
- writes one exact systemd public-bind drop-in atomically;
- enables/restarts only the HTTPS edge unit;
- verifies effective ExecStart points at the immutable candidate release;
- verifies effective public-bind/domain environment;
- performs a TLS handshake to `127.0.0.1:18443` using the approved public hostname;
- writes a durable activation receipt.
Any failure after drop-in mutation automatically invokes rollback.
Exact replay of an already-active matching approval returns the durable receipt without restarting
or creating duplicate state.
## Rollback
`rollback`:
- stops and disables only the HTTPS edge;
- removes only the controller-owned public-bind drop-in;
- daemon-reloads systemd;
- verifies edge inactive;
- writes a durable rollback receipt.
It does not delete:
- owner approval;
- TLS certificate/private key;
- ingress bearer token;
- internal loopback ingress;
- Task/Job state;
- Git branches.
Rollback is therefore replay-safe and does not require reconstructing secrets.
## Not covered
This controller intentionally does not own:
- DNS records;
- host firewall/NAT/security-group changes;
- certificate issuance/renewal;
- OpenAI Custom GPT domain verification;
- public Internet acceptance from outside the host.
Those remain separate owner/environment gates.
@@ -0,0 +1,120 @@
# Disabled-by-default HTTPS edge for external GPT Online Task access
## Purpose
The internal ingress (`127.0.0.1:18945`) remains loopback-only. This edge is the only component
allowed to become Internet-facing, and only after a separate owner approval.
It forwards only:
```text
POST /v1/read-only-tasks
GET /v1/read-only-tasks/{request_id}
GET /openapi.json
GET /health
```
It never proxies:
```text
18892 canonical Bridge
18940 Task API
WriterLease
commit
verification
release
tag
baseline
arbitrary URLs
```
## Default state
The systemd unit is disabled by policy unless the owner creates:
```text
/etc/aios-gitops/gpt-https-edge.approved
```
The base unit still binds only to:
```text
127.0.0.1:18443
AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=0
```
Non-loopback bind is rejected by code unless an owner-controlled drop-in sets:
```text
AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1
AIOS_GPT_EDGE_HOST=0.0.0.0
AIOS_GPT_EDGE_PUBLIC_DOMAIN=<approved hostname>
```
This repository candidate does not create that drop-in or approval marker.
## TLS
TLS certificate/key are owner-managed systemd credentials:
```text
/etc/aios-gitops/gpt-https-edge-tls.crt
/etc/aios-gitops/gpt-https-edge-tls.key
```
The service requires TLS 1.2 or newer, disables TLS compression, and never logs or returns key
material.
Certificate/key issuance and renewal are external owner responsibilities until a separately
governed ACME workflow is implemented.
## Request protections
The edge provides:
- 64 KiB request cap;
- 256 KiB upstream response cap;
- JSON validation before proxying;
- fixed upstream `http://127.0.0.1:18945`;
- Authorization-only header forwarding;
- no Cookie/X-Forwarded-* trust;
- route/method allowlist;
- per-source-IP in-memory rate limiting;
- bounded concurrent request handling;
- optional exact Host-header enforcement;
- HSTS / no-store / nosniff / no-referrer response headers.
The rate limiter is intentionally simple and local. Multi-node/global abuse protection is outside
the current one-host architecture.
## Authentication
The HTTPS edge does not create a new authentication scheme. It forwards the external Bearer
credential to the loopback GPT ingress, which verifies `gpt_ingress_token`.
The edge never receives the AIOS control-plane token.
## Deployment boundary
Public activation is a high-risk network operation and requires explicit owner approval. Before
activation verify:
```text
host-side GPT ingress E2E=VERIFIED
owner ingress token provisioned
TLS certificate/key provisioned
public hostname owned and DNS configured
firewall/network exposure reviewed
explicit public-bind approval recorded
rollback plan available
```
Only after those gates may an owner-authorized runtime drop-in bind this service publicly.
Until then:
```text
GPT_HTTPS_EDGE_IMPLEMENTATION=candidate
PUBLIC_NETWORK_EXPOSURE=NONE
PUBLIC_CUSTOM_GPT_DIRECT_AI0S_ACCESS=NOT_VERIFIED
```
@@ -0,0 +1,87 @@
# Release candidate capability contract remediation
## Problem
The release-preparation worker can be correctly registered with the raw capability:
```text
release:candidate
```
while the canonical Task API matcher still rejects the worker.
Before this repair, the capability matrix is:
```text
worker raw = ["release:candidate"]
worker effective= []
task required = ["exact:release:candidate"]
eligible = false
```
The Task therefore remains `QUEUED`, `attempt=0`, and cannot be claimed even though both the Task and
worker use the same literal capability.
## Root cause
`aios_capability_contract.py` is deliberately fail-closed:
- task-side unknown names become `exact:<name>`;
- worker-side unknown names are discarded;
- wildcard worker capabilities are ignored.
`release:candidate` was absent from both canonical alias tables.
## Repair
Add exactly one semantic capability to both vocabularies:
```text
release:candidate → release:candidate
```
Do not add:
```text
release:*
submitOwnerApproval
startDeployment
tag_publish
baseline_registry_finalize
runtime_switch
```
Unknown names and wildcards must continue to fail closed.
## Runtime recovery
After the governed remediation candidate is officially verified:
1. install the candidate capability-contract module as a higher-precedence Task API overlay;
2. restart only `aios-agent-task-api.service`;
3. require `/health` success;
4. require the fixed existing release Task to remain the same Task/idempotency;
5. let the already-active dedicated release worker poll and claim it automatically;
6. require heartbeat, trusted-builder result and official verification;
7. do not manually claim or create a replacement live Task.
The fixed V2R2C2 live Task may complete as the acceptance candidate it was originally bound to.
Because this remediation is a later source commit, the first release candidate that includes the
matcher repair should use the next immutable version (for example `V2R2C3`) rather than rewriting an
already-created V2R2C2 candidate.
## Governance boundary
This repair affects only Task eligibility. It grants no credential and performs no release action by
itself.
The dedicated release worker still requires its independent `release:candidate` credential, and the
following remain separate:
```text
Owner signed approval
deployment
tag publication
baseline finalization
runtime switch
```
@@ -0,0 +1,142 @@
# Release Preparation Online Agent
## Purpose
This worker closes the release-preparation execution gap without widening the ChatGPT connector
identity.
The dedicated Agent identity is:
```text
online-release-worker-cn-network-mkt-20260819
```
Its only advertised task capability is:
```text
release:candidate
```
The worker may create an immutable release candidate, but it never submits Owner approval and never
starts deployment, publishes a tag, finalizes a baseline, switches runtime or writes Git.
## Credentials
The worker uses two independently scoped credentials:
```text
agent_token
release_candidate_token
```
`agent_token` is the Online Agent registration credential. It must be encrypted into systemd's
credential store and must never be written to Git, Task metadata, evidence or logs.
`release_candidate_token` is provisioned during Owner-authorized runtime activation. Only its SHA-256 record is stored in the release API token registry; the plaintext is immediately encrypted into systemd credstore and the temporary plaintext is removed. Its only allowed scope is:
```text
release:candidate
```
It is explicitly not an Owner signing key and must not grant:
```text
submitOwnerApproval
startDeployment
tag_publish
baseline_registry_finalize
runtime_switch
```
The unit reads both credentials through systemd `LoadCredentialEncrypted`; neither plaintext token is kept in the unit, repository, Task metadata, evidence, or logs.
## Task contract
The worker will only claim Tasks whose exact required capability set is:
```json
["release:candidate"]
```
The Task must provide a bounded `constraints.release_candidate` payload (metadata is accepted only for compatibility):
```json
{
"candidate_id": "...",
"source_commit": "<40 hex>",
"version": "V2R2C2",
"projection_id": "projection-..."
}
```
No caller-supplied URL, method, headers, command, shell, environment, credential path, deployment
operation or arbitrary release route is accepted.
The release API base is host configuration, not Task input. The only release mutation invoked by the
worker is:
```text
POST /createReleaseCandidate
```
## Lifecycle
```text
poll
→ claim
→ heartbeat
→ createReleaseCandidate
→ result submit
→ official verifier
```
The result declares:
```text
writer_lease=NONE
changed_files=[]
direct_main_write=false
production_write=false
tag_write=false
baseline_write=false
runtime_switch=false
owner_approval_submitted=false
```
## Remaining high-risk gate
Successful release-candidate creation does not authorize Production.
The next stage remains:
```text
release candidate
→ immutable approval envelope
→ Owner signature
→ startDeployment
```
That separation is mandatory even when release preparation is fully unattended.
## Unattended availability
The release worker has its own systemd watchdog rather than modifying the shared worker-availability
guard contract.
Every successful Task poll and Task heartbeat emits:
```text
WATCHDOG=1
```
and the service uses:
```text
Type=notify
WatchdogSec=90
Restart=always
```
A process that is technically alive but no longer polling/heartbeating therefore gets restarted by
systemd. The worker also writes `worker-liveness.json` for operator diagnostics.
+252
View File
@@ -0,0 +1,252 @@
from __future__ import annotations
from datetime import datetime, timezone
import json
from pathlib import Path
import subprocess
import sys
import zipfile
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
import aios_all_modified_files_packager as packager
def init_repo(tmp_path: Path) -> tuple[Path, Path, str]:
bare=tmp_path/"repo.git"
work=tmp_path/"work"
subprocess.run(["git","init","--bare",str(bare)],check=True,capture_output=True)
subprocess.run(["git","clone",str(bare),str(work)],check=True,capture_output=True)
subprocess.run(["git","-C",str(work),"config","user.email","test@example.invalid"],check=True)
subprocess.run(["git","-C",str(work),"config","user.name","AIOS Test"],check=True)
(work/"README.md").write_text("base\n",encoding="utf-8")
subprocess.run(["git","-C",str(work),"add","README.md"],check=True)
subprocess.run(["git","-C",str(work),"commit","-m","base"],check=True,capture_output=True)
subprocess.run(["git","-C",str(work),"push","origin","HEAD:refs/heads/main"],check=True,capture_output=True)
base=subprocess.check_output(["git","-C",str(work),"rev-parse","HEAD"],text=True).strip()
return bare,work,base
def commit(work: Path, message: str) -> str:
subprocess.run(["git","-C",str(work),"add","-A"],check=True)
subprocess.run(["git","-C",str(work),"commit","-m",message],check=True,capture_output=True)
candidate=subprocess.check_output(["git","-C",str(work),"rev-parse","HEAD"],text=True).strip()
subprocess.run(["git","-C",str(work),"push","origin","HEAD:refs/heads/main"],check=True,capture_output=True)
return candidate
def real_pem_header() -> bytes:
return b"-----" + b"BEGIN " + b"PRIVATE KEY" + b"-----"
def bearer_value() -> bytes:
return b"Authorization" + b": Bearer " + (b"a"*40)
def key_value_secret() -> bytes:
return b"api_key" + b"=" + (b"x"*24)
def test_plain_private_key_words_are_not_secret_findings():
for value in (
b"BEGIN PRIVATE KEY",
b"PRIVATE KEY",
b"PEM private-key header",
b"assert 'BEGIN PRIVATE KEY' in message",
):
assert packager.secret_finding(value) is None
def test_complete_pem_private_key_header_is_rejected():
assert packager.secret_finding(real_pem_header())=="PEM_PRIVATE_KEY"
assert packager.secret_finding(
b"-----BEGIN OPENSSH " + b"PRIVATE KEY-----"
)=="PEM_PRIVATE_KEY"
def test_bearer_and_key_value_secret_detection_remains_fail_closed():
assert packager.secret_finding(bearer_value())=="AUTHORIZATION_BEARER"
assert packager.secret_finding(key_value_secret())=="KEY_VALUE_SECRET"
def test_short_or_descriptive_secret_words_do_not_false_positive():
assert packager.secret_finding(b"secret=short") is None
assert packager.secret_finding(b"password field is validated") is None
assert packager.secret_finding(b"api_key example placeholder") is None
def test_end_to_end_bundle_accepts_literal_begin_private_key_text(tmp_path):
bare,work,base=init_repo(tmp_path)
(work/"tests").mkdir()
(work/"tests"/"test_text.py").write_text(
'ERROR = "BEGIN PRIVATE KEY is not a complete PEM header"\n',
encoding="utf-8",
)
(work/"docs").mkdir()
(work/"docs"/"note.md").write_text("Scanner regression.\n",encoding="utf-8")
candidate=commit(work,"candidate")
extra=tmp_path/"execution"
extra.mkdir()
(extra/"evidence.json").write_text('{"status":"PASS"}\n',encoding="utf-8")
out=tmp_path/"out"
summary=packager.build_bundle(
git_dir=bare,
base=base,
candidate=candidate,
task_id="r42-packager-test",
extra_dir=extra,
output_root=out,
now=datetime(2026,8,19,tzinfo=timezone.utc),
)
assert summary["changed_file_count"]==2
assert summary["execution_file_count"]==1
assert summary["repository_document_count"]==1
assert summary["execution_document_count"]==0
bundle_path=Path(summary["bundle_path"])
assert bundle_path.is_file()
assert packager.sha256(bundle_path.read_bytes())==summary["bundle_sha256"]
with zipfile.ZipFile(bundle_path) as archive:
names=set(archive.namelist())
assert "repo/after/tests/test_text.py" in names
assert "repo/after/docs/note.md" in names
assert "execution/evidence.json" in names
assert "documents-index.json" in names
assert "all-modified-files-manifest.json" in names
assert "checksums.sha256" in names
manifest=json.loads(archive.read("all-modified-files-manifest.json"))
assert manifest["schema"]=="aios.all-modified-files.bundle.v1"
assert manifest["candidate_oid"]==candidate
packager.verify_checksums(out/"bundle")
def test_end_to_end_bundle_rejects_real_pem_header(tmp_path):
bare,work,base=init_repo(tmp_path)
(work/"credential.txt").write_bytes(
real_pem_header()+b"\nnot-a-real-key-body\n"
)
candidate=commit(work,"credential-like candidate")
with pytest.raises(
packager.PackagerError,
match="PEM_PRIVATE_KEY:after:credential.txt",
):
packager.build_bundle(
git_dir=bare,
base=base,
candidate=candidate,
task_id="r42-packager-pem-reject",
output_root=tmp_path/"out",
)
def test_execution_evidence_bearer_is_rejected(tmp_path):
bare,work,base=init_repo(tmp_path)
(work/"normal.txt").write_text("normal\n",encoding="utf-8")
candidate=commit(work,"normal")
extra=tmp_path/"execution"
extra.mkdir()
(extra/"unsafe.txt").write_bytes(bearer_value()+b"\n")
with pytest.raises(
packager.PackagerError,
match="AUTHORIZATION_BEARER:execution:unsafe.txt",
):
packager.build_bundle(
git_dir=bare,
base=base,
candidate=candidate,
task_id="r42-packager-bearer-reject",
extra_dir=extra,
output_root=tmp_path/"out",
)
def test_execution_symlink_is_rejected(tmp_path):
bare,work,base=init_repo(tmp_path)
(work/"normal.txt").write_text("normal\n",encoding="utf-8")
candidate=commit(work,"normal")
extra=tmp_path/"execution"
extra.mkdir()
target=extra/"target.txt"
target.write_text("safe\n",encoding="utf-8")
(extra/"link.txt").symlink_to(target)
with pytest.raises(
packager.PackagerError,
match="EXTRA_FILE_SYMLINK_REJECTED",
):
packager.build_bundle(
git_dir=bare,
base=base,
candidate=candidate,
task_id="r42-packager-symlink-reject",
extra_dir=extra,
output_root=tmp_path/"out",
)
def test_non_ancestor_lineage_is_rejected(tmp_path):
bare,work,base=init_repo(tmp_path)
(work/"one.txt").write_text("one\n",encoding="utf-8")
first=commit(work,"first")
subprocess.run(["git","-C",str(work),"checkout","--detach",base],check=True,capture_output=True)
(work/"other.txt").write_text("other\n",encoding="utf-8")
subprocess.run(["git","-C",str(work),"add","other.txt"],check=True)
subprocess.run(["git","-C",str(work),"commit","-m","other"],check=True,capture_output=True)
other=subprocess.check_output(["git","-C",str(work),"rev-parse","HEAD"],text=True).strip()
subprocess.run(
["git","-C",str(work),"push","origin","HEAD:refs/heads/other"],
check=True,
capture_output=True,
)
with pytest.raises(packager.PackagerError,match="LINEAGE_INVALID"):
packager.validate_lineage(bare,first,other)
def test_bundle_checksum_closure_detects_tamper(tmp_path):
bare,work,base=init_repo(tmp_path)
(work/"normal.txt").write_text("normal\n",encoding="utf-8")
candidate=commit(work,"normal")
out=tmp_path/"out"
packager.build_bundle(
git_dir=bare,
base=base,
candidate=candidate,
task_id="r42-packager-checksum",
output_root=out,
)
(out/"bundle"/"repo"/"after"/"normal.txt").write_text(
"tampered\n",
encoding="utf-8",
)
with pytest.raises(
packager.PackagerError,
match="CHECKSUM_CLOSURE_INVALID",
):
packager.verify_checksums(out/"bundle")
def test_source_does_not_contain_complete_pem_header_literal():
source=(ROOT/"active-runtime/app/aios_all_modified_files_packager.py").read_bytes()
tests=(ROOT/"tests/test_all_modified_files_packager.py").read_bytes()
complete=real_pem_header()
assert complete not in source
assert complete not in tests
def test_document_path_contract():
assert packager.is_document_path("docs/spec.md")
assert packager.is_document_path("README")
assert packager.is_document_path("CHANGELOG.md")
assert not packager.is_document_path("active-runtime/app/main.py")
+157
View File
@@ -0,0 +1,157 @@
from __future__ import annotations
import sys
from pathlib import Path
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
import aios_autonomous_control_plane as acp
import aios_capability_contract as contract
import aios_release_online_agent_worker as worker
def _ready_probes():
return {
"credential_projection": lambda: True,
"trusted_builder": lambda: True,
"authenticated_probe": lambda: True,
"release_api_health": lambda: True,
"worker_liveness": lambda: True,
}
def _task():
return {
"task_id":"r42-autonomous-release-admission-test",
"required_capabilities":["release:candidate"],
"constraints":{
"release_candidate":{
"candidate_id":"r42-release-candidate-v2r2c9-test",
"source_commit":"1"*40,
"version":"V2R2C9",
"projection_id":"projection-"+"a"*32,
}
},
}
def test_capability_contract_registry_is_closed_and_release_is_versioned():
status=contract.validate_capability_contract_registry()
assert status["valid"] is True
release=contract.capability_contract("release:candidate")
assert release is not None
assert release.min_version==1
assert release.max_version==1
assert release.permissions==("prepare_release",)
assert "credential_projection" in release.dependencies
assert "trusted_builder" in release.dependencies
assert "authenticated_probe" in release.health_requirements
assert "production_deploy" in release.forbidden
assert "runtime_switch" in release.forbidden
def test_unknown_agent_capability_cannot_register_or_match():
assert contract.registered_capability("release:future",worker=True) is False
assert contract.registered_capability("release:*",worker=True) is False
assert contract.has_capabilities(["release:future"],["release:future"]) is False
def test_runtime_admission_requires_all_release_execution_readiness():
decision=acp.runtime_admission(
worker_capabilities=["release:candidate"],
required_capabilities=["release:candidate"],
requested_operations=["prepare_release"],
probes=_ready_probes(),
)
assert decision.admitted is True
assert decision.state=="EXECUTION_READY"
def test_runtime_admission_blocks_credential_failure_before_execution():
probes=_ready_probes()
probes["credential_projection"]=lambda:False
probes["authenticated_probe"]=lambda:False
decision=acp.runtime_admission(
worker_capabilities=["release:candidate"],
required_capabilities=["release:candidate"],
requested_operations=["prepare_release"],
probes=probes,
)
assert decision.admitted is False
assert decision.state=="EXECUTION_READY_FAILED"
assert decision.failure_class==acp.FailureClass.CREDENTIAL_PROJECTION_ERROR
assert decision.repair_level==acp.RepairLevel.AUTO_REPAIR_TASK
def test_forbidden_release_operation_requires_owner_boundary():
decision=acp.runtime_admission(
worker_capabilities=["release:candidate"],
required_capabilities=["release:candidate"],
requested_operations=["production_deploy"],
probes=_ready_probes(),
)
assert decision.admitted is False
assert decision.failure_class==acp.FailureClass.POLICY_DENIED
assert decision.repair_level==acp.RepairLevel.OWNER_APPROVAL
@pytest.mark.parametrize(
"error,phase,expected,level,consumes",
[
("Permission denied: gitops.tokens","release",acp.FailureClass.CREDENTIAL_PROJECTION_ERROR,acp.RepairLevel.AUTO_REPAIR_TASK,False),
("HTTP_TRANSPORT_UNAVAILABLE","network",acp.FailureClass.DEPENDENCY_UNAVAILABLE,acp.RepairLevel.AUTO_LOCAL,False),
("TASK_CAPABILITY_SCOPE_INVALID","validation",acp.FailureClass.CAPABILITY_CONTRACT_ERROR,acp.RepairLevel.AUTO_REPAIR_TASK,False),
("lease lost","heartbeat",acp.FailureClass.LEASE_LOST,acp.RepairLevel.AUTO_LOCAL,False),
("business permanent invalid release","release",acp.FailureClass.BUSINESS_PERMANENT,acp.RepairLevel.MANUAL,True),
],
)
def test_failure_classification_drives_repair_policy(error,phase,expected,level,consumes):
failure=acp.classify_failure(error,phase=phase)
assert failure==expected
repair=acp.repair_for_failure(failure)
assert repair.level==level
assert repair.consumes_business_attempt is consumes
def test_release_worker_does_not_claim_when_admission_is_not_ready(monkeypatch):
calls=[]
def fake_http(base,method,route,token,body=None,timeout=30):
calls.append((base,method,route,body))
if route=="/health":
return 200,{"status":"ok"}
if route=="/createReleaseCandidate" and body=={}:
return 500,{"error":"Permission denied: gitops.tokens"}
if route.endswith("/claim"):
raise AssertionError("claim must not happen before EXECUTION_READY")
raise AssertionError(route)
events=[]
monkeypatch.setattr(worker,"http_json",fake_http)
monkeypatch.setattr(worker,"audit",lambda event,**fields:events.append((event,fields)))
monkeypatch.setattr(worker,"write_liveness",lambda *args,**kwargs:None)
assert worker.execute_task(
_task(),
agent_token="a"*32,
release_token="r"*32,
) is False
assert not any(route.endswith("/claim") for _,_,route,_ in calls)
blocked=[item for item in events if item[0]=="TASK_ADMISSION_BLOCKED"]
assert len(blocked)==1
assert blocked[0][1]["admission"]["state"]=="EXECUTION_READY_FAILED"
def test_system_invariant_snapshot_is_degraded_when_dependency_is_missing():
probes=_ready_probes()
probes["trusted_builder"]=lambda:False
snapshot=acp.system_invariant_snapshot(
worker_capabilities=["release:candidate"],
required_capabilities=["release:candidate"],
probes=probes,
)
assert snapshot["status"]=="DEGRADED"
assert snapshot["execution_ready"] is False
+627
View File
@@ -0,0 +1,627 @@
from __future__ import annotations
from datetime import datetime, timezone
import json
from pathlib import Path
import sqlite3
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
import aios_durable_owner_authorization as auth
NOW=datetime(2026,8,19,6,0,0,tzinfo=timezone.utc)
ACCOUNT="chatgpt-governed-implementer-singlehost"
def write_root_file(path: Path, value: object) -> Path:
path.parent.mkdir(parents=True,exist_ok=True)
path.write_text(json.dumps(value,sort_keys=True)+"\n",encoding="utf-8")
path.chmod(0o600)
return path
def policy_value(**overrides):
value={
"schema":auth.POLICY_SCHEMA,
"grant_id":"owner-permanent-preparation-20260819",
"owner_identity":"owner-singlehost",
"account":ACCOUNT,
"capabilities":[
"governed-task:project",
"release:candidate",
],
"issued_at":"2026-08-19T05:00:00Z",
"not_before":"2026-08-19T05:00:00Z",
"persistent_until_revoked":True,
"constraints":{
"repository":"aios/aios",
"allowed_target_prefix":"agent/",
"release_version_prefix":"V2R2C",
"max_task_risk":"engineering",
},
}
value.update(overrides)
return value
def load_policy(tmp_path,monkeypatch,**overrides):
path=write_root_file(tmp_path/"policy.json",policy_value(**overrides))
revocations=write_root_file(
tmp_path/"revocations.json",
{"revoked_grant_ids":[]},
)
monkeypatch.setattr(auth,"validate_root_file",lambda path,**kwargs:path)
return auth.DurablePolicy.load(
path,
revocations,
now=NOW,
),path,revocations
def governed_request(**overrides):
value={
"task_id":"r42-owner-auth-projection-test-20260819",
"repository":"aios/aios",
"base_oid":"8"*40,
"expected_branch_oid":"8"*40,
"target_branch":"agent/r42-owner-auth-projection-test-20260819",
"generation":1,
"risk":"engineering",
"description":"project existing governed task",
"idempotency_key":"binding-owner-auth-projection-v1",
"branch_mode":"CREATE_NEW",
"priority":100,
"max_attempts":3,
"require_online_task_projection":True,
"online_task":{
"goal":"Analyze source read-only.",
"instructions":"Use source capture first.",
"success_criteria":[
{"id":"source","description":"Return source snapshot."}
],
"constraints":{"repository_mutation":"forbidden"},
"required_capabilities":["git.read","git.test"],
"priority":100,
"max_attempts":3,
},
"execution":{
"action":"source_capture",
"adapter":"source_capture",
"capability_scope":["git.read","git.test"],
"source_signals":["durable authorization projection"],
},
}
value.update(overrides)
return value
def release_request(**overrides):
value={
"candidate_id":"r42-pf1-release-candidate-20260819",
"source_commit":"8"*40,
"version":"V2R2C2",
}
value.update(overrides)
return value
class FakeUpstream:
def __init__(self):
self.project_calls=[]
def project_governed_task(self,body):
self.project_calls.append(dict(body))
return 200,{
"task_id":body["task_id"],
"job_id":"job-"+"a"*32,
"state":"BOUND",
}
def broker(tmp_path,monkeypatch):
policy,path,revocations=load_policy(tmp_path,monkeypatch)
store=auth.ReceiptStore(tmp_path/"receipts.sqlite3")
upstream=FakeUpstream()
instance=auth.Broker(
policy_path=path,
revocation_path=revocations,
store=store,
upstream=upstream,
expected_account=ACCOUNT,
)
monkeypatch.setattr(
auth.DurablePolicy,
"load",
classmethod(lambda cls,*args,**kwargs:policy),
)
return instance,upstream,store,policy
def test_allowed_persistent_capabilities_are_only_preparation_scopes():
assert auth.ALLOWED_PERSISTENT_CAPABILITIES=={
"governed-task:project",
"release:candidate",
}
assert auth.ALLOWED_PERSISTENT_CAPABILITIES.isdisjoint(
auth.FORBIDDEN_PERSISTENT_CAPABILITIES
)
@pytest.mark.parametrize(
"capability",
sorted(auth.FORBIDDEN_PERSISTENT_CAPABILITIES),
)
def test_policy_rejects_high_risk_persistent_capability(
tmp_path,monkeypatch,capability
):
bad=policy_value(
capabilities=["governed-task:project",capability]
)
path=write_root_file(tmp_path/"policy.json",bad)
revocations=write_root_file(
tmp_path/"revocations.json",
{"revoked_grant_ids":[]},
)
monkeypatch.setattr(auth,"validate_root_file",lambda path,**kwargs:path)
with pytest.raises(
auth.AuthorizationError,
match="POLICY_CAPABILITY_NOT_PERSISTABLE",
):
auth.DurablePolicy.load(path,revocations,now=NOW)
def test_policy_is_persistent_until_revoked(tmp_path,monkeypatch):
policy,_,_=load_policy(tmp_path,monkeypatch)
assert policy.persistent_until_revoked is True
assert policy.account==ACCOUNT
assert policy.capabilities=={
"governed-task:project",
"release:candidate",
}
def test_revocation_fails_closed(tmp_path,monkeypatch):
path=write_root_file(tmp_path/"policy.json",policy_value())
revocations=write_root_file(
tmp_path/"revocations.json",
{"revoked_grant_ids":["owner-permanent-preparation-20260819"]},
)
monkeypatch.setattr(auth,"validate_root_file",lambda path,**kwargs:path)
with pytest.raises(auth.AuthorizationError,match="OWNER_GRANT_REVOKED"):
auth.DurablePolicy.load(path,revocations,now=NOW)
def test_future_policy_is_not_active(tmp_path,monkeypatch):
with pytest.raises(auth.AuthorizationError,match="POLICY_NOT_ACTIVE"):
load_policy(
tmp_path,
monkeypatch,
issued_at="2026-08-20T05:00:00Z",
not_before="2026-08-20T05:00:00Z",
)
def test_governed_projection_request_is_tightly_constrained(
tmp_path,monkeypatch
):
policy,_,_=load_policy(tmp_path,monkeypatch)
decision=policy.authorize(
"governed-task:project",
governed_request(),
)
assert decision["decision"]=="ALLOW"
assert decision["persistent_until_revoked"] is True
bad_caps=governed_request()
bad_caps["online_task"]={
**bad_caps["online_task"],
"required_capabilities":["release:candidate"],
}
bad_action=governed_request()
bad_action["execution"]={
**bad_action["execution"],
"action":"deterministic",
"adapter":"deterministic",
}
bad_scope=governed_request()
bad_scope["execution"]={
**bad_scope["execution"],
"capability_scope":["git.read"],
}
for bad in (
governed_request(repository="other/repo"),
governed_request(target_branch="main"),
governed_request(generation=2),
governed_request(risk="high"),
governed_request(expected_branch_oid="9"*40),
governed_request(branch_mode="CONTINUE_EXISTING"),
governed_request(priority=101),
governed_request(max_attempts=4),
governed_request(require_online_task_projection=False),
bad_caps,
bad_action,
bad_scope,
):
with pytest.raises(auth.AuthorizationError):
policy.authorize("governed-task:project",bad)
def test_write_capable_projection_is_allowed_only_as_source_capture_placeholder(
tmp_path,monkeypatch
):
policy,_,_=load_policy(tmp_path,monkeypatch)
request=governed_request()
request["online_task"]={
**request["online_task"],
"required_capabilities":[
"git.read",
"git.test",
"governed.write.branch",
"governed.write.file",
],
}
request["execution"]={
**request["execution"],
"capability_scope":[
"git.read",
"git.test",
"governed.write.branch",
"governed.write.file",
],
}
decision=policy.authorize("governed-task:project",request)
assert decision["decision"]=="ALLOW"
assert request["execution"]["action"]=="source_capture"
def test_release_candidate_request_is_tightly_constrained(
tmp_path,monkeypatch
):
policy,_,_=load_policy(tmp_path,monkeypatch)
decision=policy.authorize("release:candidate",release_request())
assert decision["decision"]=="ALLOW"
with pytest.raises(
auth.AuthorizationError,
match="RELEASE_VERSION_NOT_AUTHORIZED",
):
policy.authorize(
"release:candidate",
release_request(version="V3R1C1"),
)
with pytest.raises(
auth.AuthorizationError,
match="RELEASE_CANDIDATE_FIELDS_INVALID",
):
policy.authorize(
"release:candidate",
{**release_request(),"execute":True},
)
def test_governed_task_projection_executes_canonical_upstream(
tmp_path,monkeypatch
):
instance,upstream,store,_=broker(tmp_path,monkeypatch)
request=governed_request()
response=instance.execute(
"governed-task:project",
request,
idempotency_key="owner-auth-projection-operation-0001",
)
assert len(upstream.project_calls)==1
assert upstream.project_calls[0]==request
assert response["operation_state"]=="EXECUTED"
assert response["external_capability_projection_required"] is False
assert response["separate_owner_release_approval_still_required"] is True
def test_release_candidate_authorization_does_not_invent_builder_upstream(
tmp_path,monkeypatch
):
instance,upstream,store,_=broker(tmp_path,monkeypatch)
response=instance.execute(
"release:candidate",
release_request(),
idempotency_key="owner-auth-release-candidate-operation-0001",
)
assert upstream.project_calls==[]
assert response["operation_state"]==(
"AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION"
)
assert response["upstream_status"] is None
assert response["upstream_response"] is None
assert response["external_capability_projection_required"] is True
assert response["projection_id"].startswith("projection-")
assert response["projection_state"]=="PENDING_EXTERNAL_CONNECTOR"
assert response["separate_owner_release_approval_still_required"] is True
def test_release_authorization_creates_durable_projection_outbox(
tmp_path,monkeypatch
):
instance,_,store,policy=broker(tmp_path,monkeypatch)
request=release_request()
response=instance.execute(
"release:candidate",
request,
idempotency_key="owner-auth-release-outbox-0001",
)
projection=store.get_projection(response["projection_id"])
assert projection is not None
assert projection["schema"]==auth.PROJECTION_SCHEMA
assert projection["state"]=="PENDING_EXTERNAL_CONNECTOR"
assert projection["grant_id"]==policy.grant_id
assert projection["account"]==ACCOUNT
assert projection["capability"]=="release:candidate"
assert projection["request"]==request
assert projection["separate_owner_release_approval_still_required"] is True
listed=instance.projections()
assert listed["schema"]==auth.PROJECTION_LIST_SCHEMA
assert listed["count"]==1
assert listed["items"][0]["projection_id"]==response["projection_id"]
assert listed["external_connector_ack_supported"] is False
def test_existing_pre_outbox_release_receipt_is_backfilled_on_exact_retry(
tmp_path,monkeypatch
):
instance,_,store,policy=broker(tmp_path,monkeypatch)
request=release_request()
decision=policy.authorize("release:candidate",request)
key="owner-auth-release-pre-outbox-0001"
legacy_response={
"schema":auth.RECEIPT_SCHEMA,
"grant_id":policy.grant_id,
"owner_identity":policy.owner_identity,
"account":policy.account,
"capability":"release:candidate",
"policy_sha256":policy.policy_sha256,
"request_sha256":decision["request_sha256"],
"idempotency_key":key,
"operation_state":"AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION",
"upstream_status":None,
"upstream_response":None,
"external_capability_projection_required":True,
"separate_owner_release_approval_still_required":True,
}
store.put(
key,
"release:candidate",
decision["request_sha256"],
legacy_response,
)
response=instance.execute(
"release:candidate",
request,
idempotency_key=key,
)
assert response["projection_id"].startswith("projection-")
assert response["projection_state"]=="PENDING_EXTERNAL_CONNECTOR"
assert store.get_projection(response["projection_id"]) is not None
persisted=store.get(key)["response"]
assert persisted["projection_id"]==response["projection_id"]
def test_projection_id_is_deterministic_and_exact_retry_does_not_duplicate(
tmp_path,monkeypatch
):
instance,_,store,_=broker(tmp_path,monkeypatch)
key="owner-auth-release-outbox-0002"
first=instance.execute("release:candidate",release_request(),idempotency_key=key)
second=instance.execute("release:candidate",release_request(),idempotency_key=key)
assert second==first
con=sqlite3.connect(store.path)
rows=con.execute("select projection_id from projection_outbox").fetchall()
assert rows==[(first["projection_id"],)]
def test_projection_enqueue_recovers_crash_between_outbox_and_receipt(
tmp_path,monkeypatch
):
instance,_,store,policy=broker(tmp_path,monkeypatch)
request=release_request()
decision=policy.authorize("release:candidate",request)
key="owner-auth-release-outbox-crash-0001"
first=store.enqueue_projection(
idempotency_key=key,
grant_id=policy.grant_id,
account=policy.account,
capability="release:candidate",
policy_sha256=policy.policy_sha256,
request_sha256=decision["request_sha256"],
request=request,
)
# Simulate a later retry after the process crashed before receipt creation.
second=store.enqueue_projection(
idempotency_key=key,
grant_id=policy.grant_id,
account=policy.account,
capability="release:candidate",
policy_sha256=policy.policy_sha256,
request_sha256=decision["request_sha256"],
request=request,
)
assert second==first
def test_projection_idempotency_conflict_is_fail_closed(
tmp_path,monkeypatch
):
instance,_,store,policy=broker(tmp_path,monkeypatch)
first=release_request()
first_decision=policy.authorize("release:candidate",first)
key="owner-auth-release-outbox-conflict-0001"
store.enqueue_projection(
idempotency_key=key,
grant_id=policy.grant_id,
account=policy.account,
capability="release:candidate",
policy_sha256=policy.policy_sha256,
request_sha256=first_decision["request_sha256"],
request=first,
)
second=release_request(version="V2R2C3")
second_decision=policy.authorize("release:candidate",second)
with pytest.raises(auth.AuthorizationError,match="PROJECTION_IDEMPOTENCY_CONFLICT"):
store.enqueue_projection(
idempotency_key=key,
grant_id=policy.grant_id,
account=policy.account,
capability="release:candidate",
policy_sha256=policy.policy_sha256,
request_sha256=second_decision["request_sha256"],
request=second,
)
def test_revocation_blocks_projection_enumeration(tmp_path,monkeypatch):
instance,_,store,policy=broker(tmp_path,monkeypatch)
response=instance.execute(
"release:candidate",
release_request(),
idempotency_key="owner-auth-release-outbox-revoke-0001",
)
# Restore real policy loading and write a revocation for this grant.
monkeypatch.undo()
revocations=instance.revocation_path
revocations.write_text(
json.dumps({"revoked_grant_ids":[policy.grant_id]})+"\n",
encoding="utf-8",
)
revocations.chmod(0o600)
monkeypatch.setattr(auth,"validate_root_file",lambda path,**kwargs:path)
with pytest.raises(auth.AuthorizationError,match="OWNER_GRANT_REVOKED"):
instance.projections()
# Evidence is retained; revocation does not erase history.
assert store.get_projection(response["projection_id"]) is not None
def test_projection_outbox_contains_no_credentials(tmp_path,monkeypatch):
instance,_,store,_=broker(tmp_path,monkeypatch)
instance.execute(
"release:candidate",
release_request(),
idempotency_key="owner-auth-release-outbox-schema-0001",
)
con=sqlite3.connect(store.path)
columns=[row[1] for row in con.execute("pragma table_info(projection_outbox)")]
joined=" ".join(columns).lower()
for forbidden in ("token","credential","private_key","secret"):
assert forbidden not in joined
payload=con.execute("select payload_json from projection_outbox").fetchone()[0]
assert "Bearer " not in payload
assert "submitOwnerApproval" not in payload
assert "startDeployment" not in payload
def test_exact_retry_returns_same_receipt_without_second_projection(
tmp_path,monkeypatch
):
instance,upstream,_,_=broker(tmp_path,monkeypatch)
request=governed_request()
first=instance.execute(
"governed-task:project",
request,
idempotency_key="owner-auth-projection-operation-0002",
)
second=instance.execute(
"governed-task:project",
request,
idempotency_key="owner-auth-projection-operation-0002",
)
assert second==first
assert len(upstream.project_calls)==1
def test_idempotency_conflict_fails_closed(tmp_path,monkeypatch):
instance,_,_,_=broker(tmp_path,monkeypatch)
instance.execute(
"release:candidate",
release_request(),
idempotency_key="owner-auth-release-operation-0002",
)
with pytest.raises(auth.AuthorizationError,match="IDEMPOTENCY_CONFLICT"):
instance.execute(
"release:candidate",
release_request(version="V2R2C3"),
idempotency_key="owner-auth-release-operation-0002",
)
def test_receipt_store_contains_no_owner_policy_or_credentials(
tmp_path,monkeypatch
):
instance,_,store,_=broker(tmp_path,monkeypatch)
instance.execute(
"release:candidate",
release_request(),
idempotency_key="owner-auth-release-operation-0003",
)
con=sqlite3.connect(store.path)
columns=[row[1] for row in con.execute("pragma table_info(receipts)")]
joined=" ".join(columns).lower()
assert "token" not in joined
assert "credential" not in joined
assert "policy_json" not in columns
def test_systemd_broker_uses_unix_socket_and_only_host_control_credential():
unit=(
ROOT/"deploy/systemd/aios-owner-durable-authorization-broker.service"
).read_text()
assert "LoadCredential=agent_task_control_token:" in unit
assert "release_candidate_token" not in unit
assert "18947" not in unit
assert "AIOS_RELEASE_CANDIDATE_API_URL" not in unit
assert "AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock" in unit
assert "Group=aios-executor-exchange" in unit
assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
assert "IPAddressDeny=any" in unit
assert "IPAddressAllow=127.0.0.0/8" in unit
assert "InaccessiblePaths=/data/aios-gitops /home" in unit
def test_source_contains_no_release_builder_endpoint_or_high_risk_execution():
source=(
ROOT/"active-runtime/app/aios_durable_owner_authorization.py"
).read_text()
assert "/createReleaseCandidate" not in source
assert "18947" not in source
assert "ThreadingHTTPServer" not in source
assert "ThreadingUnixStreamServer" in source
assert "startDeployment" in source
assert "tag_publish" in source
assert "baseline_registry_finalize" in source
assert "runtime_switch" in source
assert "AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION" in source
assert "PENDING_EXTERNAL_CONNECTOR" in source
assert "projection_outbox" in source
assert "external_connector_ack_supported" in source
assert "SAFE_PROJECTED_TASK_CAPABILITIES" in source
assert "INITIAL_EXECUTION_ACTION_NOT_AUTHORIZED" in source
def test_docs_avoid_new_broker_bearer_secret():
doc=(
ROOT/"docs/superpowers/specs/2026-08-19-durable-owner-authorization.md"
).read_text()
assert "AF_UNIX" in doc
assert "avoids introducing another long-lived bearer secret" in doc
def test_docs_preserve_separate_owner_release_approval():
doc=(
ROOT/"docs/superpowers/specs/2026-08-19-durable-owner-authorization.md"
).read_text()
assert "persistent until explicitly revoked" in doc.lower()
assert "Owner signature" in doc
assert "AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION" in doc
assert "PENDING_EXTERNAL_CONNECTOR" in doc
+205
View File
@@ -0,0 +1,205 @@
from __future__ import annotations
from email.message import Message
import json
from pathlib import Path
import stat
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
import aios_gpt_https_edge as edge
class Headers(dict):
def get(self,key,default=None):
return super().get(key,default)
def test_default_and_public_bind_require_explicit_owner_approval():
assert edge.public_bind_allowed("127.0.0.1","0")=="127.0.0.1"
assert edge.public_bind_allowed("::1","0")=="::1"
with pytest.raises(RuntimeError,match="OWNER_APPROVAL_REQUIRED"):
edge.public_bind_allowed("0.0.0.0","0")
with pytest.raises(RuntimeError,match="OWNER_APPROVAL_REQUIRED"):
edge.public_bind_allowed("::","")
assert edge.public_bind_allowed("0.0.0.0","1")=="0.0.0.0"
def test_arbitrary_public_bind_address_is_denied_even_with_approval():
with pytest.raises(RuntimeError,match="BIND_ADDRESS_INVALID"):
edge.public_bind_allowed("192.0.2.10","1")
def test_public_domain_validation():
assert edge.validated_public_domain("tasks.example.com")=="tasks.example.com"
assert edge.validated_public_domain("")==""
with pytest.raises(RuntimeError,match="PUBLIC_DOMAIN_INVALID"):
edge.validated_public_domain("https://example.com")
with pytest.raises(RuntimeError,match="PUBLIC_DOMAIN_INVALID"):
edge.validated_public_domain("localhost")
def test_route_allowlist_accepts_only_readonly_api():
assert edge.allowed_route("POST","/v1/read-only-tasks")=="/v1/read-only-tasks"
assert edge.allowed_route("GET","/v1/read-only-tasks/external-request-0001")==(
"/v1/read-only-tasks/external-request-0001"
)
assert edge.allowed_route("GET","/health")=="/health"
assert edge.allowed_route("GET","/openapi.json")=="/openapi.json"
for method,path in [
("POST","/v2/governed-task"),
("GET","/v1/tasks/x"),
("POST","/v1/read-only-tasks/x"),
("DELETE","/v1/read-only-tasks/x"),
]:
with pytest.raises(edge.EdgeError,match="ROUTE_DENY"):
edge.allowed_route(method,path)
def test_query_and_fragment_are_denied():
with pytest.raises(edge.EdgeError,match="QUERY_OR_FRAGMENT_DENY"):
edge.allowed_route("GET","/health?x=1")
with pytest.raises(edge.EdgeError,match="QUERY_OR_FRAGMENT_DENY"):
edge.allowed_route("GET","/openapi.json#x")
def test_forwarded_headers_are_minimal_and_require_bearer():
incoming=Headers({
"Authorization":"Bearer "+"a"*40,
"Content-Type":"application/json; charset=utf-8",
"Cookie":"secret=1",
"X-Forwarded-For":"203.0.113.10",
"X-Api-Key":"do-not-forward",
})
forwarded=edge.forwarded_headers(incoming,body=True)
assert forwarded=={
"Accept":"application/json",
"Authorization":"Bearer "+"a"*40,
"Content-Type":"application/json",
}
with pytest.raises(edge.EdgeError,match="AUTHORIZATION_REQUIRED"):
edge.forwarded_headers(Headers({}),body=False)
def test_invalid_content_type_is_denied():
with pytest.raises(edge.EdgeError,match="CONTENT_TYPE_INVALID"):
edge.forwarded_headers(
Headers({
"Authorization":"Bearer "+"a"*40,
"Content-Type":"text/plain",
}),
body=True,
)
def test_rate_limiter_caps_window_and_allows_expiry():
now=[100.0]
limiter=edge.RateLimiter(window_seconds=60,limit=2,burst=1,now_fn=lambda:now[0])
assert limiter.allowed("client")
assert limiter.allowed("client")
assert limiter.allowed("client")
assert not limiter.allowed("client")
now[0]=161.0
assert limiter.allowed("client")
def test_rate_limiter_client_table_is_bounded():
now=[100.0]
limiter=edge.RateLimiter(
window_seconds=60,
limit=10,
burst=0,
max_clients=2,
now_fn=lambda:now[0],
)
assert limiter.allowed("a")
assert limiter.allowed("b")
assert not limiter.allowed("c")
now[0]=161.0
assert limiter.allowed("c")
assert len(limiter._events) <= 2
def test_audit_source_does_not_log_sensitive_request_material():
source=(ROOT/"active-runtime/app/aios_gpt_https_edge.py").read_text()
assert '"authorization", "body", "token", "cookie", "client_ip"' in source
assert "_client_pseudonym" in source
def test_proxy_upstream_is_fixed():
proxy=edge.Proxy()
assert proxy.upstream=="http://127.0.0.1:18945"
bad=edge.Proxy(upstream="http://127.0.0.1:18892")
with pytest.raises(edge.EdgeError,match="UPSTREAM_DRIFT_DENY"):
bad.request("GET","/health",{"Authorization":"Bearer "+"a"*40},None)
def test_public_bind_requires_approved_domain():
assert edge.validated_edge_identity("127.0.0.1","0","")==("127.0.0.1","")
with pytest.raises(RuntimeError,match="PUBLIC_DOMAIN_REQUIRED"):
edge.validated_edge_identity("0.0.0.0","1","")
assert edge.validated_edge_identity(
"0.0.0.0","1","tasks.example.com"
)==("0.0.0.0","tasks.example.com")
def test_tls_key_requires_no_group_or_other_permissions(tmp_path):
key=tmp_path/"key.pem"
key.write_text("x"*64,encoding="utf-8")
key.chmod(0o600)
assert edge.validate_tls_file(key,key=True)==key
key.chmod(0o644)
with pytest.raises(RuntimeError,match="TLS_KEY_MODE_INVALID"):
edge.validate_tls_file(key,key=True)
def test_tls_symlink_is_denied(tmp_path):
real=tmp_path/"cert.pem"
real.write_text("x"*64,encoding="utf-8")
link=tmp_path/"link.pem"
link.symlink_to(real)
with pytest.raises(RuntimeError,match="TLS_FILE_INVALID"):
edge.validate_tls_file(link,key=False)
def test_systemd_unit_is_disabled_by_owner_approval_condition_and_loopback_default():
unit=(ROOT/"deploy/systemd/aios-gpt-https-edge.service").read_text()
assert "ConditionPathExists=/etc/aios-gitops/gpt-https-edge.approved" in unit
assert "AIOS_GPT_EDGE_HOST=127.0.0.1" in unit
assert "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=0" in unit
assert "AIOS_GPT_EDGE_PORT=18443" in unit
assert "LoadCredential=tls_cert:" in unit
assert "LoadCredential=tls_key:" in unit
assert "InaccessiblePaths=/data/aios-gitops" in unit
assert "InaccessiblePaths=" in unit and "/var/lib/aios-agent-task" in unit
assert "NoNewPrivileges=yes" in unit
assert "ProtectSystem=strict" in unit
def test_systemd_unit_does_not_reference_internal_control_credentials():
unit=(ROOT/"deploy/systemd/aios-gpt-https-edge.service").read_text()
assert "agent_task_control" not in unit
assert "gitea" not in unit.lower()
assert "18892" not in unit
assert "18940" not in unit
def test_edge_source_does_not_proxy_internal_bridge_or_task_api():
source=(ROOT/"active-runtime/app/aios_gpt_https_edge.py").read_text()
assert 'UPSTREAM = "http://127.0.0.1:18945"' in source
assert "18892" not in source
assert "18940" not in source
assert "/v2/governed-task" not in source
assert "/v1/tasks/" not in source
def test_docs_keep_public_activation_separately_approved():
doc=(ROOT/"docs/superpowers/specs/2026-08-19-gpt-https-edge.md").read_text()
assert "explicit owner approval" in doc.lower()
assert "PUBLIC_NETWORK_EXPOSURE=NONE" in doc
assert "PUBLIC_CUSTOM_GPT_DIRECT_AI0S_ACCESS=NOT_VERIFIED" in doc
+362
View File
@@ -0,0 +1,362 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import hashlib
import json
from pathlib import Path
import sys
import subprocess
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
import aios_gpt_https_edge_activation as activation
CANDIDATE="69a51d50c7f6cd9045346d41e338acebe4b6f1e8"
NOW=datetime(2026,8,19,2,0,0,tzinfo=timezone.utc)
def approval_value(cert_sha: str, **overrides):
value={
"schema":"aios.gpt-https-edge.public-approval.v1",
"approval_id":"owner-edge-approval-0001",
"candidate_oid":CANDIDATE,
"domain":"tasks.example.com",
"public_bind":True,
"bind_host":"0.0.0.0",
"port":18443,
"tls_cert_sha256":cert_sha,
"issued_at":"2026-08-19T01:00:00Z",
"not_before":"2026-08-19T01:00:00Z",
"expires_at":"2026-08-20T01:00:00Z",
"owner":"owner@example",
}
value.update(overrides)
return value
def write_owner_file(path: Path, data: bytes, mode=0o600):
path.write_bytes(data)
path.chmod(mode)
return path
def load_approval(tmp_path, monkeypatch, **overrides):
tmp_path.mkdir(parents=True,exist_ok=True)
cert=write_owner_file(tmp_path/"cert.pem",b"certificate-material-"*4,0o600)
value=approval_value(hashlib.sha256(cert.read_bytes()).hexdigest(),**overrides)
approval=write_owner_file(
tmp_path/"approval.json",
(json.dumps(value,sort_keys=True)+"\n").encode(),
0o600,
)
monkeypatch.setattr(
activation,
"validate_root_file",
lambda path, **kwargs: path,
)
return activation.Approval.load(
approval,
candidate_oid=CANDIDATE,
cert_path=cert,
now=NOW,
),approval,cert
class FakeSystemd:
def __init__(self):
self.active_units={activation.INGRESS_UNIT}
self.enabled_units=set()
self.calls=[]
self.exec_start=""
self.environment=""
self.fail_restart=False
def run(self,*args,check=True):
self.calls.append(args)
class Result:
returncode=0
stdout=""
if args[:2]==("restart",activation.EDGE_UNIT):
if self.fail_restart:
raise subprocess.CalledProcessError(1,args)
self.active_units.add(activation.EDGE_UNIT)
elif args[:2]==("stop",activation.EDGE_UNIT):
self.active_units.discard(activation.EDGE_UNIT)
elif args[:2]==("enable",activation.EDGE_UNIT):
self.enabled_units.add(activation.EDGE_UNIT)
elif args[:2]==("disable",activation.EDGE_UNIT):
self.enabled_units.discard(activation.EDGE_UNIT)
return Result()
def active(self,unit):
return unit in self.active_units
def enabled(self,unit):
return unit in self.enabled_units
def show(self,unit,prop):
if prop=="ExecStart":
return self.exec_start
if prop=="Environment":
return self.environment
return ""
def test_approval_binds_candidate_domain_bind_port_and_cert(tmp_path,monkeypatch):
approval,_,cert=load_approval(tmp_path,monkeypatch)
assert approval.candidate_oid==CANDIDATE
assert approval.domain=="tasks.example.com"
assert approval.bind_host=="0.0.0.0"
assert approval.port==18443
assert approval.tls_cert_sha256==hashlib.sha256(cert.read_bytes()).hexdigest()
assert approval.expires_at>approval.not_before
def test_approval_rejects_candidate_mismatch(tmp_path,monkeypatch):
cert=write_owner_file(tmp_path/"cert.pem",b"certificate-material-"*4)
value=approval_value(hashlib.sha256(cert.read_bytes()).hexdigest())
approval=write_owner_file(tmp_path/"approval.json",json.dumps(value).encode())
monkeypatch.setattr(activation,"validate_root_file",lambda path,**kwargs:path)
with pytest.raises(activation.ActivationError,match="CANDIDATE_MISMATCH"):
activation.Approval.load(
approval,
candidate_oid="0"*40,
cert_path=cert,
now=NOW,
)
def test_approval_rejects_cert_hash_mismatch(tmp_path,monkeypatch):
cert=write_owner_file(tmp_path/"cert.pem",b"certificate-material-"*4)
value=approval_value("a"*64)
approval=write_owner_file(tmp_path/"approval.json",json.dumps(value).encode())
monkeypatch.setattr(activation,"validate_root_file",lambda path,**kwargs:path)
with pytest.raises(activation.ActivationError,match="CERT_SHA_MISMATCH"):
activation.Approval.load(
approval,
candidate_oid=CANDIDATE,
cert_path=cert,
now=NOW,
)
def test_approval_rejects_expired_not_yet_valid_and_long_lived(tmp_path,monkeypatch):
with pytest.raises(activation.ActivationError,match="APPROVAL_EXPIRED"):
load_approval(
tmp_path/"expired",
monkeypatch,
issued_at="2026-08-17T01:00:00Z",
not_before="2026-08-17T01:00:00Z",
expires_at="2026-08-18T01:00:00Z",
)
# Use distinct temp roots because helper writes fixed names.
with pytest.raises(activation.ActivationError,match="NOT_YET_VALID"):
load_approval(
tmp_path/"future",
monkeypatch,
issued_at="2026-08-19T03:00:00Z",
not_before="2026-08-19T03:00:00Z",
expires_at="2026-08-20T03:00:00Z",
)
with pytest.raises(activation.ActivationError,match="VALIDITY_TOO_LONG"):
load_approval(
tmp_path/"long",
monkeypatch,
issued_at="2026-08-18T01:00:00Z",
not_before="2026-08-18T01:00:00Z",
expires_at="2026-08-26T01:00:00Z",
)
def test_validate_release_requires_edge_and_controller_and_denies_symlink(tmp_path):
release=tmp_path/"release"
app=release/"active-runtime/app"
app.mkdir(parents=True)
edge_file=app/"aios_gpt_https_edge.py"
controller=app/"aios_gpt_https_edge_activation.py"
edge_file.write_text("edge\n")
controller.write_text("controller\n")
value=activation.validate_release(release)
assert value["edge"]==edge_file
assert value["controller"]==controller
controller.unlink()
with pytest.raises(activation.ActivationError,match="IMMUTABLE_RELEASE_INVALID"):
activation.validate_release(release)
def test_render_dropin_is_exact_and_contains_no_secret(tmp_path,monkeypatch):
approval,_,_=load_approval(tmp_path,monkeypatch)
release=Path("/opt/aios/releases")/(CANDIDATE+"-gpt-https-edge-public")
text=activation.render_dropin(approval,release)
assert "AIOS_GPT_EDGE_HOST=0.0.0.0" in text
assert "AIOS_GPT_EDGE_PORT=18443" in text
assert "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1" in text
assert "AIOS_GPT_EDGE_PUBLIC_DOMAIN=tasks.example.com" in text
assert CANDIDATE in text
for forbidden in ("Bearer ","agent_task_control","tls_key","gpt_ingress_token"):
assert forbidden not in text
def test_receipt_contains_binding_but_no_owner_secret_material(tmp_path,monkeypatch):
approval,_,_=load_approval(tmp_path,monkeypatch)
receipt=activation.receipt_for(
approval,
Path("/opt/aios/releases/test"),
state="ACTIVE",
)
assert receipt["approval_id"]==approval.approval_id
assert receipt["candidate_oid"]==CANDIDATE
assert receipt["domain"]=="tasks.example.com"
serialized=json.dumps(receipt)
assert "Bearer " not in serialized
assert "private" not in serialized.lower()
def test_activate_is_idempotent_for_same_active_approval(tmp_path,monkeypatch):
approval,_,cert=load_approval(tmp_path,monkeypatch)
release=tmp_path/"release"
app=release/"active-runtime/app/aios_gpt_https_edge.py"
controller=release/"active-runtime/app/aios_gpt_https_edge_activation.py"
app.parent.mkdir(parents=True)
app.write_text("print('x')\n")
controller.write_text("print('controller')\n")
key=write_owner_file(tmp_path/"key.pem",b"private-key-material-"*4,0o600)
dropin=tmp_path/"edge.d"/activation.DROPIN_NAME
state=tmp_path/"state.json"
systemd=FakeSystemd()
monkeypatch.setattr(activation.os,"geteuid",lambda:0)
monkeypatch.setattr(activation,"require_ingress_ready",lambda _:None)
monkeypatch.setattr(activation,"validate_tls_pair",lambda *a,**k:{"cert_sha256":approval.tls_cert_sha256})
monkeypatch.setattr(activation,"tls_probe",lambda *a,**k:None)
systemd.exec_start=str(app)
systemd.environment=(
"AIOS_GPT_EDGE_HOST=0.0.0.0 "
"AIOS_GPT_EDGE_PORT=18443 "
"AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1 "
"AIOS_GPT_EDGE_PUBLIC_DOMAIN=tasks.example.com"
)
first=activation.activate(
approval=approval,
release_root=release,
cert_path=cert,
key_path=key,
dropin_path=dropin,
state_path=state,
systemd=systemd,
)
calls_after_first=len(systemd.calls)
second=activation.activate(
approval=approval,
release_root=release,
cert_path=cert,
key_path=key,
dropin_path=dropin,
state_path=state,
systemd=systemd,
)
assert first["state"]=="ACTIVE"
assert second==json.loads(state.read_text())
assert len(systemd.calls)==calls_after_first
def test_activate_failure_after_mutation_rolls_back(tmp_path,monkeypatch):
approval,_,cert=load_approval(tmp_path,monkeypatch)
release=tmp_path/"release"
app=release/"active-runtime/app/aios_gpt_https_edge.py"
controller=release/"active-runtime/app/aios_gpt_https_edge_activation.py"
app.parent.mkdir(parents=True)
app.write_text("print('x')\n")
controller.write_text("print('controller')\n")
key=write_owner_file(tmp_path/"key.pem",b"private-key-material-"*4,0o600)
dropin=tmp_path/"edge.d"/activation.DROPIN_NAME
state=tmp_path/"state.json"
systemd=FakeSystemd()
systemd.fail_restart=True
monkeypatch.setattr(activation.os,"geteuid",lambda:0)
monkeypatch.setattr(activation,"require_ingress_ready",lambda _:None)
monkeypatch.setattr(activation,"validate_tls_pair",lambda *a,**k:{"cert_sha256":approval.tls_cert_sha256})
monkeypatch.setattr(activation,"tls_probe",lambda *a,**k:None)
with pytest.raises(subprocess.CalledProcessError):
activation.activate(
approval=approval,
release_root=release,
cert_path=cert,
key_path=key,
dropin_path=dropin,
state_path=state,
systemd=systemd,
)
assert not dropin.exists()
assert not systemd.active(activation.EDGE_UNIT)
assert json.loads(state.read_text())["state"]=="ROLLED_BACK"
def test_rollback_is_replay_safe_and_leaves_owner_assets_alone(tmp_path,monkeypatch):
approval_file=tmp_path/"approval.json"
cert=tmp_path/"cert.pem"
key=tmp_path/"key.pem"
for path in (approval_file,cert,key):
path.write_text("owner-asset"*4)
dropin=tmp_path/"edge.d"/activation.DROPIN_NAME
dropin.parent.mkdir(parents=True)
dropin.write_text("[Service]\n")
state=tmp_path/"state.json"
state.write_text(json.dumps({
"schema":activation.STATE_SCHEMA,
"state":"ACTIVE",
"candidate_oid":CANDIDATE,
"approval_id":"owner-edge-approval-0001",
"domain":"tasks.example.com",
}))
systemd=FakeSystemd()
systemd.active_units.add(activation.EDGE_UNIT)
systemd.enabled_units.add(activation.EDGE_UNIT)
monkeypatch.setattr(activation.os,"geteuid",lambda:0)
first=activation.rollback(
dropin_path=dropin,
state_path=state,
systemd=systemd,
reason="owner_requested",
)
second=activation.rollback(
dropin_path=dropin,
state_path=state,
systemd=systemd,
reason="owner_requested",
)
assert first["state"]=="ROLLED_BACK"
assert second["state"]=="ROLLED_BACK"
assert not dropin.exists()
assert not systemd.active(activation.EDGE_UNIT)
assert not systemd.enabled(activation.EDGE_UNIT)
for path in (approval_file,cert,key):
assert path.exists()
def test_activation_source_never_creates_owner_approval_or_tls_secret():
source=(ROOT/"active-runtime/app/aios_gpt_https_edge_activation.py").read_text()
assert "secrets.token" not in source
assert "openssl" not in source.lower()
assert "BEGIN PRIVATE KEY" not in source
assert "gpt-https-edge.approved" in source
assert "gpt-https-edge-tls.key" in source
def test_docs_separate_implementation_activation_and_runtime_verification():
doc=(ROOT/"docs/superpowers/specs/2026-08-19-gpt-https-edge-activation.md").read_text()
assert "IMPLEMENTED" in doc
assert "OWNER_APPROVED" in doc
assert "ACTIVATED" in doc
assert "RUNTIME_VERIFIED" in doc
assert "does not own" in doc.lower()
@@ -0,0 +1,83 @@
from __future__ import annotations
import sys
from pathlib import Path
import pytest
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/"active-runtime/app"))
from aios_capability_contract import ( # noqa: E402
has_capabilities,
normalize_required_capabilities,
normalize_worker_capabilities,
normalized_capability_matrix,
required_capabilities_subset,
)
def test_release_candidate_is_known_to_task_and_worker_vocabularies():
capability="release:candidate"
assert normalize_required_capabilities([capability])==frozenset({capability})
assert normalize_worker_capabilities([capability])==frozenset({capability})
assert has_capabilities([capability],[capability]) is True
assert normalized_capability_matrix([capability],[capability])=={
"worker_effective":["release:candidate"],
"task_required":["release:candidate"],
"eligible":True,
}
def test_release_candidate_capability_does_not_imply_high_risk_release_operations():
worker=["release:candidate"]
for forbidden in (
"submitOwnerApproval",
"startDeployment",
"tag_publish",
"baseline_registry_finalize",
"runtime_switch",
"deploy.production",
):
assert has_capabilities(worker,[forbidden]) is False
def test_release_wildcards_and_nearby_unknown_names_remain_fail_closed():
assert has_capabilities(["release:candidate"],["release:*"]) is False
assert has_capabilities(["release:candidate"],["release:candidate.extra"]) is False
assert has_capabilities(["release:*"],["release:candidate"]) is False
def test_release_candidate_delegation_subset_is_exact():
assert required_capabilities_subset(
["release:candidate"],
["release:candidate"],
) is True
assert required_capabilities_subset(
["release:candidate"],
["startDeployment"],
) is False
@pytest.mark.parametrize(
"worker,required",
[
(["agent.runtime.hermes"],["agent.runtime.hermes"]),
(["agent.runtime.openclaw"],["agent.runtime.openclaw"]),
(["git.read"],["repository_read"]),
(["git.test"],["test_execution"]),
(
["governed.write.branch","governed.write.file"],
["governed.git.write"],
),
],
)
def test_existing_known_capability_semantics_are_non_regressed(worker,required):
assert has_capabilities(worker,required) is True
def test_release_candidate_worker_does_not_match_empty_or_unrelated_task_by_escalation():
assert has_capabilities(["release:candidate"],[]) is True
assert has_capabilities(["release:candidate"],["git.read"]) is False
assert has_capabilities(["release:candidate"],["agent.runtime.hermes"]) is False
+226
View File
@@ -0,0 +1,226 @@
from __future__ import annotations
import json
from pathlib import Path
import sys
import pytest
ROOT=Path(__file__).resolve().parents[1]
APP=ROOT/"active-runtime/app"
sys.path.insert(0,str(APP))
import aios_release_online_agent_worker as worker
def task(**overrides):
value={
"task_id":"r42-release-candidate-live-acceptance-20260819",
"required_capabilities":["release:candidate"],
"constraints":{
"release_candidate":{
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
"source_commit":"5"*40,
"version":"V2R2C2",
"projection_id":"projection-"+"a"*32,
}
}
}
value.update(overrides)
return value
def test_release_request_accepts_only_exact_release_capability():
req=worker.ReleaseRequest.from_task(task())
assert req.candidate_id=="r42-release-candidate-v2r2c2-20260819"
assert req.source_commit=="5"*40
assert req.version=="V2R2C2"
assert req.projection_id=="projection-"+"a"*32
with pytest.raises(worker.ReleaseWorkerError,match="TASK_CAPABILITY_SCOPE_INVALID"):
worker.ReleaseRequest.from_task(task(required_capabilities=["release:candidate","git.read"]))
@pytest.mark.parametrize(
"payload",
[
{
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
"source_commit":"short",
"version":"V2R2C2",
"projection_id":"projection-"+"a"*32,
},
{
"candidate_id":"bad",
"source_commit":"5"*40,
"version":"V2R2C2",
"projection_id":"projection-"+"a"*32,
},
{
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
"source_commit":"5"*40,
"version":"bad",
"projection_id":"projection-"+"a"*32,
},
{
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
"source_commit":"5"*40,
"version":"V2R2C2",
"projection_id":"bad",
},
{
"candidate_id":"r42-release-candidate-v2r2c2-20260819",
"source_commit":"5"*40,
"version":"V2R2C2",
"projection_id":"projection-"+"a"*32,
"execute":True,
},
],
)
def test_release_request_rejects_invalid_or_extra_fields(payload):
value=task()
value["constraints"]={"release_candidate":payload}
with pytest.raises(worker.ReleaseWorkerError):
worker.ReleaseRequest.from_task(value)
def test_result_payload_preserves_high_risk_denials():
release=worker.ReleaseRequest.from_task(task())
result=worker.result_payload(
"r42-release-candidate-live-acceptance-20260819",
release,
{"candidate_id":release.candidate_id,"status":"BUILT"},
)
assert result["status"]=="SUCCEEDED"
assert result["classification"]=="RELEASE_CANDIDATE_CREATED"
assert result["writer_lease"]=="NONE"
assert result["changed_files"]==[]
assert result["direct_main_write"] is False
assert result["production_write"] is False
assert result["tag_write"] is False
assert result["baseline_write"] is False
assert result["runtime_switch"] is False
assert result["owner_approval_submitted"] is False
assert result["credential_material_exposed"] is False
def test_result_payload_rejects_builder_candidate_mismatch():
release=worker.ReleaseRequest.from_task(task())
with pytest.raises(worker.ReleaseWorkerError,match="BUILDER_CANDIDATE_BINDING_INVALID"):
worker.result_payload(
"r42-release-candidate-live-acceptance-20260819",
release,
{"candidate_id":"other-release-candidate"},
)
def test_worker_calls_only_fixed_release_candidate_route(monkeypatch):
release=worker.ReleaseRequest.from_task(task())
calls=[]
def fake_http(base,method,route,token,body=None,timeout=30):
calls.append((base,method,route,token,body))
if route=="/health":
return 200,{"status":"ok"}
if route=="/createReleaseCandidate" and body=={}:
return 400,{"error":"request_invalid"}
if route.endswith("/claim"):
return 200,{"lease_token":"l"*32,"attempt":1}
if route.endswith("/heartbeat"):
return 200,{}
if route=="/createReleaseCandidate":
return 200,{
"candidate_id":release.candidate_id,
"artifact_sha256":"f"*64,
}
if route.endswith("/result"):
return 200,{"status":"VERIFYING"}
raise AssertionError(route)
monkeypatch.setattr(worker,"http_json",fake_http)
monkeypatch.setattr(worker,"audit",lambda *args,**kwargs:None)
monkeypatch.setattr(worker,"write_liveness",lambda *args,**kwargs:None)
monkeypatch.setattr(worker.Heartbeat,"first",lambda self:None)
monkeypatch.setattr(worker.Heartbeat,"close",lambda self:None)
monkeypatch.setattr(worker.Heartbeat,"ensure",lambda self:None)
monkeypatch.setattr(worker.threading.Thread,"start",lambda self:None)
assert worker.execute_task(
task(),
agent_token="a"*32,
release_token="r"*32,
) is True
release_calls=[
item for item in calls
if item[0]==worker.RELEASE_API
and item[2]=="/createReleaseCandidate"
and item[4]!={}
]
assert len(release_calls)==1
assert release_calls[0][1]=="POST"
assert release_calls[0][2]=="/createReleaseCandidate"
assert release_calls[0][4]=={
"candidate_id":release.candidate_id,
"source_commit":release.source_commit,
"version":release.version,
}
def test_write_liveness_records_release_capability(tmp_path,monkeypatch):
path=tmp_path/"worker-liveness.json"
monkeypatch.setattr(worker,"STATE_ROOT",tmp_path)
monkeypatch.setattr(worker,"LIVENESS_PATH",path)
worker.write_liveness(
"IDLE",
stage="poll",
visible_count=2,
)
value=json.loads(path.read_text())
assert value["schema"]=="aios.online-agent.worker-liveness.v1"
assert value["agent_type"]=="release-preparation-worker"
assert value["capabilities"]==["release:candidate"]
assert value["state"]=="IDLE"
assert value["visible_count"]==2
def test_sd_notify_is_noop_without_systemd_socket(monkeypatch):
monkeypatch.delenv("NOTIFY_SOCKET",raising=False)
worker.sd_notify("WATCHDOG=1")
def test_systemd_unit_uses_two_encrypted_credentials_and_loopback_release_api():
unit=(ROOT/"deploy/systemd/aios-release-online-agent-worker.service").read_text()
assert "LoadCredentialEncrypted=agent_token:" in unit
assert "LoadCredentialEncrypted=release_candidate_token:" in unit
assert "AIOS_RELEASE_API_URL=http://127.0.0.1:18890" in unit
assert "AIOS_RELEASE_WORKER_AGENT_ID=online-release-worker-cn-network-mkt-20260819" in unit
assert "Type=notify" in unit
assert "WatchdogSec=90" in unit
assert "Restart=always" in unit
assert "IPAddressDeny=any" in unit
assert "IPAddressAllow=127.0.0.0/8" in unit
assert "InaccessiblePaths=/data/aios-gitops /home" in unit
assert "submitOwnerApproval" not in unit
assert "startDeployment" not in unit
def test_source_contains_no_high_risk_release_routes():
source=(ROOT/"active-runtime/app/aios_release_online_agent_worker.py").read_text()
assert '"/createReleaseCandidate"' in source
for forbidden in (
"/submitOwnerApproval",
"/startDeployment",
"/requestOrCreateTag",
"/buildProductionArtifact",
"/executeApprovedProductionRelease",
):
assert forbidden not in source
def test_docs_keep_release_candidate_distinct_from_production():
doc=(ROOT/"docs/superpowers/specs/2026-08-19-release-preparation-online-agent.md").read_text()
assert "release:candidate" in doc
assert "Owner signature" in doc
assert "never\nstarts deployment" in doc or "never starts deployment" in doc
assert "LoadCredentialEncrypted" in doc