"""访问撤销落地(PR1 of 访问对象闭环):SSO 撤销要在 IdP 侧真的发生,API Key 转手要轮换、闲置要有人问。 ## 这是在解决什么 审计(2026-09-05)里最重的一条:``SsoGrantService.revoke`` 把 ``status`` 翻成 ``revoked`` 之后,只给系统负责人和 IT 发一张「请去 IdP 手动停」的卡 —— 没有任务、 没有到期、没有复核。撤销之后这条残留反而从半年度自证的快照里消失,比撤销前更不 可见。同一条链上,离职者的 API Key 被强制转给经理时**明写不轮换**,而三条 apikey cron 里没有任何一条问过「这把 Key 还有人用吗」。 ## sso_grants 上加的一组列:把「撤了」拆成「决定撤」和「真的没了」 | 列 | 是什么 | |---|---| | ``removal_status`` | ``not_applicable`` / ``pending`` / ``confirmed`` / ``verified`` / ``contradicted`` / ``unverified_legacy`` | | ``removal_task_id`` | 挂在 ``provisioning_tasks`` 上的那条 manual_notify 任务(凭证要求走它) | | ``removal_confirmed_at`` / ``_by_person_id`` / ``removal_external_ref`` | 人签字:谁、何时、凭什么说「已在目标系统停用」 | | ``removal_verified_at`` | 厂商证据(Google Reports 的 SAML 登录审计)确认撤销后再无登录 | | ``removal_last_checked_at`` / ``removal_last_login_at`` / ``removal_check_error`` | 最近一次复核的时间、观察到的最后一次登录、以及复核失败的原因(典型:缺 Reports 权限) | | ``removal_reminder_count`` / ``removal_last_reminded_at`` | 催办阶梯的计数与时间(只能拉长间隔,不能收紧) | **历史行刻意标成 ``unverified_legacy`` 而不是 ``pending``**:在这次迁移之前撤销的 授权从没有人核实过,把它们统统开成任务是推测性回填;标成一个单独的、看得见的状态, 由管理员在台账上批量「开复核任务」,才是诚实的做法。 ## apikey_idle_alerts_dedup ``(assignment_id, billing_month)`` 唯一 —— 和 ``apikey_budget_alerts_dedup`` / ``apikey_expiry_alerts_dedup`` 同一个姿势:dedup 行的意义是「这个月问过了」, 投递失败不回滚它。 ## 两条 cron 与四个平台设置 * ``sso.removal_reconcile``(每日 08:40 上海):给每条 ``pending`` 的撤销保证有一条 打开的任务、按阶梯催办、到 14 天开合规事件;对 Google 提供方的授权拉 SAML 登录 审计做复核。 * ``apikey.idle_check``(每月 3 日 09:30 上海,账单入库之后):上自然月花费为 $0 且已发放满 45 天的 Key,IT 收一张汇总;员工那张卡由 ``apikey.idle_alert.enabled_for_employees`` 控制(默认关 —— 2026-09-05 用户拍板 「先发 IT 汇总,逐类灰度」)。 * ``apikey.rotate_on_transfer`` 默认 true:所有权变更(含离职强制转移)即入队一次 轮换,旧凭证按 ``apikey.transfer_rotation_grace_days``(默认 7)天后退役。 * ``sso.removal`` 一个 JSON:复核开关、阶梯天数、几天无登录算 verified、几天开事件。 Revision ID: 603 Revises: 602 ⚠️ 迁移号要对着 ``origin/main`` 抢,而且每次 push 之前重新抢一次。 """ from __future__ import annotations import json import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for revision = "603" down_revision = "602" branch_labels = None depends_on = None SSO = "sso_grants" IDLE = "apikey_idle_alerts_dedup" SETTINGS = "platform_settings" JOBS = "scheduled_jobs" JOB_SSO = "sso.removal_reconcile" JOB_IDLE = "apikey.idle_check" _JOB_SSO_DESC = ( "每日 08:40(Asia/Shanghai)。三件事:① 每条 removal_status='pending' 的 SSO " "撤销必须有一条打开的 provisioning 任务(manual_notify,完成要凭证),没有就补开;" "② 按阶梯(默认 3 / 7 / 14 天,之后每 7 天)催系统负责人,7 天起抄送 IT," "14 天开一条 compliance 事件;③ 对 Google 提供方的授权拉 Admin Reports 的 SAML " "登录审计:撤销之后仍有登录 → contradicted 并重开任务;已确认满 N 天无登录 → " "verified。缺 Reports 权限时不判失败,把原因写在 removal_check_error 上。" ) _JOB_IDLE_DESC = ( "每月 3 日 09:30(Asia/Shanghai,等账单入库)。扫描 status='active' 且发放满 " "min_age_days 的 API Key,取上自然月花费:$0 且数据可用 → 闲置。IT 收一张汇总卡" "(始终);员工本人那张「保留 / 释放 / 找 IT」卡由 apikey.idle_alert." "enabled_for_employees 控制,默认关。Bailian 等拿不到按 Key 花费的提供方跳过并" "计入 skipped_no_data。dedup 表按 (assignment, month) 保证每月最多问一次。" ) _SETTINGS: tuple[tuple[str, object, str], ...] = ( ( "apikey.rotate_on_transfer", True, "API Key 所有权变更(员工转交、离职强制转给经理)时是否立刻入队一次轮换。" "默认开:交出去的 Key 原持有人仍然知道明文,不轮换等于共享凭证。" "只对能自动轮换的提供方(Bedrock / Bailian / Azure OpenAI)生效,其余记为 " "manual_required 并通知新持有人与 IT。", ), ( "apikey.transfer_rotation_grace_days", 7, "转手触发的轮换里,旧凭证多少天后退役。0 = 立刻。给新持有人留出更新调用方的时间。", ), ( "apikey.idle_alert", { "enabled_for_employees": False, "min_age_days": 45, "keep_months": 6, }, "闲置 API Key 提醒。enabled_for_employees=false 时只给 IT 发汇总(2026-09-05 " "决定:新增员工触达先发 IT、逐类灰度);min_age_days 是发放满多少天才开始" "评估;keep_months 是持有人点「保留」后免打扰的月数。", ), ( "sso.removal", { "verify_via_google_reports": True, "reminder_ladder_days": [3, 7, 14], "verified_after_days": 7, "incident_after_days": 14, }, "SSO 撤销落地的节奏。verify_via_google_reports 控制是否用 Google Admin Reports " "的 SAML 登录审计复核(需要 DwD 授予 admin.reports.audit.readonly);" "reminder_ladder_days 是催办阶梯(只能拉长);verified_after_days 是确认后" "几天无登录算已核实;incident_after_days 是几天未落地开合规事件。", ), ) def _has_table(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :n" ), {"n": name}, ).scalar() is not None def _has_column(bind, table: str, column: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.columns " "WHERE table_schema = DATABASE() " "AND table_name = :t AND column_name = :c" ), {"t": table, "c": column}, ).scalar() is not None def _has_index(bind, table: str, index: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.statistics " "WHERE table_schema = DATABASE() " "AND table_name = :t AND index_name = :i" ), {"t": table, "i": index}, ).scalar() is not None def _has_fk(bind, table: str, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.table_constraints " "WHERE table_schema = DATABASE() AND table_name = :t " "AND constraint_name = :n AND constraint_type = 'FOREIGN KEY'" ), {"t": table, "n": name}, ).scalar() is not None def _seed_job(bind, *, key: str, display: str, desc: str, cron: str) -> None: bind.execute( sa.text( f"INSERT INTO {JOBS} " "(id, name, display_name, description, handler_key, " " trigger_type, cron_expr, timezone, enabled, config, " " run_count, failure_count, created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, 'cron', " ":cron, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " f"(SELECT 1 FROM {JOBS} WHERE handler_key = :handler)", ), { "id": str(uuid.uuid4()), "name": key, "display": display, "desc": desc, "handler": key, "cron": cron, "config": "{}", }, ) def upgrade() -> None: bind = op.get_bind() person_fk = id_type_for("identity_persons") task_fk = id_type_for("provisioning_tasks") # ── sso_grants: removal lifecycle ──────────────────────────────── if _has_table(bind, SSO): # Every step is guarded on its own: TiDB DDL is non-transactional, a # run that dies at column 4 must be able to resume at column 5. cols: list[tuple[str, sa.Column]] = [ ("removal_status", sa.Column( "removal_status", sa.String(24), nullable=False, server_default="not_applicable", )), ("removal_task_id", sa.Column("removal_task_id", task_fk, nullable=True)), ("removal_confirmed_at", sa.Column( "removal_confirmed_at", sa.DateTime(timezone=True), nullable=True, )), ("removal_confirmed_by_person_id", sa.Column( "removal_confirmed_by_person_id", person_fk, nullable=True, )), ("removal_external_ref", sa.Column( "removal_external_ref", sa.String(512), nullable=True, )), ("removal_verified_at", sa.Column( "removal_verified_at", sa.DateTime(timezone=True), nullable=True, )), ("removal_last_checked_at", sa.Column( "removal_last_checked_at", sa.DateTime(timezone=True), nullable=True, )), ("removal_last_login_at", sa.Column( "removal_last_login_at", sa.DateTime(timezone=True), nullable=True, )), ("removal_check_error", sa.Column( "removal_check_error", sa.String(255), nullable=True, )), ("removal_reminder_count", sa.Column( "removal_reminder_count", sa.Integer(), nullable=False, server_default="0", )), ("removal_last_reminded_at", sa.Column( "removal_last_reminded_at", sa.DateTime(timezone=True), nullable=True, )), ] for name, col in cols: if not _has_column(bind, SSO, name): op.add_column(SSO, col) if not _has_fk(bind, SSO, "fk_sso_grants_removal_task"): op.create_foreign_key( "fk_sso_grants_removal_task", SSO, "provisioning_tasks", ["removal_task_id"], ["id"], ondelete="SET NULL", ) if not _has_fk(bind, SSO, "fk_sso_grants_removal_confirmed_by"): op.create_foreign_key( "fk_sso_grants_removal_confirmed_by", SSO, "identity_persons", ["removal_confirmed_by_person_id"], ["id"], ondelete="SET NULL", ) # The reconcile cron's question, every morning: which revoked grants # still have work outstanding. if not _has_index(bind, SSO, "ix_sso_grants_removal_status"): op.create_index( "ix_sso_grants_removal_status", SSO, ["removal_status", "revoked_at"], ) # Backfill: rows revoked before this migration were never checked. # Say so — a separate visible state, not ``pending`` (which would # imply a task exists) and not ``confirmed`` (which would be a lie). bind.execute(sa.text( f"UPDATE {SSO} SET removal_status = 'unverified_legacy' " "WHERE status = 'revoked' AND removal_status = 'not_applicable'" )) # ── apikey_idle_alerts_dedup ───────────────────────────────────── if not _has_table(bind, IDLE): op.create_table( IDLE, sa.Column("id", id_type_for("identity_persons"), primary_key=True), sa.Column( "assignment_id", id_type_for("api_key_assignments"), sa.ForeignKey("api_key_assignments.id", ondelete="CASCADE"), nullable=False, ), # The calendar month that was $0, as ``YYYY-MM``. sa.Column("billing_month", sa.String(7), nullable=False), # How many consecutive $0 months this row represents (1 = first). sa.Column( "consecutive_months", sa.Integer(), nullable=False, server_default="1", ), sa.Column( "employee_notified", sa.Boolean(), nullable=False, server_default="0", ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.UniqueConstraint( "assignment_id", "billing_month", name="uq_apikey_idle_alerts_assignment_month", ), ) # ── platform settings ──────────────────────────────────────────── if _has_table(bind, SETTINGS): for key, value, description in _SETTINGS: bind.execute( sa.text( f"INSERT INTO {SETTINGS} (`key`, `value`, description, " " created_at, updated_at) " "SELECT :k, :v, :d, NOW(), NOW() FROM DUAL " f"WHERE NOT EXISTS (SELECT 1 FROM {SETTINGS} " " WHERE `key` = :k)" ), {"k": key, "v": json.dumps(value), "d": description}, ) # ── scheduled jobs ─────────────────────────────────────────────── if _has_table(bind, JOBS): _seed_job( bind, key=JOB_SSO, display="SSO 撤销落地 · 复核与催办", desc=_JOB_SSO_DESC, cron="40 8 * * *", ) _seed_job( bind, key=JOB_IDLE, display="API Key · 闲置检查", desc=_JOB_IDLE_DESC, cron="30 9 3 * *", ) def downgrade() -> None: bind = op.get_bind() if _has_table(bind, JOBS): for key in (JOB_SSO, JOB_IDLE): bind.execute( sa.text(f"DELETE FROM {JOBS} WHERE handler_key = :k"), {"k": key}, ) if _has_table(bind, SETTINGS): for key, _v, _d in _SETTINGS: bind.execute( sa.text(f"DELETE FROM {SETTINGS} WHERE `key` = :k"), {"k": key}, ) if _has_table(bind, IDLE): op.drop_table(IDLE) if _has_table(bind, SSO): if _has_index(bind, SSO, "ix_sso_grants_removal_status"): op.drop_index("ix_sso_grants_removal_status", table_name=SSO) if _has_fk(bind, SSO, "fk_sso_grants_removal_task"): op.drop_constraint("fk_sso_grants_removal_task", SSO, type_="foreignkey") if _has_fk(bind, SSO, "fk_sso_grants_removal_confirmed_by"): op.drop_constraint( "fk_sso_grants_removal_confirmed_by", SSO, type_="foreignkey", ) for name in ( "removal_last_reminded_at", "removal_reminder_count", "removal_check_error", "removal_last_login_at", "removal_last_checked_at", "removal_verified_at", "removal_external_ref", "removal_confirmed_by_person_id", "removal_confirmed_at", "removal_task_id", "removal_status", ): if _has_column(bind, SSO, name): op.drop_column(SSO, name)