symfonic.memory.backends.postgres_staged_roles

postgres_staged_roles

Opt-in PostgreSQL authority boundary between a staged turn writer and its publisher.

The claim fence trigger honours a transaction setting that any session may set, so by itself it is a trusted-host contract. This installer makes PostgreSQL the authority for one named, non-owner writer role:

• memory_nodes: the writer may INSERT a row, and UPDATE a row, only while that row is pending and carries a selected claim, before and after the statement. It cannot release a row, cannot overwrite a published row and cannot DELETE, so DELETE+INSERT of a published row is refused as well.
• memory_operation_receipts: the writer may INSERT and UPDATE write receipts only. A FLUSH receipt, and so ACTIVE, is out of its reach.
• memory_invocation_staged_proofs (CORE-339): the writer and PUBLIC have no access; the publisher is granted SELECT and INSERT. A COMMITTED FLUSH receipt the writer forges while the boundary is down has no proof, so it never reads as ACTIVE after the boundary returns.
• memory_erase_jobs (CORE-199 S1), when present: SELECT only for both roles, so both pass the native erase fence and neither can write a job (D6); neither may hold anything on memory_erase_receipts.
• edges and vectors are read-only for the writer; it holds no TRUNCATE, REFERENCES or TRIGGER privilege, and the installer refuses a writer that could run DDL or assume another role.

Every other role, the publisher included, keeps unrestricted row access, so the publisher is the only named role that can activate; any further role an operator grants these tables can as well. ensure_schema never calls this module: a default or staged journal without it behaves exactly as before.

Not covered: the writer can still set the claim setting and so change the claim, id or scope of a pending claimed row (it stays pending and claimed); it writes its own journal rows, and can spend an operation id as a write receipt to block a FLUSH. None of these releases a row or commits a FLUSH receipt. Views, SECURITY DEFINER functions or policies an operator adds over these tables are outside the attestation, except foreign policies, which it refuses.

Policy bodies are attested against a baseline the installer records, in the same transaction, as each policy's comment: a server-side fingerprint of its command, mode, roles and deparsed USING and WITH CHECK. Only the table owner can alter a policy or its comment, so the writer can read the baseline but not edit it. ALTER POLICY keeps the old comment and so mismatches; a dropped and recreated policy has none. Both refuse, as does a deparse change after a server upgrade, until install records the baseline again. The owner itself, and any superuser, is trusted: it bypasses row security in any case.

install async

install(conn: Any, *, writer: str, publisher: str) -> None

Install and attest the boundary in one transaction; run as the table owner.

Requires the staged journal schema (and so the claim fence) to exist. The roles must already exist; creating roles is a deployment decision.

Source code in src/symfonic/memory/backends/postgres_staged_roles.py

async def install(conn: Any, *, writer: str, publisher: str) -> None:
    """Install and attest the boundary in one transaction; run as the table owner.

    Requires the staged journal schema (and so the claim fence) to exist. The
    roles must already exist; creating roles is a deployment decision.
    """
    w, p = _roles(writer, publisher)
    async with conn.transaction():
        await fence.require_installed(conn)
        await _require_sound(conn, w, p)
        for statement in statements(w, p):
            await conn.execute(statement)
        if jobs := await erase.jobs_table(conn):
            await erase.grant_read_only(conn, jobs, w)
            await erase.grant_read_only(conn, jobs, p)
        await _record_baseline(conn)
        await require_boundary(conn, w, p)

require_boundary async

require_boundary(conn: Any, writer: str, publisher: str) -> None

Attest the installed boundary exactly; any drift refuses.

Source code in src/symfonic/memory/backends/postgres_staged_roles.py

async def require_boundary(conn: Any, writer: str, publisher: str) -> None:
    """Attest the installed boundary exactly; any drift refuses."""
    w, p = _roles(writer, publisher)
    await fence.require_installed(conn)
    await _require_sound(conn, w, p)
    for table, granted in WRITER_GRANTS.items():
        for privilege in _PRIVILEGES:
            check = ("has_any_column_privilege" if privilege in _COLUMN_PRIVILEGES
                     else "has_table_privilege")
            held = await conn.fetchval(
                f"SELECT {check}($1::name, $2::regclass, $3::text)", w, table, privilege)
            if held is not (privilege in granted):
                _refuse(f"writer privilege {privilege} on {table} drifted")
    for table in WRITER_ROWS:
        secured = await conn.fetchval(
            "SELECT relrowsecurity FROM pg_class WHERE oid = $1::regclass", table)
        rows = await conn.fetch(_ATTEST, table, w)
        if secured is not True or sorted(row["polname"] for row in rows) != sorted(POLICIES):
            _refuse(f"row security on {table} is absent or carries foreign policies")
        for row in rows:
            _require_policy(table, row)
    await _require_jobs_readable(conn, w, p)

statements

statements(writer: str, publisher: str) -> tuple[str, ...]

The grants and policies, in order; names are validated, never quoted.

Source code in src/symfonic/memory/backends/postgres_staged_roles.py

def statements(writer: str, publisher: str) -> tuple[str, ...]:
    """The grants and policies, in order; names are validated, never quoted."""
    w, p = _roles(writer, publisher)
    # PostgreSQL cannot store a regrole constant in a policy expression.
    # _require_sound rejects writer role memberships, so current_user is exact.
    is_writer = f"(current_user = '{w}')"
    out = [f"SELECT pg_advisory_xact_lock({_LOCK})",
           f"REVOKE ALL ON TABLE {proof.PROOFS} FROM PUBLIC"]
    for table, privileges in WRITER_GRANTS.items():
        out.append(f"REVOKE ALL ON TABLE {table} FROM {w}")
        if privileges:
            out.append(f"GRANT {', '.join(privileges)} ON TABLE {table} TO {w}")
    for table, privileges in PUBLISHER_GRANTS.items():
        out.append(f"GRANT {', '.join(privileges)} ON TABLE {table} TO {p}")
    for table, rows in WRITER_ROWS.items():
        out.append(f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY")
        out.extend(f"DROP POLICY IF EXISTS {policy} ON {table}" for policy in POLICIES)
        out.extend((
            f"CREATE POLICY {POLICIES[0]} ON {table} FOR ALL TO PUBLIC "
            f"USING (NOT {is_writer}) WITH CHECK (NOT {is_writer})",
            f"CREATE POLICY {POLICIES[1]} ON {table} FOR SELECT TO {w} USING (true)",
            f"CREATE POLICY {POLICIES[2]} ON {table} FOR INSERT TO {w} WITH CHECK {rows}",
            f"CREATE POLICY {POLICIES[3]} ON {table} FOR UPDATE TO {w} "
            f"USING {rows} WITH CHECK {rows}",
        ))
    return tuple(out)

uninstall async

uninstall(conn: Any) -> None

Drop the policies and row security; table grants are left to the operator.

Source code in src/symfonic/memory/backends/postgres_staged_roles.py

async def uninstall(conn: Any) -> None:
    """Drop the policies and row security; table grants are left to the operator."""
    async with conn.transaction():
        await conn.execute(f"SELECT pg_advisory_xact_lock({_LOCK})")
        for table in WRITER_ROWS:
            for policy in POLICIES:
                await conn.execute(f"DROP POLICY IF EXISTS {policy} ON {table}")
            await conn.execute(f"ALTER TABLE {table} DISABLE ROW LEVEL SECURITY")