791 lines
32 KiB
Python
791 lines
32 KiB
Python
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
|