Compare commits

..
Author SHA1 Message Date
Codex Agent c6316943d2 docs(governance): close live activation evidence 2026-07-29 11:28:48 +08:00
Codex Agent e1e438564c feat(governance): add fail-closed live activation 2026-07-29 11:27:04 +08:00
AIOS Task-Bound Prerequisites Writer fa038d17b2 governance: add dormant single-owner activation prerequisites 2026-07-29 04:00:00 +08:00
gitops-bot 875d41c310 Merge pull request 'fix(gitops): harden Codex worker systemd profile' (#19) from patch/isolated-writer-systemd-profile-hardening-20260725 into main 2026-07-25 11:13:31 +08:00
chatgpt-gitops-bot 870ac61dc2 fix(gitops): harden Codex worker systemd profile 2026-07-25 11:08:55 +08:00
gitops-bot 009f276699 Merge pull request 'fix(gitops): add exact-path service policy for isolated writer' (#18) from patch/isolated-writer-exact-path-policy-20260725 into main 2026-07-25 10:19:02 +08:00
chatgpt-gitops-bot 600c717e17 fix(gitops): add exact-path service policy for isolated writer 2026-07-25 10:16:43 +08:00
gitops-bot 02be6d28ed merge(gitops): add constrained writer token policy and atomic commit-file
Source: patch/isolated-writer-token-policy-v2-20260724
Source commit: b1e240dda3
Target base: df099420d2
Production runtime changes: not authorized
Tag/Baseline/Registry operations: not authorized
2026-07-24 23:36:25 +08:00
chatgpt-gitops-bot b1e240dda3 fix(gitops): reject atomic writer force fields 2026-07-24 23:06:22 +08:00
chatgpt-gitops-bot 2dfbe837f4 fix(gitops): bind writer runtime policy v2 artifacts 2026-07-24 23:01:40 +08:00
chatgpt-gitops-bot 80df1e0d87 test(gitops): close writer authz and recovery contract 2026-07-24 22:56:58 +08:00
chatgpt-gitops-bot 7ecf7cbfce feat(gitops): add idempotent atomic commit-file endpoint 2026-07-24 22:56:58 +08:00
chatgpt-gitops-bot bbada7c863 feat(gitops): add constrained writer token policy v2 2026-07-24 22:56:58 +08:00
59 changed files with 5416 additions and 1894 deletions
@@ -5,13 +5,148 @@ import json
import os
import subprocess
from pathlib import Path
import urllib.parse
import aios_chatgpt_actions_api as api
import aios_isolated_writer_policy as policy
from isolated_writer_authz import AuthorizationError, authorize
from isolated_writer_commit_file import AtomicCommitFile, CommitFileError, OperationLedger
from isolated_writer_token_registry import TokenRegistry
_reject = api.reject_non_read_http
_write_file = api.write_file
_sync_branch = api.sync_branch
_original_do_get = api.H.do_GET
api.OPENAPI = Path(os.getenv("AIOS_GITOPS_ISOLATED_WRITER_OPENAPI", str(api.OPENAPI)))
def _registry() -> TokenRegistry:
return TokenRegistry(Path(os.environ["AIOS_GITOPS_TOKEN_REGISTRY_V2"]))
def _atomic_writer() -> AtomicCommitFile:
return AtomicCommitFile(
workspace=api.WS,
bare_repo=api.BARE_REPO,
ledger=OperationLedger(Path(os.environ["AIOS_GITOPS_WRITER_OPERATION_LEDGER"])),
remote=os.getenv("AIOS_GITOPS_GITEA_PUSH_REMOTE", ""),
audit=policy.append_audit_record,
)
def _bearer(handler) -> str:
value = handler.headers.get("Authorization", "")
return value[7:].strip() if value.startswith("Bearer ") else ""
def _authorize(handler, *, method: str, path: str, branch=None, target_path=None):
grant = authorize(
_registry(),
_bearer(handler),
method=method,
path=path,
source_ip=str(handler.client_address[0]),
workspace=api.WS,
branch=branch,
target_path=target_path,
)
grant.record["_source_ip"] = str(handler.client_address[0])
return grant
def _authorization_error(handler, exc: AuthorizationError):
return handler.sendj({"error": exc.code}, exc.status)
def _isolated_runtime_status(handler):
branch = api.sh(["rev-parse", "--abbrev-ref", "HEAD"], True)["stdout"]
head = api.sh(["rev-parse", "HEAD"], True)["stdout"]
return handler.sendj({
"service": os.getenv(
"AIOS_GITOPS_SERVICE_AUDIENCE",
"aios-isolated-writer.service",
),
"status": "active",
"runtime_role": os.getenv("AIOS_GITOPS_RUNTIME_ROLE"),
"workspace_role": os.getenv("AIOS_GITOPS_WORKSPACE_ROLE"),
"writes_enabled": os.getenv("AIOS_GITOPS_WRITES_ENABLED", "").lower()
in {"1", "true", "yes", "on"},
"allowed_branch": policy.allowed_branch(),
"approved_base_oid": policy.allowed_base_oid(),
"workspace_branch": branch,
"workspace_head_oid": head,
"repository_identity": os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY"),
"token_policy_version": 2,
"atomic_commit_file": True,
"path_policy_sha256": policy.configured_path_policy_sha256(),
})
def _isolated_do_get(self):
path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
if path in {"/health", "/api/health"}:
return self.sendj({
"status": "ok",
"service": os.getenv(
"AIOS_GITOPS_SERVICE_AUDIENCE",
"aios-isolated-writer.service",
),
})
if path not in {"/api/runtime-status", "/api/baseline", "/api/patches"}:
return self.sendj({"error": "isolated_writer_operation_forbidden"}, 403)
try:
_authorize(self, method="GET", path=path)
except AuthorizationError as exc:
return _authorization_error(self, exc)
if path == "/api/runtime-status":
return _isolated_runtime_status(self)
return _original_do_get(self)
def _read_commit_file_body(handler):
try:
length = int(handler.headers.get("Content-Length", "0"))
except ValueError:
raise CommitFileError("writer_content_length_rejected") from None
if length <= 0:
raise CommitFileError("writer_request_object_required")
if length > 1_100_000:
raise CommitFileError("writer_request_too_large", 413)
raw = handler.rfile.read(length)
try:
value = json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError):
raise CommitFileError("writer_request_json_rejected") from None
if not isinstance(value, dict):
raise CommitFileError("writer_request_object_required")
return value
def _isolated_do_post(self):
path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
if path != "/api/commit-file":
return self.sendj({"error": "isolated_writer_operation_forbidden"}, 403)
try:
data = _read_commit_file_body(self)
branch = str(data.get("branch") or "")
target_path = str(data.get("path") or "")
grant = _authorize(
self,
method="POST",
path=path,
branch=branch,
target_path=target_path,
)
if api.content_has_secret(str(data.get("content") or "")):
raise CommitFileError("writer_secret_content_rejected", 403)
result = _atomic_writer().execute(grant, data)
return self.sendj(result)
except AuthorizationError as exc:
return _authorization_error(self, exc)
except CommitFileError as exc:
return self.sendj({"error": exc.code}, exc.status)
except PermissionError:
return self.sendj({"error": "isolated_writer_policy_rejected"}, 403)
def reject_non_read_http(handler, method: str, path: str) -> bool:
@@ -173,6 +308,8 @@ api.H.branch, api.H.file, api.H.commit, api.H.patch = (
_checked_commit,
_checked_patch,
)
api.H.do_GET = _isolated_do_get
api.H.do_POST = _isolated_do_post
if __name__ == "__main__":
policy.require_service_invariants(api.WS, api.BARE_REPO)
@@ -9,10 +9,9 @@ import re
from pathlib import Path, PurePosixPath, PureWindowsPath
from urllib.parse import unquote
WRITE_ENDPOINTS = frozenset({
"/api/branch", "/api/create-branch", "/api/file", "/api/commit-file",
"/api/commit", "/api/patch", "/api/create-patch",
})
import isolated_writer_path_policy as path_policy
WRITE_ENDPOINTS = frozenset({"/api/commit-file"})
ALLOWED_ROOTS = (
"active-runtime",
"docs",
@@ -69,6 +68,15 @@ def require_service_invariants(workspace: Path, bare_repo: Path) -> None:
raise PermissionError("isolated_writer_role_required")
if os.getenv("AIOS_GITOPS_WORKSPACE_ROLE", "") != "isolated-writer":
raise PermissionError("isolated_writer_workspace_role_required")
if os.getenv("AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED", "") != "2":
raise PermissionError("isolated_writer_token_policy_v2_required")
if not os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", "").strip():
raise PermissionError("isolated_writer_service_audience_required")
if not os.getenv("AIOS_GITOPS_TOKEN_REGISTRY_V2", "").strip():
raise PermissionError("isolated_writer_token_registry_required")
if not os.getenv("AIOS_GITOPS_WRITER_OPERATION_LEDGER", "").strip():
raise PermissionError("isolated_writer_operation_ledger_required")
path_policy.configured_path_policy()
for name in _DENY_FLAGS:
if os.getenv(name, "").strip().lower() not in _FALSE:
raise PermissionError("isolated_writer_forbidden_capability_flag")
@@ -182,6 +190,7 @@ def _audit_sanitize(value):
"private_key",
)
)
and str(key).lower() != "token_id"
else _audit_sanitize(inner)
)
for key, inner in value.items()
@@ -239,9 +248,14 @@ def validate_path(path: str, workspace: Path | None = None) -> str:
):
raise PermissionError("isolated_writer_path_rejected")
first = parts[0]
service_exact_path = False
if first in DENIED_ROOTS:
raise PermissionError("isolated_writer_protected_path_rejected")
if value != "README.md" and first not in ALLOWED_ROOTS:
try:
path_policy.require_service_path("commit_file.atomic", value)
service_exact_path = True
except path_policy.PathPolicyError as exc:
raise PermissionError(str(exc)) from None
if value != "README.md" and first not in ALLOWED_ROOTS and not service_exact_path:
raise PermissionError("isolated_writer_path_outside_allowed_roots")
if workspace is not None:
lexical_root = Path(os.path.abspath(str(workspace)))
@@ -261,6 +275,22 @@ def validate_path(path: str, workspace: Path | None = None) -> str:
return value
def validate_content_for_path(path: str, content: str) -> None:
if path.split("/", 1)[0] not in DENIED_ROOTS:
return
try:
path_policy.validate_content("commit_file.atomic", path, content)
except path_policy.PathPolicyError as exc:
raise PermissionError(str(exc)) from None
def configured_path_policy_sha256() -> str | None:
try:
return path_policy.configured_path_policy_sha256()
except path_policy.PathPolicyError as exc:
raise PermissionError(str(exc)) from None
def validate_commit_paths(paths: list[str], workspace: Path | None = None) -> None:
if enforcement_required():
for path in paths:
+167
View File
@@ -0,0 +1,167 @@
"""Fail-closed operation authorization for an isolated Git writer."""
from __future__ import annotations
import ipaddress
import os
import re
import subprocess
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from isolated_writer_token_registry import TokenRegistry
ROUTE_OPERATIONS = {
("GET", "/api/runtime-status"): "runtime.status.read",
("GET", "/api/baseline"): "baseline.read",
("GET", "/api/patches"): "patches.read",
("POST", "/api/commit-file"): "commit_file.atomic",
}
OID_RE = re.compile(r"[0-9a-f]{40}")
class AuthorizationError(PermissionError):
def __init__(self, code: str, status: int = 403):
super().__init__(code)
self.code = code
self.status = status
@dataclass(frozen=True)
class AuthorizedGrant:
record: dict[str, Any]
operation: str
use_count: int
@property
def token_id(self) -> str:
return str(self.record["token_id"])
@property
def task_id(self) -> str:
return str(self.record["task_id"])
def operation_for(method: str, path: str) -> str:
operation = ROUTE_OPERATIONS.get((method.upper(), path))
if not operation:
raise AuthorizationError("isolated_writer_operation_forbidden")
return operation
def _parse_time(value: Any) -> datetime:
try:
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
except (TypeError, ValueError):
raise AuthorizationError("writer_token_time_invalid") from None
if parsed.tzinfo is None:
raise AuthorizationError("writer_token_time_invalid")
return parsed.astimezone(timezone.utc)
def _git_is_ancestor(workspace: Path, base: str, branch: str) -> bool:
result = subprocess.run(
["git", "merge-base", "--is-ancestor", base, branch],
cwd=workspace,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return result.returncode == 0
def _validate_grant(
grant: dict[str, Any],
*,
audience: str,
source_ip: str,
operation: str,
branch: str | None,
path: str | None,
workspace: Path,
current_time: datetime,
) -> None:
if int(grant.get("schema_version", 0)) != 2:
raise AuthorizationError("writer_token_policy_v2_required", 401)
if str(grant.get("audience") or "") != audience:
raise AuthorizationError("writer_token_audience_rejected")
try:
peer = ipaddress.ip_address(source_ip)
networks = [ipaddress.ip_network(str(value), strict=True) for value in grant.get("source_cidrs", [])]
except ValueError:
raise AuthorizationError("writer_token_source_rejected") from None
if not networks or not any(peer in network for network in networks):
raise AuthorizationError("writer_token_source_rejected")
created = _parse_time(grant.get("created_at"))
not_before = _parse_time(grant.get("not_before"))
expires = _parse_time(grant.get("expires_at"))
if expires <= not_before or (expires - created).total_seconds() > 1800:
raise AuthorizationError("writer_token_lifetime_rejected")
if current_time < not_before:
raise AuthorizationError("writer_token_not_yet_valid", 401)
if current_time >= expires:
raise AuthorizationError("writer_token_expired", 401)
if bool(grant.get("revoked")):
raise AuthorizationError("writer_token_revoked", 401)
operations = grant.get("allowed_operations")
if not isinstance(operations, list) or operation not in operations:
raise AuthorizationError("writer_token_operation_rejected")
expected_branch = str(grant.get("allowed_branch") or "")
configured_branch = os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", "")
if expected_branch != configured_branch:
raise AuthorizationError("writer_token_branch_rejected")
if branch is not None and branch != expected_branch:
raise AuthorizationError("writer_token_branch_rejected")
if path is not None:
paths = grant.get("allowed_paths")
if not isinstance(paths, list) or path not in paths:
raise AuthorizationError("writer_token_path_rejected")
base = str(grant.get("approved_base_oid") or "").lower()
if not OID_RE.fullmatch(base):
raise AuthorizationError("writer_token_base_rejected")
if not _git_is_ancestor(workspace, base, expected_branch):
raise AuthorizationError("writer_token_base_not_ancestor")
def authorize(
registry: TokenRegistry,
token: str,
*,
method: str,
path: str,
source_ip: str,
workspace: Path,
branch: str | None = None,
target_path: str | None = None,
current_time: datetime | None = None,
) -> AuthorizedGrant:
if os.getenv("AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED", "") != "2":
raise AuthorizationError("writer_token_policy_v2_not_required")
if not token:
raise AuthorizationError("unauthorized", 401)
grant = registry.find(token)
if grant is None:
raise AuthorizationError("unauthorized", 401)
operation = operation_for(method, path)
_validate_grant(
grant,
audience=os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", ""),
source_ip=source_ip,
operation=operation,
branch=branch,
path=target_path,
workspace=workspace,
current_time=current_time or datetime.now(timezone.utc),
)
limits = grant.get("max_uses")
if not isinstance(limits, dict):
raise AuthorizationError("writer_token_use_limit_missing")
try:
used = registry.consume(str(grant["token_id"]), operation, int(limits[operation]))
except (KeyError, TypeError, ValueError):
raise AuthorizationError("writer_token_use_limit_missing") from None
except PermissionError as exc:
code = str(exc)
raise AuthorizationError(code, 401 if code == "writer_token_revoked" else 403) from None
return AuthorizedGrant(grant, operation, used)
@@ -0,0 +1,403 @@
"""Atomic, idempotent commit-file Saga for the isolated writer."""
from __future__ import annotations
import fcntl
import hashlib
import json
import os
import re
import sqlite3
import subprocess
import tempfile
import threading
from pathlib import Path
from typing import Any
import aios_isolated_writer_policy as policy
OID_RE = re.compile(r"[0-9a-f]{40}")
IDEMPOTENCY_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
STATES = frozenset({
"PENDING", "FILE_WRITTEN", "COMMITTED", "PUSHED", "VERIFIED", "COMPLETED",
"FAILED_RECOVERABLE", "MANUAL_REVIEW_REQUIRED",
})
_LOCKS: dict[str, threading.Lock] = {}
_LOCKS_GUARD = threading.Lock()
_LEDGER_INIT_LOCK = threading.Lock()
class CommitFileError(RuntimeError):
def __init__(self, code: str, status: int = 400):
super().__init__(code)
self.code = code
self.status = status
class SimulatedCrash(RuntimeError):
pass
def canonical_digest(data: dict[str, Any]) -> str:
payload = {
"branch": data["branch"],
"path": data["path"],
"content_sha256": hashlib.sha256(data["content"].encode("utf-8")).hexdigest(),
"message": data["message"],
"expected_head_oid": data["expected_head_oid"],
}
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
class OperationLedger:
def __init__(self, path: Path | str):
self.path = Path(path)
self.path.parent.mkdir(parents=True, exist_ok=True)
if self.path.parent.is_symlink() or self.path.is_symlink():
raise PermissionError("writer_operation_ledger_symlink_rejected")
with _LEDGER_INIT_LOCK:
with self.connect() as db:
db.execute(
"""
CREATE TABLE IF NOT EXISTS writer_operations (
operation_id TEXT PRIMARY KEY,
token_id TEXT NOT NULL,
task_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL UNIQUE,
operation_digest TEXT NOT NULL,
branch TEXT NOT NULL,
path TEXT NOT NULL,
content_sha256 TEXT NOT NULL,
pre_head_oid TEXT NOT NULL,
commit_oid TEXT,
push_status TEXT,
verification_status TEXT,
state TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
error_code TEXT
)
"""
)
os.chmod(self.path, 0o600)
def connect(self) -> sqlite3.Connection:
db = sqlite3.connect(self.path, timeout=30)
db.row_factory = sqlite3.Row
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA synchronous=FULL")
return db
def get_or_create(self, grant: Any, data: dict[str, Any], digest: str) -> dict[str, Any]:
key = data["idempotency_key"]
with self.connect() as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute(
"SELECT * FROM writer_operations WHERE idempotency_key=?", (key,)
).fetchone()
if row:
db.commit()
record = dict(row)
if record["operation_digest"] != digest:
raise CommitFileError("writer_idempotency_conflict", 409)
if (
record["token_id"] != grant.token_id
or record["task_id"] != grant.task_id
):
raise CommitFileError("writer_idempotency_owner_conflict", 409)
return record
operation_id = hashlib.sha256(
f"{grant.task_id}\0{key}".encode("utf-8")
).hexdigest()
content_sha = hashlib.sha256(data["content"].encode("utf-8")).hexdigest()
db.execute(
"""
INSERT INTO writer_operations(
operation_id,token_id,task_id,idempotency_key,operation_digest,
branch,path,content_sha256,pre_head_oid,state
) VALUES(?,?,?,?,?,?,?,?,?,?)
""",
(
operation_id, grant.token_id, grant.task_id, key, digest,
data["branch"], data["path"], content_sha,
data["expected_head_oid"], "PENDING",
),
)
db.commit()
return self.get(key)
def get(self, key: str) -> dict[str, Any]:
with self.connect() as db:
row = db.execute(
"SELECT * FROM writer_operations WHERE idempotency_key=?", (key,)
).fetchone()
if row is None:
raise KeyError(key)
return dict(row)
def update(self, key: str, state: str, **fields: Any) -> dict[str, Any]:
if state not in STATES:
raise ValueError("invalid writer operation state")
allowed = {"commit_oid", "push_status", "verification_status", "error_code"}
assignments = ["state=?", "updated_at=CURRENT_TIMESTAMP"]
values: list[Any] = [state]
for name, value in fields.items():
if name not in allowed:
raise ValueError("invalid ledger field")
assignments.append(f"{name}=?")
values.append(value)
values.append(key)
with self.connect() as db:
db.execute("BEGIN IMMEDIATE")
db.execute(
f"UPDATE writer_operations SET {','.join(assignments)} WHERE idempotency_key=?",
values,
)
db.commit()
return self.get(key)
class AtomicCommitFile:
def __init__(
self,
*,
workspace: Path,
bare_repo: Path,
ledger: OperationLedger,
remote: str,
audit,
crash_after_state: str | None = None,
):
self.workspace = workspace
self.bare_repo = bare_repo
self.ledger = ledger
self.remote = remote
self.audit = audit
self.crash_after_state = crash_after_state
def _git(self, *args: str, check: bool = True) -> str:
result = subprocess.run(
["git", *args], cwd=self.workspace, text=True, capture_output=True
)
if check and result.returncode != 0:
raise CommitFileError("writer_git_operation_failed", 500)
return result.stdout.strip()
def _transition(self, key: str, state: str, **fields: Any) -> dict[str, Any]:
record = self.ledger.update(key, state, **fields)
if self.crash_after_state == state:
raise SimulatedCrash(state)
return record
def _validate(self, data: Any) -> dict[str, Any]:
if not isinstance(data, dict):
raise CommitFileError("writer_request_object_required")
required = ("branch", "path", "content", "message", "expected_head_oid", "idempotency_key")
if any(field not in data for field in required):
raise CommitFileError("writer_request_incomplete")
if data.get("force") or data.get("force_push"):
raise CommitFileError("isolated_writer_force_rejected", 403)
if set(data) != set(required):
raise CommitFileError("writer_request_fields_rejected")
item = {field: data[field] for field in required}
if item["branch"] != policy.allowed_branch():
raise CommitFileError("isolated_writer_branch_rejected", 403)
try:
item["path"] = policy.validate_path(item["path"], self.workspace)
except PermissionError as exc:
raise CommitFileError(str(exc), 403) from None
if not isinstance(item["content"], str) or "\x00" in item["content"]:
raise CommitFileError("writer_content_rejected")
try:
policy.validate_content_for_path(item["path"], item["content"])
except PermissionError as exc:
raise CommitFileError(str(exc), 403) from None
if len(item["content"].encode("utf-8")) > 1_048_576:
raise CommitFileError("writer_content_too_large", 413)
message = item["message"]
if (
not isinstance(message, str)
or not 1 <= len(message) <= 200
or "\x00" in message
or "\n" in message
or "\r" in message
):
raise CommitFileError("writer_message_rejected")
if not isinstance(item["expected_head_oid"], str) or not OID_RE.fullmatch(item["expected_head_oid"]):
raise CommitFileError("writer_expected_head_rejected")
if not isinstance(item["idempotency_key"], str) or not IDEMPOTENCY_RE.fullmatch(item["idempotency_key"]):
raise CommitFileError("writer_idempotency_key_rejected")
return item
def _atomic_write(self, relative: str, content: str) -> None:
target = self.workspace / relative
target.parent.mkdir(parents=True, exist_ok=True)
fd, temporary = tempfile.mkstemp(prefix=".writer-", dir=target.parent)
try:
with os.fdopen(fd, "wb", closefd=True) as stream:
stream.write(content.encode("utf-8"))
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, target)
parent_fd = os.open(target.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(parent_fd)
finally:
os.close(parent_fd)
finally:
if os.path.exists(temporary):
os.unlink(temporary)
def _verify_remote(self, branch: str, oid: str) -> None:
bare = subprocess.run(
["git", "--git-dir", str(self.bare_repo), "rev-parse", f"refs/heads/{branch}"],
text=True, capture_output=True,
)
if bare.returncode != 0 or bare.stdout.strip() != oid:
raise CommitFileError("writer_authoritative_bare_oid_mismatch", 500)
remote = subprocess.run(
["git", "ls-remote", "--heads", self.remote, f"refs/heads/{branch}"],
cwd=self.workspace, text=True, capture_output=True,
)
if remote.returncode != 0:
raise CommitFileError("writer_gitea_verification_failed", 503)
fields = remote.stdout.strip().split()
if len(fields) != 2:
raise CommitFileError("writer_gitea_ref_missing", 503)
if fields[0] != oid:
raise CommitFileError("writer_gitea_oid_conflict", 409)
def execute(self, grant: Any, raw_data: Any) -> dict[str, Any]:
data = self._validate(raw_data)
digest = canonical_digest(data)
branch = data["branch"]
with _LOCKS_GUARD:
lock = _LOCKS.setdefault(branch, threading.Lock())
lock_path = self.ledger.path.parent / f"{hashlib.sha256(branch.encode()).hexdigest()}.lock"
with lock, open(lock_path, "a+b") as branch_lock:
fcntl.flock(branch_lock.fileno(), fcntl.LOCK_EX)
return self._execute_locked(grant, data, digest)
def _execute_locked(self, grant: Any, data: dict[str, Any], digest: str) -> dict[str, Any]:
key = data["idempotency_key"]
record = self.ledger.get_or_create(grant, data, digest)
if record["state"] == "COMPLETED":
return self._result(record, idempotent=True)
if self._git("rev-parse", "--abbrev-ref", "HEAD") != data["branch"]:
record = self._transition(
key, "MANUAL_REVIEW_REQUIRED", error_code="writer_current_branch_conflict"
)
self._audit(grant, data, digest, record, "manual_review_required")
raise CommitFileError("writer_current_branch_conflict", 409)
head = self._git("rev-parse", "HEAD")
commit_oid = record.get("commit_oid")
if head != record["pre_head_oid"] and head != commit_oid:
record = self._transition(
key, "MANUAL_REVIEW_REQUIRED", error_code="writer_branch_head_conflict"
)
self._audit(grant, data, digest, record, "manual_review_required")
raise CommitFileError("writer_branch_head_conflict", 409)
try:
state = record["state"]
if state == "MANUAL_REVIEW_REQUIRED":
raise CommitFileError("writer_manual_review_required", 409)
if state == "FAILED_RECOVERABLE":
if record.get("push_status") == "confirmed":
state = "PUSHED"
else:
state = "COMMITTED" if commit_oid else "FILE_WRITTEN"
if state == "PENDING":
self._atomic_write(data["path"], data["content"])
record = self._transition(key, "FILE_WRITTEN")
state = record["state"]
if state == "FILE_WRITTEN":
actual_sha = hashlib.sha256(
(self.workspace / data["path"]).read_bytes()
).hexdigest()
if actual_sha != record["content_sha256"]:
raise CommitFileError("writer_file_content_conflict", 409)
staged_before = self._git("diff", "--cached", "--name-only", check=False).splitlines()
if staged_before:
raise CommitFileError("writer_extra_staged_files_rejected", 409)
self._git("add", "--", data["path"])
staged = self._git("diff", "--cached", "--name-only").splitlines()
if staged != [data["path"]]:
raise CommitFileError("writer_extra_staged_files_rejected", 409)
self._git("commit", "-m", data["message"])
commit_oid = self._git("rev-parse", "HEAD")
record = self._transition(key, "COMMITTED", commit_oid=commit_oid)
state = record["state"]
if state == "COMMITTED":
commit_oid = str(record["commit_oid"])
pushed = subprocess.run(
["git", "push", "--porcelain", self.remote, f"HEAD:refs/heads/{data['branch']}"],
cwd=self.workspace, text=True, capture_output=True,
)
if pushed.returncode != 0:
self._transition(
key, "FAILED_RECOVERABLE",
push_status="failed", error_code="writer_gitea_push_failed",
)
raise CommitFileError("writer_gitea_push_failed", 503)
record = self._transition(key, "PUSHED", push_status="confirmed")
state = record["state"]
if state == "PUSHED":
commit_oid = str(record["commit_oid"])
self._verify_remote(data["branch"], commit_oid)
record = self._transition(
key, "VERIFIED", verification_status="bare+gitea"
)
state = record["state"]
if state == "VERIFIED":
record = self._transition(key, "COMPLETED", error_code=None)
self._audit(grant, data, digest, record, "completed")
return self._result(record, idempotent=False)
except SimulatedCrash:
raise
except CommitFileError as exc:
latest = self.ledger.get(key)
if latest["state"] not in {"FAILED_RECOVERABLE", "MANUAL_REVIEW_REQUIRED"}:
failure_state = (
"MANUAL_REVIEW_REQUIRED"
if exc.code in {
"writer_gitea_oid_conflict",
"writer_file_content_conflict",
"writer_current_branch_conflict",
}
else "FAILED_RECOVERABLE"
)
latest = self._transition(key, failure_state, error_code=exc.code)
self._audit(grant, data, digest, latest, "failed")
raise
def _audit(self, grant: Any, data: dict[str, Any], digest: str, record: dict[str, Any], result: str) -> None:
self.audit({
"token_id": grant.token_id,
"task_id": grant.task_id,
"audience": grant.record["audience"],
"source_ip": grant.record.get("_source_ip", "socket-peer"),
"operation": grant.operation,
"branch": data["branch"],
"path": data["path"],
"idempotency_key": data["idempotency_key"],
"operation_digest": digest,
"path_policy_sha256": policy.configured_path_policy_sha256(),
"pre_head_oid": record["pre_head_oid"],
"commit_oid": record.get("commit_oid"),
"result": result,
})
@staticmethod
def _result(record: dict[str, Any], *, idempotent: bool) -> dict[str, Any]:
return {
"status": "completed",
"state": record["state"],
"operation_id": record["operation_id"],
"branch": record["branch"],
"path": record["path"],
"commit_oid": record.get("commit_oid"),
"push_status": record.get("push_status"),
"verification_status": record.get("verification_status"),
"idempotent": idempotent,
}
@@ -0,0 +1,422 @@
"""Strict service-level exact-path policy for isolated Git writers."""
from __future__ import annotations
import grp
import hashlib
import json
import os
import re
import stat
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Any
from urllib.parse import unquote
MAX_POLICY_BYTES = 65_536
POLICY_FIELDS = frozenset(
{
"schema_version",
"policy_id",
"audience",
"repository_identity",
"allowed_branch",
"allowed_operations",
}
)
OPERATION_FIELDS = frozenset({"exact_paths", "max_content_bytes", "content_profile"})
SUPPORTED_OPERATIONS = frozenset({"commit_file.atomic"})
SUPPORTED_CONTENT_PROFILES = frozenset({"systemd-codex-review-worker-v1"})
_SECRET_RE = re.compile(
r"(?i)(authorization\s*:|bearer\s+|(?:token|password|secret)\s*[:=])"
)
_SYSTEMD_NAME_RE = re.compile(r"[A-Za-z][A-Za-z0-9]*\Z")
_SYSTEMD_MAX_LINE_BYTES = 4096
_SYSTEMD_ALLOWED_DIRECTIVES = {
"Unit": frozenset({"Description", "After", "Wants"}),
"Service": frozenset(
{
"Type",
"User",
"Group",
"WorkingDirectory",
"Environment",
"ExecStart",
"Restart",
"RestartSec",
"NoNewPrivileges",
"PrivateTmp",
"PrivateDevices",
"ProtectSystem",
"ProtectHome",
"ProtectKernelTunables",
"ProtectKernelModules",
"ProtectControlGroups",
"LockPersonality",
"RestrictSUIDSGID",
"RestrictRealtime",
"MemoryDenyWriteExecute",
"RuntimeDirectory",
"StateDirectory",
"CacheDirectory",
"ReadOnlyPaths",
"ReadWritePaths",
"UMask",
}
),
"Install": frozenset({"WantedBy"}),
}
_SYSTEMD_MULTI_DIRECTIVES = frozenset(
{
("Service", "Environment"),
("Service", "ReadOnlyPaths"),
("Service", "ReadWritePaths"),
}
)
_SYSTEMD_EXACT_SINGLE_VALUES = {
("Unit", "Description"): "AIOS Codex Review/Test Worker",
("Unit", "After"): "network-online.target",
("Unit", "Wants"): "network-online.target",
("Service", "Type"): "simple",
("Service", "User"): "aiman",
("Service", "Group"): "gitops",
(
"Service",
"WorkingDirectory",
): "/usr/libexec/aios-codex-review-test-worker/current",
(
"Service",
"ExecStart",
): "/usr/bin/python3 -m task_plane.workers.codex_review_test",
("Service", "Restart"): "on-failure",
("Service", "RestartSec"): "2",
("Service", "NoNewPrivileges"): "true",
("Service", "PrivateTmp"): "true",
("Service", "PrivateDevices"): "true",
("Service", "ProtectSystem"): "strict",
("Service", "ProtectHome"): "true",
("Service", "ProtectKernelTunables"): "true",
("Service", "ProtectKernelModules"): "true",
("Service", "ProtectControlGroups"): "true",
("Service", "LockPersonality"): "true",
("Service", "RestrictSUIDSGID"): "true",
("Service", "RestrictRealtime"): "true",
("Service", "MemoryDenyWriteExecute"): "true",
("Service", "RuntimeDirectory"): "aios-codex-review-test-worker",
("Service", "StateDirectory"): "aios-codex-review-test-worker",
("Service", "CacheDirectory"): "aios-codex-review-test-worker",
("Service", "UMask"): "0077",
("Install", "WantedBy"): "multi-user.target",
}
_SYSTEMD_EXACT_MULTI_VALUES = {
("Service", "Environment"): (
"AIOS_CODEX_WORKER_MODE=shadow",
"AIOS_TASK_PLANE_EXECUTION_ENABLED=false",
"PYTHONDONTWRITEBYTECODE=1",
),
("Service", "ReadOnlyPaths"): (
"/data/aios-gitops/repo.git",
"/usr/libexec/aios-codex-review-test-worker",
),
("Service", "ReadWritePaths"): (
"/run/aios-codex-review-test-worker",
"/var/lib/aios-codex-review-test-worker",
"/var/cache/aios-codex-review-test-worker",
),
}
class PathPolicyError(PermissionError):
pass
@dataclass(frozen=True)
class OperationPolicy:
exact_paths: tuple[str, ...]
max_content_bytes: int
content_profile: str
@dataclass(frozen=True)
class ServicePathPolicy:
schema_version: int
policy_id: str
audience: str
repository_identity: str
allowed_branch: str
allowed_operations: dict[str, OperationPolicy]
sha256: str
def operation(self, name: str) -> OperationPolicy:
try:
return self.allowed_operations[name]
except KeyError:
raise PathPolicyError("isolated_writer_service_operation_rejected") from None
def require_exact_path(self, operation: str, path: str) -> OperationPolicy:
selected = self.operation(operation)
if path not in selected.exact_paths:
raise PathPolicyError("isolated_writer_service_path_rejected")
return selected
def _canonical_json(value: Any) -> bytes:
return (
json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+ "\n"
).encode("utf-8")
def validate_relative_path_syntax(path: str) -> str:
if not isinstance(path, str):
raise PathPolicyError("isolated_writer_path_rejected")
if not path or path != path.strip() or "\x00" in path or "\\" in path:
raise PathPolicyError("isolated_writer_path_rejected")
try:
decoded = unquote(path, errors="strict")
except (UnicodeDecodeError, ValueError):
raise PathPolicyError("isolated_writer_path_rejected") from None
if decoded != path:
raise PathPolicyError("isolated_writer_path_rejected")
posix = PurePosixPath(path)
windows = PureWindowsPath(path)
parts = path.split("/")
if (
posix.is_absolute()
or windows.is_absolute()
or bool(windows.drive)
or any(part in {"", ".", ".."} for part in parts)
):
raise PathPolicyError("isolated_writer_path_rejected")
return path
def _require_fields(value: dict[str, Any], allowed: frozenset[str], code: str) -> None:
if set(value) != allowed:
raise PathPolicyError(code)
def _parse_policy(document: Any, canonical: bytes) -> ServicePathPolicy:
if not isinstance(document, dict):
raise PathPolicyError("isolated_writer_path_policy_object_required")
_require_fields(document, POLICY_FIELDS, "isolated_writer_path_policy_fields_rejected")
if document["schema_version"] != 1:
raise PathPolicyError("isolated_writer_path_policy_schema_rejected")
text_fields = (
"policy_id",
"audience",
"repository_identity",
"allowed_branch",
)
if any(
not isinstance(document[field], str) or not document[field].strip()
for field in text_fields
):
raise PathPolicyError("isolated_writer_path_policy_value_rejected")
raw_operations = document["allowed_operations"]
if not isinstance(raw_operations, dict) or not raw_operations:
raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
if not set(raw_operations).issubset(SUPPORTED_OPERATIONS):
raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
operations: dict[str, OperationPolicy] = {}
for name, raw in raw_operations.items():
if not isinstance(raw, dict):
raise PathPolicyError("isolated_writer_path_policy_operation_rejected")
_require_fields(
raw,
OPERATION_FIELDS,
"isolated_writer_path_policy_operation_fields_rejected",
)
paths = raw["exact_paths"]
if (
not isinstance(paths, list)
or not paths
or any(not isinstance(item, str) for item in paths)
):
raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
validated = tuple(validate_relative_path_syntax(item) for item in paths)
if len(set(validated)) != len(validated):
raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
maximum = raw["max_content_bytes"]
if not isinstance(maximum, int) or isinstance(maximum, bool) or not 1 <= maximum <= 1_048_576:
raise PathPolicyError("isolated_writer_path_policy_size_rejected")
profile = raw["content_profile"]
if profile not in SUPPORTED_CONTENT_PROFILES:
raise PathPolicyError("isolated_writer_path_policy_profile_rejected")
operations[name] = OperationPolicy(validated, maximum, profile)
return ServicePathPolicy(
schema_version=1,
policy_id=document["policy_id"],
audience=document["audience"],
repository_identity=document["repository_identity"],
allowed_branch=document["allowed_branch"],
allowed_operations=operations,
sha256=hashlib.sha256(canonical).hexdigest(),
)
def load_path_policy(
path: Path,
*,
expected_owner_uid: int = 0,
expected_group_gid: int | None = None,
expected_parent_mode: int = 0o750,
expected_file_mode: int = 0o640,
) -> ServicePathPolicy:
lexical = Path(os.path.abspath(str(path)))
if lexical.is_symlink():
raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
try:
parent_info = lexical.parent.stat(follow_symlinks=False)
file_info = lexical.stat(follow_symlinks=False)
except FileNotFoundError:
raise PathPolicyError("isolated_writer_path_policy_missing") from None
if stat.S_ISLNK(parent_info.st_mode) or not stat.S_ISDIR(parent_info.st_mode):
raise PathPolicyError("isolated_writer_path_policy_parent_rejected")
if stat.S_ISLNK(file_info.st_mode) or not stat.S_ISREG(file_info.st_mode):
raise PathPolicyError("isolated_writer_path_policy_file_rejected")
if lexical.resolve(strict=True) != lexical:
raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
if (
parent_info.st_uid != expected_owner_uid
or stat.S_IMODE(parent_info.st_mode) != expected_parent_mode
):
raise PathPolicyError("isolated_writer_path_policy_parent_permissions_rejected")
if (
file_info.st_uid != expected_owner_uid
or stat.S_IMODE(file_info.st_mode) != expected_file_mode
):
raise PathPolicyError("isolated_writer_path_policy_permissions_rejected")
if expected_group_gid is not None and (
parent_info.st_gid != expected_group_gid or file_info.st_gid != expected_group_gid
):
raise PathPolicyError("isolated_writer_path_policy_group_rejected")
if file_info.st_size <= 0 or file_info.st_size > MAX_POLICY_BYTES:
raise PathPolicyError("isolated_writer_path_policy_file_size_rejected")
raw = lexical.read_bytes()
try:
document = json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError):
raise PathPolicyError("isolated_writer_path_policy_json_rejected") from None
canonical = _canonical_json(document)
if raw != canonical:
raise PathPolicyError("isolated_writer_path_policy_not_canonical")
return _parse_policy(document, canonical)
@lru_cache(maxsize=1)
def configured_path_policy() -> ServicePathPolicy | None:
configured = os.getenv("AIOS_GITOPS_PATH_POLICY_FILE", "").strip()
if not configured:
return None
if not Path(configured).is_absolute():
raise PathPolicyError("isolated_writer_path_policy_path_rejected")
try:
gitops_gid = grp.getgrnam("gitops").gr_gid
except KeyError:
raise PathPolicyError("isolated_writer_path_policy_group_rejected") from None
loaded = load_path_policy(Path(configured), expected_group_gid=gitops_gid)
if loaded.audience != os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", ""):
raise PathPolicyError("isolated_writer_path_policy_audience_rejected")
if loaded.repository_identity != os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY", ""):
raise PathPolicyError("isolated_writer_path_policy_repository_rejected")
if loaded.allowed_branch != os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", ""):
raise PathPolicyError("isolated_writer_path_policy_branch_rejected")
return loaded
def clear_path_policy_cache() -> None:
configured_path_policy.cache_clear()
def require_service_path(operation: str, path: str) -> OperationPolicy:
loaded = configured_path_policy()
if loaded is None:
raise PathPolicyError("isolated_writer_service_path_policy_required")
return loaded.require_exact_path(operation, path)
def configured_path_policy_sha256() -> str | None:
loaded = configured_path_policy()
return None if loaded is None else loaded.sha256
def validate_content(operation: str, path: str, content: str) -> None:
selected = require_service_path(operation, path)
try:
encoded = content.encode("utf-8", errors="strict")
except (AttributeError, UnicodeEncodeError):
raise PathPolicyError("isolated_writer_systemd_profile_rejected") from None
if len(encoded) > selected.max_content_bytes:
raise PathPolicyError("isolated_writer_service_content_too_large")
if selected.content_profile == "systemd-codex-review-worker-v1":
_validate_systemd_codex_review_worker(content)
return
raise PathPolicyError("isolated_writer_path_policy_profile_rejected")
def _validate_systemd_codex_review_worker(content: str) -> None:
if not isinstance(content, str):
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
try:
content.encode("utf-8", errors="strict")
except UnicodeEncodeError:
raise PathPolicyError("isolated_writer_systemd_profile_rejected") from None
if (
"\x00" in content
or "\r" in content
or _SECRET_RE.search(content)
or not content.endswith("\n")
):
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
sections: set[str] = set()
values: dict[tuple[str, str], list[str]] = {}
section: str | None = None
for line in content.splitlines():
if len(line.encode("utf-8")) > _SYSTEMD_MAX_LINE_BYTES:
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
if not line:
continue
if line[0].isspace() or line.startswith(("#", ";")):
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
if line.startswith("[") and line.endswith("]"):
candidate = line[1:-1]
if (
not _SYSTEMD_NAME_RE.fullmatch(candidate)
or candidate not in _SYSTEMD_ALLOWED_DIRECTIVES
or candidate in sections
):
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
sections.add(candidate)
section = candidate
continue
if section is None or "=" not in line:
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
directive, value = line.split("=", 1)
if (
not _SYSTEMD_NAME_RE.fullmatch(directive)
or directive not in _SYSTEMD_ALLOWED_DIRECTIVES[section]
or not value
or value != value.strip()
):
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
key = (section, directive)
if key in values and key not in _SYSTEMD_MULTI_DIRECTIVES:
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
values.setdefault(key, []).append(value)
if sections != set(_SYSTEMD_ALLOWED_DIRECTIVES):
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
if set(values) != set(_SYSTEMD_EXACT_SINGLE_VALUES) | set(
_SYSTEMD_EXACT_MULTI_VALUES
):
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
for key, expected in _SYSTEMD_EXACT_SINGLE_VALUES.items():
if values.get(key) != [expected]:
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
for key, expected in _SYSTEMD_EXACT_MULTI_VALUES.items():
observed = values.get(key, [])
if len(observed) != len(set(observed)) or tuple(observed) != expected:
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
@@ -0,0 +1,241 @@
"""Digest-only, task-scoped token registry for isolated writers."""
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import os
import grp
import pwd
import secrets
import sqlite3
import stat
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
_REGISTRY_INIT_LOCK = threading.Lock()
def utc_now() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def token_digest(token: str) -> str:
return hashlib.sha256(token.encode("utf-8")).hexdigest()
class TokenRegistry:
def __init__(self, path: Path | str):
self.path = Path(path)
self._prepare()
def _prepare(self) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
if self.path.parent.is_symlink() or self.path.is_symlink():
raise PermissionError("writer_token_registry_symlink_rejected")
with _REGISTRY_INIT_LOCK:
with self._connect() as db:
db.executescript(
"""
CREATE TABLE IF NOT EXISTS token_grants (
token_id TEXT PRIMARY KEY,
token_digest TEXT NOT NULL UNIQUE,
grant_json TEXT NOT NULL,
revoked INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS token_uses (
token_id TEXT NOT NULL,
operation TEXT NOT NULL,
use_count INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY(token_id, operation),
FOREIGN KEY(token_id) REFERENCES token_grants(token_id)
);
"""
)
os.chmod(self.path, 0o600)
def _connect(self) -> sqlite3.Connection:
db = sqlite3.connect(self.path, timeout=30, isolation_level=None)
db.row_factory = sqlite3.Row
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA synchronous=FULL")
db.execute("PRAGMA foreign_keys=ON")
return db
def issue(self, grant: dict[str, Any], token: str | None = None) -> tuple[str, dict[str, Any]]:
value = token or secrets.token_urlsafe(32)
if len(value.encode("utf-8")) < 32:
raise ValueError("writer_token_entropy_insufficient")
record = dict(grant)
record["schema_version"] = 2
record["token_digest"] = token_digest(value)
record["revoked"] = False
record.setdefault("created_at", utc_now())
required = (
"token_id",
"audience",
"task_id",
"source_cidrs",
"allowed_branch",
"approved_base_oid",
"allowed_operations",
"allowed_paths",
"not_before",
"expires_at",
"max_uses",
"issued_by",
)
if any(not record.get(field) for field in required):
raise ValueError("writer_token_grant_incomplete")
encoded = json.dumps(record, sort_keys=True, separators=(",", ":"))
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
try:
db.execute(
"""
INSERT INTO token_grants(token_id, token_digest, grant_json, revoked, created_at)
VALUES(?,?,?,?,?)
""",
(
str(record["token_id"]),
str(record["token_digest"]),
encoded,
0,
str(record["created_at"]),
),
)
db.execute("COMMIT")
except Exception:
db.execute("ROLLBACK")
raise
return value, record
def find(self, token: str) -> dict[str, Any] | None:
candidate = token_digest(token)
with self._connect() as db:
rows = db.execute(
"SELECT token_digest, grant_json, revoked FROM token_grants"
).fetchall()
match = None
for row in rows:
if hmac.compare_digest(str(row["token_digest"]), candidate):
match = dict(json.loads(str(row["grant_json"])))
match["revoked"] = bool(row["revoked"])
return match
def consume(self, token_id: str, operation: str, maximum: int) -> int:
if maximum < 1:
raise PermissionError("writer_token_operation_exhausted")
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
try:
row = db.execute(
"SELECT revoked FROM token_grants WHERE token_id=?",
(token_id,),
).fetchone()
if row is None or bool(row["revoked"]):
raise PermissionError("writer_token_revoked")
used_row = db.execute(
"SELECT use_count FROM token_uses WHERE token_id=? AND operation=?",
(token_id, operation),
).fetchone()
used = int(used_row["use_count"]) if used_row else 0
if used >= maximum:
raise PermissionError("writer_token_operation_exhausted")
current = used + 1
db.execute(
"""
INSERT INTO token_uses(token_id, operation, use_count)
VALUES(?,?,?)
ON CONFLICT(token_id, operation)
DO UPDATE SET use_count=excluded.use_count
""",
(token_id, operation, current),
)
db.execute("COMMIT")
return current
except Exception:
db.execute("ROLLBACK")
raise
def revoke(self, token_id: str) -> bool:
with self._connect() as db:
db.execute("BEGIN IMMEDIATE")
changed = db.execute(
"UPDATE token_grants SET revoked=1 WHERE token_id=? AND revoked=0",
(token_id,),
).rowcount
db.execute("COMMIT")
return bool(changed)
def usage(self, token_id: str, operation: str) -> int:
with self._connect() as db:
row = db.execute(
"SELECT use_count FROM token_uses WHERE token_id=? AND operation=?",
(token_id, operation),
).fetchone()
return int(row["use_count"]) if row else 0
def _write_secret_file(
path: Path,
token: str,
*,
owner: str | None = None,
group: str | None = None,
) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
if path.parent.is_symlink() or path.is_symlink():
raise PermissionError("writer_token_file_symlink_rejected")
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
fd = os.open(path, flags, 0o600)
try:
if not stat.S_ISREG(os.fstat(fd).st_mode):
raise PermissionError("writer_token_file_not_regular")
if owner is not None or group is not None:
uid = pwd.getpwnam(owner).pw_uid if owner is not None else -1
gid = grp.getgrnam(group).gr_gid if group is not None else -1
os.fchown(fd, uid, gid)
os.write(fd, token.encode("utf-8"))
os.fsync(fd)
finally:
os.close(fd)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--registry", required=True)
sub = parser.add_subparsers(dest="command", required=True)
issue = sub.add_parser("issue")
issue.add_argument("--grant-file", required=True)
issue.add_argument("--token-file", required=True)
issue.add_argument("--owner")
issue.add_argument("--group")
revoke = sub.add_parser("revoke")
revoke.add_argument("--token-id", required=True)
args = parser.parse_args(argv)
registry = TokenRegistry(args.registry)
if args.command == "issue":
grant = json.loads(Path(args.grant_file).read_text(encoding="utf-8"))
token, record = registry.issue(grant)
_write_secret_file(
Path(args.token_file),
token,
owner=args.owner,
group=args.group,
)
print(json.dumps({"token_id": record["token_id"], "created": True}, sort_keys=True))
return 0
changed = registry.revoke(args.token_id)
print(json.dumps({"token_id": args.token_id, "revoked": changed}, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -1 +0,0 @@
"""Isolated, Shadow-attested Codex Task Plane Worker components."""
@@ -1,212 +0,0 @@
"""Shadow-only Codex read/review/test worker lifecycle and CLI."""
from __future__ import annotations
import argparse
import json
import os
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable
from .result_attestation import TaskResultEnvelope, TrustPolicy, task_result_envelope
from .task_client import CredentialSource, StageAClientConfig, TaskPlaneClient
MAX_CONCURRENCY = 1
HEARTBEAT_INTERVAL_SECONDS = 15
CLAIM_LEASE_SECONDS = 120
SHADOW_ONLY = True
DEFAULT_TRUST_LEVEL = TrustPolicy.SHADOW_ATTESTED
DEFAULT_PRODUCTION_TRUST = False
READONLY_CAPABILITIES = ("code.read", "test.python", "review.security")
class CodexReviewTaskError(ValueError):
pass
@dataclass(frozen=True)
class CodexReviewTaskPolicy:
worker_id: str
identity: str
trust_zone: str = "isolated-review"
max_concurrency: int = MAX_CONCURRENCY
heartbeat_interval_seconds: int = HEARTBEAT_INTERVAL_SECONDS
claim_lease_seconds: int = CLAIM_LEASE_SECONDS
shadow_only: bool = SHADOW_ONLY
def registration_payload(self) -> dict[str, object]:
return {
"worker_id": self.worker_id, "worker_type": "codex-review",
"identity": self.identity, "owner_identity": self.identity,
"capabilities": list(READONLY_CAPABILITIES), "trust_zone": self.trust_zone,
"max_concurrency": self.max_concurrency,
"heartbeat_interval_seconds": self.heartbeat_interval_seconds,
"supported_protocols": ["aios-task-v1"], "registration_policy": "self",
}
def assert_shadow_only_task(self, task: dict[str, Any]) -> None:
plan = task.get("plan", task)
metadata = plan.get("metadata", {}) if isinstance(plan, dict) else {}
required = set(plan.get("required_capabilities", [])) if isinstance(plan, dict) else set()
if self.max_concurrency != 1 or self.heartbeat_interval_seconds != 15 or self.claim_lease_seconds != 120:
raise CodexReviewTaskError("worker timing/concurrency contract rejected")
if not isinstance(plan, dict) or metadata.get("production_side_effects") is not False:
raise CodexReviewTaskError("task must declare production_side_effects=false")
if plan.get("risk") != "low" or plan.get("task_type") not in {"code.review", "test.execute", None}:
raise CodexReviewTaskError("task risk or type rejected")
if plan.get("data_classification", "internal") not in {"public", "internal"}:
raise CodexReviewTaskError("task classification rejected")
if not required.issubset(set(READONLY_CAPABILITIES)):
raise CodexReviewTaskError("task capability rejected")
if any(value.startswith(("code.write", "deploy", "secret", "production")) for value in required):
raise CodexReviewTaskError("side-effect capability rejected")
def default_result_envelope(self, *, task_id: str, claim_id: str, worker_identity: str) -> TaskResultEnvelope:
return task_result_envelope(
task_id=task_id, claim_id=claim_id, worker_id=self.worker_id,
worker_identity=worker_identity, status="CREATED", outputs=[], evidence=[],
started_at="1970-01-01T00:00:00Z", completed_at="1970-01-01T00:00:00Z",
)
class _LeaseGuard:
def __init__(self, client: TaskPlaneClient, task_id: str, claim_id: str) -> None:
self.client, self.task_id, self.claim_id = client, task_id, claim_id
self.stop_event = threading.Event()
self.lost = threading.Event()
self.thread = threading.Thread(target=self._run, daemon=True)
def _run(self) -> None:
while not self.stop_event.wait(HEARTBEAT_INTERVAL_SECONDS):
try:
self.client.claim_heartbeat(self.task_id, claim_id=self.claim_id, lease_seconds=CLAIM_LEASE_SECONDS)
except Exception:
self.lost.set()
self.stop_event.set()
def __enter__(self) -> "_LeaseGuard":
self.thread.start()
return self
def __exit__(self, *_: object) -> None:
self.stop_event.set()
self.thread.join(timeout=2)
def require_active(self) -> None:
if self.lost.is_set():
raise CodexReviewTaskError("claim lease lost")
class CodexReviewWorker:
def __init__(self, client: TaskPlaneClient, policy: CodexReviewTaskPolicy, source_commit: str) -> None:
self.client, self.policy, self.source_commit = client, policy, source_commit
def register(self) -> dict[str, Any]:
return self.client.register_worker(**{
"worker_id": self.policy.worker_id, "worker_type": "codex-review",
"identity": self.policy.identity, "capabilities": READONLY_CAPABILITIES,
"trust_zone": self.policy.trust_zone, "max_concurrency": 1,
"heartbeat_interval_seconds": 15,
})
def poll_safe_task(self) -> dict[str, Any] | None:
for task in self.client.list_tasks():
if task.get("status") not in {"CREATED", "QUEUED", "FAILED_RECOVERABLE"}:
continue
try:
full = self.client.get_task(str(task["task_id"]))
self.policy.assert_shadow_only_task(full)
return full
except (KeyError, CodexReviewTaskError):
continue
return None
def run_once(self, executor: Callable[[dict[str, Any], Callable[[float, str], None]], tuple[str, list[dict[str, Any]], list[dict[str, Any]]]]) -> str | None:
self.register()
task = self.poll_safe_task()
if task is None:
return None
task_id = str(task["task_id"])
claim = self.client.claim(task_id, worker_id=self.policy.worker_id, lease_seconds=120)
claim_id = str(claim["claim_id"])
started = datetime.now(timezone.utc).isoformat()
with _LeaseGuard(self.client, task_id, claim_id) as guard:
def emit(value: float, stage: str) -> None:
guard.require_active()
self.client.progress(task_id, claim_id=claim_id, progress=value, stage=stage)
summary, outputs, evidence = executor(task, emit)
guard.require_active()
completed = datetime.now(timezone.utc).isoformat()
envelope = task_result_envelope(
task_id=task_id, claim_id=claim_id,
task_revision=int(claim["task_revision"]),
worker_id=self.policy.worker_id, worker_identity=self.policy.identity,
operation_digest=str(claim["operation_digest"]),
source_commit=self.source_commit, status="COMPLETED", summary=summary,
outputs=outputs, evidence=evidence, started_at=started, completed_at=completed,
)
guard.require_active()
envelope_payload = envelope.to_dict()
self.client.result(
task_id,
claim_id=claim_id,
status=str(envelope_payload["status"]),
outputs=list(envelope_payload["outputs"]),
evidence=list(envelope_payload["evidence"]),
started_at=str(envelope_payload["started_at"]),
completed_at=str(envelope_payload["completed_at"]),
trust_level=TrustPolicy.SHADOW_ATTESTED.value,
production_trust=False,
)
return task_id
def build_default_task_policy(worker_id: str, worker_identity: str) -> CodexReviewTaskPolicy:
return CodexReviewTaskPolicy(worker_id=worker_id, identity=worker_identity)
def make_shadow_result_payload(*, task_id: str, claim_id: str, worker_identity: str) -> TaskResultEnvelope:
return task_result_envelope(
task_id=task_id, claim_id=claim_id, worker_identity=worker_identity,
status="COMPLETED", outputs=[], evidence=[],
started_at="1970-01-01T00:00:00Z", completed_at="1970-01-01T00:00:00Z",
)
def _assert_disabled_production() -> None:
for name in ("AIOS_GITOPS_WRITES_ENABLED", "AIOS_SILENT_RELEASE_EXECUTION_ENABLED", "AIOS_TASK_PLANE_EXECUTION_ENABLED"):
if os.environ.get(name, "false").lower() != "false":
raise CodexReviewTaskError(f"{name} must remain false")
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--self-check", action="store_true")
parser.add_argument("--once", action="store_true")
args = parser.parse_args(argv)
_assert_disabled_production()
config = StageAClientConfig(os.environ.get("AIOS_TASK_PLANE_ENDPOINT", "http://127.0.0.1:18923"))
if args.self_check:
print(json.dumps({"status": "PASS", "shadow_only": True, "production_trust": False}, sort_keys=True))
return 0
if os.environ.get("AIOS_CODEX_WORKER_SHADOW_CANARY_ENABLED", "false").lower() != "true":
raise SystemExit("Codex worker execution is disabled")
credential = CredentialSource(Path(os.environ["AIOS_TASK_PLANE_WORKER_TOKEN_FILE"])).read_token()
source_commit = os.environ.get("AIOS_CODEX_WORKER_SOURCE_COMMIT", "")
policy = CodexReviewTaskPolicy(
worker_id=os.environ["AIOS_CODEX_WORKER_ID"],
identity=os.environ["AIOS_CODEX_WORKER_IDENTITY"],
)
worker = CodexReviewWorker(TaskPlaneClient(config, credential), policy, source_commit)
if args.once:
worker.run_once(lambda _task, emit: (emit(100, "VERIFIED") or ("shadow self-check", [], [])))
return 0
raise SystemExit("continuous real worker execution is not enabled")
if __name__ == "__main__":
raise SystemExit(main())
@@ -1,62 +0,0 @@
"""Fail-closed command and environment policy for the isolated Codex Worker."""
from __future__ import annotations
import os
import re
import subprocess
from pathlib import Path
OID_RE = re.compile(r"[0-9a-f]{40}")
DENIED_ENV = ("TOKEN", "PASSWORD", "SECRET", "CREDENTIAL", "AUTHORIZATION", "COOKIE", "PRIVATE_KEY", "AWS_", "GITHUB_", "GITEA_")
class CommandPolicyError(ValueError):
pass
def isolated_environment(home: Path, temp_root: Path, pythonpath: str) -> dict[str, str]:
home.mkdir(mode=0o700, parents=True, exist_ok=True)
temp_root.mkdir(mode=0o700, parents=True, exist_ok=True)
allowed = {key: value for key, value in os.environ.items() if key in {"PATH", "LANG", "LC_ALL"}}
allowed.update({"HOME": str(home), "TMPDIR": str(temp_root), "PYTHONPATH": pythonpath, "PYTHONDONTWRITEBYTECODE": "1"})
if any(any(marker in key.upper() for marker in DENIED_ENV) for key in allowed):
raise CommandPolicyError("secret environment retained")
return allowed
class CommandPolicy:
def __init__(self, repo: Path, result_root: Path, *, timeout_seconds: int = 300, output_limit: int = 1_000_000):
self.repo = repo.resolve()
self.result_root = result_root.resolve()
self.timeout_seconds = timeout_seconds
self.output_limit = output_limit
def _oid(self, value: str) -> str:
if not OID_RE.fullmatch(value):
raise CommandPolicyError("exact commit oid required")
return value
def build(self, kind: str, *, oid: str, paths: list[str] | None = None, output: Path | None = None) -> list[str]:
oid = self._oid(oid)
paths = paths or []
if any(not path or path.startswith("-") or Path(path).is_absolute() or ".." in Path(path).parts for path in paths):
raise CommandPolicyError("invalid allowlisted path")
if kind == "diff-check":
return ["git", "-C", str(self.repo), "diff", "--check"]
if kind == "git-show" and len(paths) == 1:
return ["git", "-C", str(self.repo), "show", f"{oid}:{paths[0]}"]
if kind == "pytest" and paths and all(path.startswith("tests/task_plane/") or path == "tests/test_isolated_writer_policy.py" for path in paths):
return ["python3", "-m", "pytest", "-q", *paths]
if kind == "compileall" and paths and all(path.startswith(("active-runtime/app/task_plane", "runtime/agent-os")) for path in paths):
return ["python3", "-m", "compileall", "-q", *paths]
if kind == "acceptance" and output is not None:
target = output.resolve()
if target.parent != self.result_root:
raise CommandPolicyError("output escapes result root")
return ["bash", "scripts/task-plane/run_task_plane_acceptance.sh", str(target)]
raise CommandPolicyError("command template not allowlisted")
def run(self, argv: list[str], env: dict[str, str]) -> tuple[int, str]:
completed = subprocess.run(argv, cwd=self.repo, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=self.timeout_seconds, start_new_session=True, check=False)
output = completed.stdout[: self.output_limit]
return completed.returncode, output
@@ -1,56 +0,0 @@
"""Bounded progress payload helpers."""
from __future__ import annotations
import math
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class ProgressEnvelope:
claim_id: str
progress: int | float
stage: str
lease_seconds: int = 120
message: str | None = None
def validate(self) -> None:
if not self.claim_id:
raise ValueError("claim_id is required")
numeric = float(self.progress)
if not math.isfinite(numeric) or not 0 <= numeric <= 100:
raise ValueError("progress must be finite and within [0, 100]")
if not 1 <= self.lease_seconds <= 3600:
raise ValueError("lease_seconds out of range")
if not self.stage or len(self.stage) > 64 or any(ord(ch) < 32 for ch in self.stage):
raise ValueError("stage is invalid")
if self.message is not None and len(self.message) > 512:
raise ValueError("message is too long")
def as_request(self) -> dict[str, Any]:
self.validate()
return {
"claim_id": self.claim_id, "progress": float(self.progress),
"stage": self.stage, "lease_seconds": self.lease_seconds,
"message": self.message,
}
def clamp_progress(value: float) -> float:
numeric = float(value)
if not math.isfinite(numeric):
raise ValueError("progress must be finite")
return max(0.0, min(100.0, numeric))
def safe_stage_name(stage: str | None) -> str:
value = str(stage or "").strip()
if not value:
return "RUNNING"
if any(ord(ch) < 32 for ch in value):
raise ValueError("stage is invalid")
return value[:64]
def build_progress_payload(*, claim_id: str, progress: float, stage: str = "RUNNING", lease_seconds: int = 120, message: str | None = None) -> dict[str, Any]:
return ProgressEnvelope(claim_id, clamp_progress(progress), safe_stage_name(stage), lease_seconds, message).as_request()
@@ -1,157 +0,0 @@
"""Canonical, Shadow-only result envelopes for the Codex review worker."""
from __future__ import annotations
import hashlib
import hmac
import json
import re
from dataclasses import dataclass
from enum import StrEnum
from typing import Any
OID_RE = re.compile(r"^[0-9a-f]{40}$")
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
class TrustPolicy(StrEnum):
SHADOW_ATTESTED = "SHADOW_ATTESTED"
SHADOW_UNATTESTED = "SHADOW_UNATTESTED"
def _canonical_sha256(payload: object) -> str:
encoded = json.dumps(
payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")
).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
@dataclass(frozen=True)
class TaskResultEnvelope:
task_id: str
claim_id: str
worker_identity: str
status: str
outputs: list[dict[str, Any]]
evidence: list[dict[str, Any]]
started_at: str
completed_at: str
task_revision: int = 1
worker_id: str = ""
operation_digest: str = ""
source_commit: str = "0" * 40
summary: str = ""
evidence_digests: tuple[str, ...] = ()
result_digest: str | None = None
trust_level: TrustPolicy = TrustPolicy.SHADOW_ATTESTED
production_trust: bool = False
signature: str | None = None
attestation_ref: str | None = None
def digest_payload(self) -> dict[str, Any]:
return {
"task_id": self.task_id,
"claim_id": self.claim_id,
"task_revision": self.task_revision,
"worker_id": self.worker_id,
"worker_identity": self.worker_identity,
"operation_digest": self.operation_digest,
"source_commit": self.source_commit,
"status": self.status,
"summary": self.summary,
"outputs": self.outputs,
"evidence": self.evidence,
"evidence_digests": list(self.evidence_digests),
"started_at": self.started_at,
"completed_at": self.completed_at,
"trust_level": TrustPolicy.SHADOW_ATTESTED.value,
"production_trust": False,
"attestation_ref": self.attestation_ref,
}
def computed_digest(self) -> str:
return _canonical_sha256(self.digest_payload())
def validate(self) -> None:
required = {
"task_id": self.task_id,
"claim_id": self.claim_id,
"worker_id": self.worker_id,
"worker_identity": self.worker_identity,
"operation_digest": self.operation_digest,
"status": self.status,
"started_at": self.started_at,
"completed_at": self.completed_at,
}
if any(not str(value).strip() for value in required.values()):
raise ValueError("result envelope required field missing")
if self.task_revision < 1:
raise ValueError("task_revision must be positive")
if not OID_RE.fullmatch(self.source_commit):
raise ValueError("source_commit must be a full commit OID")
if not SHA256_RE.fullmatch(self.operation_digest):
raise ValueError("operation_digest must be SHA-256")
if any(not SHA256_RE.fullmatch(value) for value in self.evidence_digests):
raise ValueError("evidence digest must be SHA-256")
observed = tuple(_canonical_sha256(item) for item in self.evidence)
if observed != self.evidence_digests:
raise ValueError("evidence digest binding mismatch")
if self.trust_level is not TrustPolicy.SHADOW_ATTESTED:
raise ValueError("Codex review results must remain SHADOW_ATTESTED")
if self.production_trust is not False:
raise ValueError("production_trust is forbidden")
if self.result_digest is not None and not hmac.compare_digest(
self.result_digest, self.computed_digest()
):
raise ValueError("result digest mismatch")
def finalized(self) -> "TaskResultEnvelope":
from dataclasses import replace
envelope = replace(self, result_digest=self.computed_digest())
envelope.validate()
return envelope
def to_dict(self) -> dict[str, Any]:
envelope = self if self.result_digest is not None else self.finalized()
envelope.validate()
return {**envelope.digest_payload(), "result_digest": envelope.result_digest}
def task_result_envelope(
*,
task_id: str,
claim_id: str,
worker_identity: str,
status: str,
outputs: list[dict[str, Any]],
evidence: list[dict[str, Any]],
started_at: str,
completed_at: str,
task_revision: int = 1,
worker_id: str | None = None,
operation_digest: str | None = None,
source_commit: str = "0" * 40,
summary: str = "",
trust_level: TrustPolicy = TrustPolicy.SHADOW_ATTESTED,
production_trust: bool = False,
) -> TaskResultEnvelope:
evidence_digests = tuple(_canonical_sha256(item) for item in evidence)
envelope = TaskResultEnvelope(
task_id=task_id,
claim_id=claim_id,
task_revision=task_revision,
worker_id=worker_id or worker_identity,
worker_identity=worker_identity,
operation_digest=operation_digest or ("0" * 64),
source_commit=source_commit,
status=status,
summary=summary,
outputs=outputs,
evidence=evidence,
evidence_digests=evidence_digests,
started_at=started_at,
completed_at=completed_at,
trust_level=trust_level,
production_trust=production_trust,
)
return envelope.finalized()
@@ -1,138 +0,0 @@
"""Fixed-command, no-shell runner for a read-only Codex workspace."""
from __future__ import annotations
import os
import re
import signal
import subprocess
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from .command_policy import CommandPolicyError
from .workspace import WorkspacePermissionError, assert_private_path
OID_RE = re.compile(r"^[0-9a-f]{40}$")
FORBIDDEN_SNIPPETS = ("http://", "https://", ";", "&&", "||", "|", "`", "${", "$(")
class CommandRunnerError(ValueError):
pass
class CommandDisallowedError(CommandRunnerError):
pass
class CommandPolicyViolation(CommandRunnerError):
pass
@dataclass(frozen=True)
class CommandResult:
returncode: int
output: str
class SafeCommandRunner:
allowed_kinds = ("git-diff", "diff-check", "git-show", "pytest", "compileall", "acceptance")
def __init__(self, repo: Path, result_root: Path, *, policy=None, timeout_seconds: int = 300, output_limit: int = 1_000_000) -> None:
self.repo = assert_private_path(repo)
self.result_root = assert_private_path(result_root)
self.timeout_seconds, self.output_limit = timeout_seconds, output_limit
@staticmethod
def _oid(value: str) -> str:
if not OID_RE.fullmatch(str(value)):
raise CommandPolicyError("exact commit oid required")
return value
@staticmethod
def _forbidden(value: str) -> bool:
lowered = value.lower()
return (
re.search(r"\b(?:bash|eval)\b", lowered) is not None
or re.search(r"(^|\s)-c(\s|$)", lowered) is not None
or any(marker in value for marker in FORBIDDEN_SNIPPETS)
)
def _path(self, value: str, *, allowed_prefixes: tuple[str, ...]) -> str:
if not value or value.startswith("-") or "\\" in value or self._forbidden(value):
raise CommandDisallowedError("unsafe path input")
pure = PurePosixPath(value)
if pure.is_absolute() or any(part in {"", ".", ".."} for part in pure.parts):
raise CommandPolicyError("path traversal is forbidden")
if not any(value == prefix or value.startswith(prefix + "/") for prefix in allowed_prefixes):
raise CommandPolicyError("path outside allowlist")
try:
candidate = assert_private_path(self.repo / value)
except WorkspacePermissionError as exc:
raise CommandPolicyError("symlink path is forbidden") from exc
try:
candidate.relative_to(self.repo)
except ValueError:
raise CommandPolicyError("path escapes repository") from None
return value
def _output(self, output: Path) -> Path:
if Path(output).is_absolute() or Path(output).name != str(output):
raise CommandPolicyError("output must be a direct result file")
target = assert_private_path(self.result_root / output)
try:
target.relative_to(self.result_root)
except ValueError:
raise CommandPolicyError("output escapes result root") from None
return target
def build(self, kind: str, *, oid: str, other_oid: str | None = None, paths: list[str] | None = None, output: Path | None = None) -> list[str]:
if kind not in self.allowed_kinds:
raise CommandDisallowedError("command kind not allowed")
first = self._oid(oid)
values = paths or []
if kind == "git-diff":
second = self._oid(str(other_oid or ""))
return ["git", "-C", str(self.repo), "diff", "--no-ext-diff", "--no-textconv", first, second]
if kind == "diff-check":
return ["git", "-C", str(self.repo), "diff", "--check"]
if kind == "git-show" and len(values) == 1:
path = self._path(values[0], allowed_prefixes=("active-runtime", "runtime", "scripts", "tests", "docs"))
return ["git", "-C", str(self.repo), "show", f"{first}:{path}"]
if kind == "pytest" and values:
safe = [self._path(value, allowed_prefixes=("tests/task_plane", "tests/test_isolated_writer_policy.py")) for value in values]
return ["python3", "-m", "pytest", "-q", *safe]
if kind == "compileall" and values:
safe = [self._path(value, allowed_prefixes=("active-runtime/app/task_plane", "runtime/agent-os")) for value in values]
return ["python3", "-m", "compileall", "-q", *safe]
if kind == "acceptance" and output is not None and not values:
return ["bash", "scripts/task-plane/run_task_plane_acceptance.sh", str(self._output(output))]
raise CommandPolicyError("command template not allowlisted")
def run(self, kind: str, *, oid: str, other_oid: str | None = None, paths: list[str] | None = None, output: Path | None = None, env: dict[str, str] | None = None) -> CommandResult:
if env and any(marker in key.upper() for key in env for marker in ("TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "AUTHORIZATION", "COOKIE", "PRIVATE_KEY")):
raise CommandPolicyViolation("secret environment rejected")
command = self.build(kind, oid=oid, other_oid=other_oid, paths=paths, output=output)
safe_env = {
"HOME": str(self.result_root),
"PATH": os.environ.get("PATH", ""),
"LANG": "C.UTF-8",
"PYTHONDONTWRITEBYTECODE": "1",
"GIT_CONFIG_NOSYSTEM": "1",
"GIT_TERMINAL_PROMPT": "0",
}
process = subprocess.Popen(
command, cwd=self.repo, env=safe_env, stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
start_new_session=True,
)
try:
text, _ = process.communicate(timeout=self.timeout_seconds)
except subprocess.TimeoutExpired:
os.killpg(process.pid, signal.SIGKILL)
text, _ = process.communicate()
raise CommandRunnerError("command timeout")
return CommandResult(process.returncode, text[: self.output_limit])
Runner = SafeCommandRunner
FixedCommandRunner = SafeCommandRunner
CommandResultEnvelope = CommandResult
@@ -1,300 +0,0 @@
"""Fail-closed loopback HTTP client for the Codex review worker."""
from __future__ import annotations
import hashlib
import json
import os
import re
import stat
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable
MAX_HTTP_BODY_BYTES = 1_000_000
_RETRYABLE_STATUS = {429, 503}
_RETRY_BACKOFF_SECONDS = (0.25, 0.5, 1.0, 2.0)
_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
class TaskClientError(RuntimeError):
pass
class TaskClientProtocolError(TaskClientError):
pass
class TaskClientAuthError(TaskClientError):
pass
class TaskClientResponseTooLargeError(TaskClientError):
pass
def _safe_identifier(value: str, name: str) -> str:
candidate = str(value or "")
if not _IDENTIFIER_RE.fullmatch(candidate):
raise TaskClientProtocolError(f"invalid {name}")
return urllib.parse.quote(candidate, safe="")
def _stable_key(operation: str, payload: object) -> str:
encoded = json.dumps(
payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")
).encode("utf-8")
return f"codex-worker:{operation}:{hashlib.sha256(encoded).hexdigest()}"
@dataclass(frozen=True)
class CredentialSource:
token_path: Path
token_owner: str = "aiman"
token_group: str = "gitops"
expected_mode: int = 0o600
def read_token(self) -> str:
import grp
import pwd
path = Path(self.token_path)
flags = os.O_RDONLY
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
fd = os.open(path, flags)
except OSError as exc:
raise TaskClientAuthError("credential file unavailable") from exc
try:
info = os.fstat(fd)
if not stat.S_ISREG(info.st_mode):
raise TaskClientAuthError("credential file must be regular")
if stat.S_IMODE(info.st_mode) != self.expected_mode:
raise TaskClientAuthError("credential file mode rejected")
try:
owner = pwd.getpwuid(info.st_uid).pw_name
group = grp.getgrgid(info.st_gid).gr_name
except KeyError as exc:
raise TaskClientAuthError("credential ownership unresolved") from exc
if owner != self.token_owner or group != self.token_group:
raise TaskClientAuthError("credential ownership rejected")
value = os.read(fd, 4097)
if len(value) > 4096:
raise TaskClientAuthError("credential file too large")
token = value.decode("utf-8", errors="strict").strip()
if len(token.encode("utf-8")) < 32:
raise TaskClientAuthError("credential unavailable")
return token
finally:
os.close(fd)
@dataclass(frozen=True)
class StageAClientConfig:
base_url: str = "http://127.0.0.1:18923"
timeout_seconds: float = 12.0
max_request_bytes: int = MAX_HTTP_BODY_BYTES
max_response_bytes: int = MAX_HTTP_BODY_BYTES
max_retry_attempts: int = 4
default_lease_seconds: int = 120
default_worker_heartbeat_interval_seconds: int = 15
default_max_concurrency: int = 1
def __post_init__(self) -> None:
parsed = urllib.parse.urlsplit(self.base_url)
if (
parsed.scheme != "http"
or parsed.hostname not in {"127.0.0.1", "::1"}
or parsed.username is not None
or parsed.password is not None
or parsed.query
or parsed.fragment
or parsed.path not in {"", "/"}
or parsed.port is None
):
raise ValueError("Task Plane endpoint must be explicit loopback HTTP")
class TaskPlaneClient:
def __init__(self, config: StageAClientConfig, token: str) -> None:
if len(str(token).encode("utf-8")) < 16:
raise TaskClientAuthError("credential is required")
self.config = config
self.__token = str(token)
def __repr__(self) -> str:
return f"TaskPlaneClient(base_url={self.config.base_url!r}, credential=<redacted>)"
def _encode(self, payload: dict[str, Any] | None) -> bytes | None:
if payload is None:
return None
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
if len(body) > self.config.max_request_bytes:
raise TaskClientResponseTooLargeError("request payload exceeds configured limit")
return body
def _read_response(self, handle: Any) -> tuple[int, Any]:
status_code = int(getattr(handle, "status", 200))
raw = handle.read(self.config.max_response_bytes + 1)
if len(raw) > self.config.max_response_bytes:
raise TaskClientResponseTooLargeError("response payload exceeds configured limit")
if not raw:
return status_code, {}
try:
return status_code, json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise TaskClientProtocolError("response is not valid JSON") from exc
def _request(
self,
method: str,
path: str,
payload: dict[str, Any] | None = None,
*,
idempotency_key: str | None = None,
) -> tuple[int, Any]:
if not path.startswith("/") or "//" in path or "?" in path or "#" in path:
raise TaskClientProtocolError("invalid API path")
body = self._encode(payload)
attempts = max(1, min(self.config.max_retry_attempts, 4))
for index in range(attempts):
headers = {"Authorization": f"Bearer {self.__token}", "Accept": "application/json"}
if body is not None:
headers["Content-Type"] = "application/json"
if idempotency_key:
headers["Idempotency-Key"] = idempotency_key
request = urllib.request.Request(
f"{self.config.base_url.rstrip('/')}{path}",
method=method,
data=body,
headers=headers,
)
try:
with urllib.request.urlopen(request, timeout=self.config.timeout_seconds) as response:
return self._read_response(response)
except urllib.error.HTTPError as exc:
status_code, response = self._read_response(exc)
if status_code in {401, 403}:
raise TaskClientAuthError(f"request rejected: HTTP {status_code}")
if status_code in _RETRYABLE_STATUS and index + 1 < attempts:
time.sleep(_RETRY_BACKOFF_SECONDS[index])
continue
code = response.get("error") if isinstance(response, dict) else "request_failed"
raise TaskClientProtocolError(f"request failed: HTTP {status_code} ({code})")
except urllib.error.URLError as exc:
if index + 1 < attempts:
time.sleep(_RETRY_BACKOFF_SECONDS[index])
continue
raise TaskClientProtocolError("request transport error") from exc
raise TaskClientProtocolError("request failed")
def _post(self, path: str, payload: dict[str, Any], *, idempotency_key: str | None = None) -> dict[str, Any]:
status_code, response = self._request(
"POST", path, payload, idempotency_key=idempotency_key or _stable_key(path, payload)
)
if status_code not in {200, 201} or not isinstance(response, dict):
raise TaskClientProtocolError("unexpected API response")
return response
def _get(self, path: str) -> Any:
status_code, response = self._request("GET", path)
if status_code != 200:
raise TaskClientProtocolError("unexpected API response")
return response
def list_tasks(self) -> list[dict[str, Any]]:
response = self._get("/tasks")
if not isinstance(response, list) or any(not isinstance(item, dict) for item in response):
raise TaskClientProtocolError("invalid task list")
return response
def get_task(self, task_id: str) -> dict[str, Any]:
response = self._get(f"/tasks/{_safe_identifier(task_id, 'task_id')}")
if not isinstance(response, dict):
raise TaskClientProtocolError("invalid task response")
return response
def register_worker(
self, *, worker_id: str, worker_type: str, identity: str,
capabilities: Iterable[str], trust_zone: str, max_concurrency: int,
heartbeat_interval_seconds: int, supported_protocols: Iterable[str] = ("aios-task-v1",),
) -> dict[str, Any]:
payload = {
"worker_id": worker_id, "worker_type": worker_type, "identity": identity,
"capabilities": sorted(set(capabilities)), "trust_zone": trust_zone,
"max_concurrency": max_concurrency,
"heartbeat_interval_seconds": heartbeat_interval_seconds,
"supported_protocols": list(supported_protocols),
"owner_identity": identity, "registration_policy": "self",
}
return self._post("/workers/register", payload, idempotency_key=_stable_key("register", payload))
def worker_heartbeat(self, worker_id: str, *, heartbeat_interval_seconds: int | None = None) -> dict[str, Any]:
payload = {} if heartbeat_interval_seconds is None else {"heartbeat_interval_seconds": heartbeat_interval_seconds}
return self._post(f"/workers/{_safe_identifier(worker_id, 'worker_id')}/heartbeat", payload)
def claim(self, task_id: str, *, worker_id: str, lease_seconds: int | None = None) -> dict[str, Any]:
payload = {"worker_id": worker_id, "lease_seconds": lease_seconds or self.config.default_lease_seconds}
return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/claim", payload)
def claim_heartbeat(self, task_id: str, *, claim_id: str, lease_seconds: int | None = None) -> dict[str, Any]:
if not claim_id:
raise TaskClientProtocolError("claim_id is required")
payload = {"claim_id": claim_id, "lease_seconds": lease_seconds or self.config.default_lease_seconds}
return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/heartbeat", payload)
def progress(self, task_id: str, *, claim_id: str, progress: int | float, stage: str, message: str | None = None, lease_seconds: int | None = None) -> dict[str, Any]:
if not claim_id:
raise TaskClientProtocolError("claim_id is required for progress")
payload = {"claim_id": claim_id, "progress": progress, "stage": stage, "message": message, "lease_seconds": lease_seconds or self.config.default_lease_seconds}
return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/progress", payload)
def result(self, task_id: str, *, claim_id: str, status: str, outputs: list[dict[str, Any]], evidence: list[dict[str, Any]], started_at: str, completed_at: str, result_digest: str | None = None, operation_digest: str | None = None, trust_level: str | None = None, production_trust: bool | None = None, **_: Any) -> dict[str, Any]:
if not claim_id:
raise TaskClientProtocolError("claim_id is required for result submission")
if production_trust not in {None, False}:
raise TaskClientProtocolError("production_trust is forbidden")
payload = {"claim_id": claim_id, "status": status, "outputs": outputs, "evidence": evidence, "started_at": started_at, "completed_at": completed_at}
if result_digest:
payload["result_digest"] = result_digest
if operation_digest:
payload["operation_digest"] = operation_digest
if trust_level:
payload["trust_level"] = trust_level
payload["production_trust"] = False
return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/result", payload)
def evidence(self, task_id: str, *, claim_id: str, evidence: list[dict[str, Any]], status: str = "VERIFYING", started_at: str, completed_at: str) -> dict[str, Any]:
return self.result(task_id, claim_id=claim_id, status=status, outputs=[], evidence=evidence, started_at=started_at, completed_at=completed_at)
def fail(self, task_id: str, *, claim_id: str, reason: str = "failed", recoverable: bool = False) -> dict[str, Any]:
if not claim_id:
raise TaskClientProtocolError("claim_id is required for fail")
return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/fail", {"claim_id": claim_id, "reason": reason, "recoverable": recoverable})
@dataclass(frozen=True)
class WorkerConfig:
worker_id: str
worker_type: str
identity: str
trust_zone: str
max_concurrency: int = 1
heartbeat_interval_seconds: int = 15
claim_lease_seconds: int = 120
supported_protocols: tuple[str, ...] = ("aios-task-v1",)
capabilities: tuple[str, ...] = ("code.read", "test.python", "review.security")
def to_registration_payload(self) -> dict[str, Any]:
return {
"worker_id": self.worker_id, "worker_type": self.worker_type,
"identity": self.identity, "capabilities": list(self.capabilities),
"trust_zone": self.trust_zone, "max_concurrency": self.max_concurrency,
"heartbeat_interval_seconds": self.heartbeat_interval_seconds,
"supported_protocols": list(self.supported_protocols),
"owner_identity": self.identity, "registration_policy": "self",
}
@@ -1,162 +0,0 @@
"""Private, symlink-safe per-claim workspaces."""
from __future__ import annotations
import os
import shutil
import stat
from dataclasses import dataclass
from pathlib import Path
from .command_policy import isolated_environment
WORKSPACE_DIR_MODE = 0o700
RESULT_FILE_MODE = 0o600
class WorkspaceError(RuntimeError):
pass
class WorkspacePermissionError(WorkspaceError):
pass
class WorkspaceCleanupError(WorkspaceError):
pass
def _lexical(path: Path) -> Path:
return Path(os.path.abspath(os.fspath(path)))
def assert_private_path(path: Path, *, expected_mode: int = WORKSPACE_DIR_MODE) -> Path:
"""Reject every existing symlink before resolving the final path."""
lexical = _lexical(Path(path))
current = Path(lexical.anchor)
for component in lexical.parts[1:]:
current = current / component
try:
info = current.lstat()
except FileNotFoundError:
continue
if stat.S_ISLNK(info.st_mode):
raise WorkspacePermissionError("symlink path is forbidden")
return lexical
def _ensure_directory(path: Path, *, mode: int = WORKSPACE_DIR_MODE) -> None:
lexical = assert_private_path(path)
lexical.mkdir(mode=mode, parents=True, exist_ok=True)
if lexical.is_symlink() or not lexical.is_dir():
raise WorkspacePermissionError("workspace directory rejected")
os.chmod(lexical, mode)
if stat.S_IMODE(lexical.stat().st_mode) != mode:
raise WorkspacePermissionError("workspace directory mode rejected")
def _ensure_result_name(name: str) -> str:
safe = str(name).strip()
if not safe or safe in {".", ".."} or "/" in safe or "\\" in safe or "\0" in safe:
raise WorkspaceError("invalid result file name")
return safe
def _within(path: Path, base: Path) -> Path:
lexical = assert_private_path(path)
root = assert_private_path(base)
try:
lexical.relative_to(root)
except ValueError:
raise WorkspacePermissionError("path escapes workspace") from None
return lexical
def _copy_readonly_tree(source: Path, destination: Path) -> None:
source = assert_private_path(source)
if not source.is_dir():
raise WorkspaceError("source archive tree unavailable")
for item in source.rglob("*"):
if item.is_symlink():
raise WorkspacePermissionError("source archive symlink rejected")
shutil.copytree(source, destination, dirs_exist_ok=True, symlinks=False)
for item in destination.rglob("*"):
if item.is_file():
os.chmod(item, 0o400)
elif item.is_dir():
os.chmod(item, 0o500)
@dataclass(frozen=True)
class WorkspaceLayout:
workspace_root: Path
repo_root: Path
result_root: Path
home: Path
tmp: Path
@dataclass
class TaskWorkspace:
layout: WorkspaceLayout
task_id: str
claim_id: str
def __init__(self, root: Path, task_id: str, claim_id: str, *, repo_subdir: str = "repo") -> None:
workspace_root = assert_private_path(root)
_ensure_directory(workspace_root)
self.layout = WorkspaceLayout(
workspace_root, workspace_root / repo_subdir, workspace_root / "results",
workspace_root / "home", workspace_root / "tmp",
)
for path in (self.layout.repo_root, self.layout.result_root, self.layout.home, self.layout.tmp):
_ensure_directory(path)
self.task_id, self.claim_id = task_id, claim_id
@classmethod
def create(cls, root: Path, task_id: str, claim_id: str, *, repo_subdir: str = "repo") -> "TaskWorkspace":
return cls(root, task_id, claim_id, repo_subdir=repo_subdir)
def __enter__(self) -> "TaskWorkspace":
return self
def __exit__(self, *_: object) -> None:
self.cleanup()
def materialize_readonly_archive(self, source: Path) -> None:
if any(self.layout.repo_root.iterdir()):
raise WorkspaceError("workspace repository is not empty")
_copy_readonly_tree(source, self.layout.repo_root)
def environ(self) -> dict[str, str]:
return isolated_environment(self.layout.home, self.layout.tmp, "active-runtime")
def result_path(self, name: str) -> Path:
return _within(self.layout.result_root / _ensure_result_name(name), self.layout.result_root)
def write_result(self, *, name: str, payload: str) -> Path:
path = self.result_path(name)
flags = os.O_CREAT | os.O_WRONLY | os.O_TRUNC
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
fd = os.open(path, flags, RESULT_FILE_MODE)
try:
if not stat.S_ISREG(os.fstat(fd).st_mode):
raise WorkspacePermissionError("result path is not regular")
os.write(fd, payload.encode("utf-8"))
os.fsync(fd)
os.fchmod(fd, RESULT_FILE_MODE)
finally:
os.close(fd)
return path
def cleanup(self) -> None:
root = assert_private_path(self.layout.workspace_root)
if not root.exists():
return
try:
shutil.rmtree(root)
except OSError as exc:
raise WorkspaceCleanupError("workspace cleanup failed") from exc
Workspace = TaskWorkspace
@@ -1,40 +0,0 @@
[Unit]
Description=AIOS Codex Review/Test Worker
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=aiman
Group=gitops
WorkingDirectory=/usr/libexec/aios-codex-review-test-worker/current
Environment=AIOS_CODEX_WORKER_MODE=shadow
Environment=AIOS_TASK_PLANE_EXECUTION_ENABLED=false
Environment=PYTHONDONTWRITEBYTECODE=1
ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test
Restart=on-failure
RestartSec=2
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
LockPersonality=true
RestrictSUIDSGID=true
RestrictRealtime=true
MemoryDenyWriteExecute=true
RuntimeDirectory=aios-codex-review-test-worker
StateDirectory=aios-codex-review-test-worker
CacheDirectory=aios-codex-review-test-worker
ReadOnlyPaths=/data/aios-gitops/repo.git
ReadOnlyPaths=/usr/libexec/aios-codex-review-test-worker
ReadWritePaths=/run/aios-codex-review-test-worker
ReadWritePaths=/var/lib/aios-codex-review-test-worker
ReadWritePaths=/var/cache/aios-codex-review-test-worker
UMask=0077
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,26 @@
[Unit]
Description=AIOS dormant single-owner machine-status prerequisite
RefuseManualStop=no
[Service]
Type=oneshot
User=aiman
Group=gitops
Environment=AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED=false
Environment=AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE=false
Environment=AIOS_GITOPS_WRITES_ENABLED=false
ExecStart=/usr/bin/python3 -m aios_single_owner_governance.service --component machine-status-producer --self-check --checkpoint-dir /var/lib/aios-gitops/runtime/single-owner-governance
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
RestrictSUIDSGID=true
LockPersonality=true
MemoryDenyWriteExecute=true
RestrictAddressFamilies=AF_UNIX
ReadOnlyPaths=/data/aios-gitops/repo.git
ReadWritePaths=/var/lib/aios-gitops/runtime/single-owner-governance
@@ -0,0 +1,26 @@
[Unit]
Description=AIOS dormant single-owner review adapter prerequisite
RefuseManualStop=no
[Service]
Type=oneshot
User=aiman
Group=gitops
Environment=AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED=false
Environment=AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE=false
Environment=AIOS_GITOPS_WRITES_ENABLED=false
ExecStart=/usr/bin/python3 -m aios_single_owner_governance.service --component owner-review-adapter --self-check --checkpoint-dir /var/lib/aios-gitops/runtime/single-owner-governance
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
RestrictSUIDSGID=true
LockPersonality=true
MemoryDenyWriteExecute=true
RestrictAddressFamilies=AF_UNIX
ReadOnlyPaths=/data/aios-gitops/repo.git
ReadWritePaths=/var/lib/aios-gitops/runtime/single-owner-governance
@@ -0,0 +1,20 @@
[Unit]
Description=AIOS single-owner governance activation verifier
ConditionPathExists=/etc/aios-gitops/single-owner-activation-bindings.json
[Service]
Type=oneshot
User=root
Group=gitops
Environment=AIOS_GITOPS_WRITES_ENABLED=false
ExecStart=/usr/bin/aios-single-owner-governance --verify --bindings /etc/aios-gitops/single-owner-activation-bindings.json --state-root /var/lib/aios-gitops/governance-activation --lock-path /run/aios-gitops/governance-activation.lock
NoNewPrivileges=true
PrivateTmp=true
ProtectHome=true
ProtectSystem=strict
ReadWritePaths=/var/lib/aios-gitops/governance-activation /run/aios-gitops
RestrictSUIDSGID=true
UMask=0027
# Deliberately not installable. Deployment remains disabled/inactive;
# activation is an explicit approved CLI transaction, never an always-on unit.
@@ -0,0 +1,4 @@
d /var/lib/aios-gitops/governance-activation 0750 root gitops -
d /var/lib/aios-gitops/governance-activation/wal 0750 root gitops -
d /var/lib/aios-gitops/governance-activation/receipts 0750 root gitops -
d /run/aios-gitops 0750 root gitops -
@@ -0,0 +1,68 @@
# Isolated Writer Exact-Path Policy V1
This policy is an additional, service-owned authorization layer for
`POST /api/commit-file`. It never expands a Token Policy V2 grant.
An atomic write is permitted only when all of the following agree:
1. socket peer, audience, operation, branch, base and exact path in the
Token Policy V2 grant;
2. service audience, repository identity, branch, operation and exact path
in `AIOS_GITOPS_PATH_POLICY_FILE`;
3. lexical path, resolved containment and symlink checks;
4. the configured content-size and content-profile checks;
5. operation digest, expected branch HEAD and durable operation ledger.
Any failed layer returns a fail-closed policy error. Coarse `/api/file` and
`/api/commit` routes remain forbidden.
## File contract
The configured path must be absolute. The parent directory must be
`root:gitops 0750`; the regular, non-symlink file must be
`root:gitops 0640`. JSON must use UTF-8 canonical encoding with sorted keys,
compact separators and one trailing newline. Unknown fields, unsupported
operations or profiles, duplicate paths, oversized policies and
group/world-writable files are rejected during Writer startup.
Example:
```json
{"allowed_branch":"patch/codex-review-test-worker-20260724","allowed_operations":{"commit_file.atomic":{"content_profile":"systemd-codex-review-worker-v1","exact_paths":["deploy/systemd/aios-codex-review-test-worker.service"],"max_content_bytes":32768}},"audience":"aios-codex-review-worker-writer.service","policy_id":"codex-review-worker-path-policy-v1","repository_identity":"aios/aios","schema_version":1}
```
The SHA-256 of the canonical policy is exposed in the isolated runtime
status and written to the operation audit record. The request body cannot
select or override the policy file.
## Content profile
`systemd-codex-review-worker-v1` is an exact, fail-closed unit grammar. The
only sections are `[Unit]`, `[Service]` and `[Install]`. Each section has a
directive allowlist; every single-value directive and every repeated
`Environment`, `ReadOnlyPaths` and `ReadWritePaths` entry must match the
governed Shadow unit contract exactly.
The profile fixes the Python module
`task_plane.workers.codex_review_test`, `User=aiman`, `Group=gitops`,
the controlled `/usr/libexec/aios-codex-review-test-worker/current`
working directory, all hardening switches and the three permitted
environment variables. It grants read access only to the authoritative
bare repository and the versioned worker artifact, and write access only
to the worker runtime, state and cache directories.
Unknown sections or directives are rejected, including lifecycle hooks,
credential injection, environment imports, root images, bind mounts,
capability expansion, devices, sockets and top-level writable paths. The
parser also rejects duplicate sections, duplicate single-value directives,
empty resets, continuation lines, CRLF, NUL, non-UTF-8 surrogate content,
overlong lines and Unicode lookalike directive names. It does not use
`ConfigParser(strict=False)` or silently merge repeated values.
Production credentials, if later authorized, must be supplied by a
separately governed deployment drop-in. They are not permitted in the
source unit or this content profile.
Policy changes require a separate maintenance branch and Owner-reviewed
ordinary merge commit. A task-scoped Writer must not edit its own policy
file.
+44
View File
@@ -51,6 +51,31 @@ components:
path: {type: string, example: docs/gitops/chatgpt-change.md}
content: {type: string}
message: {type: string}
AtomicFileCommitRequest:
type: object
additionalProperties: false
required: [branch, path, content, message, expected_head_oid, idempotency_key]
properties:
branch:
type: string
pattern: '^patch/[A-Za-z0-9._/-]+$'
path:
type: string
minLength: 1
maxLength: 512
content:
type: string
maxLength: 1048576
message:
type: string
minLength: 1
maxLength: 200
expected_head_oid:
type: string
pattern: '^[0-9a-f]{40}$'
idempotency_key:
type: string
pattern: '^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$'
CommitRequest:
type: object
properties:
@@ -428,6 +453,25 @@ paths:
schema: {$ref: '#/components/schemas/FileCommitRequest'}
responses: {'200': {description: File committed}, '400': {description: Secret rejected or invalid path}, '401': {description: Unauthorized}}
x-openai-isConsequential: true
/api/commit-file:
post:
operationId: commitFileAtomic
summary: Atomically write one allowlisted file, commit, push, and verify using Token Policy V2
security: [{bearerAuth: []}]
requestBody:
required: true
content:
application/json:
schema: {$ref: '#/components/schemas/AtomicFileCommitRequest'}
responses:
'200': {description: Completed or idempotently replayed}
'400': {description: Invalid canonical request}
'401': {description: Missing, invalid, expired, or revoked Token Policy V2 grant}
'403': {description: Audience, source, branch, operation, path, or policy denied}
'409': {description: Idempotency conflict or manual review required}
'413': {description: Request too large}
'503': {description: Push or remote verification unavailable; recoverable}
x-openai-isConsequential: true
/api/commit:
post:
operationId: commitChanges
@@ -1,6 +0,0 @@
{
"schema_version": 1,
"task": "codex-review-test-worker",
"writer_role": "isolated-writer",
"status": "READY"
}
@@ -1,36 +0,0 @@
# Codex Read/Review/Test Worker
This package defines a Shadow-only Task Plane worker for allowlisted code reading,
review, Python tests, compile checks, and Task Plane acceptance. It is not a
production execution authorization.
Safety invariants:
- `max_concurrency=1`, heartbeat 15 seconds, claim lease 120 seconds.
- Task Plane endpoint must be explicit loopback HTTP.
- credentials are read only from a regular, non-symlink `0600` file.
- only low-risk `code.review` and `test.execute` tasks with
`production_side_effects=false` and public/internal data are eligible.
- credential-secret, write, deployment, production, arbitrary shell, network,
path traversal, symlink escape, revision expressions, and secret environment
variables are rejected.
- `ResultEnvelope` is canonical-digest-bound and always has
`trust_level=SHADOW_ATTESTED`, `production_trust=false`.
The committed systemd unit is deliberately non-installable and runs only
`--self-check`. Enabling a real Worker or changing any production execution flag
requires a separate Owner approval and a separately bound artifact.
Artifact build:
```bash
python3 scripts/build_codex_review_worker_artifact.py \
--git-dir /data/aios-gitops/repo.git \
--commit <exact-branch-commit> \
--output <new-empty-output-directory>
```
The temporary canary uses `127.0.0.1:18923`, an isolated state root, an ephemeral
credential file, and a no-`.git` source tree. Remove the unit, token, and state
after acceptance. Do not deploy this package to the existing Task Plane Shadow
runtime as part of build/canary acceptance.
@@ -0,0 +1,50 @@
{
"artifact_sha256": "01e0226f0302bb8e2f8d81299393075bf0332c6e22e00877469fb297c51f3802",
"branch": "patch/single-owner-governance-live-activation-20260729",
"controller_visible_governance_oid": "8b7f9d9028f6ff9beb225bf85f53dbeaac6830e7",
"created_at": "2026-07-29T03:28:15Z",
"implementation_commit_oid": "e1e438564ce8b987a36df0947e6eaa324dfede74",
"implementation_tree_oid": "dd93dc1df10cb8b8fe59b09d3214a9ae6fee671e",
"independent_validation": {
"activation_py_sha256": "86cde2079b4f3493957ba346e71b02b387693e63a3f38be418710ff7c348d19a",
"approval_backend_dual_database": "DENIED",
"expiry_replay": "DENIED",
"operation_replay": "DENIED",
"result": "PASS",
"target_replay": "DENIED",
"task_replay": "DENIED"
},
"mirror_closure": {
"pr30_merge_oid": "f6bef90633f428454e85721f353cd82cdf04caf0",
"pr31_merge_oid": "8b7f9d9028f6ff9beb225bf85f53dbeaac6830e7",
"receipt_sha256": "b3edc4eb0701efcb448b981029ab9a67a0637f8349bf4a047f430f647e4a8fd0",
"result": "PASS"
},
"production_isolation": {
"aios_gitops_writes_enabled": false,
"live_activation_service_active": false,
"live_activation_service_enabled": false,
"main_modified": false,
"production_modified": false,
"protected_refs_unchanged": true
},
"rollback_artifact_sha256": "ed9307cc2f07767b75866fcb370fa3f98b76c5147e4b7fc49ecf04cf21cf7c31",
"signed_merge_audit": {
"merge_signature_policy": "NOT_REQUIRED",
"pr31_signature_policy_result": "UNSIGNED_BUT_POLICY_VALID",
"signed_merge_requirement_effective": false
},
"task_id": "AIOS-SINGLE-OWNER-GOVERNANCE-ACTIVATION-20260729",
"test_evidence_sha256": "172a7cf812c6b901524138fb72851a9b7d2e895dedc98893407d0bab9b52c6c1",
"tests": {
"collection_errors": 0,
"compileall": "PASS",
"copy_of_live_gitea_acceptance": "PASS",
"secret_scan": "PASS",
"subtests_passed": 17,
"tests_failed": 0,
"tests_passed": 75,
"transaction_tests": "PASS",
"rollback_tests": "PASS"
}
}
@@ -1,20 +0,0 @@
{
"schema_version": 1,
"component": "aios-codex-review-test-worker",
"mode": "SHADOW_ONLY",
"production_trust": false,
"production_execution_authorized": false,
"required_acceptance": [
"exact_commit_artifact",
"no_git_canary",
"worker_registration",
"claim_lease_heartbeat",
"progress",
"result_envelope",
"lease_loss_stops_result",
"unsafe_tasks_rejected",
"secret_scan",
"production_side_effects_false"
],
"runtime_result": "GENERATED_OUTSIDE_GIT_DURING_CANARY"
}
@@ -1,138 +0,0 @@
#!/usr/bin/env python3
"""Build a no-.git, exact-commit Codex review worker artifact."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import shutil
import subprocess
import tarfile
from pathlib import Path
OID_RE = re.compile(r"^[0-9a-f]{40}$")
SECRET_PATTERNS = (
re.compile(rb"Authorization:\s*Bearer\s+[A-Za-z0-9._~+/=-]{16,}", re.I),
re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
re.compile(rb"(?:ghp|glpat)_[A-Za-z0-9_-]{20,}", re.I),
)
PACKAGE_PATHS = (
"active-runtime/app/task_plane",
"runtime/agent-os",
"tests/task_plane",
"tests/test_isolated_writer_policy.py",
"scripts/task-plane",
"scripts/run_codex_review_worker_acceptance.py",
"scripts/build_codex_review_worker_artifact.py",
"deploy/systemd/aios-codex-review-test-worker.service",
"docs/task-plane",
"reports/codex-review-worker-acceptance.json",
)
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def atomic_json(path: Path, payload: object) -> None:
temporary = path.with_suffix(path.suffix + ".tmp")
with temporary.open("w", encoding="utf-8", newline="\n") as stream:
json.dump(payload, stream, sort_keys=True, separators=(",", ":"))
stream.write("\n")
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, path)
def git(git_dir: Path, *args: str) -> str:
return subprocess.run(
["git", f"--git-dir={git_dir}", *args], check=True,
capture_output=True, text=True, timeout=60,
env={"PATH": os.environ.get("PATH", "")},
).stdout.strip()
def build(git_dir: Path, commit: str, output: Path) -> dict[str, str]:
commit = git(git_dir, "rev-parse", f"{commit}^{{commit}}")
if not OID_RE.fullmatch(commit):
raise ValueError("exact commit OID required")
tree_oid = git(git_dir, "rev-parse", f"{commit}^{{tree}}")
for required in PACKAGE_PATHS:
probe = subprocess.run(
["git", f"--git-dir={git_dir}", "cat-file", "-e", f"{commit}:{required}"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
)
if probe.returncode != 0:
raise RuntimeError(f"required artifact path missing: {required}")
if output.exists():
raise FileExistsError("output already exists")
output.mkdir(parents=True, mode=0o700)
source_tar = output / "codex-review-worker-source.tar"
subprocess.run(
[
"git", f"--git-dir={git_dir}", "archive", "--format=tar",
f"--output={source_tar}", commit, "--", *PACKAGE_PATHS,
],
check=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
)
source = output / "source"
source.mkdir(mode=0o700)
with tarfile.open(source_tar, "r") as archive:
archive.extractall(source, filter="data")
if any(item.name == ".git" for item in source.rglob("*")):
raise RuntimeError("artifact contains Git metadata")
findings = []
for item in sorted(path for path in source.rglob("*") if path.is_file() and path.stat().st_size < 5_000_000):
data = item.read_bytes()
if any(pattern.search(data) for pattern in SECRET_PATTERNS):
findings.append(item.relative_to(source).as_posix())
if findings:
raise RuntimeError("secret scan failed")
tracked = git(
git_dir, "ls-tree", "-r", "--name-only", commit, "--", *PACKAGE_PATHS
).splitlines()
files = [
{"path": name, "sha256": sha256(source / name), "size": (source / name).stat().st_size}
for name in tracked if (source / name).is_file()
]
manifest = {"schema_version": 1, "commit": commit, "tree_oid": tree_oid, "files": files}
atomic_json(output / "manifest.json", manifest)
shutil.copyfile(source_tar, output / "rollback-source.tar")
unit = source / "deploy/systemd/aios-codex-review-test-worker.service"
closure = {
"schema_version": 1,
"commit": commit,
"tree_oid": tree_oid,
"artifact_sha256": sha256(source_tar),
"manifest_sha256": sha256(output / "manifest.json"),
"rollback_artifact_sha256": sha256(output / "rollback-source.tar"),
"systemd_unit_sha256": sha256(unit),
"no_git": True,
"secret_scan": "PASS",
}
atomic_json(output / "package-closure.json", closure)
closure["package_closure_sha256"] = sha256(output / "package-closure.json")
with (output / "checksums.sha256").open("w", encoding="utf-8", newline="\n") as stream:
for item in sorted(path for path in output.iterdir() if path.is_file() and path.name != "checksums.sha256"):
stream.write(f"{sha256(item)} {item.name}\n")
return closure
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--git-dir", required=True, type=Path)
parser.add_argument("--commit", required=True)
parser.add_argument("--output", required=True, type=Path)
args = parser.parse_args()
print(json.dumps(build(args.git_dir.resolve(), args.commit, args.output.resolve()), sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -1,159 +0,0 @@
#!/usr/bin/env python3
"""Run or serve an isolated Codex Worker Shadow canary."""
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import os
import stat
import sys
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime"))
from app.task_plane.api import TaskPlaneService, run_server
from app.task_plane.policy import IdentityVerifier, PolicyEngine, ReadonlyWorkerProfile
from app.task_plane.store import TaskPlaneStore
from app.task_plane.workers.codex_review_test import (
CodexReviewTaskPolicy,
CodexReviewWorker,
READONLY_CAPABILITIES,
)
from app.task_plane.workers.task_client import CredentialSource, StageAClientConfig, TaskPlaneClient
IDENTITY = "spiffe://aios/agent/codex/review-test-canary"
WORKER_ID = "codex-review-test-canary"
def _token(path: Path) -> str:
info = path.lstat()
if path.is_symlink() or not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o600:
raise SystemExit("canary credential reference rejected")
value = path.read_text(encoding="utf-8").strip()
if len(value.encode("utf-8")) < 32:
raise SystemExit("canary credential unavailable")
return value
def serve(host: str, port: int, state_root: Path, token_file: Path) -> None:
if host not in {"127.0.0.1", "::1"}:
raise SystemExit("canary must bind loopback")
token = _token(token_file)
profile = ReadonlyWorkerProfile(
worker_id=WORKER_ID, identity=IDENTITY,
capabilities=frozenset(READONLY_CAPABILITIES),
)
policy = PolicyEngine(
identity_verifier=IdentityVerifier({IDENTITY}),
execution_enabled=False,
readonly_worker_execution_enabled=True,
readonly_worker_profiles=[profile],
identity_capabilities={
IDENTITY: {"task.creator", "task.read", "task.audit", "task.claim", "worker.self_register"}
},
)
state_root.mkdir(mode=0o700, parents=True, exist_ok=True)
store = TaskPlaneStore(str(state_root / "task-plane.sqlite3"), high_assurance=True)
service = TaskPlaneService(store, policy=policy)
run_server(host, port, service, authenticator=lambda candidate: IDENTITY if hmac.compare_digest(candidate, token) else None)
def _post(base_url: str, token: str, path: str, payload: dict, key: str | None = None) -> dict:
body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
if key:
headers["Idempotency-Key"] = key
request = urllib.request.Request(base_url + path, data=body, headers=headers, method="POST")
with urllib.request.urlopen(request, timeout=15) as response:
return json.loads(response.read().decode("utf-8"))
def run(base_url: str, token_file: Path, source_commit: str, report: Path) -> dict:
token = _token(token_file)
task_id = "CODEX-CANARY-" + source_commit[:12]
task = {
"task_id": task_id, "task_type": "code.review", "risk": "low",
"required_capabilities": ["code.read"],
"metadata": {"production_side_effects": False},
"data_classification": "internal", "network_policy": "deny",
"allowed_input_paths": ["active-runtime/app/task_plane"],
"allowed_output_paths": ["reports/canary.json"],
}
_post(base_url, token, "/tasks", task, "codex-canary-create-" + source_commit)
client = TaskPlaneClient(StageAClientConfig(base_url), token)
progress = []
def executor(_task, emit):
emit(25, "CLAIMED")
progress.append(25)
emit(100, "VERIFIED")
progress.append(100)
return "shadow canary completed", [{"kind": "summary", "status": "PASS"}], [{"kind": "test", "sha256": hashlib.sha256(source_commit.encode()).hexdigest()}]
worker = CodexReviewWorker(
client, CodexReviewTaskPolicy(WORKER_ID, IDENTITY), source_commit
)
completed_task = worker.run_once(executor)
task_after = client.get_task(task_id)
unsafe_rejected = 0
for unsafe in (
{"risk": "high", "task_type": "code.review", "metadata": {"production_side_effects": False}},
{"risk": "low", "task_type": "code.review", "metadata": {"production_side_effects": True}},
{"risk": "low", "task_type": "code.review", "metadata": {"production_side_effects": False}, "data_classification": "credential-secret"},
{"risk": "low", "task_type": "code.review", "metadata": {"production_side_effects": False}, "required_capabilities": ["code.write"]},
):
try:
worker.policy.assert_shadow_only_task(unsafe)
except Exception:
unsafe_rejected += 1
checks = {
"worker_registration": "PASS",
"task_claim": "PASS" if completed_task == task_id else "FAIL",
"heartbeat": "PASS",
"progress": "PASS" if progress == [25, 100] else "FAIL",
"result_envelope": "PASS" if task_after.get("status") == "COMPLETED" else "FAIL",
"unsafe_task_rejections": "PASS" if unsafe_rejected == 4 else "FAIL",
"network_endpoint_loopback": "PASS",
"production_side_effects": False,
"git_write_count": 0,
"tag_write_count": 0,
"baseline_write_count": 0,
"registry_write_count": 0,
"deployment_count": 0,
}
result = {
"schema_version": 1, "source_commit": source_commit,
"task_id": task_id, "worker_id": WORKER_ID, "checks": checks,
"passed": not any(value == "FAIL" for value in checks.values()),
}
encoded = (json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")
report.write_bytes(encoded)
result["report_sha256"] = hashlib.sha256(encoded).hexdigest()
return result
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--serve-canary", action="store_true")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=18923)
parser.add_argument("--state-root", type=Path)
parser.add_argument("--token-file", required=True, type=Path)
parser.add_argument("--source-commit")
parser.add_argument("--report", type=Path)
args = parser.parse_args()
if args.serve_canary:
serve(args.host, args.port, args.state_root, args.token_file)
return 0
if not args.source_commit or args.report is None:
raise SystemExit("--source-commit and --report are required")
print(json.dumps(run(f"http://{args.host}:{args.port}", args.token_file, args.source_commit, args.report), sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,24 @@
# AIOS Single-Owner Governance Prerequisites
This package is a dormant candidate produced under the PR #29 prerequisite
authorization. It validates exact-head Owner approvals and machine gates,
models protection-policy compare-and-swap with compensation, builds evidence,
and exercises every mutation path only against disposable fixtures.
The prerequisite shadow services remain static and refuse live mutation. This
branch adds a separate, one-shot activation executor with `--plan`,
`--activate`, `--verify`, `--rollback`, and `--status` commands. It is not an
always-on writer and never enables global GitOps writes.
The executor is fail-closed: every consequential command is bound to an exact
approval merge, governance OID, evidence closure, Owner review, component
digests, full branch-protection snapshot, operation ID, and plan ID. It writes
a fsynced WAL before the single-row SQLite CAS, verifies the full applied row,
and only rolls back with a target-to-previous CAS. Audit records are
hash-chained and credentials are redacted.
Installation must remain disabled and inactive until a merged
`SINGLE_OWNER_GOVERNANCE_LIVE_ACTIVATION_IMPLEMENTATION_AND_EXECUTION`
approval authorizes one explicit execution. The approved runtime paths are
`/var/lib/aios-gitops/governance-activation/` and
`/run/aios-gitops/governance-activation.lock`.
@@ -0,0 +1,80 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"additionalProperties": false,
"properties": {
"activation_approval_merge_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
"activation_approval_pr_number": {"minimum": 1, "type": "integer"},
"activation_manifest_blob_oid": {"pattern": "^[0-9a-f]{40,64}$", "type": "string"},
"activation_manifest_path": {"pattern": "^approvals/.+\\.json$", "type": "string"},
"activation_manifest_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
"approval_db_path": {"minLength": 1, "type": "string"},
"approval_request_head_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
"backend": {"enum": ["file", "sqlite"]},
"backend_path": {"minLength": 1, "type": "string"},
"branch": {"const": "governance/approvals"},
"evidence_closure_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
"evidence_git_dir": {"minLength": 1, "type": "string"},
"executor": {"not": {"const": "owner"}, "type": "string"},
"expected_branch_protection_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
"expected_gitea_schema_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
"expected_governance_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
"expected_state_machine_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
"expected_validator_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
"expires_at": {"format": "date-time", "type": "string"},
"forbidden_operations": {"additionalProperties": {"const": false}, "type": "object"},
"governance_git_dir": {"minLength": 1, "type": "string"},
"governance_ref": {"const": "refs/heads/governance/approvals"},
"machine_checks": {
"additionalProperties": false,
"properties": {
"aios/branch-protection": {"$ref": "#/$defs/machineCheck"},
"aios/evidence-closure": {"$ref": "#/$defs/machineCheck"},
"aios/secret-scan": {"$ref": "#/$defs/machineCheck"},
"aios/state-machine": {"$ref": "#/$defs/machineCheck"},
"aios/validator": {"$ref": "#/$defs/machineCheck"}
},
"required": ["aios/branch-protection", "aios/evidence-closure", "aios/secret-scan", "aios/state-machine", "aios/validator"],
"type": "object"
},
"observed_evidence_closure_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
"observed_governance_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
"observed_owner_review_oid": {"type": ["integer", "string"]},
"observed_state_machine_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
"observed_validator_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
"operation_id": {"minLength": 1, "type": "string"},
"owner_review_oid": {"type": ["integer", "string"]},
"repo_id": {"minimum": 1, "type": "integer"},
"requester": {"not": {"const": "owner"}, "type": "string"},
"state_machine_path": {"minLength": 1, "type": "string"},
"target_branch_protection": {"type": "object"},
"task_id": {"minLength": 1, "type": "string"},
"validator_path": {"minLength": 1, "type": "string"},
"wip": {"const": false}
},
"$defs": {
"machineCheck": {
"additionalProperties": false,
"properties": {
"head_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
"state": {"const": "success"}
},
"required": ["state", "head_oid"],
"type": "object"
}
},
"required": [
"task_id", "operation_id", "expected_governance_oid", "observed_governance_oid",
"expected_validator_sha256", "observed_validator_sha256", "validator_path",
"expected_state_machine_sha256", "observed_state_machine_sha256", "state_machine_path",
"expected_branch_protection_sha256", "expected_gitea_schema_sha256",
"activation_approval_merge_oid", "activation_approval_pr_number", "approval_request_head_oid",
"activation_manifest_path", "activation_manifest_blob_oid", "activation_manifest_sha256",
"evidence_closure_oid", "observed_evidence_closure_oid", "evidence_git_dir",
"owner_review_oid", "observed_owner_review_oid", "approval_db_path", "expires_at",
"governance_git_dir", "governance_ref", "machine_checks", "forbidden_operations",
"target_branch_protection", "backend", "backend_path", "repo_id", "branch",
"requester", "executor", "wip"
],
"title": "AIOS Single Owner Live Activation Bindings",
"type": "object"
}
@@ -0,0 +1,20 @@
"""Dormant single-owner governance activation prerequisites."""
from .models import (
ApprovalContext,
GovernanceState,
MachineGate,
ProtectionSnapshot,
Review,
)
from .validator import SingleOwnerApprovalValidator, ValidationDenied
__all__ = [
"ApprovalContext",
"GovernanceState",
"MachineGate",
"ProtectionSnapshot",
"Review",
"SingleOwnerApprovalValidator",
"ValidationDenied",
]
@@ -0,0 +1,686 @@
from __future__ import annotations
import argparse
from contextlib import contextmanager
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import sqlite3
import subprocess
from typing import Any, Iterator, Mapping, Protocol
try:
import fcntl
except ImportError: # pragma: no cover
fcntl = None
from .gitea_protected_branch import (
CASConflict,
FileProtectedBranchAdapter,
SQLiteProtectedBranchAdapter,
calculate_snapshot_sha256,
)
from .journal import JournalStore, atomic_write_json, read_json, redact, sha256_json
DEFAULT_STATE_ROOT = Path("/var/lib/aios-gitops/governance-activation")
DEFAULT_LOCK_PATH = Path("/run/aios-gitops/governance-activation.lock")
LIVE_GITEA_DB = Path("/data/aios-gitops/runtime/gitea-bin/data/gitea.db")
LIVE_AUTHORITATIVE_GIT = Path("/data/aios-gitops/repo.git")
HEX40 = re.compile(r"^[0-9a-f]{40}$")
HEX64 = re.compile(r"^[0-9a-f]{64}$")
REQUIRED_BINDINGS = {
"task_id",
"operation_id",
"expected_governance_oid",
"expected_validator_sha256",
"expected_state_machine_sha256",
"expected_branch_protection_sha256",
"activation_approval_merge_oid",
"evidence_closure_oid",
"owner_review_oid",
"observed_governance_oid",
"observed_validator_sha256",
"observed_state_machine_sha256",
"observed_evidence_closure_oid",
"observed_owner_review_oid",
"approval_request_head_oid",
"activation_approval_pr_number",
"expires_at",
"validator_path",
"state_machine_path",
"governance_git_dir",
"governance_ref",
"evidence_git_dir",
"approval_db_path",
"expected_gitea_schema_sha256",
"backend",
"backend_path",
"repo_id",
"branch",
"activation_manifest_path",
"activation_manifest_blob_oid",
"activation_manifest_sha256",
}
REQUIRED_MACHINE_CONTEXTS = {
"aios/validator",
"aios/state-machine",
"aios/branch-protection",
"aios/secret-scan",
"aios/evidence-closure",
}
class InjectedCrash(BaseException):
pass
class Adapter(Protocol):
def read_current(self) -> dict[str, Any]: ...
def calculate_snapshot_sha256(self, snapshot: Mapping[str, Any]) -> str: ...
def apply_atomic(self, expected_sha256: str, target: Mapping[str, Any]) -> dict[str, Any]: ...
def verify_applied(self, expected_sha256: str) -> bool: ...
def rollback_atomic(self, expected_target_sha256: str, previous: Mapping[str, Any]) -> dict[str, Any]: ...
def verify_rollback(self, expected_sha256: str) -> bool: ...
@contextmanager
def exclusive_lock(path: Path) -> Iterator[None]:
path.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
flags = os.O_RDWR | os.O_CREAT
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
fd = os.open(path, flags, 0o640)
try:
os.fchmod(fd, 0o640)
if fcntl is not None:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
yield
finally:
if fcntl is not None:
fcntl.flock(fd, fcntl.LOCK_UN)
os.close(fd)
def _require_hex(name: str, value: object, lengths: tuple[int, ...]) -> None:
if not isinstance(value, str) or len(value) not in lengths or not re.fullmatch(r"[0-9a-f]+", value):
raise RuntimeError(f"{name.upper()}_INVALID")
def _sha256_file(path: Path) -> str:
if not path.is_absolute() or not path.is_file() or path.is_symlink():
raise RuntimeError("VERIFIED_COMPONENT_PATH_INVALID")
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def _git(git_dir: Path, *args: str) -> str:
if not git_dir.is_absolute() or not git_dir.exists() or git_dir.is_symlink():
raise RuntimeError("GIT_OBJECT_STORE_INVALID")
result = subprocess.run(
["git", f"--git-dir={git_dir}", *args],
text=True,
capture_output=True,
check=False,
)
if result.returncode != 0:
raise RuntimeError("GIT_EVIDENCE_VERIFICATION_FAILED")
return result.stdout.strip()
def _parse_utc(value: object) -> datetime:
if not isinstance(value, str):
raise RuntimeError("EXPIRY_INVALID")
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as exc:
raise RuntimeError("EXPIRY_INVALID") from exc
if parsed.tzinfo is None:
raise RuntimeError("EXPIRY_INVALID")
return parsed.astimezone(timezone.utc)
def _approved_execution_binding(bindings: Mapping[str, Any]) -> dict[str, Any]:
return {
"task_id": bindings["task_id"],
"operation_id": bindings["operation_id"],
"expected_validator_sha256": bindings["expected_validator_sha256"],
"expected_state_machine_sha256": bindings["expected_state_machine_sha256"],
"expected_branch_protection_sha256": bindings["expected_branch_protection_sha256"],
"expected_gitea_schema_sha256": bindings["expected_gitea_schema_sha256"],
"target_branch_protection_sha256": calculate_snapshot_sha256(
bindings["target_branch_protection"]
),
"evidence_closure_oid": bindings["evidence_closure_oid"],
"backend": bindings["backend"],
"repo_id": bindings["repo_id"],
"branch": bindings["branch"],
"requester": bindings["requester"],
"executor": bindings["executor"],
"forbidden_operations": bindings["forbidden_operations"],
}
def _same_path(left: Path, right: Path) -> bool:
try:
return left.resolve(strict=True) == right.resolve(strict=True) and os.path.samefile(left, right)
except (FileNotFoundError, OSError):
return False
def _verify_authority_paths(bindings: Mapping[str, Any]) -> None:
if bindings["backend"] != "sqlite":
return
approval_db = Path(str(bindings["approval_db_path"]))
backend_db = Path(str(bindings["backend_path"]))
if not _same_path(approval_db, backend_db):
raise RuntimeError("APPROVAL_DB_BACKEND_DB_IDENTITY_MISMATCH")
live = _same_path(backend_db, LIVE_GITEA_DB)
if live:
if not _same_path(Path(str(bindings["governance_git_dir"])), LIVE_AUTHORITATIVE_GIT):
raise RuntimeError("AUTHORITATIVE_GOVERNANCE_GIT_REQUIRED")
if not _same_path(Path(str(bindings["evidence_git_dir"])), LIVE_AUTHORITATIVE_GIT):
raise RuntimeError("AUTHORITATIVE_EVIDENCE_GIT_REQUIRED")
return
acceptance_root = os.environ.get("AIOS_GOVERNANCE_ACTIVATION_ACCEPTANCE_ROOT")
if not acceptance_root:
raise RuntimeError("NON_LIVE_SQLITE_BACKEND_DENIED")
root = Path(acceptance_root).resolve(strict=True)
if not str(root).startswith("/run/aios-live-activation-acceptance-"):
raise RuntimeError("ACCEPTANCE_ROOT_INVALID")
for path in (approval_db, backend_db, Path(str(bindings["governance_git_dir"])), Path(str(bindings["evidence_git_dir"]))):
try:
path.resolve(strict=True).relative_to(root)
except (FileNotFoundError, ValueError) as exc:
raise RuntimeError("ACCEPTANCE_AUTHORITY_ESCAPE_DENIED") from exc
def _load_approved_manifest(bindings: Mapping[str, Any], git_dir: Path) -> dict[str, Any]:
path = str(bindings["activation_manifest_path"])
if not path.startswith("approvals/") or not path.endswith(".json") or ".." in Path(path).parts:
raise RuntimeError("ACTIVATION_MANIFEST_PATH_INVALID")
merge_oid = str(bindings["activation_approval_merge_oid"])
blob_oid = _git(git_dir, "rev-parse", f"{merge_oid}:{path}")
if blob_oid != bindings["activation_manifest_blob_oid"]:
raise RuntimeError("ACTIVATION_MANIFEST_BLOB_MISMATCH")
raw_result = subprocess.run(
["git", f"--git-dir={git_dir}", "show", f"{merge_oid}:{path}"],
capture_output=True,
check=False,
)
if raw_result.returncode != 0:
raise RuntimeError("ACTIVATION_MANIFEST_READ_FAILED")
if hashlib.sha256(raw_result.stdout).hexdigest() != bindings["activation_manifest_sha256"]:
raise RuntimeError("ACTIVATION_MANIFEST_SHA256_MISMATCH")
try:
manifest = json.loads(raw_result.stdout)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise RuntimeError("ACTIVATION_MANIFEST_INVALID") from exc
if not isinstance(manifest, dict):
raise RuntimeError("ACTIVATION_MANIFEST_INVALID")
return manifest
def _verify_external_evidence(bindings: Mapping[str, Any]) -> None:
_verify_authority_paths(bindings)
validator_sha = _sha256_file(Path(str(bindings["validator_path"])))
state_machine_sha = _sha256_file(Path(str(bindings["state_machine_path"])))
if validator_sha != bindings["expected_validator_sha256"] or validator_sha != bindings["observed_validator_sha256"]:
raise RuntimeError("VALIDATOR_DIGEST_MISMATCH")
if state_machine_sha != bindings["expected_state_machine_sha256"] or state_machine_sha != bindings["observed_state_machine_sha256"]:
raise RuntimeError("STATE_MACHINE_DIGEST_MISMATCH")
governance_git_dir = Path(str(bindings["governance_git_dir"]))
governance_oid = _git(governance_git_dir, "rev-parse", "--verify", str(bindings["governance_ref"]))
if governance_oid != bindings["expected_governance_oid"] or governance_oid != bindings["observed_governance_oid"]:
raise RuntimeError("WRONG_GOVERNANCE_OID_DENIED")
merge_oid = str(bindings["activation_approval_merge_oid"])
_git(governance_git_dir, "cat-file", "-e", f"{merge_oid}^{{commit}}")
if subprocess.run(
["git", f"--git-dir={governance_git_dir}", "merge-base", "--is-ancestor", merge_oid, governance_oid],
capture_output=True,
check=False,
).returncode != 0:
raise RuntimeError("ACTIVATION_APPROVAL_NOT_IN_GOVERNANCE_REF")
manifest = _load_approved_manifest(bindings, governance_git_dir)
if manifest.get("approval_type") != "SINGLE_OWNER_GOVERNANCE_LIVE_ACTIVATION_IMPLEMENTATION_AND_EXECUTION":
raise RuntimeError("ACTIVATION_APPROVAL_TYPE_MISMATCH")
if manifest.get("execution_binding") != _approved_execution_binding(bindings):
raise RuntimeError("APPROVED_EXECUTION_BINDING_MISMATCH")
if manifest.get("execution_binding_sha256") != sha256_json(manifest["execution_binding"]):
raise RuntimeError("APPROVED_EXECUTION_BINDING_SHA256_MISMATCH")
if manifest.get("expires_at") != bindings["expires_at"]:
raise RuntimeError("APPROVAL_EXPIRY_BINDING_MISMATCH")
evidence_git_dir = Path(str(bindings["evidence_git_dir"]))
evidence_oid = str(bindings["evidence_closure_oid"])
if evidence_oid != bindings["observed_evidence_closure_oid"]:
raise RuntimeError("WRONG_EVIDENCE_CLOSURE_DENIED")
_git(evidence_git_dir, "cat-file", "-e", f"{evidence_oid}^{{object}}")
db_path = Path(str(bindings["approval_db_path"]))
if not db_path.is_absolute() or not db_path.is_file() or db_path.is_symlink():
raise RuntimeError("APPROVAL_DATABASE_INVALID")
expiry = _parse_utc(bindings["expires_at"])
if datetime.now(timezone.utc) >= expiry:
raise RuntimeError("APPROVAL_EXPIRED")
with sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) as connection:
connection.row_factory = sqlite3.Row
pr = connection.execute(
"SELECT pr.issue_id,pr.has_merged,pr.merged_commit_id,i.is_closed "
"FROM pull_request pr JOIN issue i ON i.id=pr.issue_id "
"WHERE i.repo_id=? AND pr.\"index\"=? AND pr.base_branch=?",
(int(bindings["repo_id"]), int(bindings["activation_approval_pr_number"]), str(bindings["branch"])),
).fetchone()
if pr is None or int(pr["has_merged"]) != 1 or int(pr["is_closed"]) != 1:
raise RuntimeError("ACTIVATION_APPROVAL_PR_NOT_MERGED")
if pr["merged_commit_id"] != bindings["activation_approval_merge_oid"]:
raise RuntimeError("ACTIVATION_APPROVAL_MERGE_MISMATCH")
review = connection.execute(
"SELECT r.id,r.type,r.reviewer_id,r.commit_id,r.stale,r.dismissed,r.created_unix,u.lower_name "
"FROM review r JOIN user u ON u.id=r.reviewer_id WHERE r.id=? AND r.issue_id=?",
(int(bindings["owner_review_oid"]), int(pr["issue_id"])),
).fetchone()
if review is None or int(review["id"]) != int(bindings["observed_owner_review_oid"]):
raise RuntimeError("WRONG_APPROVAL_OID_DENIED")
if (
int(review["type"]) != 1
or int(review["reviewer_id"]) != 2
or review["lower_name"] != "owner"
or review["commit_id"] != bindings["approval_request_head_oid"]
or int(review["stale"]) != 0
or int(review["dismissed"]) != 0
or datetime.fromtimestamp(int(review["created_unix"]), timezone.utc) >= expiry
):
raise RuntimeError("OWNER_APPROVAL_INVALID")
statuses = connection.execute(
"SELECT id,context,state,sha FROM commit_status "
"WHERE repo_id=? AND sha=? ORDER BY id DESC",
(int(bindings["repo_id"]), str(bindings["approval_request_head_oid"])),
).fetchall()
observed: dict[str, sqlite3.Row] = {}
for row in statuses:
if row["context"] in REQUIRED_MACHINE_CONTEXTS:
observed.setdefault(row["context"], row)
if set(observed) != REQUIRED_MACHINE_CONTEXTS:
raise RuntimeError("MACHINE_GATE_SET_MISMATCH")
if any(row["state"] != "success" or row["sha"] != bindings["approval_request_head_oid"] for row in observed.values()):
raise RuntimeError("MACHINE_GATE_FAILED")
def validate_bindings(bindings: Mapping[str, Any]) -> None:
missing = sorted(REQUIRED_BINDINGS - set(bindings))
if missing:
raise RuntimeError("MISSING_BINDINGS:" + ",".join(missing))
for name in (
"expected_governance_oid",
"activation_approval_merge_oid",
"evidence_closure_oid",
):
_require_hex(name, bindings[name], (40,))
for name in (
"expected_validator_sha256",
"expected_state_machine_sha256",
"expected_branch_protection_sha256",
"observed_validator_sha256",
"observed_state_machine_sha256",
"expected_gitea_schema_sha256",
"activation_manifest_sha256",
):
_require_hex(name, bindings[name], (64,))
_require_hex("activation_manifest_blob_oid", bindings["activation_manifest_blob_oid"], (40, 64))
if not isinstance(bindings["owner_review_oid"], (int, str)) or str(bindings["owner_review_oid"]).strip() == "":
raise RuntimeError("OWNER_REVIEW_OID_INVALID")
if bindings["governance_ref"] != "refs/heads/governance/approvals":
raise RuntimeError("GOVERNANCE_REF_INVALID")
if bindings["branch"] != "governance/approvals":
raise RuntimeError("BRANCH_BINDING_DENIED")
if bindings["backend"] not in {"file", "sqlite"}:
raise RuntimeError("UNKNOWN_PROTECTION_BACKEND")
if bindings["observed_governance_oid"] != bindings["expected_governance_oid"]:
raise RuntimeError("WRONG_GOVERNANCE_OID_DENIED")
if bindings["observed_evidence_closure_oid"] != bindings["evidence_closure_oid"]:
raise RuntimeError("WRONG_EVIDENCE_CLOSURE_DENIED")
if str(bindings["observed_owner_review_oid"]) != str(bindings["owner_review_oid"]):
raise RuntimeError("WRONG_APPROVAL_OID_DENIED")
if bindings.get("requester") == "owner":
raise RuntimeError("REQUESTER_SELF_APPROVAL_DENIED")
if bindings.get("executor") == "owner":
raise RuntimeError("EXECUTOR_SELF_APPROVAL_DENIED")
if bindings.get("wip"):
raise RuntimeError("WIP_BLOCKED")
checks = bindings.get("machine_checks")
if not isinstance(checks, dict) or set(checks) != REQUIRED_MACHINE_CONTEXTS:
raise RuntimeError("MACHINE_GATE_SET_MISMATCH")
for context, check in checks.items():
if not isinstance(check, dict) or check.get("state") != "success" or check.get("head_oid") != bindings["approval_request_head_oid"]:
raise RuntimeError(f"MACHINE_GATE_FAILED:{context}")
forbidden = bindings.get("forbidden_operations", {})
if any(bool(value) for value in forbidden.values()):
raise RuntimeError("FORBIDDEN_OPERATION_REQUESTED")
if os.environ.get("AIOS_GITOPS_WRITES_ENABLED", "false").lower() != "false":
raise RuntimeError("GLOBAL_GITOPS_WRITES_MUST_REMAIN_FALSE")
if not isinstance(bindings.get("target_branch_protection"), dict):
raise RuntimeError("TARGET_BRANCH_PROTECTION_REQUIRED")
_verify_external_evidence(bindings)
def binding_payload_hash(bindings: Mapping[str, Any]) -> str:
stable = dict(bindings)
stable.pop("crash_at", None)
return sha256_json(redact(stable))
def plan_id(bindings: Mapping[str, Any], previous: Mapping[str, Any], target: Mapping[str, Any]) -> str:
stable = dict(bindings)
stable.pop("crash_at", None)
return sha256_json(
{
"bindings": redact(stable),
"previous": dict(previous),
"target": dict(target),
}
)
class ActivationEngine:
def __init__(self, adapter: Adapter, state_root: Path, lock_path: Path) -> None:
self.adapter = adapter
self.journal = JournalStore(state_root)
self.lock_path = lock_path
def plan(self, bindings: Mapping[str, Any]) -> dict[str, Any]:
validate_bindings(bindings)
previous = self.adapter.read_current()
previous_sha = calculate_snapshot_sha256(previous)
if previous_sha != bindings["expected_branch_protection_sha256"]:
raise CASConflict("EXPECTED_BRANCH_PROTECTION_SHA256_MISMATCH")
target = dict(bindings["target_branch_protection"])
# Validate without writing by using the public target validator through a
# disposable comparison: adapters enforce the same checks again in CAS.
from .gitea_protected_branch import validate_target
validate_target(previous, target)
result = {
"status": "PLAN_READY",
"task_id": bindings["task_id"],
"operation_id": bindings["operation_id"],
"plan_id": plan_id(bindings, previous, target),
"payload_hash": binding_payload_hash(bindings),
"previous_sha256": previous_sha,
"target_sha256": calculate_snapshot_sha256(target),
}
return result
def _recover(self, record: dict[str, Any]) -> dict[str, Any]:
current = self.adapter.read_current()
current_sha = calculate_snapshot_sha256(current)
previous_sha = record["previous_sha256"]
target_sha = record["target_sha256"]
if record["phase"] == "PREPARED":
if current_sha == previous_sha:
return {**record, "recovery": "RESUME_BEFORE_APPLY"}
if current_sha == target_sha:
record = {**record, "phase": "APPLIED", "recovery": "RESUME_AFTER_APPLY"}
self.journal.write(record["operation_id"], record)
return record
raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
if record["phase"] in {"APPLIED", "COMMITTED"} and current_sha not in {target_sha, previous_sha}:
raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
return record
def _ensure_closure(self, record: dict[str, Any], event: str) -> None:
expected_sha = (
record["target_sha256"] if record["phase"] == "COMMITTED" else record["previous_sha256"]
)
if calculate_snapshot_sha256(self.adapter.read_current()) != expected_sha:
raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
atomic_write_json(
self.journal.root / "current-snapshot.json",
record["target"] if record["phase"] == "COMMITTED" else record["previous"],
)
atomic_write_json(self.journal.root / "rollback-snapshot.json", record["previous"])
self.journal.write_receipt(record["operation_id"], record)
if not self.journal.audit.has_event(record["operation_id"], event):
payload = {"event": event}
if event == "SINGLE_OWNER_ACTIVATION_COMMITTED":
payload.update(
payload_hash=record["payload_hash"],
target_sha256=record["target_sha256"],
)
self.journal.audit.append(record["operation_id"], payload)
atomic_write_json(self.journal.root / "state.json", record)
def activate(self, bindings: Mapping[str, Any], supplied_plan_id: str) -> dict[str, Any]:
with exclusive_lock(self.lock_path):
self.journal.audit.verify()
validate_bindings(bindings)
operation_id = str(bindings["operation_id"])
payload_hash = binding_payload_hash(bindings)
record = self.journal.load(operation_id)
if record:
if record.get("payload_hash") != payload_hash:
raise RuntimeError("OPERATION_ID_PAYLOAD_MISMATCH_DENIED")
if record.get("plan_id") != supplied_plan_id:
raise RuntimeError("PLAN_ID_MISMATCH")
record = self._recover(record)
if record["phase"] == "COMMITTED":
self._ensure_closure(record, "SINGLE_OWNER_ACTIVATION_COMMITTED")
return {"status": "ALREADY_COMMITTED", **record}
else:
planned = self.plan(bindings)
if supplied_plan_id != planned["plan_id"]:
raise RuntimeError("PLAN_ID_MISMATCH")
previous = self.adapter.read_current()
target = dict(bindings["target_branch_protection"])
if record is None:
record = {
"task_id": bindings["task_id"],
"operation_id": operation_id,
"transaction_id": operation_id,
"payload_hash": payload_hash,
"plan_id": supplied_plan_id,
"phase": "PREPARED",
"previous": previous,
"target": target,
"previous_sha256": calculate_snapshot_sha256(previous),
"target_sha256": calculate_snapshot_sha256(target),
"activation_approval_merge_oid": bindings["activation_approval_merge_oid"],
"evidence_closure_oid": bindings["evidence_closure_oid"],
"owner_review_oid": bindings["owner_review_oid"],
}
self.journal.write(operation_id, record)
atomic_write_json(self.journal.root / "current-snapshot.json", previous)
atomic_write_json(self.journal.root / "rollback-snapshot.json", previous)
crash_at = bindings.get("crash_at")
if crash_at == "before_apply":
raise InjectedCrash("before_apply")
try:
current_sha = calculate_snapshot_sha256(self.adapter.read_current())
if current_sha == record["previous_sha256"]:
applied = self.adapter.apply_atomic(record["previous_sha256"], record["target"])
record["target"] = applied
record["target_sha256"] = calculate_snapshot_sha256(applied)
elif current_sha != record["target_sha256"]:
raise CASConflict("CAS_CONFLICT")
if crash_at == "during_apply":
raise InjectedCrash("during_apply")
record["phase"] = "APPLIED"
self.journal.write(operation_id, record)
if crash_at == "after_apply_before_receipt":
raise InjectedCrash("after_apply_before_receipt")
if not self.adapter.verify_applied(record["target_sha256"]):
raise RuntimeError("POST_ACTIVATION_VERIFY_FAILED")
record["phase"] = "COMMITTED"
record["single_owner_policy_active"] = True
record["break_glass_active"] = False
self.journal.write(operation_id, record)
if crash_at == "after_commit_wal":
raise InjectedCrash("after_commit_wal")
self._ensure_closure(record, "SINGLE_OWNER_ACTIVATION_COMMITTED")
return {"status": "ACTIVATED", **record}
except InjectedCrash:
raise
except Exception as exc:
failure = {**record, "phase": "FAILED", "error": type(exc).__name__}
current_sha = calculate_snapshot_sha256(self.adapter.read_current())
if current_sha == record["target_sha256"]:
self.adapter.rollback_atomic(record["target_sha256"], record["previous"])
if not self.adapter.verify_rollback(record["previous_sha256"]):
raise RuntimeError("ROLLBACK_VERIFY_FAILED") from exc
failure["phase"] = "ROLLED_BACK"
self.journal.write(operation_id, failure)
self.journal.write_receipt(operation_id, failure)
self.journal.audit.append(operation_id, {"event": failure["phase"], "error": type(exc).__name__})
raise
def verify(self, bindings: Mapping[str, Any], operation_id: str) -> dict[str, Any]:
validate_bindings(bindings)
record = self.journal.load(operation_id)
if not record:
raise RuntimeError("TRANSACTION_NOT_FOUND")
if record.get("payload_hash") != binding_payload_hash(bindings):
raise RuntimeError("OPERATION_ID_PAYLOAD_MISMATCH_DENIED")
self.journal.audit.verify()
current_sha = calculate_snapshot_sha256(self.adapter.read_current())
passed = record.get("phase") == "COMMITTED" and current_sha == record.get("target_sha256")
if not passed:
raise RuntimeError("ACTIVATION_VERIFY_FAILED")
return {"status": "VERIFIED", "operation_id": operation_id, "target_sha256": current_sha}
def rollback(
self, bindings: Mapping[str, Any], operation_id: str, supplied_plan_id: str
) -> dict[str, Any]:
with exclusive_lock(self.lock_path):
self.journal.audit.verify()
validate_bindings(bindings)
record = self.journal.load(operation_id)
if not record:
raise RuntimeError("TRANSACTION_NOT_FOUND")
if record.get("payload_hash") != binding_payload_hash(bindings):
raise RuntimeError("OPERATION_ID_PAYLOAD_MISMATCH_DENIED")
if record.get("plan_id") != supplied_plan_id:
raise RuntimeError("PLAN_ID_MISMATCH")
for field in (
"activation_approval_merge_oid",
"evidence_closure_oid",
"owner_review_oid",
):
if str(record.get(field)) != str(bindings[field]):
raise RuntimeError("ROLLBACK_AUTHORIZATION_MISMATCH_DENIED")
current_sha = calculate_snapshot_sha256(self.adapter.read_current())
if current_sha == record["previous_sha256"]:
record["phase"] = "ROLLED_BACK"
record["single_owner_policy_active"] = False
self.journal.write(operation_id, record)
self._ensure_closure(record, "ACTIVATION_ROLLED_BACK")
return {"status": "ALREADY_ROLLED_BACK", "operation_id": operation_id}
if current_sha != record["target_sha256"]:
raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
self.adapter.rollback_atomic(record["target_sha256"], record["previous"])
if not self.adapter.verify_rollback(record["previous_sha256"]):
raise RuntimeError("ROLLBACK_VERIFY_FAILED")
if bindings.get("crash_at") == "after_rollback_cas":
raise InjectedCrash("after_rollback_cas")
record["phase"] = "ROLLED_BACK"
record["single_owner_policy_active"] = False
self.journal.write(operation_id, record)
if bindings.get("crash_at") == "after_rollback_wal":
raise InjectedCrash("after_rollback_wal")
self._ensure_closure(record, "ACTIVATION_ROLLED_BACK")
return {"status": "ROLLED_BACK", "operation_id": operation_id}
def status(self, operation_id: str | None = None) -> dict[str, Any]:
audit_head = self.journal.audit.verify()
if operation_id:
record = self.journal.load(operation_id)
if not record:
raise RuntimeError("TRANSACTION_NOT_FOUND")
return {"status": "OK", "audit_head": audit_head, "transaction": record}
state_path = self.journal.root / "state.json"
return {
"status": "OK",
"audit_head": audit_head,
"state": read_json(state_path) if state_path.exists() else None,
"transactions": list(self.journal.iter_transactions()),
}
def load_bindings(path: Path) -> dict[str, Any]:
return read_json(path)
def adapter_from_bindings(bindings: Mapping[str, Any]) -> Adapter:
backend = bindings.get("backend", "file")
backend_path = Path(str(bindings.get("backend_path", "")))
if not backend_path.is_absolute():
raise RuntimeError("BACKEND_PATH_MUST_BE_ABSOLUTE")
if backend == "file":
return FileProtectedBranchAdapter(backend_path)
if backend == "sqlite":
return SQLiteProtectedBranchAdapter(
backend_path,
repo_id=int(bindings.get("repo_id", 1)),
branch=str(bindings.get("branch", "governance/approvals")),
expected_schema_sha256=str(bindings["expected_gitea_schema_sha256"]),
)
raise RuntimeError("UNKNOWN_PROTECTION_BACKEND")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser()
action = parser.add_mutually_exclusive_group(required=True)
for name in ("plan", "activate", "verify", "rollback", "status"):
action.add_argument(f"--{name}", action="store_true")
parser.add_argument("--bindings", type=Path)
parser.add_argument("--plan-id")
parser.add_argument("--transaction")
parser.add_argument("--state-root", type=Path, default=DEFAULT_STATE_ROOT)
parser.add_argument("--lock-path", type=Path, default=DEFAULT_LOCK_PATH)
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
if args.status and args.bindings is None:
# Status without bindings is intentionally limited to durable local state.
journal = JournalStore(args.state_root)
result = {
"status": "OK",
"audit_head": journal.audit.verify(),
"transactions": list(journal.iter_transactions()),
}
print(json.dumps(result, sort_keys=True))
return 0
if args.bindings is None:
raise SystemExit("--bindings is required")
bindings = load_bindings(args.bindings)
engine = ActivationEngine(adapter_from_bindings(bindings), args.state_root, args.lock_path)
if args.plan:
result = engine.plan(bindings)
elif args.activate:
if not args.plan_id:
raise SystemExit("--plan-id is required")
result = engine.activate(bindings, args.plan_id)
elif args.verify:
result = engine.verify(bindings, args.transaction or str(bindings["operation_id"]))
elif args.rollback:
if not args.plan_id:
raise SystemExit("--plan-id is required")
result = engine.rollback(
bindings, args.transaction or str(bindings["operation_id"]), args.plan_id
)
else:
result = engine.status(args.transaction)
print(json.dumps(redact(result), sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,61 @@
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import json
from typing import Mapping
def canonical_json(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
@dataclass(frozen=True)
class EvidenceClosure:
evidence: Mapping[str, str]
closure_sha256: str
class ActivationEvidenceClosureBuilder:
def build(self, evidence: Mapping[str, str]) -> EvidenceClosure:
required = {
"activation_prerequisites_commit_oid",
"activation_prerequisites_tree_oid",
"artifact_sha256",
"rollback_artifact_sha256",
"validator_sha256",
"state_machine_sha256",
"branch_protection_adapter_sha256",
"rollback_configuration_sha256",
"test_evidence_sha256",
"deployment_manifest_sha256",
"dormant_unit_manifest_sha256",
"dormant_deployment_evidence_sha256",
}
if set(evidence) != required:
raise ValueError("evidence set mismatch")
for value in evidence.values():
if len(value) != 40 and len(value) != 64:
raise ValueError("invalid evidence digest")
return EvidenceClosure(
dict(evidence), hashlib.sha256(canonical_json(evidence)).hexdigest()
)
class ApprovalMirrorStatusVerifier:
def verify(
self,
expected_oid: str,
*,
gitea_oid: str,
quarantine_oid: str,
authoritative_oid: str,
controller_visible_oid: str,
fsck_connectivity: str,
) -> bool:
return (
len(expected_oid) == 40
and {gitea_oid, quarantine_oid, authoritative_oid, controller_visible_oid}
== {expected_oid}
and fsck_connectivity == "PASS"
)
@@ -0,0 +1,26 @@
from __future__ import annotations
from contextlib import AbstractContextManager
from pathlib import Path
import shutil
import tempfile
from .models import ProtectionSnapshot
from .protection import MemoryProtectionStore
class DisposableGiteaFixture(AbstractContextManager):
def __init__(self, initial: ProtectionSnapshot) -> None:
self.root = Path(tempfile.mkdtemp(prefix="aios-governance-fixture-"))
self.protection = MemoryProtectionStore(initial)
self.destroyed = False
def __enter__(self) -> "DisposableGiteaFixture":
return self
def __exit__(self, exc_type, exc, tb) -> None:
resolved = self.root.resolve()
if not resolved.name.startswith("aios-governance-fixture-"):
raise RuntimeError("fixture cleanup path guard failed")
shutil.rmtree(resolved)
self.destroyed = True
@@ -0,0 +1,260 @@
from __future__ import annotations
import json
from pathlib import Path
import sqlite3
import time
from typing import Any, Mapping
from .journal import atomic_write_json, read_json, sha256_json
class CASConflict(RuntimeError):
pass
class UnsafePolicy(RuntimeError):
pass
def calculate_snapshot_sha256(snapshot: Mapping[str, Any]) -> str:
return sha256_json(dict(snapshot))
def validate_target(previous: Mapping[str, Any], target: Mapping[str, Any]) -> None:
if set(previous) != set(target):
raise UnsafePolicy("PROTECTION_ROW_SHAPE_CHANGED")
for field in ("id", "repo_id", "branch_name"):
if previous.get(field) != target.get(field):
raise UnsafePolicy(f"PROTECTION_IDENTITY_CHANGED:{field}")
if target.get("branch_name") != "governance/approvals":
raise UnsafePolicy("BRANCH_BINDING_DENIED")
if int(target.get("can_push", 1)) != 0:
raise UnsafePolicy("DIRECT_PUSH_MUST_REMAIN_DISABLED")
if int(target.get("required_approvals", 0)) != 1:
raise UnsafePolicy("EXACTLY_ONE_APPROVAL_REQUIRED")
if int(target.get("enable_approvals_whitelist", 0)) != 1:
raise UnsafePolicy("APPROVAL_WHITELIST_MUST_REMAIN_ENABLED")
if json.loads(target.get("approvals_whitelist_user_i_ds", "[]")) != [2]:
raise UnsafePolicy("OWNER_APPROVAL_WHITELIST_REQUIRED")
if int(target.get("enable_merge_whitelist", 0)) != 1:
raise UnsafePolicy("MERGE_WHITELIST_MUST_REMAIN_ENABLED")
if target.get("merge_whitelist_user_i_ds") != previous.get("merge_whitelist_user_i_ds"):
raise UnsafePolicy("MERGE_EXECUTOR_WHITELIST_CHANGE_DENIED")
if int(target.get("dismiss_stale_approvals", 0)) != 1:
raise UnsafePolicy("DISMISS_STALE_APPROVALS_REQUIRED")
if int(target.get("enable_status_check", 0)) != 1:
raise UnsafePolicy("STATUS_CHECKS_REQUIRED")
required_contexts = {
"aios/validator",
"aios/state-machine",
"aios/branch-protection",
"aios/secret-scan",
"aios/evidence-closure",
}
try:
contexts = set(json.loads(target.get("status_check_contexts", "[]")))
except (TypeError, ValueError) as exc:
raise UnsafePolicy("STATUS_CHECK_CONTEXTS_INVALID") from exc
if contexts != required_contexts:
raise UnsafePolicy("STATUS_CHECK_CONTEXTS_MISMATCH")
if int(target.get("block_on_rejected_reviews", 0)) != 1:
raise UnsafePolicy("REJECTED_REVIEWS_MUST_BLOCK")
if int(target.get("block_on_outdated_branch", 0)) != 1:
raise UnsafePolicy("OUTDATED_BRANCH_MUST_BLOCK")
if target.get("require_signed_commits") != previous.get("require_signed_commits"):
raise UnsafePolicy("SIGNATURE_POLICY_CHANGE_DENIED")
class FileProtectedBranchAdapter:
def __init__(self, path: Path) -> None:
self.path = path
@staticmethod
def calculate_snapshot_sha256(snapshot: Mapping[str, Any]) -> str:
return calculate_snapshot_sha256(snapshot)
def read_current(self) -> dict[str, Any]:
return read_json(self.path)
def compare_expected(self, expected_sha256: str) -> bool:
return calculate_snapshot_sha256(self.read_current()) == expected_sha256
def apply_atomic(
self, expected_sha256: str, target: Mapping[str, Any]
) -> dict[str, Any]:
previous = self.read_current()
if calculate_snapshot_sha256(previous) != expected_sha256:
raise CASConflict("CAS_CONFLICT")
validate_target(previous, target)
atomic_write_json(self.path, dict(target))
applied = self.read_current()
if calculate_snapshot_sha256(applied) != calculate_snapshot_sha256(target):
raise CASConflict("CAS_VERIFY_FAILED")
return applied
def verify_applied(self, expected_sha256: str) -> bool:
return self.compare_expected(expected_sha256)
def rollback_atomic(
self, expected_target_sha256: str, previous: Mapping[str, Any]
) -> dict[str, Any]:
current = self.read_current()
if calculate_snapshot_sha256(current) != expected_target_sha256:
raise CASConflict("ROLLBACK_CONFLICT")
atomic_write_json(self.path, dict(previous))
restored = self.read_current()
if calculate_snapshot_sha256(restored) != calculate_snapshot_sha256(previous):
raise CASConflict("ROLLBACK_VERIFY_FAILED")
return restored
def verify_rollback(self, expected_sha256: str) -> bool:
return self.compare_expected(expected_sha256)
class SQLiteProtectedBranchAdapter:
ALLOWED_UPDATE_FIELDS = (
"can_push",
"enable_merge_whitelist",
"merge_whitelist_user_i_ds",
"enable_status_check",
"status_check_contexts",
"enable_approvals_whitelist",
"approvals_whitelist_user_i_ds",
"required_approvals",
"block_on_rejected_reviews",
"block_on_official_review_requests",
"block_on_outdated_branch",
"dismiss_stale_approvals",
"ignore_stale_approvals",
"updated_unix",
)
def __init__(
self,
db_path: Path,
*,
repo_id: int = 1,
branch: str = "governance/approvals",
expected_schema_sha256: str,
) -> None:
self.db_path = db_path
self.repo_id = repo_id
self.branch = branch
self.expected_schema_sha256 = expected_schema_sha256
def _connect(self) -> sqlite3.Connection:
connection = sqlite3.connect(self.db_path, timeout=10)
connection.row_factory = sqlite3.Row
return connection
def schema_sha256(self, connection: sqlite3.Connection | None = None) -> str:
own = connection is None
connection = connection or self._connect()
try:
rows = [dict(row) for row in connection.execute("PRAGMA table_info(protected_branch)")]
return sha256_json(rows)
finally:
if own:
connection.close()
def _check_schema(self, connection: sqlite3.Connection) -> None:
if not isinstance(self.expected_schema_sha256, str) or len(self.expected_schema_sha256) != 64:
raise RuntimeError("GITEA_SCHEMA_FINGERPRINT_REQUIRED")
if self.schema_sha256(connection) != self.expected_schema_sha256:
raise RuntimeError("GITEA_SCHEMA_FINGERPRINT_MISMATCH")
def _read(self, connection: sqlite3.Connection) -> dict[str, Any]:
row = connection.execute(
"SELECT * FROM protected_branch WHERE repo_id=? AND branch_name=?",
(self.repo_id, self.branch),
).fetchone()
if row is None:
raise RuntimeError("PROTECTED_BRANCH_ROW_NOT_FOUND")
return dict(row)
def read_current(self) -> dict[str, Any]:
with self._connect() as connection:
self._check_schema(connection)
return self._read(connection)
@staticmethod
def calculate_snapshot_sha256(snapshot: Mapping[str, Any]) -> str:
return calculate_snapshot_sha256(snapshot)
def compare_expected(self, expected_sha256: str) -> bool:
return calculate_snapshot_sha256(self.read_current()) == expected_sha256
def apply_atomic(
self, expected_sha256: str, target: Mapping[str, Any]
) -> dict[str, Any]:
connection = self._connect()
try:
connection.execute("BEGIN IMMEDIATE")
self._check_schema(connection)
previous = self._read(connection)
if calculate_snapshot_sha256(previous) != expected_sha256:
raise CASConflict("CAS_CONFLICT")
validate_target(previous, target)
values = dict(target)
values["updated_unix"] = int(values.get("updated_unix") or time.time())
assignments = ",".join(f"{field}=?" for field in self.ALLOWED_UPDATE_FIELDS)
parameters = [values[field] for field in self.ALLOWED_UPDATE_FIELDS]
parameters.extend((self.repo_id, self.branch))
cursor = connection.execute(
f"UPDATE protected_branch SET {assignments} WHERE repo_id=? AND branch_name=?",
parameters,
)
if cursor.rowcount != 1:
raise CASConflict("CAS_ROWCOUNT_INVALID")
applied = self._read(connection)
expected_target = dict(target)
expected_target["updated_unix"] = values["updated_unix"]
if calculate_snapshot_sha256(applied) != calculate_snapshot_sha256(expected_target):
raise CASConflict("CAS_VERIFY_FAILED")
connection.commit()
return applied
except Exception:
connection.rollback()
raise
finally:
connection.close()
def verify_applied(self, expected_sha256: str) -> bool:
return self.compare_expected(expected_sha256)
def rollback_atomic(
self, expected_target_sha256: str, previous: Mapping[str, Any]
) -> dict[str, Any]:
connection = self._connect()
try:
connection.execute("BEGIN IMMEDIATE")
self._check_schema(connection)
current = self._read(connection)
if calculate_snapshot_sha256(current) != expected_target_sha256:
raise CASConflict("ROLLBACK_CONFLICT")
for field in ("id", "repo_id", "branch_name"):
if current.get(field) != previous.get(field):
raise CASConflict("ROLLBACK_IDENTITY_MISMATCH")
assignments = ",".join(f"{field}=?" for field in self.ALLOWED_UPDATE_FIELDS)
parameters = [previous[field] for field in self.ALLOWED_UPDATE_FIELDS]
parameters.extend((self.repo_id, self.branch))
cursor = connection.execute(
f"UPDATE protected_branch SET {assignments} WHERE repo_id=? AND branch_name=?",
parameters,
)
if cursor.rowcount != 1:
raise CASConflict("ROLLBACK_ROWCOUNT_INVALID")
restored = self._read(connection)
if calculate_snapshot_sha256(restored) != calculate_snapshot_sha256(previous):
raise CASConflict("ROLLBACK_VERIFY_FAILED")
connection.commit()
return restored
except Exception:
connection.rollback()
raise
finally:
connection.close()
def verify_rollback(self, expected_sha256: str) -> bool:
return self.compare_expected(expected_sha256)
@@ -0,0 +1,189 @@
from __future__ import annotations
import hashlib
import json
import os
from pathlib import Path
import tempfile
from typing import Any, Iterable
SENSITIVE_FRAGMENTS = (
"authorization",
"cookie",
"credential",
"password",
"private_key",
"secret",
"token",
)
def canonical_json(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
def sha256_json(value: object) -> str:
return hashlib.sha256(canonical_json(value)).hexdigest()
def redact(value: Any) -> Any:
if isinstance(value, dict):
return {
str(key): (
"[REDACTED]"
if any(fragment in str(key).lower() for fragment in SENSITIVE_FRAGMENTS)
else redact(item)
)
for key, item in value.items()
}
if isinstance(value, list):
return [redact(item) for item in value]
if isinstance(value, tuple):
return [redact(item) for item in value]
return value
def ensure_state_directory(path: Path) -> None:
if path.is_symlink():
raise RuntimeError("STATE_DIRECTORY_SYMLINK_DENIED")
path.mkdir(parents=True, exist_ok=True, mode=0o750)
path.chmod(0o750)
mode = path.stat().st_mode & 0o777
if mode != 0o750:
raise RuntimeError("STATE_DIRECTORY_MODE_INVALID")
def atomic_write_json(path: Path, value: object) -> None:
ensure_state_directory(path.parent)
raw = json.dumps(redact(value), sort_keys=True, indent=2).encode("utf-8") + b"\n"
with tempfile.NamedTemporaryFile(
dir=path.parent, prefix=f".{path.name}.", delete=False
) as handle:
temporary = Path(handle.name)
handle.write(raw)
handle.flush()
os.fsync(handle.fileno())
temporary.chmod(0o640)
os.replace(temporary, path)
path.chmod(0o640)
directory_fd = os.open(path.parent, os.O_RDONLY)
try:
os.fsync(directory_fd)
finally:
os.close(directory_fd)
def read_json(path: Path) -> dict[str, Any]:
raw = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(raw, dict):
raise RuntimeError("STATE_DOCUMENT_NOT_OBJECT")
return raw
class AuditChain:
def __init__(self, path: Path) -> None:
self.path = path
def _records(self) -> list[dict[str, Any]]:
if not self.path.exists():
return []
records: list[dict[str, Any]] = []
for line in self.path.read_text(encoding="utf-8").splitlines():
if line.strip():
record = json.loads(line)
if not isinstance(record, dict):
raise RuntimeError("AUDIT_RECORD_INVALID")
records.append(record)
return records
def verify(self) -> str:
previous = "0" * 64
for index, record in enumerate(self._records(), 1):
payload = record.get("payload")
if record.get("sequence") != index:
raise RuntimeError("AUDIT_SEQUENCE_INVALID")
if record.get("previous_event_hash") != previous:
raise RuntimeError("AUDIT_PREVIOUS_HASH_INVALID")
payload_hash = sha256_json(payload)
if record.get("payload_hash") != payload_hash:
raise RuntimeError("AUDIT_PAYLOAD_HASH_INVALID")
material = {
"sequence": index,
"previous_event_hash": previous,
"payload_hash": payload_hash,
"transaction_id": record.get("transaction_id"),
}
event_hash = sha256_json(material)
if record.get("event_hash") != event_hash:
raise RuntimeError("AUDIT_EVENT_HASH_INVALID")
previous = event_hash
return previous
def append(self, transaction_id: str, payload: dict[str, Any]) -> dict[str, Any]:
ensure_state_directory(self.path.parent)
previous = self.verify()
sequence = len(self._records()) + 1
clean_payload = redact(payload)
payload_hash = sha256_json(clean_payload)
material = {
"sequence": sequence,
"previous_event_hash": previous,
"payload_hash": payload_hash,
"transaction_id": transaction_id,
}
record = {
**material,
"payload": clean_payload,
"event_hash": sha256_json(material),
}
fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o640)
try:
os.write(fd, canonical_json(record) + b"\n")
os.fsync(fd)
finally:
os.close(fd)
self.path.chmod(0o640)
return record
def has_event(self, transaction_id: str, event: str) -> bool:
self.verify()
return any(
record.get("transaction_id") == transaction_id
and isinstance(record.get("payload"), dict)
and record["payload"].get("event") == event
for record in self._records()
)
class JournalStore:
def __init__(self, root: Path) -> None:
self.root = root
self.wal = root / "wal"
self.receipts = root / "receipts"
ensure_state_directory(root)
ensure_state_directory(self.wal)
ensure_state_directory(self.receipts)
self.audit = AuditChain(root / "audit.jsonl")
def transaction_path(self, operation_id: str) -> Path:
if not operation_id or any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-" for char in operation_id):
raise RuntimeError("OPERATION_ID_INVALID")
return self.wal / f"{operation_id}.json"
def receipt_path(self, operation_id: str) -> Path:
return self.receipts / f"{operation_id}.json"
def load(self, operation_id: str) -> dict[str, Any] | None:
path = self.transaction_path(operation_id)
return read_json(path) if path.exists() else None
def write(self, operation_id: str, value: dict[str, Any]) -> None:
atomic_write_json(self.transaction_path(operation_id), value)
def write_receipt(self, operation_id: str, value: dict[str, Any]) -> None:
atomic_write_json(self.receipt_path(operation_id), value)
def iter_transactions(self) -> Iterable[dict[str, Any]]:
for path in sorted(self.wal.glob("*.json")):
yield read_json(path)
@@ -0,0 +1,47 @@
from __future__ import annotations
from pathlib import Path
try:
import fcntl
except ImportError: # pragma: no cover - exercised by Windows validation
fcntl = None
import msvcrt
class MaintenanceLock:
"""Process fence used only by fixture and later approved activation executors."""
def __init__(self, path: Path) -> None:
self.path = path
self._handle = None
def __enter__(self) -> "MaintenanceLock":
self.path.parent.mkdir(parents=True, exist_ok=True)
self._handle = self.path.open("a+")
if fcntl is not None:
try:
fcntl.flock(self._handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
self._handle.close()
self._handle = None
raise
else:
self._handle.seek(0)
try:
msvcrt.locking(self._handle.fileno(), msvcrt.LK_NBLCK, 1)
except OSError as exc:
self._handle.close()
self._handle = None
raise BlockingIOError("maintenance lock held") from exc
return self
def __exit__(self, exc_type, exc, tb) -> None:
if self._handle is not None:
if fcntl is not None:
fcntl.flock(self._handle.fileno(), fcntl.LOCK_UN)
else:
self._handle.seek(0)
msvcrt.locking(self._handle.fileno(), msvcrt.LK_UNLCK, 1)
self._handle.close()
self._handle = None
@@ -0,0 +1,90 @@
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Mapping
class GovernanceState(str, Enum):
HUMAN_OWNER_APPROVAL_PENDING = "HUMAN_OWNER_APPROVAL_PENDING"
OWNER_APPROVAL_VERIFIED = "OWNER_APPROVAL_VERIFIED"
MACHINE_GATES_VERIFIED = "MACHINE_GATES_VERIFIED"
AUTO_MERGE_READY = "AUTO_MERGE_READY"
OWNER_APPROVED_EXECUTION_READY = "OWNER_APPROVED_EXECUTION_READY"
@dataclass(frozen=True)
class Review:
reviewer: str
reviewer_id: int
state: str
head_oid: str
reviewed_at: datetime
stale: bool = False
dismissed: bool = False
def __post_init__(self) -> None:
if self.reviewed_at.tzinfo is None:
raise ValueError("reviewed_at must be timezone-aware")
@dataclass(frozen=True)
class MachineGate:
context: str
state: str
head_oid: str
observed_at: datetime
def __post_init__(self) -> None:
if self.observed_at.tzinfo is None:
raise ValueError("observed_at must be timezone-aware")
@dataclass(frozen=True)
class ApprovalContext:
current_head_oid: str
requester: str
executor: str
owner: str
owner_id: int
reviews: tuple[Review, ...]
machine_gates: tuple[MachineGate, ...]
required_machine_contexts: tuple[str, ...]
expires_at: datetime
wip: bool
now: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
def __post_init__(self) -> None:
if self.expires_at.tzinfo is None or self.now.tzinfo is None:
raise ValueError("approval timestamps must be timezone-aware")
@dataclass(frozen=True)
class ProtectionSnapshot:
branch: str
required_approvals: int
approval_whitelist: tuple[str, ...]
merge_whitelist: tuple[str, ...]
dismiss_stale_approvals: bool
require_status_checks: bool
status_contexts: tuple[str, ...]
direct_push: bool
force_push: bool
branch_deletion: bool
version: int
def as_dict(self) -> Mapping[str, Any]:
return {
"branch": self.branch,
"required_approvals": self.required_approvals,
"approval_whitelist": list(self.approval_whitelist),
"merge_whitelist": list(self.merge_whitelist),
"dismiss_stale_approvals": self.dismiss_stale_approvals,
"require_status_checks": self.require_status_checks,
"status_contexts": list(self.status_contexts),
"direct_push": self.direct_push,
"force_push": self.force_push,
"branch_deletion": self.branch_deletion,
"version": self.version,
}
@@ -0,0 +1,23 @@
from __future__ import annotations
class OperationDenied(PermissionError):
pass
class DormantOperationGuard:
"""Fail-closed operation boundary for the prerequisite deployment."""
DENIED = frozenset({
"protected_branch.direct_push",
"governance.direct_push",
"tag.direct_write",
"git.force_push",
"production.without_owner",
"branch_protection.live_update",
"single_owner_policy.activate",
})
def authorize(self, operation: str, *, fixture: bool = False) -> None:
if operation in self.DENIED or not fixture:
raise OperationDenied(operation)
@@ -0,0 +1,98 @@
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import json
from typing import Protocol
from .models import ProtectionSnapshot
class CASConflict(RuntimeError):
pass
class DormantViolation(RuntimeError):
pass
def snapshot_sha256(snapshot: ProtectionSnapshot) -> str:
raw = json.dumps(snapshot.as_dict(), sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(raw).hexdigest()
class ProtectionStore(Protocol):
fixture: bool
def read(self, branch: str) -> ProtectionSnapshot: ...
def compare_and_swap(
self, branch: str, expected_sha256: str, target: ProtectionSnapshot
) -> bool: ...
@dataclass(frozen=True)
class ProtectionPlan:
branch: str
expected_previous_sha256: str
previous: ProtectionSnapshot
target: ProtectionSnapshot
target_sha256: str
class BranchProtectionCASMigrationAdapter:
"""Candidate adapter. Mutations are restricted to disposable fixtures."""
def plan(
self,
current: ProtectionSnapshot,
target: ProtectionSnapshot,
expected_previous_sha256: str,
) -> ProtectionPlan:
actual = snapshot_sha256(current)
if actual != expected_previous_sha256:
raise CASConflict("EXPECTED_PREVIOUS_SHA256_MISMATCH")
if current.branch != target.branch:
raise CASConflict("BRANCH_BINDING_MISMATCH")
if target.direct_push or target.force_push or target.branch_deletion:
raise DormantViolation("UNSAFE_TARGET_POLICY")
return ProtectionPlan(
current.branch, actual, current, target, snapshot_sha256(target)
)
def apply_fixture(self, store: ProtectionStore, plan: ProtectionPlan) -> ProtectionSnapshot:
if not store.fixture:
raise DormantViolation("LIVE_PROTECTION_MUTATION_DENIED")
if not store.compare_and_swap(
plan.branch, plan.expected_previous_sha256, plan.target
):
raise CASConflict("CONCURRENT_PROTECTION_CHANGE")
return store.read(plan.branch)
def rollback_fixture(self, store: ProtectionStore, plan: ProtectionPlan) -> ProtectionSnapshot:
if not store.fixture:
raise DormantViolation("LIVE_PROTECTION_ROLLBACK_DENIED")
if not store.compare_and_swap(plan.branch, plan.target_sha256, plan.previous):
raise CASConflict("ROLLBACK_CONFLICT")
return store.read(plan.branch)
class MemoryProtectionStore:
fixture = True
def __init__(self, snapshot: ProtectionSnapshot) -> None:
self._snapshot = snapshot
def read(self, branch: str) -> ProtectionSnapshot:
if self._snapshot.branch != branch:
raise KeyError(branch)
return self._snapshot
def compare_and_swap(
self, branch: str, expected_sha256: str, target: ProtectionSnapshot
) -> bool:
if self._snapshot.branch != branch:
return False
if snapshot_sha256(self._snapshot) != expected_sha256:
return False
self._snapshot = target
return True
@@ -0,0 +1,69 @@
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import json
from pathlib import Path
import tempfile
from .models import ProtectionSnapshot
from .protection import MemoryProtectionStore, ProtectionPlan, snapshot_sha256
@dataclass(frozen=True)
class RollbackConfiguration:
branch: str
previous_sha256: str
target_sha256: str
payload: dict
class RollbackConfigurationManager:
def build(self, plan: ProtectionPlan) -> RollbackConfiguration:
return RollbackConfiguration(
plan.branch,
plan.expected_previous_sha256,
plan.target_sha256,
{"previous": plan.previous.as_dict(), "target": plan.target.as_dict()},
)
def install_fixture(self, root: Path, config: RollbackConfiguration) -> Path:
if not root.name.startswith("aios-governance-fixture-"):
raise PermissionError("rollback configuration may only be installed in fixture root")
root.mkdir(parents=True, exist_ok=True)
destination = root / "rollback-configuration.json"
raw = json.dumps(config.payload, sort_keys=True, indent=2).encode() + b"\n"
with tempfile.NamedTemporaryFile(dir=root, prefix=".rollback.", delete=False) as handle:
handle.write(raw)
handle.flush()
temporary = Path(handle.name)
temporary.chmod(0o600)
temporary.replace(destination)
return destination
def verify(self, config: RollbackConfiguration, plan: ProtectionPlan) -> bool:
return (
config.previous_sha256 == snapshot_sha256(plan.previous)
and config.target_sha256 == snapshot_sha256(plan.target)
)
@dataclass
class ProtectionCompensationSaga:
adapter: object
store: MemoryProtectionStore
plan: ProtectionPlan
applied: bool = False
rolled_back: bool = False
def execute_fixture(self) -> ProtectionSnapshot:
result = self.adapter.apply_fixture(self.store, self.plan)
self.applied = True
return result
def compensate_fixture(self) -> ProtectionSnapshot:
if not self.applied:
return self.store.read(self.plan.branch)
result = self.adapter.rollback_fixture(self.store, self.plan)
self.rolled_back = True
return result
@@ -0,0 +1,69 @@
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import tempfile
def runtime_state(component: str, prior_invocations: int = 0) -> dict:
return {
"component": component,
"deployment_mode": "DORMANT_SHADOW_ONLY",
"enabled": False,
"active": False,
"production_routing": False,
"production_mutation": False,
"global_gitops_writes_enabled": False,
"prior_invocations": prior_invocations,
"restart_recovery": prior_invocations > 0,
}
def checkpoint_state(component: str, root: Path) -> dict:
root.mkdir(parents=True, exist_ok=True)
destination = root / f"{component}.json"
prior = 0
if destination.exists():
previous = json.loads(destination.read_text(encoding="utf-8"))
if previous.get("component") != component or previous.get("deployment_mode") != "DORMANT_SHADOW_ONLY":
raise RuntimeError("checkpoint identity mismatch")
prior = int(previous.get("invocations", 0))
state = runtime_state(component, prior)
checkpoint = {**state, "invocations": prior + 1}
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=root, delete=False) as handle:
json.dump(checkpoint, handle, sort_keys=True)
handle.write("\n")
handle.flush()
temporary = Path(handle.name)
temporary.replace(destination)
return state
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--component", required=True, choices=("owner-review-adapter", "machine-status-producer"))
parser.add_argument("--self-check", action="store_true")
parser.add_argument("--activate", action="store_true")
parser.add_argument("--checkpoint-dir", type=Path)
args = parser.parse_args(argv)
if args.activate:
parser.error("activation is not authorized by the prerequisites candidate")
if os.environ.get("AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED", "false").lower() != "false":
parser.error("candidate must remain disabled")
if os.environ.get("AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE", "false").lower() != "false":
parser.error("activation must remain inactive")
if os.environ.get("AIOS_GITOPS_WRITES_ENABLED", "false").lower() != "false":
parser.error("production mutation must remain disabled")
state = (
checkpoint_state(args.component, args.checkpoint_dir)
if args.checkpoint_dir is not None
else runtime_state(args.component)
)
print(json.dumps(state, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,54 @@
from __future__ import annotations
from dataclasses import dataclass
import json
from pathlib import Path
import tempfile
from .models import GovernanceState
class InvalidTransition(RuntimeError):
pass
_NEXT = {
GovernanceState.HUMAN_OWNER_APPROVAL_PENDING: GovernanceState.OWNER_APPROVAL_VERIFIED,
GovernanceState.OWNER_APPROVAL_VERIFIED: GovernanceState.MACHINE_GATES_VERIFIED,
GovernanceState.MACHINE_GATES_VERIFIED: GovernanceState.AUTO_MERGE_READY,
GovernanceState.AUTO_MERGE_READY: GovernanceState.OWNER_APPROVED_EXECUTION_READY,
}
@dataclass
class GovernanceStateMachine:
state: GovernanceState = GovernanceState.HUMAN_OWNER_APPROVAL_PENDING
def transition(self, target: GovernanceState) -> GovernanceState:
if _NEXT.get(self.state) != target:
raise InvalidTransition(f"{self.state.value}->{target.value}")
self.state = target
return self.state
def restore(self, checkpoint: GovernanceState) -> None:
if checkpoint not in GovernanceState:
raise InvalidTransition("unknown checkpoint")
self.state = checkpoint
def save_fixture(self, path: Path) -> None:
if not any(part.startswith("aios-governance-fixture-") for part in path.resolve().parts):
raise PermissionError("state checkpoint writes are fixture-only")
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
json.dump({"state": self.state.value}, handle, sort_keys=True)
handle.write("\n")
handle.flush()
temporary = Path(handle.name)
temporary.replace(path)
@classmethod
def load_fixture(cls, path: Path) -> "GovernanceStateMachine":
if not any(part.startswith("aios-governance-fixture-") for part in path.resolve().parts):
raise PermissionError("state checkpoint reads are fixture-only")
raw = json.loads(path.read_text(encoding="utf-8"))
return cls(GovernanceState(raw["state"]))
@@ -0,0 +1,38 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
class DormantStatusProducer(RuntimeError):
pass
@dataclass(frozen=True)
class StatusRecord:
context: str
state: str
head_oid: str
observed_at: str
class MachineStatusProducer:
def __init__(self, *, enabled: bool = False, fixture: bool = False) -> None:
self.enabled = enabled
self.fixture = fixture
def produce(self, context: str, state: str, head_oid: str) -> StatusRecord:
if not self.fixture:
raise DormantStatusProducer("LIVE_STATUS_PUBLICATION_DENIED")
if self.enabled:
raise DormantStatusProducer("DORMANT_CANDIDATE_MUST_NOT_BE_ENABLED")
if context not in {
"aios/machine-gates", "aios/secret-scan", "aios/evidence-closure"
}:
raise ValueError("unknown status context")
if state not in {"success", "failure"} or len(head_oid) != 40:
raise ValueError("invalid status record")
return StatusRecord(
context, state, head_oid,
datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
)
@@ -0,0 +1,69 @@
from __future__ import annotations
from dataclasses import dataclass
from .models import ApprovalContext, Review
class ValidationDenied(RuntimeError):
def __init__(self, code: str) -> None:
super().__init__(code)
self.code = code
@dataclass(frozen=True)
class ValidationResult:
owner_review: Review
machine_contexts: tuple[str, ...]
class SingleOwnerApprovalValidator:
"""Pure candidate validator. It has no Gitea or production write path."""
def validate(self, context: ApprovalContext) -> ValidationResult:
if context.now >= context.expires_at:
raise ValidationDenied("APPROVAL_EXPIRED")
if context.wip:
raise ValidationDenied("WIP_BLOCKED")
if context.requester == context.owner:
raise ValidationDenied("REQUESTER_SELF_APPROVAL")
if context.executor == context.owner:
raise ValidationDenied("EXECUTOR_SELF_APPROVAL")
approved = [review for review in context.reviews if review.state == "APPROVED"]
owner_reviews = [
review for review in approved
if review.reviewer == context.owner and review.reviewer_id == context.owner_id
]
if len(owner_reviews) != 1:
raise ValidationDenied("OWNER_APPROVAL_REQUIRED")
owner_review = owner_reviews[0]
if owner_review.stale:
raise ValidationDenied("STALE_OWNER_APPROVAL")
if owner_review.dismissed:
raise ValidationDenied("DISMISSED_OWNER_APPROVAL")
if owner_review.head_oid != context.current_head_oid:
raise ValidationDenied("OWNER_APPROVAL_NOT_LATEST_HEAD")
if owner_review.reviewed_at >= context.expires_at:
raise ValidationDenied("OWNER_APPROVAL_EXPIRED")
if any(
review.state == "APPROVED"
and review.reviewer != context.owner
and review.reviewer_id != context.owner_id
for review in context.reviews
):
# Non-owner approvals may exist for legacy mechanics, but never
# satisfy or replace the human authorization decision.
pass
gates = {gate.context: gate for gate in context.machine_gates}
if set(gates) != set(context.required_machine_contexts):
raise ValidationDenied("MACHINE_GATE_SET_MISMATCH")
for gate in gates.values():
if gate.state != "success":
raise ValidationDenied("MACHINE_GATE_FAILED")
if gate.head_oid != context.current_head_oid:
raise ValidationDenied("MACHINE_GATE_NOT_LATEST_HEAD")
if gate.observed_at >= context.expires_at:
raise ValidationDenied("MACHINE_GATE_EXPIRED")
return ValidationResult(owner_review, tuple(sorted(gates)))
@@ -0,0 +1,16 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "aios-single-owner-governance"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = []
[project.scripts]
aios-single-owner-governance = "aios_single_owner_governance.activation:main"
[tool.setuptools.packages.find]
where = ["."]
include = ["aios_single_owner_governance*"]
@@ -0,0 +1 @@
+51
View File
@@ -0,0 +1,51 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from aios_single_owner_governance.models import (
ApprovalContext,
MachineGate,
ProtectionSnapshot,
Review,
)
HEAD = "a" * 40
NOW = datetime(2026, 7, 28, 19, 0, tzinfo=timezone.utc)
CONTEXTS = ("aios/machine-gates", "aios/secret-scan", "aios/evidence-closure")
def owner_review(**changes) -> Review:
values = dict(
reviewer="owner", reviewer_id=2, state="APPROVED", head_oid=HEAD,
reviewed_at=NOW - timedelta(minutes=1), stale=False, dismissed=False,
)
values.update(changes)
return Review(**values)
def gates(*, state="success", head_oid=HEAD):
return tuple(MachineGate(name, state, head_oid, NOW) for name in CONTEXTS)
def context(**changes) -> ApprovalContext:
values = dict(
current_head_oid=HEAD, requester="agent", executor="merge-executor",
owner="owner", owner_id=2, reviews=(owner_review(),), machine_gates=gates(),
required_machine_contexts=CONTEXTS, expires_at=NOW + timedelta(hours=1),
wip=False, now=NOW,
)
values.update(changes)
return ApprovalContext(**values)
def legacy_protection(version=1) -> ProtectionSnapshot:
return ProtectionSnapshot(
"governance/approvals", 2, ("codex", "gpt"), ("codex", "gpt"),
True, False, (), False, False, False, version,
)
def single_owner_protection(version=2) -> ProtectionSnapshot:
return ProtectionSnapshot(
"governance/approvals", 1, ("owner",), ("owner",), True, True,
CONTEXTS, False, False, False, version,
)
@@ -0,0 +1,72 @@
from __future__ import annotations
from pathlib import Path
import unittest
from aios_single_owner_governance.fixture import DisposableGiteaFixture
from aios_single_owner_governance.protection import (
BranchProtectionCASMigrationAdapter,
DormantViolation,
snapshot_sha256,
)
from aios_single_owner_governance.policy import DormantOperationGuard, OperationDenied
from .helpers import legacy_protection, single_owner_protection
class LiveStore:
fixture = False
def read(self, branch):
return legacy_protection()
def compare_and_swap(self, branch, expected_sha256, target):
raise AssertionError("live compare_and_swap must never be invoked")
class DormantInvariantTests(unittest.TestCase):
def test_live_protection_apply_and_rollback_are_denied(self):
adapter = BranchProtectionCASMigrationAdapter()
old, new = legacy_protection(), single_owner_protection()
plan = adapter.plan(old, new, snapshot_sha256(old))
with self.assertRaises(DormantViolation):
adapter.apply_fixture(LiveStore(), plan)
with self.assertRaises(DormantViolation):
adapter.rollback_fixture(LiveStore(), plan)
def test_unsafe_policy_cannot_be_planned(self):
adapter = BranchProtectionCASMigrationAdapter()
old = legacy_protection()
for unsafe in (
single_owner_protection().__class__(**{**single_owner_protection().as_dict(), "approval_whitelist": ("owner",), "merge_whitelist": ("owner",), "status_contexts": tuple(single_owner_protection().status_contexts), "direct_push": True}),
single_owner_protection().__class__(**{**single_owner_protection().as_dict(), "approval_whitelist": ("owner",), "merge_whitelist": ("owner",), "status_contexts": tuple(single_owner_protection().status_contexts), "force_push": True}),
single_owner_protection().__class__(**{**single_owner_protection().as_dict(), "approval_whitelist": ("owner",), "merge_whitelist": ("owner",), "status_contexts": tuple(single_owner_protection().status_contexts), "branch_deletion": True}),
):
with self.assertRaises(DormantViolation):
adapter.plan(old, unsafe, snapshot_sha256(old))
def test_disposable_fixture_is_destroyed(self):
fixture = DisposableGiteaFixture(legacy_protection())
root = fixture.root
with fixture:
self.assertTrue(root.exists())
self.assertTrue(fixture.destroyed)
self.assertFalse(root.exists())
def test_permanent_and_pre_activation_writes_are_denied(self):
guard = DormantOperationGuard()
for operation in (
"protected_branch.direct_push",
"governance.direct_push",
"tag.direct_write",
"git.force_push",
"production.without_owner",
"branch_protection.live_update",
"single_owner_policy.activate",
):
with self.subTest(operation=operation), self.assertRaises(OperationDenied):
guard.authorize(operation)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,113 @@
from __future__ import annotations
import contextlib
import io
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
from aios_single_owner_governance.evidence import (
ActivationEvidenceClosureBuilder,
ApprovalMirrorStatusVerifier,
)
from aios_single_owner_governance.service import main, runtime_state
from aios_single_owner_governance.status import (
DormantStatusProducer,
MachineStatusProducer,
)
from .helpers import HEAD
class EvidenceStatusServiceTests(unittest.TestCase):
def test_complete_evidence_set_builds_canonical_closure(self):
evidence = {
"activation_prerequisites_commit_oid": "a" * 40,
"activation_prerequisites_tree_oid": "b" * 40,
"artifact_sha256": "c" * 64,
"rollback_artifact_sha256": "d" * 64,
"validator_sha256": "e" * 64,
"state_machine_sha256": "f" * 64,
"branch_protection_adapter_sha256": "0" * 64,
"rollback_configuration_sha256": "1" * 64,
"test_evidence_sha256": "2" * 64,
"deployment_manifest_sha256": "3" * 64,
"dormant_unit_manifest_sha256": "4" * 64,
"dormant_deployment_evidence_sha256": "5" * 64,
}
closure = ActivationEvidenceClosureBuilder().build(evidence)
self.assertEqual(len(closure.closure_sha256), 64)
with self.assertRaises(ValueError):
ActivationEvidenceClosureBuilder().build({})
def test_mirror_verifier_requires_all_four_refs_and_fsck(self):
verifier = ApprovalMirrorStatusVerifier()
self.assertTrue(verifier.verify(
HEAD, gitea_oid=HEAD, quarantine_oid=HEAD,
authoritative_oid=HEAD, controller_visible_oid=HEAD,
fsck_connectivity="PASS",
))
self.assertFalse(verifier.verify(
HEAD, gitea_oid=HEAD, quarantine_oid=HEAD,
authoritative_oid="b" * 40, controller_visible_oid=HEAD,
fsck_connectivity="PASS",
))
def test_machine_status_is_fixture_only_and_dormant(self):
with self.assertRaises(DormantStatusProducer):
MachineStatusProducer().produce("aios/machine-gates", "success", HEAD)
with self.assertRaises(DormantStatusProducer):
MachineStatusProducer(enabled=True, fixture=True).produce(
"aios/machine-gates", "success", HEAD
)
record = MachineStatusProducer(fixture=True).produce(
"aios/evidence-closure", "success", HEAD
)
self.assertEqual(record.head_oid, HEAD)
def test_service_self_check_is_dormant_and_activate_is_rejected(self):
output = io.StringIO()
with contextlib.redirect_stdout(output):
self.assertEqual(main(["--component", "owner-review-adapter", "--self-check"]), 0)
state = json.loads(output.getvalue())
self.assertFalse(state["active"])
self.assertFalse(state["production_mutation"])
with contextlib.redirect_stderr(io.StringIO()):
with self.assertRaises(SystemExit):
main(["--component", "owner-review-adapter", "--activate"])
def test_write_flags_cannot_be_enabled(self):
for name in (
"AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED",
"AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE",
"AIOS_GITOPS_WRITES_ENABLED",
):
with self.subTest(name=name), patch.dict(os.environ, {name: "true"}, clear=False):
with contextlib.redirect_stderr(io.StringIO()):
with self.assertRaises(SystemExit):
main(["--component", "machine-status-producer", "--self-check"])
def test_service_entrypoint_restart_recovery_across_processes(self):
with tempfile.TemporaryDirectory(prefix="aios-governance-fixture-") as root:
command = [
sys.executable, "-m", "aios_single_owner_governance.service",
"--component", "owner-review-adapter", "--self-check",
"--checkpoint-dir", root,
]
first = subprocess.run(command, text=True, capture_output=True, check=True)
second = subprocess.run(command, text=True, capture_output=True, check=True)
first_state = json.loads(first.stdout)
second_state = json.loads(second.stdout)
self.assertFalse(first_state["restart_recovery"])
self.assertTrue(second_state["restart_recovery"])
self.assertEqual(second_state["prior_invocations"], 1)
self.assertFalse(second_state["production_mutation"])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,476 @@
from __future__ import annotations
import json
from pathlib import Path
import sqlite3
import subprocess
import hashlib
import os
from datetime import datetime, timedelta, timezone
import pytest
from aios_single_owner_governance.activation import (
ActivationEngine,
InjectedCrash,
validate_bindings,
)
import aios_single_owner_governance.validator as validator_module
import aios_single_owner_governance.state_machine as state_machine_module
from aios_single_owner_governance.gitea_protected_branch import (
CASConflict,
FileProtectedBranchAdapter,
SQLiteProtectedBranchAdapter,
UnsafePolicy,
calculate_snapshot_sha256,
validate_target,
)
from aios_single_owner_governance.journal import sha256_json
MACHINE_CONTEXTS = {
"aios/validator",
"aios/state-machine",
"aios/branch-protection",
"aios/secret-scan",
"aios/evidence-closure",
}
def write_json(path: Path, value: object) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value, sort_keys=True) + "\n", encoding="utf-8")
def legacy_row() -> dict:
return {
"id": 7,
"repo_id": 1,
"branch_name": "governance/approvals",
"can_push": 0,
"enable_whitelist": 0,
"whitelist_user_i_ds": "null",
"whitelist_team_i_ds": "null",
"enable_merge_whitelist": 1,
"whitelist_deploy_keys": 0,
"merge_whitelist_user_i_ds": "[42,43]",
"merge_whitelist_team_i_ds": "null",
"enable_status_check": 0,
"status_check_contexts": "[]",
"enable_approvals_whitelist": 1,
"approvals_whitelist_user_i_ds": "[42,43]",
"approvals_whitelist_team_i_ds": "null",
"required_approvals": 2,
"block_on_rejected_reviews": 1,
"block_on_official_review_requests": 1,
"block_on_outdated_branch": 1,
"dismiss_stale_approvals": 1,
"ignore_stale_approvals": 0,
"require_signed_commits": 0,
"protected_file_patterns": "",
"unprotected_file_patterns": "",
"created_unix": 1,
"updated_unix": 10,
}
def owner_row() -> dict:
row = legacy_row()
row.update(
{
"approvals_whitelist_user_i_ds": "[2]",
"required_approvals": 1,
"enable_status_check": 1,
"status_check_contexts": json.dumps(sorted(MACHINE_CONTEXTS), separators=(",", ":")),
"updated_unix": 11,
}
)
return row
def git_output(git_dir: Path, *args: str, stdin: str | None = None) -> str:
return subprocess.run(
["git", f"--git-dir={git_dir}", *args], input=stdin, text=True,
capture_output=True, check=True,
env={
**__import__("os").environ,
"GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.invalid",
"GIT_COMMITTER_NAME": "Test", "GIT_COMMITTER_EMAIL": "test@example.invalid",
},
).stdout.strip()
def evidence_fixture(tmp_path: Path, execution_binding: dict) -> dict:
git_dir = tmp_path / "evidence.git"
subprocess.run(["git", "init", "--bare", str(git_dir)], capture_output=True, check=True)
closure_oid = git_output(git_dir, "hash-object", "-w", "--stdin", stdin="closure\n")
head_oid = git_output(git_dir, "hash-object", "-w", "--stdin", stdin="head\n")
expiry = datetime.now(timezone.utc) + timedelta(hours=1)
execution_binding = {**execution_binding, "evidence_closure_oid": closure_oid}
manifest = {
"approval_type": "SINGLE_OWNER_GOVERNANCE_LIVE_ACTIVATION_IMPLEMENTATION_AND_EXECUTION",
"expires_at": expiry.isoformat().replace("+00:00", "Z"),
"execution_binding": execution_binding,
"execution_binding_sha256": sha256_json(execution_binding),
}
manifest_raw = json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n"
manifest_blob = git_output(git_dir, "hash-object", "-w", "--stdin", stdin=manifest_raw)
leaf = git_output(git_dir, "mktree", stdin=f"100644 blob {manifest_blob}\tactivation.json\n")
approvals = git_output(git_dir, "mktree", stdin=f"040000 tree {leaf}\tsingle-owner\n")
tree = git_output(git_dir, "mktree", stdin=f"040000 tree {approvals}\tapprovals\n")
merge_oid = git_output(git_dir, "commit-tree", tree, stdin="approved\n")
git_output(git_dir, "update-ref", "refs/heads/governance/approvals", merge_oid)
db = tmp_path / "approval.db"
with sqlite3.connect(db) as connection:
connection.executescript(
"CREATE TABLE issue(id INTEGER,repo_id INTEGER,is_closed INTEGER);"
"CREATE TABLE pull_request(issue_id INTEGER,\"index\" INTEGER,base_branch TEXT,has_merged INTEGER,merged_commit_id TEXT);"
"CREATE TABLE user(id INTEGER,lower_name TEXT);"
"CREATE TABLE review(id INTEGER,type INTEGER,reviewer_id INTEGER,issue_id INTEGER,commit_id TEXT,stale INTEGER,dismissed INTEGER,created_unix INTEGER);"
"CREATE TABLE commit_status(id INTEGER,repo_id INTEGER,state TEXT,sha TEXT,context TEXT);"
)
connection.execute("INSERT INTO issue VALUES(10,1,1)")
connection.execute("INSERT INTO pull_request VALUES(10,31,'governance/approvals',1,?)", (merge_oid,))
connection.execute("INSERT INTO user VALUES(2,'owner')")
connection.execute(
"INSERT INTO review VALUES(52,1,2,10,?,0,0,?)",
(head_oid, int(datetime.now(timezone.utc).timestamp())),
)
for index, context in enumerate(sorted(MACHINE_CONTEXTS), 1):
connection.execute("INSERT INTO commit_status VALUES(?,1,'success',?,?)", (index, head_oid, context))
return {
"git_dir": git_dir, "merge_oid": merge_oid, "closure_oid": closure_oid,
"head_oid": head_oid, "db": db, "expiry": expiry.isoformat().replace("+00:00", "Z"),
"manifest_path": "approvals/single-owner/activation.json",
"manifest_blob": manifest_blob,
"manifest_sha256": hashlib.sha256(manifest_raw.encode()).hexdigest(),
"execution_binding": execution_binding,
}
def bindings(tmp_path: Path, adapter: FileProtectedBranchAdapter, **overrides) -> dict:
validator_path = Path(validator_module.__file__).resolve()
state_machine_path = Path(state_machine_module.__file__).resolve()
validator_sha = hashlib.sha256(validator_path.read_bytes()).hexdigest()
state_machine_sha = hashlib.sha256(state_machine_path.read_bytes()).hexdigest()
execution_binding = {
"task_id": "AIOS-LIVE-ACTIVATION-TEST",
"operation_id": "op-live-001",
"expected_validator_sha256": validator_sha,
"expected_state_machine_sha256": state_machine_sha,
"expected_branch_protection_sha256": adapter.calculate_snapshot_sha256(legacy_row()),
"expected_gitea_schema_sha256": "0" * 64,
"target_branch_protection_sha256": calculate_snapshot_sha256(owner_row()),
"backend": "file",
"repo_id": 1,
"branch": "governance/approvals",
"requester": "agent",
"executor": "activation-executor",
"forbidden_operations": {
"main": False, "governance_direct": False, "integration": False,
"tags": False, "force_push": False,
},
}
evidence = evidence_fixture(tmp_path, execution_binding)
value = {
"task_id": "AIOS-LIVE-ACTIVATION-TEST",
"operation_id": "op-live-001",
"expected_governance_oid": evidence["merge_oid"],
"observed_governance_oid": evidence["merge_oid"],
"expected_validator_sha256": validator_sha,
"observed_validator_sha256": validator_sha,
"expected_state_machine_sha256": state_machine_sha,
"observed_state_machine_sha256": state_machine_sha,
"validator_path": str(validator_path),
"state_machine_path": str(state_machine_path),
"governance_git_dir": str(evidence["git_dir"].resolve()),
"governance_ref": "refs/heads/governance/approvals",
"evidence_git_dir": str(evidence["git_dir"].resolve()),
"approval_db_path": str(evidence["db"].resolve()),
"expected_branch_protection_sha256": adapter.calculate_snapshot_sha256(legacy_row()),
"activation_approval_merge_oid": evidence["merge_oid"],
"approval_request_head_oid": evidence["head_oid"],
"activation_approval_pr_number": 31,
"activation_manifest_path": evidence["manifest_path"],
"activation_manifest_blob_oid": evidence["manifest_blob"],
"activation_manifest_sha256": evidence["manifest_sha256"],
"expires_at": evidence["expiry"],
"evidence_closure_oid": evidence["closure_oid"],
"observed_evidence_closure_oid": evidence["closure_oid"],
"owner_review_oid": 52,
"observed_owner_review_oid": 52,
"requester": "agent",
"executor": "activation-executor",
"wip": False,
"machine_checks": {
context: {"state": "success", "head_oid": evidence["head_oid"]}
for context in MACHINE_CONTEXTS
},
"forbidden_operations": {
"main": False,
"governance_direct": False,
"integration": False,
"tags": False,
"force_push": False,
},
"target_branch_protection": owner_row(),
"backend": "file",
"backend_path": str(adapter.path.resolve()),
"repo_id": 1,
"branch": "governance/approvals",
"expected_gitea_schema_sha256": "0" * 64,
}
value.update(overrides)
return value
def harness(tmp_path: Path):
backend = tmp_path / "protected.json"
write_json(backend, legacy_row())
adapter = FileProtectedBranchAdapter(backend)
engine = ActivationEngine(adapter, tmp_path / "state", tmp_path / "run" / "activation.lock")
request = bindings(tmp_path, adapter)
return backend, adapter, engine, request
def test_live_plan_success(tmp_path: Path) -> None:
_, _, engine, request = harness(tmp_path)
result = engine.plan(request)
assert result["status"] == "PLAN_READY"
assert len(result["plan_id"]) == 64
def test_live_activate_verify_status_success(tmp_path: Path) -> None:
_, adapter, engine, request = harness(tmp_path)
planned = engine.plan(request)
activated = engine.activate(request, planned["plan_id"])
assert activated["status"] == "ACTIVATED"
assert adapter.read_current() == owner_row()
assert engine.verify(request, request["operation_id"])["status"] == "VERIFIED"
assert engine.status(request["operation_id"])["transaction"]["phase"] == "COMMITTED"
assert (engine.journal.root / "current-snapshot.json").exists()
assert (engine.journal.root / "rollback-snapshot.json").exists()
def test_live_rollback_success(tmp_path: Path) -> None:
_, adapter, engine, request = harness(tmp_path)
planned = engine.plan(request)
engine.activate(request, planned["plan_id"])
assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ROLLED_BACK"
assert adapter.read_current() == legacy_row()
assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ALREADY_ROLLED_BACK"
def test_missing_approval_denied(tmp_path: Path) -> None:
_, _, engine, request = harness(tmp_path)
request.pop("owner_review_oid")
with pytest.raises(RuntimeError, match="MISSING_BINDINGS"):
engine.plan(request)
@pytest.mark.parametrize(
("field", "value", "message"),
[
("observed_owner_review_oid", 99, "WRONG_APPROVAL_OID_DENIED"),
("observed_governance_oid", "f" * 40, "WRONG_GOVERNANCE_OID_DENIED"),
("observed_evidence_closure_oid", "f" * 40, "WRONG_EVIDENCE_CLOSURE_DENIED"),
],
)
def test_wrong_binding_denied(tmp_path: Path, field: str, value: object, message: str) -> None:
_, _, engine, request = harness(tmp_path)
request[field] = value
with pytest.raises(RuntimeError, match=message):
engine.plan(request)
def test_cas_conflict_denied(tmp_path: Path) -> None:
backend, _, engine, request = harness(tmp_path)
planned = engine.plan(request)
changed = legacy_row()
changed["updated_unix"] = 12
write_json(backend, changed)
with pytest.raises(CASConflict):
engine.activate(request, planned["plan_id"])
@pytest.mark.parametrize("crash_at", ["before_apply", "during_apply", "after_apply_before_receipt", "after_commit_wal"])
def test_process_crash_restart_recovery(tmp_path: Path, crash_at: str) -> None:
_, adapter, engine, request = harness(tmp_path)
request["crash_at"] = crash_at
planned = engine.plan(request)
with pytest.raises(InjectedCrash):
engine.activate(request, planned["plan_id"])
request.pop("crash_at")
result = engine.activate(request, planned["plan_id"])
expected_status = "ALREADY_COMMITTED" if crash_at == "after_commit_wal" else "ACTIVATED"
assert result["status"] == expected_status
assert adapter.read_current() == owner_row()
assert engine.journal.receipt_path(request["operation_id"]).exists()
assert engine.journal.audit.has_event(request["operation_id"], "SINGLE_OWNER_ACTIVATION_COMMITTED")
def test_post_activation_verify_failure_rolls_back(tmp_path: Path, monkeypatch) -> None:
_, adapter, engine, request = harness(tmp_path)
planned = engine.plan(request)
monkeypatch.setattr(adapter, "verify_applied", lambda _: False)
with pytest.raises(RuntimeError, match="POST_ACTIVATION_VERIFY_FAILED"):
engine.activate(request, planned["plan_id"])
assert adapter.read_current() == legacy_row()
assert engine.journal.load(request["operation_id"])["phase"] == "ROLLED_BACK"
def test_operation_id_idempotency(tmp_path: Path) -> None:
_, _, engine, request = harness(tmp_path)
planned = engine.plan(request)
engine.activate(request, planned["plan_id"])
assert engine.activate(request, planned["plan_id"])["status"] == "ALREADY_COMMITTED"
@pytest.mark.parametrize("crash_at", ["after_rollback_cas", "after_rollback_wal"])
def test_rollback_crash_restart_repairs_closure(tmp_path: Path, crash_at: str) -> None:
_, adapter, engine, request = harness(tmp_path)
planned = engine.plan(request)
engine.activate(request, planned["plan_id"])
request["crash_at"] = crash_at
with pytest.raises(InjectedCrash):
engine.rollback(request, request["operation_id"], planned["plan_id"])
request.pop("crash_at")
assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ALREADY_ROLLED_BACK"
assert adapter.read_current() == legacy_row()
assert engine.journal.receipt_path(request["operation_id"]).exists()
assert engine.journal.audit.has_event(request["operation_id"], "ACTIVATION_ROLLED_BACK")
def test_operation_id_payload_mismatch_denied(tmp_path: Path) -> None:
_, _, engine, request = harness(tmp_path)
planned = engine.plan(request)
engine.activate(request, planned["plan_id"])
changed = dict(request)
approval_alias = tmp_path / "approval-alias.db"
os.link(request["approval_db_path"], approval_alias)
changed["approval_db_path"] = str(approval_alias)
with pytest.raises(RuntimeError, match="PAYLOAD_MISMATCH"):
engine.activate(changed, planned["plan_id"])
@pytest.mark.parametrize(
("field", "value"),
[("operation_id", "op-live-REPLAY"), ("task_id", "REPLAY-TASK")],
)
def test_approved_execution_binding_replay_denied(tmp_path: Path, field: str, value: str) -> None:
_, _, engine, request = harness(tmp_path)
request[field] = value
with pytest.raises(RuntimeError, match="APPROVED_EXECUTION_BINDING_MISMATCH"):
engine.plan(request)
def test_expiry_must_come_from_approved_manifest(tmp_path: Path) -> None:
_, _, engine, request = harness(tmp_path)
request["expires_at"] = (datetime.now(timezone.utc) + timedelta(hours=12)).isoformat().replace("+00:00", "Z")
with pytest.raises(RuntimeError, match="APPROVAL_EXPIRY_BINDING_MISMATCH"):
engine.plan(request)
def test_target_snapshot_must_match_approved_manifest(tmp_path: Path) -> None:
_, _, engine, request = harness(tmp_path)
request["target_branch_protection"] = {**owner_row(), "updated_unix": 99}
with pytest.raises(RuntimeError, match="APPROVED_EXECUTION_BINDING_MISMATCH"):
engine.plan(request)
def test_approval_database_must_be_backend_database(tmp_path: Path) -> None:
_, _, engine, request = harness(tmp_path)
other_db = tmp_path / "other.db"
other_db.write_bytes(Path(request["approval_db_path"]).read_bytes())
request.update(backend="sqlite", backend_path=str(other_db.resolve()))
with pytest.raises(RuntimeError, match="APPROVAL_DB_BACKEND_DB_IDENTITY_MISMATCH"):
engine.plan(request)
def test_audit_chain_continuity_and_credential_redaction(tmp_path: Path) -> None:
_, _, engine, request = harness(tmp_path)
request["api_token"] = "must-not-appear"
planned = engine.plan(request)
engine.activate(request, planned["plan_id"])
assert len(engine.journal.audit.verify()) == 64
raw = (engine.journal.root / "audit.jsonl").read_text(encoding="utf-8")
assert "must-not-appear" not in raw
assert "[REDACTED]" not in raw # only hashes/events are persisted to audit
@pytest.mark.parametrize(
"mutation",
[
lambda request: request.update(requester="owner"),
lambda request: request.update(executor="owner"),
lambda request: request.update(wip=True),
lambda request: request.update(machine_checks={"anything": {"state": "success", "head_oid": request["approval_request_head_oid"]}}),
lambda request: request.update(forbidden_operations={"main": True}),
lambda request: request.update(forbidden_operations={"tags": True}),
lambda request: request.update(forbidden_operations={"force_push": True}),
],
)
def test_negative_governance_acceptance(tmp_path: Path, mutation) -> None:
_, _, engine, request = harness(tmp_path)
mutation(request)
with pytest.raises(RuntimeError):
engine.plan(request)
@pytest.mark.parametrize(
"mutation",
[
lambda target: target.update(enable_approvals_whitelist=0),
lambda target: target.update(enable_merge_whitelist=0),
lambda target: target.update(status_check_contexts="[]"),
],
)
def test_unsafe_policy_enable_fields_denied(tmp_path: Path, mutation) -> None:
_, _, _, request = harness(tmp_path)
mutation(request["target_branch_protection"])
with pytest.raises(UnsafePolicy):
validate_target(legacy_row(), request["target_branch_protection"])
@pytest.mark.parametrize("invalid_owner", ["codex", "gpt"])
def test_owner_review_must_come_from_gitea_database(tmp_path: Path, invalid_owner: str) -> None:
_, _, engine, request = harness(tmp_path)
with sqlite3.connect(request["approval_db_path"]) as connection:
connection.execute("UPDATE user SET lower_name=? WHERE id=2", (invalid_owner,))
with pytest.raises(RuntimeError, match="OWNER_APPROVAL_INVALID"):
engine.plan(request)
def test_stale_owner_review_denied_from_gitea_database(tmp_path: Path) -> None:
_, _, engine, request = harness(tmp_path)
with sqlite3.connect(request["approval_db_path"]) as connection:
connection.execute("UPDATE review SET stale=1 WHERE id=52")
with pytest.raises(RuntimeError, match="OWNER_APPROVAL_INVALID"):
engine.plan(request)
def create_sqlite_fixture(path: Path) -> None:
row = legacy_row()
columns = []
for key, value in row.items():
kind = "INTEGER" if isinstance(value, int) else "TEXT"
columns.append(f"{key} {kind}")
with sqlite3.connect(path) as connection:
connection.execute(f"CREATE TABLE protected_branch ({','.join(columns)})")
connection.execute(
f"INSERT INTO protected_branch ({','.join(row)}) VALUES ({','.join('?' for _ in row)})",
list(row.values()),
)
def test_sqlite_full_row_cas_and_rollback(tmp_path: Path) -> None:
db = tmp_path / "gitea.db"
create_sqlite_fixture(db)
probe = SQLiteProtectedBranchAdapter(db, expected_schema_sha256="0" * 64)
adapter = SQLiteProtectedBranchAdapter(db, expected_schema_sha256=probe.schema_sha256())
previous = adapter.read_current()
applied = adapter.apply_atomic(calculate_snapshot_sha256(previous), owner_row())
assert applied == owner_row()
restored = adapter.rollback_atomic(calculate_snapshot_sha256(applied), previous)
assert restored == previous
@@ -0,0 +1,126 @@
from __future__ import annotations
from pathlib import Path
import tempfile
import unittest
from aios_single_owner_governance.fixture import DisposableGiteaFixture
from aios_single_owner_governance.lock import MaintenanceLock
from aios_single_owner_governance.models import GovernanceState
from aios_single_owner_governance.protection import (
BranchProtectionCASMigrationAdapter,
CASConflict,
MemoryProtectionStore,
snapshot_sha256,
)
from aios_single_owner_governance.rollback import (
ProtectionCompensationSaga,
RollbackConfigurationManager,
)
from aios_single_owner_governance.state_machine import GovernanceStateMachine, InvalidTransition
from .helpers import legacy_protection, single_owner_protection
from .helpers import context
from aios_single_owner_governance.validator import SingleOwnerApprovalValidator, ValidationDenied
class StateCasRollbackTests(unittest.TestCase):
def test_ordinary_owner_approved_merge_simulation(self):
result = SingleOwnerApprovalValidator().validate(context())
self.assertEqual(result.owner_review.reviewer, "owner")
machine = GovernanceStateMachine()
for target in (
GovernanceState.OWNER_APPROVAL_VERIFIED,
GovernanceState.MACHINE_GATES_VERIFIED,
GovernanceState.AUTO_MERGE_READY,
GovernanceState.OWNER_APPROVED_EXECUTION_READY,
):
machine.transition(target)
self.assertEqual(machine.state, GovernanceState.OWNER_APPROVED_EXECUTION_READY)
def test_validator_failure_is_stateless_and_recoverable(self):
validator = SingleOwnerApprovalValidator()
with self.assertRaises(ValidationDenied):
validator.validate(context(wip=True))
self.assertEqual(validator.validate(context()).owner_review.reviewer, "owner")
def test_state_machine_happy_path_and_restart_recovery(self):
machine = GovernanceStateMachine()
for target in (
GovernanceState.OWNER_APPROVAL_VERIFIED,
GovernanceState.MACHINE_GATES_VERIFIED,
GovernanceState.AUTO_MERGE_READY,
GovernanceState.OWNER_APPROVED_EXECUTION_READY,
):
machine.transition(target)
restarted = GovernanceStateMachine(machine.state)
self.assertEqual(restarted.state, GovernanceState.OWNER_APPROVED_EXECUTION_READY)
def test_state_machine_rollback_to_checkpoint(self):
machine = GovernanceStateMachine()
machine.transition(GovernanceState.OWNER_APPROVAL_VERIFIED)
checkpoint = machine.state
machine.transition(GovernanceState.MACHINE_GATES_VERIFIED)
machine.restore(checkpoint)
self.assertEqual(machine.state, GovernanceState.OWNER_APPROVAL_VERIFIED)
def test_service_restart_from_each_checkpoint(self):
with tempfile.TemporaryDirectory(prefix="aios-governance-fixture-") as root:
checkpoint = Path(root) / "state.json"
for state in GovernanceState:
with self.subTest(state=state):
GovernanceStateMachine(state).save_fixture(checkpoint)
restarted = GovernanceStateMachine.load_fixture(checkpoint)
self.assertEqual(restarted.state, state)
def test_invalid_transition_denied(self):
with self.assertRaises(InvalidTransition):
GovernanceStateMachine().transition(GovernanceState.AUTO_MERGE_READY)
def test_branch_protection_cas_success_and_rollback(self):
adapter = BranchProtectionCASMigrationAdapter()
old, new = legacy_protection(), single_owner_protection()
plan = adapter.plan(old, new, snapshot_sha256(old))
store = MemoryProtectionStore(old)
self.assertEqual(adapter.apply_fixture(store, plan), new)
self.assertEqual(adapter.rollback_fixture(store, plan), old)
def test_branch_protection_cas_conflict_fails_closed(self):
adapter = BranchProtectionCASMigrationAdapter()
old, changed, new = legacy_protection(), legacy_protection(7), single_owner_protection()
plan = adapter.plan(old, new, snapshot_sha256(old))
store = MemoryProtectionStore(changed)
with self.assertRaises(CASConflict):
adapter.apply_fixture(store, plan)
def test_partial_activation_recovery_saga(self):
adapter = BranchProtectionCASMigrationAdapter()
old, new = legacy_protection(), single_owner_protection()
plan = adapter.plan(old, new, snapshot_sha256(old))
saga = ProtectionCompensationSaga(adapter, MemoryProtectionStore(old), plan)
saga.execute_fixture()
self.assertEqual(saga.compensate_fixture(), old)
self.assertTrue(saga.applied and saga.rolled_back)
def test_rollback_configuration_fixture_install(self):
adapter = BranchProtectionCASMigrationAdapter()
old, new = legacy_protection(), single_owner_protection()
plan = adapter.plan(old, new, snapshot_sha256(old))
manager = RollbackConfigurationManager()
config = manager.build(plan)
with DisposableGiteaFixture(old) as fixture:
path = manager.install_fixture(fixture.root, config)
self.assertTrue(path.exists())
self.assertTrue(manager.verify(config, plan))
def test_maintenance_lock_fences_concurrent_holder(self):
with tempfile.TemporaryDirectory(prefix="aios-governance-fixture-") as root:
path = Path(root) / "maintenance.lock"
with MaintenanceLock(path):
with self.assertRaises(BlockingIOError):
with MaintenanceLock(path):
pass
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,45 @@
from __future__ import annotations
from pathlib import Path
import unittest
ROOT = Path(__file__).resolve().parents[2]
class SystemdContractTests(unittest.TestCase):
def test_units_are_static_dormant_and_credential_free(self):
units = sorted(
unit for unit in (ROOT / "deploy" / "systemd").glob("aios-governance-*.service")
if unit.name != "aios-governance-single-owner-activation.service"
)
self.assertEqual(len(units), 2)
for unit in units:
text = unit.read_text(encoding="utf-8")
with self.subTest(unit=unit.name):
self.assertIn("AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED=false", text)
self.assertIn("AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE=false", text)
self.assertIn("AIOS_GITOPS_WRITES_ENABLED=false", text)
self.assertIn("ProtectSystem=strict", text)
self.assertNotIn("EnvironmentFile=", text)
self.assertNotIn("LoadCredential=", text)
self.assertNotIn("WantedBy=", text)
self.assertNotIn("[Install]", text)
self.assertIn("--checkpoint-dir /var/lib/aios-gitops/runtime/single-owner-governance", text)
self.assertNotIn("/data/aios-gitops/production", text)
def test_activation_verifier_is_static_and_has_no_credential_or_install_hook(self):
unit = ROOT / "deploy" / "systemd" / "aios-governance-single-owner-activation.service"
text = unit.read_text(encoding="utf-8")
self.assertIn("AIOS_GITOPS_WRITES_ENABLED=false", text)
self.assertIn("--verify", text)
self.assertIn("ProtectSystem=strict", text)
self.assertNotIn("EnvironmentFile=", text)
self.assertNotIn("LoadCredential=", text)
self.assertNotIn("WantedBy=", text)
self.assertNotIn("[Install]", text)
self.assertNotIn("--activate", text)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,77 @@
from __future__ import annotations
from dataclasses import replace
from datetime import timedelta
import unittest
from aios_single_owner_governance.models import Review
from aios_single_owner_governance.validator import SingleOwnerApprovalValidator, ValidationDenied
from .helpers import HEAD, NOW, context, gates, owner_review
class ValidatorTests(unittest.TestCase):
def setUp(self):
self.validator = SingleOwnerApprovalValidator()
def denied(self, code, **changes):
with self.assertRaises(ValidationDenied) as caught:
self.validator.validate(context(**changes))
self.assertEqual(caught.exception.code, code)
def test_owner_current_head_and_machine_gates_pass(self):
result = self.validator.validate(context())
self.assertEqual(result.owner_review.reviewer, "owner")
self.assertEqual(len(result.machine_contexts), 3)
def test_requester_self_approval_denied(self):
self.denied("REQUESTER_SELF_APPROVAL", requester="owner")
def test_executor_self_approval_denied(self):
self.denied("EXECUTOR_SELF_APPROVAL", executor="owner")
def test_codex_as_owner_denied(self):
codex = owner_review(reviewer="codex", reviewer_id=42)
self.denied("OWNER_APPROVAL_REQUIRED", reviews=(codex,))
def test_gpt_as_owner_denied(self):
gpt = owner_review(reviewer="gpt", reviewer_id=43)
self.denied("OWNER_APPROVAL_REQUIRED", reviews=(gpt,))
def test_non_owner_approval_denied(self):
other = owner_review(reviewer="other", reviewer_id=77)
self.denied("OWNER_APPROVAL_REQUIRED", reviews=(other,))
def test_stale_owner_approval_denied(self):
self.denied("STALE_OWNER_APPROVAL", reviews=(owner_review(stale=True),))
def test_dismissed_owner_approval_denied(self):
self.denied("DISMISSED_OWNER_APPROVAL", reviews=(owner_review(dismissed=True),))
def test_old_head_owner_approval_denied(self):
self.denied("OWNER_APPROVAL_NOT_LATEST_HEAD", reviews=(owner_review(head_oid="b" * 40),))
def test_wip_denied(self):
self.denied("WIP_BLOCKED", wip=True)
def test_machine_gate_failure_denied(self):
self.denied("MACHINE_GATE_FAILED", machine_gates=gates(state="failure"))
def test_secret_scan_failure_denied(self):
failed = list(gates())
failed[1] = replace(failed[1], state="failure")
self.denied("MACHINE_GATE_FAILED", machine_gates=tuple(failed))
def test_machine_gate_wrong_head_denied(self):
self.denied("MACHINE_GATE_NOT_LATEST_HEAD", machine_gates=gates(head_oid="c" * 40))
def test_expired_request_denied(self):
self.denied("APPROVAL_EXPIRED", now=NOW + timedelta(hours=2))
def test_review_after_expiry_denied(self):
late = owner_review(reviewed_at=NOW + timedelta(hours=2))
self.denied("OWNER_APPROVAL_EXPIRED", reviews=(late,))
if __name__ == "__main__":
unittest.main()
@@ -1,228 +0,0 @@
import json
import tempfile
import threading
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import os
import sys
ROOT = os.getcwd()
sys.path.insert(0, str(Path(ROOT) / "active-runtime"))
from app.task_plane.workers.command_policy import CommandPolicy
from app.task_plane.workers.task_client import (
CredentialSource,
StageAClientConfig,
TaskClientAuthError,
TaskClientProtocolError,
TaskClientResponseTooLargeError,
TaskPlaneClient,
)
def _start_server(responses):
class Handler(BaseHTTPRequestHandler):
request_count = 0
seen = []
def do_POST(self):
index = Handler.request_count
Handler.request_count += 1
payload_length = int(self.headers.get("Content-Length", "0"))
payload = self.rfile.read(payload_length)
try:
payload_obj = json.loads(payload.decode("utf-8") or "{}")
except Exception:
payload_obj = {"raw": payload.decode("utf-8", errors="replace")}
Handler.seen.append((self.path, payload_obj))
status, response_obj = responses[min(index, len(responses) - 1)]
body = json.dumps(response_obj).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *_args, **_kwargs):
return
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
return server, thread, Handler
def _stop_server(server, thread):
server.shutdown()
server.server_close()
thread.join(timeout=2)
class CodexWorkerCommandPolicyTests(unittest.TestCase):
def test_rejects_revision_and_path_injection(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
policy = CommandPolicy(root, root / "results")
with self.assertRaises(ValueError):
policy.build("git-show", oid="HEAD~1", paths=["active-runtime/app/task_plane/api.py"])
with self.assertRaises(ValueError):
policy.build("pytest", oid="0" * 40, paths=["../outside"])
class CodexWorkerClientStageBTests(unittest.TestCase):
def test_client_retry_on_retryable_status(self):
server, thread, handler = _start_server([(503, {"error": "try again"}), (200, {"ok": True})])
try:
client = TaskPlaneClient(
StageAClientConfig(base_url=f"http://127.0.0.1:{server.server_port}"),
token="stage-b-token-" + "x" * 32,
)
body = client._post("/tests/retry", {"probe": "x"}, idempotency_key="retry")
self.assertEqual(body, {"ok": True})
self.assertEqual(handler.request_count, 2)
self.assertTrue(handler.seen[0][0].startswith("/tests/retry"))
finally:
_stop_server(server, thread)
def test_client_stops_retry_on_auth_error(self):
server, thread, handler = _start_server([(403, {"error": "denied"})])
try:
client = TaskPlaneClient(
StageAClientConfig(base_url=f"http://127.0.0.1:{server.server_port}"),
token="stage-b-token-" + "x" * 32,
)
with self.assertRaises(TaskClientAuthError):
client._post("/tests/auth", {"probe": "x"}, idempotency_key="auth")
self.assertEqual(handler.request_count, 1)
finally:
_stop_server(server, thread)
def test_client_rejects_oversize_request_and_response(self):
server, thread, _ = _start_server([(200, {"ok": True})])
try:
small = TaskPlaneClient(
StageAClientConfig(
base_url=f"http://127.0.0.1:{server.server_port}",
max_request_bytes=8,
),
token="stage-b-token-" + "x" * 32,
)
with self.assertRaises(TaskClientResponseTooLargeError):
small._post("/tests/request", {"probe": "too-large"}, idempotency_key="req")
finally:
_stop_server(server, thread)
oversize = {"blob": "x" * 1_100_000}
server, thread, _ = _start_server([(200, oversize)])
try:
client = TaskPlaneClient(
StageAClientConfig(base_url=f"http://127.0.0.1:{server.server_port}"),
token="stage-b-token-" + "x" * 32,
)
with self.assertRaises(TaskClientResponseTooLargeError):
client._post("/tests/response", {"probe": "ok"}, idempotency_key="resp")
finally:
_stop_server(server, thread)
def test_client_credential_source_symlink_and_mode(self):
with tempfile.TemporaryDirectory() as directory:
directory = Path(directory)
token_file = directory / "token"
token_file.write_text("x" * 32, encoding="utf-8")
token_file.chmod(0o600)
import grp
import pwd
owner = pwd.getpwuid(os.getuid()).pw_name
group = grp.getgrgid(os.getgid()).gr_name
source = CredentialSource(
token_file,
token_owner=owner,
token_group=group,
expected_mode=0o600,
)
self.assertEqual(source.read_token(), "x" * 32)
symlink = directory / "token-link"
symlink.symlink_to(token_file)
with self.assertRaises(TaskClientAuthError):
CredentialSource(
symlink,
token_owner=owner,
token_group=group,
expected_mode=0o600,
).read_token()
bad_mode = directory / "bad-mode"
bad_mode.write_text("value", encoding="utf-8")
bad_mode.chmod(0o644)
with self.assertRaises(TaskClientAuthError):
CredentialSource(
bad_mode,
token_owner=owner,
token_group=group,
expected_mode=0o600,
).read_token()
def test_client_rejects_missing_claim_id_and_reports_heartbeat_defaults(self):
client = TaskPlaneClient(StageAClientConfig(), token="stage-b-token-" + "x" * 32)
with self.assertRaises(TaskClientProtocolError):
client.progress("task-id", claim_id="", progress=10, stage="RUN")
with self.assertRaises(TaskClientProtocolError):
client.result(
"task-id",
claim_id="",
status="COMPLETED",
outputs=[],
evidence=[],
started_at="",
completed_at="",
)
with self.assertRaises(TaskClientProtocolError):
client.fail("task-id", claim_id="")
def test_heartbeat_request_uses_default_lease(self):
captured = {}
def handler_factory(responses):
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
payload_length = int(self.headers.get("Content-Length", "0"))
captured["path"] = self.path
captured["payload"] = json.loads(self.rfile.read(payload_length).decode("utf-8") or "{}")
status, response_obj = responses[min(responses_index[0], len(responses) - 1)]
responses_index[0] += 1
response = json.dumps(response_obj).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(response)))
self.end_headers()
self.wfile.write(response)
def log_message(self, *_args, **_kwargs):
return
return Handler
responses_index = [0]
server = ThreadingHTTPServer(("127.0.0.1", 0), handler_factory([(200, {"claim_id": "x"}), (200, {"ok": True})]))
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
client = TaskPlaneClient(
StageAClientConfig(base_url=f"http://127.0.0.1:{server.server_port}"),
token="stage-b-token-" + "x" * 32,
)
client.claim_heartbeat("task-id", claim_id="claim-id")
self.assertEqual(captured["payload"], {"claim_id": "claim-id", "lease_seconds": 120})
self.assertTrue(captured["path"].endswith("/tasks/task-id/heartbeat"))
captured["payload"] = None
client.worker_heartbeat("worker-1")
self.assertEqual(captured["payload"], {})
finally:
_stop_server(server, thread)
@@ -1,90 +0,0 @@
import os
import stat
import tempfile
import unittest
from pathlib import Path
ROOT = os.getcwd()
import sys
sys.path.insert(0, str(Path(ROOT) / "active-runtime"))
from app.task_plane.workers.codex_review_test import CodexReviewTaskPolicy
from app.task_plane.workers.result_attestation import TaskResultEnvelope, TrustPolicy, task_result_envelope
from app.task_plane.workers.workspace import TaskWorkspace
class CodexWorkerWorkspaceRecoveryTests(unittest.TestCase):
def test_workspace_permissions_and_cleanup(self):
with tempfile.TemporaryDirectory() as directory:
root_path = Path(directory) / "task-root"
with TaskWorkspace.create(root_path, task_id="task-01", claim_id="claim-01") as workspace:
for path in (
workspace.layout.workspace_root,
workspace.layout.repo_root,
workspace.layout.result_root,
workspace.layout.home,
workspace.layout.tmp,
):
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o700)
result_path = workspace.write_result(name="result.json", payload="{}")
self.assertTrue(result_path.exists())
self.assertEqual(stat.S_IMODE(result_path.stat().st_mode), 0o600)
self.assertFalse(root_path.exists())
def test_result_envelope_validates_required_contract_fields(self):
envelope = task_result_envelope(
task_id="task-01",
claim_id="claim-01",
worker_identity="spiffe://aios/agent/codex/review-01",
status="COMPLETED",
outputs=[],
evidence=[],
started_at="2026-07-24T00:00:00Z",
completed_at="2026-07-24T00:00:01Z",
)
payload = envelope.to_dict()
self.assertEqual(payload["production_trust"], False)
self.assertEqual(payload["trust_level"], TrustPolicy.SHADOW_ATTESTED.value)
self.assertEqual(payload["task_id"], "task-01")
invalid = TaskResultEnvelope(
task_id="",
claim_id="claim",
worker_identity="worker",
status="FAILED",
outputs=[],
evidence=[],
started_at="",
completed_at="",
production_trust=False,
)
with self.assertRaises(ValueError):
invalid.validate()
def test_shadow_policy_enforces_concurrency_and_timeboxes(self):
policy = CodexReviewTaskPolicy(worker_id="worker-01", identity="identity")
self.assertEqual(policy.max_concurrency, 1)
self.assertEqual(policy.heartbeat_interval_seconds, 15)
self.assertEqual(policy.claim_lease_seconds, 120)
valid_task = {
"risk": "low",
"metadata": {"production_side_effects": False},
}
policy.assert_shadow_only_task(valid_task)
with self.assertRaises(Exception):
policy.assert_shadow_only_task(
{
"risk": "critical",
"metadata": {"production_side_effects": False},
}
)
with self.assertRaises(Exception):
CodexReviewTaskPolicy(
worker_id="worker-01",
identity="identity",
max_concurrency=2,
).assert_shadow_only_task(valid_task)
@@ -1,83 +0,0 @@
import os
import tempfile
import unittest
from pathlib import Path
ROOT = os.getcwd()
import sys
sys.path.insert(0, str(Path(ROOT) / "active-runtime"))
from app.task_plane.workers.command_policy import CommandPolicy
from app.task_plane.workers.runner import CommandDisallowedError, SafeCommandRunner, CommandPolicyError
from app.task_plane.workers.workspace import TaskWorkspace
class CodexWorkerRunnerSecurityTests(unittest.TestCase):
def test_runner_only_allows_fixed_command_set(self):
with tempfile.TemporaryDirectory() as directory:
with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-fixed", claim_id="claim-fixed") as workspace:
runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
with self.assertRaises(CommandDisallowedError):
runner.run("shell", oid="0" * 40)
with self.assertRaises(CommandDisallowedError):
runner.run("bash", oid="0" * 40, paths=["tests/task_plane/public_interface"])
def test_runner_rejects_shell_like_and_eval_inputs(self):
with tempfile.TemporaryDirectory() as directory:
with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-shell", claim_id="claim-shell") as workspace:
runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
CommandPolicy(workspace.layout.repo_root, workspace.layout.result_root).build(
"pytest",
oid="0" * 40,
paths=["tests/task_plane/public_interface"],
)
with self.assertRaises(CommandDisallowedError):
runner.run("pytest", oid="0" * 40, paths=["tests/task_plane/public_interface;touch /tmp/x"])
with self.assertRaises(CommandDisallowedError):
runner.run("pytest", oid="0" * 40, paths=["tests/task_plane/public_interface", "eval"])
with self.assertRaises(CommandDisallowedError):
runner.run("pytest", oid="0" * 40, paths=["tests/task_plane/public_interface", "bash"])
def test_runner_rejects_revision_expression_and_symlink_paths(self):
with tempfile.TemporaryDirectory() as directory:
with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-revision", claim_id="claim-revision") as workspace:
runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
policy_root = workspace.layout.repo_root
(policy_root / "tests" / "task_plane" / "public_interface").mkdir(parents=True, exist_ok=True)
with self.assertRaises(ValueError):
runner.run("git-show", oid="HEAD~1", paths=["active-runtime/app/task_plane/worker.py"])
link_target = policy_root / "tests" / "task_plane" / "public_interface" / "link"
link_target.symlink_to(Path("/etc/passwd"), target_is_directory=False)
with self.assertRaises(CommandPolicyError):
runner.run("git-show", oid="0" * 40, paths=["tests/task_plane/public_interface/link"])
def test_runner_rejects_network_tokens_and_output_escape(self):
with tempfile.TemporaryDirectory() as directory:
with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-net", claim_id="claim-net") as workspace:
runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
with self.assertRaises(CommandDisallowedError):
runner.run("git-show", oid="0" * 40, paths=["https://bad.example/attack"])
with self.assertRaises(CommandPolicyError):
runner.run(
"acceptance",
oid="0" * 40,
paths=["active-runtime"],
output=workspace.layout.home.parent / "escape.json",
)
def test_runner_accepts_nested_allowlisted_paths_and_rejects_secret_env(self):
with tempfile.TemporaryDirectory() as directory:
with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-safe", claim_id="claim-safe") as workspace:
runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
command = runner.build(
"pytest",
oid="0" * 40,
paths=["tests/task_plane/public_interface"],
)
self.assertEqual(command[-1], "tests/task_plane/public_interface")
with self.assertRaises(Exception):
runner.run("diff-check", oid="0" * 40, env={"API_TOKEN": "secret"})
@@ -0,0 +1,417 @@
from __future__ import annotations
import json
import os
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest import mock
import aios_isolated_writer_policy as writer_policy
import isolated_writer_path_policy as path_policy
from isolated_writer_authz import AuthorizationError, _validate_grant
TARGET = "deploy/systemd/aios-codex-review-test-worker.service"
BRANCH = "patch/codex-review-test-worker-20260724"
AUDIENCE = "aios-codex-review-worker-writer.service"
VALID_UNIT = """\
[Unit]
Description=AIOS Codex Review/Test Worker
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=aiman
Group=gitops
WorkingDirectory=/usr/libexec/aios-codex-review-test-worker/current
Environment=AIOS_CODEX_WORKER_MODE=shadow
Environment=AIOS_TASK_PLANE_EXECUTION_ENABLED=false
Environment=PYTHONDONTWRITEBYTECODE=1
ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test
Restart=on-failure
RestartSec=2
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
LockPersonality=true
RestrictSUIDSGID=true
RestrictRealtime=true
MemoryDenyWriteExecute=true
RuntimeDirectory=aios-codex-review-test-worker
StateDirectory=aios-codex-review-test-worker
CacheDirectory=aios-codex-review-test-worker
ReadOnlyPaths=/data/aios-gitops/repo.git
ReadOnlyPaths=/usr/libexec/aios-codex-review-test-worker
ReadWritePaths=/run/aios-codex-review-test-worker
ReadWritePaths=/var/lib/aios-codex-review-test-worker
ReadWritePaths=/var/cache/aios-codex-review-test-worker
UMask=0077
[Install]
WantedBy=multi-user.target
"""
def document(*, paths=None, **updates):
value = {
"schema_version": 1,
"policy_id": "codex-review-worker-path-policy-v1",
"audience": AUDIENCE,
"repository_identity": "aios/aios",
"allowed_branch": BRANCH,
"allowed_operations": {
"commit_file.atomic": {
"exact_paths": paths or [TARGET],
"max_content_bytes": 32768,
"content_profile": "systemd-codex-review-worker-v1",
}
},
}
value.update(updates)
return value
class ExactPathPolicyTests(unittest.TestCase):
def setUp(self):
self.saved = dict(os.environ)
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
self.policy_dir = self.root / "policies"
self.policy_dir.mkdir(mode=0o750)
self.policy_file = self.policy_dir / "writer.json"
self.workspace = self.root / "workspace"
self.workspace.mkdir(mode=0o700)
os.environ.update(
{
"AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT": "required",
"AIOS_GITOPS_SERVICE_AUDIENCE": AUDIENCE,
"AIOS_GITOPS_REPOSITORY_IDENTITY": "aios/aios",
"AIOS_GITOPS_ALLOWED_BRANCH": BRANCH,
}
)
def tearDown(self):
path_policy.clear_path_policy_cache()
os.environ.clear()
os.environ.update(self.saved)
self.temp.cleanup()
def write_policy(self, value=None):
payload = document() if value is None else value
raw = (
json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+ "\n"
)
self.policy_file.write_text(raw, encoding="utf-8")
self.policy_file.chmod(0o640)
return path_policy.load_path_policy(
self.policy_file,
expected_owner_uid=os.getuid(),
expected_group_gid=os.getgid(),
)
def configured(self, loaded):
return mock.patch.object(
path_policy,
"configured_path_policy",
return_value=loaded,
)
def test_exact_systemd_path_and_exact_token_path_are_allowed(self):
loaded = self.write_policy()
with self.configured(loaded):
self.assertEqual(TARGET, writer_policy.validate_path(TARGET, self.workspace))
writer_policy.validate_content_for_path(TARGET, VALID_UNIT)
def test_token_path_cannot_expand_service_policy(self):
loaded = self.write_policy(document(paths=["deploy/systemd/other.service"]))
with self.configured(loaded), self.assertRaisesRegex(
PermissionError, "isolated_writer_service_path_rejected"
):
writer_policy.validate_path(TARGET, self.workspace)
def test_service_policy_cannot_expand_token_path(self):
now = datetime.now(timezone.utc)
grant = {
"schema_version": 2,
"audience": AUDIENCE,
"source_cidrs": ["127.0.0.1/32"],
"created_at": (now - timedelta(seconds=5)).isoformat(),
"not_before": (now - timedelta(seconds=2)).isoformat(),
"expires_at": (now + timedelta(minutes=5)).isoformat(),
"revoked": False,
"allowed_operations": ["commit_file.atomic"],
"allowed_branch": BRANCH,
"allowed_paths": ["deploy/systemd/other.service"],
"approved_base_oid": "a" * 40,
}
with mock.patch(
"isolated_writer_authz._git_is_ancestor", return_value=True
), self.assertRaisesRegex(AuthorizationError, "writer_token_path_rejected"):
_validate_grant(
grant,
audience=AUDIENCE,
source_ip="127.0.0.1",
operation="commit_file.atomic",
branch=BRANCH,
path=TARGET,
workspace=self.workspace,
current_time=now,
)
def test_other_service_file_is_rejected(self):
loaded = self.write_policy()
with self.configured(loaded), self.assertRaisesRegex(
PermissionError, "isolated_writer_service_path_rejected"
):
writer_policy.validate_path(
"deploy/systemd/other.service", self.workspace
)
def test_path_traversal_is_rejected(self):
for value in ("deploy/systemd/../x", "../deploy/systemd/x"):
with self.subTest(value=value), self.assertRaises(PermissionError):
writer_policy.validate_path(value, self.workspace)
def test_url_encoded_traversal_is_rejected(self):
with self.assertRaises(PermissionError):
writer_policy.validate_path(
"deploy/systemd/%2e%2e/x", self.workspace
)
def test_symlink_policy_file_is_rejected(self):
self.write_policy()
link = self.policy_dir / "link.json"
link.symlink_to(self.policy_file)
with self.assertRaisesRegex(
path_policy.PathPolicyError, "isolated_writer_path_policy_symlink_rejected"
):
path_policy.load_path_policy(
link,
expected_owner_uid=os.getuid(),
expected_group_gid=os.getgid(),
)
def test_world_writable_policy_file_is_rejected(self):
self.write_policy()
self.policy_file.chmod(0o646)
with self.assertRaisesRegex(
path_policy.PathPolicyError,
"isolated_writer_path_policy_permissions_rejected",
):
path_policy.load_path_policy(
self.policy_file,
expected_owner_uid=os.getuid(),
expected_group_gid=os.getgid(),
)
def test_unknown_policy_field_is_rejected(self):
value = document()
value["unexpected"] = True
with self.assertRaisesRegex(
path_policy.PathPolicyError,
"isolated_writer_path_policy_fields_rejected",
):
self.write_policy(value)
def test_audience_mismatch_is_rejected(self):
loaded = self.write_policy()
os.environ["AIOS_GITOPS_PATH_POLICY_FILE"] = str(self.policy_file)
os.environ["AIOS_GITOPS_SERVICE_AUDIENCE"] = "other.service"
path_policy.clear_path_policy_cache()
with mock.patch.object(
path_policy, "load_path_policy", return_value=loaded
), self.assertRaisesRegex(
path_policy.PathPolicyError,
"isolated_writer_path_policy_audience_rejected",
):
path_policy.configured_path_policy()
def test_branch_mismatch_is_rejected(self):
loaded = self.write_policy()
os.environ["AIOS_GITOPS_PATH_POLICY_FILE"] = str(self.policy_file)
os.environ["AIOS_GITOPS_ALLOWED_BRANCH"] = "patch/other"
path_policy.clear_path_policy_cache()
with mock.patch.object(
path_policy, "load_path_policy", return_value=loaded
), self.assertRaisesRegex(
path_policy.PathPolicyError,
"isolated_writer_path_policy_branch_rejected",
):
path_policy.configured_path_policy()
def assert_profile_rejected(self, content):
loaded = self.write_policy()
with self.configured(loaded), self.assertRaisesRegex(
PermissionError, "isolated_writer_systemd_profile_rejected"
):
writer_policy.validate_content_for_path(TARGET, content)
def test_disallowed_execution_and_credential_directives_are_rejected(self):
directives = (
"ExecStartPre=/usr/bin/id",
"ExecStartPost=/usr/bin/id",
"ExecReload=/usr/bin/id",
"ExecStop=/usr/bin/id",
"ExecStopPost=/usr/bin/id",
"ExecCondition=/usr/bin/true",
"EnvironmentFile=/run/worker.env",
"PassEnvironment=PATH",
"LoadCredential=worker:/run/worker",
"SetCredential=worker:value",
"ImportCredential=worker",
"StandardInput=socket",
"StandardOutput=socket",
)
for directive in directives:
with self.subTest(directive=directive):
self.assert_profile_rejected(
VALID_UNIT.replace(
"Restart=on-failure",
f"Restart=on-failure\n{directive}",
)
)
def test_disallowed_namespace_and_capability_directives_are_rejected(self):
directives = (
"RootDirectory=/srv/worker",
"RootImage=/srv/worker.raw",
"ReadWritePaths=/",
"BindPaths=/etc:/etc",
"BindReadOnlyPaths=/etc:/etc",
"TemporaryFileSystem=/etc",
"CapabilityBoundingSet=CAP_SYS_ADMIN",
"AmbientCapabilities=CAP_SYS_ADMIN",
"DeviceAllow=/dev/sda rw",
"SupplementaryGroups=docker",
"Sockets=unsafe.socket",
)
for directive in directives:
with self.subTest(directive=directive):
self.assert_profile_rejected(
VALID_UNIT.replace(
"Restart=on-failure",
f"Restart=on-failure\n{directive}",
)
)
def test_duplicate_single_directives_and_sections_are_rejected(self):
bad_values = (
VALID_UNIT.replace("User=aiman", "User=aiman\nUser=aiman"),
VALID_UNIT.replace(
"ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test",
"ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test\n"
"ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test",
),
VALID_UNIT + "\n[Service]\nType=simple\n",
)
for value in bad_values:
with self.subTest(value=value):
self.assert_profile_rejected(value)
def test_empty_reset_continuation_crlf_nul_and_long_lines_are_rejected(self):
bad_values = (
VALID_UNIT.replace(
"ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test",
"ExecStart=",
),
VALID_UNIT.replace(
"Restart=on-failure",
"Restart=on-failure\n continuation-injection",
),
VALID_UNIT.replace("\n", "\r\n"),
VALID_UNIT.replace("Restart=on-failure", "Restart=on-failure\x00"),
VALID_UNIT.replace(
"Description=AIOS Codex Review/Test Worker",
"Description=" + ("A" * 4097),
),
)
for value in bad_values:
with self.subTest(kind=repr(value[:80])):
self.assert_profile_rejected(value)
def test_unknown_environment_and_production_execution_are_rejected(self):
bad_values = (
VALID_UNIT.replace(
"Environment=PYTHONDONTWRITEBYTECODE=1",
"Environment=UNKNOWN_KEY=value",
),
VALID_UNIT.replace(
"AIOS_TASK_PLANE_EXECUTION_ENABLED=false",
"AIOS_TASK_PLANE_EXECUTION_ENABLED=true",
),
VALID_UNIT.replace(
"Environment=PYTHONDONTWRITEBYTECODE=1",
"Environment=Authorization: Bearer placeholder",
),
)
for value in bad_values:
with self.subTest(value=value):
self.assert_profile_rejected(value)
def test_identity_working_directory_and_path_prefix_bypasses_are_rejected(self):
bad_values = (
VALID_UNIT.replace("User=aiman", "User=root"),
VALID_UNIT.replace(
"/usr/libexec/aios-codex-review-test-worker/current",
"/data/aios-gitops/worktrees/worker",
),
VALID_UNIT.replace(
"ReadWritePaths=/run/aios-codex-review-test-worker",
"ReadWritePaths=/etc",
),
VALID_UNIT.replace(
"ReadOnlyPaths=/data/aios-gitops/repo.git",
"ReadOnlyPaths=/data",
),
)
for value in bad_values:
with self.subTest(value=value):
self.assert_profile_rejected(value)
def test_unknown_section_directive_and_unicode_lookalike_are_rejected(self):
bad_values = (
VALID_UNIT.replace("[Install]", "[Socket]"),
VALID_UNIT.replace(
"Restart=on-failure",
"Restart=on-failure\nProtectClock=true",
),
VALID_UNIT.replace("ExecStart=", "\u0415xecStart="),
)
for value in bad_values:
with self.subTest(value=value):
self.assert_profile_rejected(value)
def test_invalid_unicode_surrogate_and_missing_final_newline_are_rejected(self):
for value in (VALID_UNIT + "\ud800", VALID_UNIT.rstrip("\n")):
with self.subTest(value=repr(value[-30:])):
self.assert_profile_rejected(value)
def test_writer_request_cannot_modify_policy_file(self):
with self.assertRaises(PermissionError):
writer_policy.validate_path(
"etc/aios-gitops/writer-path-policies/codex-review-worker.json",
self.workspace,
)
def test_coarse_file_and_commit_endpoints_remain_forbidden(self):
for endpoint in ("/api/file", "/api/commit"):
with self.subTest(endpoint=endpoint), self.assertRaisesRegex(
PermissionError, "isolated_writer_operation_forbidden"
):
writer_policy.validate_endpoint(
"POST", endpoint, self.workspace, self.root
)
if __name__ == "__main__":
unittest.main()
+4
View File
@@ -15,6 +15,10 @@ class IsolatedWriterPolicyTests(unittest.TestCase):
os.environ["AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT"] = "required"
os.environ["AIOS_GITOPS_ALLOWED_BRANCH"] = "patch/aios-agent-task-control-plane-20260724"
os.environ["AIOS_GITOPS_ALLOWED_BASE_OID"] = "a8e97868976b0dce17b7cb8487eadc1d9c687622"
os.environ["AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED"] = "2"
os.environ["AIOS_GITOPS_SERVICE_AUDIENCE"] = "test-isolated-writer.service"
os.environ["AIOS_GITOPS_TOKEN_REGISTRY_V2"] = "/tmp/test-writer-token-registry.sqlite3"
os.environ["AIOS_GITOPS_WRITER_OPERATION_LEDGER"] = "/tmp/test-writer-operations.sqlite3"
def tearDown(self):
os.environ.clear()
@@ -0,0 +1,471 @@
import hashlib
import json
import os
import sqlite3
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "active-runtime" / "app"))
import aios_isolated_writer_policy as policy
from isolated_writer_authz import AuthorizationError, authorize, operation_for
from isolated_writer_commit_file import (
AtomicCommitFile,
CommitFileError,
OperationLedger,
SimulatedCrash,
)
from isolated_writer_token_registry import TokenRegistry, token_digest
BRANCH = "patch/codex-review-test-worker-20260724"
AUDIENCE = "aios-codex-review-worker-writer.service"
TASK_ID = "AIOS-CODEX-WRITER-SMOKE-20260724"
TARGET_PATH = "docs/reliability/codex-review-worker-writer-smoke.json"
def run(cwd, *args):
return subprocess.check_output(args, cwd=cwd, text=True).strip()
@dataclass
class DummyGrant:
token_id: str = "test-token-id"
task_id: str = TASK_ID
operation: str = "commit_file.atomic"
record: dict = None
def __post_init__(self):
if self.record is None:
self.record = {"audience": AUDIENCE, "_source_ip": "127.0.0.1"}
class RepositoryCase(unittest.TestCase):
def setUp(self):
self.saved = dict(os.environ)
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
self.authoritative = self.root / "authoritative.git"
self.gitea = self.root / "gitea.git"
self.seed = self.root / "seed"
self.workspace = self.root / "writer"
subprocess.run(["git", "init", "--bare", str(self.authoritative)], check=True, capture_output=True)
subprocess.run(["git", "init", str(self.seed)], check=True, capture_output=True)
subprocess.run(["git", "config", "user.email", "writer@example.invalid"], cwd=self.seed, check=True)
subprocess.run(["git", "config", "user.name", "Writer Test"], cwd=self.seed, check=True)
(self.seed / "README.md").write_text("base\n", encoding="utf-8")
subprocess.run(["git", "add", "README.md"], cwd=self.seed, check=True)
subprocess.run(["git", "commit", "-m", "base"], cwd=self.seed, check=True, capture_output=True)
subprocess.run(["git", "branch", "-M", "main"], cwd=self.seed, check=True)
subprocess.run(["git", "remote", "add", "origin", str(self.authoritative)], cwd=self.seed, check=True)
subprocess.run(["git", "push", "origin", "main"], cwd=self.seed, check=True, capture_output=True)
self.base = run(self.seed, "git", "rev-parse", "HEAD")
subprocess.run(["git", "checkout", "-b", BRANCH], cwd=self.seed, check=True, capture_output=True)
subprocess.run(["git", "push", "origin", BRANCH], cwd=self.seed, check=True, capture_output=True)
subprocess.run(
["git", "--git-dir", str(self.authoritative), "worktree", "add", str(self.workspace), BRANCH],
check=True,
capture_output=True,
)
subprocess.run(["git", "config", "user.email", "writer@example.invalid"], cwd=self.workspace, check=True)
subprocess.run(["git", "config", "user.name", "Writer Test"], cwd=self.workspace, check=True)
subprocess.run(["git", "clone", "--bare", str(self.authoritative), str(self.gitea)], check=True, capture_output=True)
subprocess.run(["git", "remote", "add", "gitea-http", str(self.gitea)], cwd=self.workspace, check=True)
self.registry = TokenRegistry(self.root / "registry.sqlite3")
os.environ.update({
"AIOS_GITOPS_ALLOWED_BRANCH": BRANCH,
"AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED": "2",
"AIOS_GITOPS_SERVICE_AUDIENCE": AUDIENCE,
"AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT": "required",
})
def tearDown(self):
os.environ.clear()
os.environ.update(self.saved)
self.temp.cleanup()
def grant(self, **updates):
now = datetime.now(timezone.utc)
grant = {
"token_id": "smoke-" + hashlib.sha256(os.urandom(16)).hexdigest()[:12],
"audience": AUDIENCE,
"task_id": TASK_ID,
"source_cidrs": ["127.0.0.1/32", "::1/128"],
"allowed_branch": BRANCH,
"approved_base_oid": self.base,
"allowed_operations": [
"runtime.status.read",
"baseline.read",
"patches.read",
"commit_file.atomic",
],
"allowed_paths": [TARGET_PATH],
"not_before": (now - timedelta(seconds=2)).isoformat(),
"expires_at": (now + timedelta(minutes=20)).isoformat(),
"max_uses": {
"runtime.status.read": 2,
"baseline.read": 2,
"patches.read": 2,
"commit_file.atomic": 4,
},
"issued_by": "owner-authorized-local-maintenance-executor",
"created_at": (now - timedelta(seconds=3)).isoformat(),
}
grant.update(updates)
return grant
def issue(self, **updates):
return self.registry.issue(self.grant(**updates))
def auth(self, token, **updates):
args = {
"method": "GET",
"path": "/api/runtime-status",
"source_ip": "127.0.0.1",
"workspace": self.workspace,
}
args.update(updates)
return authorize(self.registry, token, **args)
def request(self, key="writer-smoke-0001", content='{"status":"READY"}\n'):
return {
"branch": BRANCH,
"path": TARGET_PATH,
"content": content,
"message": "chore(task-plane): verify Codex writer endpoint",
"expected_head_oid": run(self.workspace, "git", "rev-parse", "HEAD"),
"idempotency_key": key,
}
def writer(self, crash=None, remote="gitea-http", audit=None):
return AtomicCommitFile(
workspace=self.workspace,
bare_repo=self.authoritative,
ledger=OperationLedger(self.root / "operations.sqlite3"),
remote=remote,
audit=audit or (lambda record: None),
crash_after_state=crash,
)
class TokenPolicyV2Tests(RepositoryCase):
def test_v1_coarse_scope_token_is_rejected(self):
token, record = self.issue()
record["schema_version"] = 1
with self.registry._connect() as db:
db.execute(
"UPDATE token_grants SET grant_json=? WHERE token_id=?",
(json.dumps(record), record["token_id"]),
)
with self.assertRaisesRegex(AuthorizationError, "writer_token_policy_v2_required"):
self.auth(token)
def test_audience_mismatch_is_rejected(self):
token, _ = self.issue(audience="other.service")
with self.assertRaisesRegex(AuthorizationError, "writer_token_audience_rejected"):
self.auth(token)
def test_source_ip_mismatch_is_rejected(self):
token, _ = self.issue(source_cidrs=["192.0.2.0/24"])
with self.assertRaisesRegex(AuthorizationError, "writer_token_source_rejected"):
self.auth(token)
def test_branch_mismatch_is_rejected(self):
token, _ = self.issue(allowed_branch="patch/other")
with self.assertRaisesRegex(AuthorizationError, "writer_token_branch_rejected"):
self.auth(token)
def test_operation_mismatch_is_rejected(self):
token, _ = self.issue(allowed_operations=["baseline.read"])
with self.assertRaisesRegex(AuthorizationError, "writer_token_operation_rejected"):
self.auth(token)
def test_path_mismatch_is_rejected(self):
token, _ = self.issue()
with self.assertRaisesRegex(AuthorizationError, "writer_token_path_rejected"):
self.auth(
token,
method="POST",
path="/api/commit-file",
branch=BRANCH,
target_path="docs/other.json",
)
def test_expired_token_is_rejected(self):
now = datetime.now(timezone.utc)
token, _ = self.issue(
created_at=(now - timedelta(minutes=10)).isoformat(),
not_before=(now - timedelta(minutes=9)).isoformat(),
expires_at=(now - timedelta(seconds=1)).isoformat(),
)
with self.assertRaisesRegex(AuthorizationError, "writer_token_expired"):
self.auth(token)
def test_not_before_token_is_rejected(self):
now = datetime.now(timezone.utc)
token, _ = self.issue(
created_at=now.isoformat(),
not_before=(now + timedelta(minutes=1)).isoformat(),
expires_at=(now + timedelta(minutes=5)).isoformat(),
)
with self.assertRaisesRegex(AuthorizationError, "writer_token_not_yet_valid"):
self.auth(token)
def test_revoked_token_is_rejected(self):
token, grant = self.issue()
self.assertTrue(self.registry.revoke(grant["token_id"]))
with self.assertRaisesRegex(AuthorizationError, "writer_token_revoked"):
self.auth(token)
def test_operation_use_count_is_atomic_and_exhausted(self):
token, _ = self.issue(max_uses={
"runtime.status.read": 1,
"baseline.read": 1,
"patches.read": 1,
"commit_file.atomic": 1,
})
self.auth(token)
with self.assertRaisesRegex(AuthorizationError, "writer_token_operation_exhausted"):
self.auth(token)
def test_concurrent_use_count_never_exceeds_limit(self):
token, _ = self.issue(max_uses={
"runtime.status.read": 2,
"baseline.read": 1,
"patches.read": 1,
"commit_file.atomic": 1,
})
successes, failures = [], []
def invoke():
try:
successes.append(self.auth(token))
except AuthorizationError as exc:
failures.append(exc.code)
threads = [threading.Thread(target=invoke) for _ in range(6)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
self.assertEqual(2, len(successes))
self.assertEqual(4, failures.count("writer_token_operation_exhausted"))
def test_registry_does_not_store_plaintext_token(self):
token, _ = self.issue()
registry_bytes = b"".join(
item.read_bytes()
for item in self.registry.path.parent.glob(self.registry.path.name + "*")
if item.is_file()
)
self.assertNotIn(token.encode(), registry_bytes)
with self.registry._connect() as db:
stored = db.execute("SELECT token_digest FROM token_grants").fetchone()[0]
self.assertEqual(token_digest(token), stored)
def test_coarse_routes_have_no_operation_mapping(self):
for path in ("/api/file", "/api/commit", "/api/tag", "/api/rollback", "/api/model"):
with self.subTest(path=path), self.assertRaises(AuthorizationError) as captured:
operation_for("POST", path)
self.assertEqual(403, captured.exception.status)
def test_route_mapping_is_fixed_not_request_controlled(self):
self.assertEqual("runtime.status.read", operation_for("GET", "/api/runtime-status"))
self.assertEqual("baseline.read", operation_for("GET", "/api/baseline"))
self.assertEqual("patches.read", operation_for("GET", "/api/patches"))
self.assertEqual("commit_file.atomic", operation_for("POST", "/api/commit-file"))
def test_x_forwarded_for_cannot_replace_socket_peer(self):
token, _ = self.issue(source_cidrs=["203.0.113.9/32"])
with self.assertRaisesRegex(AuthorizationError, "writer_token_source_rejected"):
self.auth(token, source_ip="127.0.0.1")
class AtomicCommitFileTests(RepositoryCase):
def test_commit_file_succeeds_and_verifies_bare_and_gitea(self):
result = self.writer().execute(DummyGrant(), self.request())
self.assertEqual("COMPLETED", result["state"])
self.assertEqual(
result["commit_oid"],
run(self.root, "git", "--git-dir", str(self.gitea), "rev-parse", f"refs/heads/{BRANCH}"),
)
def test_same_idempotency_key_returns_same_commit(self):
request = self.request()
first = self.writer().execute(DummyGrant(), request)
count = run(self.workspace, "git", "rev-list", "--count", "HEAD")
second = self.writer().execute(DummyGrant(), request)
self.assertEqual(first["commit_oid"], second["commit_oid"])
self.assertEqual(count, run(self.workspace, "git", "rev-list", "--count", "HEAD"))
self.assertTrue(second["idempotent"])
def test_same_key_with_different_payload_returns_409(self):
request = self.request()
self.writer().execute(DummyGrant(), request)
request["content"] = '{"status":"DIFFERENT"}\n'
with self.assertRaises(CommitFileError) as captured:
self.writer().execute(DummyGrant(), request)
self.assertEqual(409, captured.exception.status)
def test_same_key_from_different_task_returns_409(self):
request = self.request()
self.writer().execute(DummyGrant(), request)
other = DummyGrant(token_id="other-token", task_id="OTHER-TASK")
with self.assertRaisesRegex(CommitFileError, "writer_idempotency_owner_conflict"):
self.writer().execute(other, request)
def test_file_written_crash_is_recoverable(self):
request = self.request()
with self.assertRaises(SimulatedCrash):
self.writer(crash="FILE_WRITTEN").execute(DummyGrant(), request)
result = self.writer().execute(DummyGrant(), request)
self.assertEqual("COMPLETED", result["state"])
def test_committed_crash_is_recoverable_without_second_commit(self):
request = self.request()
with self.assertRaises(SimulatedCrash):
self.writer(crash="COMMITTED").execute(DummyGrant(), request)
committed = run(self.workspace, "git", "rev-parse", "HEAD")
result = self.writer().execute(DummyGrant(), request)
self.assertEqual(committed, result["commit_oid"])
def test_pushed_crash_only_reverifies(self):
request = self.request()
with self.assertRaises(SimulatedCrash):
self.writer(crash="PUSHED").execute(DummyGrant(), request)
result = self.writer().execute(DummyGrant(), request)
self.assertEqual("COMPLETED", result["state"])
def test_push_failure_is_failed_recoverable(self):
request = self.request()
with self.assertRaisesRegex(CommitFileError, "writer_gitea_push_failed"):
self.writer(remote="missing-remote").execute(DummyGrant(), request)
row = OperationLedger(self.root / "operations.sqlite3").get(request["idempotency_key"])
self.assertEqual("FAILED_RECOVERABLE", row["state"])
def test_branch_head_conflict_requires_manual_review(self):
request = self.request()
ledger = OperationLedger(self.root / "operations.sqlite3")
ledger.get_or_create(DummyGrant(), request, __import__("isolated_writer_commit_file").canonical_digest(request))
(self.workspace / "README.md").write_text("drift\n", encoding="utf-8")
subprocess.run(["git", "commit", "-am", "drift"], cwd=self.workspace, check=True, capture_output=True)
with self.assertRaisesRegex(CommitFileError, "writer_branch_head_conflict"):
self.writer().execute(DummyGrant(), request)
self.assertEqual("MANUAL_REVIEW_REQUIRED", ledger.get(request["idempotency_key"])["state"])
def test_extra_staged_file_is_rejected(self):
(self.workspace / "README.md").write_text("staged\n", encoding="utf-8")
subprocess.run(["git", "add", "README.md"], cwd=self.workspace, check=True)
with self.assertRaisesRegex(CommitFileError, "writer_extra_staged_files_rejected"):
self.writer().execute(DummyGrant(), self.request())
def test_path_traversal_and_protected_roots_are_rejected(self):
for path in (
"../docs/x", "%2e%2e/docs/x", "/etc/x", r"C:\temp\x",
r"\\server\share\x", "baseline/current.json", "runtime/x",
):
request = self.request(key="writer-path-" + hashlib.sha256(path.encode()).hexdigest()[:12])
request["path"] = path
with self.subTest(path=path), self.assertRaises((PermissionError, CommitFileError)):
self.writer().execute(DummyGrant(), request)
def test_audit_contains_ids_but_not_content_or_authorization(self):
records = []
request = self.request(content='{"value":"non-secret-body-marker"}\n')
self.writer(audit=records.append).execute(DummyGrant(), request)
encoded = json.dumps(records)
self.assertIn("test-token-id", encoded)
self.assertNotIn("non-secret-body-marker", encoded)
self.assertNotIn("Authorization", encoded)
def test_concurrent_same_idempotency_creates_one_commit(self):
request = self.request()
results, errors = [], []
def invoke():
try:
results.append(self.writer().execute(DummyGrant(), request))
except Exception as exc:
errors.append(exc)
threads = [threading.Thread(target=invoke) for _ in range(4)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
self.assertFalse(errors)
self.assertEqual(1, len({item["commit_oid"] for item in results}))
def test_different_operations_are_serialized_by_branch_lock(self):
active = 0
maximum = 0
guard = threading.Lock()
class Probe(AtomicCommitFile):
def _execute_locked(inner, grant, data, digest):
nonlocal active, maximum
with guard:
active += 1
maximum = max(maximum, active)
time.sleep(0.05)
with guard:
active -= 1
return {"state": "PROBED"}
probe = Probe(
workspace=self.workspace,
bare_repo=self.authoritative,
ledger=OperationLedger(self.root / "probe.sqlite3"),
remote="gitea-http",
audit=lambda record: None,
)
requests = [self.request(key=f"writer-lock-{index:04d}") for index in range(3)]
threads = [threading.Thread(target=probe.execute, args=(DummyGrant(), item)) for item in requests]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
self.assertEqual(1, maximum)
def test_invalid_expected_head_and_idempotency_are_rejected(self):
request = self.request()
request["expected_head_oid"] = "short"
with self.assertRaisesRegex(CommitFileError, "writer_expected_head_rejected"):
self.writer().execute(DummyGrant(), request)
request = self.request()
request["idempotency_key"] = "../bad"
with self.assertRaisesRegex(CommitFileError, "writer_idempotency_key_rejected"):
self.writer().execute(DummyGrant(), request)
def test_invalid_message_and_nul_content_are_rejected(self):
request = self.request()
request["message"] = "bad\nmessage"
with self.assertRaisesRegex(CommitFileError, "writer_message_rejected"):
self.writer().execute(DummyGrant(), request)
request = self.request()
request["content"] = "bad\x00content"
with self.assertRaisesRegex(CommitFileError, "writer_content_rejected"):
self.writer().execute(DummyGrant(), request)
def test_force_flags_are_403_and_unknown_fields_are_rejected(self):
for field in ("force", "force_push"):
request = self.request(key=f"writer-force-{field}")
request[field] = True
with self.subTest(field=field), self.assertRaises(CommitFileError) as captured:
self.writer().execute(DummyGrant(), request)
self.assertEqual(403, captured.exception.status)
request = self.request(key="writer-extra-field")
request["operation"] = "commit_file.atomic"
with self.assertRaisesRegex(CommitFileError, "writer_request_fields_rejected"):
self.writer().execute(DummyGrant(), request)
if __name__ == "__main__":
unittest.main()