async def attest_working_profile(conn, schema: str, guard: WorkingEndpointGuard,
*, publication: bool) -> None:
"""Fail closed on any extra event hook, changed owner, or role escalation."""
if (type(guard) is not WorkingEndpointGuard or guard.schema != schema
or type(guard.working_host_role) is not str):
raise MemoryContractError("edge publication Working profile binding mismatch")
namespace = '"' + schema.replace('"', '""') + '"'
ddl = WORKING_GUARD_DDL.replace('__GUARD_SCHEMA__', namespace)
try:
await attest_working_guard(conn, schema, namespace, ddl, guard.working_host_role)
except RuntimeError as exc:
raise MemoryContractError("edge publication Working profile: " + str(exc)) from exc
rows = await conn.fetch(
"SELECT c.relname, t.tgname FROM pg_catalog.pg_class c "
"JOIN pg_catalog.pg_namespace n ON n.oid=c.relnamespace "
"LEFT JOIN pg_catalog.pg_trigger t ON t.tgrelid=c.oid AND NOT t.tgisinternal "
"WHERE n.nspname=$1 AND c.relname=ANY($2::text[])",
schema, ["memory_edges", "memory_nodes"],
)
expected = {name: set(names) for name, names in _WORKING_TRIGGERS.items()}
if publication:
expected["memory_edges"].add("symfonic_edge_publication_guard")
actual = {name: set() for name in expected}
for row in rows:
if row["tgname"] is not None:
actual[row["relname"]].add(row["tgname"])
if actual != expected or len(rows) != sum(map(len, expected.values())):
raise MemoryContractError("edge publication Working profile trigger inventory changed")
rules = await conn.fetchval(
"SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_rewrite r "
"JOIN pg_catalog.pg_class c ON c.oid=r.ev_class "
"JOIN pg_catalog.pg_namespace n ON n.oid=c.relnamespace "
"WHERE n.nspname=$1 AND c.relname=ANY($2::text[]))",
schema, ["memory_edges", "memory_nodes"],
)
if rules:
raise MemoryContractError("edge publication Working profile rule inventory changed")
role = await conn.fetchrow(
"SELECT h.oid,h.rolsuper,h.rolcreaterole,h.rolcreatedb,h.rolreplication,"
"h.rolbypassrls,h.rolcanlogin,n.nspowner,n.oid AS namespace_oid "
"FROM pg_catalog.pg_roles h CROSS JOIN pg_catalog.pg_namespace n "
"WHERE h.rolname=$1 AND n.nspname=$2",
guard.working_host_role, schema,
)
if (role is None or role["rolsuper"] or role["rolcreaterole"]
or role["rolcreatedb"] or role["rolreplication"]
or role["rolbypassrls"] or not role["rolcanlogin"]
or await conn.fetchval("SELECT pg_catalog.pg_has_role($1::oid,$2::oid,'MEMBER')",
role["oid"], role["nspowner"])):
raise MemoryContractError("edge publication Working profile host role changed")
unsafe = await conn.fetchval(
"SELECT pg_catalog.has_schema_privilege($1::oid,$2::oid,'CREATE') OR "
"EXISTS(SELECT 1 FROM unnest(ARRAY['memory_edges','memory_nodes']) name "
"CROSS JOIN unnest(ARRAY['TRIGGER','TRUNCATE']) privilege "
"WHERE pg_catalog.has_table_privilege($1::oid,"
"pg_catalog.to_regclass(pg_catalog.format('%I.%I',$3::text,name)),privilege))",
role["oid"], role["namespace_oid"], schema,
)
if unsafe:
raise MemoryContractError("edge publication Working profile unsafe role grants")
if publication:
owner = await conn.fetchval(
"SELECT p.proowner FROM pg_catalog.pg_proc p "
"WHERE p.oid=pg_catalog.to_regprocedure($1)",
namespace + ".symfonic_edge_publication_guard()",
)
if owner != role["nspowner"]:
raise MemoryContractError("edge publication Working profile function owner changed")