"""SyncService: upsert engine + mark-and-sweep for CMDB resources. Cloud-tenant credentials now live in ``integrations(kind='cloud')``; this module loads an ``Integration`` row, decrypts its secrets, and feeds a concrete :class:`CloudProvider`. The resource-table plumbing below is unchanged — the column used to be ``provider_account_id`` and is now ``integration_id``, both holding the same UUID that identifies the tenant. """ from __future__ import annotations import asyncio import json as _json import logging import re import statistics import uuid from datetime import UTC, datetime, timedelta from typing import Any # Pre-launch hardening 2026-05-16: hard ceiling on how long a single # capability sync can run. Boto3 / Azure SDK / google-api-python-client # each have their own per-request timeouts but a multi-region fan-out # (AWS describe_instances across 24 enabled regions) or a per-project # walk (GCP CRM v3 + compute aggregated list across 100 projects) can # still string together a multi-hour blocking call if the cloud is # slow but not actually erroring. Without a top-level bound a single # hung capability locks up the worker pod's connection pool until the # k8s liveness probe kills it. 15 minutes is well above any # legitimate cloud-fan-out time (Aliyun multi-region IAM crawls land # in <90s in prod, Azure 80-sub fan-out lands in <180s) while still # being short enough that we recover within a single cron cadence. _CAPABILITY_SYNC_TIMEOUT_S = 15 * 60 from sqlalchemy import and_, func, or_, select, text, update from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import async_session_factory from app.core.txn import best_effort from app.domains.cmdb.credential_crypto import decrypt_fields from app.domains.cmdb.iam_upsert import normalize_iam_sub_account_id from app.domains.cmdb.models import ( CERT_SOURCE_CLOUD, CmdbCdnDomain, CmdbCertificate, CmdbDnsRecord, CmdbDomain, CmdbIamUser, CmdbLoadBalancer, CmdbObjectStorage, CmdbServer, CmdbSyncJob, ) from app.domains.cmdb.owner_history import ( CHANGE_REASON_SYNC_INITIAL_ATTRIBUTION, RESOURCE_TYPE_CDN_DOMAIN, RESOURCE_TYPE_DNS_RECORD, RESOURCE_TYPE_DOMAIN, RESOURCE_TYPE_LOAD_BALANCER, record_owner_change, ) from app.domains.cmdb.providers.base import ( ProviderCapability, ProviderCredentials, get_provider_class, ) from app.domains.cmdb.providers.exceptions import ( CapabilityNotSupported, ProviderAuthError, ProviderTransientError, redact_leg_entry, redact_provider_error, ) from app.domains.cmdb.sweep_settings import ( drop_exceeds_tolerance, get_sweep_drop_tolerance_percent, ) from app.integrations.models import Integration logger = logging.getLogger(__name__) async def _maybe_seed_initial_owner_history( db: AsyncSession, *, result: Any, # SQLAlchemy CursorResult from a mysql_insert ON DUPLICATE KEY UPDATE model_cls: Any, resource_type: str, lookup_filters: list[Any], ) -> None: """Write a ``sync_initial_attribution`` history row when sync's upsert lands a fresh INSERT for a row that already has an owner pointer. Today the cmdb sync path does NOT pull owner from provider data (the dataclasses in :mod:`app.domains.cmdb.providers.base` carry no owner field for domain / dns / cdn / lb), so this helper is inert — the row's ``owner_person_id`` will always be ``None`` on INSERT and the early-return at the second `if` fires. The wiring exists so that any future code path which lands a fresh network row with an owner already set (e.g. a provider extension resolving cloud-side resource owner → Helix person, or an admin bulk-import script) automatically writes the matching first-row attribution in :data:`cmdb_network_owner_history`. Without this, the admin-UI owner-timeline for that resource would show ``admin_patch`` as the first row instead of the canonical ``sync_initial_attribution`` that migration 175's backfill writes for legacy rows. Uses MySQL's ``ON DUPLICATE KEY UPDATE`` rowcount semantics: * ``rowcount == 1`` → fresh INSERT * ``rowcount == 2`` → an existing row matched and was updated * ``rowcount == 0`` → UPDATE was a noop (all values identical) so we only consult the row on INSERT and avoid an extra SELECT per upsert on the steady-state UPDATE path (which is the hot path for repeat syncs). """ if (result.rowcount or 0) != 1: return row = await db.scalar(select(model_cls).where(*lookup_filters)) if row is None or row.owner_person_id is None: return await record_owner_change( db, resource_type=resource_type, resource_id=str(row.id), previous_owner_person_id=None, new_owner_person_id=row.owner_person_id, actor_person_id=None, change_reason=CHANGE_REASON_SYNC_INITIAL_ATTRIBUTION, trace_id=None, ) # Leg-failure ``reason`` tokens that indicate "credential simply can't # observe this object" rather than "transient or per-leg outage". When # 100% of legs return one of these tokens AND the capability upserted # zero rows, we degrade the capability result to ``skipped`` with a # structured reason (see ``_is_structural_all_legs_auth_failure``) so # the admin UI surfaces one honest "structurally inaccessible" signal # instead of an hourly red ``partial_failure``. Adding tokens here is # a deliberate signal-vs-noise choice — keep the set narrow but cover # every "auth-shaped" prefix a sync helper emits (round-21 review caught # the original set being object_storage-only, leaving GCP DNS / IAM # legs to keep flooding red). _STRUCTURAL_AUTH_REASONS = frozenset({ "auth_error", "ProviderAuthError", "http_401", "http_403", "buckets.list_http_401", "buckets.list_http_403", }) # Provider-agnostic auth-shaped reason pattern: matches # ``(.)*_http_401`` and ``_http_403`` (e.g. # ``managedZones.list_http_403``, ``rrsets.list_http_403``, # ``cdn_domains.list_http_403``, ``iam.users.list_http_401``). # Lets every per-API leg label share the structural-skip degrade # without us having to enumerate each by hand. Restricted to 4xx # auth status codes — 5xx stays transient (partial_failure) so we # notice real cloud-side outages. _STRUCTURAL_AUTH_REASON_RE = re.compile( r"^[a-zA-Z][a-zA-Z0-9_.]*_http_(401|403)$", ) def _is_structural_all_legs_auth_failure( legs: list[dict[str, Any]], ) -> bool: """Return True when every leg failed for a credential-scope auth reason. Production case (2026-05-16): AliCloud-intl ``object_storage`` capability returned ``auth_error`` on EVERY sub_account leg because the integration's root-tenant AK cannot list buckets owned by sub_account-scoped OSS endpoints without an AssumeRole hop we don't implement. The capability fundamentally cannot succeed under the current credentials — hourly red ``partial_failure`` was noise, not signal. Contract: every distinct ``reason`` token observed across legs must EITHER be a member of ``_STRUCTURAL_AUTH_REASONS`` (curated tokens like ``auth_error``) OR match ``_STRUCTURAL_AUTH_REASON_RE`` (any ``_http_(401|403)`` shape — covers every per-API leg label that current and future drivers emit). A SINGLE transient reason mixed in (``InternalError``, ``Throttling``, 5xx http_*) keeps the result as ``partial_failure`` so transient issues remain visible — the structural-skip branch is only correct when every leg failed for a reason we know the cron tick will fail again on next run with the same shape. """ if not legs: return False reasons = { str(entry.get("reason") or "").strip() for entry in legs if isinstance(entry, dict) } if not reasons: return False return all( reason in _STRUCTURAL_AUTH_REASONS or _STRUCTURAL_AUTH_REASON_RE.match(reason) is not None for reason in reasons ) async def _send_sync_notification( job_id: str, provider_alias: str, status: str, total_upserted: int, total_deleted: int, total_errors: int, ) -> None: """Log and optionally Feishu-alert when a sync job finalizes with **real** changes or failure. Previous trigger ``total_upserted > 0`` over-counted: every UPSERT pass rewrites ``last_synced_at`` so an unchanged sync still reports N upserted rows, firing Feishu hourly per integration. Real-delta indicators we keep: - ``total_deleted > 0`` (mark-and-sweep actually soft-deleted rows — that's a true change worth flagging) - ``status != 'success'`` (any failure / partial_failure) Successful sync of an unchanged cloud is silent. Errors in this function are caught and never propagate to the caller. """ try: should_notify = total_deleted > 0 or status != "success" if not should_notify: return message = ( f"📊 云资源同步完成 - {provider_alias}\n" f"状态: {status}\n" f"新增/更新: {total_upserted} 条\n" f"删除: {total_deleted} 条\n" f"错误: {total_errors} 条" ) logger.info( "sync delta notification: job=%s provider=%s status=%s upserted=%d deleted=%d errors=%d", job_id, provider_alias, status, total_upserted, total_deleted, total_errors, ) from app.integrations._url_safety import ( UnsafeURLError, safe_async_http_transport, ) from app.integrations.compat import get_feishu_config feishu_cfg = await get_feishu_config() webhook_url = str(feishu_cfg.get("alert_webhook_url") or "") if not webhook_url: logger.info("feishu.alert_webhook_url not configured in 配置中心 — skipping Feishu sync notification") return # SSRF guard — async variant so getaddrinfo runs in a thread pool. try: transport = await safe_async_http_transport( webhook_url, allowed_suffixes=("feishu.cn", "larksuite.com"), ) except UnsafeURLError as exc: logger.warning( "feishu.alert_webhook_url rejected by SSRF guard: %s — " "skipping sync delta notification job=%s", exc.message, job_id, ) return import httpx # noqa: PLC0415 — lazy import keeps top-level deps minimal payload = {"msg_type": "text", "content": {"text": message}} # ``trust_env=False`` disables HTTP_PROXY env-var injection — a # compromised sidecar / debug override can otherwise tunnel the # webhook bearer-equivalent key through an attacker-controlled # proxy. Defense-in-depth on top of the SSRF allowlist. async with httpx.AsyncClient( timeout=15, trust_env=False, transport=transport, ) as client: resp = await client.post(webhook_url, json=payload) result = resp.json() if result.get("StatusCode") != 0 and result.get("code") not in (None, 0): logger.warning("Feishu sync notification webhook returned non-zero: %s", result) else: logger.info("sync delta notification sent via Feishu webhook: job=%s", job_id) except Exception: # noqa: BLE001 logger.exception("failed to send sync delta notification: job=%s", job_id) def _build_capability_overlap_filter(capabilities: list[str]): """Build an SQL expression matching jobs whose ``resource_types`` JSON array overlaps this run's capability list. Returns None when the list is empty or contains the ``*`` wildcard (expand-to-all) — in those cases the caller should fall back to per-integration locking (original pre-2026-04-24 behaviour) since a wildcard job overlaps anything. Uses MySQL 8.0+ / TiDB ``JSON_OVERLAPS(col, CAST(:json AS JSON))`` — both Helix production DBs support it. """ if not capabilities or "*" in capabilities: return None payload = _json.dumps(capabilities) return text( "JSON_OVERLAPS(cmdb_sync_jobs.resource_types, CAST(:caps AS JSON))" ).bindparams(caps=payload) async def run_sync( integration_id: str, job_id: str, capabilities: list[str], triggered_by_user_id: str | None, *, force_sweep: bool = False, ) -> None: """Entry point — called as FastAPI BackgroundTask or scheduled job. Opens its own AsyncSession because FastAPI closes the request-scoped session (from ``get_db``) before background tasks run in FastAPI 0.106+. ``force_sweep`` is the admin-confirmed override for ``_safe_sweep``'s anti-data-loss guards (the 5%-drop floor and the 0-rows-but-existing guard). Set ONLY by ``POST /cmdb/sync/{integration_id}/force-sweep`` after the admin has confirmed that the cloud-side row drop is the real new truth (e.g. they manually closed 11 of 20 IAM users in the console). Otherwise the next sync re-fails the same way and the failure notifier re-pages every cron tick — the "loop" the user reported 2026-05-14. """ async with async_session_factory() as db: try: await _run_sync_impl( db, integration_id, job_id, capabilities, force_sweep=force_sweep, ) except Exception as exc: # noqa: BLE001 — last-resort guard await db.rollback() await _fail_job(db, job_id, f"Unhandled sync error: {exc}") await _send_sync_notification(job_id, integration_id, "failed", 0, 0, 1) async def run_sync_with_lock( integration_id: str, job_id: str, capabilities: list[str], triggered_by_user_id: str | None, *, force_sweep: bool = False, lock_ttl: int = 1200, ) -> None: """``run_sync`` wrapped in the ``cloud_sync:`` distributed lock. Why this exists: ``POST /cmdb/sync/{id}/force-sweep`` used to hold the lock and run ``run_sync`` *synchronously* inside the request so the response could carry the refreshed job row. On a multi-project GCP fan-out that single request blocks for 60–120s — well past nginx's default 60s ``proxy_read_timeout`` — so the proxy returned 504, the frontend saw an empty error envelope, and the user saw "后端返回错误，但没有详细消息" (bug 2026-05-18). The fix is to queue ``run_sync`` as a FastAPI BackgroundTask and return the initial job row immediately; the frontend polls ``GET /cmdb/sync-jobs/{id}`` until terminal. The lock can no longer be acquired in the request handler (BG tasks run *after* the response is sent, so the ``async with`` context would have exited before the work started), so we acquire it here, inside the task. LockBusy contract: the endpoint previously surfaced this as 409 ``CLOUD_SYNC_ALREADY_RUNNING``. After the async refactor the request always returns 202, so we surface the contention by writing the job row as ``status='failed'`` with a clear retry message — the polling UI sees it and the failure_notifier's dedup keeps the alert from re-paging on each retry. ``lock_ttl`` defaults to 1200s — ``_CAPABILITY_SYNC_TIMEOUT_S`` is 15 min (900s) so the lock has to outlive a worst-case single capability run, otherwise a slow sync's lock could auto-expire mid- work and a concurrent scope / grants path acquires and races writes against the same integration's resource tables (review feedback 2026-05-18: the pre-fix default was 600s — well under the per-capability timeout). 1200s gives ~33% margin over the timeout. Any failure to acquire the lock OR run ``run_sync`` under it is materialized on the job row so the polling UI always observes a terminal status. A bare ``except Exception`` is mandatory: without it a Redis-down ``ConnectionRefusedError`` at lock acquisition would leave the BG task dead silent, the job row stranded in ``status='queued'``, and the failure_notifier blind to the breakage (review feedback 2026-05-18). """ from app.core.distributed_lock import LockBusy, distributed_lock try: async with distributed_lock( f"cloud_sync:{integration_id}", ttl=lock_ttl, ): await run_sync( integration_id, job_id, capabilities, triggered_by_user_id, force_sweep=force_sweep, ) except LockBusy: logger.info( "force-sweep job %s deferred: cloud_sync lock busy on " "integration %s", job_id, integration_id, ) await _mark_force_sweep_job_failed( job_id, error_message=( "Another cloud-sync operation is already running " "for this integration; wait for it to complete " "and retry the force-sweep." ), ) await _send_sync_notification( job_id, integration_id, "failed", 0, 0, 1, ) except Exception as exc: # noqa: BLE001 — last-resort guard # ``run_sync`` has its own ``except Exception`` that marks the # job failed via ``_fail_job``, so this branch only catches # failures that escape it — primarily ``distributed_lock`` # itself raising on Redis down (e.g. ``redis.exceptions. # ConnectionError`` carrying ``redis://:password@host:6379`` in # ``str(exc)``), or a programming error inside the wrapper # before ``run_sync`` was reached. Without this last-resort # guard the job row would remain ``status='queued'`` # indefinitely and the failure notifier would be blind. # # Visible error_message uses only the exception class name — # ``str(exc)`` from third-party libs can carry connection URLs # with embedded credentials, and the job row is read by every # ``can_read_cloud`` user (review feedback 2026-05-18). The # verbose detail goes to worker logs via ``logger.exception`` # where ops can grep but the admin UI cannot expose it. logger.exception( "force-sweep job %s wrapper failed on integration %s", job_id, integration_id, ) await _mark_force_sweep_job_failed( job_id, error_message=( "Force-sweep BG task failed before sync ran " f"({type(exc).__name__}); see worker logs for detail." ), ) await _send_sync_notification( job_id, integration_id, "failed", 0, 0, 1, ) async def _mark_force_sweep_job_failed( job_id: str, *, error_message: str, ) -> None: """Update a ``cmdb_sync_jobs`` row to ``status='failed'`` from a fresh session — used by ``run_sync_with_lock``'s LockBusy and last-resort branches. The ``WHERE status IN ('queued', 'running')`` predicate is a belt- and-braces guard: if the row already landed terminal (``success`` / ``partial_failure`` / ``failed`` / ``skipped``) by the time this fires, the UPDATE no-ops instead of clobbering a real terminal state with a misleading "failed". A fresh session is used because the request-scoped session was already closed when the BG task began (mirrors ``run_sync``'s own session discipline). Also stamps ``total_errors=1`` so the sync-history UI and downstream aggregations show this row consistently with ``_fail_job``'s failures — without this, a force-sweep that hit the LockBusy / last-resort branch would land ``status='failed'`` but ``total_errors=0`` (from ``create_sync_job``'s default) and the admin sees a "failed but 0 errors" mismatch (review feedback 2026-05-18). """ async with async_session_factory() as db: await db.execute( update(CmdbSyncJob) .where( CmdbSyncJob.id == uuid.UUID(job_id), CmdbSyncJob.status.in_(("queued", "running")), ) .values( status="failed", finished_at=datetime.now(UTC), error_message=error_message, total_errors=1, ) ) await db.commit() async def _run_sync_impl( db: AsyncSession, integration_id: str, job_id: str, capabilities: list[str], *, force_sweep: bool = False, ) -> None: # Mark job as running await db.execute( update(CmdbSyncJob) .where(CmdbSyncJob.id == uuid.UUID(job_id)) .values(status="running", started_at=datetime.now(UTC)) ) await db.commit() # Guard: skip if another sync for the SAME capabilities is already # running on this tenant. Previously the lock checked # ``integration_id`` alone — correct for the legacy "one sync per # tenant covering all capabilities" design, but wrong after the # per-capability scheduled_jobs split (migration 046): every # capability has its own cron row, they fire in the same minute, # and all-but-one would bail with status=failed. We now check # JSON overlap on ``resource_types`` so parallel runs of distinct # capabilities (e.g. ``cdn`` alongside ``dns_records``) coexist # cleanly, and only TRUE overlaps (e.g. manual full-sync while # per-cap cron is running) serialize. # # Also: "another sync already running" is a SKIP, not a FAILURE — # lands as ``status='skipped'`` so scheduled-job history doesn't # turn red on a benign collision. Resource-sync handler's "raise # on failed" convention does NOT fire on ``skipped``. conflict_expr = _build_capability_overlap_filter(capabilities) running_q = select(func.count(CmdbSyncJob.id)).where( CmdbSyncJob.integration_id == integration_id, CmdbSyncJob.status == "running", CmdbSyncJob.id != uuid.UUID(job_id), ) if conflict_expr is not None: running_q = running_q.where(conflict_expr) running = await db.scalar(running_q) if running and running > 0: await db.execute( update(CmdbSyncJob) .where(CmdbSyncJob.id == uuid.UUID(job_id)) .values( status="skipped", finished_at=datetime.now(UTC), error_message=( "Another sync with overlapping capabilities is " "already running for this integration — skipped to " "avoid data races. Safe to ignore; next cron tick " "will sync." ), ) ) await db.commit() return # Load the integration row (cloud tenant) integration = await db.get(Integration, uuid.UUID(integration_id)) if integration is None or integration.kind != "cloud": await _fail_job(db, job_id, "Cloud integration not found") await _send_sync_notification(job_id, integration_id, "failed", 0, 0, 1) return # Cache every ORM attribute we may need AFTER a rollback or commit into # plain locals. ``AsyncSession.rollback()`` expires ALL attributes in # its identity map — including on instances we loaded BEFORE the # transaction that rolled back. Re-accessing an expired column in # async code triggers SQLAlchemy's sync lazy-refresh path and raises # ``greenlet_spawn has not been called; can't call await_only()``. # That was the inner-job ``Unhandled sync error`` we saw on every # failing capability (prod 2026-04-23 to 2026-04-24, 37 occurrences). # Reading these up-front means the post-rollback paths never touch # the ORM instance again. integration_alias: str = integration.name or integration_id integration_provider: str = integration.provider integration_secrets_encrypted = integration.secrets_encrypted integration_secrets_nonce = integration.secrets_nonce integration_config_public = integration.config_public or {} if not integration_secrets_encrypted: await _fail_job(db, job_id, "Cloud integration has no credentials configured") await _send_sync_notification( job_id, integration_alias, "failed", 0, 0, 1, ) return # Decrypt credentials try: cred_fields = decrypt_fields( integration_secrets_encrypted, integration_secrets_nonce, integration_id=integration_id, ) except Exception as exc: logger.warning("Credential decryption failed for integration %s: %s", integration_id, exc) await _fail_job( db, job_id, "Credential decryption failed — check that CREDENTIAL_ENCRYPTION_KEY matches the key used when the integration was created", ) await _send_sync_notification( job_id, integration_alias, "failed", 0, 0, 1, ) return # Merge public config (regions, default_region, …) so providers that # need non-secret tuning values still get them via ``fields``. merged: dict[str, str] = {} if integration_config_public: merged.update( {k: str(v) for k, v in integration_config_public.items() if v is not None}, ) merged.update(cred_fields) creds = ProviderCredentials( provider_id=integration_id, provider_type=integration_provider, account_alias=integration_alias, fields=merged, ) provider_cls = get_provider_class(integration_provider) provider = provider_cls(creds) capability_results: dict[str, Any] = {} total_upserted = 0 total_deleted = 0 total_errors = 0 # Expand "*" wildcard to all capabilities supported by the provider if "*" in capabilities: resolved = [c.value for c in ProviderCapability] else: resolved = list(capabilities) # Wrap the per-capability loop in try/finally so that an unhandled # exception (decryption error mid-loop, ORM unexpected raise, etc.) # cannot leak the provider's httpx.AsyncClient / SDK clients. Without # this wrapper a hot-path failure leaks one TCP/TLS pool per failed # job; over thousands of cron retries that exhausts file descriptors # on the worker pod. try: for cap_str in resolved: try: cap = ProviderCapability(cap_str) except ValueError: capability_results[cap_str] = {"status": "skipped", "reason": "unknown capability"} continue if not provider.supports(cap): capability_results[cap_str] = {"status": "skipped", "reason": "not supported"} continue try: forced_meta: dict[str, Any] = {} # ``asyncio.wait_for`` enforces the # ``_CAPABILITY_SYNC_TIMEOUT_S`` ceiling — see the # constant docstring for the rationale. if cap == ProviderCapability.SERVERS: upserted, deleted = await asyncio.wait_for( _sync_servers( db, integration, job_id, provider, force_sweep=force_sweep, forced_meta=forced_meta, ), timeout=_CAPABILITY_SYNC_TIMEOUT_S, ) elif cap == ProviderCapability.DOMAINS: upserted, deleted = await asyncio.wait_for( _sync_domains( db, integration, job_id, provider, force_sweep=force_sweep, forced_meta=forced_meta, ), timeout=_CAPABILITY_SYNC_TIMEOUT_S, ) elif cap == ProviderCapability.DNS_RECORDS: upserted, deleted = await asyncio.wait_for( _sync_dns_records( db, integration, job_id, provider, force_sweep=force_sweep, forced_meta=forced_meta, ), timeout=_CAPABILITY_SYNC_TIMEOUT_S, ) elif cap == ProviderCapability.CERTIFICATES: upserted, deleted = await asyncio.wait_for( _sync_certificates( db, integration, job_id, provider, force_sweep=force_sweep, forced_meta=forced_meta, ), timeout=_CAPABILITY_SYNC_TIMEOUT_S, ) elif cap == ProviderCapability.LOAD_BALANCERS: upserted, deleted = await asyncio.wait_for( _sync_load_balancers( db, integration, job_id, provider, force_sweep=force_sweep, forced_meta=forced_meta, ), timeout=_CAPABILITY_SYNC_TIMEOUT_S, ) elif cap == ProviderCapability.CDN: upserted, deleted = await asyncio.wait_for( _sync_cdn_domains( db, integration, job_id, provider, force_sweep=force_sweep, forced_meta=forced_meta, ), timeout=_CAPABILITY_SYNC_TIMEOUT_S, ) elif cap == ProviderCapability.OBJECT_STORAGE: upserted, deleted = await asyncio.wait_for( _sync_object_storage( db, integration, job_id, provider, force_sweep=force_sweep, forced_meta=forced_meta, ), timeout=_CAPABILITY_SYNC_TIMEOUT_S, ) elif cap == ProviderCapability.IAM: upserted, deleted = await asyncio.wait_for( _sync_iam_users( db, integration, job_id, provider, force_sweep=force_sweep, forced_meta=forced_meta, ), timeout=_CAPABILITY_SYNC_TIMEOUT_S, ) else: capability_results[cap_str] = {"status": "skipped", "reason": "not implemented"} continue # Pre-launch hardening 2026-05-16: drain any partial # failures the provider's multi-region / multi-sub / # multi-project fan-out accumulated. A capability that # successfully upserts 50 servers but skipped 3 of 20 # AWS regions (or 5 of 100 GCP projects) should land # ``status='partial_failure'`` on the job row — not a # green ``success`` that silently hides 15% missing # rows. ``consume_partial_failures()`` clears the list # between capabilities so each result reflects only # that capability's legs. # Redact each entry's ``leg`` and ``reason`` fields # symmetrically with the test-connection path # (``_CloudDriverBase.test_connection``) before # persisting into ``cmdb_sync_jobs.capability_results``. # ``capability_results`` is admin-readable verbatim via # the sync-history UI, so the same credential-safe # surface that protects ``last_test_message`` applies # here. Today every current driver emits bounded # ``reason``/``leg`` strings (``http_403`` / # ``region:us-east-1``), but the wrap protects against # future drivers that build leg labels from raw SDK # error payloads — same rationale as the ``error`` # field's existing ``redact_provider_error`` wrap. leg_failures = [ redact_leg_entry(e) for e in provider.consume_partial_failures() if isinstance(e, dict) ] # Pre-launch hardening 2026-05-17: legs that the driver # has classified as "structurally expected skip" (e.g. # GCP project where the capability's underlying API was # never enabled) are tracked separately from # ``partial_failures``. They surface for admin visibility # but do NOT flip the capability status to # ``partial_failure`` and do NOT increment ``total_errors`` # (which feeds the failure-notifier). Production # incident 2026-05-17: a 500-employee org with 474 # BigQuery/Storage-only GCP projects saw the GCP servers # cron go red on every tick — the legs were technically # correct skips (no Compute API on those projects) but # the engine couldn't distinguish them from real 403s. skipped_legs: list[dict[str, Any]] = [] try: skipped_legs = [ redact_leg_entry(e) for e in provider.consume_skipped_legs() if isinstance(e, dict) ] except AttributeError: # Older provider that pre-dates the ``skipped_legs`` # contract — treat as empty list. Safe default keeps # forward-compatibility for any future external # provider that doesn't inherit from the updated # CloudProvider base. skipped_legs = [] cap_status = "partial_failure" if leg_failures else "success" # Pre-launch hardening 2026-05-17: detect the # "structural-impossibility" pattern — every leg failed # with the same auth reason AND zero rows were upserted. # This is what AliCloud-intl object_storage looked like # in prod (every sub_account leg returned ``auth_error`` # because the integration AK is root-tenant scoped and # the OSS REST endpoint needs per-sub-account credentials # OR cross-account AssumeRole that isn't implemented). # The capability fundamentally cannot succeed under the # current integration shape — flooding the admin UI # with hourly ``partial_failure`` red is misleading. # Degrade to ``skipped`` with a structured reason so the # admin sees one signal, not 24 daily; the cron-level # failure-notifier likewise stays quiet rather than # paging every tick. Honest "all legs failed structurally" # signal, not silencing — the reason payload carries # the leg breakdown so the admin can act. if ( leg_failures and upserted == 0 and _is_structural_all_legs_auth_failure(leg_failures) ): cap_status = "skipped" result_entry: dict[str, Any] = { "status": "skipped", "reason": ( "all-legs auth_error — capability " "structurally inaccessible under current " "integration credentials (typical: cross-" "account access required but not configured)" ), "partial_failures": leg_failures, "leg_count": len(leg_failures), "upserted": 0, "deleted": 0, } # Structured WARNING so the log-aggregation surface # has one signal-grade event per cron tick (we don't # emit a Feishu alert — by design — to avoid flooding # the admin chat with the same "structurally # impossible" reminder every hour). The persisted # ``capability_results.status='skipped'`` is the # primary signal; this log line is the secondary # forensic breadcrumb for ops correlation. logger.warning( "cmdb.sync capability degraded to skipped " "(structural all-legs auth_error): job=%s " "integration=%s capability=%s legs=%d", job_id, integration_id, cap_str, len(leg_failures), ) else: result_entry = { "status": cap_status, "upserted": upserted, "deleted": deleted, } if leg_failures: result_entry["partial_failures"] = leg_failures # Count the capability as having errors so the # job-level rollup ends in ``partial_failure``, # not ``success`` — admins reading # ``cmdb_sync_jobs.status`` need a single signal # to filter on. total_errors += 1 if skipped_legs: # Surface "structurally expected skip" legs # alongside upsert/delete counts so admins can # see WHICH leg legitimately has no resources # (e.g. which GCP project doesn't have Compute # Engine enabled) without the capability going # red. Counted separately from upsert/deleted — # the leg returned zero rows by design, no # mutation happened. Does NOT increment # ``total_errors`` (deliberately, see # ``skipped_legs`` rationale above). result_entry["skipped_legs"] = skipped_legs result_entry["skipped_leg_count"] = len(skipped_legs) # Surface the bypass on the job row's capability_results so # the admin (and the audit ingest) can see WHICH guard was # bypassed and the pre-sweep existing count — soft-deletes # are reversible (set ``deleted_at=NULL``) but only if you # know which rows to look at. if forced_meta: result_entry["forced_sweep"] = True result_entry.update(forced_meta) capability_results[cap_str] = result_entry total_upserted += upserted total_deleted += deleted except TimeoutError: # Pre-launch hardening 2026-05-16: bounded by # ``_CAPABILITY_SYNC_TIMEOUT_S``. The capability is marked # failed (not skipped) so the failure-notifier alerts and # the next sweep retries; we record the timeout value so # the admin can correlate against the sync-history row # and tell apart "cloud was slow" vs "cloud hung # entirely". await db.rollback() await db.refresh(integration) capability_results[cap_str] = { "status": "failed", "error": ( f"capability sync exceeded " f"{_CAPABILITY_SYNC_TIMEOUT_S}s and was aborted " "(cloud API hung or fan-out too large) — " "investigate the provider's status page and " "retry on the next cron tick." ), "timed_out": True, "timeout_s": _CAPABILITY_SYNC_TIMEOUT_S, } total_errors += 1 except ProviderAuthError as exc: # ``rollback()`` expires every ORM instance in the session; # ``refresh()`` un-expires so the next capability in a # wildcard run (and the post-loop notification path) can # read cached columns without triggering a sync lazy-load. await db.rollback() await db.refresh(integration) # Pre-launch hardening 2026-05-16: ``capability_results`` # is persisted to ``cmdb_sync_jobs`` and surfaced verbatim # on the admin sync-history page — boto3 / GCP API client # error strings can carry access-key prefixes, signed-URL # tokens, or stringified request bodies. Run every error # payload through ``redact_provider_error`` before it # lands in the row. capability_results[cap_str] = { "status": "failed", "error": f"auth error: {redact_provider_error(exc)}", } total_errors += 1 except CapabilityNotSupported as exc: # A driver's capability can be runtime-gated (e.g. UCloud # ``dns_records`` needs ``dns_zones`` config) — surface as # ``skipped`` not ``failed`` so scheduled-job history doesn't # turn red for a legitimate config-gated state. Crucially # the exception fires BEFORE the mark-and-sweep in each # ``_sync_*`` helper, so historical rows survive untouched. await db.rollback() await db.refresh(integration) capability_results[cap_str] = { "status": "skipped", "reason": redact_provider_error(exc), } except Exception as exc: await db.rollback() await db.refresh(integration) # Drain any partial_failures the driver populated BEFORE # the exception fired. Pre-fix the error-path threw away # this signal — symptomatic in PR #340's GCP zone-warning # rollout: the driver correctly recorded each problem # zone as a leg in ``provider.partial_failures``, but # ``_safe_sweep`` raised ``ProviderTransientError`` on # the >5%-drop guard BEFORE the success-path leg-drain # at line 522 ran, so ``capability_results.partial_failures`` # stayed NULL and the admin sync-history surface still # showed "refusing to sweep" with no clue which zone # was bad — exactly the visibility goal the GCP fix was # meant to achieve. Symmetric leg-drain on the error # branch closes the gap; same redaction wrap as the # success path. Capabilities that never populated # partial_failures (the historically common shape) # produce an empty list and the entry shape is # unchanged from pre-fix. err_leg_failures: list[dict[str, Any]] = [] try: err_leg_failures = [ redact_leg_entry(e) for e in provider.consume_partial_failures() if isinstance(e, dict) ] except Exception: # noqa: BLE001 # Driver internals broken — never let leg collection # mask the original exception we're handling. logger.exception( "consume_partial_failures() raised in error " "branch of capability %s (job=%s)", cap_str, job_id, ) # Pre-launch hardening 2026-05-17: symmetric drain of the # skipped-legs queue. If the capability raised mid-stream # (e.g. safe_sweep refused) AFTER the driver had already # classified some legs as "API not enabled" skips, the # success-path drain at line ~545 never ran and the # information was lost. Surface them on the error entry # too so an admin investigating the failure can still see # "by the way, 12 projects had Compute API disabled" as # forensic context. err_skipped_legs: list[dict[str, Any]] = [] try: err_skipped_legs = [ redact_leg_entry(e) for e in provider.consume_skipped_legs() if isinstance(e, dict) ] except AttributeError: err_skipped_legs = [] except Exception: # noqa: BLE001 logger.exception( "consume_skipped_legs() raised in error " "branch of capability %s (job=%s)", cap_str, job_id, ) err_entry: dict[str, Any] = { "status": "failed", "error": redact_provider_error(exc), } if err_leg_failures: err_entry["partial_failures"] = err_leg_failures err_entry["leg_count"] = len(err_leg_failures) if err_skipped_legs: err_entry["skipped_legs"] = err_skipped_legs err_entry["skipped_leg_count"] = len(err_skipped_legs) capability_results[cap_str] = err_entry total_errors += 1 finally: try: await provider.close() except Exception: # noqa: BLE001 logger.exception("provider.close() failed for job=%s", job_id) final_status = ( "success" if total_errors == 0 else ("partial_failure" if total_upserted + total_deleted > 0 else "failed") ) await db.execute( update(CmdbSyncJob) .where(CmdbSyncJob.id == uuid.UUID(job_id)) .values( status=final_status, finished_at=datetime.now(UTC), capability_results=capability_results, total_upserted=total_upserted, total_deleted=total_deleted, total_errors=total_errors, ) ) await db.commit() # Notify on any delta or failure — errors must not propagate. # NOTE: ``integration_alias`` is the cached local from line 219; DO # NOT access ``integration.name`` here — any capability that # rolled back above has already expired this ORM instance, and a # post-rollback column re-read would raise greenlet_spawn in async # code. await _send_sync_notification( job_id, integration_alias, final_status, total_upserted, total_deleted, total_errors, ) # Anomaly threshold for the sweep guard: if the provider returned far # fewer rows than we already had, the most likely explanation is a # partial fetch failure (driver swallowed an exception, returned ``[]``) # rather than a real bulk-deletion. Refuse to sweep in that case so we # do not soft-delete real production rows. # # History: 0.5 pre 2026-05-05 → 0.95 (5% hardcoded drop tolerance) → # settings-derived 2026-05-19 first pass (.env) → platform_settings- # backed 2026-05-19 second pass. Tolerance resolves via # :func:`app.domains.cmdb.sweep_settings.get_sweep_drop_tolerance_percent` # which reads ``platform_settings['cmdb_sync.sweep_drop_tolerance_percent']`` # (admin-editable in Platform Settings UI; takes effect on the next # sweep call, no restart) and falls back to # ``settings.CMDB_SYNC_SWEEP_DROP_TOLERANCE_PERCENT`` when missing / # malformed. The zero-rows guard below is unconditional — only the # drop-ratio leg is tunable. Tolerance semantics (% of drop tolerated # before refusing to sweep): 0 = STRICTEST (any drop trips), 20 = # default, 100 = DISABLED (floor collapses to 0 → never trips). # How many recent CmdbSyncJob rows to scan when looking for the # previous successful upserted-count baseline for a (integration, # capability) pair. 20 covers a multi-day streak of failures even on # an hourly cron — well past the point where we'd give up and use the # Helix-existing count fallback. Larger windows make the baseline # query slower without buying real recovery (a 7-day-old baseline is # already too stale to be authoritative). _BASELINE_SCAN_WINDOW = 20 # How many recent successful per-capability upsert counts to take the # median of. See ``_baseline_cap_upserted`` — the single-previous-run # baseline it replaces could not tell a daily-autoscaling fleet apart # from a partial driver failure. _BASELINE_SAMPLE_TARGET = 5 async def _baseline_cap_upserted( db: AsyncSession, *, integration_id: str, capability_key: str, ) -> int | None: """Return the MEDIAN successful upserted count for this ``(integration, capability)`` pair, or None if no prior success. **Median of recent runs, not the single previous run** (2026-08-09). The previous shape compared against whatever the last successful sync happened to report, which is only meaningful for a fleet that holds still between ticks. Prod counter-example — the GCP servers cron, ``failure_count`` 47, tripping intermittently with ``provider persisted 834 unique rows vs 1140 baseline (previous_sync) (>20% drop) — refusing to sweep``. That tenant's GKE fleet autoscales on China business hours, and 14 samples per cron tick show the cycle cleanly: 09:55 UTC (17:55 Beijing) median 986 peak 1140 21:55 UTC (05:55 Beijing) median 803 trough 740 740…1140 is an amplitude of **35% of peak** against a **20%** drop tolerance, so any trough run whose baseline landed on a peak run read as a partial driver failure. Nothing was wrong with the driver, the leg set was stable at 20 skipped projects for the whole window, and the sweep was refused on a completely healthy snapshot. Taking the median over the recent successful runs makes the baseline describe "what a healthy run looks like for this pair" instead of "what the last one happened to catch". It is robust in both directions: a one-off inflated run (duplicate provider pages) no longer poisons the next tick's baseline, and a one-off trough no longer lowers the bar for the tick after it. A genuine partial failure — the thing the guard exists for — still trips, because losing a third of the fleet is far outside the band a median smooths over. Samples are collected newest-first and capped at ``_BASELINE_SAMPLE_TARGET``; with a single prior success the median IS that value, so a freshly-onboarded integration behaves exactly as before. Callers still fall back to the Helix-existing comparison when this returns None, and still cap the result by current active cardinality (see ``_safe_sweep``). This is the sync-to-sync baseline the floor compares against, introduced 2026-05-14 (third pass after PR #299 force-sweep + PR #300 active-filter): the older "compare upserted to Helix's existing count" math was systemically broken for any tenant that accumulated stale rows from a previous floor-trip. Each failed sync left rows un-swept; the next sync's existing count stayed inflated; the floor read the same drop again and stayed locked forever. Comparing to what the CLOUD actually returned is self-correcting — the baseline reflects the fleet, not what Helix accumulated. The 2026-08-09 change keeps that property and only swaps which cloud-returned number is authoritative. Lookup walks the most recent ``_BASELINE_SCAN_WINDOW`` jobs (job status in ``success`` / ``partial_failure``) and collects per-cap entries whose ``status == 'success'`` carry an integer ``upserted``, newest first, up to ``_BASELINE_SAMPLE_TARGET``. Returns None if no such entry is found in the window — callers fall back to the Helix-existing comparison (still useful for the first-ever sync of a new integration). The (integration_id, status, finished_at) shape is already indexed via ``ix_cmdb_sync_jobs_integration_status_finished``; the scan is bounded to 20 rows so query cost is constant. """ # ``status IN (success, partial_failure)`` already implies # ``finished_at IS NOT NULL`` (those statuses are only set at job # completion); no need for a NULLS-LAST clause, which MySQL/TiDB # don't accept anyway (``ORDER BY ... NULLS LAST`` is a PostgreSQL # extension). rows = (await db.execute( select(CmdbSyncJob.capability_results) .where(CmdbSyncJob.integration_id == integration_id) .where(CmdbSyncJob.status.in_(["success", "partial_failure"])) .order_by(CmdbSyncJob.finished_at.desc()) .limit(_BASELINE_SCAN_WINDOW) )).scalars().all() samples: list[int] = [] for cap_results in rows: if not isinstance(cap_results, dict): continue entry = cap_results.get(capability_key) if not isinstance(entry, dict): continue if entry.get("status") != "success": continue upserted = entry.get("upserted") if isinstance(upserted, int) and upserted >= 0: samples.append(upserted) if len(samples) >= _BASELINE_SAMPLE_TARGET: break if not samples: return None # ``median_low`` rather than ``median``: an even-sized sample must # resolve to a real observed run count, not the average of two. The # baseline feeds an integer floor that authorizes soft-deletes, and # "a number the cloud actually returned once" is the defensible # thing to point at in a post-incident review. Erring low also keeps # the guard marginally more permissive, i.e. biased away from # refusing to sweep a healthy snapshot. return statistics.median_low(samples) async def _audit_resurrections( db: AsyncSession, *, model: Any, integration_id: str, key_column: Any, incoming_keys: list[Any], job_id: str, capability_key: str, audit_resource_type: str, secondary_column: Any | None = None, secondary_values: list[Any] | None = None, ) -> int: """Detect rows the upcoming ``on_duplicate_key_update`` will un-soft-delete and emit one ``cmdb.sync.resurrected`` audit row. Round-27 pre-launch (2026-05-25). Per ``feedback_no_speculative_backfill`` + ``feedback_on_duplicate_key_preserve_consistency``: silently clearing ``deleted_at`` on a row that was explicitly soft-deleted (operator decommissioned it; previous safe-sweep stamped it) loses provenance — the next maintainer has no way to know an instance ID was reused or a cloud-side resurrection happened. The pre-flight SELECT is cheap (one round-trip per capability per sync, the ``(integration_id, key)`` lookup is primary-key indexed) and the audit row gives operators a forensic trail. Composite-key support: pass ``secondary_column`` + ``secondary_values`` (same length as ``incoming_keys``) when the resource's natural key is a tuple, e.g. DNS records whose unique key is ``(integration_id, zone_name, record_id)`` — same ``record_id`` can legitimately exist in multiple zones, so matching by ``record_id`` alone would produce false-positive resurrection audits. Round-27 regression fix (2026-05-25): introduced after the DNS pass triggered cross-zone false positives during the post-fix review. Returns the count of rows about to be resurrected. Caller does NOT need to act on the count — the audit row is the artefact. """ if not incoming_keys: return 0 use_composite = secondary_column is not None if use_composite: if secondary_values is None or len(secondary_values) != len(incoming_keys): raise ValueError( "_audit_resurrections: secondary_values must be a list of " "the same length as incoming_keys when secondary_column " "is given.", ) # Composite key match: row must match BOTH (key_column == k1) # AND (secondary_column == k2) for the SAME source row. Cannot # use ``IN`` on each independently because that would match the # full cross-product, which is exactly the false-positive bug # we're fixing. pair_predicate = or_(*[ and_(key_column == k1, secondary_column == k2) for k1, k2 in zip( incoming_keys, secondary_values, strict=True, ) ]) select_cols = [ model.id, key_column, secondary_column, model.deleted_at, ] where_clause = and_( model.integration_id == str(integration_id), pair_predicate, model.deleted_at.is_not(None), ) else: select_cols = [model.id, key_column, model.deleted_at] where_clause = and_( model.integration_id == str(integration_id), key_column.in_(incoming_keys), model.deleted_at.is_not(None), ) rows_q = await db.execute( select(*select_cols).where(where_clause).limit(500), ) rows = rows_q.all() if not rows: return 0 if use_composite: sample = [ { "id": str(r[0]), "key": str(r[1]), "secondary": str(r[2]), "was_deleted_at": r[3].isoformat() if r[3] else None, } for r in rows ] else: sample = [ { "id": str(r[0]), "key": str(r[1]), "was_deleted_at": r[2].isoformat() if r[2] else None, } for r in rows ] from app.audit.service import AuditService # noqa: PLC0415 audit = AuditService(db) await audit.log( trace_id=str(job_id), actor_type="system", actor_user_id=None, action="cmdb.sync.resurrected", status="success", resource_type=audit_resource_type, resource_id=None, detail={ "integration_id": str(integration_id), "sync_job_id": str(job_id), "capability": capability_key, "resurrected_count": len(rows), "resurrected_sample": sample, }, ) return len(rows) async def _safe_sweep( db: AsyncSession, *, model, integration_id: str, job_id: str, upserted: int, capability_label: str, capability_key: str | None = None, force_sweep: bool = False, forced_meta: dict[str, Any] | None = None, extra_active_filter: list | None = None, ) -> int: """Mark-and-sweep with anti-data-loss guards. Refuses to soft-delete in two anomaly cases. ``observed`` below means unique active rows actually stamped with this job id; the raw driver-loop ``upserted`` counter is deliberately not trusted because duplicate pages can update the same row repeatedly: 1. ``observed == 0`` and there's a non-zero baseline (previous successful sync's upserted, or Helix-existing count when no prior success) → driver almost certainly partial-failed. 2. ``observed < baseline * (1 - tolerance/100)`` → looks like a partial fetch even though some rows came back. Tolerance resolves via ``platform_settings['cmdb_sync.sweep_drop_tolerance_percent']`` (DB-first; admin-editable, takes effect on next sweep) with ``settings.CMDB_SYNC_SWEEP_DROP_TOLERANCE_PERCENT`` fallback — default 20% drop tolerance → 0.80 keep-ratio. Tolerance is resolved exactly ONCE per ``_safe_sweep`` call so the floor and the error-message percentage cannot disagree if an admin edits the setting mid-sweep. In both cases, raise ``ProviderTransientError`` so the calling capability marks ``failed`` and the next sync re-tries — preferable to silently soft-deleting half the fleet. **Baseline: sync-to-sync, NOT cumulative Helix state.** When ``capability_key`` is provided AND prior successful syncs for this ``(integration, capability)`` pair are on record, the floor compares the current unique observed count to the MEDIAN of those runs' reported upsert counts (``_baseline_cap_upserted``; it was the single previous run until 2026-08-09 — see that function for the autoscaling-fleet false-positive that motivated the change). Why: the older "compare to Helix's existing count" math read EC2 / domains / DNS-records the same way it read IAM — but for every capability, once the floor tripped once, each failed sync left more stale rows in DB, the next sync saw an even bigger inflated existing count, and the floor stayed locked forever. A 5% accumulation of cloud-side deletions across a hot capability (AWS pruning terminated instances after ~1h, DNS records deleted in a batch, domains transferred out) was enough to permanently deadlock that capability's cron. Sync-to-sync is self-correcting: a one-off driver hiccup (current << previous) still trips; a steady-state cloud that lost 1 of 100 between syncs reads as 99% of previous = no trip = sweep proceeds = baseline updates to 99 for next time. The trade-off is the very first sync after a new integration / never-succeeded streak falls back to existing-count comparison; on a tenant that's already in a stuck state, one force-sweep override is needed to seed the baseline. ``extra_active_filter`` (caller-supplied list of SQLAlchemy expressions) narrows BOTH the ``existing`` count AND the sweep's UPDATE so the guard compares cloud-active vs Helix-active, not cloud-active vs Helix-anything-not-deleted. Used by capabilities that retain rows in non-active lifecycle states (e.g. ``CmdbIamUser.revoked_at IS NOT NULL``). Why this matters for IAM specifically (which is the only model with this lifecycle today): ``provider.list_iam_users()`` returns ACTIVE cloud-side IAM users. Helix-revoked users (Helix called ``DeleteUser`` and stamped ``revoked_at``) are gone from the cloud's list AND still present in ``cmdb_iam_users`` (kept as history for the 已被撤销 UI tab + audit trail). The sweep WHERE must skip them so a stale ``last_seen_sync_job_id`` on a revoked row doesn't soft-delete the history record. ``force_sweep`` (admin override, default False): when True, BOTH guards are bypassed and the sweep proceeds with whatever the provider returned. The route that flips this — ``POST /cmdb/sync/{integration_id}/force-sweep`` — is admin-only and writes its own ``cmdb.sync.force_sweep_override`` audit row. ``forced_meta``, when provided, records which guard would have tripped, the baseline source (``previous_sync`` vs ``existing``), and the pre-sweep counts so an admin can find the affected rows. ``forced_meta`` key set is **asymmetric by branch** — downstream readers (UI, MCP, audit consumers) MUST NOT assume a fixed shape: * Zero-rows guard bypass: ``forced_sweep_guard='zero_rows'`` plus ``forced_sweep_baseline``, ``forced_sweep_baseline_source``, ``forced_sweep_existing``. No observed/upserted (both are zero) and no tolerance (irrelevant for this guard). * Drop-ratio guard bypass: ``forced_sweep_guard='drop_ratio'`` plus the above 3 keys, ``forced_sweep_upserted``, ``forced_sweep_observed``, AND ``forced_sweep_tolerance_percent`` — the tolerance that was ACTIVE at trip time, so a future audit can reconstruct what threshold the admin chose to bypass even if the platform_setting changes later. Returns the number of rows soft-deleted. """ base_active_where = [ model.integration_id == integration_id, model.deleted_at.is_(None), ] if extra_active_filter: base_active_where.extend(extra_active_filter) # ``existing`` counts only rows the driver has EVER successfully # observed (``last_seen_sync_job_id IS NOT NULL``). That matches # the sweep scope below (which the 2026-05-28 fix narrowed to # ``IS NOT NULL`` so migration / manual rows are NEVER auto-swept). # Without aligning the two, the drop-ratio guard on a tenant with # a freshly-imported batch of ``IS NULL`` rows would trip on every # sync (those rows inflate ``existing`` but are never reachable by # the upsert path → ratio is always low). existing_q = select(func.count(model.id)).where( *base_active_where, model.last_seen_sync_job_id.is_not(None), ) existing = int((await db.execute(existing_q)).scalar() or 0) # ``upserted`` is a driver-loop throughput counter, not a cardinality: # duplicate pages and tenant-wide fan-out can UPDATE the same unique row # many times. A destructive guard must compare the number of unique, # active rows actually stamped by THIS job, otherwise duplicates inflate # the numerator and let missing rows pass as a healthy snapshot. observed_q = select(func.count(model.id)).where( *base_active_where, model.last_seen_sync_job_id == job_id, ) observed = int((await db.execute(observed_q)).scalar() or 0) # Prefer sync-to-sync baseline. Falls back to existing-count when # no prior successful sync of this capability exists for this # integration (first-ever sync, integration just enabled, or # streak of failures longer than _BASELINE_SCAN_WINDOW). prev_upserted: int | None = None if capability_key: prev_upserted = await _baseline_cap_upserted( db, integration_id=integration_id, capability_key=capability_key, ) if prev_upserted is None: baseline_source = "existing" baseline = existing elif prev_upserted > existing: # Historical capability results only stored the raw driver-loop # upsert counter. A duplicate page can therefore claim 100 upserts # for five unique persisted rows and poison every future baseline. # The active DB cardinality is a safe upper bound; capping here keeps # sync-to-sync's anti-stale-accumulation behavior when the historical # count is lower, while recovering from raw-counter inflation. baseline_source = "previous_sync_capped_by_existing" baseline = existing else: # Label kept as ``previous_sync`` (2026-08-09): the value is now # the median of recent successful runs rather than the single # previous one, but the label is a STABLE IDENTIFIER for "the # cloud-returned baseline" as opposed to ``existing``. It is # persisted in ``forced_meta`` on historical audit rows and # asserted by consumers; renaming it would make old and new # audit rows describe the same branch differently for no gain. baseline_source = "previous_sync" baseline = prev_upserted if observed == 0 and baseline > 0: if not force_sweep: raise ProviderTransientError( f"{capability_label}: provider returned 0 rows but " f"baseline is {baseline} ({baseline_source}) for this " "integration — refusing to sweep (likely partial driver " "failure). Investigate the provider error log before retrying." ) if forced_meta is not None: forced_meta["forced_sweep_guard"] = "zero_rows" forced_meta["forced_sweep_baseline"] = baseline forced_meta["forced_sweep_baseline_source"] = baseline_source forced_meta["forced_sweep_existing"] = existing logger.warning( "force_sweep override: %s zero-rows guard bypassed " "(integration=%s, baseline=%d source=%s existing=%d upserted=0)", capability_label, integration_id, baseline, baseline_source, existing, ) elif baseline > 0 and observed < baseline: # ``upserted < baseline`` is a cheap pre-check — keep_ratio is # always in ``[0, 1]`` so the floor cannot exceed baseline; a # sync where upserted >= baseline can never trip the drop-ratio # guard. Short-circuiting here skips a DB roundtrip on every # healthy / growing-fleet sync (the steady-state case for most # integrations) without changing the trip semantics. # # Read tolerance ONCE and derive both the floor and the # error-message percentage from the same value — closes the # TOCTOU window where an admin editing the setting between two # resolver calls could land a message that disagrees with the # floor that actually tripped. tolerance_pct = await get_sweep_drop_tolerance_percent(db) if drop_exceeds_tolerance( baseline=baseline, observed=observed, tolerance_percent=tolerance_pct, ): if not force_sweep: raise ProviderTransientError( f"{capability_label}: provider persisted {observed} unique " f"rows ({upserted} raw upserts) vs " f"{baseline} baseline ({baseline_source}) " f"(>{tolerance_pct}% drop) " "— refusing to sweep (likely partial driver failure)." ) if forced_meta is not None: forced_meta["forced_sweep_guard"] = "drop_ratio" forced_meta["forced_sweep_baseline"] = baseline forced_meta["forced_sweep_baseline_source"] = baseline_source forced_meta["forced_sweep_existing"] = existing forced_meta["forced_sweep_upserted"] = upserted forced_meta["forced_sweep_observed"] = observed forced_meta["forced_sweep_tolerance_percent"] = tolerance_pct logger.warning( "force_sweep override: %s drop-ratio guard bypassed " "(integration=%s, baseline=%d source=%s existing=%d " "observed=%d raw_upserted=%d tolerance=%d%%)", capability_label, integration_id, baseline, baseline_source, existing, observed, upserted, tolerance_pct, ) # Sweep narrows by the same ``extra_active_filter`` so we never # soft-delete a row we excluded from the "existing" count. For IAM # this preserves the revoked-history rows indefinitely (the # 已被撤销 tab in the UI keeps showing them); only an admin's # explicit purge / data-cleanup migration removes them. # # ``last_seen_sync_job_id IS NOT NULL`` (added 2026-05-28 after the # legacy-CMDB migration → UCloud sync collision incident): # # * BEFORE: the predicate was # ``last_seen_sync_job_id IS NULL OR last_seen_sync_job_id != job_id``. # Rows whose ``last_seen_sync_job_id`` is NULL — i.e. NEVER touched # by any sync (rows inserted by the legacy migration script, or # manual API inserts that didn't appear in the cloud response) — # would match the ``IS NULL`` branch and get swept. That's exactly # the bug: the legacy-CMDB migration writes rows with # ``last_seen_sync_job_id=NULL`` and instance_ids the live driver # can't generate, so the first sync after migration silently # soft-deleted all 116 UCloud servers (10.9.84.114 / DLP-test # incident). # # * NOW: only rows that WERE once observed by ``some`` prior sync # of this driver (``last_seen_sync_job_id IS NOT NULL``) AND are # no longer observed by the current job are candidates for # soft-delete. Rows the live driver has never seen are # structurally outside its scope — they're either bare-metal # parked under a disabled integration, future-migration imports # that haven't been driver-observed yet, or admin-curated # manual entries. None of those should be auto-swept; the # ``_safe_sweep`` baseline tolerance was never the right # mechanism to protect them (the legacy import wiped its way # under the floor before anyone could intervene). # # Downstream effect: the very first sync of a freshly-imported # integration no longer sweeps the imported rows; instead they're # picked up one-by-one as the live driver upserts each into the # ``(integration_id, instance_id)`` slot and stamps a # ``last_seen_sync_job_id``. Rows the driver never sees stay alive. # If they should NOT stay alive (genuinely decommissioned), the # admin removes them via ``POST /cmdb/{resource}/{id}/delete`` or # the future recycle-bin UI — explicit > implicit. sweep_where = list(base_active_where) + [ model.last_seen_sync_job_id.is_not(None), model.last_seen_sync_job_id != job_id, ] # Round-28 (2026-05-18): protect rows freshly inserted by a parallel # request from being swept. When an admin or employee creates a DNS # record / domain etc. via HTTP between the sync's provider-fetch and # this sweep, the new row has ``last_seen_sync_job_id=NULL`` (it was # never touched by this sync). Without the time guard the row matches # the OR clause above and gets soft-deleted immediately. Excluding # rows whose ``first_seen_at`` is newer than the sync's ``started_at`` # closes the race for every model that carries that column (every # network resource type does — see ``first_seen_at`` declarations in # models.py). Models without the column skip this guard cleanly. if hasattr(model, "first_seen_at"): sync_job = await db.get(CmdbSyncJob, uuid.UUID(str(job_id))) if sync_job is not None and sync_job.started_at is not None: sweep_where.append(model.first_seen_at < sync_job.started_at) # Capture the IDs about to be soft-deleted BEFORE running the # UPDATE so the audit row can name them. Without this, a routine # sync sweep that legitimately wipes a few rows leaves zero # forensic trail — admin asking "what swept this DNS record?" # has only ``last_seen_sync_job_id`` to crawl back through. # The audit row caps at 500 ids to keep the JSON sane; a sweep # that exceeds 500 is unusual enough that the per-tenant count is # the meaningful forensic signal anyway. id_select = select(model.id).where(*sweep_where).limit(500) swept_ids_result = await db.execute(id_select) swept_ids = [str(r) for (r,) in swept_ids_result.all()] sweep = update(model).where(*sweep_where).values(deleted_at=datetime.now(UTC)) result = await db.execute(sweep) swept_count = result.rowcount or 0 if swept_count > 0: # Only audit network-asset sweeps — sync of IAM users / cost # facts / billing already has its own per-row audit-equivalent # state machine and a generic sweep audit there would be noise. # The four "owned" resource types match the user-explicit # surface (domain / DNS / CDN / LB). audit_resource_type = { "CmdbDomain": "cmdb_domain", "CmdbDnsRecord": "cmdb_dns_record", "CmdbCdnDomain": "cmdb_cdn_domain", "CmdbLoadBalancer": "cmdb_load_balancer", }.get(model.__name__) if audit_resource_type is not None: from app.audit.service import AuditService # noqa: PLC0415 audit = AuditService(db) await audit.log( trace_id=str(job_id), actor_type="system", actor_user_id=None, action="cmdb.sync.sweep_soft_deleted", status="success", resource_type=audit_resource_type, resource_id=None, detail={ "integration_id": str(integration_id), "sync_job_id": str(job_id), "capability": capability_key, "swept_count": swept_count, "swept_ids_sample": swept_ids, "force_sweep": force_sweep, "baseline": baseline, "baseline_source": baseline_source, "upserted": upserted, }, ) return swept_count # ── Write-transaction bounding (TiDB 8176 incident, 2026-08-08) ── # Rows upserted per write transaction inside a capability loop. # # Production incident 2026-08-08: ``cmdb.resource_sync`` on the GCP # integration died with # ``(8176, 'Your query has been cancelled due to exceeding the allowed # memory limit for the tidb-server instance and this query is currently # using the most memory')``. # # The killed statement was never the expensive one. Every ``_sync_*`` # helper below ran its ENTIRE capability — 784 GCP instances — as one # transaction and only committed after the sweep. TiDB buffers a # transaction's whole write set in tidb-server memory until commit, and # charges that membuffer to the session, so the accumulated bytes are # attributed to whatever statement happens to be running. The evidence # from ``information_schema.statements_summary_history`` for the failing # run (job 21547529, integration c1a4f301 / GCP): # # Insert cmdb_servers exec=663 max_mem=64.5MB <- membuffer growing # Insert cmdb_servers exec=93 max_mem=129.0MB # Select count(id) … max_mem=129.0MB <- a COUNT(*)! # Select id … limit 500 max_mem=129.0MB # Update cmdb_servers (sweep) max_mem=157.1MB <- killed here # # The same three guard queries re-run afterwards in a fresh session with # an empty transaction report **0.0MB** — proof the memory was the # membuffer, not the query plans. 8176 is an *instance*-level guard, so # whichever session tops the list gets killed when the shared serverless # tidb-server comes under pressure: the sync was simply the fattest # session on the instance. # # Committing every N rows caps the membuffer at N × row-size regardless # of fleet size. 200 keeps the peak in the low single-digit MB for every # capability while staying coarse enough that per-sync commit overhead # is negligible (4 commits for a 784-instance GCP fleet, 27 for AWS's # 5.2k). # # Failure semantics change, deliberately and safely: an exception # mid-loop used to roll back the whole capability, now it rolls back # only the current batch. No data-loss path opens up — the mark-and- # sweep runs strictly AFTER the loop, so a partial run soft-deletes # nothing; the already-committed rows simply carry this job's # ``last_seen_sync_job_id`` and get re-stamped by the next successful # tick. Preserving partial progress is strictly better than discarding # it. IAM is deliberately NOT batched — see ``_sync_iam_users``. _UPSERT_COMMIT_BATCH = 200 async def _commit_upsert_batch(db: AsyncSession, upserted: int) -> None: """Commit every ``_UPSERT_COMMIT_BATCH`` rows of a capability loop. Call at the END of a loop iteration, after every write belonging to that row (upsert + owner-history seed) has been issued, so a batch boundary can never split one resource's writes across two transactions. """ if upserted and upserted % _UPSERT_COMMIT_BATCH == 0: await db.commit() # ── Oversized-payload guard for ``raw_data`` snapshots ─────────── # Longest single string kept verbatim inside a server's ``raw_data``. # # Same 2026-08-08 incident, other half. GCP instances averaged **95 KB** # of ``raw_data`` each — 16× AWS (6 KB) and 34× Aliyun (2.8 KB) — because # GKE nodes carry their bootstrap payload in ``metadata.items``: # ``configure-sh`` 64 KB, ``kube-env`` 22 KB, ``user-data`` 5.9 KB per # instance. That is what turned 784 rows into a 74 MB write set (157 MB # of membuffer). Nothing reads those blobs: the only consumers of a # server's ``raw_data`` are ``CmdbService`` (``_project_id``, the GCP # project → cloud-scope link) and ``efficiency.py`` (``id``, for cost # matching) — both short strings. # # Keeping them was also poor secret hygiene: ``kube-env`` is the GKE # node bootstrap envelope (CA cert + kubelet credential material), and # ``raw_data`` is a plain JSON column read by any admin-tier query. # # 4096 is far above every legitimate value in a compute-instance # payload (``selfLink`` ~150 chars, licence URLs, the 1.2 KB # ``kube-labels``) and far below every bootstrap blob. Elision keeps the # key and records the original length so the shape stays diagnosable. # # Applied to SERVERS ONLY, on purpose. Certificates legitimately store # multi-KB PEM chains in ``raw_data`` and must not be trimmed; the other # capabilities are already small. Widen only with the same kind of # measurement that justified this one. _RAW_STRING_MAX_CHARS = 4096 # Depth ceiling for the recursive walk — a cyclic or pathologically # nested provider payload must not blow the Python stack inside a cron. _RAW_TRIM_MAX_DEPTH = 16 def _trim_oversized_strings(value: Any, _depth: int = 0) -> Any: """Replace every string longer than ``_RAW_STRING_MAX_CHARS``. Returns a structurally identical payload with oversized leaves swapped for ````. Dict keys are left alone (provider payload keys are field names, never blobs). """ if isinstance(value, str): if len(value) > _RAW_STRING_MAX_CHARS: return f"" return value if _depth >= _RAW_TRIM_MAX_DEPTH: return value if isinstance(value, dict): return { k: _trim_oversized_strings(v, _depth + 1) for k, v in value.items() } if isinstance(value, list): return [_trim_oversized_strings(v, _depth + 1) for v in value] return value async def _sync_servers( db: AsyncSession, integration: Integration, job_id: str, provider, *, force_sweep: bool = False, forced_meta: dict[str, Any] | None = None, ) -> tuple[int, int]: servers = await provider.list_servers() # Audit any rows the upcoming on_duplicate_key_update will # un-soft-delete (Round-27 pre-launch). See _audit_resurrections. await _audit_resurrections( db, model=CmdbServer, integration_id=str(integration.id), key_column=CmdbServer.instance_id, incoming_keys=[s.instance_id for s in servers], job_id=job_id, capability_key=ProviderCapability.SERVERS.value, audit_resource_type="cmdb_server", ) upserted = 0 for s in servers: # Elide bootstrap-script blobs before they reach the write set — # see ``_RAW_STRING_MAX_CHARS``. raw_data = _trim_oversized_strings(s.raw) stmt = mysql_insert(CmdbServer).values( id=str(uuid.uuid4()), integration_id=str(integration.id), instance_id=s.instance_id, name=s.name, status=s.status, region=s.region, zone=s.zone, vcpu=s.vcpu, memory_mb=s.memory_mb, public_ips=s.public_ips, private_ips=s.private_ips, os_type=s.os_type, image_id=s.image_id, tags=s.tags, raw_data=raw_data, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), ).on_duplicate_key_update( # PR #559: ``deleted_at`` is intentionally OMITTED from the # SET list. The earlier shape was ``deleted_at=None`` which # silently un-soft-deleted any row whose key matched — # including rows an admin had just explicitly destroyed via # ``POST /cmdb/servers/{id}/destroy`` (the cloud-side # terminate is async; UCloud / Aliyun / AWS keep returning # the instance in ``stopping`` / ``shutting-down`` state for # up to a few minutes after the API call accepts). The next # sync tick would reset ``deleted_at=NULL`` and the row # would silently reappear in the active list — same shape # as the ``feedback_on_duplicate_key_preserve_consistency`` # rule. # # A CASE-based "preserve when status=terminated" guard does # NOT work here because MySQL evaluates ON DUPLICATE KEY # UPDATE SET clauses sequentially and ``cmdb_servers.status`` # in a CASE WHEN reads the value AS UPDATED by the same # statement's earlier SET — by the time the deleted_at CASE # fires, ``status`` is already 'stopping' (new), not # 'terminated' (old). SQLAlchemy also doesn't preserve # kwarg order in ``on_duplicate_key_update`` so manual # reordering doesn't pin the column ordering either. # # The recycle bin (PR #558, ``/admin/cloud/recycle-bin``) # is the canonical un-delete path. Sweeps that mistakenly # soft-delete a healthy instance now require an explicit # admin Restore click rather than auto-resurrecting on the # next tick. Safe-sweep's 95% drop-tolerance floor + the # PR-558 sweep predicate sharpening (no ``OR is_(None)``) # already make spurious sweeps rare; the recycle bin # closes the recovery loop without coupling sync semantics # to admin-destroy semantics. name=s.name, status=s.status, region=s.region, zone=s.zone, vcpu=s.vcpu, memory_mb=s.memory_mb, public_ips=s.public_ips, private_ips=s.private_ips, os_type=s.os_type, image_id=s.image_id, tags=s.tags, raw_data=raw_data, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), ) await db.execute(stmt) upserted += 1 await _commit_upsert_batch(db, upserted) deleted = await _safe_sweep( db, model=CmdbServer, integration_id=str(integration.id), job_id=job_id, upserted=upserted, capability_label="servers", capability_key=ProviderCapability.SERVERS.value, force_sweep=force_sweep, forced_meta=forced_meta, ) await db.commit() return upserted, deleted async def _sync_domains( db: AsyncSession, integration: Integration, job_id: str, provider, *, force_sweep: bool = False, forced_meta: dict[str, Any] | None = None, ) -> tuple[int, int]: domains = await provider.list_domains() await _audit_resurrections( db, model=CmdbDomain, integration_id=str(integration.id), key_column=CmdbDomain.domain_name, incoming_keys=[d.domain_name for d in domains], job_id=job_id, capability_key=ProviderCapability.DOMAINS.value, audit_resource_type="cmdb_domain", ) upserted = 0 for d in domains: stmt = mysql_insert(CmdbDomain).values( id=str(uuid.uuid4()), integration_id=str(integration.id), domain_name=d.domain_name, registrar=d.registrar, expiry_date=d.expiry_date, auto_renew=d.auto_renew, icp_filing_no=d.icp_filing_no, status=d.status, raw_data=d.raw, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), # Migration 218: any value that came from a real driver # call is by definition ``synced``. The same flag goes # into ON DUPLICATE KEY so a row that had been ``manual`` # or ``migrated`` flips back to ``synced`` the moment a # real cloud driver speaks — cloud is authoritative. data_source="synced", ).on_duplicate_key_update( registrar=d.registrar, expiry_date=d.expiry_date, auto_renew=d.auto_renew, icp_filing_no=d.icp_filing_no, status=d.status, raw_data=d.raw, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), # ``deleted_at`` intentionally OMITTED (same rule as servers/ # certs/dns, PR #559): ``domain`` is a recycle-bin-managed type # (router ``_RECYCLE_BIN_TYPES``), so clearing deleted_at here # would silently resurrect a row an admin soft-deleted — the # registrar keeps listing it. Un-delete is the explicit Restore # path only. ``_audit_resurrections`` above already flags this. data_source="synced", ) result = await db.execute(stmt) upserted += 1 await _maybe_seed_initial_owner_history( db, result=result, model_cls=CmdbDomain, resource_type=RESOURCE_TYPE_DOMAIN, lookup_filters=[ CmdbDomain.integration_id == str(integration.id), CmdbDomain.domain_name == d.domain_name, CmdbDomain.deleted_at.is_(None), ], ) await _commit_upsert_batch(db, upserted) deleted = await _safe_sweep( db, model=CmdbDomain, integration_id=str(integration.id), job_id=job_id, upserted=upserted, capability_label="domains", capability_key=ProviderCapability.DOMAINS.value, force_sweep=force_sweep, forced_meta=forced_meta, # Migration 218: ``manual`` and ``migrated`` rows are NOT cloud- # backed (off-platform registrars like GoDaddy / Markmonitor / # Namecheap, or the legacy CMDB import for which no driver # exists — e.g. ~50 domains tagged under UCloud which has no # ``list_domains`` API). Letting the sweep see them means the # very first sync after this migration would soft-delete every # legacy row whose registrar slug doesn't match the driver # output, since they will not appear in # ``provider.list_domains()``. Mirrors the IAM sweep's # ``revoked_at IS NOT NULL`` exclusion — same rationale: the # row is intentionally kept by Helix outside the cloud's # authoritative list. extra_active_filter=[ CmdbDomain.data_source == "synced", ], ) await db.commit() return upserted, deleted async def _sync_certificates( db: AsyncSession, integration: Integration, job_id: str, provider, *, force_sweep: bool = False, forced_meta: dict[str, Any] | None = None, ) -> tuple[int, int]: """Sync cloud-cert-manager certificates (AWS ACM / Aliyun / Tencent SSL) into ``cmdb_certificates`` with ``source='cloud'``. Dedup anchor is ``dedup_key = cloud:{integration_id}:{cloud_cert_id}`` — distinct from probed rows (``probed:{host}:{port}``), so a cloud cert and a probed cert for the same hostname coexist as separate rows (different provenance). Cross-source reconciliation + fingerprint-dedup of alerts lands in PR3. """ from app.domains.cmdb.cert_service import derive_cert_status certs = await provider.list_certificates() now = datetime.now(UTC) # Forensic trail for rows this sync is about to touch that were # previously soft-deleted (mirrors servers / domains / DNS, which all # audit resurrections before their upsert). The cert path was the only # synced resource type missing this call. cert_dedup_keys = [ f"{CERT_SOURCE_CLOUD}:{integration.id}:{c.cloud_cert_id}" for c in certs ] await _audit_resurrections( db, model=CmdbCertificate, integration_id=str(integration.id), key_column=CmdbCertificate.dedup_key, incoming_keys=cert_dedup_keys, job_id=job_id, capability_key=ProviderCapability.CERTIFICATES.value, audit_resource_type="cmdb_certificate", ) upserted = 0 for c in certs: dedup_key = f"{CERT_SOURCE_CLOUD}:{integration.id}:{c.cloud_cert_id}" status = derive_cert_status(c.not_after, now) stmt = mysql_insert(CmdbCertificate).values( id=str(uuid.uuid4()), dedup_key=dedup_key, source=CERT_SOURCE_CLOUD, integration_id=str(integration.id), cloud_cert_id=c.cloud_cert_id, common_name=c.common_name, sans=c.sans, issuer=c.issuer, serial_number=c.serial_number, fingerprint_sha256=c.fingerprint_sha256, not_before=c.not_before, not_after=c.not_after, status=status, raw_data=c.raw, last_seen_sync_job_id=job_id, last_synced_at=now, first_seen_at=now, ).on_duplicate_key_update( common_name=c.common_name, sans=c.sans, issuer=c.issuer, serial_number=c.serial_number, fingerprint_sha256=c.fingerprint_sha256, not_before=c.not_before, not_after=c.not_after, status=status, raw_data=c.raw, last_seen_sync_job_id=job_id, last_synced_at=now, # ``deleted_at`` is intentionally OMITTED from the update set # (was previously ``deleted_at=None``). An admin can explicitly # retire a cloud cert via ``DELETE /cmdb/certificates/{id}`` # (router sets ``deleted_at=now()``), but the cloud provider # keeps returning the cert for minutes afterwards — clearing # ``deleted_at`` here silently revived the just-retired row on # the next sync. Same fix shipped for servers (PR #559) and DNS # records. The recycle bin is the canonical un-delete path; a # genuinely-returned swept cert is recovered by admin Restore, # not by auto-resurrection. See ``_audit_resurrections`` above # for the forensic trail. ) await db.execute(stmt) upserted += 1 await _commit_upsert_batch(db, upserted) deleted = await _safe_sweep( db, model=CmdbCertificate, integration_id=str(integration.id), job_id=job_id, upserted=upserted, capability_label="certificates", capability_key=ProviderCapability.CERTIFICATES.value, force_sweep=force_sweep, forced_meta=forced_meta, # Only sweep this integration's CLOUD rows — probed / manual rows # carry NULL integration_id and are not in the cloud's authoritative # list, so they must never be swept by a cloud sync. extra_active_filter=[ CmdbCertificate.source == CERT_SOURCE_CLOUD, ], ) await db.commit() return upserted, deleted async def _sync_dns_records( db: AsyncSession, integration: Integration, job_id: str, provider, *, force_sweep: bool = False, forced_meta: dict[str, Any] | None = None, ) -> tuple[int, int]: records = await provider.list_dns_records() # DNS unique key is ``(integration_id, zone_name, record_id)`` — # same record_id can legitimately exist in multiple zones (Aliyun / # Tencent DNSPod / Route 53 all only require record_id uniqueness # within a zone). Use composite-key resurrection check; matching by # record_id alone produced false-positive cross-zone audits during # post-fix regression review. await _audit_resurrections( db, model=CmdbDnsRecord, integration_id=str(integration.id), key_column=CmdbDnsRecord.record_id, incoming_keys=[r.record_id for r in records], secondary_column=CmdbDnsRecord.zone_name, secondary_values=[r.zone_name for r in records], job_id=job_id, capability_key=ProviderCapability.DNS_RECORDS.value, audit_resource_type="cmdb_dns_record", ) upserted = 0 for r in records: stmt = mysql_insert(CmdbDnsRecord).values( id=str(uuid.uuid4()), integration_id=str(integration.id), zone_name=r.zone_name, record_id=r.record_id, name=r.name, record_type=r.record_type, value=r.value, ttl=r.ttl, line=r.line, status=r.status, raw_data=r.raw, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), ).on_duplicate_key_update( name=r.name, record_type=r.record_type, value=r.value, ttl=r.ttl, line=r.line, status=r.status, raw_data=r.raw, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), # ``deleted_at`` is intentionally OMITTED from the update set # (was previously ``deleted_at=None`` which half-revived # tombstoned rows during the cloud-side eventual-consistency # window: admin soft-deletes record at T=0, cloud delete # propagates at T=1, but a sync started at T=0.5 still saw # the record alive and on_duplicate_key_update would clear # the freshly-set ``deleted_at`` — record reappears in lists # / acted on as if never deleted, until next sync re-sweeps). # Mirrors the lifecycle_service.py:5934 fix (scope-sync 2026- # 05-21 incident) and the on_duplicate_key_preserve_ # consistency rule. Owner / remark / last_modified_* / # last_recertify_nag_at / last_owner_review_at were already # excluded for the same "admin-set wins" semantic. ) result = await db.execute(stmt) upserted += 1 await _maybe_seed_initial_owner_history( db, result=result, model_cls=CmdbDnsRecord, resource_type=RESOURCE_TYPE_DNS_RECORD, lookup_filters=[ CmdbDnsRecord.integration_id == str(integration.id), CmdbDnsRecord.zone_name == r.zone_name, CmdbDnsRecord.record_id == r.record_id, CmdbDnsRecord.deleted_at.is_(None), ], ) await _commit_upsert_batch(db, upserted) deleted = await _safe_sweep( db, model=CmdbDnsRecord, integration_id=str(integration.id), job_id=job_id, upserted=upserted, capability_label="dns_records", capability_key=ProviderCapability.DNS_RECORDS.value, force_sweep=force_sweep, forced_meta=forced_meta, ) await db.commit() return upserted, deleted async def _sync_load_balancers( db: AsyncSession, integration: Integration, job_id: str, provider, *, force_sweep: bool = False, forced_meta: dict[str, Any] | None = None, ) -> tuple[int, int]: lbs = await provider.list_load_balancers() await _audit_resurrections( db, model=CmdbLoadBalancer, integration_id=str(integration.id), key_column=CmdbLoadBalancer.lb_id, incoming_keys=[lb.lb_id for lb in lbs], job_id=job_id, capability_key=ProviderCapability.LOAD_BALANCERS.value, audit_resource_type="cmdb_load_balancer", ) upserted = 0 for lb in lbs: stmt = mysql_insert(CmdbLoadBalancer).values( id=str(uuid.uuid4()), integration_id=str(integration.id), lb_id=lb.lb_id, name=lb.name, lb_type=lb.lb_type, network_type=lb.network_type, ip_address=lb.ip_address, region=lb.region, status=lb.status, raw_data=lb.raw, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), ).on_duplicate_key_update( name=lb.name, lb_type=lb.lb_type, network_type=lb.network_type, ip_address=lb.ip_address, region=lb.region, status=lb.status, raw_data=lb.raw, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), # ``deleted_at`` OMITTED — load_balancer is recycle-bin-managed; # clearing it would resurrect an admin-soft-deleted row (see # servers/domains, PR #559). Restore is the only un-delete path. ) result = await db.execute(stmt) upserted += 1 await _maybe_seed_initial_owner_history( db, result=result, model_cls=CmdbLoadBalancer, resource_type=RESOURCE_TYPE_LOAD_BALANCER, lookup_filters=[ CmdbLoadBalancer.integration_id == str(integration.id), CmdbLoadBalancer.lb_id == lb.lb_id, CmdbLoadBalancer.deleted_at.is_(None), ], ) await _commit_upsert_batch(db, upserted) deleted = await _safe_sweep( db, model=CmdbLoadBalancer, integration_id=str(integration.id), job_id=job_id, upserted=upserted, capability_label="load_balancers", capability_key=ProviderCapability.LOAD_BALANCERS.value, force_sweep=force_sweep, forced_meta=forced_meta, ) await db.commit() return upserted, deleted async def _sync_cdn_domains( db: AsyncSession, integration: Integration, job_id: str, provider, *, force_sweep: bool = False, forced_meta: dict[str, Any] | None = None, ) -> tuple[int, int]: domains = await provider.list_cdn_domains() await _audit_resurrections( db, model=CmdbCdnDomain, integration_id=str(integration.id), key_column=CmdbCdnDomain.domain_name, incoming_keys=[d.domain_name for d in domains], job_id=job_id, capability_key=ProviderCapability.CDN.value, audit_resource_type="cmdb_cdn_domain", ) upserted = 0 for d in domains: stmt = mysql_insert(CmdbCdnDomain).values( id=str(uuid.uuid4()), integration_id=str(integration.id), domain_name=d.domain_name, cname=d.cname, status=d.status, source_type=d.source_type, source_address=d.source_address, is_https=d.is_https, raw_data=d.raw, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), ).on_duplicate_key_update( cname=d.cname, status=d.status, source_type=d.source_type, source_address=d.source_address, is_https=d.is_https, raw_data=d.raw, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), # ``deleted_at`` OMITTED — cdn_domain is recycle-bin-managed; # clearing it would resurrect an admin-soft-deleted row (see # servers/domains, PR #559). Restore is the only un-delete path. ) result = await db.execute(stmt) upserted += 1 await _maybe_seed_initial_owner_history( db, result=result, model_cls=CmdbCdnDomain, resource_type=RESOURCE_TYPE_CDN_DOMAIN, lookup_filters=[ CmdbCdnDomain.integration_id == str(integration.id), CmdbCdnDomain.domain_name == d.domain_name, CmdbCdnDomain.deleted_at.is_(None), ], ) await _commit_upsert_batch(db, upserted) deleted = await _safe_sweep( db, model=CmdbCdnDomain, integration_id=str(integration.id), job_id=job_id, upserted=upserted, capability_label="cdn_domains", capability_key=ProviderCapability.CDN.value, force_sweep=force_sweep, forced_meta=forced_meta, ) await db.commit() return upserted, deleted async def _sync_object_storage( db: AsyncSession, integration: Integration, job_id: str, provider, *, force_sweep: bool = False, forced_meta: dict[str, Any] | None = None, ) -> tuple[int, int]: buckets = await provider.list_object_storage() # A bucket dict missing its name is unusable — skip it rather than # KeyError out of the whole capability (every other field already uses # ``.get()``; the name was the lone direct-index inconsistency). buckets = [b for b in buckets if b.get("bucket_name")] await _audit_resurrections( db, model=CmdbObjectStorage, integration_id=str(integration.id), key_column=CmdbObjectStorage.bucket_name, incoming_keys=[b["bucket_name"] for b in buckets], job_id=job_id, capability_key=ProviderCapability.OBJECT_STORAGE.value, audit_resource_type="cmdb_object_storage", ) upserted = 0 for b in buckets: stmt = mysql_insert(CmdbObjectStorage).values( id=str(uuid.uuid4()), integration_id=str(integration.id), bucket_name=b["bucket_name"], region=b.get("region"), storage_class=b.get("storage_class"), access_type=b.get("access_type"), size_bytes=b.get("size_bytes"), object_count=b.get("object_count"), status=b.get("status", "active"), owner_raw=b.get("owner_raw"), raw_data=b.get("raw", {}), last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), ).on_duplicate_key_update( region=b.get("region"), storage_class=b.get("storage_class"), access_type=b.get("access_type"), size_bytes=b.get("size_bytes"), object_count=b.get("object_count"), status=b.get("status", "active"), # Round-49: do NOT overwrite ``owner_raw`` on duplicate-key # update. The cloud-side value (Azure subscription-id, GCP # bucket-ACL entity) lands here on first discovery via # ``.values(...)`` above. The admin PATCH path # ``CmdbService.annotate_object_storage`` is the only writer # that updates this column from the human side and audits # before/after — letting hourly sync stomp the admin's # annotation back to the provider string defeats that # workflow (same bug-class as the Round 33 family). raw_data=b.get("raw", {}), last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), deleted_at=None, ) await db.execute(stmt) upserted += 1 await _commit_upsert_batch(db, upserted) deleted = await _safe_sweep( db, model=CmdbObjectStorage, integration_id=str(integration.id), job_id=job_id, upserted=upserted, capability_label="object_storage", capability_key=ProviderCapability.OBJECT_STORAGE.value, force_sweep=force_sweep, forced_meta=forced_meta, ) await db.commit() return upserted, deleted def _email_from_iam_user_payload( user_name: str, tags: dict | None, ) -> str | None: """Same logic as :func:`lifecycle_service._email_from_iam_user`, duplicated here to avoid a circular import (sync.py is imported by every cmdb module).""" if user_name and "@" in user_name: return user_name.lower() tags = tags or {} norm = {str(k).lower(): str(v) for k, v in tags.items() if v is not None} for key in ("email", "user_email", "email_address"): v = norm.get(key, "").strip() if "@" in v: return v.lower() v = norm.get("owner", "").strip() if "@" in v: return v.lower() return None async def _resolve_iam_owner_person_id( db: AsyncSession, integration_id: str, ) -> int: """Backfill ``cmdb_iam_users.owner_person_id`` from the user_name/tags email mapping, with email-local-part fallback. Called at the end of :func:`_sync_iam_users`. The IAM sync itself only writes the cloud-side payload; identity resolution is a Helix concern, kept separate so the upsert stays a pure mirror. Two-tier resolution: 1. **Direct email match.** Extract email from user_name (if it has ``@``) or tags (``email`` / ``Owner``). Look up by ``IdentityPerson.work_email``. 2. **Email-local-part fallback.** AliCloud RAM creates users like ``li.da`` / ``xiaowu.wang`` (no ``@``) and admins rarely set the optional ``Email`` field on RAM users. Match these against ``IdentityPerson.work_email`` whose local-part (chars before ``@``) equals the IAM user_name **and** there's exactly one matching person. The "exactly one" guard is important — a generic IAM user like ``svc-key`` could otherwise bind to whichever ``svc-key@…`` happens to exist, which is a wrong attribution we'd never automatically undo. Returns the count of rows newly bound to a person. """ from app.identity.person_models import IdentityPerson rows = (await db.execute( select(CmdbIamUser).where( CmdbIamUser.integration_id == integration_id, CmdbIamUser.owner_person_id.is_(None), CmdbIamUser.deleted_at.is_(None), ) )).scalars().all() if not rows: return 0 bound = 0 # ── Tier 1: direct email match ───────────────────────────── emails: dict[str, list[str]] = {} # email -> list[iam_user_id] no_email_user_names: dict[str, list[str]] = {} # local-part -> list[iam_user_id] for r in rows: email = _email_from_iam_user_payload(r.user_name or "", r.tags) if email: emails.setdefault(email, []).append(str(r.id)) elif r.user_name: no_email_user_names.setdefault(r.user_name.lower(), []).append(str(r.id)) if emails: # Round-49: exclude leavers from the auto-resolve. A freshly- # discovered IAM user whose email matches a recent leaver # would otherwise be bound to that leaver — downstream # offboarding sweeps already-processed them, so the IAM user # ends up owned by someone HR considers gone, and the leaver- # grant sweep needs an extra cycle to catch it. person_rows = (await db.execute( select(IdentityPerson.id, IdentityPerson.work_email).where( func.lower(IdentityPerson.work_email).in_(list(emails.keys())), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ) )).all() email_to_person: dict[str, str] = { str(row[1]).strip().lower(): str(row[0]) for row in person_rows } for email, iam_user_ids in emails.items(): person_id = email_to_person.get(email) if not person_id: continue await db.execute( update(CmdbIamUser) .where(CmdbIamUser.id.in_(iam_user_ids)) .values(owner_person_id=person_id) ) bound += len(iam_user_ids) # ── Tier 2: email-local-part fallback ────────────────────── if no_email_user_names: local_parts = list(no_email_user_names.keys()) # Pull every active person whose work_email's local-part # matches one of our candidates. ``substring_index`` is # MySQL/TiDB-portable. # Round-49: exclude leavers — same rationale as Tier-1. person_rows = (await db.execute( select(IdentityPerson.id, IdentityPerson.work_email).where( func.lower( func.substring_index(IdentityPerson.work_email, "@", 1) ).in_(local_parts), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ) )).all() local_to_persons: dict[str, list[str]] = {} for pid, email in person_rows: email_str = str(email or "").strip().lower() if "@" not in email_str: continue local = email_str.split("@", 1)[0] local_to_persons.setdefault(local, []).append(str(pid)) for user_name, iam_user_ids in no_email_user_names.items(): persons = local_to_persons.get(user_name) or [] # Only bind when EXACTLY one Helix person owns this # local-part. ``shuai`` could be ``shuai.feng@`` AND # ``shuai.zhang@`` — a wrong bind on auto-resolve has # security weight (grants flow through this), so we # prefer to leave it unbound and let an admin assign. if len(persons) != 1: continue await db.execute( update(CmdbIamUser) .where(CmdbIamUser.id.in_(iam_user_ids)) .values(owner_person_id=persons[0]) ) bound += len(iam_user_ids) if bound: await db.commit() return bound async def _sync_iam_users( db: AsyncSession, integration: Integration, job_id: str, provider, *, force_sweep: bool = False, forced_meta: dict[str, Any] | None = None, ) -> tuple[int, int]: # NOTE — deliberately NOT wired to ``_commit_upsert_batch`` (2026-08-08), # unlike the other seven capabilities. IAM's upsert clears # ``revoked_at`` on cloud-side revivals and the matching shadow-IAM # security alert is enqueued into the outbox AFTER the loop, on # purpose, so the de-revocation and its alert commit together. A # batch boundary inside the loop would let a crash land "revoked_at # cleared, nobody told" — trading a security-signal guarantee for # memory headroom this capability does not need: IAM payloads are # a few hundred bytes and the largest tenant lists ~2k users, three # orders of magnitude below the membuffer pressure that motivated # batching. Revisit only if that stops being true. users = await provider.list_iam_users() # Round-27 pre-launch (2026-05-25): IAM has its own purpose-built # ``revoked_at`` revival audit + IT-admin notification (line ~2167 # below). That path covers the case where Helix explicitly revoked # the IAM user and the cloud re-listed it (shadow-IAM). But the # plain ``deleted_at IS NOT NULL`` resurrection — where safe-sweep # had soft-deleted the row because the cloud previously dropped # the user — is also a meaningful forensic event that the other 6 # ``_sync_*`` capabilities audit via ``_audit_resurrections``. Add # it here for parity. Uses composite key # ``(user_name, sub_account_id)`` because the unique constraint # spans that pair (multi-sub-account RAM iteration would otherwise # cross-flag same-name users from different sub-accounts). await _audit_resurrections( db, model=CmdbIamUser, integration_id=str(integration.id), key_column=CmdbIamUser.user_name, incoming_keys=[u["user_name"] for u in users], secondary_column=CmdbIamUser.sub_account_id, secondary_values=[ normalize_iam_sub_account_id( (u.get("tags") or {}).get("__sub_account_id"), ) for u in users ], job_id=job_id, capability_key=ProviderCapability.IAM.value, audit_resource_type="cmdb_iam_user", ) # Detect revival-after-Helix-revoke: if the cloud lists a user that # Helix had previously marked ``revoked_at IS NOT NULL`` (i.e. an # admin manually re-created the user in the cloud console after our # offboarding deleted them), we must clear ``revoked_at`` so the # row stops hiding in the "revoked" tab — otherwise the user holds # real cloud credentials but Helix has zero visibility (shadow IAM). # Snapshot the names that were previously revoked, so we can log # the security-relevant transition after the upsert pass. seen_user_names = {u["user_name"] for u in users} # #1779: the revival snapshot MUST key on the same composite # ``(user_name, sub_account_id)`` the unique constraint / upsert use. # Selecting by ``user_name`` alone cross-flagged a same-named user who # is active in sub-account A but revoked+deleted in sub-account B: the # B row (never touched by the upsert, which matches the full triple) # was reported "revived", firing a `[Security] Shadow-IAM revival` # alert + a ``cmdb.iam_user.revival_detected`` audit row on EVERY sync # — while its ``revoked_at`` stayed set, so the alert claimed a clear # that never happened and re-fired forever. Only a row whose exact # (user_name, sub_account_id) appears in THIS listing is genuinely # being un-revoked by the upsert below. seen_iam_pairs = { ( u["user_name"], normalize_iam_sub_account_id( (u.get("tags") or {}).get("__sub_account_id"), ), ) for u in users } # Snapshot revival candidates BEFORE the upsert: rows whose # revoked_at is set AND > 5 minutes old. The upsert below will clear # revoked_at for exactly these rows. We need pre-state to emit a # loud audit + IT-admin notification — without it the de-revocation # is whispered (logger.warning only) and the shadow-IAM incident # the warning was added to surface stays invisible. revival_cutoff_pre = datetime.now(UTC) - timedelta(minutes=5) revived_rows: list[CmdbIamUser] = [] if seen_user_names: revived_rows = [ r for r in ( await db.execute( select(CmdbIamUser).where( CmdbIamUser.integration_id == str(integration.id), CmdbIamUser.user_name.in_(seen_user_names), CmdbIamUser.revoked_at.isnot(None), CmdbIamUser.revoked_at < revival_cutoff_pre, ) ) ).scalars().all() # Composite match: the DB upsert only clears revoked_at for the # exact (user_name, sub_account_id) triple in this listing. if (r.user_name, normalize_iam_sub_account_id(r.sub_account_id)) in seen_iam_pairs ] revived_names: set[str] = {r.user_name for r in revived_rows} if revived_names: logger.warning( "iam_sync %s: cloud-side revival detected for previously " "revoked IAM users %s — clearing revoked_at; investigate " "out-of-band re-creation", integration.name, sorted(revived_names), ) upserted = 0 # Cleanup-then-sync race protection: a freshly-revoked IAM user # (e.g. offboarding T+0 just ran ``user_revoke``) is still listed by # the cloud's eventually-consistent IAM API for tens of seconds. # Without a guard, the next sync's ``ON DUPLICATE KEY UPDATE`` would # unconditionally clear ``revoked_at``/``revoke_reason``, undoing the # offboarding action. Only clear when the existing revocation is # older than 5 minutes — past that we trust the warning logged # above and treat it as a real cloud-side revival. revival_cutoff = datetime.now(UTC) - timedelta(minutes=5) for u in users: create_date = u.get("create_date") last_login = u.get("last_login_at") # boto3 returns tz-aware datetimes; SQLAlchemy is fine with those # directly, but driver-internal calls may have already ISO-stringified # them when _json_safe walked the raw dict — normalize. if isinstance(create_date, str): try: create_date = datetime.fromisoformat(create_date.replace("Z", "+00:00")) except ValueError: create_date = None if isinstance(last_login, str): try: last_login = datetime.fromisoformat(last_login.replace("Z", "+00:00")) except ValueError: last_login = None # Conditional clear of revoked_at: keep the existing value when # the revocation is fresh (<5 min old, likely a cleanup-then-sync # race), otherwise clear it (real revival). Expressed as a MySQL # ``IF`` so the decision happens in-DB without a pre-read. # Note: in ``ON DUPLICATE KEY UPDATE``, bare column references # resolve to the EXISTING (pre-update) row value, which is what # we need here to compare the pre-existing ``revoked_at`` to the # cutoff. revoked_at_keep = func.if_( CmdbIamUser.revoked_at < revival_cutoff, None, CmdbIamUser.revoked_at, ) revoke_reason_keep = func.if_( CmdbIamUser.revoked_at < revival_cutoff, None, CmdbIamUser.revoke_reason, ) # Resolve the sub-account this RAM user lives in. Provider-side # tagging uses ``__sub_account_id`` (master sentinel ``"__master__"`` # or the numeric Aliyun AccountId of the RD member). See # ``normalize_iam_sub_account_id`` for the canonical mapping — # any drift from it re-breaks migration 120 / migration 157. sub_account_id = normalize_iam_sub_account_id( (u.get("tags") or {}).get("__sub_account_id"), ) stmt = mysql_insert(CmdbIamUser).values( id=str(uuid.uuid4()), integration_id=str(integration.id), user_name=u["user_name"], sub_account_id=sub_account_id, arn=u.get("arn"), user_id_external=u.get("user_id_external"), create_date=create_date, last_login_at=last_login, attached_policies=u.get("attached_policies") or [], groups=u.get("groups") or [], mfa_enabled=bool(u.get("mfa_enabled")), # Tri-state. A provider that cannot read MFA says # ``unknown`` rather than borrowing the bool's False, # which the compliance view reads as "MFA is off". mfa_state=str(u.get("mfa_state") or "unknown"), tags=u.get("tags") or {}, raw_data=u.get("raw") or {}, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), ).on_duplicate_key_update( arn=u.get("arn"), user_id_external=u.get("user_id_external"), create_date=create_date, last_login_at=last_login, attached_policies=u.get("attached_policies") or [], groups=u.get("groups") or [], mfa_enabled=bool(u.get("mfa_enabled")), # Tri-state. A provider that cannot read MFA says # ``unknown`` rather than borrowing the bool's False, # which the compliance view reads as "MFA is off". mfa_state=str(u.get("mfa_state") or "unknown"), tags=u.get("tags") or {}, raw_data=u.get("raw") or {}, last_seen_sync_job_id=job_id, last_synced_at=datetime.now(UTC), deleted_at=None, # Cloud is source of truth for liveness — if the user is # back AND the revocation is older than 5 min, clear the # Helix revocation record so the UI stops hiding the row. # The warning logged above flags the transition for review. revoked_at=revoked_at_keep, revoke_reason=revoke_reason_keep, ) await db.execute(stmt) upserted += 1 # Pre-launch P0 fix (Round 45, CR-202): emit a loud audit row + # IT-admin notification for every detected IAM revival. Previously # only logger.warning fired — the shadow-IAM condition (the very # incident the warning was added to surface) was therefore invisible # in audit_logs and no human was paged. Write inside the same # transaction so the audit chain is atomic with the upsert. revival_alerts_enqueued = 0 if revived_rows: async with best_effort( db, label="sync best-effort step", log=logger, ) as be_2577_step: from app.audit.service import AuditService # noqa: PLC0415 audit_svc = AuditService(db) for revived in revived_rows: await audit_svc.log( trace_id=job_id, actor_type="system", actor_user_id=None, action="cmdb.iam_user.revival_detected", resource_type="cmdb_iam_user", resource_id=str(revived.id), status="warning", detail={ "integration_id": str(integration.id), "integration_name": integration.name, "user_name": revived.user_name, "prior_revoked_at": ( revived.revoked_at.isoformat() if revived.revoked_at else None ), "prior_revoke_reason": revived.revoke_reason, "note": ( "Cloud-side admin re-created an IAM user " "Helix had marked revoked. Investigate " "out-of-band re-creation; this is the " "shadow-IAM threat pattern." ), }, ) if not be_2577_step.ok: logger.warning( "iam_sync %s: failed to write revival audit rows: %s", integration.name, be_2577_step.reason(), ) # Dispatch a Feishu + email push to IT admins so the revival # becomes a paged event, not a buried log line. Best-effort — # never block the sync on push failure. async with best_effort( db, label="sync best-effort step", log=logger, ) as be_2615_step: from app.auth.models import AuthRole, AuthUserRole # noqa: PLC0415 # #1779: was ``app.identity.models`` — a module that does not # exist, so this block ALWAYS raised ImportError (caught below) # and the shadow-IAM revival IT-admin push NEVER sent, leaving # the alert as a buried log line — defeating the whole point of # the paged-event channel this fix restores. from app.identity.person_models import IdentityPerson # noqa: PLC0415 from app.notifications.service import ( # noqa: PLC0415 NotificationEnvelope, NotificationService, ) admin_ids = list( ( await db.execute( select(AuthUserRole.user_id) .join( AuthRole, AuthRole.id == AuthUserRole.role_id, ) .where( AuthRole.name.in_(("global_admin", "it_admin")) ) ) ).scalars().all() ) if admin_ids: notify = NotificationService(db) names_preview = ", ".join(sorted(revived_names)[:5]) + ( "…" if len(revived_names) > 5 else "" ) body = ( f"[Shadow-IAM Alert] {len(revived_rows)} previously-" f"revoked IAM user(s) on {integration.name} " f"reappeared in cloud listing: {names_preview}. " "Helix is clearing the revocation tombstone. " "Investigate out-of-band re-creation immediately. " "Audit: action=cmdb.iam_user.revival_detected" ) for admin_id in admin_ids: admin = await db.get(IdentityPerson, admin_id) if admin is None or not admin.work_email: continue if (admin.employment_status or "").lower() != "active": continue try: # Outbox-only ENQUEUE, not dispatch: the enclosing # IAM sync holds INSERT/ON-DUPLICATE row locks on # ``cmdb_iam_users`` (and each outbox row FK-locks the # recipient's ``identity_persons`` row) until the # commit below. dispatch()'s synchronous Feishu+email # HTTP would hold all of that open across N_admins×2 # external calls — the "长事务群发锁" login-outage # class. Enqueue the durable rows now; the spawned # drain after commit (below) delivers them promptly, # and the hourly ``notification.outbox_retry`` backstops. await notify.enqueue( NotificationEnvelope( recipient=admin, title=( f"[Security] Shadow-IAM revival on " f"{integration.name}" ), body_text=body, channels=["feishu", "email"], template_key="cmdb_iam_revival", # Land the click on the IAM-users page so # the admin can investigate the revived # users; without ``url`` the inbox row was # a dead click. url="/admin/cloud/iam-users", context={ "integration_name": integration.name, "count": len(revived_rows), "user_names": sorted(revived_names), # ``user_names`` is a list, so a template # that interpolated it would render a # Python repr. ``names_preview`` is the # same truncated join the body above # composes — the template placeholder. "names_preview": names_preview, }, ), ) revival_alerts_enqueued += 1 except Exception as push_exc: # noqa: BLE001 logger.warning( "iam_sync %s: revival enqueue to admin %s " "failed: %s", integration.name, admin_id, push_exc, ) if not be_2615_step.ok: logger.warning( "iam_sync %s: revival notification dispatch failed: %s", integration.name, be_2615_step.reason(), ) # IAM is the one capability with a "tracked but cloud-deleted" # lifecycle state: Helix-revoked users (admin disowned / offboarding # called ``DeleteUser`` + stamped ``revoked_at``) are gone from the # cloud's ``list_iam_users`` AND still present here as history. The # safety floor must compare cloud-active vs Helix-active to match # what the provider actually returns — otherwise a tenant that has # ever revoked a user trips the 5%-drop guard forever (admin report # 2026-05-14: 9 active + 11 revoked → 9 vs 20 = 55% drop locked # every cron tick). deleted = await _safe_sweep( db, model=CmdbIamUser, integration_id=str(integration.id), job_id=job_id, upserted=upserted, capability_label="iam_users", capability_key=ProviderCapability.IAM.value, force_sweep=force_sweep, forced_meta=forced_meta, extra_active_filter=[CmdbIamUser.revoked_at.is_(None)], ) await db.commit() # Revival alerts were ENQUEUED (durable outbox rows) inside the now- # committed transaction; kick a background drain so a shadow-IAM # security alert delivers in seconds rather than waiting up to an hour # for the ``notification.outbox_retry`` cron. Fire-and-forget — the # cron is the backstop if this drain dies mid-flight. if revival_alerts_enqueued: from app.core.background import spawn from app.notifications.service import drain_due_outbox_in_background spawn( drain_due_outbox_in_background(), name=f"iam_revival_drain:{job_id}", ) # After the upsert/sweep, backfill owner_person_id for any IAM-user # row whose user_name or Owner-tag email maps to an active Helix # person. We do NOT block the sync on identity resolution failure — # an unmapped row is still a valid mirror of the cloud-side data. try: bound = await _resolve_iam_owner_person_id(db, str(integration.id)) if bound: logger.info( "iam_sync %s: bound %d previously-unowned IAM users to " "their identity persons", integration.name, bound, ) except Exception as exc: # noqa: BLE001 logger.warning( "iam_sync %s: owner backfill failed (non-fatal): %s", integration.name, exc, ) return upserted, deleted async def _fail_job(db: AsyncSession, job_id: str, message: str) -> None: await db.execute( update(CmdbSyncJob) .where(CmdbSyncJob.id == uuid.UUID(job_id)) .values(status="failed", finished_at=datetime.now(UTC), error_message=message) ) await db.commit()