"""TLS/SSL certificate inventory + expiry-alert dedup state. Revision ID: 267 Revises: 266 Create Date: 2026-06-09 (Renumbered 266 → 267 during merge — PR #862 ``266_email_group_is_dynamic.py`` took 266 first.) Why this exists --------------- Helix tracked domains (``cmdb_domains.expiry_date`` + the ``cmdb.domain_expiry_alert`` cron) and DNS records, but had **zero** visibility into the TLS/SSL certificates those hostnames actually serve. A certificate that lapses turns a working ``https://`` site into a browser "not secure" wall — and unlike a domain renewal (one click at the registrar) a cert must be re-issued AND re-deployed, so it deserves EARLIER, owner-routed reminders. This migration adds two tables: ``cmdb_certificates`` One row per observed certificate. Three provenance sources (``source`` column): * ``probed`` — Helix opened a TLS connection to ``probe_host:probe_port`` (SNI = host) and read the leaf cert the endpoint served. Covers every PUBLIC https endpoint regardless of where the cert lives. Target list is built from ``cmdb_dns_records`` (A/AAAA/CNAME) + ``cmdb_domains`` by the ``cmdb.cert_probe_sweep`` cron. * ``cloud`` — synced from a cloud cert manager (AWS ACM / Aliyun / Tencent SSL) — added in PR2. * ``manual`` — admin-entered for a cert on a box Helix can't reach. ``dedup_key`` is the idempotent upsert key (UNIQUE): * probed → ``probed:{host}:{port}`` * cloud → ``cloud:{integration_id}:{cloud_cert_id}`` * manual → ``manual:{uuid}`` A renewal re-issues the cert (``fingerprint_sha256`` + ``not_after`` change) but keeps the SAME dedup_key, so the upsert UPDATES the row and the moved-forward ``not_after`` auto-clears the alert state. ``cmdb_cert_expiry_alert_state`` Per-certificate dedup for the ``cmdb.cert_expiry_alert`` cron — identical shape to ``cmdb_domain_expiry_alert_state``. The cron only re-fires a (cert, owner) when the bucket transitions (``safe`` / ``<=30d`` / ``<=7d`` / ``<=1d`` / ``expired``) or the owner changes. It also seeds the ``network.owner_person_id`` platform setting (the "网络负责人" — catch-all owner for owner-less domains / DNS / certs and the fallback recipient for their expiry alerts). Best-effort resolves 孟凡生 by display name; leaves the row absent when not found (fresh local / CI DBs) so the resolver falls back gracefully. Idempotency ----------- ``CREATE TABLE IF NOT EXISTS`` semantics via ``_table_exists``. Seed is guarded by an existence check. """ 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, normalize_fk_column_collation, table_kw_for, ) revision = "267" down_revision = "266" branch_labels = None depends_on = None CERT_TABLE = "cmdb_certificates" STATE_TABLE = "cmdb_cert_expiry_alert_state" NETWORK_OWNER_KEY = "network.owner_person_id" # Two crons seeded with this feature (hyphenated UUID() id to avoid the # no-dash-UUID attach bug — migrations 200/248). _PROBE_HANDLER = "cmdb.cert_probe_sweep" _ALERT_HANDLER = "cmdb.cert_expiry_alert" 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 upgrade() -> None: bind = op.get_bind() persons_id = id_type_for("identity_persons") integ_id = id_type_for("integrations") kw = table_kw_for("identity_persons") if not _table_exists(bind, CERT_TABLE): op.create_table( CERT_TABLE, sa.Column("id", persons_id, primary_key=True), sa.Column("dedup_key", sa.String(512), nullable=False), sa.Column("source", sa.String(16), nullable=False), # NULL for probed / manual; set for cloud-synced certs. sa.Column("integration_id", integ_id, nullable=True), sa.Column("cloud_cert_id", sa.String(256), nullable=True), # Probe coordinates (NULL for cloud / manual). sa.Column("probe_host", sa.String(256), nullable=True), sa.Column( "probe_port", sa.Integer(), nullable=False, server_default=sa.text("443"), ), # Parsed certificate fields. sa.Column("common_name", sa.String(256), nullable=True), sa.Column("sans", sa.JSON(), nullable=True), sa.Column("issuer", sa.String(256), nullable=True), sa.Column("serial_number", sa.String(128), nullable=True), sa.Column("fingerprint_sha256", sa.String(64), nullable=True), sa.Column("not_before", sa.DateTime(timezone=True), nullable=True), sa.Column("not_after", sa.DateTime(timezone=True), nullable=True), sa.Column("key_algorithm", sa.String(32), nullable=True), sa.Column("key_bits", sa.Integer(), nullable=True), sa.Column("signature_algorithm", sa.String(64), nullable=True), # valid / expiring / expired / unreachable / unknown sa.Column( "status", sa.String(24), nullable=False, server_default="unknown", ), sa.Column("last_error", sa.Text(), nullable=True), sa.Column("owner_person_id", persons_id, nullable=True), sa.Column("environment", sa.String(16), nullable=True), sa.Column("raw_data", sa.JSON(), nullable=True), sa.Column("last_seen_sync_job_id", sa.String(36), nullable=True), sa.Column("last_probed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.Column("remark", sa.Text(), nullable=True), sa.Column("last_modified_by_person_id", persons_id, nullable=True), sa.Column("last_modified_at", sa.DateTime(timezone=True), nullable=True), sa.ForeignKeyConstraint( ["owner_person_id"], ["identity_persons.id"], name="fk_cmdb_certificates_owner", ondelete="SET NULL", ), sa.ForeignKeyConstraint( ["last_modified_by_person_id"], ["identity_persons.id"], name="fk_cmdb_certificates_modifier", ondelete="SET NULL", ), sa.UniqueConstraint("dedup_key", name="uq_cmdb_certificates_dedup"), **kw, ) # integration_id FK created separately so we can normalize its # collation to integrations.id first (TiDB error 3780 guard). normalize_fk_column_collation( CERT_TABLE, "integration_id", "integrations", nullable=True, ) op.create_foreign_key( "fk_cmdb_certificates_integration", CERT_TABLE, "integrations", ["integration_id"], ["id"], ondelete="CASCADE", ) op.create_index( "ix_cmdb_certificates_owner", CERT_TABLE, ["owner_person_id"], ) op.create_index( "ix_cmdb_certificates_not_after", CERT_TABLE, ["not_after"], ) op.create_index( "ix_cmdb_certificates_status", CERT_TABLE, ["status"], ) op.create_index( "ix_cmdb_certificates_integration", CERT_TABLE, ["integration_id"], ) if not _table_exists(bind, STATE_TABLE): op.create_table( STATE_TABLE, sa.Column("cert_id", sa.String(36), primary_key=True), sa.Column("last_bucket", sa.String(16), nullable=False), sa.Column("last_owner_person_id", sa.String(36), nullable=True), sa.Column( "last_alerted_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), **kw, ) # Seed the network-owner platform setting (best-effort resolve 孟凡生). exists = bind.execute( sa.text("SELECT 1 FROM platform_settings WHERE `key` = :k LIMIT 1"), {"k": NETWORK_OWNER_KEY}, ).scalar() if not exists: owner_id = bind.execute( sa.text( "SELECT id FROM identity_persons " "WHERE display_name = :n AND employment_status = 'active' " "ORDER BY created_at LIMIT 1" ), {"n": "孟凡生"}, ).scalar() if owner_id is not None: bind.execute( sa.text( "INSERT INTO platform_settings (`key`, value, description) " "VALUES (:k, :v, :d)" ), { "k": NETWORK_OWNER_KEY, # value is a JSON column — store the person id as a JSON string. "v": json.dumps(str(owner_id)), "d": "网络负责人 person_id — 域名/DNS/证书无主对象的兜底负责人与到期告警兜底接收人。", }, ) # Seed the two crons (idempotent — skip when handler_key exists). if _table_exists(bind, "scheduled_jobs"): _seed_cron( bind, handler=_PROBE_HANDLER, display="证书 · 探测扫描", desc=( "每天 03:00 UTC 对启用租户 DNS(A/AAAA/CNAME)+ 域名 apex 主机" "逐个 443 TLS 握手读取叶子证书并 upsert 到 cmdb_certificates。" ), cron="0 3 * * *", config={"concurrency": 20, "timeout_seconds": 8}, ) _seed_cron( bind, handler=_ALERT_HANDLER, display="证书 · 到期提醒", desc=( "每天 04:00 UTC 按 owner 推送证书到期飞书私信(<=30d/<=7d/<=1d/" "expired,状态表 dedup,无主兜底网络负责人)。受 " "network.user_notifications_enabled 控制。" ), cron="0 4 * * *", config={"expired_lookback_days": 30}, ) def _seed_cron(bind, *, handler, display, desc, cron, config) -> 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, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)" ), { "id": str(uuid.uuid4()), "name": handler, "display": display, "desc": desc, "handler": handler, "cron": cron, "config": json.dumps(config), }, ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE handler_key IN (:a, :b)"), {"a": _PROBE_HANDLER, "b": _ALERT_HANDLER}, ) if _table_exists(bind, STATE_TABLE): op.drop_table(STATE_TABLE) if _table_exists(bind, CERT_TABLE): op.drop_table(CERT_TABLE) bind.execute( sa.text("DELETE FROM platform_settings WHERE `key` = :k"), {"k": NETWORK_OWNER_KEY}, )