Migration-owner DDL; runtime must have SELECT/INSERT only on operator audit.
Do not run this migration with the disposition runtime role. The installer does
not grant deployment privileges or choose identity/key custody. A DBA retains
power to alter/drop audit objects: immutability is enforced against the checked
runtime role, not claimed against database administrators or backup loss.
install_pending_decision_schema
async
install_pending_decision_schema(migration_pool)
Explicit migration-owner operation, never called from request dispatch.
Source code in src/symfonic/memory/backends/postgres_pending_decision_schema.py
| async def install_pending_decision_schema(migration_pool):
"""Explicit migration-owner operation, never called from request dispatch."""
async with migration_pool.transaction() as conn:
await conn.execute(
"SELECT pg_advisory_xact_lock(2147483646, hashtext(current_schema()))")
await conn.execute(_DDL)
|
require_immutable_runtime
async
require_immutable_runtime(conn)
Refuse owner/superuser/DDL or mutable audit privileges before decision I/O.
Source code in src/symfonic/memory/backends/postgres_pending_decision_schema.py
| async def require_immutable_runtime(conn):
"""Refuse owner/superuser/DDL or mutable audit privileges before decision I/O."""
safe = await conn.fetchval("""
SELECT current_user=session_user AND NOT r.rolsuper AND NOT r.rolcreaterole
AND c.relkind='r' AND NOT pg_has_role(current_user,c.relowner,'MEMBER')
AND NOT has_schema_privilege(current_user,c.relnamespace,'CREATE')
AND has_table_privilege(current_user,c.oid,'SELECT')
AND has_table_privilege(current_user,c.oid,'INSERT')
AND NOT has_table_privilege(current_user,c.oid,'UPDATE')
AND NOT has_table_privilege(current_user,c.oid,'DELETE')
AND NOT has_table_privilege(current_user,c.oid,'TRUNCATE')
AND NOT has_table_privilege(current_user,c.oid,'TRIGGER')
AND (SELECT count(*) FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid
WHERE t.tgrelid=c.oid AND p.proowner=c.relowner
AND t.tgfoid=to_regprocedure('symfonic_pending_decision_immutable()')
AND NOT t.tgisinternal AND t.tgenabled='O'
AND ((t.tgname='pending_decision_immutable' AND t.tgtype=27)
OR (t.tgname='pending_decision_no_truncate' AND t.tgtype=34)))=2
FROM pg_class c JOIN pg_roles r ON r.rolname=current_user
WHERE c.oid=to_regclass('memory_pending_quick_decisions')
""")
if safe is not True:
raise MemoryContractError("operator audit requires immutable runtime/migration role split")
|