feat(identity): persist administrator authority with scoped database roles

backend/migrations/versions/0005_identity_authority.py

@@ -0,0 +1,150 @@
+"""Default-denied administrator authority and separate local operator/import roles."""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0005_identity_authority"
+down_revision = "0004_performance_watermark"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "accounts",
+        sa.Column("is_administrator", sa.Boolean(), server_default="false", nullable=False),
+    )
+    op.add_column(
+        "accounts",
+        sa.Column("authorization_version", sa.BigInteger(), server_default="1", nullable=False),
+    )
+    op.create_check_constraint(
+        op.f("ck_accounts_authorization_version"), "accounts", "authorization_version >= 1"
+    )
+    op.create_index(
+        "ix_accounts_active_administrators",
+        "accounts",
+        ["id"],
+        postgresql_where=sa.text("is_administrator AND enabled"),
+    )
+    # Logins and secrets are provisioned separately by an operator. Neither group
+    # may inherit an existing privileged role or quietly reuse a privileged login.
+    for role in ("kf_oj_identity_operator", "kf_oj_problem_import"):
+        op.execute(f"""
+            DO $$ BEGIN
+                IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname='{role}') THEN
+                    CREATE ROLE {role} NOLOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE
+                        NOREPLICATION NOBYPASSRLS;
+                ELSIF EXISTS (SELECT 1 FROM pg_roles WHERE rolname='{role}' AND
+                    (rolcanlogin OR rolsuper OR rolcreatedb OR rolcreaterole
+                        OR rolreplication OR rolbypassrls))
+                    OR EXISTS (SELECT 1 FROM pg_auth_members m JOIN pg_roles r ON r.oid=m.member
+                        WHERE r.rolname='{role}') THEN
+                    RAISE EXCEPTION 'unsafe existing authority or import role';
+                END IF;
+            END $$
+        """)
+    op.execute(
+        "REVOKE ALL ON ALL TABLES IN SCHEMA public "
+        "FROM kf_oj_identity_operator,kf_oj_problem_import"
+    )
+    # Table REVOKE does not remove explicit column grants. Remove those too,
+    # including grants left by an earlier downgrade or manual role provisioning.
+    op.execute("""
+        DO $$ DECLARE t record; BEGIN
+            FOR t IN SELECT c.relname, string_agg(quote_ident(a.attname),',' ORDER BY a.attnum) cols
+                FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace
+                JOIN pg_attribute a ON a.attrelid=c.oid
+                WHERE n.nspname='public' AND c.relkind IN ('r','p','v','f')
+                    AND a.attnum>0 AND NOT a.attisdropped
+                GROUP BY c.oid,c.relname LOOP
+                EXECUTE format('REVOKE SELECT (%s),INSERT (%s),UPDATE (%s),REFERENCES (%s) '
+                    'ON TABLE public.%I FROM kf_oj_identity_operator,kf_oj_problem_import',
+                    t.cols,t.cols,t.cols,t.cols,t.relname);
+            END LOOP;
+        END $$
+    """)
+    op.execute("GRANT USAGE ON SCHEMA public TO kf_oj_identity_operator,kf_oj_problem_import")
+    op.execute("REVOKE INSERT,UPDATE,DELETE ON accounts FROM kf_oj_platform")
+    op.execute(
+        "REVOKE INSERT(id,username,password_hash,enabled,created_at,"
+        "is_administrator,authorization_version),"
+        "UPDATE(id,username,password_hash,enabled,created_at,"
+        "is_administrator,authorization_version) "
+        "ON accounts FROM kf_oj_platform"
+    )
+    op.execute(
+        "GRANT INSERT(id,username,password_hash,enabled,created_at) ON accounts TO kf_oj_platform"
+    )
+    op.execute("GRANT UPDATE(enabled) ON accounts TO kf_oj_platform")
+    op.execute("GRANT SELECT ON accounts,sessions TO kf_oj_identity_operator")
+    op.execute("GRANT UPDATE(is_administrator,enabled) ON accounts TO kf_oj_identity_operator")
+    op.execute("GRANT UPDATE(revoked_at) ON sessions TO kf_oj_identity_operator")
+    op.execute("REVOKE ALL ON accounts,sessions FROM kf_oj_coordinator,kf_oj_problem_import")
+    # No import queue exists yet. Its role deliberately has no table privileges
+    # until the import migration grants only the queue/staging tables it needs.
+    op.execute("""
+        CREATE FUNCTION public.kf_account_authority_version() RETURNS trigger LANGUAGE plpgsql
+        SET search_path = pg_catalog, public AS $$
+        BEGIN
+            IF NEW.authorization_version IS DISTINCT FROM OLD.authorization_version THEN
+                RAISE EXCEPTION 'authorization version is server derived' USING ERRCODE='23514';
+            END IF;
+            IF OLD.is_administrator AND NEW.enabled IS DISTINCT FROM OLD.enabled
+                AND NOT pg_has_role(current_user,'kf_oj_identity_operator','USAGE') THEN
+                RAISE EXCEPTION 'administrator enablement requires the local operator'
+                    USING ERRCODE='42501';
+            END IF;
+            IF NEW.is_administrator IS DISTINCT FROM OLD.is_administrator
+                OR NEW.enabled IS DISTINCT FROM OLD.enabled THEN
+                NEW.authorization_version := OLD.authorization_version + 1;
+            END IF;
+            RETURN NEW;
+        END $$
+    """)
+    op.execute("""
+        CREATE FUNCTION public.kf_revoke_authority_sessions() RETURNS trigger LANGUAGE plpgsql
+        SET search_path = pg_catalog, public AS $$
+        BEGIN
+            IF NEW.authorization_version != OLD.authorization_version THEN
+                UPDATE public.sessions SET revoked_at=coalesce(revoked_at,clock_timestamp())
+                    WHERE account_id=NEW.id AND revoked_at IS NULL;
+            END IF;
+            RETURN NEW;
+        END $$
+    """)
+    for function in ("kf_account_authority_version", "kf_revoke_authority_sessions"):
+        op.execute(f"REVOKE ALL ON FUNCTION public.{function}() FROM PUBLIC")
+    op.execute("""
+        CREATE TRIGGER kf_account_authority_version BEFORE UPDATE ON public.accounts
+        FOR EACH ROW EXECUTE FUNCTION public.kf_account_authority_version()
+    """)
+    op.execute("""
+        CREATE TRIGGER kf_revoke_authority_sessions AFTER UPDATE ON public.accounts
+        FOR EACH ROW EXECUTE FUNCTION public.kf_revoke_authority_sessions()
+    """)
+
+
+def downgrade() -> None:
+    op.execute("""
+        DO $$ BEGIN
+            IF EXISTS(SELECT 1 FROM accounts
+                WHERE is_administrator OR authorization_version != 1) THEN
+                RAISE EXCEPTION 'cannot discard authority or revocation history'
+                    USING ERRCODE='23514';
+            END IF;
+        END $$
+    """)
+    for name in ("kf_revoke_authority_sessions", "kf_account_authority_version"):
+        op.execute(f"DROP TRIGGER {name} ON public.accounts")
+        op.execute(f"DROP FUNCTION public.{name}()")
+    op.execute("REVOKE ALL ON accounts,sessions FROM kf_oj_identity_operator")
+    op.execute("REVOKE UPDATE(is_administrator,enabled) ON accounts FROM kf_oj_identity_operator")
+    op.execute("REVOKE UPDATE(revoked_at) ON sessions FROM kf_oj_identity_operator")
+    op.drop_index("ix_accounts_active_administrators", table_name="accounts")
+    op.drop_constraint(op.f("ck_accounts_authorization_version"), "accounts", type_="check")
+    op.drop_column("accounts", "authorization_version")
+    op.drop_column("accounts", "is_administrator")
+    op.execute("GRANT INSERT,UPDATE,DELETE ON accounts TO kf_oj_platform")
+    # Shared NOLOGIN groups survive downgrade, as in the foundation migration;
+    # removing a cluster role could break another database using the same groups.

backend/src/kf_oj/modules/identity/README.md

@@ -14,6 +14,15 @@
 登录接口为 `/api/v1/auth/login`，退出 `/logout`，当前主体 `/me`。会话使用随机 256-bit token，
 数据库只保存 token / CSRF 摘要；默认 12 小时有效，单账号保留至多 10 个有效会话。
 账号禁用立即拒绝登录和旧会话。请求只采用 ASGI 连接 peer，不读取客户端转发地址。
+
+`0005_identity_authority` 为旧账号保留普通用户默认值，并加入管理员标记与服务端派生的授权版本。
+启用状态或管理员标记实际变化时，同一事务递增授权版本并撤销旧会话；事务回滚也回滚撤销。
+平台账号仅可插入普通账号列、更新启用状态，不能写管理员标记或授权版本；管理员的启用状态
+需要独立本地运维身份。`kf_oj_identity_operator` 与 `kf_oj_problem_import` 为无登录的独立权限组，
+口令与登录身份由运维另行配置，平台不继承运维组。导入组目前没有表权限，后续导入迁移只授予
+队列与候选所需权限，不授予评测状态写入权。迁移不创建默认管理员；管理员 CLI、公开能力端口
+及管理 API 尚待接通。保留授权或撤销历史后拒绝回退这次迁移。
+
 受控启动使用 `--no-proxy-headers`，避免服务器中间件预先替换 peer；反向代理后的用户共享
 该连接预算，引入可信代理的独立限流策略之前不能直接打开客户端转发头信任。
 每 peer 至多 10 次登录/15 分钟，预算通过真实数据库串行更新，失败尝试也消耗预算。

backend/src/kf_oj/modules/identity/models.py

@@ -3,7 +3,17 @@
 from datetime import datetime
 from uuid import UUID
 
-from sqlalchemy import Boolean, CheckConstraint, DateTime, ForeignKey, Index, String, Text
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    CheckConstraint,
+    DateTime,
+    ForeignKey,
+    Index,
+    String,
+    Text,
+    text,
+)
 from sqlalchemy.orm import Mapped, mapped_column
 
 from kf_oj.infra.db.base import Base
@@ -15,12 +25,20 @@ class Account(CreatedAt, Base):
     __table_args__ = (
         CheckConstraint("username ~ '^[a-z0-9_]{3,24}$'", name="canonical_username"),
         CheckConstraint("char_length(password_hash) BETWEEN 40 AND 256", name="password_hash_size"),
+        CheckConstraint("authorization_version >= 1", name="authorization_version"),
+        Index(
+            "ix_accounts_active_administrators",
+            "id",
+            postgresql_where=text("is_administrator AND enabled"),
+        ),
     )
 
     id: Mapped[UUID] = mapped_column(primary_key=True)
     username: Mapped[str] = mapped_column(String(24), unique=True)
     password_hash: Mapped[str] = mapped_column(Text)
     enabled: Mapped[bool] = mapped_column(Boolean, server_default="true")
+    is_administrator: Mapped[bool] = mapped_column(Boolean, server_default="false")
+    authorization_version: Mapped[int] = mapped_column(BigInteger, server_default="1")
 
 
 class Session(CreatedAt, Base):

backend/tests/test_administrator_storage.py

@@ -0,0 +1,235 @@
+"""Actual temporary PG migration and authority storage; API/CLI acceptance is separate."""
+
+import asyncio
+import subprocess
+from datetime import timedelta
+from uuid import uuid4
+
+import asyncpg
+import pytest
+from conftest import connect, migrate
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from test_database_foundation import fixture_rows
+
+from kf_oj.modules.identity.public import AuthenticationFailed, Principal
+from kf_oj.modules.identity.service import IdentityService, token_digest
+
+
+class RollBackAuthorityChange(Exception):
+    """Abort the test transaction before any authority change commits."""
+
+
+async def submission_snapshot(connection):
+    # Synthetic queued metadata only; this does not claim real program execution.
+    return {
+        table: list(
+            await connection.fetch(
+                f"SELECT to_jsonb(t)::text FROM {table} t ORDER BY to_jsonb(t)::text"
+            )
+        )
+        for table in (
+            "artifacts",
+            "problems",
+            "problem_versions",
+            "problem_cases",
+            "submissions",
+            "evaluations",
+            "outbox",
+        )
+    }
+
+
+async def account_and_session(connection):
+    account = uuid4()
+    token, csrf = uuid4().hex * 2, uuid4().hex * 2
+    await connection.execute(
+        "INSERT INTO accounts(id,username,password_hash) VALUES($1,'storage_user',$2)",
+        account,
+        "synthetic-irreversible-placeholder-hash-" * 2,
+    )
+    now = await connection.fetchval("SELECT clock_timestamp()")
+    await connection.execute(
+        "INSERT INTO sessions(token_hash,csrf_hash,account_id,expires_at) VALUES($1,$2,$3,$4)",
+        token_digest(token),
+        token_digest(csrf),
+        account,
+        now + timedelta(hours=1),
+    )
+    return account, token, csrf
+
+
+def test_upgrade_preserves_ordinary_account_and_current_session(postgres_url):
+    migrate(postgres_url, "downgrade", "0004_performance_watermark")
+
+    async def seed():
+        connection = await connect(postgres_url)
+        try:
+            account, token, csrf = await account_and_session(connection)
+            await fixture_rows(connection)
+            return account, token, csrf, await submission_snapshot(connection)
+        finally:
+            await connection.close()
+
+    account, token, csrf, before = asyncio.run(seed())
+    migrate(postgres_url, "upgrade", "head")
+    migrate(postgres_url, "check")
+
+    async def inspect():
+        connection = await connect(postgres_url)
+        engine = create_async_engine(postgres_url)
+        try:
+            row = await connection.fetchrow("SELECT * FROM accounts WHERE id=$1", account)
+            assert row["enabled"] and not row["is_administrator"]
+            assert row["authorization_version"] == 1
+            assert await submission_snapshot(connection) == before
+            service = IdentityService(async_sessionmaker(engine, expire_on_commit=False))
+            assert await service.authenticate(token, csrf_token=csrf) == Principal(
+                account, "storage_user"
+            )
+        finally:
+            await connection.close()
+            await engine.dispose()
+
+    asyncio.run(inspect())
+
+
+def test_authority_columns_and_import_role_are_not_runtime_write_permissions(postgres_url):
+    async def inspect():
+        connection = await connect(postgres_url)
+        try:
+            for role in ("kf_oj_platform", "kf_oj_coordinator", "kf_oj_problem_import"):
+                for column in ("is_administrator", "authorization_version"):
+                    assert not await connection.fetchval(
+                        "SELECT has_column_privilege($1,'accounts',$2,'UPDATE')", role, column
+                    )
+                    assert not await connection.fetchval(
+                        "SELECT has_column_privilege($1,'accounts',$2,'INSERT')", role, column
+                    )
+            for table in (
+                "evaluations",
+                "judge_attempts",
+                "case_results",
+                "evaluation_performance",
+            ):
+                for privilege in ("SELECT", "INSERT", "UPDATE", "DELETE"):
+                    assert not await connection.fetchval(
+                        "SELECT has_table_privilege('kf_oj_problem_import',$1,$2)", table, privilege
+                    )
+            assert await connection.fetchval(
+                "SELECT has_column_privilege('kf_oj_platform','accounts','enabled','UPDATE')"
+            )
+            assert await connection.fetchval(
+                "SELECT has_column_privilege('kf_oj_identity_operator',"
+                "'accounts','is_administrator','UPDATE')"
+            )
+            assert not await connection.fetchval(
+                "SELECT has_column_privilege('kf_oj_identity_operator',"
+                "'accounts','authorization_version','UPDATE')"
+            )
+            assert not await connection.fetchval(
+                "SELECT pg_has_role('kf_oj_platform','kf_oj_identity_operator','USAGE')"
+            )
+        finally:
+            await connection.close()
+
+    asyncio.run(inspect())
+
+
+def test_authority_change_derives_version_and_revokes_old_session_atomically(postgres_url):
+    async def inspect():
+        connection = await connect(postgres_url)
+        engine = create_async_engine(postgres_url)
+        try:
+            account, token, csrf = await account_and_session(connection)
+            service = IdentityService(async_sessionmaker(engine, expire_on_commit=False))
+            await service.authenticate(token, csrf_token=csrf)
+            async with connection.transaction():
+                await connection.execute("SET LOCAL ROLE kf_oj_identity_operator")
+                await connection.execute(
+                    "UPDATE accounts SET is_administrator=true WHERE id=$1", account
+                )
+            assert (
+                await connection.fetchval(
+                    "SELECT authorization_version FROM accounts WHERE id=$1", account
+                )
+                == 2
+            )
+            with pytest.raises(AuthenticationFailed):
+                await service.authenticate(token, csrf_token=csrf)
+            async with connection.transaction():
+                await connection.execute("SET LOCAL ROLE kf_oj_identity_operator")
+                await connection.execute(
+                    "UPDATE accounts SET is_administrator=true WHERE id=$1", account
+                )
+            assert (
+                await connection.fetchval(
+                    "SELECT authorization_version FROM accounts WHERE id=$1", account
+                )
+                == 2
+            )
+            fresh_token, fresh_csrf = uuid4().hex * 2, uuid4().hex * 2
+            await connection.execute(
+                "INSERT INTO sessions(token_hash,csrf_hash,account_id,expires_at) "
+                "VALUES($1,$2,$3,clock_timestamp()+interval '1 hour')",
+                token_digest(fresh_token),
+                token_digest(fresh_csrf),
+                account,
+            )
+            with pytest.raises(RollBackAuthorityChange):
+                async with connection.transaction():
+                    await connection.execute(
+                        "UPDATE accounts SET is_administrator=false WHERE id=$1", account
+                    )
+                    assert (
+                        await connection.fetchval(
+                            "SELECT authorization_version FROM accounts WHERE id=$1", account
+                        )
+                        == 3
+                    )
+                    raise RollBackAuthorityChange
+            assert await connection.fetchval(
+                "SELECT is_administrator FROM accounts WHERE id=$1", account
+            )
+            assert (
+                await connection.fetchval(
+                    "SELECT authorization_version FROM accounts WHERE id=$1", account
+                )
+                == 2
+            )
+            await service.authenticate(fresh_token, csrf_token=fresh_csrf)
+            async with connection.transaction():
+                await connection.execute("SET LOCAL ROLE kf_oj_identity_operator")
+                await connection.execute(
+                    "UPDATE accounts SET is_administrator=false WHERE id=$1", account
+                )
+            assert (
+                await connection.fetchval(
+                    "SELECT authorization_version FROM accounts WHERE id=$1", account
+                )
+                == 3
+            )
+            with pytest.raises(AuthenticationFailed):
+                await service.authenticate(fresh_token, csrf_token=fresh_csrf)
+        finally:
+            await connection.close()
+            await engine.dispose()
+
+    asyncio.run(inspect())
+
+
+def test_authority_history_cannot_be_erased_by_downgrade(postgres_url):
+    async def change():
+        connection = await connect(postgres_url)
+        try:
+            account, _, _ = await account_and_session(connection)
+            await connection.execute("UPDATE accounts SET enabled=false WHERE id=$1", account)
+            with pytest.raises(asyncpg.CheckViolationError):
+                await connection.execute(
+                    "UPDATE accounts SET authorization_version=1 WHERE id=$1", account
+                )
+        finally:
+            await connection.close()
+
+    asyncio.run(change())
+    with pytest.raises(subprocess.CalledProcessError):
+        migrate(postgres_url, "downgrade", "0004_performance_watermark")

backend/tests/test_database_foundation.py

@@ -12,9 +12,14 @@ def test_migration_roundtrip_and_role_safety(postgres_url):
         c = await connect(postgres_url)
         try:
             tables = set(await c.fetch("SELECT tablename FROM pg_tables WHERE schemaname='public'"))
-            assert len(tables) == 13  # Twelve business tables and Alembic history.
+            assert len(tables) == 14  # Thirteen business tables and Alembic history.
             roles = await c.fetch("SELECT * FROM pg_roles WHERE rolname LIKE 'kf_oj_%'")
-            assert {row["rolname"] for row in roles} == {"kf_oj_platform", "kf_oj_coordinator"}
+            assert {row["rolname"] for row in roles} == {
+                "kf_oj_platform",
+                "kf_oj_coordinator",
+                "kf_oj_identity_operator",
+                "kf_oj_problem_import",
+            }
             assert all(
                 not any(
                     r[k]