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()