from __future__ import annotations import dataclasses import json import sys import threading import urllib.request from datetime import datetime, timedelta, timezone from http.server import ThreadingHTTPServer from pathlib import Path 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")) sys.path.insert(0, str(ROOT / "scripts")) from silent_release_hardening import ( # noqa: E402 CapabilityAuthorizationError, CapabilityController, CapabilityLedger, ClaimProbeResult, ApprovalKeyRing, ApprovalRevocationStore, BuilderAttestationVerifier, DeploymentPlan, FinalizationPlan, ProbeEvidence, ReleasePlan, SignedApprovalVerifier, SignedReleasePlanApproval, SilentReleaseActionService, TagPlan, TagPublicationProbeEvidence, TrustedCandidateBundle, TrustedReleasePolicy, classify_tag_publication, ) from silent_release_controller import ReleaseCandidate, ReleaseState # noqa: E402 from silent_release_execution import ( # noqa: E402 LocalTestApprovalSigner, LocalTestBuilderSigner, PersistentSilentReleaseController, ShadowActionHandler, SignedTrustedBuilderAdapter, UnavailableProductionApprovalBackend, ) from build_silent_release_artifact import ( # noqa: E402 redact_output, secret_scan, ) COMMIT = "1" * 40 TAG_OID = "2" * 40 def approval() -> SignedReleasePlanApproval: return SignedReleasePlanApproval( release_plan_id="release-plan-readiness-001", candidate_id="rc-readiness-001", source_commit=COMMIT, tree_oid="3" * 40, artifact_sha256="4" * 64, manifest_sha256="5" * 64, deployment_plan_sha256="6" * 64, policy_sha256="7" * 64, approval_plan_id="owner-plan-readiness-001", evidence_closure_oid="8" * 40, owner_approval_record_oid="9" * 40, capabilities=( "runtime_switch", "tag_publish", "baseline_registry_finalize", ), expires_at="2099-01-01T00:00:00Z", nonce="nonce-readiness-000000001", key_id="owner-key-readiness-1", signature="test-signature", ) def tag_plan() -> TagPlan: return TagPlan( version="shadow-readiness-001", tag_type="annotated", expected_tag_oid=TAG_OID, expected_target_oid=COMMIT, evidence_closure_oid="8" * 40, approval_plan_id="owner-plan-readiness-001", repository_identity="aios/aios", gitea_owner="aios", gitea_repository="aios", current_remote_tag_oid=None, remote_absent_evidence_sha256="a" * 64, verification_contract="git+api+listing/v1", ) def make_failed_claim( ledger: CapabilityLedger, *, operation: str, ) -> dict: claimed = ledger.claim( approval(), "runtime_switch", operation=operation, executor_id="worker-initial", lease_seconds=10, now=100, ) ledger.fail_recoverable( approval(), "runtime_switch", operation=operation, claim_id=claimed["claim_id"], reason="adapter_error", ) return claimed def test_failed_recoverable_not_applied_retries_once(tmp_path): ledger = CapabilityLedger( tmp_path / "ledger", maximum_probe_age_seconds=30 ) make_failed_claim(ledger, operation="runtime-retry") calls: list[str] = [] controller = CapabilityController( capability="runtime_switch", ledger=ledger, adapter=lambda: calls.append("adapter") or {"verified": True}, recovery_probe=lambda: ClaimProbeResult.NOT_APPLIED, executor_id="worker-recovery", ) assert controller.execute( approval(), operation="runtime-retry" )["status"] == "EXECUTED" assert controller.execute( approval(), operation="runtime-retry" )["status"] == "IDEMPOTENT" assert calls == ["adapter"] def test_failed_recoverable_applied_verified_skips_adapter(tmp_path): ledger = CapabilityLedger(tmp_path / "ledger") make_failed_claim(ledger, operation="runtime-applied") calls: list[str] = [] controller = CapabilityController( capability="runtime_switch", ledger=ledger, adapter=lambda: calls.append("adapter") or {"verified": True}, recovery_probe=lambda: ClaimProbeResult.APPLIED_VERIFIED, executor_id="worker-recovery", ) assert controller.execute( approval(), operation="runtime-applied" )["status"] == "IDEMPOTENT" assert calls == [] @pytest.mark.parametrize( ("probe", "expected"), ( ( ClaimProbeResult.INDETERMINATE, "RECONCILIATION_REQUIRED", ), ( ClaimProbeResult.CONFLICT, "MANUAL_OWNER_REVIEW_REQUIRED", ), ), ) def test_failed_recoverable_unknown_or_conflict_stops_writes( tmp_path, probe, expected ): ledger = CapabilityLedger(tmp_path / "ledger") make_failed_claim(ledger, operation=f"runtime-{probe.value.lower()}") calls: list[str] = [] controller = CapabilityController( capability="runtime_switch", ledger=ledger, adapter=lambda: calls.append("adapter") or {"verified": True}, recovery_probe=lambda: probe, executor_id="worker-recovery", ) assert controller.execute( approval(), operation=f"runtime-{probe.value.lower()}" )["status"] == expected assert calls == [] def test_probe_must_bind_operation_and_be_fresh(tmp_path): ledger = CapabilityLedger( tmp_path / "ledger", maximum_probe_age_seconds=10 ) make_failed_claim(ledger, operation="runtime-freshness") record = ledger.read( approval(), "runtime_switch", operation="runtime-freshness" ) assert record is not None with pytest.raises( CapabilityAuthorizationError, match="operation digest" ): ledger.claim( approval(), "runtime_switch", operation="runtime-freshness", executor_id="worker-recovery", lease_seconds=10, now=120, side_effect_probe=lambda: ProbeEvidence( result=ClaimProbeResult.NOT_APPLIED, operation_digest="0" * 64, observed_at=120, ), ) with pytest.raises(CapabilityAuthorizationError, match="freshness"): ledger.claim( approval(), "runtime_switch", operation="runtime-freshness", executor_id="worker-recovery", lease_seconds=10, now=120, side_effect_probe=lambda: ProbeEvidence( result=ClaimProbeResult.NOT_APPLIED, operation_digest=record["operation_digest"], observed_at=100, ), ) def test_probe_journal_is_hash_chained_and_tamper_fails_closed(tmp_path): ledger = CapabilityLedger(tmp_path / "ledger") make_failed_claim(ledger, operation="runtime-journal") ledger.claim( approval(), "runtime_switch", operation="runtime-journal", executor_id="worker-recovery", lease_seconds=10, now=101, side_effect_probe=lambda: ClaimProbeResult.INDETERMINATE, ) operation_path = ledger._operation_path( approval(), "runtime_switch", "runtime-journal" ) records = ledger.verify_probe_journal(operation_path) assert records[0]["payload"]["result"] == "INDETERMINATE" journal = ledger._probe_journal_path(operation_path) payload = json.loads(journal.read_text(encoding="utf-8")) payload["payload"]["result"] = "NOT_APPLIED" journal.write_text(json.dumps(payload) + "\n", encoding="utf-8") with pytest.raises(Exception, match="hash chain"): ledger.verify_probe_journal(operation_path) def tag_evidence(**overrides) -> TagPublicationProbeEvidence: values = { "repository_identity": "aios/aios", "remote_reachable": True, "gitea_api_reachable": True, "gitea_listing_reachable": True, "remote_tag_oid": TAG_OID, "annotated_tag_object_oid": TAG_OID, "remote_peeled_target_oid": COMMIT, "gitea_api_tag_oid": TAG_OID, "gitea_api_peeled_target_oid": COMMIT, "gitea_listing_visible": True, "gitea_listing_tag_oid": TAG_OID, "gitea_listing_peeled_target_oid": COMMIT, "observed_at": 100, } values.update(overrides) return TagPublicationProbeEvidence(**values) def test_tag_probe_exact_absence_is_not_applied(): evidence = tag_evidence( remote_tag_oid=None, annotated_tag_object_oid=None, remote_peeled_target_oid=None, gitea_api_tag_oid=None, gitea_api_peeled_target_oid=None, gitea_listing_visible=False, gitea_listing_tag_oid=None, gitea_listing_peeled_target_oid=None, ) assert ( classify_tag_publication(evidence, tag_plan()) == ClaimProbeResult.NOT_APPLIED ) def test_tag_probe_exact_publication_is_applied_verified(): assert ( classify_tag_publication(tag_evidence(), tag_plan()) == ClaimProbeResult.APPLIED_VERIFIED ) def test_tag_probe_unavailable_or_divergent_is_indeterminate(): assert ( classify_tag_publication( tag_evidence(gitea_api_reachable=False), tag_plan() ) == ClaimProbeResult.INDETERMINATE ) assert ( classify_tag_publication( tag_evidence(gitea_listing_tag_oid=None), tag_plan() ) == ClaimProbeResult.INDETERMINATE ) def test_tag_probe_same_name_different_oid_is_conflict(): assert ( classify_tag_publication( tag_evidence(remote_tag_oid="f" * 40), tag_plan() ) == ClaimProbeResult.CONFLICT ) assert ( classify_tag_publication( tag_evidence(remote_peeled_target_oid="e" * 40), tag_plan() ) == ClaimProbeResult.CONFLICT ) def readiness_release_plan(source_commit: str = COMMIT) -> ReleasePlan: policy_sha = "7" * 64 return ReleasePlan( deployment=DeploymentPlan( service="aios-silent-release-controller", release_path="/usr/libexec/aios-silent-release-shadow/release-1", release_root="/usr/libexec/aios-silent-release-shadow", cutover_link="/usr/libexec/aios-silent-release-shadow/current", dropin_path=( "/usr/libexec/aios-silent-release-shadow/release-1/" "deploy/systemd/aios-silent-release-controller.service" ), dropin_sha256="a" * 64, systemd_unit_sha256="a" * 64, rollback_path=( "/usr/libexec/aios-silent-release-shadow/rollback-1" ), rollback_sha256="b" * 64, source_commit=source_commit, tree_oid="3" * 40, artifact_sha256="4" * 64, manifest_sha256="5" * 64, package_closure_sha256="c" * 64, authoritative_git_dir="/data/aios-gitops/repo.git", repository_identity="aios/aios", writer_expected=False, baseline_before_sha256="d" * 64, registry_before_sha256="e" * 64, production_tag_refset_before_sha256="f" * 64, health_contract_version="shadow-health/v1", timeout_policy={"health_seconds": 10}, retry_policy={"health_attempts": 1}, policy_sha256=policy_sha, health_contract={"shadow_mode": True}, ), tag=TagPlan( version="shadow-readiness-001", tag_type="annotated", expected_tag_oid=TAG_OID, expected_target_oid=source_commit, evidence_closure_oid="8" * 40, approval_plan_id="owner-plan-readiness-001", repository_identity="aios/aios", gitea_owner="aios", gitea_repository="aios", current_remote_tag_oid=None, remote_absent_evidence_sha256="a" * 64, verification_contract="shadow-only/v1", ), finalization=FinalizationPlan( version="shadow-readiness-001", expected_tag_oid=TAG_OID, expected_target_oid=source_commit, expected_previous_baseline_sha256="d" * 64, registry_expected_record={ "shadow": True, "target_oid": source_commit, }, baseline_expected_record={ "shadow": True, "target_oid": source_commit, }, final_consistency_contract="shadow-simulated/v1", ), policy_sha256=policy_sha, ) def readiness_candidate(source_commit: str = COMMIT) -> ReleaseCandidate: return ReleaseCandidate( candidate_id="rc-shadow-" + source_commit[:12], source_commit=source_commit, tree_oid="3" * 40, artifact_sha256="4" * 64, manifest_sha256="5" * 64, tests={"passed": True, "source": "trusted-builder"}, canary={"passed": True, "without_git": True}, rollback_ready=True, git_consistency={"consistent": True}, gitea_consistency={"consistent": True}, approval_state="PENDING", state=ReleaseState.READY_FOR_APPROVAL, ) def trusted_service(tmp_path): builder_signer = LocalTestBuilderSigner.generate() builder_public = builder_signer.public_key.public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw, ) def collector(intent): source_commit = str(intent["source_commit"]) return TrustedCandidateBundle( candidate=readiness_candidate(source_commit), release_plan=readiness_release_plan(source_commit), checks={ "authoritative_git": True, "artifact": True, "canary": True, "gitea": True, "manifest": True, "package_closure": True, "rollback": True, "secret_scan": True, "tests": True, }, ) owner_private = Ed25519PrivateKey.generate() owner_public = owner_private.public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw, ) verifier = SignedApprovalVerifier( ApprovalKeyRing({"shadow-owner-test-key": owner_public}), ApprovalRevocationStore(tmp_path / "revocations"), allowed_trust_domains={"aios.shadow"}, ) policy = TrustedReleasePolicy( services=frozenset({"aios-silent-release-controller"}), release_roots=frozenset( {"/usr/libexec/aios-silent-release-shadow"} ), cutover_links=frozenset( {"/usr/libexec/aios-silent-release-shadow/current"} ), dropin_paths=frozenset( { "/usr/libexec/aios-silent-release-shadow/release-1/" "deploy/systemd/aios-silent-release-controller.service" } ), authoritative_git_dirs=frozenset( {"/data/aios-gitops/repo.git"} ), repository_identities=frozenset({"aios/aios"}), ) service = SilentReleaseActionService( tmp_path / "state", execution_enabled=False, approval_verifier=verifier, candidate_builder=SignedTrustedBuilderAdapter( collector, builder_signer ), builder_attestation_verifier=BuilderAttestationVerifier( {builder_signer.key_id: builder_public} ), trusted_policy=policy, expected_repository="aios/aios", policy_sha256="7" * 64, ) return service, LocalTestApprovalSigner( "shadow-owner-test-key", owner_private ) def signed_shadow_approval( service: SilentReleaseActionService, signer: LocalTestApprovalSigner, candidate_id: str, ) -> SignedReleasePlanApproval: candidate = service.candidates.load(candidate_id) plan = service.release_plans.load(candidate_id) now = datetime.now(timezone.utc) value = SignedReleasePlanApproval( release_plan_id="release-plan-shadow-readiness-001", candidate_id=candidate_id, source_commit=candidate.source_commit, tree_oid=candidate.tree_oid, artifact_sha256=candidate.artifact_sha256, manifest_sha256=candidate.manifest_sha256, deployment_plan_sha256=plan.deployment.sha256, release_plan_sha256=plan.sha256, policy_sha256=plan.policy_sha256, approval_plan_id=plan.tag.approval_plan_id, evidence_closure_oid=plan.tag.evidence_closure_oid, owner_approval_record_oid="9" * 40, capabilities=( "runtime_switch", "tag_publish", "baseline_registry_finalize", ), owner_identity="owner:shadow-test", trust_domain="aios.shadow", issued_at=now.isoformat().replace("+00:00", "Z"), not_before=now.isoformat().replace("+00:00", "Z"), expires_at=(now + timedelta(minutes=5)) .isoformat() .replace("+00:00", "Z"), nonce="nonce-shadow-readiness-000001", key_id="shadow-owner-test-key", signature="", ) return signer.sign(value) def test_trusted_builder_attestation_and_caller_evidence_gate(tmp_path): service, _ = trusted_service(tmp_path) result = service.create_release_intent( {"event_id": "webhook-main-001", "source_commit": COMMIT} ) acceptance = service.get_acceptance(result["candidate_id"]) assert acceptance["attestation"]["builder_attestation_id"].startswith( "builder-attestation-" ) assert acceptance["checks"]["tests"] is True with pytest.raises( CapabilityAuthorizationError, match="caller-supplied evidence" ): service.create_release_intent( { "event_id": "webhook-main-002", "source_commit": COMMIT, "tests": {"passed": True}, } ) def test_builder_attestation_tamper_is_rejected(tmp_path): service, _ = trusted_service(tmp_path) bundle = service.candidate_builder.build( {"event_id": "webhook-main-001", "source_commit": COMMIT} ) tampered = dataclasses.replace( bundle.attestation, artifact_sha256="0" * 64 ) with pytest.raises( CapabilityAuthorizationError, match="does not bind" ): dataclasses.replace(bundle, attestation=tampered).validate( service.builder_attestation_verifier ) def test_production_approval_backend_missing_disables_execution(): backend = UnavailableProductionApprovalBackend() assert backend.available is False with pytest.raises( CapabilityAuthorizationError, match="not configured" ): backend.verify() def test_controller_duplicate_webhook_notification_and_restart_are_idempotent( tmp_path, ): service, _ = trusted_service(tmp_path) first_controller = PersistentSilentReleaseController( tmp_path / "state", service, main_probe=lambda: COMMIT, ) first = first_controller.run_once() second = first_controller.run_once() restarted = PersistentSilentReleaseController( tmp_path / "state", service, main_probe=lambda: COMMIT, ).run_once() assert first["intent"]["candidate_id"] == second["intent"]["candidate_id"] assert second["intent"]["idempotent"] is True assert first["notification"]["idempotent"] is False assert second["notification"]["idempotent"] is True assert restarted["notification"]["idempotent"] is True assert restarted["side_effects_executed"] is False def test_controller_startup_journal_failure_stops_before_main_event(tmp_path): service, _ = trusted_service(tmp_path) service.sagas.create( "rc-shadow-" + COMMIT[:12], "release-plan-shadow-readiness-001" ) journal = next(service.sagas.root.glob("*.events.jsonl")) journal.write_text('{"tampered":true}\n', encoding="utf-8") probes: list[str] = [] result = PersistentSilentReleaseController( tmp_path / "state", service, main_probe=lambda: probes.append("main") or COMMIT, ).run_once() assert result["status"] == "FAIL_CLOSED" assert result["intent"] is None assert result["side_effects_executed"] is False assert probes == [] def test_action_http_shadow_flow_has_zero_side_effects(tmp_path): service, signer = trusted_service(tmp_path) handler = type( "TestShadowHandler", (ShadowActionHandler,), {"service": service, "controller": None}, ) server = ThreadingHTTPServer(("127.0.0.1", 0), handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() base = f"http://127.0.0.1:{server.server_port}" def post(path, payload): request = urllib.request.Request( base + path, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, ) with urllib.request.urlopen(request, timeout=5) as response: return json.load(response) try: intent = post( "/createReleaseIntent", {"event_id": "http-main-001", "source_commit": COMMIT}, ) candidate_id = intent["candidate_id"] approval = signed_shadow_approval(service, signer, candidate_id) accepted = post("/submitSignedApproval", approval.to_dict()) deployed = post( "/startDeployment", {"candidate_id": candidate_id} ) with urllib.request.urlopen( base + "/getReleasePlan?candidate_id=" + candidate_id, timeout=5, ) as response: plan = json.load(response) finally: server.shutdown() server.server_close() thread.join(timeout=5) assert accepted["status"] == "VERIFIED_AND_STORED" assert deployed == { "candidate_id": candidate_id, "execution_enabled": False, "shadow_mode": True, "side_effects": [], "side_effects_executed": False, "status": "SHADOW_COMPLETED", } assert plan["shadow_mode"] is True assert plan["side_effects_executed"] is False def test_secret_scan_redacts_placeholders_but_rejects_real_material(tmp_path): (tmp_path / "safe.txt").write_text( "Authorization: Bearer test-placeholder-token-000000", encoding="utf-8", ) assert secret_scan(tmp_path) == {"findings": [], "passed": True} (tmp_path / "unsafe.txt").write_text( "-----BEGIN " + "OPENSSH PRIVATE KEY-----\n" + ("A" * 80) + "\n-----END " + "OPENSSH PRIVATE KEY-----", encoding="utf-8", ) result = secret_scan(tmp_path) assert result["passed"] is False assert result["findings"] == ["unsafe.txt"] secret = "live-" + ("A" * 40) redacted = redact_output("Authorization: Bearer " + secret) assert secret not in redacted assert "" in redacted (tmp_path / "unsafe.txt").write_text( "Authorization: " + "Bearer live-test-looking-secret-000001", encoding="utf-8", ) result = secret_scan(tmp_path) assert result["passed"] is False assert result["findings"] == ["unsafe.txt"] def test_systemd_shadow_unit_has_both_gates_and_no_secret_environment(): unit = ( ROOT / "deploy" / "systemd" / "aios-silent-release-controller.service" ).read_text(encoding="utf-8") assert "AIOS_GITOPS_WRITES_ENABLED=false" in unit assert "AIOS_SILENT_RELEASE_EXECUTION_ENABLED=false" in unit assert "--host 127.0.0.1" in unit assert "PRIVATE_KEY=" not in unit assert "TOKEN=" not in unit assert "PASSWORD=" not in unit