"""离职账号「终态保障」的两张表 + 两个 cron。 在这之前,Helix 能证明的只有一件事:**我把我知道的权限撤了**。 它证明不了审计真正会问的那件事:**这个人现在还能不能进去。** 两者的差别在 2026-08-13 具体化了一次:Atlassian 侧的 Google 目录同步 被一次设置页保存关掉,而 Helix 的 Jira 撤权一直工作得好好的 —— 于是 连续 16 天,每个离职者的产品访问都是空的、账号却全都活着,而每日对账 天天报告「干净」。它看的是 `product_access`,而出事的是 `account_status`, 两个字段,两件事。 `offboarding_account_assurance` 把第二个问题落成数据:每个(人 × 身份 系统)一行,记下**厂商此刻怎么说**、Helix 做过什么、以及两者合起来是否 意味着这个人真的进不去了。由各系统的离职对账 cron 整体刷新(它们本来 每轮就要拉一次厂商全量名册),所以读接口是一次普通查询,不是一次对 厂商的扇出调用。 `identity_chain_heartbeats` 解决的是另一半:那次故障之所以 16 天不可见, 是因为唯一的信号是**某件事没有发生**,而没有人会盯着「没发生」。这张表 把每一条 Helix 依赖、但不拥有的身份链路变成一次正向心跳 + 一个期望周期; 扫描发现周期内没有心跳就置 stale 并告警 —— 不管那一周有没有人离职。 两个 cron: * `jira.idp_sync_heartbeat` —— 读 Atlassian 组织审计事件里的 `scim_auto_sync_complete`(真实 4 小时节奏),写心跳。 * `google.leaver_account_audit` —— 拉 Google Workspace 全量名册,和 Helix 的离职者对账,写 google 那一半的 assurance 行。它是这次唯一 查出「3 个离职 4~5 个月的人 Google 账号至今可登录、且过去 4 天还在 登录」的通道。 Revision ID: 575 Revises: 574 """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "575" down_revision = "574" branch_labels = None depends_on = None ASSURANCE_TABLE = "offboarding_account_assurance" HEARTBEAT_TABLE = "identity_chain_heartbeats" HEARTBEAT_JOB = "jira.idp_sync_heartbeat" GOOGLE_AUDIT_JOB = "google.leaver_account_audit" _HEARTBEAT_DESC = ( "盯住 Atlassian 侧的 Google Workspace / SCIM 用户供给还在不在跑。" "把离职者账号置为 inactive 的一直是这条同步,不是 Helix ——" "Helix 按既定策略只撤 group 与产品访问、永远保留账号(保 issue 历史" "归属)。2026-08-13 这条同步被一次设置页保存静默关掉,16 天没人发现," "期间所有离职者账号全活着、8-16 之后入职的人一个账号都没建。\n" "做法:读组织审计事件里的 scim_auto_sync_complete(真实节奏 4 小时" "一轮),超过期望周期没有新的一条 → 心跳置 stale + 告警 IT,并顺带" "报出最近一次 scim_sync_disabled 是谁、什么时候关的。" "它不依赖「恰好这周有人离职」—— 链路一断就叫,这正是上次缺的那半。" ) _GOOGLE_AUDIT_DESC = ( "拉 Google Workspace 全量名册,与 Helix 标记为 left 的人对账,回答" "「这个离职的人还能不能登录公司 Google 账号」,结果写进" "『账号终态』。\n" "已知会命中的三类:①离职流程从没跑过(早于该流程上线);②跑了但" "suspend 没确认(删号闸因此一直不放行);③被 external_account_registry" "的『保留』分类豁免掉 —— 而那条豁免可能既没有负责人也没有理由。" "三类在页面上分别显示,不混成一个数。" ) 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 _seed_job(bind, *, key: str, display: str, desc: str, cron: str) -> None: bind.execute( sa.text( "INSERT INTO scheduled_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, 'UTC', 1, '{}', 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)", ), { "id": str(uuid.uuid4()), "name": key, "display": display, "desc": desc, "handler": key, "cron": cron, }, ) def upgrade() -> None: conn = op.get_bind() if not sa.inspect(conn).has_table(ASSURANCE_TABLE): op.create_table( ASSURANCE_TABLE, sa.Column("id", id_type_for("identity_persons"), primary_key=True), sa.Column( "person_id", id_type_for("identity_persons"), nullable=False, ), # 'google' | 'atlassian' sa.Column("system", sa.String(32), nullable=False), # absent | disabled | active | unknown ← 厂商此刻的真话 sa.Column( "vendor_state", sa.String(16), nullable=False, server_default="unknown", ), sa.Column("vendor_account_id", sa.String(128), nullable=True), # Helix 自己那一侧做过什么(从离职台账抄过来) sa.Column("helix_action", sa.String(48), nullable=True), # ok | at_risk | exempt | unknown sa.Column( "verdict", sa.String(16), nullable=False, server_default="unknown", ), # 只在 exempt 时有值:**哪一条决定**在把这个账号留着。 # 说不出是谁的决定的豁免不是决定,是缺口 —— 这一列就是让 # 这个差别在页面上看得见的东西。 sa.Column("exempt_reason", sa.String(64), nullable=True), sa.Column("detail", sa.JSON(), nullable=True), # 厂商最后一次被读到的时间。它自己变旧就是一个结论 # (扫描停了),所以单独一列、不藏在 updated_at 后面。 sa.Column("evidence_at", sa.DateTime(timezone=True), nullable=True), # 第一次被判定 at_risk 的时间 —— 「这个口子开了多久」 # 才是让人真的去处理的那个数字。 sa.Column( "at_risk_since", sa.DateTime(timezone=True), nullable=True, ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), onupdate=sa.text("CURRENT_TIMESTAMP"), ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], ondelete="CASCADE", ), sa.UniqueConstraint( "person_id", "system", name="uq_account_assurance_person_system", ), **table_kw_for("identity_persons"), ) op.create_index( "ix_account_assurance_verdict", ASSURANCE_TABLE, ["verdict"], ) op.create_index( "ix_account_assurance_system", ASSURANCE_TABLE, ["system"], ) if not sa.inspect(conn).has_table(HEARTBEAT_TABLE): op.create_table( HEARTBEAT_TABLE, sa.Column("id", id_type_for("identity_persons"), primary_key=True), sa.Column("chain_key", sa.String(64), nullable=False), # ok | stale | unknown sa.Column( "status", sa.String(16), nullable=False, server_default="unknown", ), # 厂商侧最新的一次心跳(**不是**「Helix 上次检查」) sa.Column("last_beat_at", sa.DateTime(timezone=True), nullable=True), sa.Column("checked_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "expected_period_minutes", sa.Integer(), nullable=False, server_default="240", ), sa.Column("detail", sa.JSON(), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), onupdate=sa.text("CURRENT_TIMESTAMP"), ), sa.UniqueConstraint("chain_key", name="uq_identity_chain_key"), **table_kw_for("identity_persons"), ) if _has_table(conn, "scheduled_jobs"): # 心跳每小时跑:同步本身 4 小时一轮,按小时看能把「断了」的发现 # 时间压到一个周期以内,而它一次只读一页审计事件,极便宜。 _seed_job( conn, key=HEARTBEAT_JOB, display="Atlassian · 身份目录同步心跳", desc=_HEARTBEAT_DESC, cron="7 * * * *", ) # 03:20 UTC —— 排在 jira.leaver_account_audit(03:30)之前, # 这样『账号终态』页上两个系统的证据是同一个早上的。 _seed_job( conn, key=GOOGLE_AUDIT_JOB, display="Google · 离职账号终态对账", desc=_GOOGLE_AUDIT_DESC, cron="20 3 * * *", ) def downgrade() -> None: conn = op.get_bind() if _has_table(conn, "scheduled_jobs"): conn.execute( sa.text("DELETE FROM scheduled_jobs WHERE handler_key IN (:a, :b)"), {"a": HEARTBEAT_JOB, "b": GOOGLE_AUDIT_JOB}, ) if sa.inspect(conn).has_table(HEARTBEAT_TABLE): op.drop_table(HEARTBEAT_TABLE) if sa.inspect(conn).has_table(ASSURANCE_TABLE): op.drop_index("ix_account_assurance_system", ASSURANCE_TABLE) op.drop_index("ix_account_assurance_verdict", ASSURANCE_TABLE) op.drop_table(ASSURANCE_TABLE)