async def attest_working_guard(
conn: Any, schema: str, quoted_schema: str, ddl: str, host_role: str,
) -> None:
"""Fail closed if an existing install no longer matches the admitted DDL."""
tables = await conn.fetch(
"SELECT c.relname, c.oid, c.relowner, c.relkind::text AS relkind, "
"c.relrowsecurity, c.relforcerowsecurity "
"FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace "
"WHERE n.nspname=$1 AND c.relname=ANY($2::text[])",
schema, ["memory_nodes", "memory_edges", "memory_working_endpoint_guard"],
)
table = {row["relname"]: row for row in tables}
if set(table) != {"memory_nodes", "memory_edges", "memory_working_endpoint_guard"}:
raise RuntimeError("Working guard catalog attestation failed: tables")
guard = table["memory_working_endpoint_guard"]
if any(row["relowner"] != guard["relowner"] for row in table.values()):
raise RuntimeError("Working guard catalog attestation failed: table ownership")
namespace = await conn.fetchrow(
"SELECT n.nspowner, n.nspacl FROM pg_namespace n WHERE n.nspname=$1", schema,
)
if namespace is None or namespace["nspowner"] != guard["relowner"]:
raise RuntimeError("Working guard catalog attestation failed: schema ownership")
schema_create = await conn.fetchval(
"SELECT EXISTS (SELECT 1 FROM pg_namespace n CROSS JOIN LATERAL "
"aclexplode(COALESCE(n.nspacl, acldefault('n', n.nspowner))) acl "
"WHERE n.nspname=$1 AND acl.grantee<>n.nspowner "
"AND acl.privilege_type='CREATE')", schema,
)
if schema_create:
raise RuntimeError("Working guard catalog attestation failed: schema grants")
if any(row["relkind"] != "r" for row in table.values()):
raise RuntimeError("Working guard catalog attestation failed: relation kind")
if any(row["relrowsecurity"] or row["relforcerowsecurity"] for row in table.values()):
raise RuntimeError("Working guard catalog attestation failed: row security")
grants = await conn.fetch(
"SELECT c.relname, acl.grantee, acl.privilege_type "
"FROM pg_class c CROSS JOIN LATERAL "
"aclexplode(COALESCE(c.relacl, acldefault('r', c.relowner))) acl "
"WHERE c.oid=ANY($1::oid[]) AND acl.grantee<>c.relowner "
"AND (c.relname='memory_working_endpoint_guard' "
"OR acl.privilege_type IN ('TRIGGER', 'TRUNCATE'))",
[row["oid"] for row in table.values()],
)
host_oid = await conn.fetchval(
"SELECT oid FROM pg_roles WHERE rolname=$1", host_role,
)
expected_grants = {
(host_oid, privilege)
for privilege in ("SELECT", "INSERT", "UPDATE")
}
guard_grants = {
(row["grantee"], row["privilege_type"])
for row in grants if row["relname"] == "memory_working_endpoint_guard"
}
if host_oid is None or guard_grants != expected_grants or any(
row["relname"] != "memory_working_endpoint_guard" for row in grants
):
raise RuntimeError("Working guard catalog attestation failed: unsafe table grants")
column_grants = await conn.fetch(
"SELECT a.attname, acl.grantee FROM pg_attribute a "
"CROSS JOIN LATERAL aclexplode(a.attacl) acl "
"WHERE a.attrelid=$1 AND a.attnum>0 AND NOT a.attisdropped "
"AND acl.grantee<>$2",
guard["oid"], guard["relowner"],
)
if column_grants:
raise RuntimeError("Working guard catalog attestation failed: column grants")
columns = await conn.fetch(
"SELECT a.attname, a.atttypid::regtype::text AS datatype, "
"a.attnotnull, col.collname "
"FROM pg_attribute a LEFT JOIN pg_collation col ON col.oid=a.attcollation "
"WHERE a.attrelid=$1 AND a.attnum>0 AND NOT a.attisdropped",
guard["oid"],
)
expected_columns = {
"tenant_id": ("text", True, "C"),
"record_id": ("text", True, "C"),
"scope_path": ("text", True, "C"),
"retired": ("boolean", True, None),
}
actual_columns = {
row["attname"]: (row["datatype"], row["attnotnull"], row["collname"])
for row in columns
}
if actual_columns != expected_columns:
raise RuntimeError("Working guard catalog attestation failed: columns")
primary = await conn.fetchval(
"SELECT pg_get_constraintdef(oid) FROM pg_constraint "
"WHERE conrelid=$1 AND contype='p'",
guard["oid"],
)
if primary != "PRIMARY KEY (tenant_id, record_id)":
raise RuntimeError("Working guard catalog attestation failed: primary key")
functions = await conn.fetch(
"SELECT p.oid, p.proname, p.prosrc, p.prosecdef, p.proconfig, p.proowner, l.lanname, "
"EXISTS (SELECT 1 FROM aclexplode(COALESCE(p.proacl, "
"acldefault('f',p.proowner))) acl "
"WHERE acl.grantee=0 AND acl.privilege_type='EXECUTE') AS public_execute "
"FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace "
"JOIN pg_language l ON l.oid=p.prolang "
"WHERE n.nspname=$1 AND p.proname=ANY($2::text[]) "
"AND p.pronargs=0",
schema, ["symfonic_working_edge_guard", "symfonic_working_node_guard"],
)
function = {row["proname"]: row for row in functions}
if set(function) != {"symfonic_working_edge_guard", "symfonic_working_node_guard"}:
raise RuntimeError("Working guard catalog attestation failed: functions")
for name, row in function.items():
if (
row["lanname"] != "plpgsql"
or not row["prosecdef"]
or row["proowner"] != guard["relowner"]
or row["public_execute"]
or list(row["proconfig"] or []) != ["search_path=pg_catalog, pg_temp"]
or row["prosrc"].strip() != _expected_body(ddl, quoted_schema, name)
):
raise RuntimeError("Working guard catalog attestation failed: function body")
triggers = await conn.fetch(
"SELECT t.tgname, t.tgtype, t.tgenabled::text AS tgenabled, "
"t.tgfoid, t.tgattr::text AS attrs, t.tgisinternal, "
"t.tgqual IS NULL AS no_when, t.tgconstraint, c.relname "
"FROM pg_trigger t JOIN pg_class c ON c.oid=t.tgrelid "
"JOIN pg_namespace n ON n.oid=c.relnamespace "
"WHERE n.nspname=$1 AND t.tgname=ANY($2::text[])",
schema, [
"symfonic_working_edge_guard_before_write",
"symfonic_working_node_guard_before_insert",
"symfonic_working_node_guard_before_change",
],
)
expected_triggers = {
"symfonic_working_edge_guard_before_write": (
"memory_edges", 31, function["symfonic_working_edge_guard"]["oid"],
{"tenant_id", "source", "target", "scope_path"},
),
"symfonic_working_node_guard_before_insert": (
"memory_nodes", 7, function["symfonic_working_node_guard"]["oid"], set(),
),
"symfonic_working_node_guard_before_change": (
"memory_nodes", 27, function["symfonic_working_node_guard"]["oid"],
{"id", "tenant_id", "scope_path", "layer"},
),
}
if {row["tgname"] for row in triggers} != set(expected_triggers):
raise RuntimeError("Working guard catalog attestation failed: triggers")
for row in triggers:
relation, kind, oid, names = expected_triggers[row["tgname"]]
attrs = await conn.fetch(
"SELECT attname, attnum FROM pg_attribute "
"WHERE attrelid=$1 AND attname=ANY($2::text[])",
table[relation]["oid"], list(names),
)
expected_numbers = {record["attnum"] for record in attrs}
actual_numbers = {int(number) for number in row["attrs"].split()}
if (
row["relname"] != relation
or row["tgtype"] != kind
or row["tgenabled"] != "A"
or row["tgfoid"] != oid
or row["tgisinternal"]
or not row["no_when"]
or row["tgconstraint"] != 0
or {record["attname"] for record in attrs} != names
or actual_numbers != expected_numbers
):
raise RuntimeError("Working guard catalog attestation failed: trigger wiring")