awoooi/apps/api/tests/test_executor_trust_boundary_readback.py

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",
        "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": "4",
        "worker_db_connection_limit": "2",
        "broker_db_connection_limit": "2",
        "api_required_connection_budget": "4",
        "worker_required_connection_budget": "2",
        "broker_required_connection_budget": "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_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": 8,
        "required_total": 8,
    }
    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


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"]
    )