diff --git a/backend/migrations/versions/0005_identity_authority.py b/backend/migrations/versions/0005_identity_authority.py new file mode 100644 index 0000000..c0015b4 --- /dev/null +++ b/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. diff --git a/backend/src/kf_oj/modules/identity/README.md b/backend/src/kf_oj/modules/identity/README.md index 81b9bcf..722ca9a 100644 --- a/backend/src/kf_oj/modules/identity/README.md +++ b/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 分钟，预算通过真实数据库串行更新，失败尝试也消耗预算。 diff --git a/backend/src/kf_oj/modules/identity/models.py b/backend/src/kf_oj/modules/identity/models.py index 7556bda..23859f9 100644 --- a/backend/src/kf_oj/modules/identity/models.py +++ b/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): diff --git a/backend/tests/test_administrator_storage.py b/backend/tests/test_administrator_storage.py new file mode 100644 index 0000000..22220f3 --- /dev/null +++ b/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") diff --git a/backend/tests/test_database_foundation.py b/backend/tests/test_database_foundation.py index 761ecf1..b7a2db0 100644 --- a/backend/tests/test_database_foundation.py +++ b/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]