"""Point-in-time holding ledgers for user permissions and auth roles. Revision ID: 484 Revises: 483 「张三 2026-03 时是不是 global_admin」「李四 5 月持有过哪些软件权限」—— 这两个问题今天**无法用 SQL 回答**,而它们恰好是 SOC2 CC6.1 / CC6.3 每次 取证都要问的: * ``auth_user_roles`` 撤销即 **DELETE 行**,没有 ``revoked_at``。角色一旦 收回,「他曾经是管理员」这件事在表里不留任何痕迹。 * ``user_permissions`` 上有 ``UNIQUE(user_id, permission_id)``,re-request 走的是 **UPDATE 现有行**(``PermissionService.request_permission`` / ``admin_grant`` 的终态行复用分支):``granted_at`` 被改写、``revoked_at`` 清空,上一个持有周期**整段消失**。云权限(``cmdb_cloud_access_grants``) 遇到同一问题时至少在 ``detail._resurrection_history`` 里留了 JSON 补偿, 软件权限连这个都没有。 两张 append-only 台账各记录一个**持有区间**(而不是事件流):时点查询就是 一句 ``granted_at <= D AND (closed_at IS NULL OR closed_at > D)``,不需要 窗口函数配对 open/close 事件,也不需要在应用层解析 JSON。 **故意不建外键**(与 ``audit_logs`` 同理由,见 ``app/audit/models.py``): ① 取证行必须比它描述的人/权限行活得久 —— 目录行软删、person 行被 HRIS purge 之后,「他当时持有过」这件事本身仍然是证据;② 子表 INSERT 会给父行 上共享锁,而这两张表的写入点在授权/回收热路径上,加 FK 等于把 03:00 批量 回收和登录路径拉进同一把锁(2026 年已有两起 1205 事故是这么来的)。 列类型仍用 ``id_type_for`` 与父表对齐,JOIN 不受影响。 存量不回填(no speculative backfill):台账从上线时刻开始记账,上线前的 历史仍然只能回扫 ``audit_logs``。回填需要从 829 种 action 的非结构化 ``detail`` 里反推区间,准确性不可证明,宁可留空也不造假证据。 """ import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for from app.core.sql_types import PreciseDateTime as _PRECISE_TS revision = "484" down_revision = "483" branch_labels = None depends_on = None PERMISSION_TABLE = "user_permission_cycles" ROLE_TABLE = "auth_user_role_cycles" def _table_exists(bind, name: str) -> bool: return bool( bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t" ), {"t": name}, ).fetchone() ) def upgrade() -> None: bind = op.get_bind() row_id = id_type_for("identity_persons") kw = table_kw_for("identity_persons") if not _table_exists(bind, PERMISSION_TABLE): op.create_table( PERMISSION_TABLE, sa.Column("id", row_id, primary_key=True), # The ``user_permissions`` row this cycle belongs to. That row # is reused across cycles (UNIQUE(user_id, permission_id)), so # it is a stable handle rather than a per-cycle identity. sa.Column("user_permission_id", row_id, nullable=False), sa.Column("user_id", row_id, nullable=False), sa.Column("permission_id", row_id, nullable=False), # opened_at = the request/grant row entered a non-terminal # state. granted_at = it actually became ACTIVE (the vendor # side landed). A cycle with opened_at but NULL granted_at is # a request that never became access — it belongs in the # audit trail but must NOT count as "held access". sa.Column("opened_at", _PRECISE_TS, nullable=False), sa.Column("granted_at", _PRECISE_TS, nullable=True), sa.Column("closed_at", _PRECISE_TS, nullable=True), sa.Column("open_status", sa.String(32), nullable=False), sa.Column("close_status", sa.String(32), nullable=True), sa.Column("opened_by_person_id", row_id, nullable=True), sa.Column("closed_by_person_id", row_id, nullable=True), sa.Column("approval_request_id", row_id, nullable=True), sa.Column("provisioning_task_id", row_id, nullable=True), sa.Column("reason", sa.Text(), nullable=True), sa.Column("open_trace_id", row_id, nullable=True), sa.Column("close_trace_id", row_id, nullable=True), sa.Column("detail", sa.JSON(), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), # Deliberately NO uniqueness on (user_permission_id, # opened_at). It was tried as a double-emit backstop and came # out: any UNIQUE here can turn a ledger bug into a 1062 on a # legitimate grant, and an evidence ledger must never be able # to fail the transaction it observes. The invariant "at most # one open cycle per row" is held instead by the listener # closing before it opens (locked by tests) and surfaced as a # drift finding rather than as a write error. **kw, ) # Point-in-time roster for one permission: "who held P on D". op.create_index( "ix_user_permission_cycles_perm_window", PERMISSION_TABLE, ["permission_id", "granted_at", "closed_at"], ) # Point-in-time footprint for one person: "what did X hold on D". op.create_index( "ix_user_permission_cycles_person_window", PERMISSION_TABLE, ["user_id", "granted_at", "closed_at"], ) op.create_index( "ix_user_permission_cycles_row", PERMISSION_TABLE, ["user_permission_id"], ) if not _table_exists(bind, ROLE_TABLE): op.create_table( ROLE_TABLE, sa.Column("id", row_id, primary_key=True), sa.Column("user_id", row_id, nullable=False), sa.Column("role_id", row_id, nullable=False), # Snapshot of the role name at grant time. ``auth_roles`` rows # are stable today, but an evidence ledger that resolves names # by live JOIN would silently rewrite history if a role were # ever renamed. Mirrors the ``data_access_bindings`` # ``*_snapshot`` columns. sa.Column("role_name", sa.String(32), nullable=False), sa.Column("granted_at", _PRECISE_TS, nullable=False), sa.Column("granted_by_person_id", row_id, nullable=True), # manual | derived — copied from auth_user_roles.grant_source # so a reconciler-created grant is distinguishable from an # admin decision without joining to a row that may be gone. sa.Column( "grant_source", sa.String(16), nullable=False, server_default="manual", ), sa.Column("revoked_at", _PRECISE_TS, nullable=True), sa.Column("revoked_by_person_id", row_id, nullable=True), sa.Column("open_trace_id", row_id, nullable=True), sa.Column("close_trace_id", row_id, nullable=True), sa.Column("detail", sa.JSON(), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), # No uniqueness here either — same reasoning as the # permission ledger above. **kw, ) # "who held role R on D" — the SOC2 CC6.1 privileged-roster query. op.create_index( "ix_auth_user_role_cycles_role_window", ROLE_TABLE, ["role_id", "granted_at", "revoked_at"], ) op.create_index( "ix_auth_user_role_cycles_person_window", ROLE_TABLE, ["user_id", "granted_at", "revoked_at"], ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, ROLE_TABLE): op.drop_table(ROLE_TABLE) if _table_exists(bind, PERMISSION_TABLE): op.drop_table(PERMISSION_TABLE)