472 lines
20 KiB
Python
472 lines
20 KiB
Python
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()
|