"""Graduated expiry-alert engine — shared by the cert + domain crons. Replaces the old "one Feishu DM per coarse bucket transition, owner only" model with a production-grade lifecycle: cadence T-30 once → T-14 once → DAILY inside T-7 → DAILY while expired recipients 本人 always; inside the last-mile window (≤7d / expired) ALSO escalate to the 域名负责人 (zone domain owner) + the lead (本人's direct manager) ack/snooze an owner who marks the resource "renewing" suppresses all alerts (incl. escalation) through ``acknowledged_until`` first-enable when the notification kill switch is first turned on, the existing expired/expiring BACKLOG must not carpet-bomb — a resource first observed already inside the escalation window is stamped (``escalation_started_on`` left NULL) and stays suppressed; only resources that cross a threshold UNDER the cron's watch (go-forward) escalate. The engine is resource-agnostic: callers build a list of :class:`ExpiryItem` (certs via ``CertOwnerResolver``, domains via their own owner) and call :func:`run_expiry_alerts`. Notification delivery (in-app inbox bell + a best-effort Feishu DM carrying the one-click "mark renewing" magic link) goes through the governed notification outbox. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import UTC, date, datetime, timedelta from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import CmdbExpiryAlertState from app.identity.person_models import IdentityPerson # Bucket labels. The two "once" buckets fire a single alert on entry; the # two escalation buckets fire DAILY and fan out to the escalation tier. BUCKET_SAFE = "safe" BUCKET_30D = "<=30d" BUCKET_14D = "<=14d" BUCKET_7D = "<=7d" BUCKET_EXPIRED = "expired" ESCALATION_BUCKETS = frozenset({BUCKET_7D, BUCKET_EXPIRED}) NOTIFY_BUCKETS = frozenset({BUCKET_30D, BUCKET_14D, BUCKET_7D, BUCKET_EXPIRED}) # Default snooze length when an owner marks a resource "renewing". DEFAULT_ACK_DAYS = 14 @dataclass class ExpiryItem: """One thing that can expire, with its resolved responsibility chain.""" resource_type: str # EXPIRY_RESOURCE_CERTIFICATE | EXPIRY_RESOURCE_DOMAIN resource_id: str label: str # human label (CN / domain_name) expiry_date: date | None #: 本人 — the primary owner (cert: resolved record/domain owner; domain: #: the domain owner). May be ``None`` (owner-less). primary_owner_id: str | None #: 域名负责人 — the zone domain owner, escalated to in the last-mile #: window. ``None`` for domains (the domain owner IS 本人) or unknown. domain_owner_id: str | None = None #: Extra context for the notification body (e.g. probe_host). detail: str | None = None #: True when this resource was first discovered <24h ago. A fresh + #: already-urgent resource alerts on first sight rather than waiting #: for a bucket transition it may never get (e.g. born expired). Old #: backlog (not fresh) stays suppressed on first observation — the #: first-enable storm guard. first_seen_fresh: bool = False @dataclass class _Recipient: person: IdentityPerson role: str # "owner" | "domain_owner" | "lead" def bucket_for(expiry: date | None, today: date, expired_lookback_days: int) -> str: """Cadence buckets — see module docstring. >30d safe · 15-30 ``<=30d`` (once) · 8-14 ``<=14d`` (once) · 0-7 ``<=7d`` (daily, escalate) · expired-within-lookback (daily, escalate). """ if expiry is None: return BUCKET_SAFE delta = (expiry - today).days if delta < 0: within = (today - expiry).days <= expired_lookback_days return BUCKET_EXPIRED if within else BUCKET_SAFE if delta <= 7: return BUCKET_7D if delta <= 14: return BUCKET_14D if delta <= 30: return BUCKET_30D return BUCKET_SAFE async def _resolve_people( db: AsyncSession, ids: set[str], ) -> dict[str, IdentityPerson]: clean = {str(i) for i in ids if i} if not clean: return {} rows = list((await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(list(clean))) )).scalars().all()) return {str(p.id): p for p in rows} def _is_active(p: IdentityPerson | None) -> bool: return p is not None and p.employment_status == "active" @dataclass class _Decision: """What to do with one item this tick.""" send: bool = False escalate: bool = False # include domain_owner + lead new_bucket: str = BUCKET_SAFE new_escalation_started_on: date | None = None counters: list[str] = field(default_factory=list) def _decide( *, bucket: str, today: date, prev: CmdbExpiryAlertState | None, effective_owner: str | None, first_seen_fresh: bool, ) -> _Decision: """Pure state-machine decision for one item (no I/O). ``first_seen_fresh`` = the resource was discovered <24h ago (so a genuinely-new urgent resource still alerts on first sight rather than waiting for a transition that will never come). """ prev_bucket = prev.last_bucket if prev else None prev_esc_started = prev.escalation_started_on if prev else None prev_alerted_on = prev.last_alerted_on if prev else None prev_owner = str(prev.last_owner_person_id) if (prev and prev.last_owner_person_id) else None # Acknowledged / snoozed → suppress everything until the ack expires. if prev and prev.acknowledged_until and prev.acknowledged_until >= today: return _Decision(send=False, new_bucket=bucket, new_escalation_started_on=prev_esc_started, counters=["acknowledged_suppressed"]) if bucket == BUCKET_SAFE: # Renewed / far out → reset escalation arming so a future expiry # re-arms cleanly. return _Decision(send=False, new_bucket=BUCKET_SAFE, new_escalation_started_on=None) owner_changed = prev is not None and prev_owner != (str(effective_owner) if effective_owner else None) in_escalation = bucket in ESCALATION_BUCKETS # First observation ever: suppress the backlog (stamp only) UNLESS a # freshly-discovered urgent resource that would otherwise never get a # transition before it lapses. if prev is None: if first_seen_fresh and in_escalation: return _Decision(send=True, escalate=True, new_bucket=bucket, new_escalation_started_on=today, counters=["first_observation_emitted"]) return _Decision(send=False, new_bucket=bucket, new_escalation_started_on=None, counters=["suppressed_first_observation"]) if not in_escalation: # Normal "once" bucket (<=30d / <=14d): fire on entry or owner change. if prev_bucket != bucket or owner_changed: return _Decision(send=True, escalate=False, new_bucket=bucket, new_escalation_started_on=None) return _Decision(send=False, new_bucket=bucket, new_escalation_started_on=None, counters=["deduped_unchanged"]) # Escalation window (<=7d / expired): DAILY while armed. entered_now = prev_bucket not in ESCALATION_BUCKETS if entered_now: # Go-forward crossing into the window under our watch → arm + send. # NOTE: a backlog cert that was suppressed while ≤7d and then # expires WITHOUT first going safe stays in ESCALATION_BUCKETS the # whole time, so it never hits this branch — intended first-enable # suppression. A renew→safe→re-expire round-trip DOES pass through # ``safe`` (prev_bucket='safe'), so it correctly re-arms here. return _Decision(send=True, escalate=True, new_bucket=bucket, new_escalation_started_on=today) # Already in the window. if prev_esc_started is None: # Seeded backlog (first-enable) — stay suppressed. return _Decision(send=False, new_bucket=bucket, new_escalation_started_on=None, counters=["backlog_suppressed"]) # Owner changed mid-window → the NEW owner must be paged now, even if # the old owner was already alerted today. if prev_alerted_on == today and not owner_changed: return _Decision(send=False, escalate=True, new_bucket=bucket, new_escalation_started_on=prev_esc_started, counters=["deduped_today"]) # New day in an armed window (or owner change) → escalated resend. return _Decision(send=True, escalate=True, new_bucket=bucket, new_escalation_started_on=prev_esc_started) async def run_expiry_alerts( db: AsyncSession, *, resource_type: str, items: list[ExpiryItem], config: dict[str, Any], notify: Any, ) -> dict[str, Any]: """Drive the cadence/escalation/ack state machine over ``items``. ``notify`` is an async callable ``(item, recipients, bucket) -> bool`` returning whether at least one channel was delivered — injected so the crons wire it to the real Feishu+inbox sender and tests can assert on recipients without external I/O. """ today = datetime.now(UTC).date() expired_lookback_days = max( 1, min(90, int(config.get("expired_lookback_days") or 30)), ) summary: dict[str, Any] = { "scanned": 0, "owners_notified": 0, "escalated": 0, "skipped_no_owner": 0, "skipped_leaver": 0, "send_failures": 0, "notify_buckets": {BUCKET_30D: 0, BUCKET_14D: 0, BUCKET_7D: 0, BUCKET_EXPIRED: 0}, } if not items: return summary # Prior state for all items. ids = [it.resource_id for it in items] states = list((await db.execute( select(CmdbExpiryAlertState).where( CmdbExpiryAlertState.resource_type == resource_type, CmdbExpiryAlertState.resource_id.in_(ids), ) )).scalars().all()) state_by_id = {s.resource_id: s for s in states} # Resolve every candidate person (owners + domain owners + their # managers) in one batch, then their managers in a second. owner_ids = {it.primary_owner_id for it in items if it.primary_owner_id} dom_owner_ids = {it.domain_owner_id for it in items if it.domain_owner_id} people = await _resolve_people(db, owner_ids | dom_owner_ids) manager_ids = { str(p.manager_person_id) for p in people.values() if p.manager_person_id } people |= await _resolve_people(db, manager_ids) now = datetime.now(UTC) for it in items: summary["scanned"] += 1 bucket = bucket_for(it.expiry_date, today, expired_lookback_days) prev = state_by_id.get(it.resource_id) owner = people.get(str(it.primary_owner_id)) if it.primary_owner_id else None # Owner-less: nothing to DM. Record state so a future # owner-attribution + threshold pair fires cleanly; surfaced in the # admin governance console regardless. if it.primary_owner_id is None or owner is None: summary["skipped_no_owner"] += 1 _stamp(db, prev, resource_type, it.resource_id, bucket, owner_id=None, alerted_on=None, escalation_started_on=None, now=now) continue decision = _decide( bucket=bucket, today=today, prev=prev, effective_owner=it.primary_owner_id, first_seen_fresh=it.first_seen_fresh, ) for c in decision.counters: summary[c] = summary.get(c, 0) + 1 if not decision.send: # Leaver owner in a notify bucket is worth surfacing as a skip. if decision.new_bucket in NOTIFY_BUCKETS and not _is_active(owner) \ and "acknowledged_suppressed" not in decision.counters: summary["skipped_leaver"] += 1 _stamp(db, prev, resource_type, it.resource_id, decision.new_bucket, owner_id=str(it.primary_owner_id), alerted_on=(prev.last_alerted_on if prev else None), escalation_started_on=decision.new_escalation_started_on, now=now) continue # Build the recipient list. 本人 always; escalate adds 域名负责人 + lead. recipients = _build_recipients(it, owner, people, escalate=decision.escalate) active = [r for r in recipients if _is_active(r.person)] if not active: summary["skipped_leaver"] += 1 _stamp(db, prev, resource_type, it.resource_id, decision.new_bucket, owner_id=str(it.primary_owner_id), alerted_on=(prev.last_alerted_on if prev else None), escalation_started_on=decision.new_escalation_started_on, now=now) continue try: sent = await notify(it, active, decision.new_bucket) except Exception: # noqa: BLE001 — one bad send must not abort the batch sent = False if not sent: summary["send_failures"] += 1 # Do NOT advance last_alerted_on on a failed send, so the next # tick retries; but still stamp the bucket so transition logic # stays consistent. _stamp(db, prev, resource_type, it.resource_id, decision.new_bucket, owner_id=str(it.primary_owner_id), alerted_on=(prev.last_alerted_on if prev else None), escalation_started_on=decision.new_escalation_started_on, now=now) continue summary["owners_notified"] += 1 if decision.escalate: summary["escalated"] += 1 if decision.new_bucket in summary["notify_buckets"]: summary["notify_buckets"][decision.new_bucket] += 1 _stamp(db, prev, resource_type, it.resource_id, decision.new_bucket, owner_id=str(it.primary_owner_id), alerted_on=today, escalation_started_on=decision.new_escalation_started_on, now=now) await db.flush() return summary def _build_recipients( it: ExpiryItem, owner: IdentityPerson, people: dict[str, IdentityPerson], *, escalate: bool, ) -> list[_Recipient]: recipients: list[_Recipient] = [_Recipient(owner, "owner")] if not escalate: return recipients seen = {str(owner.id)} # 域名负责人 (zone domain owner), if distinct from 本人. if it.domain_owner_id and str(it.domain_owner_id) not in seen: dom = people.get(str(it.domain_owner_id)) if dom is not None: recipients.append(_Recipient(dom, "domain_owner")) seen.add(str(dom.id)) # lead = 本人's direct manager, if distinct. if owner.manager_person_id and str(owner.manager_person_id) not in seen: lead = people.get(str(owner.manager_person_id)) if lead is not None: recipients.append(_Recipient(lead, "lead")) seen.add(str(lead.id)) return recipients def _stamp( db: AsyncSession, prev: CmdbExpiryAlertState | None, resource_type: str, resource_id: str, bucket: str, *, owner_id: str | None, alerted_on: date | None, escalation_started_on: date | None, now: datetime, ) -> None: """Upsert the dedup/escalation state row (never touches ack fields).""" if prev is None: db.add(CmdbExpiryAlertState( resource_type=resource_type, resource_id=resource_id, last_bucket=bucket, last_owner_person_id=owner_id, last_alerted_on=alerted_on, escalation_started_on=escalation_started_on, last_alerted_at=now, )) return prev.last_bucket = bucket prev.last_owner_person_id = owner_id prev.last_alerted_on = alerted_on prev.escalation_started_on = escalation_started_on prev.last_alerted_at = now # ── Acknowledge / snooze ─────────────────────────────────────────────── async def resolve_expiry_owners( db: AsyncSession, resource_type: str, resource_id: str, ) -> set[str]: """The set of person ids responsible for a resource (本人 + 域名负责人 + lead) — used to authorise who may acknowledge it. Resolves the SINGLE resource directly (one cert/domain row + one resolver build) rather than rebuilding the whole expiry inventory — this is an on-request ACL check, not a sweep. """ from app.domains.cmdb.cert_owner_resolver import CertOwnerResolver from app.domains.cmdb.models import ( EXPIRY_RESOURCE_CERTIFICATE, EXPIRY_RESOURCE_DOMAIN, CmdbCertificate, CmdbDomain, ) from app.domains.cmdb.network_notifications import resolve_network_owner_id primary: str | None = None domain_owner: str | None = None if resource_type == EXPIRY_RESOURCE_CERTIFICATE: cert = await db.get(CmdbCertificate, resource_id) if cert is None or cert.deleted_at is not None: return set() network_owner_id = await resolve_network_owner_id(db) resolver = await CertOwnerResolver.build(db, network_owner_id=network_owner_id) res = resolver.resolve(cert) primary, domain_owner = res.owner_person_id, res.domain_owner_person_id elif resource_type == EXPIRY_RESOURCE_DOMAIN: dom = await db.get(CmdbDomain, resource_id) if dom is None or dom.deleted_at is not None: return set() primary = str(dom.owner_person_id) if dom.owner_person_id else ( await resolve_network_owner_id(db) ) else: return set() out: set[str] = set() if primary: out.add(str(primary)) if domain_owner: out.add(str(domain_owner)) if primary: owner = (await _resolve_people(db, {primary})).get(str(primary)) if owner is not None and owner.manager_person_id: out.add(str(owner.manager_person_id)) return out async def acknowledge_expiry( db: AsyncSession, *, resource_type: str, resource_id: str, actor_person_id: str, days: int = DEFAULT_ACK_DAYS, ) -> CmdbExpiryAlertState: """Mark a cert / domain "renewing" — suppress all expiry alerts (incl. escalation) for ``days`` days. Idempotent: re-acking extends the window from today. Creates the state row if the resource has not yet been observed by the engine.""" from sqlalchemy.dialects.mysql import insert as mysql_insert days = max(1, min(90, int(days))) today = datetime.now(UTC).date() now = datetime.now(UTC) until = today + timedelta(days=days) # Atomic upsert on the ``(resource_type, resource_id)`` PK. A plain # SELECT-then-INSERT raced two concurrent *first-ever* acks for the same # resource into a duplicate-key 500 (one click would 500 the other). The # UPDATE branch only touches the ack fields — ``last_bucket`` / # ``last_alerted_*`` are seeded on first insert and otherwise owned by the # alert sweep, exactly as before. stmt = mysql_insert(CmdbExpiryAlertState.__table__).values( resource_type=resource_type, resource_id=resource_id, last_bucket=BUCKET_SAFE, acknowledged_until=until, acknowledged_by_person_id=actor_person_id, acknowledged_at=now, last_alerted_at=now, ).on_duplicate_key_update( acknowledged_until=until, acknowledged_by_person_id=actor_person_id, acknowledged_at=now, ) await db.execute(stmt) # Re-read the canonical row to return. ``populate_existing`` refreshes any # instance already in the session identity map with the just-written values. row = (await db.execute( select(CmdbExpiryAlertState) .where( CmdbExpiryAlertState.resource_type == resource_type, CmdbExpiryAlertState.resource_id == resource_id, ) .execution_options(populate_existing=True) )).scalar_one() return row # ── Item builders ──────────────────────────────────────────────────── # These translate the cert / domain inventories into ``ExpiryItem``s with # their resolved responsibility chains. They live here (cmdb domain) with # only cmdb-domain dependencies; the notification sender + cron entrypoints # live in the scheduled_jobs orchestration layer (which may import the # notifications / approval domains). async def build_cert_expiry_items(db: AsyncSession) -> list[ExpiryItem]: """Build deduped certificate :class:`ExpiryItem`s with live-resolved ownership (本人 = record/domain owner, 域名负责人 = zone domain owner). One logical certificate can appear as several probed rows (one per host) plus a cloud row; they are collapsed by covered-name set so an owner is paged ONCE per logical cert. The representative is the lowest id (stable across runs); the soonest expiry in the group wins. """ from app.domains.cmdb.cert_owner_resolver import CertOwnerResolver from app.domains.cmdb.models import ( CERT_STATUS_UNREACHABLE, EXPIRY_RESOURCE_CERTIFICATE, PROD_ENVIRONMENTS, CmdbCertificate, ) from app.domains.cmdb.network_notifications import resolve_network_owner_id certs = list((await db.execute( select(CmdbCertificate).where( CmdbCertificate.deleted_at.is_(None), CmdbCertificate.not_after.is_not(None), CmdbCertificate.status != CERT_STATUS_UNREACHABLE, ) )).scalars().all()) if not certs: return [] def _identity_key(c: CmdbCertificate): names = [n for n in (list(c.sans or []) + [c.common_name]) if n] if names: return ("subj", tuple(sorted({n.lower() for n in names}))) if c.fingerprint_sha256: return ("fp", c.fingerprint_sha256) return ("id", str(c.id)) groups: dict[tuple, list[CmdbCertificate]] = {} for c in certs: groups.setdefault(_identity_key(c), []).append(c) network_owner_id = await resolve_network_owner_id(db) resolver = await CertOwnerResolver.build(db, network_owner_id=network_owner_id) items: list[ExpiryItem] = [] for members in groups.values(): # Production certs are deliberately out of scope for the reminder # lifecycle — prod TLS renewal is owned by the R&D/security team # outside Helix self-service (mirrors the prod cloud-access opt-out, # ``PROD_ENVIRONMENTS``). Skip a logical cert if ANY of its probed / # cloud rows is tagged prod, so a single prod-tagged member suppresses # the whole group rather than leaking a page via an untagged sibling. if any((m.environment or "").lower() in PROD_ENVIRONMENTS for m in members): continue members.sort(key=lambda c: str(c.id)) rep = members[0] soonest = min( (m.not_after.date() for m in members if m.not_after), default=None, ) # Resolve ownership on the member that actually carries an owner # (a probed row with a host beats a cloud row with none); fall back # to the representative. best = next( (m for m in members if resolver.resolve(m).owner_person_id), rep, ) res = resolver.resolve(best) # Freshness: discovered <24h ago. ``first_seen_at`` comes back # NAIVE on a fresh MySQL/TiDB load (asyncmy strips tzinfo) — coerce # to UTC before comparing or the aware-vs-naive subtraction raises. fseen = rep.first_seen_at if fseen is not None and fseen.tzinfo is None: fseen = fseen.replace(tzinfo=UTC) fresh = fseen is not None and fseen >= datetime.now(UTC) - timedelta(days=1) items.append(ExpiryItem( resource_type=EXPIRY_RESOURCE_CERTIFICATE, resource_id=str(rep.id), label=rep.common_name or rep.probe_host or str(rep.id), expiry_date=soonest, primary_owner_id=res.owner_person_id, domain_owner_id=res.domain_owner_person_id, detail=rep.probe_host, first_seen_fresh=fresh, )) return items async def build_domain_expiry_items(db: AsyncSession) -> list[ExpiryItem]: """Build domain (registrar) :class:`ExpiryItem`s. The domain owner IS 本人; the lead escalation is their manager. There is no separate 域名负责人 tier (the domain itself is the zone). Owner-less domains fall back to the network owner (兜底), mirroring certs. """ from app.domains.cmdb.models import ( EXPIRY_RESOURCE_DOMAIN, PROD_ENVIRONMENTS, CmdbDomain, ) from app.domains.cmdb.network_notifications import resolve_network_owner_id from app.domains.cmdb.service import _active_cloud_integration_ids # Restrict to enabled cloud tenants — a disabled integration's domain # rows are frozen at last sync and frequently appear as cross-tenant # duplicates; without this gate an owner whose domain straddles the # dupe gets double-paged. Mirrors the /domains list gate. active = await _active_cloud_integration_ids(db) if not active: return [] domains = list((await db.execute( select(CmdbDomain).where( CmdbDomain.deleted_at.is_(None), CmdbDomain.expiry_date.is_not(None), CmdbDomain.integration_id.in_(active), ) )).scalars().all()) if not domains: return [] network_owner_id = await resolve_network_owner_id(db) items: list[ExpiryItem] = [] for dom in domains: # Production domains are out of scope for expiry reminders — prod # registrar renewal is owned outside Helix self-service (mirrors the # cert gate + prod cloud-access opt-out, ``PROD_ENVIRONMENTS``). if (dom.environment or "").lower() in PROD_ENVIRONMENTS: continue owner = dom.owner_person_id or network_owner_id items.append(ExpiryItem( resource_type=EXPIRY_RESOURCE_DOMAIN, resource_id=str(dom.id), label=dom.domain_name, expiry_date=dom.expiry_date, primary_owner_id=str(owner) if owner else None, domain_owner_id=None, detail=dom.registrar, )) return items