"""Internal system registry + the bounty report's link to it. Two tables (``registry_systems``, ``registry_system_events``) plus ``bounty_reports.system_id``, and a seed of the systems that already appear in real vulnerability correspondence so the register is not born empty. Why a new table rather than reusing CMDB: ``cmdb_*`` is provider-synced machine inventory (servers, domains, DNS, certs), ``asset_software_catalog`` is purchased SaaS with seats, ``sso_apps`` is a SAML catalogue, and ``domain_registrations`` records who registered an external domain. None of them answers "who is responsible for the TiCard application" — which is the question an inbound vulnerability report asks. Revision ID: 542 Revises: 541 """ 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, table_kw_for revision = "542" down_revision = "541" branch_labels = None depends_on = None SYSTEMS = "registry_systems" EVENTS = "registry_system_events" REPORTS = "bounty_reports" #: Systems that already appear in real vulnerability correspondence #: (transcribed 2026-08-25 from the security mailbox), plus the two whose #: owners the platform owner named. Seeded WITHOUT owners: an invented #: owner is worse than an honest blank, and the ownerless read exists to #: chase exactly these. ``xuhuaiyu`` / ``xiaopeng.zhu`` are set by email #: because those two were given explicitly. SEED = ( # slug, name, kind, exposure, hostnames, in_scope, track, owner_email, note ("ticard", "TiCard", "demo", "public", ["ticard.pingcap.com"], True, "product", "xiaopeng.zhu@pingcap.com", "活动现场制作参会证的实验性半公开系统。不承载核心业务或生产负载 —— " "这一点在赏金定级里被反复引用,所以写在台账上而不是每次凭记忆解释。"), ("clinic", "PingCAP Clinic", "support", "authenticated", ["clinic.pingcap.com"], True, "product", "xuhuaiyu@pingcap.com", "客户诊断数据采集与分析。接触生产 S3 与客户诊断包,外部可达面需要授权校验。"), ("tidb-cloud", "TiDB Cloud", "product", "authenticated", ["tidbcloud.com", "tidb-cloud.com", "tidbapi.com"], True, "product", None, "核心产品面。赏金政策里明确列为高优先范围。"), ("pingcap-www", "PingCAP 官网", "marketing", "public", ["www.pingcap.com", "pingcap.com"], True, "it", None, "市场官网(WordPress)。历史报告集中在目录列举、XML-RPC、" "分析脚本 key、DNS 配置一类,归 IT 预算。"), ("jira-crm", "jira-crm", "internal_tool", "authenticated", ["jira-crm.tidbcloud.com"], False, "product", None, "内部工具。曾被报告在生产暴露 Werkzeug 调试控制台。"), ("one-engine", "One Engine Demo", "demo", "public", ["one-engine.pingcap.com"], True, "product", None, "TiDB Cloud Serverless 演示应用。曾被报告 /api/oltp 未认证 SQL 注入。"), ("docs", "PingCAP Docs", "marketing", "public", ["docs.pingcap.com"], True, "it", None, "文档站。"), ("labs", "PingCAP Labs", "demo", "public", ["labs.pingcap.com"], True, "product", None, "实验性站点。曾被报告开放重定向与 XSS。"), ) def _table_exists(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 _column_exists(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 _index_exists(bind, table: str, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.statistics " "WHERE table_schema = DATABASE() AND table_name = :t " "AND index_name = :n LIMIT 1" ), {"t": table, "n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() # Every FK here points at ``identity_persons`` or an intra-file # parent, so one table charset satisfies all of them. pid = id_type_for("identity_persons") tkw = table_kw_for("identity_persons") if not _table_exists(bind, SYSTEMS): op.create_table( SYSTEMS, sa.Column("id", pid, nullable=False), sa.Column("slug", sa.String(64), nullable=False), sa.Column("name", sa.String(128), nullable=False), sa.Column("kind", sa.String(24), nullable=False, server_default="other"), sa.Column("status", sa.String(16), nullable=False, server_default="active"), sa.Column("exposure", sa.String(16), nullable=False, server_default="unknown"), sa.Column("hostnames", sa.JSON), sa.Column("description", sa.Text), sa.Column("repo", sa.String(255)), sa.Column("owner_person_id", pid), sa.Column("backup_owner_person_id", pid), sa.Column("owner_department", sa.String(255)), sa.Column("owner_assigned_at", sa.DateTime(timezone=True)), sa.Column("budget_track", sa.String(16), nullable=False, server_default="product"), sa.Column("in_bounty_scope", sa.Boolean, nullable=False, server_default="0"), sa.Column("notes", sa.Text), sa.Column("source", sa.String(24), nullable=False, server_default="manual"), sa.Column("created_by", pid), 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()), sa.Column("deleted_at", sa.DateTime(timezone=True)), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("slug", name="uq_registry_systems_slug"), sa.ForeignKeyConstraint(["owner_person_id"], ["identity_persons.id"], ondelete="SET NULL"), sa.ForeignKeyConstraint(["backup_owner_person_id"], ["identity_persons.id"], ondelete="SET NULL"), sa.ForeignKeyConstraint(["created_by"], ["identity_persons.id"], ondelete="SET NULL"), **tkw, ) op.create_index("ix_registry_systems_owner", SYSTEMS, ["owner_person_id", "deleted_at"]) op.create_index("ix_registry_systems_status", SYSTEMS, ["status", "deleted_at"]) op.create_index("ix_registry_systems_exposure", SYSTEMS, ["exposure", "status"]) op.create_index("ix_registry_systems_scope", SYSTEMS, ["in_bounty_scope", "status"]) if not _table_exists(bind, EVENTS): op.create_table( EVENTS, sa.Column("id", pid, nullable=False), sa.Column("system_id", pid, nullable=False), sa.Column("event_type", sa.String(32), nullable=False), sa.Column("actor_person_id", pid), sa.Column("actor_name", sa.String(255)), sa.Column("body", sa.Text), sa.Column("meta", sa.JSON), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint(["system_id"], [f"{SYSTEMS}.id"], ondelete="CASCADE"), sa.ForeignKeyConstraint(["actor_person_id"], ["identity_persons.id"], ondelete="SET NULL"), **tkw, ) op.create_index("ix_registry_events_system", EVENTS, ["system_id", "created_at"]) # ── The bounty link ── if _table_exists(bind, REPORTS) and not _column_exists( bind, REPORTS, "system_id", ): # Nullable on purpose: ``affected_system`` free text stays the # fallback (a report can name a host we have never registered), # and backfilling a guess would put a wrong owner on real reports. op.add_column(REPORTS, sa.Column("system_id", pid)) op.create_foreign_key( "fk_bounty_reports_system", REPORTS, SYSTEMS, ["system_id"], ["id"], ondelete="SET NULL", ) op.create_index("ix_bounty_reports_system", REPORTS, ["system_id", "status"]) _seed(bind) def _seed(bind) -> None: """Insert any missing seed system. Existing slugs are left untouched — an operator's edited row must survive a re-run.""" existing = { r[0] for r in bind.execute(sa.text(f"SELECT slug FROM {SYSTEMS}")).all() } for (slug, name, kind, exposure, hosts, in_scope, track, owner_email, note) in SEED: if slug in existing: continue owner_id = None owner_name = None department = None if owner_email: row = bind.execute( sa.text( "SELECT id, display_name FROM identity_persons " "WHERE work_email = :e AND deleted_at IS NULL LIMIT 1" ), {"e": owner_email}, ).first() if row is None: print(f"[542] {slug}: owner {owner_email} not found — left blank") else: owner_id = row[0] owner_name = row[1] department = bind.execute( sa.text( "SELECT org_node_name FROM identity_org_memberships " "WHERE person_id = :p AND is_primary = 1 " "AND membership_status = 'active' " "AND deleted_at IS NULL LIMIT 1" ), {"p": owner_id}, ).scalar() system_id = str(uuid.uuid4()) bind.execute( sa.text( f"INSERT INTO {SYSTEMS} " "(id, slug, name, kind, status, exposure, hostnames, notes, " " owner_person_id, owner_department, owner_assigned_at, " " budget_track, in_bounty_scope, source, created_at, updated_at) " "VALUES (:id, :slug, :name, :kind, 'active', :exposure, :hosts, " " :notes, :owner, :dept, " " CASE WHEN :owner2 IS NULL THEN NULL ELSE UTC_TIMESTAMP() END, " " :track, :in_scope, 'seed', UTC_TIMESTAMP(), UTC_TIMESTAMP())" ), { "id": system_id, "slug": slug, "name": name, "kind": kind, "exposure": exposure, "hosts": json.dumps(hosts), "notes": note, "owner": owner_id, "owner2": owner_id, "dept": department, "track": track, "in_scope": 1 if in_scope else 0, }, ) # Seeded rows get their trail too. Without it the detail page # shows an ownership date with an empty history, which reads as # "nothing was ever recorded" rather than "recorded at seed # time" — and the trail is the reason the events table exists. _event(bind, system_id, "created", {"slug": slug, "source": "seed"}) if owner_id is not None: _event( bind, system_id, "owner_assigned", { "from_person_id": None, "from_name": None, # Name snapshotted at write time, like every other # event: the trail must still read correctly after # the person leaves and the join goes away. "to_person_id": str(owner_id), "to_name": owner_name, "source": "seed", }, ) def _event(bind, system_id: str, event_type: str, meta: dict) -> None: bind.execute( sa.text( f"INSERT INTO {EVENTS} " "(id, system_id, event_type, meta, created_at) " "VALUES (:id, :sid, :t, :meta, UTC_TIMESTAMP())" ), { "id": str(uuid.uuid4()), "sid": system_id, "t": event_type, "meta": json.dumps(meta), }, ) def downgrade() -> None: bind = op.get_bind() # The bounty link first, and STRICTLY in this order: FK → index → # column. MySQL/TiDB back the FK with the composite index whose # leading column is ``system_id``, so dropping the index while the # constraint still exists fails with 1553 ("needed in a foreign key # constraint") — verified against a real downgrade, not assumed. if _column_exists(bind, REPORTS, "system_id"): try: op.drop_constraint("fk_bounty_reports_system", REPORTS, type_="foreignkey") except Exception: # pragma: no cover - already absent pass if _index_exists(bind, REPORTS, "ix_bounty_reports_system"): op.drop_index("ix_bounty_reports_system", table_name=REPORTS) op.drop_column(REPORTS, "system_id") # Children before parents. for name in (EVENTS, SYSTEMS): if _table_exists(bind, name): op.drop_table(name)