from __future__ import annotations import dataclasses import hashlib import hmac import json import sys from pathlib import Path from types import SimpleNamespace import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "active-runtime" / "app")) import aios_trusted_candidate_builder as builder_runtime # noqa: E402 from aios_trusted_candidate_builder import ( # noqa: E402 BuilderError, BuilderKeyring, DeterministicEvidenceCollector, PeerIdentity, ServerOwnedEvidenceSource, TrustedCandidateBuilder, TrustedCandidateBuilderConfig, UnixTrustedCandidateBuilderClient, load_builder_policy, ) from silent_release_controller import ReleaseCandidate, ReleaseState # noqa: E402 from silent_release_hardening import ( # noqa: E402 BuilderAttestation, BuilderAttestationKeyRecord, BuilderAttestationKeyRing, BuilderAttestationVerifier, DeploymentPlan, FinalizationPlan, ReleasePlan, TagPlan, TrustedCandidateBundle, ) POLICY_PATH = ROOT / "config" / "gitops" / "trusted-candidate-builder-policy.json" NOW = 1_725_000_000 class RecordingDomainSigner: key_id = "test-builder-key" key_generation = 1 def __init__(self) -> None: self.calls: list[tuple[str, bytes]] = [] def sign_for_purpose(self, purpose: str, payload: bytes) -> bytes: self.calls.append((purpose, payload)) return hashlib.sha256(b"test-signature\0" + payload).digest() def make_seam(*, peer: PeerIdentity | None = None, cgroup: str = "/system.slice/aios-gitops-model-api-v2r2c1-staging.service", process=("/usr/bin/python3.12", ("/usr/bin/python3", "/opt/aios/gitops-model-api/test/runtime/app/aios_model_api_staging.py")), server_clock=lambda: NOW, client_clock=lambda: NOW, collector=None, nonce_ledger_path=None, signer_override=None) -> tuple[TrustedCandidateBuilder, UnixTrustedCandidateBuilderClient, RecordingDomainSigner]: policy = load_builder_policy(POLICY_PATH) config = TrustedCandidateBuilderConfig.from_policy(policy) keyring = BuilderKeyring(builder_hmac=b"h" * 32) signer = signer_override or RecordingDomainSigner() server = TrustedCandidateBuilder( config, keyring, collector=collector or DeterministicEvidenceCollector(config, evidence_source=passing_bundle), signer=signer, peer_identity_resolver=lambda _connection: peer or PeerIdentity(pid=77, uid=1000, gid=1000), cgroup_resolver=lambda _pid: cgroup, process_identity_resolver=lambda _pid: process, clock=server_clock, nonce_ledger_path=nonce_ledger_path, ) return server, UnixTrustedCandidateBuilderClient(config, keyring, clock=client_clock), signer def passing_bundle(recipe_name, recipe): base_oid = recipe["base_oid"] digest = lambda label: hashlib.sha256(f"{label}\0{base_oid}".encode()).hexdigest() candidate = ReleaseCandidate( candidate_id=recipe["candidate_id"], source_commit=base_oid, tree_oid=digest("tree")[:40], artifact_sha256=digest("artifact"), manifest_sha256=digest("manifest"), tests={"passed": True, "receipt": digest("tests")}, canary={"passed": True, "receipt": digest("canary")}, rollback_ready=True, git_consistency={"consistent": True}, gitea_consistency={"consistent": True}, state=ReleaseState.READY_FOR_APPROVAL, ) policy_sha = digest("policy") plan = ReleasePlan( deployment=DeploymentPlan( service="aios-silent-release-controller", release_path="/srv/release/r1", release_root="/srv/release", cutover_link="/srv/release/current", dropin_path="/srv/release/r1/deploy/systemd/controller.service", dropin_sha256=digest("dropin"), systemd_unit_sha256=digest("unit"), rollback_path="/srv/release/rollback-r1", rollback_sha256=digest("rollback"), source_commit=base_oid, tree_oid=candidate.tree_oid, artifact_sha256=candidate.artifact_sha256, manifest_sha256=candidate.manifest_sha256, package_closure_sha256=digest("closure"), authoritative_git_dir="/data/aios-gitops/repo.git", repository_identity="aios/aios", baseline_before_sha256=digest("baseline"), registry_before_sha256=digest("registry"), production_tag_refset_before_sha256=digest("tagset"), health_contract_version="trusted-builder/v1", timeout_policy={"seconds": 10}, retry_policy={"attempts": 1}, policy_sha256=policy_sha, health_contract={"checked": True}, ), tag=TagPlan( version="trusted-v2r2c2-001", tag_type="annotated", expected_tag_oid=digest("tag")[:40], expected_target_oid=base_oid, evidence_closure_oid=digest("closure-oid")[:40], approval_plan_id="owner-plan-trusted-v2r2c2", repository_identity="aios/aios", gitea_owner="aios", gitea_repository="aios", current_remote_tag_oid=None, remote_absent_evidence_sha256=digest("tag-absence"), verification_contract="trusted-builder/v1", ), finalization=FinalizationPlan( version="trusted-v2r2c2-001", expected_tag_oid=digest("tag")[:40], expected_target_oid=base_oid, expected_previous_baseline_sha256=digest("baseline"), registry_expected_record={"base": base_oid}, baseline_expected_record={"base": base_oid}, final_consistency_contract="trusted-builder/v1", ), policy_sha256=policy_sha, ) checks = {name: True for name in [ "authoritative_git", "artifact", "canary", "gitea", "manifest", "package_closure", "rollback", "secret_scan", "tests", ]} TrustedCandidateBundle(candidate, plan, checks).validate() canonical_digest = lambda value: hashlib.sha256( json.dumps(value, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() evidence_digests = { "artifact": candidate.artifact_sha256, "manifest": candidate.manifest_sha256, "package_closure": plan.deployment.package_closure_sha256, "tests": canonical_digest(candidate.tests), "canary": canonical_digest(candidate.canary), "gitea": canonical_digest(candidate.gitea_consistency), "rollback": canonical_digest({ "ready": True, "rollback_sha256": plan.deployment.rollback_sha256, "source_commit": candidate.source_commit, }), "policy": plan.policy_sha256, "secret_scan": digest("secret-scan"), } return { "candidate": candidate.to_dict(), "release_plan": plan.to_dict(), "checks": checks, "evidence_digests": evidence_digests, } def rehmac(request: dict) -> None: unsigned = {key: value for key, value in request.items() if key != "hmac"} request["hmac"] = hmac.new( b"h" * 32, json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode(), hashlib.sha256, ).hexdigest() def test_public_build_seam_accepts_only_the_fixed_v2r2c2_recipe_and_returns_server_owned_evidence() -> None: server, client, signer = make_seam() request = client.create_request("V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="a" * 32) result = server.handle_request(request, connection=object()) assert result["schema"] == "aios.trusted-candidate-builder.response.v1" assert result["status"] == "BUILT" assert result["recipe"] == "V2R2C2" assert result["repository"] == "aios/aios" assert result["task_id"] == "v12-durable-trusted-release-controller-activation-20260829" assert result["silent_release_bundle"]["candidate"]["candidate_id"] == "rc-v2r2c2-d2b5a8339a34" assert result["base_oid"] == "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6" assert set(result["checks"]) == set(server.config.required_checks) assert all(result["checks"].values()) assert result["attestation"]["trust_domain"] == "aios.trusted-builder.attestation.v1" assert signer.calls and signer.calls[0][0] == "AIOS_TRUSTED_RELEASE_BUILDER" assert result["attestation"]["fingerprint"] == hashlib.sha256(signer.calls[0][1]).hexdigest() assert BuilderAttestation.from_dict(result["attestation"]).canonical_bytes() == signer.calls[0][1] silent_bundle = result["silent_release_bundle"] round_trip = TrustedCandidateBundle( ReleaseCandidate.from_dict(silent_bundle["candidate"]), ReleasePlan.from_dict(silent_bundle["release_plan"]), silent_bundle["checks"], ) assert round_trip.validate().candidate.candidate_id == result["evidence"]["candidate_id"] @pytest.mark.parametrize( "mutate, error", [ (lambda request: request.update({"base_oid": "f" * 40}), "REQUEST_HMAC_INVALID"), (lambda request: request.update({"artifact_sha256": "0" * 64}), "REQUEST_SCHEMA_INVALID"), ], ) def test_public_build_seam_rejects_tampering_and_caller_owned_evidence(mutate, error) -> None: server, client, _ = make_seam() request = client.create_request("V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="b" * 32) mutate(request) with pytest.raises(BuilderError, match=error): server.handle_request(request, connection=object()) @pytest.mark.parametrize("field,value", [ ("repository", "other/repository"), ("task_id", "other-task"), ("base_oid", "f" * 40), ("purpose", "owner_approval"), ]) def test_public_build_seam_rejects_reauthenticated_scope_rebinding(field, value) -> None: server, client, _ = make_seam() request = client.create_request( "V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="s" * 32, ) request[field] = value rehmac(request) with pytest.raises(BuilderError, match="REQUEST_SCOPE_INVALID"): server.handle_request(request, connection=object()) def test_public_build_seam_rejects_wrong_hmac() -> None: server, client, _ = make_seam() request = client.create_request( "V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="w" * 32, ) request["hmac"] = "0" * 64 with pytest.raises(BuilderError, match="REQUEST_HMAC_INVALID"): server.handle_request(request, connection=object()) def test_public_build_seam_returns_the_original_result_for_exact_nonce_replay_and_rejects_conflicts() -> None: server, client, _ = make_seam() request = client.create_request("V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="c" * 32) first = server.handle_request(request, connection=object()) assert server.handle_request(request, connection=object()) == first conflict = client.create_request("V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="c" * 32) conflict["issued_at"] += 1 unsigned = {key: value for key, value in conflict.items() if key != "hmac"} conflict["hmac"] = hmac.new(b"h" * 32, json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode(), hashlib.sha256).hexdigest() with pytest.raises(BuilderError, match="REPLAY_CONFLICT"): server.handle_request(conflict, connection=object()) def test_exact_replay_survives_restart_and_the_original_freshness_window(tmp_path) -> None: ledger = tmp_path / "nonce-ledger.json" server, client, _ = make_seam(nonce_ledger_path=ledger) request = client.create_request( "V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="r" * 32, ) first = server.handle_request(request, connection=object()) restarted, _, _ = make_seam( server_clock=lambda: NOW + 3600, nonce_ledger_path=ledger, ) assert restarted.handle_request(request, connection=object()) == first def test_readiness_probe_checks_collector_and_signer_without_signing() -> None: server, client, signer = make_seam() request = client.create_request( "V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="q" * 32, operation=server.config.readiness_operation, ) result = server.handle_request(request, connection=object()) assert result["status"] == "READY" assert result["collector_ready"] is True assert result["signer_ready"] is True assert result["builder_key_id"] == signer.key_id assert result["policy_sha256"] == server.config.policy_sha256 assert "signature" not in result assert "silent_release_bundle" not in result assert signer.calls == [] def test_readiness_fails_closed_when_signer_is_unavailable() -> None: class UnavailableSigner: @property def key_id(self): raise BuilderError("SIGNER_UNAVAILABLE") @property def key_generation(self): return 1 def sign_for_purpose(self, purpose, payload): raise AssertionError("readiness must not attempt signing") server, client, _ = make_seam(signer_override=UnavailableSigner()) request = client.create_request( "V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="u" * 32, operation=server.config.readiness_operation, ) with pytest.raises(BuilderError, match="SIGNER_UNAVAILABLE"): server.handle_request(request, connection=object()) def test_unix_client_applies_bounded_timeout_and_fails_closed(monkeypatch) -> None: observed = {} class TimeoutSocket: def __enter__(self): return self def __exit__(self, *_args): return False def settimeout(self, seconds): observed["timeout"] = seconds def connect(self, _path): raise TimeoutError("bounded test timeout") monkeypatch.setattr(builder_runtime.socket, "socket", lambda *_args: TimeoutSocket()) monkeypatch.setattr(builder_runtime.socket, "AF_UNIX", 1, raising=False) monkeypatch.setattr(builder_runtime.socket, "SOCK_STREAM", 1, raising=False) config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH)) client = UnixTrustedCandidateBuilderClient( config, BuilderKeyring(builder_hmac=b"h" * 32), socket_path=Path("/run/aios-trusted-candidate-builder.sock"), clock=lambda: NOW, ) with pytest.raises(BuilderError, match="BUILDER_TRANSPORT_UNAVAILABLE"): client.build( "V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="t" * 32, ) assert observed["timeout"] == config.request_timeout_seconds def test_backend_attestation_round_trip_verifies_with_production_keyring() -> None: private = Ed25519PrivateKey.generate() class Ed25519Signer: key_id = "builder-key-001" key_generation = 1 def sign_for_purpose(self, purpose, payload): assert purpose == "AIOS_TRUSTED_RELEASE_BUILDER" return private.sign(payload) server, client, _ = make_seam(signer_override=Ed25519Signer()) response = server.handle_request( client.create_request( "V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="v" * 32, ), connection=object(), ) public_key = private.public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw, ) verifier = BuilderAttestationVerifier( BuilderAttestationKeyRing( [BuilderAttestationKeyRecord( key_id="builder-key-001", public_key=public_key, algorithm="Ed25519", purpose="AIOS_TRUSTED_RELEASE_BUILDER", generation=1, not_before="2020-01-01T00:00:00Z", expires_at="2099-01-01T00:00:00Z", status="ACTIVE", )], require_attestation_metadata=True, ) ) raw = response["silent_release_bundle"] bundle = TrustedCandidateBundle( ReleaseCandidate.from_dict(raw["candidate"]), ReleasePlan.from_dict(raw["release_plan"]), raw["checks"], BuilderAttestation.from_dict(raw["attestation"], require_fingerprint=True), ) assert bundle.validate(verifier).candidate.candidate_id == response["evidence"]["candidate_id"] def test_public_build_seam_rejects_direct_unbound_v2r2c3_and_authentication_boundaries() -> None: server, client, _ = make_seam() with pytest.raises(BuilderError, match="GOVERNED_V2R2C3_DERIVATION_REQUIRED"): client.create_request("V2R2C3", "unknown", "f" * 40, nonce="d" * 32) denied_peer, denied_client, _ = make_seam(peer=PeerIdentity(pid=77, uid=0, gid=0)) with pytest.raises(BuilderError, match="PEER_IDENTITY_DENIED"): denied_peer.handle_request(denied_client.create_request("V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="e" * 32), connection=object()) denied_cgroup, denied_client, _ = make_seam(cgroup="/system.slice/untrusted.service") with pytest.raises(BuilderError, match="PEER_CGROUP_DENIED"): denied_cgroup.handle_request(denied_client.create_request("V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="f" * 32), connection=object()) denied_process, denied_client, _ = make_seam( process=("/usr/bin/python3.12", ("/usr/bin/python3", "/tmp/aios_model_api_staging.py")) ) with pytest.raises(BuilderError, match="PEER_PROCESS_DENIED"): denied_process.handle_request(denied_client.create_request("V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="p" * 32), connection=object()) stale_server, stale_client, _ = make_seam(client_clock=lambda: NOW - 61) with pytest.raises(BuilderError, match="REQUEST_STALE"): stale_server.handle_request(stale_client.create_request("V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="g" * 32), connection=object()) def test_public_build_seam_fails_closed_when_server_owned_collector_is_incomplete() -> None: def incomplete_evidence(recipe_name, recipe): value = passing_bundle(recipe_name, recipe) value["checks"]["canary"] = False return value policy = load_builder_policy(POLICY_PATH) config = TrustedCandidateBuilderConfig.from_policy(policy) server, client, _ = make_seam( collector=DeterministicEvidenceCollector(config, evidence_source=incomplete_evidence) ) with pytest.raises(BuilderError, match="COLLECTOR_CHECKS_INCOMPLETE"): server.handle_request(client.create_request("V2R2C2", "rc-v2r2c2-d2b5a8339a34", "d2b5a8339a34a8f958a177f8a3b9e8296a57a7a6", nonce="h" * 32), connection=object()) def test_server_owned_source_rehashes_every_fixed_evidence_file_and_rejects_tampering(tmp_path) -> None: config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH)) recipe = config.fixed_recipe("V2R2C2") raw = passing_bundle("V2R2C2", recipe) candidate = ReleaseCandidate.from_dict(raw["candidate"]) plan = ReleasePlan.from_dict(raw["release_plan"]) evidence_root = tmp_path / "evidence" version_root = evidence_root / "V2R2C2" version_root.mkdir(parents=True) def write_bytes(name: str, value: bytes) -> str: path = version_root / name path.write_bytes(value) path.chmod(0o600) return hashlib.sha256(value).hexdigest() def canonical(value: dict) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() tests = {"passed": True, "receipt": "tests-fixed"} canary = {"passed": True, "receipt": "canary-fixed"} gitea = { "consistent": True, "remote_ref_oid": recipe["source_commit"], "repository": "aios/aios", "source_commit": recipe["source_commit"], "tree_oid": candidate.tree_oid, } git_consistency = { "consistent": True, "repository": "aios/aios", "source_commit": recipe["source_commit"], "tree_oid": candidate.tree_oid, } rollback = { "ready": True, "rollback_sha256": plan.deployment.rollback_sha256, "source_commit": recipe["source_commit"], } artifact_bytes = b"immutable-artifact" manifest_bytes = b'{"manifest":"fixed"}\n' closure_bytes = b"closure artifact.tar.gz\n" policy_bytes = b'{"policy":"fixed"}\n' artifact_sha = write_bytes("artifact.tar.gz", artifact_bytes) manifest_sha = write_bytes("manifest.json", manifest_bytes) closure_sha = write_bytes("checksums.sha256", closure_bytes) tests_sha = write_bytes("tests.json", canonical(tests)) canary_sha = write_bytes("canary.json", canonical(canary)) gitea_sha = write_bytes("gitea.json", canonical(gitea)) rollback_sha = write_bytes("rollback.json", canonical(rollback)) policy_sha = write_bytes("trusted-release-policy.json", policy_bytes) secret_scan = {"artifact_sha256": artifact_sha, "passed": True} secret_sha = write_bytes("secret-scan.json", canonical(secret_scan)) candidate.artifact_sha256 = artifact_sha candidate.manifest_sha256 = manifest_sha candidate.tests = tests candidate.canary = canary candidate.git_consistency = git_consistency candidate.gitea_consistency = gitea deployment = dataclasses.replace( plan.deployment, artifact_sha256=artifact_sha, manifest_sha256=manifest_sha, package_closure_sha256=closure_sha, policy_sha256=policy_sha, ) plan = dataclasses.replace(plan, deployment=deployment, policy_sha256=policy_sha) receipt = { "schema": "aios.trusted-candidate-builder.evidence.v1", "version": "V2R2C2", "candidate": candidate.to_dict(), "release_plan": plan.to_dict(), } write_bytes("receipt.json", canonical(receipt)) git_state = {"tree": candidate.tree_oid, "ancestry": 0} def git_runner(argv, **_kwargs): if "rev-parse" in argv: return SimpleNamespace(stdout=git_state["tree"] + "\n", returncode=0) return SimpleNamespace(stdout="", returncode=git_state["ancestry"]) source = ServerOwnedEvidenceSource( config, tmp_path / "repo.git", evidence_root, expected_owner_uid=None, git_runner=git_runner, ) collected = source("V2R2C2", recipe) assert collected["candidate"]["artifact_sha256"] == artifact_sha assert collected["evidence_digests"] == { "artifact": artifact_sha, "manifest": manifest_sha, "package_closure": closure_sha, "tests": tests_sha, "canary": canary_sha, "gitea": gitea_sha, "rollback": rollback_sha, "policy": policy_sha, "secret_scan": secret_sha, } (version_root / "manifest.json").write_bytes(b"tampered") with pytest.raises(BuilderError, match="COLLECTOR_DIGEST_MISMATCH"): source("V2R2C2", recipe) (version_root / "manifest.json").write_bytes(manifest_bytes) for name, tampered, expected in ( ("artifact.tar.gz", b"tampered", "COLLECTOR_DIGEST_MISMATCH"), ("checksums.sha256", b"tampered", "COLLECTOR_DIGEST_MISMATCH"), ("trusted-release-policy.json", b"tampered", "COLLECTOR_DIGEST_MISMATCH"), ("tests.json", canonical({"passed": True, "receipt": "wrong"}), "COLLECTOR_TESTS_MISMATCH"), ("canary.json", canonical({"passed": True, "receipt": "wrong"}), "COLLECTOR_CANARY_MISMATCH"), ("gitea.json", canonical({**gitea, "consistent": False}), "COLLECTOR_GITEA_MISMATCH"), ("rollback.json", canonical({**rollback, "ready": False}), "COLLECTOR_ROLLBACK_MISMATCH"), ("secret-scan.json", canonical({**secret_scan, "passed": False}), "COLLECTOR_SECRET_SCAN_MISMATCH"), ): path = version_root / name original = path.read_bytes() path.write_bytes(tampered) with pytest.raises(BuilderError, match=expected): source("V2R2C2", recipe) path.write_bytes(original) git_state["tree"] = "f" * 40 with pytest.raises(BuilderError, match="COLLECTOR_GIT_MISMATCH"): source("V2R2C2", recipe) git_state["tree"] = candidate.tree_oid git_state["ancestry"] = 1 with pytest.raises(BuilderError, match="COLLECTOR_GIT_MISMATCH"): source("V2R2C2", recipe) def test_systemd_socket_and_service_are_af_unix_hardened_and_credential_bound() -> None: socket_unit = (ROOT / "deploy/systemd/aios-trusted-candidate-builder.socket").read_text() service_unit = (ROOT / "deploy/systemd/aios-trusted-candidate-builder.service").read_text() assert "ListenStream=/run/aios-trusted-candidate-builder.sock" in socket_unit assert "SocketMode=0660" in socket_unit assert "LoadCredentialEncrypted=signing_private_key:" in service_unit assert "LoadCredentialEncrypted=builder_hmac:" in service_unit assert "NoNewPrivileges=true" in service_unit assert "ProtectSystem=strict" in service_unit assert "PrivateNetwork=true" in service_unit assert "RestrictAddressFamilies=AF_UNIX" in service_unit assert "StateDirectory=aios-trusted-candidate-builder" in service_unit assert "AIOS_TRUSTED_BUILDER_KEY_GENERATION=1" in service_unit assert "aios-approved-release-executor" not in service_unit assert "--socket-fd 3" in service_unit policy = load_builder_policy(POLICY_PATH) assert policy["allowed_peer_uid"] == 1000 assert policy["allowed_peer_cgroups"] == [ "/system.slice/aios-gitops-model-api-v2r2c1-staging.service" ] def test_governed_source_admission_resolves_unknown_version_without_overriding_fixed_recipe() -> None: policy = load_builder_policy(POLICY_PATH) calls: list[tuple[str, str | None, str]] = [] def resolver(version: str, candidate_id: str | None, source_commit: str): calls.append((version, candidate_id, source_commit)) return { "mode": "governed-source-admission", "task_id": "AIOS-BASELINE-1.1-PRODUCTION-CLOSURE", "candidate_id": "rc-v2r2c4-" + source_commit[:12], "base_oid": "1" * 40, "source_commit": source_commit, "version": version, "purpose": "AIOS_TRUSTED_RELEASE_BUILDER", } config = dataclasses.replace( TrustedCandidateBuilderConfig.from_policy(policy), source_admission_resolver=resolver, ) client = UnixTrustedCandidateBuilderClient( config, BuilderKeyring(builder_hmac=b"h" * 32), clock=lambda: NOW, ) source = "2" * 40 request = client.create_request( "V2R2C4", None, source, nonce="z" * 32, ) assert request["candidate_id"] == "rc-v2r2c4-" + source[:12] assert request["base_oid"] == "1" * 40 assert request["source_commit"] == source assert request["task_id"] == "AIOS-BASELINE-1.1-PRODUCTION-CLOSURE" assert calls == [("V2R2C4", None, source)] with pytest.raises(BuilderError, match="REQUEST_RELEASE_BINDING_INVALID"): config.resolve_recipe( "V2R2C2", None, source, ) # Fixed policy versions never fall through to the admission resolver. assert calls == [("V2R2C4", None, source)] def test_unknown_release_source_requires_explicit_admission() -> None: config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH)) with pytest.raises(BuilderError, match="RECIPE_UNKNOWN"): config.resolve_recipe("V2R2C4", None, "2" * 40) def test_governed_source_admission_rejects_caller_candidate_rebinding() -> None: policy = load_builder_policy(POLICY_PATH) def resolver(version: str, candidate_id: str | None, source_commit: str): return { "mode": "governed-source-admission", "task_id": "task-admitted-source", "candidate_id": "rc-v2r2c4-" + source_commit[:12], "base_oid": "1" * 40, "source_commit": source_commit, "version": version, "purpose": "AIOS_TRUSTED_RELEASE_BUILDER", } config = dataclasses.replace( TrustedCandidateBuilderConfig.from_policy(policy), source_admission_resolver=resolver, ) with pytest.raises(BuilderError, match="REQUEST_RELEASE_BINDING_INVALID"): config.resolve_recipe("V2R2C4", "rc-attacker", "2" * 40)