import hashlib import json import os import stat import subprocess from collections.abc import Mapping from pathlib import Path import pytest from ops import verify_gcp_leoclean_runtime_permissions as verifier RUN_ID = "20260715-leo-security" SOURCE_REF = f"leo-runtime-permission:{RUN_ID}" SECRET_VALUE = "unit-only-secret-value%42" AGENT_ID = "11111111-1111-4111-8111-111111111111" class FakeRunner: def __init__( self, *, sqlstate_override: tuple[str, str] | None = None, sqlstate_missing_for: str | None = None, unexpected_success_for: str | None = None, scoped_failure: bool = False, administrator_mode: str = "denied", identity_override: Mapping[str, object] | None = None, role_posture_override: Mapping[str, object] | None = None, stage_owner_role_posture_override: Mapping[str, object] | None = None, function_posture_override: Mapping[str, Mapping[str, object]] | None = None, stage_definition_override: Mapping[str, object] | None = None, catalog_posture_override: Mapping[str, object] | None = None, ) -> None: self.sqlstate_override = sqlstate_override self.sqlstate_missing_for = sqlstate_missing_for self.unexpected_success_for = unexpected_success_for self.scoped_failure = scoped_failure self.administrator_mode = administrator_mode self.identity_override = dict(identity_override or {}) self.role_posture_override = dict(role_posture_override or {}) self.stage_owner_role_posture_override = dict(stage_owner_role_posture_override or {}) self.function_posture_override = { name: dict(values) for name, values in (function_posture_override or {}).items() } self.stage_definition_override = dict(stage_definition_override or {}) self.catalog_posture_override = dict(catalog_posture_override or {}) self.calls: list[tuple[list[str], dict[str, str], int, bool]] = [] self.negative = {check.sql: check for check in verifier._negative_checks(RUN_ID, SOURCE_REF)} def __call__( self, command: list[str], env: Mapping[str, str], timeout: int, discard_stdout: bool, ) -> verifier.CommandResult: self.calls.append((list(command), dict(env), timeout, discard_stdout)) if command[0] == verifier.GCLOUD_BIN: scoped = f"--secret={verifier.SCOPED_PASSWORD_SECRET}" in command if scoped: if self.scoped_failure: return verifier.CommandResult(1, SECRET_VALUE, f"scoped failure {SECRET_VALUE}") return verifier.CommandResult(0, f"{SECRET_VALUE}\n", f"ignored warning {SECRET_VALUE}") assert f"--secret={verifier.ADMINISTRATOR_PASSWORD_SECRET}" in command assert discard_stdout is True if self.administrator_mode == "success": return verifier.CommandResult(0, SECRET_VALUE, f"ignored {SECRET_VALUE}") if self.administrator_mode == "wrong-error": return verifier.CommandResult(1, SECRET_VALUE, f"network unavailable {SECRET_VALUE}") return verifier.CommandResult( 1, SECRET_VALUE, ( "ERROR: PERMISSION_DENIED: Permission " f"'{verifier.IAM_PERMISSION}' denied for the administrator secret; {SECRET_VALUE}" ), ) assert command[0] == verifier.PSQL_BIN assert discard_stdout is False sql = next(part.removeprefix("--command=") for part in command if part.startswith("--command=")) if sql == verifier._identity_sql(): identity: dict[str, object] = { "current_user": verifier.RUNTIME_DATABASE_ROLE, "database": verifier.CANONICAL_DATABASE, "leak": SECRET_VALUE, "server_addr": verifier.PRIVATE_CLOUDSQL_HOST, "server_port": verifier.PRIVATE_CLOUDSQL_PORT, "session_user": verifier.RUNTIME_DATABASE_ROLE, "ssl": True, "ssl_version": "TLSv1.3", "system_identifier": verifier.EXPECTED_SYSTEM_IDENTIFIER, } identity.update(self.identity_override) return verifier.CommandResult(0, json.dumps(identity), f"ignored {SECRET_VALUE}") if sql == verifier._allowed_read_sql(): return verifier.CommandResult( 0, json.dumps( { "claims_rows_sampled": 1, "leak": SECRET_VALUE, "proposal_rows_sampled": 0, } ), f"ignored {SECRET_VALUE}", ) if sql == verifier._role_posture_sql(): role_posture: dict[str, object] = { "bypasses_rls": False, "can_create_db": False, "can_create_role": False, "can_login": True, "can_replicate": False, "connection_limit": 8, "direct_membership_edges": 0, "inherits_privileges": False, "is_superuser": False, "leak": SECRET_VALUE, "role": verifier.RUNTIME_DATABASE_ROLE, } role_posture.update(self.role_posture_override) return verifier.CommandResult(0, json.dumps(role_posture), f"ignored {SECRET_VALUE}") if sql == verifier._stage_owner_role_posture_sql(): owner_posture: dict[str, object] = { "bypasses_rls": False, "can_create_db": False, "can_create_role": False, "can_login": False, "can_replicate": False, "connection_limit": -1, "direct_membership_edges": 0, "inherits_privileges": False, "is_superuser": False, "leak": SECRET_VALUE, "role": verifier.STAGE_OWNER_DATABASE_ROLE, } owner_posture.update(self.stage_owner_role_posture_override) return verifier.CommandResult(0, json.dumps(owner_posture), f"ignored {SECRET_VALUE}") if sql == verifier._function_privilege_posture_sql(): function_posture: dict[str, dict[str, object]] = { name: { "execute": expected_execute, "exists": expected_exists, "leak": SECRET_VALUE, } for name, _signature, expected_exists, expected_execute in verifier.FUNCTION_PRIVILEGE_EXPECTATIONS } for name, override in self.function_posture_override.items(): function_posture.setdefault(name, {}).update(override) function_posture["unexpected_leak"] = {"value": SECRET_VALUE} return verifier.CommandResult(0, json.dumps(function_posture), f"ignored {SECRET_VALUE}") if sql == verifier._stage_function_definition_sql(): stage_definition: dict[str, object] = { "acl_exact": True, "argument_modes": None, "configuration": ["search_path=pg_catalog, pg_temp"], "exists": True, "leak": SECRET_VALUE, "runtime_execute": True, "source": verifier.EXPECTED_STAGE_FUNCTION_SOURCE, **dict(verifier.STAGE_FUNCTION_METADATA_EXPECTATIONS), } stage_definition.update(self.stage_definition_override) return verifier.CommandResult(0, json.dumps(stage_definition), f"ignored {SECRET_VALUE}") if sql == verifier._catalog_privilege_posture_sql(): catalog_posture: dict[str, object] = {field: 0 for field in verifier.CATALOG_ZERO_COUNT_FIELDS} catalog_posture.update( { "leak": SECRET_VALUE, **{field: True for field in verifier.CATALOG_TRUE_FIELDS}, } ) catalog_posture.update(self.catalog_posture_override) return verifier.CommandResult(0, json.dumps(catalog_posture), f"ignored {SECRET_VALUE}") if sql == verifier._canary_count_sql(SOURCE_REF): return verifier.CommandResult(0, "0\n", f"ignored {SECRET_VALUE}") if sql == verifier._stage_sql(RUN_ID, SOURCE_REF): return verifier.CommandResult( 0, json.dumps( { "agent_id_matches": True, "proposal": { "leak": SECRET_VALUE, "proposed_by_agent_id": AGENT_ID, "proposed_by_handle": "leo", "source_ref": SOURCE_REF, "status": "pending_review", }, } ), f"ignored {SECRET_VALUE}", ) check = self.negative[sql] if check.name == self.unexpected_success_for: return verifier.CommandResult(0, SECRET_VALUE, f"unexpected success {SECRET_VALUE}") if check.name == self.sqlstate_missing_for: return verifier.CommandResult(1, SECRET_VALUE, f"denied without state {SECRET_VALUE}") if check.expected_connection_denial: return verifier.CommandResult( 2, SECRET_VALUE, ( f'connection failed: FATAL: permission denied for database "{check.database}"\n' f"DETAIL: User does not have CONNECT privilege. {SECRET_VALUE}" ), ) observed = check.expected_sqlstate if self.sqlstate_override is not None and check.name == self.sqlstate_override[0]: observed = self.sqlstate_override[1] return verifier.CommandResult( 1, SECRET_VALUE, f"ERROR: {observed}: expected test denial; {SECRET_VALUE}\nLOCATION: test", ) def poisoned_environment() -> dict[str, str]: return { "CLOUDSDK_AUTH_ACCESS_TOKEN": "must-not-reach-gcloud", "GOOGLE_APPLICATION_CREDENTIALS": "/tmp/must-not-reach-gcloud.json", "HOME": "/home/teleo", "LANG": "C.UTF-8", "LD_PRELOAD": "/tmp/must-not-load.so", "OTHER": "preserved", "PGDATABASE": "wrong-db", "PGHOST": "public.example.invalid", "PGPASSWORD": "administrator-password", "PGSERVICE": "broad-service", "pgsslmode": "disable", } def run_success(runner: FakeRunner) -> dict[str, object]: return verifier.verify_runtime_permissions( RUN_ID, runner=runner, base_env=poisoned_environment(), effective_user="teleo", dependency_validator=lambda: None, ) def assert_child_boundaries(runner: FakeRunner) -> None: assert runner.calls for command, env, timeout, discard_stdout in runner.calls: assert SECRET_VALUE not in "\n".join(command) assert timeout == verifier.COMMAND_TIMEOUT_SECONDS pg_keys = sorted(key for key in env if key.upper().startswith("PG")) if command[0] == verifier.PSQL_BIN: assert pg_keys == ["PGPASSWORD", "PGSSLMODE", "PGSSLROOTCERT"] assert env["PGPASSWORD"] == SECRET_VALUE assert env["PGSSLMODE"] == "verify-ca" assert env["PGSSLROOTCERT"] == str(verifier.SERVER_CA_PATH) assert discard_stdout is False else: assert pg_keys == [] assert "CLOUDSDK_AUTH_ACCESS_TOKEN" not in env assert "GOOGLE_APPLICATION_CREDENTIALS" not in env assert env["HOME"] == "/home/teleo" assert env["LANG"] == "C" assert env["LC_ALL"] == "C" assert env["PATH"] == verifier.CHILD_PATH assert "LD_PRELOAD" not in env assert "OTHER" not in env def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments() -> None: runner = FakeRunner() receipt = run_success(runner) serialized = verifier.canonical_json(receipt) assert receipt["status"] == "pass" assert receipt["current_tier"] == verifier.REQUIRED_TIER assert receipt["database_identity"] == { "current_user": verifier.RUNTIME_DATABASE_ROLE, "database": verifier.CANONICAL_DATABASE, "server_addr": verifier.PRIVATE_CLOUDSQL_HOST, "server_port": verifier.PRIVATE_CLOUDSQL_PORT, "session_user": verifier.RUNTIME_DATABASE_ROLE, "ssl": True, "ssl_version": "TLSv1.3", "system_identifier": verifier.EXPECTED_SYSTEM_IDENTIFIER, } assert receipt["checks"]["rolled_back_proposal"] == { "agent_id_matches": True, "proposed_by_agent_id": AGENT_ID, "proposed_by_handle": "leo", "source_ref": SOURCE_REF, "status": "pending_review", "transaction": "rolled_back", } assert receipt["checks"]["canary_rows_before"] == 0 assert receipt["checks"]["canary_rows_after"] == 0 assert receipt["checks"]["role_posture"] == { "bypasses_rls": False, "can_create_db": False, "can_create_role": False, "can_login": True, "can_replicate": False, "connection_limit": 8, "direct_membership_edges": 0, "inherits_privileges": False, "is_superuser": False, "role": verifier.RUNTIME_DATABASE_ROLE, } assert receipt["checks"]["stage_owner_role_posture"] == { "bypasses_rls": False, "can_create_db": False, "can_create_role": False, "can_login": False, "can_replicate": False, "connection_limit": -1, "direct_membership_edges": 0, "inherits_privileges": False, "is_superuser": False, "role": verifier.STAGE_OWNER_DATABASE_ROLE, } assert receipt["checks"]["function_privileges"] == { name: { "execute": expected_execute, "exists": expected_exists, "signature": signature, } for name, signature, expected_exists, expected_execute in verifier.FUNCTION_PRIVILEGE_EXPECTATIONS } assert receipt["checks"]["catalog_privileges"] == { **{field: 0 for field in verifier.CATALOG_ZERO_COUNT_FIELDS}, **{field: True for field in verifier.CATALOG_TRUE_FIELDS}, } assert receipt["checks"]["stage_function_definition"] == { "acl_exact": True, "argument_modes": None, "configuration": ["search_path=pg_catalog, pg_temp"], "exists": True, **dict(verifier.STAGE_FUNCTION_METADATA_EXPECTATIONS), "runtime_execute": True, "signature": verifier.STAGE_FUNCTION_SIGNATURE, "source_sha256": verifier.EXPECTED_STAGE_FUNCTION_SOURCE_SHA256, } assert len(receipt["checks"]["negative_permissions"]) == len(verifier._negative_checks(RUN_ID, SOURCE_REF)) assert receipt["secret_access"]["administrator"]["classification"] == "iam_permission_denied" assert receipt["secret_access"]["administrator"]["stdout_discarded"] is True assert SECRET_VALUE not in serialized assert "administrator-password" not in serialized assert "stage_leoclean_proposal is restricted" not in serialized assert_child_boundaries(runner) administrator_call = runner.calls[-1] assert administrator_call[0][0] == verifier.GCLOUD_BIN assert administrator_call[3] is True @pytest.mark.parametrize( ("attribute", "unsafe_value"), [ ("can_login", False), ("is_superuser", True), ("can_create_db", True), ("can_create_role", True), ("inherits_privileges", True), ("can_replicate", True), ("bypasses_rls", True), ("connection_limit", 7), ], ) def test_any_unsafe_runtime_role_attribute_fails_closed(attribute: str, unsafe_value: object) -> None: runner = FakeRunner(role_posture_override={attribute: unsafe_value}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "runtime_role_posture_mismatch" assert caught.value.check == "runtime_role_posture" assert SECRET_VALUE not in str(caught.value) def test_any_direct_runtime_role_membership_edge_fails_closed() -> None: runner = FakeRunner(role_posture_override={"direct_membership_edges": 1}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "runtime_role_posture_mismatch" assert caught.value.check == "runtime_role_posture" @pytest.mark.parametrize( ("attribute", "unsafe_value"), [ ("role", "unexpected-owner"), ("can_login", True), ("is_superuser", True), ("can_create_db", True), ("can_create_role", True), ("inherits_privileges", True), ("can_replicate", True), ("bypasses_rls", True), ("connection_limit", 0), ("direct_membership_edges", 1), ], ) def test_any_unsafe_stage_owner_role_posture_fails_closed(attribute: str, unsafe_value: object) -> None: runner = FakeRunner(stage_owner_role_posture_override={attribute: unsafe_value}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "stage_owner_role_posture_mismatch" assert caught.value.check == "stage_owner_role_posture" assert SECRET_VALUE not in str(caught.value) @pytest.mark.parametrize( "function_name", ["approve_strict_proposal", "assert_approved_proposal", "finish_approved_proposal"], ) def test_reviewer_or_apply_execute_grant_fails_closed(function_name: str) -> None: runner = FakeRunner(function_posture_override={function_name: {"execute": True}}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "function_privilege_posture_mismatch" assert caught.value.check == "function_privilege_posture" @pytest.mark.parametrize( "function_name", [ "stage_leoclean_proposal_5_arg", "approve_strict_proposal", "assert_approved_proposal", "finish_approved_proposal", ], ) def test_missing_expected_exact_function_fails_closed(function_name: str) -> None: runner = FakeRunner(function_posture_override={function_name: {"execute": False, "exists": False}}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "function_privilege_posture_mismatch" assert caught.value.check == "function_privilege_posture" def test_forged_overload_presence_or_missing_stage_execute_fails_closed() -> None: forged = FakeRunner(function_posture_override={"stage_leoclean_proposal_6_arg": {"exists": True}}) missing_execute = FakeRunner(function_posture_override={"stage_leoclean_proposal_5_arg": {"execute": False}}) for runner in (forged, missing_execute): with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "function_privilege_posture_mismatch" def _drifted_value(expected: object) -> object: if isinstance(expected, bool): return not expected if isinstance(expected, int): return expected + 1 if isinstance(expected, str): return f"{expected}-drift" if isinstance(expected, list): return [*expected, "drift"] raise AssertionError(f"unsupported test expectation type: {type(expected).__name__}") @pytest.mark.parametrize( ("field", "unsafe_value"), [(field, _drifted_value(expected)) for field, expected in verifier.STAGE_FUNCTION_METADATA_EXPECTATIONS], ) def test_any_stage_function_metadata_drift_fails_closed(field: str, unsafe_value: object) -> None: runner = FakeRunner(stage_definition_override={field: unsafe_value}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "stage_function_definition_mismatch" assert caught.value.check == "stage_function_definition" assert SECRET_VALUE not in str(caught.value) @pytest.mark.parametrize( ("field", "unsafe_value"), [ ("exists", False), ("argument_modes", ["i", "i", "i", "i", "i"]), ("configuration", ["search_path=public"]), ("acl_exact", False), ("runtime_execute", False), ("leakproof", 0), ("argument_defaults", False), ], ) def test_any_stage_function_contract_drift_fails_closed(field: str, unsafe_value: object) -> None: runner = FakeRunner(stage_definition_override={field: unsafe_value}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "stage_function_definition_mismatch" assert caught.value.check == "stage_function_definition" @pytest.mark.parametrize( "unsafe_source", [ None, 42, verifier.EXPECTED_STAGE_FUNCTION_SOURCE.replace("return to_jsonb(staged);", "return '{}'::jsonb;"), "x" * (verifier.MAX_STAGE_FUNCTION_SOURCE_BYTES + 1), ], ) def test_missing_malformed_oversized_or_semantically_drifted_stage_body_fails_closed( unsafe_source: object, ) -> None: runner = FakeRunner(stage_definition_override={"source": unsafe_source}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "stage_function_definition_mismatch" assert caught.value.check == "stage_function_definition" failure = verifier.canonical_json(verifier.failure_receipt(RUN_ID, caught.value)) assert "stage_leoclean_proposal is restricted" not in failure assert SECRET_VALUE not in failure def test_stage_body_normalization_accepts_only_line_ending_and_framing_newline_changes() -> None: crlf_source = "\r\n" + verifier.EXPECTED_STAGE_FUNCTION_SOURCE.strip("\n").replace("\n", "\r\n") + "\r\n" receipt = run_success(FakeRunner(stage_definition_override={"source": crlf_source})) assert ( receipt["checks"]["stage_function_definition"]["source_sha256"] == verifier.EXPECTED_STAGE_FUNCTION_SOURCE_SHA256 ) def test_embedded_stage_body_attestation_is_tied_to_the_provisioning_sql_source() -> None: provisioning_sql = (Path(__file__).resolve().parents[1] / "ops" / "gcp_leoclean_runtime_role.sql").read_text( encoding="utf-8" ) function_anchor = "create or replace function kb_stage.stage_leoclean_proposal(" assert provisioning_sql.count(function_anchor) == 1 function_section = provisioning_sql.split(function_anchor, 1)[1] assert function_section.count("as $function$") == 1 provisioned_source = function_section.split("as $function$", 1)[1].split("$function$;", 1)[0] assert provisioned_source == verifier.EXPECTED_STAGE_FUNCTION_SOURCE assert ( hashlib.sha256(verifier.normalize_stage_function_source(provisioned_source).encode("utf-8")).hexdigest() == verifier.EXPECTED_STAGE_FUNCTION_SOURCE_SHA256 ) def test_catalog_queries_use_exact_regprocedure_signatures_and_both_membership_directions() -> None: function_sql = verifier._function_privilege_posture_sql() role_sql = verifier._role_posture_sql() owner_role_sql = verifier._stage_owner_role_posture_sql() stage_definition_sql = verifier._stage_function_definition_sql() catalog_sql = verifier._catalog_privilege_posture_sql() assert "pg_catalog.to_regprocedure(signature)" in function_sql for _name, signature, _exists, _execute in verifier.FUNCTION_PRIVILEGE_EXPECTATIONS: assert signature in function_sql assert "membership.roleid = role_row.oid" in role_sql assert "membership.member = role_row.oid" in role_sql assert "membership.roleid = role_row.oid" in owner_role_sql assert "membership.member = role_row.oid" in owner_role_sql assert verifier.STAGE_OWNER_DATABASE_ROLE in owner_role_sql assert verifier.STAGE_FUNCTION_SIGNATURE in stage_definition_sql assert "left join pg_catalog.pg_proc" in stage_definition_sql assert "function_catalog.prosrc" in stage_definition_sql assert "function_catalog.prokind" in stage_definition_sql assert "function_catalog.proargtypes" in stage_definition_sql assert "function_catalog.prosecdef" in stage_definition_sql assert "pg_catalog.pg_get_function_result" in stage_definition_sql assert "pg_catalog.aclexplode" in stage_definition_sql assert "pg_catalog.pg_shdepend" in catalog_sql assert "pg_catalog.has_table_privilege" in catalog_sql assert "pg_catalog.has_column_privilege" in catalog_sql assert "pg_catalog.has_sequence_privilege" in catalog_sql assert "pg_catalog.has_function_privilege" in catalog_sql assert "pg_catalog.aclexplode" in catalog_sql assert "nspname !~ '^pg_'" in catalog_sql assert "nspname <> 'information_schema'" in catalog_sql assert verifier.STAGE_OWNER_DATABASE_ROLE in catalog_sql assert "public_database_connect_grants" in catalog_sql assert "public_large_object_mutation_routine_execute" in catalog_sql assert "'lo_creat'" in catalog_sql assert "'lo_open'" in catalog_sql for schema, relation in verifier.READ_TABLE_ALLOWLIST: assert schema in catalog_sql assert relation in catalog_sql @pytest.mark.parametrize("field", verifier.CATALOG_ZERO_COUNT_FIELDS) def test_any_unexpected_catalog_privilege_fails_closed(field: str) -> None: runner = FakeRunner(catalog_posture_override={field: 1}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "catalog_privilege_posture_mismatch" assert caught.value.check == "catalog_privilege_posture" assert SECRET_VALUE not in str(caught.value) @pytest.mark.parametrize("field", verifier.CATALOG_TRUE_FIELDS) def test_any_missing_required_catalog_privilege_fails_closed(field: str) -> None: runner = FakeRunner(catalog_posture_override={field: False}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "catalog_privilege_posture_mismatch" def test_legacy_and_template_database_connect_are_both_behaviorally_denied() -> None: runner = FakeRunner() receipt = run_success(runner) denials = {check["check"]: check for check in receipt["checks"]["negative_permissions"]} for name in ("legacy_database_connect", "template_database_connect"): assert denials[name] == { "check": name, "expected_sqlstate": "42501", "observed_error_class": "connect_privilege_denied", "observed_sqlstate": "not_exposed_by_libpq_startup", "result": "denied", } database_calls = [ " ".join(command) for command, _env, _timeout, _discard in runner.calls if command[0] == verifier.PSQL_BIN ] assert any(f"dbname={verifier.LEGACY_DATABASE}" in command for command in database_calls) assert any(f"dbname={verifier.TEMPLATE_DATABASE}" in command for command in database_calls) def test_required_sqlstate_mismatch_fails_closed_without_secret_in_error() -> None: runner = FakeRunner(sqlstate_override=("canonical_insert", "23505")) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) error = caught.value assert error.code == "negative_permission_sqlstate_mismatch" assert error.check == "canonical_insert" assert error.expected_sqlstate == "42501" assert error.observed_sqlstate == "23505" assert SECRET_VALUE not in str(error) assert SECRET_VALUE not in verifier.canonical_json(verifier.failure_receipt(RUN_ID, error)) assert_child_boundaries(runner) def test_missing_verbose_sqlstate_fails_closed_without_raw_stderr() -> None: runner = FakeRunner(sqlstate_missing_for="forged_proposer_overload") with pytest.raises(verifier.VerificationError) as caught: run_success(runner) error = caught.value assert error.code == "negative_permission_sqlstate_missing" assert error.expected_sqlstate == "42883" assert error.observed_sqlstate == "missing" assert SECRET_VALUE not in json.dumps(error.as_dict()) def test_unclassified_database_startup_failure_cannot_masquerade_as_connect_denial() -> None: runner = FakeRunner(sqlstate_missing_for="legacy_database_connect") with pytest.raises(verifier.VerificationError) as caught: run_success(runner) error = caught.value assert error.code == "negative_connection_denial_mismatch" assert error.check == "legacy_database_connect" assert error.observed_sqlstate == "startup_error_unclassified" assert SECRET_VALUE not in json.dumps(error.as_dict()) def test_unexpected_permission_success_fails_closed_and_transaction_sql_has_rollback() -> None: runner = FakeRunner(unexpected_success_for="proposal_update") with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "negative_permission_unexpectedly_succeeded" assert caught.value.observed_sqlstate == "success" proposal_update = next( check for check in verifier._negative_checks(RUN_ID, SOURCE_REF) if check.name == "proposal_update" ) assert proposal_update.sql.startswith("begin;\n") assert proposal_update.sql.endswith("\nrollback;") assert SECRET_VALUE not in str(caught.value) @pytest.mark.parametrize("administrator_mode", ["success", "wrong-error"]) def test_administrator_secret_must_be_exact_iam_denial_and_stdout_is_discarded(administrator_mode: str) -> None: runner = FakeRunner(administrator_mode=administrator_mode) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) expected_code = ( "administrator_secret_unexpectedly_readable" if administrator_mode == "success" else "administrator_secret_denial_not_iam_permission" ) assert caught.value.code == expected_code assert SECRET_VALUE not in str(caught.value) assert SECRET_VALUE not in verifier.canonical_json(verifier.failure_receipt(RUN_ID, caught.value)) assert runner.calls[-1][3] is True def test_scoped_secret_failure_never_repeats_command_output() -> None: runner = FakeRunner(scoped_failure=True) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "scoped_secret_access_failed" assert SECRET_VALUE not in str(caught.value) assert SECRET_VALUE not in verifier.canonical_json(verifier.failure_receipt(RUN_ID, caught.value)) assert len(runner.calls) == 1 @pytest.mark.parametrize( "invalid", ["short", "UPPERCASE-RUN", "contains_underscore", "contains.dot", " leading-space", "a" * 65], ) def test_run_id_validation_is_strict(invalid: str) -> None: with pytest.raises(ValueError, match="8-64 lowercase"): verifier.validate_run_id(invalid) def test_wrong_identity_or_non_ssl_connection_fails_before_permission_probes() -> None: runner = FakeRunner(identity_override={"ssl": False}) with pytest.raises(verifier.VerificationError) as caught: run_success(runner) assert caught.value.code == "private_ssl_identity_mismatch" assert len(runner.calls) == 2 assert SECRET_VALUE not in str(caught.value) def test_optional_receipt_is_canonical_mode_0600_and_refuses_symlink(tmp_path: Path) -> None: payload = {"z": 2, "a": {"safe": True}} output = tmp_path / "receipt.json" verifier.write_receipt(output, payload) assert output.read_text(encoding="utf-8") == verifier.canonical_json(payload) assert stat.S_IMODE(output.stat().st_mode) == 0o600 assert output.read_text(encoding="utf-8").startswith('{"a":') target = tmp_path / "target.json" target.write_text("untouched", encoding="utf-8") symlink = tmp_path / "symlink.json" symlink.symlink_to(target) with pytest.raises(OSError): verifier.write_receipt(symlink, payload) assert target.read_text(encoding="utf-8") == "untouched" def test_subprocess_runner_discards_administrator_stdout_at_file_descriptor(monkeypatch: pytest.MonkeyPatch) -> None: observed: dict[str, object] = {} def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: observed["command"] = command observed.update(kwargs) return subprocess.CompletedProcess(command, 1, stdout=None, stderr="PERMISSION_DENIED") monkeypatch.setattr(verifier.subprocess, "run", fake_run) result = verifier.subprocess_runner([verifier.GCLOUD_BIN, "test"], os.environ, 3, True) assert observed["stdout"] is subprocess.DEVNULL assert observed["stdin"] is subprocess.DEVNULL assert observed["stderr"] is subprocess.PIPE assert result.stdout == "" assert result.returncode == 1