teleo-infrastructure/tests/test_gcp_leoclean_runtime_permissions.py

import hashlib
import json
import os
import re
import stat
import subprocess
from collections.abc import Mapping
from pathlib import Path

import pytest

from ops import derive_leoclean_runtime_query_contract as query_contract
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,
                "owner_execute": False,
                "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),
        "owner_execute": False,
        "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),
        ("owner_execute", True),
        ("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_query_minimal_runtime_read_columns_match_the_provisioning_manifest() -> None:
    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
    runtime_source = runtime_path.read_text(encoding="utf-8")
    expected = {(schema, relation): frozenset(columns) for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST}
    observed = query_contract.verify_runtime_read_columns(runtime_source, expected)

    assert observed == expected
    assert sum(len(columns) for columns in observed.values()) == 92
    assert "created_by" not in observed[("public", "claims")]
    assert "reviewed_by_agent_id" not in observed[("kb_stage", "kb_proposals")]
    assert "applied_by_agent_id" not in observed[("kb_stage", "kb_proposals")]

    provisioning_sql = (Path(__file__).resolve().parents[1] / "ops" / "gcp_leoclean_runtime_role.sql").read_text(
        encoding="utf-8"
    )
    manifest = provisioning_sql.split("insert into pg_temp.leoclean_runtime_read_column_allowlist (", 1)[1].split(
        ";", 1
    )[0]
    rows = re.findall(r"\('([^']+)', '([^']+)', '([^']+)', ([0-9]+)\)", manifest)
    manifest_columns: dict[tuple[str, str], list[tuple[int, str]]] = {}
    for schema, relation, column, ordinal in rows:
        manifest_columns.setdefault((schema, relation), []).append((int(ordinal), column))
    parsed = {
        relation: frozenset(column for _ordinal, column in columns) for relation, columns in manifest_columns.items()
    }

    assert len(rows) == 92
    assert parsed == expected
    assert not re.search(
        r"grant\s+select\s+on\b[^;]*\bto\s+leoclean_kb_runtime\b",
        provisioning_sql,
        re.IGNORECASE | re.DOTALL,
    )
    assert "leoclean_kb_runtime unexpectedly has table-level SELECT" in provisioning_sql
    assert "leoclean_kb_runtime is missing required column SELECT" in provisioning_sql


def test_runtime_query_contract_rejects_a_new_unallowlisted_column() -> None:
    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
    runtime_source = runtime_path.read_text(encoding="utf-8")
    expected = {(schema, relation): frozenset(columns) for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST}
    mutated = runtime_source.replace("         c.status,\n", "         c.status,\n         c.future_private_text,\n", 1)
    assert mutated != runtime_source

    with pytest.raises(query_contract.QueryContractError, match="runtime query surface differs"):
        query_contract.verify_runtime_read_columns(mutated, expected)


def test_runtime_query_contract_rejects_unqualified_canonical_columns() -> None:
    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
    runtime_source = runtime_path.read_text(encoding="utf-8")
    mutated = runtime_source.replace("         c.status,\n", "         c.status,\n         future_private_text,\n", 1)
    assert mutated != runtime_source

    with pytest.raises(query_contract.QueryContractError, match="unqualified SQL identifiers"):
        query_contract.derive_runtime_read_columns(mutated)


@pytest.mark.parametrize(
    ("anchor", "replacement", "error"),
    [
        (
            '    claims_sql = f"""\n',
            '    run_psql(args, "update public.claims set status = \'closed\'")\n    claims_sql = f"""\n',
            "direct canonical DML",
        ),
        (
            '    claims_sql = f"""\n',
            '    run_psql(args, "insert into public.claims (id) values (gen_random_uuid())")\n    claims_sql = f"""\n',
            "direct canonical DML",
        ),
        (
            '    claims_sql = f"""\n',
            '    run_psql(args, "truncate table public.claims")\n    claims_sql = f"""\n',
            "direct canonical DML",
        ),
        (
            "  select c.id,\n",
            "  select c.*,\n         c.id,\n",
            "canonical wildcard reads",
        ),
        (
            "  select c.id,\n",
            "  select to_jsonb(c) as hidden_row,\n         c.id,\n",
            "whole-row canonical alias",
        ),
        (
            "  select c.id,\n",
            "  select c.{getattr(args, 'column', 'status')} as dynamic_value,\n         c.id,\n",
            "dynamic SQL interpolation",
        ),
        (
            "  select c.id,\n",
            "  select future_private_text as future_private_text,\n         c.id,\n",
            "masked by an output alias",
        ),
    ],
    ids=(
        "update",
        "insert",
        "truncate",
        "qualified-wildcard",
        "whole-row-json",
        "dynamic-column",
        "self-alias-masking",
    ),
)
def test_runtime_query_contract_fails_closed_for_adversarial_query_mutations(
    anchor: str,
    replacement: str,
    error: str,
) -> None:
    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
    runtime_source = runtime_path.read_text(encoding="utf-8")
    expected = {(schema, relation): frozenset(columns) for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST}
    mutated = runtime_source.replace(anchor, replacement, 1)
    assert mutated != runtime_source

    with pytest.raises(query_contract.QueryContractError, match=error):
        query_contract.verify_runtime_read_columns(mutated, expected)


def test_runtime_query_contract_rejects_unqualified_dynamic_column_interpolation() -> None:
    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
    runtime_source = runtime_path.read_text(encoding="utf-8")
    mutated = runtime_source.replace(
        "  select c.id,\n",
        "  select {getattr(args, 'column', 'status')} as dynamic_value,\n         c.id,\n",
        1,
    )
    assert mutated != runtime_source

    with pytest.raises(query_contract.QueryContractError, match="not part of the reviewed runtime query surface"):
        query_contract.derive_runtime_read_columns(mutated)


def test_runtime_query_contract_rejects_alias_target_dml_across_block_comments() -> None:
    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
    runtime_source = runtime_path.read_text(encoding="utf-8")
    mutated = runtime_source.replace(
        '    claims_sql = f"""\n',
        "    run_psql(args, \"update /* canonical target */ c set status = 'closed' "
        'from public.claims as c")\n    claims_sql = f"""\n',
        1,
    )
    assert mutated != runtime_source

    with pytest.raises(query_contract.QueryContractError, match="direct canonical DML"):
        query_contract.derive_runtime_read_columns(mutated)


@pytest.mark.parametrize(
    "unsafe_interpolation",
    ("{min_score}", "{max(1, min(min_score, 2))}"),
    ids=("raw-name", "raw-max-expression"),
)
def test_runtime_query_contract_requires_an_explicit_sql_serializer(unsafe_interpolation: str) -> None:
    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
    runtime_source = runtime_path.read_text(encoding="utf-8")
    reviewed = "{sql_integer(min_score, minimum=1, maximum=2)}"
    mutated = runtime_source.replace(reviewed, unsafe_interpolation, 1)
    assert mutated != runtime_source

    with pytest.raises(query_contract.QueryContractError, match="not part of the reviewed runtime query surface"):
        query_contract.derive_runtime_read_columns(mutated)


def test_runtime_query_contract_rejects_dynamic_canonical_namespace_interpolation() -> None:
    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
    runtime_source = runtime_path.read_text(encoding="utf-8")
    mutated = runtime_source.replace(
        "  select c.id,\n",
        "  select dynamic_alias.id,\n    from {getattr(args, 'schema', 'public')}.claims as dynamic_alias;\n"
        "  select c.id,\n",
        1,
    )
    assert mutated != runtime_source

    with pytest.raises(query_contract.QueryContractError, match="not part of the reviewed runtime query surface"):
        query_contract.derive_runtime_read_columns(mutated)


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
    for _schema, _relation, columns in verifier.READ_COLUMN_ALLOWLIST:
        for column in columns:
            assert column 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