from src.services.executor_trust_boundary_readback import ( RECEIPT_SCHEMA_VERSION, build_executor_trust_boundary_readback, ) def _verified_receipt(source_sha: str = "abc123") -> dict[str, str]: return { "schema_version": RECEIPT_SCHEMA_VERSION, "verified_source_sha": source_sha, "verified_at": "2026-07-11T01:00:00Z", "workflow_run_id": "4870", "verifier": "kubectl_auth_can_i_and_live_mount_readback", "api_service_account": "awoooi-api", "worker_service_account": "awoooi-worker", "broker_service_account": "awoooi-executor-broker", "api_mutation_denied": "true", "api_ssh_mount_absent": "true", "worker_mutation_denied": "true", "worker_ssh_mount_absent": "true", "broker_kubernetes_token_absent": "true", "broker_ssh_mount_present": "true", "legacy_executor_binding_absent": "true", "api_ssh_egress_denied": "true", "worker_ssh_egress_denied": "true", "broker_ssh_egress_allowlisted": "true", "legacy_namespace_egress_allow_absent": "true", "single_writer_source_verifier": ( "cd_tests_and_production_source_sha_v1" ), "decision_manager_queue_only": "true", "webhook_router_queue_only": "true", "failure_watcher_queue_only": "true", "auto_approved_direct_execution_blocked": "true", "candidate_idempotency_contract_verified": "true", "apply_idempotency_contract_verified": "true", "critical_break_glass_contract_verified": "true", "independent_post_verifier_source_verifier": ( "cd_tests_and_production_source_sha_v1" ), "asset_postcondition_registry_complete": "true", "independent_post_verifier_read_only": "true", "executor_returncode_not_success_source": "true", "verifier_failure_marks_apply_failed": "true", "verifier_failure_retry_first": "true", "post_verifier_raw_output_not_persisted": "true", "canonical_learning_source_verifier": ( "cd_tests_and_production_source_sha_v1" ), "canonical_playbook_resolver_complete": "true", "km_row_readback_required": "true", "playbook_row_readback_required": "true", "learning_failure_fail_closed": "true", "trust_failure_fail_closed": "true", "learning_receipts_public_safe": "true", "db_identity_verifier": ( "runtime_role_fingerprint_secret_projection_and_" "representative_preflight_v2" ), "api_database_secret_ref": "awoooi-api-db", "worker_database_secret_ref": "awoooi-worker-db", "broker_database_secret_ref": "awoooi-broker-db", "api_db_role_fingerprint": "a" * 64, "worker_db_role_fingerprint": "b" * 64, "broker_db_role_fingerprint": "c" * 64, "api_db_connection_limit": "12", "worker_db_connection_limit": "8", "broker_db_connection_limit": "4", "api_required_connection_budget": "8", "worker_required_connection_budget": "5", "broker_required_connection_budget": "2", "api_concurrent_connection_probe_count": "8", "worker_concurrent_connection_probe_count": "5", "broker_concurrent_connection_probe_count": "2", "api_representative_db_preflight_passed": "true", "worker_representative_db_preflight_passed": "true", "broker_representative_db_preflight_passed": "true", "api_table_privilege_gap_count": "0", "worker_table_privilege_gap_count": "0", "broker_table_privilege_gap_count": "0", "api_sequence_privilege_gap_count": "0", "worker_sequence_privilege_gap_count": "0", "broker_sequence_privilege_gap_count": "0", "representative_db_preflights_verified": "true", "api_http_concurrency_probe_passed": "true", "api_database_null_pool": "true", "worker_database_null_pool": "true", "broker_database_null_pool": "true", "api_monolithic_secret_env_from_absent": "true", "worker_monolithic_secret_env_from_absent": "true", "broker_monolithic_secret_env_from_absent": "true", "distinct_database_secret_refs": "true", "distinct_db_role_fingerprints": "true", "db_identity_isolated": "true", "connection_budget_verified": "true", } def test_executor_trust_boundary_requires_live_controls_and_matching_source() -> None: readback = build_executor_trust_boundary_readback( _verified_receipt(), deployed_source_sha="abc123", ) assert readback["status"] == "verified_ready" assert readback["production_boundary_verified"] is True assert readback["full_workload_boundary_verified"] is True assert readback["source_sha_matches_deployment"] is True assert readback["active_blockers"] == [] assert readback["secret_values_read"] is False database_boundary = readback["database_identity_boundary"] assert database_boundary["db_identity_isolated"] is True assert database_boundary["connection_budget_verified"] is True assert database_boundary["connection_budget"] == { "observed_total": 24, "required_total": 15, } assert database_boundary["controls"][ "representative_db_preflights_verified" ] is True assert database_boundary["secret_values_read"] is False assert database_boundary["role_names_exposed"] is False source_boundary = readback["autonomous_single_writer_source_boundary"] assert source_boundary["source_boundary_verified"] is True assert source_boundary["active_blockers"] == [] assert all(source_boundary["controls"].values()) verifier_boundary = readback["independent_post_verifier_source_boundary"] assert verifier_boundary["source_boundary_verified"] is True assert verifier_boundary["active_blockers"] == [] assert all(verifier_boundary["controls"].values()) learning_boundary = readback["canonical_learning_source_boundary"] assert learning_boundary["source_boundary_verified"] is True assert learning_boundary["active_blockers"] == [] assert all(learning_boundary["controls"].values()) def test_executor_trust_boundary_single_writer_source_controls_fail_closed() -> None: receipt = _verified_receipt() receipt["failure_watcher_queue_only"] = "false" readback = build_executor_trust_boundary_readback( receipt, deployed_source_sha="abc123", ) source_boundary = readback["autonomous_single_writer_source_boundary"] assert readback["production_boundary_verified"] is True assert source_boundary["source_boundary_verified"] is False assert "failure_watcher_queue_only_not_verified" in source_boundary[ "active_blockers" ] def test_executor_trust_boundary_post_verifier_source_controls_fail_closed() -> None: receipt = _verified_receipt() receipt["executor_returncode_not_success_source"] = "false" readback = build_executor_trust_boundary_readback( receipt, deployed_source_sha="abc123", ) verifier_boundary = readback["independent_post_verifier_source_boundary"] assert readback["production_boundary_verified"] is True assert verifier_boundary["source_boundary_verified"] is False assert "executor_returncode_not_success_source_not_verified" in ( verifier_boundary["active_blockers"] ) def test_executor_trust_boundary_canonical_learning_controls_fail_closed() -> None: receipt = _verified_receipt() receipt["playbook_row_readback_required"] = "false" readback = build_executor_trust_boundary_readback( receipt, deployed_source_sha="abc123", ) learning_boundary = readback["canonical_learning_source_boundary"] assert readback["production_boundary_verified"] is True assert learning_boundary["source_boundary_verified"] is False assert "playbook_row_readback_required_not_verified" in ( learning_boundary["active_blockers"] ) def test_executor_trust_boundary_rejects_stale_or_incomplete_receipt() -> None: stale = build_executor_trust_boundary_readback( _verified_receipt("old-source"), deployed_source_sha="new-source", ) assert stale["production_boundary_verified"] is False assert "verified_source_sha_mismatch" in stale["active_blockers"] incomplete_receipt = _verified_receipt() incomplete_receipt["worker_ssh_egress_denied"] = "false" incomplete = build_executor_trust_boundary_readback( incomplete_receipt, deployed_source_sha="abc123", ) assert incomplete["production_boundary_verified"] is False assert "worker_ssh_egress_denied_not_verified" in incomplete["active_blockers"] def test_executor_trust_boundary_keeps_shared_database_identity_in_progress() -> None: shared = _verified_receipt() for workload_id in ("api", "worker", "broker"): shared[f"{workload_id}_database_secret_ref"] = "awoooi-secrets" shared[f"{workload_id}_db_role_fingerprint"] = "d" * 64 shared["api_monolithic_secret_env_from_absent"] = "false" shared["worker_monolithic_secret_env_from_absent"] = "false" shared["distinct_database_secret_refs"] = "false" shared["distinct_db_role_fingerprints"] = "false" shared["db_identity_isolated"] = "false" shared["connection_budget_verified"] = "false" readback = build_executor_trust_boundary_readback( shared, deployed_source_sha="abc123", ) assert readback["production_boundary_verified"] is True assert readback["full_workload_boundary_verified"] is False assert readback["status"] == ( "network_boundary_verified_database_identity_in_progress" ) database_boundary = readback["database_identity_boundary"] assert database_boundary["db_identity_isolated"] is False assert database_boundary["connection_budget_verified"] is False assert "api_monolithic_secret_env_from_present" in database_boundary[ "active_blockers" ] assert "database_secret_refs_not_distinct" in database_boundary[ "active_blockers" ] assert "database_role_fingerprints_not_distinct" in database_boundary[ "active_blockers" ] def test_executor_trust_boundary_requires_representative_database_preflight() -> None: receipt = _verified_receipt() receipt["worker_representative_db_preflight_passed"] = "false" receipt["worker_table_privilege_gap_count"] = "1" receipt["representative_db_preflights_verified"] = "false" readback = build_executor_trust_boundary_readback( receipt, deployed_source_sha="abc123", ) database_boundary = readback["database_identity_boundary"] assert database_boundary["db_identity_isolated"] is True assert database_boundary["connection_budget_verified"] is False assert "worker_representative_db_preflight_not_verified" in ( database_boundary["active_blockers"] ) def test_executor_trust_boundary_rejects_static_budget_without_concurrency_probe() -> None: receipt = _verified_receipt() receipt["api_concurrent_connection_probe_count"] = "4" readback = build_executor_trust_boundary_readback( receipt, deployed_source_sha="abc123", ) database_boundary = readback["database_identity_boundary"] assert database_boundary["connection_budget_verified"] is False assert "api_representative_db_preflight_not_verified" in ( database_boundary["active_blockers"] ) def test_executor_trust_boundary_requires_api_http_concurrency_probe() -> None: receipt = _verified_receipt() receipt["api_http_concurrency_probe_passed"] = "false" readback = build_executor_trust_boundary_readback( receipt, deployed_source_sha="abc123", ) database_boundary = readback["database_identity_boundary"] assert database_boundary["connection_budget_verified"] is False assert "api_http_concurrency_probe_not_verified" in ( database_boundary["active_blockers"] )