symfonic.memory.backends.postgres_runtime_roles

postgres_runtime_roles

Opt-in schema-migrator / DML-only runtime login split (CORE-127 candidate).

Import explicitly. Nothing here creates roles, sets passwords, terminates sessions, resolves prepared transactions or swaps indexes; those are operator acts. grant_runtime and revoke_runtime run as the schema owner; require_split_runtime runs read-only as the runtime; require_login_retired runs as an administrative observer. A passing probe is point-in-time evidence for the named login only, never a fleet barrier.

RuntimeRoleNotReady

Bases: RuntimeError

A login violates the migrator/runtime split contract.

grant_runtime async

grant_runtime(connection: Any, runtime_login: str, *, schema: str = 'public', ledger: str = MIGRATION_LEDGER) -> tuple[str, ...]

Owner side: reset the runtime to exact DML on every owned table, atomically.

Every table in schema gets SELECT/INSERT/UPDATE/DELETE and nothing else, column grants included; the CORE-199 erase job table gets SELECT only, its receipts nothing, its memory erase generations SELECT/INSERT/UPDATE and its erase log SELECT/INSERT; the migration ledger, views, materialized views and foreign tables get nothing; sequences get USAGE only. No default privileges are installed, so each migration must be followed by this call. Returns the tables granted DML.

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

async def grant_runtime(
    connection: Any, runtime_login: str, *, schema: str = "public", ledger: str = MIGRATION_LEDGER
) -> tuple[str, ...]:
    """Owner side: reset the runtime to exact DML on every owned table, atomically.

    Every table in ``schema`` gets SELECT/INSERT/UPDATE/DELETE and nothing else,
    column grants included; the CORE-199 erase job table gets SELECT only,
    its receipts nothing, its memory erase generations SELECT/INSERT/UPDATE and its
    erase log SELECT/INSERT; the
    migration ledger, views, materialized views and foreign tables get nothing;
    sequences get USAGE only. No default privileges are installed, so each
    migration must be followed by this call. Returns the tables granted DML.
    """
    runtime, quoted_schema = _identifier(runtime_login), _identifier(schema)
    async with connection.transaction():
        if await connection.fetchval("SELECT current_user=$1", runtime_login):
            raise RuntimeRoleNotReady("the owner cannot grant to itself")
        await _unprivileged(connection, runtime_login)
        rows = await _owned_schema(connection, schema)
        await connection.execute(f"REVOKE CREATE ON SCHEMA {quoted_schema} FROM {runtime}")
        await connection.execute(f"GRANT USAGE ON SCHEMA {quoted_schema} TO {runtime}")
        granted = []
        for row in rows:
            # Revoking a table's privileges also revokes its column privileges.
            await connection.execute(f"REVOKE ALL ON {_target(schema, row)} FROM {runtime}")
            if row["relkind"] == "S":
                await connection.execute(f"GRANT USAGE ON {_target(schema, row)} TO {runtime}")
            elif row["relkind"] in _TABLE_KINDS and row["relname"] in _READ_ONLY:
                if privileges := ", ".join(_READ_ONLY[row["relname"]]):
                    await connection.execute(
                        f"GRANT {privileges} ON {_target(schema, row)} TO {runtime}"
                    )
            elif row["relkind"] in _TABLE_KINDS and row["relname"] != ledger:
                await connection.execute(
                    f"GRANT {', '.join(_DML)} ON {_target(schema, row)} TO {runtime}"
                )
                granted.append(row["relname"])
        await _unprivileged(connection, runtime_login)
    return tuple(granted)

require_login_retired async

require_login_retired(connection: Any, old_login: str) -> None

Administrative observer: is old_login disabled, unreachable and drained?

Point-in-time and named-login only. It cannot see a session that already ran SET ROLE from a former member, privileged operators who may re-enable the login, credentials copied into another login, or other databases' prepared transactions beyond this cluster's view. It changes nothing.

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

async def require_login_retired(connection: Any, old_login: str) -> None:
    """Administrative observer: is ``old_login`` disabled, unreachable and drained?

    Point-in-time and named-login only. It cannot see a session that already
    ran SET ROLE from a former member, privileged operators who may re-enable
    the login, credentials copied into another login, or other databases'
    prepared transactions beyond this cluster's view. It changes nothing.
    """
    row = await connection.fetchrow(
        "SELECT oid, rolcanlogin FROM pg_roles WHERE rolname=$1", old_login
    )
    if not row:
        raise RuntimeRoleNotReady("old login does not exist; name the login to retire")
    if row["rolcanlogin"]:
        raise RuntimeRoleNotReady("old login still has LOGIN")
    if await connection.fetchval(
        "SELECT EXISTS(SELECT 1 FROM pg_auth_members WHERE roleid=$1 OR member=$1)", row["oid"]
    ):
        raise RuntimeRoleNotReady("old login still has member roles or memberships")
    if await connection.fetchval(
        "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE usesysid=$1)", row["oid"]
    ):
        raise RuntimeRoleNotReady("old login still has sessions")
    if await connection.fetchval(
        "SELECT EXISTS(SELECT 1 FROM pg_prepared_xacts WHERE owner=$1)", old_login
    ):
        raise RuntimeRoleNotReady("old login still has prepared transactions")

require_split_runtime async

require_split_runtime(connection: Any, *, owner_login: str, tables: tuple[str, ...], schema: str = 'public', ledger: str = MIGRATION_LEDGER) -> None

Runtime side, read-only: refuse startup unless this login is DML-only.

The connected login must be a direct, unprivileged, membership-free LOGIN owning nothing (large objects included), able to create nothing, holding no large object grant and unable to EXECUTE any SECURITY DEFINER function outside pg_catalog and information_schema. schema and every relation in it must belong to owner_login; each table carries exactly the table-level DML grant and no column grant beyond it (the ledger none, the erase job table SELECT only, its receipts none, the memory erase generations SELECT/INSERT/UPDATE, the erase log SELECT/INSERT), every tables entry and memory_nodes resolve on this login's search_path to schema, views, materialized views and foreign tables carry none, every tables entry exists, and sequences are USAGE without UPDATE.

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

async def require_split_runtime(
    connection: Any,
    *,
    owner_login: str,
    tables: tuple[str, ...],
    schema: str = "public",
    ledger: str = MIGRATION_LEDGER,
) -> None:
    """Runtime side, read-only: refuse startup unless this login is DML-only.

    The connected login must be a direct, unprivileged, membership-free LOGIN
    owning nothing (large objects included), able to create nothing, holding no
    large object grant and unable to EXECUTE any SECURITY DEFINER function
    outside ``pg_catalog`` and ``information_schema``. ``schema`` and every
    relation in it must belong to ``owner_login``; each table carries exactly
    the table-level DML grant and no column grant beyond it (the ledger none,
    the erase job table SELECT only, its receipts none, the memory erase
    generations SELECT/INSERT/UPDATE, the erase log SELECT/INSERT), every ``tables`` entry
    and ``memory_nodes`` resolve on this login's search_path to ``schema``,
    views, materialized views and foreign tables carry none, every ``tables``
    entry exists, and sequences are USAGE without UPDATE.
    """
    if not await connection.fetchval("SELECT session_user=current_user"):
        raise RuntimeRoleNotReady("runtime must not run under SET ROLE")
    login = await connection.fetchval("SELECT current_user")
    if login == owner_login:
        raise RuntimeRoleNotReady("runtime must not be the schema owner")
    await _unprivileged(connection, login)
    if not await connection.fetchval(
        "SELECT pg_get_userbyid(nspowner)=$2 FROM pg_namespace WHERE nspname=$1",
        schema,
        owner_login,
    ):
        raise RuntimeRoleNotReady("application schema must belong to the named owner")
    rows = await connection.fetch(_RELATIONS, schema)
    present = {row["relname"] for row in rows if row["relkind"] in _TABLE_KINDS}
    if not all(row["schema_owned"] for row in rows) or not set(tables) <= present:
        raise RuntimeRoleNotReady(
            "required tables must exist and every relation belong to the owner"
        )
    # The native erase fence reads jobs beside the memory_nodes this login
    # resolves; attesting another schema's grants would prove nothing there.
    for name in sorted({*tables, "memory_nodes"} & present):
        if await connection.fetchval(_RESOLVES_TO, name) != schema:
            raise RuntimeRoleNotReady(f"{name} must resolve to the attested schema {schema}")
    for held in await connection.fetch(_HELD, schema, list(_DML + _EXCESS)):
        table, privilege = held["relname"], held["privilege"]
        wanted = _READ_ONLY.get(table, _DML)
        if held["relkind"] in _TABLE_KINDS and table != ledger and privilege in wanted:
            ok = held["on_table"]
        else:
            ok = not held["anywhere"]
        if not ok:
            raise RuntimeRoleNotReady(
                f"runtime privilege {privilege} on {table} "
                f"must be {'absent' if held['anywhere'] else 'held'}"
            )
    for sequence in (row for row in rows if row["relkind"] == "S"):
        target = f'{_identifier(schema)}."{sequence["relname"]}"'
        if not await connection.fetchval(
            "SELECT has_sequence_privilege(current_user, $1, 'USAGE') AND NOT "
            "has_sequence_privilege(current_user, $1, 'UPDATE')",
            target,
        ):
            raise RuntimeRoleNotReady("runtime sequences must be USAGE without UPDATE")

revoke_runtime async

revoke_runtime(connection: Any, runtime_login: str, *, schema: str = 'public') -> tuple[str, ...]

Owner side rollback: withdraw every runtime privilege in schema at once.

It does not re-enable any old login; reopening one is a separate, explicitly authorised operator act that require_login_retired reports.

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

async def revoke_runtime(
    connection: Any, runtime_login: str, *, schema: str = "public"
) -> tuple[str, ...]:
    """Owner side rollback: withdraw every runtime privilege in ``schema`` at once.

    It does not re-enable any old login; reopening one is a separate,
    explicitly authorised operator act that ``require_login_retired`` reports.
    """
    runtime, quoted_schema = _identifier(runtime_login), _identifier(schema)
    async with connection.transaction():
        rows = await _owned_schema(connection, schema)
        for row in rows:
            await connection.execute(f"REVOKE ALL ON {_target(schema, row)} FROM {runtime}")
        await connection.execute(f"REVOKE ALL ON SCHEMA {quoted_schema} FROM {runtime}")
    return tuple(row["relname"] for row in rows if row["relkind"] in _TABLE_KINDS)