"""Adapter shared by all cloud drivers. Wraps an existing :class:`CloudProvider` implementation (AWS / Aliyun / Tencent) with the :class:`CloudDriver` interface. The only per-provider work is the field schema + how to instantiate the underlying class — the cloud discovery logic stays where it lives today. """ from __future__ import annotations import asyncio import dataclasses import logging from collections.abc import Callable from typing import Any from app.domains.cmdb.providers.base import ( CloudProvider, ProviderCapability, ProviderCredentials, ) from app.domains.cmdb.providers.exceptions import ( CapabilityNotSupported, ProviderAuthError, redact_leg_entry, redact_provider_error, ) from app.integrations.driver_base import CloudDriver from app.integrations.schemas import IntegrationTestResult logger = logging.getLogger(__name__) # Cap the per-leg breakdown shown in connection-test messages so a # 60-project GCP integration with org-wide 403 doesn't render a # multi-kilobyte ``last_test_message`` row in the integrations table. _LEG_SUMMARY_MAX = 5 # Hard cap on the per-call leg payload returned in # ``IntegrationTestResult.detail`` (which the test-connection endpoint # echoes verbatim to the admin browser). A 500-project GCP integration # under org-wide 403 produced 57 entries in the recorded prod incident; # a malicious sub-account / pathological tenant could in principle push # this much higher. ``_LEG_SUMMARY_MAX`` only caps the rendered # *message* — without an independent cap on the persisted list the # admin browser would receive multi-MB responses. _LEG_DETAIL_MAX = 50 # Per-capability probe timeout in test_connection. Each cap may fan out # (AWS regions, GCP projects) so we apply the same 60s bound the sync # layer uses for fast feedback — admins clicking ``测试`` expect a # response within a few seconds when credentials work; a per-cap hang # means the credential observably can't reach that surface. _CAPABILITY_PROBE_TIMEOUT_S = 60 # Capability → CloudProvider attribute name. The mapping is the same # one ``cmdb.sync`` uses (the per-capability ``_sync_X`` helpers # directly call these ``list_X`` methods). Keeping it here lets # ``test_connection`` walk the same surface area the sync path will, # without each driver knowing which provider methods exist. _CAPABILITY_PROBE_METHOD: dict[str, str] = { "servers": "list_servers", "domains": "list_domains", "dns_records": "list_dns_records", "load_balancers": "list_load_balancers", "cdn": "list_cdn_domains", "object_storage": "list_object_storage", "iam": "list_iam_users", } # Capabilities we deliberately exclude from test_connection. ``servers`` # is always probed (the universal cheapest call); ``user_revoke`` / # billing / cost_facts are write-side or money-cost paths we won't # burn quota / dollars on for a health check. Admin can still use the # sync history to see those capabilities' state. _CAPABILITY_PROBE_EXCLUDE: frozenset[str] = frozenset({ "user_revoke", "billing", "billing_daily", "cost_facts", }) def account_id_candidates( raw_id: str, derive: Callable[[str], str], ) -> list[str]: """Provider user names to try for a revoke, most-literal first. ``CloudAccountRevokeContext.account_id`` is documented as *the provider's user handle*, with an email accepted as a convenience. The three IAM-mode drivers implemented that convenience as "if it contains ``@``, take the local part" — which silently destroys the common case where the handle **is** an address. Prod 2026-08-12: `revoke_account("tangpeng@pingcap.com")` looked up CAM user ``tangpeng``, got ``UserNotExist``, and returned ``ok=True, note="user already absent"`` while the real sub-user ``tangpeng@pingcap.com`` kept **AdministratorAccess**. 17 of the 22 live CAM users on that tenant have ``@`` in their name, so the offboarding kill-switch was a no-op for 77% of it — reporting success every time. Same shape in the Aliyun RAM and UCloud drivers. Returning candidates instead of one guess lets each driver probe existence and act on the handle that is really there, while keeping the documented convenience for tenants whose users genuinely are named after the local part (AliCloud RAM: ``li.da``, ``xiaowu.wang``). "Neither candidate exists" is the only state that may still be reported as already-absent. """ raw = (raw_id or "").strip() if not raw: return [] out = [raw] if "@" in raw: derived = (derive(raw) or "").strip() if derived and derived != raw: out.append(derived) return out def _snapshot_partial_failures(provider: CloudProvider) -> list[dict[str, Any]]: """Drain ``provider.partial_failures`` and return a redacted copy. The probe loop calls multiple ``list_X`` methods on the same provider instance; each call may append to the shared ``partial_failures`` list. Snapshot between calls so each cap's legs reflect ONLY that cap's work, then clear so the next cap starts clean. Mirrors ``CloudProvider.consume_partial_failures`` on the sync path. """ raw = getattr(provider, "partial_failures", None) if not isinstance(raw, list): return [] snapshot = [ redact_leg_entry(e) for e in raw if isinstance(e, dict) ] raw.clear() return snapshot # Re-export the shared leg redactor under a module-local alias so tests # importing ``_redact_leg`` (the original location) keep working. The # canonical home is ``app.domains.cmdb.providers.exceptions`` because # both this module AND ``app.domains.cmdb.sync`` use it. _redact_leg = redact_leg_entry def _summarize_leg_failures(legs: list[dict[str, Any]]) -> str: """Human-readable, length-bounded leg-failure summary. Groups by ``reason`` so an all-403 sweep collapses to a single line (``UnauthorizedOperation × 5``) instead of repeating the same reason once per region/project. Each named leg is redacted through ``redact_provider_error`` defensively — current providers emit deterministic leg strings, but the wrap costs nothing and protects the on-disk + on-wire surface against future drivers that build leg labels from raw SDK error payloads. Round-26 enhancement: when legs carry the optional ``error_message`` field (currently emitted by the GCP provider for 4xx with a parseable ``error.message``), the most-common message in each reason group is appended to the rendered line so the admin sees actionable text directly: ``http_403 × 1 [project:tidb-376913] (msg: Permission compute.instances.list denied on resource…)`` Most-common (mode) rather than the first message keeps the summary stable across cron ticks when the leg arrival order varies. Tiebroken by lexicographic-min for determinism. Three independent caps keep the rendered string bounded even when a pathological tenant produces thousands of distinct reasons: * ``_LEG_SUMMARY_MAX`` legs per reason group (inner sample) * ``_LEG_SUMMARY_MAX`` distinct reason groups (outer list) — with a ``+M more reason(s)`` tail when truncated * 2000-char hard ceiling on the joined string as a final guard """ if not legs: return "" by_reason: dict[str, list[str]] = {} msgs_by_reason: dict[str, list[str]] = {} for entry in legs: reason = redact_provider_error(entry.get("reason") or "unknown") leg = redact_provider_error(entry.get("leg") or "?") by_reason.setdefault(reason, []).append(leg) msg = entry.get("error_message") if isinstance(msg, str) and msg.strip(): # Round-26 review P1-5: tighten the per-message budget to # 120 chars (down from 200) so a ≤5-reason summary stays # under the ~1KB ``last_test_message`` cap even when every # reason group carries an error_message. The full text is # still available verbatim in ``capability_results.partial_failures`` # — this is just the alert-card sample. msgs_by_reason.setdefault(reason, []).append( redact_provider_error(msg, max_len=120), ) # Round-21 review: deterministic ordering before slicing. Pre-fix # the slice took ``dict insertion order``, which is leg-input # order — same {reason set} could produce different signatures # across cron ticks depending on per-region fan-out variance, # making the failure-notifier dedup key flip and re-paging admin # for the same conceptual failure. Sort by ``(count desc, reason # asc)`` so the same multiset of reasons always renders the same # summary string regardless of how the legs arrived. items = sorted( by_reason.items(), key=lambda kv: (-len(kv[1]), kv[0]), ) rendered = items[:_LEG_SUMMARY_MAX] truncated_reasons = len(items) - len(rendered) parts: list[str] = [] for reason, leg_names in rendered: # Sort the inner leg list too — without this, the SAME # ``{reason: count}`` multiset would still render different # strings across cron ticks (per-region fan-out variance # changes the leg arrival order). The dedup-signature # stability the outer ordering buys is wasted if the inner # sample still varies. sorted_legs = sorted(leg_names) sample = ", ".join(sorted_legs[:_LEG_SUMMARY_MAX]) if len(sorted_legs) > _LEG_SUMMARY_MAX: sample += f" (+{len(sorted_legs) - _LEG_SUMMARY_MAX} more)" line = f"{reason} × {len(sorted_legs)} [{sample}]" # Append the most-common error_message for this reason group, # if any. Mode (highest count) with lexicographic-min tiebreak # keeps the rendered summary stable across cron ticks. msgs = msgs_by_reason.get(reason) or [] if msgs: counts: dict[str, int] = {} for m in msgs: counts[m] = counts.get(m, 0) + 1 # Sort by (count desc, message asc) so the same multiset of # messages always picks the same sample. ``next(iter(...))`` # on a sorted list gives the deterministic mode. top = sorted( counts.items(), key=lambda kv: (-kv[1], kv[0]), )[0][0] line += f" (msg: {top})" parts.append(line) if truncated_reasons > 0: parts.append(f"+{truncated_reasons} more reason(s)") joined = "; ".join(parts) # Final guard against an unforeseen reason-explosion shape. if len(joined) > 2000: joined = joined[:1997] + "..." return joined class _CloudDriverBase(CloudDriver): """Intermediate base: derivatives just set ``provider_cls``.""" provider_cls: type[CloudProvider] # Per-driver allowlist of config keys that get forwarded into the # underlying CloudProvider. Empty means "forward everything" # (legacy behaviour kept for drivers that haven't declared their # surface yet). Drivers declaring this list strip stray fields the # admin may have added under arbitrary keys, so credentials and # private settings never leak past the boundary they're declared at. required_credential_fields: list[str] = [] def _provider(self) -> CloudProvider: cfg = self.config allowlist = type(self).required_credential_fields if allowlist: allowed = set(allowlist) # ``_alias`` is the display alias the integration row sets; # always pass-through so account_alias keeps working. allowed.add("_alias") fields = { k: str(v) for k, v in cfg.items() if k in allowed and v is not None } else: fields = {k: str(v) for k, v in cfg.items() if v is not None} return self.provider_cls( ProviderCredentials( provider_id=str(self.integration_id), provider_type=self.provider, account_alias=str(cfg.get("_alias") or self.integration_id), fields=fields, ) ) async def test_connection(self) -> IntegrationTestResult: provider = self._provider() try: # Pre-flight credential probe: when the provider exposes a # cheap "verify caller identity" method (AWS STS today, # other providers can adopt the same interface), call it # FIRST. Benefits: # 1. Auth failures short-circuit before the expensive # per-region ``list_servers`` fan-out (5min+ on AWS). # 2. Burns no service-specific API quota — STS Get # CallerIdentity is free and IAM-permission-less. # 3. Returns the bound account_id so a future schema # change can compare it against the integration's # expected account (currently only logged; "wrong # account" detection requires storing the expected # ARN on the integration row). # Missing method (other drivers) falls through to the # legacy list_servers path unchanged. verify = getattr(provider, "verify_caller_identity", None) if callable(verify): # Let any failure propagate to the outer ``except # Exception`` so we don't continue to list_servers on a # broken AK (which would burn quota for the same # negative result). Successful identity is logged for # forensic correlation. identity = await verify() logger.info( "cloud test_connection identity: provider=%s " "account_id=%s arn=%s", self.provider, identity.get("account_id") or "(unset)", identity.get("arn") or "(unset)", ) # Probe ``servers`` first — it's the universal cheapest # call every cloud driver implements and proves IAM / # network reachability before we burn quota on # capability-specific endpoints. ``servers`` is implicitly # part of every cloud integration's enabled set. servers_probe = await self._probe_capability( provider, "servers", ) # Per-capability dry-run — the prior implementation reported # ``OK · enumerated 692 servers`` for an integration whose # ``object_storage`` capability returned 100% auth_error on # every leg (prod 2026-05-16). The single-capability test # let an integration "pass" while a configured capability # was structurally broken. Walk the admin-enabled set and # probe each one with the cheapest list_* surface; aggregate # to ``capability_probes`` so the admin UI can render a # per-capability table. ``servers`` is already covered # above; ``user_revoke`` / billing / cost_facts are excluded # (write-side or paid-API). probes: dict[str, dict[str, Any]] = {"servers": servers_probe} for cap in self.enabled_capabilities: if cap == "servers" or cap in _CAPABILITY_PROBE_EXCLUDE: continue if cap in probes: continue probes[cap] = await self._probe_capability(provider, cap) return self._aggregate_test_result(probes) except Exception as e: # noqa: BLE001 # Run the raw exception string through the credential-safe # redactor before it lands in ``last_test_message`` (which # is rendered on the admin UI verbatim). boto3 ClientError # / SDK error strings can carry signed-URL fragments and # ``AccessKeyId=`` prefixes — same rationale as the sync # path's ``redact_provider_error`` wrap. return IntegrationTestResult( ok=False, message=f"{type(e).__name__}: {redact_provider_error(e)}", detail={"type": type(e).__name__}, ) finally: try: await provider.close() except Exception: # noqa: BLE001, S110 pass async def _probe_capability( self, provider: CloudProvider, capability: str, ) -> dict[str, Any]: """Dry-run one ``list_X`` capability and structure the result. Returns a dict with ``status`` ∈ {ok, partial_failure, skipped, failed} plus a sample count and (when applicable) the leg-failure breakdown. The shape mirrors ``cmdb_sync_jobs.capability_results`` so the admin UI can render the same table from either source. Bounded by ``_CAPABILITY_PROBE_TIMEOUT_S`` — a slow probe is as actionable as a fast failure: a 60s ceiling per cap keeps the test-connection click responsive even for AWS multi-region fan-outs. """ method_name = _CAPABILITY_PROBE_METHOD.get(capability) if not method_name: return {"status": "skipped", "reason": "no probe defined"} fn = getattr(provider, method_name, None) if not callable(fn): return {"status": "skipped", "reason": "not implemented"} # Provider supports() check: capability declared but list method # explicitly raises CapabilityNotSupported on some flavours. cap_enum = None try: cap_enum = ProviderCapability(capability) except ValueError: cap_enum = None if cap_enum is not None and hasattr(provider, "supports"): try: if not provider.supports(cap_enum): return { "status": "skipped", "reason": "not supported by provider", } except Exception: # noqa: BLE001 # ``supports`` is metadata — never let it fail the probe. pass try: rows = await asyncio.wait_for( fn(), timeout=_CAPABILITY_PROBE_TIMEOUT_S, ) except TimeoutError: # Snapshot any legs accumulated before the timeout so the # admin sees what made progress. legs = _snapshot_partial_failures(provider) return { "status": "failed", "reason": ( f"capability probe exceeded " f"{_CAPABILITY_PROBE_TIMEOUT_S}s (cloud API " f"hung or fan-out too large)" ), "timed_out": True, "partial_failures": legs[:_LEG_DETAIL_MAX], } except CapabilityNotSupported as exc: _snapshot_partial_failures(provider) # clear for next cap return { "status": "skipped", "reason": redact_provider_error(exc), } except ProviderAuthError as exc: _snapshot_partial_failures(provider) return { "status": "failed", "reason": f"auth error: {redact_provider_error(exc)}", } except Exception as exc: # noqa: BLE001 legs = _snapshot_partial_failures(provider) # Surface the exception type name alongside the redacted # message — the legacy ``test_connection`` exception arm # composed ``f"{type(e).__name__}: {redact_provider_error(e)}"`` # so admins reading ``last_test_message`` could distinguish # ``RuntimeError: …`` (driver-side) from ``HttpError: …`` # (cloud-side). Dropping the type-prefix in the per-cap # rewrite was a regression caught in code review. return { "status": "failed", "reason": ( f"{type(exc).__name__}: {redact_provider_error(exc)}" ), "partial_failures": legs[:_LEG_DETAIL_MAX], } # Drain partial_failures for this capability and classify. # Per-cap `partial_failures` stores the full leg list — the # aggregate seam (`_aggregate_test_result`) does the # `_LEG_DETAIL_MAX + sentinel` truncation on the legacy # ``detail.leg_failures`` union surface, so the per-cap probe # carries the accurate count for ``_summarize_leg_failures`` # to render. legs = _snapshot_partial_failures(provider) sample_count = len(rows) if rows is not None else 0 if sample_count == 0 and legs: # Same classifier the sync layer uses (see # ``cmdb.sync._is_structural_all_legs_auth_failure``) — # 100% auth_error with zero rows means the credential # structurally can't observe this capability. The test # result MUST surface this as failed so the admin sees # the gap that ``test_connection: ok`` previously hid. return { "status": "failed", "reason": ( f"0 rows enumerated · {len(legs)} leg(s) failed: " f"{_summarize_leg_failures(legs)}" ), "sample_count": 0, "leg_count": len(legs), "partial_failures": legs, } if legs: return { "status": "partial_failure", "sample_count": sample_count, "leg_count": len(legs), "partial_failures": legs, "reason": _summarize_leg_failures(legs), } return {"status": "ok", "sample_count": sample_count} def _aggregate_test_result( self, probes: dict[str, dict[str, Any]], ) -> IntegrationTestResult: """Merge per-capability probes into the legacy test result shape. Top-level ``ok`` is True iff every probed capability ended in ``ok``, ``partial_failure``, or ``skipped`` — a single ``failed`` capability (auth error, structural all-leg auth failure, timeout) flips the whole integration to red so the admin UI surfaces what previously hid behind ``OK · 692 servers``. The legacy ``leg_failures`` / ``sample_server_count`` keys stay populated so callers that haven't been updated to consume ``capability_probes`` keep working. """ failed_caps = [c for c, p in probes.items() if p["status"] == "failed"] ok = not failed_caps servers_probe = probes.get("servers") or {} servers_count = int(servers_probe.get("sample_count") or 0) # Compose a one-line summary the legacy ``last_test_message`` # field can hold. Each cap contributes a token; ``servers`` # leads because it's the universal probe. parts: list[str] = [] ordered_caps = ["servers"] + [ c for c in probes.keys() if c != "servers" ] for cap in ordered_caps: entry = probes.get(cap) if not entry: continue status = entry.get("status") if status == "ok": parts.append( f"{cap}=ok({entry.get('sample_count', 0)})" ) elif status == "partial_failure": parts.append( f"{cap}=partial({entry.get('sample_count', 0)};" f"{entry.get('leg_count', 0)} legs failed)" ) elif status == "skipped": reason = entry.get("reason") or "skipped" parts.append(f"{cap}=skipped({reason})") else: # Embed the (already-redacted) reason in the # one-line message so the failure-notifier dedup key # and the integrations table's ``last_test_message`` # column carry the actionable detail, not a generic # ``servers=failed`` token. The reason was either # written by ``_probe_capability`` (which threads # exceptions through ``redact_provider_error``) or # composed from the leg summary; both surfaces are # credential-safe by construction. reason = entry.get("reason") or "failed" parts.append(f"{cap}=failed({reason})") message_head = "OK · " if ok else "FAIL · " message = message_head + " · ".join(parts) # Truncate at 1000 chars to fit ``last_test_message``'s column. if len(message) > 1000: message = message[:997] + "..." # Cap legs to detail max symmetrically with the prior code path. # ``leg_failures`` on the top-level detail is the union of every # cap's partial_failures, capped — the per-cap breakdown lives # inside ``capability_probes`` for fine-grained inspection. all_legs: list[dict[str, Any]] = [] for entry in probes.values(): for leg in entry.get("partial_failures") or []: all_legs.append(leg) if len(all_legs) > _LEG_DETAIL_MAX: top_legs = all_legs[:_LEG_DETAIL_MAX] + [{ "leg": "_truncated_", "reason": ( f"{len(all_legs) - _LEG_DETAIL_MAX} additional " f"leg(s) omitted from this payload (cap " f"{_LEG_DETAIL_MAX}); full breakdown available " f"in cmdb_sync_jobs.capability_results" ), }] else: top_legs = all_legs return IntegrationTestResult( ok=ok, message=message, detail={ # Legacy keys preserved for callers / tests that # haven't been updated. "sample_server_count": servers_count, "leg_failures": top_legs, # New: structured per-capability breakdown the admin # UI renders as a table. "capability_probes": probes, }, ) async def list_servers(self) -> list[dict[str, Any]]: provider = self._provider() try: results = await provider.list_servers() return [dataclasses.asdict(s) for s in results] finally: await provider.close() async def list_domains(self) -> list[dict[str, Any]]: provider = self._provider() try: results = await provider.list_domains() return [dataclasses.asdict(d) for d in results] finally: await provider.close() async def list_dns_records(self) -> list[dict[str, Any]]: provider = self._provider() try: results = await provider.list_dns_records() return [dataclasses.asdict(r) for r in results] finally: await provider.close()