"""GCP provider — Compute Engine instances, Cloud DNS, Cloud Storage. One provider instance per GCP organization. All API calls use the Service Account JSON credentials passed via ``ProviderCredentials.fields["service_account_json"]``. Projects are discovered recursively below the configured organization via parent-bounded ``projects.list`` and ``folders.list`` calls. This makes the organization a verified tenant boundary; a service account without complete organization/folder list authority fails closed instead of silently returning its credential-wide visible project set. Capabilities: * SERVERS — compute.instances.aggregatedList per project * DNS_RECORDS — dns.managedZones.list + rrsets.list per project * OBJECT_STORAGE — storage.buckets.list per project GCP doesn't have a "registered domains" concept comparable to AWS Route 53 Registered Domains / Aliyun domain registration, so DOMAINS is intentionally absent — Cloud Domains exists but is a separate paid product with different semantics, and most orgs don't use it. Likewise CDN and load balancers are skipped in this first pass because GCP spans multiple LB types (global/regional/classic) and Cloud CDN is attached to LBs — enumerating them usefully is a larger design conversation. """ from __future__ import annotations import asyncio import json import re from collections import deque from concurrent.futures import ThreadPoolExecutor from datetime import UTC, date, datetime, timedelta from typing import Any try: import google_auth_httplib2 import httplib2 from google.oauth2 import service_account from googleapiclient.discovery import build from googleapiclient.errors import HttpError _GOOGLE_AVAILABLE = True except ImportError: # pragma: no cover — deps declared in pyproject google_auth_httplib2 = None # type: ignore[assignment] httplib2 = None # type: ignore[assignment] service_account = None # type: ignore[assignment] build = None # type: ignore[assignment] HttpError = Exception # type: ignore[assignment, misc] _GOOGLE_AVAILABLE = False # Pre-launch hardening 2026-05-16: httplib2 (the transport # google-api-python-client uses by default) has NO socket timeout. # A slow GCP region or a hung TCP connection would otherwise pin an # executor thread forever — sync.py's asyncio task cancellation only # cancels the future, not the underlying socket. Every service built # via ``_svc`` gets a freshly authorized Http with this hard cap. _GCP_HTTP_TIMEOUT_S = 30 # 5xx + transport retry budget for googleapiclient ``.execute()`` calls. # googleapiclient's ``HttpRequest.execute(num_retries=N)`` retries on # HTTP 5xx, SSLError, socket.timeout, HttpLib2Error, ServerNotFoundError # with built-in exponential backoff + jitter. Defaults to 0 retries (no # retry); set to 3 here so a transient 5xx mid-pagination (e.g. compute # aggregatedList page 3 of 12) doesn't drop the rest of that project's # instances + flip the cron red. Round-30 parity (2026-05-24) with # Azure ``_arm_request`` and Aliyun ``_call`` 5xx-retry semantics. _GCP_EXECUTE_NUM_RETRIES = 3 _MAX_BILLED_PROJECT_PAGES = 500 _MAX_ORG_HIERARCHY_PAGES = 10_000 _MAX_ORG_HIERARCHY_NODES = 20_000 from app.core.safe_logging import get_safe_logger from app.domains.cmdb.providers.base import ( CloudDnsRecord, CloudProvider, CloudResourceInventoryItem, CloudServer, ProviderCapability, ProviderCredentials, ServerMetricPoint, ServerMetricTarget, classify_resource_category, register_provider, ) from app.domains.cmdb.providers.exceptions import ( CapabilityNotSupported, ProviderAuthError, ProviderTransientError, redact_provider_error, ) logger = get_safe_logger(__name__) # Read scopes cover everything the CMDB capabilities need. # # Defense-in-depth: pre-launch audit P0 flagged the prior bare # ``cloud-platform`` (full read/write across every GCP API) as overly # broad given the credentials are only ever used for inventory + # billing reads. Swap it for ``cloud-platform.read-only`` (covers # every-API-read at the SA-token boundary) plus per-service narrow # scopes already enumerated below. # # BigQuery scope MUST be ``bigquery`` (not ``bigquery.readonly``): the # billing-export queries go through ``client.query()`` which calls # ``Jobs.InsertJob`` — a Job-creation operation that ``bigquery.readonly`` # does not permit. Per memory ``reference_aws_cur_in_bq``, the AWS CUR # driver hit this exact 403 in prod and the OAuth scope must include # write-to-jobs. Effective read-only is enforced at the IAM-role layer # (``roles/bigquery.dataViewer`` on the dataset) — the SA can create # jobs but not write tables. The narrow OAuth scope is the wrong place # to enforce that. # # Cloud Billing scope MUST be added explicitly: ``cloud-platform. # read-only`` does NOT grant access to the Cloud Billing API # (``cloudbilling.googleapis.com``). Per the API reference, the # accepted scopes for ``billingAccounts.projects.list`` are # ``cloud-platform`` / ``cloud-billing`` / ``cloud-billing.readonly`` # only — the read-only variant of cloud-platform is NOT in that list. # PR #397 added ``_fetch_billed_project_ids`` (org-root scope-sync # gate) which calls cloudbilling on every ``_list_projects`` — and # every resource_sync / scope_sync tick 403'd with "Request had # insufficient authentication scopes" until this was added. Same bug # class as the bigquery scope above: new API surface → must extend # the OAuth scope list explicitly. The integration driver in # ``app/integrations/drivers/gcp_cloud.py:_BASE_SCOPES`` already has # ``cloud-billing.readonly`` — these two scope lists must stay # congruent on the cloudbilling axis. _READ_SCOPES = [ "https://www.googleapis.com/auth/cloud-platform.read-only", "https://www.googleapis.com/auth/cloud-billing.readonly", "https://www.googleapis.com/auth/compute.readonly", "https://www.googleapis.com/auth/ndev.clouddns.readonly", "https://www.googleapis.com/auth/devstorage.read_only", "https://www.googleapis.com/auth/bigquery", # Cloud Monitoring read — instance CPU / network time series for the # efficiency engine. ``cloud-platform.read-only`` above already implies # monitoring read, but declaring it explicitly documents the surface # and survives a future tightening of the broad scope. "https://www.googleapis.com/auth/monitoring.read", ] # GCP auto-creates "system projects" with id prefix ``sys-`` to back # Apigee / AppEngine / Firebase / Workspace and a handful of other # managed services. They surface in ``cloudresourcemanager.projects. # search`` but their IAM, Cloud Storage and most other APIs refuse # external callers by design — even a project Owner cannot get # ``resourcemanager.projects.getIamPolicy`` or ``storage.buckets.list`` # on them (Google manages the policies internally). Treating them as # regular projects produced a daily noise floor of 403s on # ``cmdb.grants_sync`` / ``cmdb.resource_sync object_storage`` / # ``cmdb.scopes_sync`` — production incident 2026-05-16 surfaced this # as "GCP grants sync partial_failure on every tick". Filter them # everywhere the GCP driver/provider enumerates projects. GCP_SYSTEM_PROJECT_PREFIX = "sys-" def is_gcp_system_project(project_id: str | None) -> bool: """Return True for GCP system projects (``sys-XXXX``). System projects back managed services (Apigee, AppEngine, Firebase, Workspace) and their IAM / Storage / BigQuery surfaces are walled off from external callers by Google. Calls return 403 regardless of the integration SA's bindings. Skip them everywhere they would otherwise be enumerated. Regular project ids end in a 4-char hash suffix (e.g. ``encoded-net-496415-d6``) so they cannot collide with this prefix by accident. """ if not project_id: return False return str(project_id).startswith(GCP_SYSTEM_PROJECT_PREFIX) # Employee-owned sandbox projects: GCP auto-creates these when an # employee enables Gemini API / Google AI Studio or first opens the # Cloud Console. They surface in organization project listings # node alongside legitimate business projects, but they are 100% # personal — never owned by IT, never billed centrally, never wanted # in the platform's "云订阅" list. Production incident 2026-05-16: # one GCP sync added 1028 such projects to ``cmdb_cloud_scopes``, # ballooning the visible scope count from 58 → 1086. SQL soft-deleted # 1018 rows post-hoc; this filter is the durable defensive fix so the # flood cannot reappear on the next sync tick. # # Three signals are highly specific to personal sandboxes; matching # any one short-circuits the project from every enumeration path: # 1. ``projectId`` prefix ``gen-lang-client-`` — Google AI Studio # auto-create pattern (179 projects in the 2026-05-16 flood). # 2. ``displayName == "My First Project"`` — Cloud Console default # name on a brand-new GCP account (153 projects). # 3. ``displayName`` starts with ``"Untitled project"`` — alternate # Console default name in some regions / UI flows. # # We intentionally do NOT filter on ``parent.startswith("organizations/")``: # legitimate business projects CAN live at the org root (e.g. shared # infra projects), so a parent-only signal would risk dropping them. # Org-root projects ARE additionally gated by the billing-account # check (see ``fetch_gcp_billed_project_ids`` below) — that's the # durable signal for "this is a corp-paid project, not a personal # sandbox". The displayName / projectId signals here remain as # defense-in-depth (catches projects that haven't been linked to a # billing account yet, including pre-launch sandboxes). _PERSONAL_PROJECT_ID_PREFIXES = ("gen-lang-client-",) _PERSONAL_PROJECT_DISPLAY_NAMES = ("My First Project",) _PERSONAL_PROJECT_DISPLAY_NAME_PREFIXES = ("Untitled project",) def is_gcp_personal_sandbox_project(project: dict | None) -> bool: """Return True for an employee's personal / sandbox GCP project. See the ``_PERSONAL_PROJECT_*`` constants in this module for the full rationale and the 2026-05-16 incident reference. ``project`` is the raw dict returned by Cloud Resource Manager project listing; we look at ``projectId`` AND ``displayName`` because the three signals are spread across both fields. Missing fields → False (be permissive — if Google ever changes the response shape we'd rather keep a legit project than silently lose it). """ if not project or not isinstance(project, dict): return False project_id = project.get("projectId") if project_id and isinstance(project_id, str): for prefix in _PERSONAL_PROJECT_ID_PREFIXES: if project_id.startswith(prefix): return True display_name = project.get("displayName") if display_name and isinstance(display_name, str): if display_name in _PERSONAL_PROJECT_DISPLAY_NAMES: return True for prefix in _PERSONAL_PROJECT_DISPLAY_NAME_PREFIXES: if display_name.startswith(prefix): return True return False # ── Billing-account gate (2026-05-19) ───────────────────────────── # # The displayName / projectId pattern filter above caught ~3 named # families of personal sandboxes, but a 2026-05-16 post-mortem found # ~111 more org-root projects with NO regular naming pattern (random # employee names, hackathon codenames, "test" / "demo" / personal # brand handles). PR #389 acknowledged those as "manual decommission # only" — unacceptable, since employees keep creating new ones and # the IT Admin can't keep up. # # The durable signal for "this is a corp-paid project" is **billing # account membership**. PingCAP's main billing account # (``billingAccounts/01D088-8F9CF2-8AF1C6``) is the source of truth # for what's centrally paid for. Personal sandboxes are either on # free trial billing, on individual employees' own billing accounts, # or have billing disabled entirely — none of which match the corp # account. # # Filter logic (applied in ``_list_active_projects_sync`` and the # driver's ``_list_projects_sync``): # # 1. Projects whose ``parent`` starts with ``folders/*`` are # fast-path-accepted. The folder hierarchy IS the corp org # structure, so a folder-hosted project is by construction # corp-managed. (Empirically in PingCAP prod 2026-05-19: 67/67 # folder-hosted projects are legitimate; zero false positives.) # 2. Projects whose ``parent`` starts with ``organizations/*`` MUST # additionally be in the billing-account allowlist (the set # returned by ``fetch_gcp_billed_project_ids``). Otherwise: skip. # # Failure modes — callers MUST raise on billing API failure, never # fall back to "allow all": that would silently re-introduce the # 1028-project flood the moment PingCAP's SA loses billing perms. # Per memory ``handler_status_must_raise``. def normalize_gcp_billing_account_name(raw: str | None) -> str | None: """Normalize a billing-account config string to the API form. Accepts: * ``"billingAccounts/01D088-8F9CF2-8AF1C6"`` → unchanged * ``"01D088-8F9CF2-8AF1C6"`` → ``"billingAccounts/01D088-8F9CF2-8AF1C6"`` * empty / None → None (caller decides whether absence is OK) """ if not raw: return None s = str(raw).strip() if not s: return None if s.startswith("billingAccounts/"): return s return f"billingAccounts/{s}" def list_gcp_organization_projects( crm: Any, organization_id: str, ) -> list[dict[str, Any]]: """Return the complete project set below one configured organization. ``projects.search`` is credential-wide: Google returns every project on which the caller has ``resourcemanager.projects.get``. It therefore cannot prove tenant ownership when a service account is shared or has a stale cross-organization grant. The authoritative path recursively uses strongly-consistent ``folders.list(parent=...)`` and ``projects.list(parent=...)`` calls, both of which are scoped to direct children of the supplied parent. Missing list permission, malformed identities, repeated tokens, duplicate resources, or traversal caps all raise. Callers must treat that as an incomplete snapshot and must not sweep persisted scopes/resources. """ org_id = str(organization_id or "").strip() if not org_id or not org_id.isdigit(): raise RuntimeError( "GCP organization_id must be a numeric organization identifier" ) parents: deque[str] = deque([f"organizations/{org_id}"]) seen_parents: set[str] = set() seen_project_ids: set[str] = set() projects: list[dict[str, Any]] = [] pages = 0 def _next_page( *, collection: str, parent: str, page_token: str | None, seen_tokens: set[str], ) -> tuple[dict[str, Any], str | None]: nonlocal pages if pages >= _MAX_ORG_HIERARCHY_PAGES: raise RuntimeError( "GCP organization hierarchy page cap " f"({_MAX_ORG_HIERARCHY_PAGES}) reached; snapshot is incomplete" ) if collection == "projects": request = crm.projects().list( parent=parent, pageToken=page_token, showDeleted=False, ) else: request = crm.folders().list( parent=parent, pageToken=page_token, showDeleted=False, ) response = request.execute(num_retries=_GCP_EXECUTE_NUM_RETRIES) pages += 1 if not isinstance(response, dict): raise RuntimeError( f"GCP {collection}.list returned a non-object response; " "snapshot is incomplete" ) next_token_raw = response.get("nextPageToken") next_token = str(next_token_raw).strip() if next_token_raw else None if next_token: if next_token in seen_tokens: raise RuntimeError( f"GCP {collection}.list repeated nextPageToken under " f"{parent}; snapshot is incomplete" ) seen_tokens.add(next_token) return response, next_token while parents: parent = parents.popleft() if parent in seen_parents: raise RuntimeError( f"GCP folder hierarchy repeated parent {parent}; " "snapshot is incomplete" ) seen_parents.add(parent) if len(seen_parents) > _MAX_ORG_HIERARCHY_NODES: raise RuntimeError( "GCP organization hierarchy node cap " f"({_MAX_ORG_HIERARCHY_NODES}) reached; snapshot is incomplete" ) project_token: str | None = None seen_project_tokens: set[str] = set() while True: response, project_token = _next_page( collection="projects", parent=parent, page_token=project_token, seen_tokens=seen_project_tokens, ) entries = response.get("projects") or [] if not isinstance(entries, list): raise RuntimeError( "GCP projects.list returned a non-list projects field; " "snapshot is incomplete" ) for project in entries: if not isinstance(project, dict): raise RuntimeError( "GCP projects.list returned a non-object project; " "snapshot is incomplete" ) project_id = str(project.get("projectId") or "").strip() if not project_id: raise RuntimeError( "GCP projects.list returned a project without projectId; " "snapshot is incomplete" ) if str(project.get("parent") or "").strip() != parent: raise RuntimeError( f"GCP projects.list returned project {project_id!r} " f"outside requested parent {parent}; snapshot is incomplete" ) if project_id in seen_project_ids: raise RuntimeError( f"GCP projects.list repeated projectId {project_id!r}; " "snapshot is incomplete" ) seen_project_ids.add(project_id) projects.append(project) if project_token is None: break folder_token: str | None = None seen_folder_tokens: set[str] = set() while True: response, folder_token = _next_page( collection="folders", parent=parent, page_token=folder_token, seen_tokens=seen_folder_tokens, ) entries = response.get("folders") or [] if not isinstance(entries, list): raise RuntimeError( "GCP folders.list returned a non-list folders field; " "snapshot is incomplete" ) for folder in entries: if not isinstance(folder, dict): raise RuntimeError( "GCP folders.list returned a non-object folder; " "snapshot is incomplete" ) folder_name = str(folder.get("name") or "").strip() if not re.fullmatch(r"folders/[0-9]+", folder_name): raise RuntimeError( "GCP folders.list returned a folder without a canonical " "name; snapshot is incomplete" ) if str(folder.get("parent") or "").strip() != parent: raise RuntimeError( f"GCP folders.list returned {folder_name!r} outside " f"requested parent {parent}; snapshot is incomplete" ) if folder_name in seen_parents or folder_name in parents: raise RuntimeError( f"GCP folders.list repeated folder {folder_name!r}; " "snapshot is incomplete" ) parents.append(folder_name) if folder_token is None: break return projects def fetch_gcp_billed_project_ids( cloudbilling_svc: Any, billing_account_name: str, ) -> set[str]: """Return projectIds linked to ``billing_account_name`` with billing ON. Paginates ``cloudbilling.billingAccounts.projects.list`` and returns only projects where ``billingEnabled=True`` — a disabled link means the project was once on this billing account but has been disassociated (defunct projects, deprovisioned sandboxes) and should NOT pass the gate. Required IAM: ``roles/billing.viewer`` (or any role with ``billing.resourceAssociations.list``) on the billing account. Raises on API error — callers must propagate to ensure the sync job fails loud rather than falling back to "allow all". A silent fallback would defeat the entire filter on the first perm hiccup. """ out: set[str] = set() page_token: str | None = None seen_page_tokens: set[str] = set() pages = 0 while True: if pages >= _MAX_BILLED_PROJECT_PAGES: raise RuntimeError( "GCP billingAccounts.projects.list page cap " f"({_MAX_BILLED_PROJECT_PAGES}) reached after collecting " f"{len(out)} project(s); allowlist is incomplete" ) req = cloudbilling_svc.billingAccounts().projects().list( name=billing_account_name, pageToken=page_token, ) resp = req.execute(num_retries=_GCP_EXECUTE_NUM_RETRIES) for entry in resp.get("projectBillingInfo") or []: if not isinstance(entry, dict): raise RuntimeError( "GCP billingAccounts.projects.list returned a non-object " "entry; allowlist is incomplete" ) if not entry.get("billingEnabled"): continue pid = entry.get("projectId") if not str(pid or "").strip(): raise RuntimeError( "GCP billingAccounts.projects.list returned a billing-enabled " "entry without projectId; allowlist is incomplete" ) out.add(str(pid).strip()) next_page_token = resp.get("nextPageToken") pages += 1 if not next_page_token: break if next_page_token in seen_page_tokens: raise RuntimeError( "GCP billingAccounts.projects.list repeated nextPageToken; " "allowlist is incomplete" ) seen_page_tokens.add(next_page_token) page_token = next_page_token return out def is_gcp_org_root_unbilled( project: dict | None, billed_project_ids: set[str] | None, ) -> bool: """Return True if ``project`` is org-root AND not in the billed set. ``billed_project_ids=None`` disables the gate (caller had no billing account configured; we don't second-guess that — useful for orgs that haven't yet adopted central billing). When the set is provided (even if empty), the gate is active: org-root projects not in the set are rejected. Folder-hosted projects are NOT gated by this — they're already inside the corp folder hierarchy, which IS the corp structure. """ if billed_project_ids is None: return False if not project or not isinstance(project, dict): return False parent = project.get("parent") if not isinstance(parent, str) or not parent.startswith("organizations/"): return False project_id = project.get("projectId") if not isinstance(project_id, str) or not project_id: return False return project_id not in billed_project_ids # How many projects we fan out to concurrently when crawling per-project # APIs. GCP's per-minute quota on aggregatedList is generous (thousands), # but we cap concurrency to keep the thread pool bounded. _MAX_CONCURRENT_PROJECTS = 10 def _gcp_point_value(pt: dict[str, Any]) -> float | None: """Extract the scalar value from a Cloud Monitoring point (double or int64; int64 arrives as a string).""" value = pt.get("value") or {} if "doubleValue" in value and value["doubleValue"] is not None: try: return float(value["doubleValue"]) except (TypeError, ValueError): return None if "int64Value" in value and value["int64Value"] is not None: try: return float(value["int64Value"]) except (TypeError, ValueError): return None return None def _parse_rfc3339(value: str) -> datetime | None: """Parse a Monitoring RFC3339 timestamp. GCP may emit up to 9 fractional digits (nanoseconds); ``datetime.fromisoformat`` only accepts 3 or 6, so truncate the fraction to microseconds.""" s = str(value).strip().replace("Z", "+00:00") m = re.match(r"^(.*\.\d{6})\d*(\+\d{2}:\d{2})$", s) if m: s = m.group(1) + m.group(2) try: return datetime.fromisoformat(s) except ValueError: return None # Zone-level warning codes emitted by ``compute.instances.aggregatedList`` # under ``returnPartialSuccess=true`` that are SAFE to silently ignore. # Every other warning code means the zone partially or fully failed and # instances may be missing from the response. # # Background: ``aggregatedList`` returns a response shaped like # ``items["zones/"] = {"instances": [...], "warning": {...}}``. # When ``returnPartialSuccess=true`` (which we set so a single dead zone # can't 503 the whole call), the API replaces a failing zone's # ``instances`` list with a ``warning`` block carrying a code such as # ``UNREACHABLE`` / ``DEPRECATED_RESOURCE_USED`` / ``ZONE_NOT_AVAILABLE``. # WITHOUT inspecting that block, the driver silently treats the failed # zone as "zero instances" — feeding the safe-sweep guard a false low # count that either triggers the configurable drop-tolerance refusal # (cron stays red — tolerance lives at # ``platform_settings['cmdb_sync.sweep_drop_tolerance_percent']``, # default 20%) or slips under and lets mark-and-sweep wipe live rows. # # Production incident 2026-05-17: GCP servers cron blocked for 8+ hours # because ~60 instances live in zones returning warnings every tick; # ``capability_results.servers.partial_failures = NULL`` despite a # real driver failure — fix surfaces each problem zone as a leg so the # admin UI shows WHICH zone is dropping rows. _BENIGN_ZONE_WARNING_CODES = frozenset({ # Zone has no instances yet — expected & sent on every empty zone. "NO_RESULTS_ON_PAGE", # Variant returned for some operations; same meaning. "NO_RESULTS_FOR_LIST", }) # Cloud-controlled warning ``code`` values are interpolated into the # ``reason`` string written to ``cmdb_sync_jobs.capability_results`` # (JSON column) and surfaced to admins. GCP documents codes as # ``UPPER_SNAKE`` (e.g. ``UNREACHABLE`` / ``ZONE_NOT_AVAILABLE``); any # char outside that class is stripped before interpolation as # defense-in-depth against: # * log injection — a stray ``\n`` in the value would forge a separate # log line via downstream ``logger.warning("...%s...", reason)`` # callsites which truncate but do NOT strip newlines. # * sync_jobs JSON-row hygiene — an unbounded code with embedded # punctuation would balloon the JSON and confuse the admin UI. # Length cap (64 chars) matches the longest legitimately observed code # with comfortable headroom; trailing ``_`` left by the sanitizer is # stripped to keep the reason canonical. _SANITIZE_WARNING_CODE_RE = re.compile(r"[^A-Z0-9_]") @register_provider class GcpProvider(CloudProvider): """GCP CMDB data source.""" @classmethod def provider_type(cls) -> str: return "gcp" @classmethod def supported_capabilities(cls) -> frozenset[ProviderCapability]: return frozenset({ ProviderCapability.SERVERS, ProviderCapability.DNS_RECORDS, ProviderCapability.OBJECT_STORAGE, ProviderCapability.METRICS, }) def __init__(self, credentials: ProviderCredentials) -> None: if not _GOOGLE_AVAILABLE: raise RuntimeError( "google-api-python-client not installed. Add it to pyproject." ) super().__init__(credentials) raw = str(credentials.fields.get("service_account_json") or "").strip() if not raw: raise ProviderAuthError( "GCP service_account_json is not configured on this integration", ) try: info = json.loads(raw) except json.JSONDecodeError as exc: raise ProviderAuthError( f"service_account_json is not valid JSON: {redact_provider_error(exc)}", ) from exc try: self._creds = service_account.Credentials.from_service_account_info( info, scopes=_READ_SCOPES, ) except Exception as exc: # noqa: BLE001 — SA parse errors vary raise ProviderAuthError( f"service_account_json rejected by Google: {redact_provider_error(exc)}", ) from exc self._executor = ThreadPoolExecutor( max_workers=_MAX_CONCURRENT_PROJECTS, thread_name_prefix="gcp-provider", ) self._resource_inventory_enabled = str( credentials.fields.get("resource_inventory_enabled") or "", ).strip().lower() in ("1", "true", "yes", "on") async def close(self) -> None: self._executor.shutdown(wait=False, cancel_futures=True) # ── Auth helpers ────────────────────────────────────────────── def _svc(self, service: str, version: str) -> Any: # Build a fresh authorized Http with a hard socket timeout so a # hung GCP endpoint can't pin this executor thread. httplib2.Http # is not thread-safe across concurrent requests, but each call # site uses its own _svc() result and runs in a single executor # thread — safe. http = google_auth_httplib2.AuthorizedHttp( # type: ignore[union-attr] self._creds, http=httplib2.Http(timeout=_GCP_HTTP_TIMEOUT_S), # type: ignore[union-attr] ) return build( # type: ignore[misc] service, version, http=http, cache_discovery=False, ) def _list_active_projects_sync(self) -> list[str]: """Every project in the configured GCP organization hierarchy. GCP ``sys-*`` system projects (Apigee / AppEngine / Firebase / Workspace) are excluded — their IAM and Storage surfaces are managed internally by Google and return 403 to every external caller, polluting ``cmdb_sync_jobs.capability_results`` with a permanent ``partial_failure`` floor. See ``is_gcp_system_project`` (this module) for the canonical filter. Production incident 2026-05-16 (object_storage capability red on every cron tick from 60-odd sys-* 403s). Employee-owned personal / sandbox projects (Gemini auto-create, Cloud Console default-name) are also excluded — they 100% belong to individual employees, never to centrally-managed infrastructure, and should not surface in CMDB. See ``is_gcp_personal_sandbox_project`` for the patterns matched. Production incident 2026-05-16 (1028 personal projects flooded ``cmdb_cloud_scopes`` on a single sync tick). Additionally (PR 2026-05-19): when the integration has a ``create_scope_billing_account`` configured, org-root projects (``parent`` starts with ``organizations/``) are gated by billing-account membership. Folder-hosted projects always pass — the folder hierarchy IS the corp structure. See ``is_gcp_org_root_unbilled`` for the predicate. This catches the ~111 unfilterable org-root personal sandboxes that the 2026-05-16 name-pattern filter could not. """ organization_id = str( self._credentials.fields.get("organization_id") or "", ).strip() if not organization_id: raise ProviderAuthError( "GCP organization_id is required for tenant-bounded discovery" ) billed_project_ids = self._fetch_billed_project_ids() crm = self._svc("cloudresourcemanager", "v3") out: list[str] = [] try: projects = list_gcp_organization_projects(crm, organization_id) except HttpError as exc: if _is_auth_error(exc): raise ProviderAuthError( "GCP organization hierarchy listing denied: " f"{redact_provider_error(exc)}", ) from exc raise ProviderTransientError( "GCP organization hierarchy listing failed: " f"{redact_provider_error(exc)}", ) from exc except RuntimeError as exc: raise ProviderTransientError(redact_provider_error(exc)) from exc for proj in projects: project_id = str(proj["projectId"]).strip() if is_gcp_system_project(project_id): continue if is_gcp_personal_sandbox_project(proj): continue if is_gcp_org_root_unbilled(proj, billed_project_ids): continue out.append(project_id) return out def _fetch_billed_project_ids(self) -> set[str] | None: """Resolve the billing-account allowlist for this integration. Returns ``None`` when ``create_scope_billing_account`` is not configured — gate is disabled (current behavior, kept for orgs that haven't centralized billing yet). When configured, returns the set of projectIds with billing-enabled links to that billing account. Raises ``ProviderAuthError`` on 401/403 (SA missing ``roles/billing.viewer`` on the billing account) and ``ProviderTransientError`` on other API errors. No silent fallback — see module docstring above ``fetch_gcp_billed_project_ids``. Note on 403: cloudbilling — unlike cloudresourcemanager — has no "API disabled" ambiguity on 403 (the Cloud Billing API is always available on the project that hosts the SA). So a 403 here is unambiguously a missing IAM grant on the billing account itself and routes to ProviderAuthError. This is different from per-project 403 handling elsewhere in this module which uses ``_classify_gcp_403_category``. """ billing_account = normalize_gcp_billing_account_name( self._credentials.fields.get("create_scope_billing_account"), ) if billing_account is None: return None cloudbilling = self._svc("cloudbilling", "v1") try: return fetch_gcp_billed_project_ids(cloudbilling, billing_account) except HttpError as exc: status = getattr(getattr(exc, "resp", None), "status", None) if status in (401, 403): raise ProviderAuthError( f"GCP cloudbilling auth failed on {billing_account!r}: " f"{redact_provider_error(exc)}", ) from exc raise ProviderTransientError( f"GCP cloudbilling list failed on {billing_account!r}: " f"{redact_provider_error(exc)}", ) from exc async def _list_projects(self) -> list[str]: return await asyncio.to_thread(self._list_active_projects_sync) def _list_resource_inventory_project_sync( self, project_id: str, ) -> list[CloudResourceInventoryItem]: cloudasset = self._svc("cloudasset", "v1") out: list[CloudResourceInventoryItem] = [] page_token: str | None = None while True: kwargs: dict[str, Any] = { "scope": f"projects/{project_id}", "pageSize": 500, } if page_token: kwargs["pageToken"] = page_token response = cloudasset.v1().searchAllResources(**kwargs).execute( num_retries=_GCP_EXECUTE_NUM_RETRIES, ) for row in response.get("results") or []: resource_id = str(row.get("name") or "").strip() resource_type = str(row.get("assetType") or "unknown").strip() if not resource_id: continue service = resource_type.split("/", 1)[0].split(".", 1)[0].casefold() category, sensitivity = classify_resource_category(resource_type, service) out.append(CloudResourceInventoryItem( provider_resource_id=resource_id, resource_type=resource_type, service=service, category=category, name=str(row.get("displayName") or resource_id.rsplit("/", 1)[-1]), scope_external_id=project_id, region=str(row.get("location") or "") or None, status=str(row.get("state") or "") or None, tags=row.get("labels") if isinstance(row.get("labels"), dict) else None, metadata={ "parent_full_resource_name": row.get("parentFullResourceName"), "additional_attributes": row.get("additionalAttributes"), "create_time": row.get("createTime"), "update_time": row.get("updateTime"), }, sensitivity=sensitivity, )) page_token = str(response.get("nextPageToken") or "") or None if not page_token: return out async def list_resource_inventory(self) -> list[CloudResourceInventoryItem]: if not self._resource_inventory_enabled: raise CapabilityNotSupported( "GCP Cloud Asset inventory is disabled; enable it after " "cloudasset.assets.searchAllResources is granted", ) return await self._fan_out_per_project( self._list_resource_inventory_project_sync, capability="resource_inventory", ) async def _fan_out_per_project( self, fn: Any, # callable (project_id) -> list[...] *, capability: str | None = None, ) -> list[Any]: """Run ``fn`` for each visible project with bounded concurrency. Exceptions from individual projects are logged and that project's results are skipped — we don't want a single 403 on one project to nuke the whole sync. ``capability`` is stamped onto any ``partial_failures`` leg emitted by the catch-all exception handler so admins reading ``cmdb_sync_jobs.capability_results`` can see WHICH capability the failure belongs to (mirrors the per-branch 403 legs which already set ``capability`` explicitly). Round-1 review finding 2026-05-17 — the catch-all was emitting capability-less legs and downstream UI rendered them as ``(unknown capability)``. """ projects = await self._list_projects() if not projects: return [] sem = asyncio.Semaphore(_MAX_CONCURRENT_PROJECTS) loop = asyncio.get_running_loop() async def _one(project_id: str) -> list[Any]: async with sem: try: return await loop.run_in_executor( self._executor, fn, project_id, ) except ProviderAuthError: # Auth errors propagate — credentials broken globally. raise except Exception as exc: # noqa: BLE001 redacted = redact_provider_error(exc) logger.warning( "GCP per-project call failed (project=%s capability=%s): %s", project_id, capability or "unknown", redacted, ) # Pre-launch hardening 2026-05-16: surface per-project # failure to ``capability_results['partial_failures']`` # so a sync that 403s on 5 of 100 GCP projects shows # up as a partial-success on the admin sync history # instead of an apparent green sync silently missing # those projects' resources. # # Pre-launch hardening 2026-05-17: preserve the # redacted error message (capped at 200 chars) in # the leg's ``reason`` instead of just the bare # exception class name. Previously a network timeout, # JSON decode failure, and credential expiry all # surfaced as ``"HttpError"`` (or worse, ``"Exception"``) # forcing the admin to grep server logs to figure # out what failed. The redactor strips creds; the # leg is persisted to ``cmdb_sync_jobs.capability_results`` # which is admin-readable, so credential-safe wrapping # is mandatory. leg_entry: dict[str, Any] = { "leg": f"project:{project_id}", "reason": f"{type(exc).__name__}: {redacted}"[:240], } if capability: leg_entry["capability"] = capability self.partial_failures.append(leg_entry) return [] results = await asyncio.gather(*[_one(p) for p in projects]) flat: list[Any] = [] for bundle in results: flat.extend(bundle) return flat # ── Servers (Compute Engine) ───────────────────────────────── def _list_servers_for_project(self, project_id: str) -> list[CloudServer]: compute = self._svc("compute", "v1") out: list[CloudServer] = [] # Per-zone warning codes observed across the pagination loop — # deduped so a problem zone that persists across N pages emits # ONE leg, not N. The pre-fix path called ``_record_zone_warning`` # inline-per-page and a >500-instance project with one bad zone # produced N duplicate legs in ``partial_failures``; the engine's # JSON column would balloon with redundant rows and the # failure-notifier's dedup-by-signature would still fire once # but the admin UI rendered the duplicates. zone_warnings_seen: set[tuple[str, str]] = set() page_token: str | None = None while True: try: resp = compute.instances().aggregatedList( project=project_id, pageToken=page_token, # ``returnPartialSuccess=True`` makes GCP return a # partial response when one or more zones error # internally — the failed zone gets a ``warning`` # block instead of raising. We surface those warnings # explicitly below (``_record_zone_warning``) so a # dead zone doesn't silently drop instances. returnPartialSuccess=True, ).execute(num_retries=_GCP_EXECUTE_NUM_RETRIES) except HttpError as exc: if _is_auth_error(exc): raise ProviderAuthError( f"GCP compute auth failed on project {project_id!r}: {redact_provider_error(exc)}", ) from exc # 403 on a single project means one of: # (a) Compute Engine API is not enabled on the project # (BigQuery / Storage / AppEngine / Firebase-only # projects — common in mature orgs with hundreds # of project IDs), OR # (b) the SA actually lacks ``compute.instances.list`` # on the project (real permission issue). # # ``_classify_gcp_api_disabled`` parses the GCP error body # to distinguish — case (a) routes to ``skipped_legs`` # (capability stays green; surfaced for visibility but # not counted as a failure) while case (b) stays in # ``partial_failures`` (real failure, capability flips to # partial_failure and the failure-notifier pages). # # 404 is structurally similar but rare (project deleted # between organization discovery and instances.aggregatedList); # treat as a real partial failure so the admin can # confirm the project actually went away. if exc.resp.status == 403: classified = _classify_gcp_403_category(exc) if classified is not None: category, detail = classified logger.info( "GCP compute list skipped for project %s " "(%s: %s)", project_id, category, detail, ) self.skipped_legs.append({ "leg": f"project:{project_id}", "capability": "servers", "reason": _gcp_skip_leg_reason(category, detail), }) return out logger.info( "GCP compute list skipped for project %s: %s", project_id, exc, ) self.partial_failures.append(_gcp_403_leg( project_id, "servers", exc, bare_reason="http_403", )) return out if exc.resp.status == 404: logger.info( "GCP compute list skipped for project %s: %s", project_id, exc, ) self.partial_failures.append(_gcp_403_leg( project_id, "servers", exc, bare_reason="http_404", )) return out raise items = resp.get("items", {}) or {} for zone_key, zone_block in items.items(): for inst in zone_block.get("instances") or []: out.append(_server_from_instance(inst, project_id, zone_key)) # Collect zone-level warnings — see set declaration above # for the cross-pagination dedup rationale and # ``_collect_zone_warnings`` for the parsing contract. # We collect (not append) here so per-page dupes collapse; # final ``partial_failures`` append happens once after # the loop. Surfacing each problem zone is critical — # production incident 2026-05-17 (see # ``_BENIGN_ZONE_WARNING_CODES`` docstring): 8+ hours of # red ``servers`` cron with NO leg in ``partial_failures`` # to identify the bad zone, because the pre-fix path # only iterated ``instances`` and never read ``warning``. # Strip ``zones/`` prefix at insert time so the dedup # set is robust to upstream-shape drift — if GCP ever # returns the bare zone name in some response variant, # both forms collapse to the same set entry. zone_name = zone_key.removeprefix("zones/") for code in self._collect_zone_warnings(zone_block): zone_warnings_seen.add((zone_name, code)) page_token = resp.get("nextPageToken") if not page_token: break # Flush the dedup-set into the per-project leg list. Leg shape # mirrors the existing GCP DNS leg at line 467 (``project:X:zone:Y``) # so the failure-notifier signature space stays consistent across # capabilities. ``zone_name`` was already normalized at insert # time (``zones/`` prefix stripped) for shape-drift robustness. for zone_name, code in sorted(zone_warnings_seen): self.partial_failures.append({ "leg": f"project:{project_id}:zone:{zone_name}", "capability": "servers", "reason": f"aggregatedList.zone_warning_{code}", }) return out def _collect_zone_warnings(self, zone_block: dict[str, Any]) -> list[str]: """Return the non-benign warning codes carried by an aggregatedList zone block. GCP's documented shape is ``warning`` (singular, dict with ``code`` + ``message``). Some responses additionally carry ``warnings`` (plural, list of dicts) when multiple sub-errors occur. Handle both defensively — silently dropping a plural-form warning would re-introduce the same blind spot the singular fix was meant to close. Cloud-controlled ``code`` is sanitized to ``[A-Z0-9_]`` + length bounded — defense against log-injection (a stray ``\\n`` would forge an entire log line downstream via the failure-notifier's ``logger.warning(..%s..)`` callsites) and against unbounded reasons polluting the sync_jobs JSON column. GCP documents codes as ``UPPER_SNAKE`` so the strip is a no-op for legitimate values; a pathological value gets sanitized losslessly enough to still identify the rough class while neutralising injection. """ candidates: list[dict[str, Any]] = [] singular = zone_block.get("warning") if isinstance(singular, dict): candidates.append(singular) plural = zone_block.get("warnings") if isinstance(plural, list): for item in plural: if isinstance(item, dict): candidates.append(item) out: list[str] = [] for warn in candidates: raw_code = str(warn.get("code") or "") code = _SANITIZE_WARNING_CODE_RE.sub("", raw_code)[:64].strip("_") if not code or code in _BENIGN_ZONE_WARNING_CODES: continue out.append(code) return out async def list_servers(self) -> list[CloudServer]: return await self._fan_out_per_project( self._list_servers_for_project, capability="servers", ) # ── Utilization metrics (Cloud Monitoring) ──────────────────────── # # Cloud Monitoring ``timeSeries.list`` is per-project. Instead of # sweeping every active project (474 in this org, most GCE-less), we # query only the projects that actually own a target instance — # derived from ``target.project_id`` or parsed out of the synced # instance's zone/selfLink URL. CPU utilization + network byte counts # are agentless; memory needs the Ops agent and is left NULL. # ``cmdb_servers.instance_id`` for GCP is the numeric instance id, # which equals Monitoring's ``resource.labels.instance_id``. _MON_SPECS = ( # (field, metric_type, aligner, is_fraction) ("cpu", "compute.googleapis.com/instance/cpu/utilization", "ALIGN_MEAN", True), ("cpu_max", "compute.googleapis.com/instance/cpu/utilization", "ALIGN_MAX", True), ("net_in", "compute.googleapis.com/instance/network/received_bytes_count", "ALIGN_SUM", False), ("net_out", "compute.googleapis.com/instance/network/sent_bytes_count", "ALIGN_SUM", False), ) @staticmethod def _project_for_target(t: ServerMetricTarget) -> str | None: if t.project_id: return t.project_id raw = t.raw or {} for key in ("zone", "selfLink", "machineType", "networkInterfaces"): m = re.search(r"/projects/([^/]+)/", str(raw.get(key) or "")) if m: return m.group(1) return None async def list_server_metrics( self, targets: list[ServerMetricTarget], *, lookback_days: int = 14, period_seconds: int = 3600, ) -> list[ServerMetricPoint]: by_project: dict[str, set[str]] = {} for t in targets: if not t.instance_id: continue project = self._project_for_target(t) if not project: continue by_project.setdefault(project, set()).add(t.instance_id) if not by_project: return [] end = datetime.now(UTC) start = end - timedelta(days=lookback_days) sem = asyncio.Semaphore(_MAX_CONCURRENT_PROJECTS) loop = asyncio.get_running_loop() async def _one(project: str, want: set[str]) -> list[ServerMetricPoint]: async with sem: try: return await loop.run_in_executor( self._executor, self._list_metrics_for_project_sync, project, want, start, end, period_seconds, ) except ProviderAuthError: raise except Exception as exc: # noqa: BLE001 self.partial_failures.append({ "leg": f"project:{project}", "capability": "metrics", "reason": f"{type(exc).__name__}: {redact_provider_error(exc)}"[:240], }) return [] results = await asyncio.gather( *[_one(p, w) for p, w in by_project.items()], ) flat: list[ServerMetricPoint] = [] for bundle in results: flat.extend(bundle) return flat def _list_metrics_for_project_sync( self, project_id: str, want: set[str], start: datetime, end: datetime, period_seconds: int, ) -> list[ServerMetricPoint]: monitoring = self._svc("monitoring", "v3") start_iso = start.strftime("%Y-%m-%dT%H:%M:%SZ") end_iso = end.strftime("%Y-%m-%dT%H:%M:%SZ") merged: dict[tuple[str, str], dict[str, float]] = {} for field, mtype, aligner, is_fraction in self._MON_SPECS: page_token: str | None = None pages = 0 while True: pages += 1 if pages > 100: # safety ceiling break resp = monitoring.projects().timeSeries().list( name=f"projects/{project_id}", filter=f'metric.type="{mtype}"', interval_startTime=start_iso, interval_endTime=end_iso, aggregation_alignmentPeriod=f"{period_seconds}s", aggregation_perSeriesAligner=aligner, view="FULL", pageToken=page_token, ).execute(num_retries=_GCP_EXECUTE_NUM_RETRIES) for series in resp.get("timeSeries", []): labels = (series.get("resource") or {}).get("labels") or {} inst_id = labels.get("instance_id") if inst_id not in want: continue for pt in series.get("points", []): end_time = (pt.get("interval") or {}).get("endTime") val = _gcp_point_value(pt) if not end_time or val is None: continue bucket = merged.setdefault((inst_id, end_time), {}) bucket[field] = val * 100 if is_fraction else val page_token = resp.get("nextPageToken") if not page_token: break points: list[ServerMetricPoint] = [] for (inst_id, end_time), f in merged.items(): ts = _parse_rfc3339(end_time) if ts is None: continue points.append(ServerMetricPoint( instance_id=inst_id, timestamp=ts, cpu_pct=f.get("cpu"), cpu_pct_max=f.get("cpu_max"), net_in_bytes=f.get("net_in"), net_out_bytes=f.get("net_out"), )) return points # ── Cloud DNS records ──────────────────────────────────────── def _list_dns_records_for_project( self, project_id: str, ) -> list[CloudDnsRecord]: dns = self._svc("dns", "v1") out: list[CloudDnsRecord] = [] managed_zones: list[dict] = [] try: zones_page_token: str | None = None while True: # Paginate managedZones.list — a project with more zones than # one API page would otherwise silently drop the trailing # zones (and all their records), corrupting the CMDB DNS # inventory + downstream cert/orphan logic. Mirrors the # resourceRecordSets pageToken loop below. zones_resp = dns.managedZones().list( project=project_id, pageToken=zones_page_token, ).execute(num_retries=_GCP_EXECUTE_NUM_RETRIES) managed_zones.extend(zones_resp.get("managedZones") or []) zones_page_token = zones_resp.get("nextPageToken") if not zones_page_token: break except HttpError as exc: if _is_auth_error(exc): raise ProviderAuthError( f"GCP dns auth failed on project {project_id!r}: {redact_provider_error(exc)}", ) from exc # Same SERVICE_DISABLED-vs-permission disambiguation as # ``_list_servers_for_project``; see that branch's comment # for the full rationale. Cloud DNS API is similarly opt-in # per project, so an org with many BigQuery-only projects # routinely hits "DNS API not enabled" 403s that should not # turn the sync red. if exc.resp.status == 403: classified = _classify_gcp_403_category(exc) if classified is not None: category, detail = classified logger.info( "GCP DNS managedZones skipped for project %s " "(%s: %s)", project_id, category, detail, ) self.skipped_legs.append({ "leg": f"project:{project_id}", "capability": "dns_records", "reason": _gcp_skip_leg_reason(category, detail), }) return out self.partial_failures.append(_gcp_403_leg( project_id, "dns_records", exc, bare_reason="managedZones.list_http_403", )) return out if exc.resp.status == 404: self.partial_failures.append(_gcp_403_leg( project_id, "dns_records", exc, bare_reason="managedZones.list_http_404", )) return out raise for zone in managed_zones: zone_name = str(zone.get("name")) dns_name = str(zone.get("dnsName") or zone_name) zone_visibility = str(zone.get("visibility") or "public") page_token: str | None = None while True: try: rrsets_resp = dns.resourceRecordSets().list( project=project_id, managedZone=zone_name, pageToken=page_token, ).execute(num_retries=_GCP_EXECUTE_NUM_RETRIES) except HttpError as exc: # rrsets-level 403 with SERVICE_DISABLED is rare in # practice (if the API was disabled, managedZones.list # would have returned 403 first and we never get here) # but still possible if DNS was disabled mid-page or # the zone-level IAM was revoked while we paginated. # Apply the same disambiguation for symmetry and to # keep this branch parallel to the other 3. if exc.resp.status == 403: classified = _classify_gcp_403_category(exc) if classified is not None: category, detail = classified logger.info( "GCP rrsets skipped (project=%s zone=%s) " "— %s: %s", project_id, zone_name, category, detail, ) self.skipped_legs.append({ "leg": f"project:{project_id}:zone:{zone_name}", "capability": "dns_records", "reason": _gcp_skip_leg_reason( category, detail, ), }) break logger.info( "GCP rrsets skipped (project=%s zone=%s): %s", project_id, zone_name, exc, ) self.partial_failures.append(_gcp_403_leg( project_id, "dns_records", exc, bare_reason="rrsets.list_http_403", zone_name=zone_name, )) break if exc.resp.status == 404: logger.info( "GCP rrsets skipped (project=%s zone=%s): %s", project_id, zone_name, exc, ) self.partial_failures.append(_gcp_403_leg( project_id, "dns_records", exc, bare_reason="rrsets.list_http_404", zone_name=zone_name, )) break raise for rr in rrsets_resp.get("rrsets") or []: name = str(rr.get("name") or "") record_type = str(rr.get("type") or "") ttl = int(rr.get("ttl") or 0) rrdatas = rr.get("rrdatas") or [] # GCP returns one rrset containing N rrdata values; # Helix stores one row per value so emit one record # per rrdata, deterministically keyed by zone+name+type+value. for value in rrdatas: out.append(CloudDnsRecord( zone_name=dns_name, record_id=f"{zone_name}::{name}::{record_type}::{value}", name=name, record_type=record_type, value=str(value), ttl=ttl, line=None, status=( "enable" if zone_visibility == "public" else "disable" ), raw={ "project_id": project_id, "zone_name": zone_name, "visibility": zone_visibility, "kind": rr.get("kind"), }, )) page_token = rrsets_resp.get("nextPageToken") if not page_token: break return out async def list_dns_records(self) -> list[CloudDnsRecord]: return await self._fan_out_per_project( self._list_dns_records_for_project, capability="dns_records", ) # ── Cloud Storage buckets ──────────────────────────────────── def _list_buckets_for_project(self, project_id: str) -> list[dict[str, Any]]: storage = self._svc("storage", "v1") out: list[dict[str, Any]] = [] page_token: str | None = None while True: try: resp = storage.buckets().list( project=project_id, pageToken=page_token, ).execute(num_retries=_GCP_EXECUTE_NUM_RETRIES) except HttpError as exc: if _is_auth_error(exc): raise ProviderAuthError( f"GCP storage auth failed on project {project_id!r}: {redact_provider_error(exc)}", ) from exc # Same SERVICE_DISABLED-vs-permission disambiguation as # ``_list_servers_for_project``. Cloud Storage API is # similarly opt-in per project; orgs with many compute- # only or BigQuery-only projects routinely hit 403 here # without it being a permission issue. if exc.resp.status == 403: classified = _classify_gcp_403_category(exc) if classified is not None: category, detail = classified logger.info( "GCP buckets.list skipped for project %s " "(%s: %s)", project_id, category, detail, ) self.skipped_legs.append({ "leg": f"project:{project_id}", "capability": "object_storage", "reason": _gcp_skip_leg_reason(category, detail), }) return out self.partial_failures.append(_gcp_403_leg( project_id, "object_storage", exc, bare_reason="buckets.list_http_403", )) return out if exc.resp.status == 404: self.partial_failures.append(_gcp_403_leg( project_id, "object_storage", exc, bare_reason="buckets.list_http_404", )) return out raise for bucket in resp.get("items") or []: out.append(_bucket_to_dict(bucket, project_id)) page_token = resp.get("nextPageToken") if not page_token: break return out async def list_object_storage(self) -> list[dict[str, Any]]: return await self._fan_out_per_project( self._list_buckets_for_project, capability="object_storage", ) # ── Monthly billing via BigQuery billing export ────────────── # # GCP doesn't expose a "give me last month's total cost" REST API # the way AWS Cost Explorer does. The supported way is to enable # BigQuery billing export on the billing account, then query the # ``gcp_billing_export_v1_`` table. The # integration's ``billing_bq_table`` field carries the fully- # qualified table ID: ``project.dataset.gcp_billing_export_v1_X``. # # Multi-billing-account support: # Orgs with multiple billing accounts get one export table per # billing account, all under the same dataset. ``billing_bq_table`` # accepts a comma-separated list to span all of them in a single # monthly query (UNION ALL). Wildcards in BigQuery would also # work but produce slow / unpredictable scans on real prod data # (verified: a ``gcp_billing_export_v1_*`` wildcard query hung # for 20+ min on a 3-table org); explicit UNION ALL is faster # and more deterministic. # # Schema (well-defined, stable): # * ``cost`` — FLOAT64, native account currency # * ``currency`` — STRING ("USD" / "CNY" / …) # * ``currency_conversion_rate`` — FLOAT64, USD → local rate at # billing time (USD = cost / rate; equals 1.0 for USD-billed # accounts). NB: GCP does NOT export a ``cost_in_usd`` column # — that's AWS CUR. Cost-facts callers compute USD via # ``SAFE_DIVIDE(cost, currency_conversion_rate)``. # * ``service`` — RECORD with ``description`` field # * ``invoice`` — RECORD with ``month`` STRING (YYYYMM) # * ``credits`` — REPEATED RECORD; subtract for net cost # # We sum ``cost + IFNULL((SELECT SUM(amount) FROM # UNNEST(credits)), 0)`` to match the GCP console ``Reports`` # view (which is net of credits/refunds). @staticmethod def _split_bq_tables(value: str) -> list[str]: """Split a comma- or whitespace-separated billing_bq_table config into a clean list of fully-qualified table ids. Accepts: ``project.ds.t1`` ``project.ds.t1,project.ds.t2`` ``project.ds.t1, project.ds.t2 project.ds.t3`` (mixed) Filters empty fragments so a stray trailing comma is harmless. """ out: list[str] = [] for chunk in value.replace(",", " ").split(): s = chunk.strip() if s: out.append(s) return out @staticmethod def _is_detailed_export(table_id: str) -> bool: """Detect whether a billing-export table is the *detailed* flavour. GCP's Cloud Billing export to BigQuery comes in two schemas: * **Standard usage cost** — table named ``gcp_billing_export_v1_``. Carries ``project`` / ``service`` / ``sku`` / ``location`` / ``cost`` / ``currency`` / ``currency_conversion_rate`` / ``usage`` / ``labels`` — but **no** ``resource`` STRUCT and **no** ``cost_in_usd`` column (callers compute USD via ``SAFE_DIVIDE(cost, currency_conversion_rate)``; the ``cost_in_usd`` column belongs to AWS CUR, not GCP exports). * **Detailed usage cost** — table named ``gcp_billing_export_resource_v1_``. Same columns *plus* a ``resource`` STRUCT (``name``, ``global_name``) giving per-resource breakdown. Refs: https://cloud.google.com/billing/docs/how-to/export-data-bigquery-tables/standard-usage https://cloud.google.com/billing/docs/how-to/export-data-bigquery-tables/detailed-usage ``GcpCloudDriver._list_cost_facts_sync`` references ``resource.name`` / ``resource.global_name``. Querying those against a standard export hard-fails with ``400 Unrecognized name: resource at [21:22]`` and the cost- facts sync records the scope as failed (observed against ``gcp-qa-perf-testing-dev`` / ``gcp-serverless-engine-dev`` / ``gcp-pm-preview`` 2026-05-10). Detection by table name is reliable because the canonical export naming is fixed by GCP and not user-customizable on creation. Match strictly on the canonical ``gcp_billing_export_resource_v1_`` prefix. Round-3 review removed an earlier ``_resource_v1_`` substring backstop because it over-matched custom names like ``mycorp_resource_v1_billing`` and would have falsely treated them as detailed — re-introducing the original ``400 Unrecognized name: resource`` failure on those tables. GCP's export-creation flow does not allow custom names, so anchoring on the canonical prefix is sufficient. """ # Last segment after the final '.' is the table name. The # caller has already passed shape validation, so we don't # re-validate here. table_name = table_id.rsplit(".", 1)[-1] return table_name.startswith("gcp_billing_export_resource_v1_") @staticmethod def _validate_bq_table_shapes(tables: list[str]) -> None: """Reject any ``billing_bq_table`` id that doesn't match the ``project.dataset.table`` shape. ``billing_bq_table`` is admin-controlled config; without validation a privileged-admin compromise (or typo) could inject backticks / semicolons / unicode / SQL comments via f-string concatenation in the BigQuery query template. BigQuery table ids follow ``project.dataset.table`` with ``[A-Za-z0-9_-]`` in project and ``[A-Za-z0-9_]`` in dataset and table; refuse anything else. Raises :class:`ValueError` on the first non-conforming id; callers wrap as ``ProviderAuthError`` / ``RuntimeError`` per their surface contract. Shared between :meth:`GcpProvider._get_monthly_billing_sync` (daily billing rollup) and ``GcpCloudDriver._list_cost_facts_sync`` (line-item cost facts) — same SQL-injection surface, same validation. """ import re as _re # noqa: PLC0415 _BQ_TABLE_RE = _re.compile( r"\A[A-Za-z0-9][A-Za-z0-9_\-]*" # project r"\.[A-Za-z0-9_]+" # dataset r"\.[A-Za-z0-9_]+\Z" # table ) for t in tables: if not _BQ_TABLE_RE.fullmatch(t): raise ValueError( f"billing_bq_table {t!r} does not match the " "required ``project.dataset.table`` shape (only " "[A-Za-z0-9_-] in project; [A-Za-z0-9_] in " "dataset / table). Refusing to interpolate " "into SQL.", ) def _get_monthly_billing_sync(self, month: date) -> dict: from google.cloud import bigquery # type: ignore[attr-defined] raw = str( self._credentials.fields.get("billing_bq_table") or "", ).strip() if not raw: raise ProviderAuthError( "GCP billing_bq_table not configured. Set it on the " "integration to a fully-qualified BigQuery table id " "(project.dataset.gcp_billing_export_v1_XXXXXX). " "Multiple billing accounts: comma-separate (e.g. " "``proj.ds.t1,proj.ds.t2``). See " "https://cloud.google.com/billing/docs/how-to/export-data-bigquery-setup", ) tables = self._split_bq_tables(raw) if not tables: raise ProviderAuthError( f"GCP billing_bq_table parsed empty from value {raw!r}", ) # Strict shape validation BEFORE interpolating into SQL. # See ``_validate_bq_table_shapes`` for the rationale; same # check used by the line-item driver path so admin-config # compromise can't escape via either query template. try: self._validate_bq_table_shapes(tables) except ValueError as exc: raise ProviderAuthError(f"GCP {redact_provider_error(exc)}") from exc invoice_month = month.strftime("%Y%m") # ``invoice.month`` is the right pivot — credits/adjustments # are stamped with the invoice month they apply to (vs. # ``usage_start_time`` which is per-row consumption time). # # CRITICAL: GCP billing exports are partitioned on # ``_PARTITIONTIME`` (which tracks usage time, NOT invoice # month). A bare ``WHERE invoice.month = X`` triggers a FULL # TABLE SCAN over all history — verified prod 2026-04-26 a # 3-table UNION ALL hung 20+ min before timing out. Adding a # ``_PARTITIONTIME`` predicate spanning the target month plus # ±45 days (window for invoice.month divergence: late-arriving # usage rolls into following invoice month, and SUD / committed- # use credits can back-date 30-45 days for some service # categories — verified vs. prod backfill 2026-05) drops the # scan from hundreds of GB to a few GB and the query from 20+ # min to ~3-5s. from datetime import timedelta as _td anchor = month.replace(day=1) partition_lo = anchor - _td(days=45) # End of target month + 45 days for late credits. if anchor.month == 12: next_month_first = anchor.replace( year=anchor.year + 1, month=1, day=1, ) else: next_month_first = anchor.replace(month=anchor.month + 1, day=1) partition_hi = next_month_first + _td(days=45) partition_lo_str = partition_lo.isoformat() partition_hi_str = partition_hi.isoformat() # Build one SELECT per table then UNION ALL into the outer # aggregator. Each inner SELECT pre-aggregates (svc, cost, # credit, currency) so the outer query stays small even on # tenants with hundreds of millions of rows per export. per_table_selects = " UNION ALL ".join( f""" SELECT service.description AS svc, cost, IFNULL((SELECT SUM(amount) FROM UNNEST(credits)), 0) AS credit_amount, currency FROM `{t}` WHERE _PARTITIONTIME >= TIMESTAMP('{partition_lo_str}') AND _PARTITIONTIME < TIMESTAMP('{partition_hi_str}') AND invoice.month = @invoice_month """ for t in tables ) # ``currency`` is in the GROUP BY, not ``ANY_VALUE`` (#2177). # The daily and cost-facts twins in ``gcp_cloud.py`` were fixed # this way in the Round-2 audit, for a real configuration: an # org with both a USD-billed and a CNY-billed billing account # exports into these same tables. With ``ANY_VALUE`` the monthly # rollup added CNY and USD as bare numbers and then labelled the # sum with whichever currency the last row happened to carry — # if that was "USD", the read side never converts and the # discount / alert gates all compare against a fabricated total. sql = f""" SELECT svc, currency, SUM(cost) AS cost_sum, SUM(credit_amount) AS credit_sum FROM ( {per_table_selects} ) GROUP BY svc, currency ORDER BY cost_sum DESC """ # Quota project = first table's project. All tables in the # same org's billing exports typically share one project so # this is consistent with what the user owns. If they don't, # the user can rotate by listing the desired project's table # first. bq_project = tables[0].split(".", 1)[0] if "." in tables[0] else None client = bigquery.Client( credentials=self._creds, project=bq_project, ) try: job = client.query( sql, job_config=bigquery.QueryJobConfig( query_parameters=[ bigquery.ScalarQueryParameter( "invoice_month", "STRING", invoice_month, ), ], ), ) rows = list(job.result()) finally: client.close() cost_by_service: dict[str, float] = {} total = 0.0 by_currency: dict[str, float] = {} for r in rows: svc = str(r["svc"] or "Unknown") net = float(r["cost_sum"] or 0) + float(r["credit_sum"] or 0) cur = str(r["currency"] or "USD").upper() by_currency[cur] = by_currency.get(cur, 0.0) + net cost_by_service[svc] = cost_by_service.get(svc, 0.0) + net total += net # A single monthly row cannot honestly describe two currencies: # the return shape is one ``total_cost_usd`` + one ``currency``, # and the read side converts by that one label. Fail loudly # rather than pick a label and hand back a number that is the # sum of two different units. if len(by_currency) > 1: raise RuntimeError( "GCP monthly billing spans multiple currencies " f"({sorted(by_currency)}) for {invoice_month}; the " "monthly rollup returns a single (total, currency) pair " "and cannot represent that. Split the billing accounts " "per currency, or extend this path to return per-" "currency rows the way list_billing_daily does.", ) currency = next(iter(by_currency), "USD") return { "month": month.replace(day=1), "total_cost_usd": total, "cost_by_service": cost_by_service, "currency": currency, "raw_response": { "bq_tables": tables, "invoice_month": invoice_month, "row_count": len(rows), }, } async def get_monthly_billing(self, month: date | None = None) -> dict: target = month or (date.today().replace(day=1) - timedelta(days=1)) # Use ``get_running_loop`` — ``get_event_loop`` is deprecated # on 3.12 and will raise on 3.14 when no current loop is bound # to the calling thread. The other call sites in this file # already use ``get_running_loop`` (line 589); keep it # consistent so a future thread-pool migration doesn't surface # this as a sudden P0. loop = asyncio.get_running_loop() return await loop.run_in_executor( self._executor, self._get_monthly_billing_sync, target, ) # ── Helpers ───────────────────────────────────────────────────── def _is_auth_error(exc: HttpError) -> bool: status = getattr(getattr(exc, "resp", None), "status", None) return status in (401,) # Maximum bytes of the HttpError response body we attempt to parse as # JSON when classifying a 403. GCP error bodies are routinely <2 KB; # anything larger is almost certainly an HTML error page (load-balancer # or auth-proxy interstitial) and we'd just be burning JSON-parse cycles # on garbage. The cap also bounds memory / CPU if some future GCP API # starts streaming a very large body on 403. _GCP_ERROR_BODY_PARSE_MAX_BYTES = 64 * 1024 # Sanitizer for any GCP-side string we interpolate into ``skipped_legs[].reason`` # or ``partial_failures[].reason`` — the leg is persisted to # ``cmdb_sync_jobs.capability_results`` (JSON column) and rendered verbatim # on the admin sync-history UI. Defense-in-depth against a malformed # GCP response carrying control chars / log-injection newlines or an # unbounded value. GCP service names follow ``foo.googleapis.com`` and # constraint ids follow ``constraints/foo.bar``; this character class # covers both. _GCP_SAFE_TOKEN_RE = re.compile(r"[^a-zA-Z0-9._/-]") _GCP_SAFE_TOKEN_MAX = 128 # Cap on the bytes of ``error.message`` we surface in a leg reason. # Long enough for the actionable error code + project id GCP echoes, # short enough that the sync_jobs JSON row stays scannable in the # admin UI without the failure-notifier text overflowing the Feishu # card body. Trailing ``…`` ellipsis attached by ``redact_provider_error`` # already accounts for truncation. _GCP_LEG_MESSAGE_MAX = 160 # Round-31 (2026-05-18): the prod ``project:tidb-376913`` incident # kept paging on bare ``http_403`` for >24h because GCP Compute V1 returned # the LEGACY REST error body (``error.errors[*].reason == "billingNotEnabled"`` # + top-level ``error.message`` beginning with "This API method requires # billing to be enabled") rather than the modern gRPC ErrorInfo body # (``error.details[*].reason == "BILLING_DISABLED"``) the Round-26 # classifier was built against. Two new defenses below — first the # legacy ``errors[]`` reasons get the same precedence-aware treatment # as the modern ``details[]`` markers, then a strict message-pattern # fallback catches body shapes where only the human-readable message # carries the signal (some older GCP services strip both ``errors`` and # ``details`` on 403). The message pattern is intentionally narrow and # guarded by a real-permission-text exclusion so a legitimate IAM # denial whose message happens to mention "billing" as remediation # text never silently downgrades to a skip. _GCP_BILLING_NOT_ENABLED_RE = re.compile( # Round-31 review MEDIUM: imperative / present-tense phrasings only. # ``enabled billing`` (past participle) appears in IAM-denial # remediation hints ("If you recently enabled billing on project X, # wait a few minutes for IAM propagation"); matching it would route # a real IAM gap to billing_disabled and silence the failure-notifier # — exactly the failure mode this classifier is meant to prevent. # Restricting to ``requires billing to be enabled`` / # ``please enable billing`` (verb-first imperative GCP uses ONLY # when billing is actually off) eliminates that overlap. The # ``billing account ... closed/disabled`` arm uses a bounded # ``[^\n]{0,80}`` instead of greedy ``.*`` for defense-in-depth even # though the input is GCP-controlled (not attacker-controlled). r"(requires billing to be enabled" r"|please enable billing" r"|billing account [^\n]{0,80}(?:closed|disabled))", re.IGNORECASE, ) _GCP_REAL_PERMISSION_RE = re.compile( r"(Permission '?[a-zA-Z0-9.]+'? denied on" r"|caller does not have permission" r"|IAM_PERMISSION_DENIED" r"|does not have access to)", re.IGNORECASE, ) # Round-26 classifier result. Categories ordered by precedence (top # entries win over later ones in the same body — see classifier docstring). # # ``iam_denied`` — real IAM permission gap. NEVER routed to skipped_legs, # even when other "structural" markers also appear; the admin must # fix the IAM binding and silently masking the gap would defeat the # entire purpose of the failure-notifier (cmdb.resource_sync incident # 2026-05-17). # ``api_disabled`` — SERVICE_DISABLED / accessNotConfigured. The # project legitimately doesn't have the underlying API turned on # (BigQuery / Storage / AppEngine / Firebase-only project). Route # to skipped_legs — retrying won't help, it's a project-config fact. # ``vpc_sc`` — VPC Service Controls perimeter blocks this call. Route # to skipped_legs — the IAM binding is correct, the org's perimeter # just doesn't permit cross-perimeter access for this SA. Per # memory ``feedback_gcp_403_audit_log_source_of_truth``: pre-Round-26 # the only way to discriminate was pulling Cloud Audit Logs; the # body now carries enough structured markers to classify in-driver. # ``org_policy`` — ``constraints/iam.disableServiceAccountKeyCreation`` # or similar Org Policy constraint denying the call. Skip; admin # config, won't change cron-to-cron. # ``billing_disabled`` — billing not enabled on project. Skip; admin # must enable billing first. (Compute / Storage / DNS APIs all 403 # on a project with billing disabled even when the API itself is # enabled, with reason BILLING_DISABLED on FAILED_PRECONDITION.) # ``consumer_invalid`` — project state weird (recently deleted, in # delete pending, suspended for legal/abuse). Skip; project lifecycle. # # Return shape: ``(category, detail)`` where detail is the canonical # service name when known (``compute.googleapis.com``), the constraint # id for org policy (``constraints/iam.foo``), or the literal # ``"unknown"`` when the body matched the category but no detail was # attached. Callers route every non-None category to skipped_legs. _GCP_403_SKIP_CATEGORIES = frozenset({ "api_disabled", "vpc_sc", "org_policy", "billing_disabled", "consumer_invalid", }) def _gcp_error_body(exc: HttpError) -> dict | None: """Parse the JSON body from a googleapiclient ``HttpError``. Returns the top-level ``error`` dict, or None if absent / malformed / oversized. Shared helper for the classifier + the leg-reason enrichment — both need the body and we don't want to parse it twice. """ content = getattr(exc, "content", None) if not content: return None # ``content`` is bytes for googleapiclient HttpError. Guard against # other shapes (some SDK paths set it as str, low-level transport # paths occasionally hand back a ``memoryview`` over a recv buffer) # and against bodies too large to be a structured error. if isinstance(content, (bytes, bytearray, memoryview)): if len(content) > _GCP_ERROR_BODY_PARSE_MAX_BYTES: return None try: text = bytes(content).decode("utf-8", errors="replace") except Exception: # noqa: BLE001 return None elif isinstance(content, str): if len(content) > _GCP_ERROR_BODY_PARSE_MAX_BYTES: return None text = content else: return None try: body = json.loads(text) except (TypeError, ValueError): return None if not isinstance(body, dict): return None error = body.get("error") if not isinstance(error, dict): return None return error def _classify_gcp_api_disabled(exc: HttpError) -> str | None: """Backwards-compat shim — returns the API-disabled service name only. Round-26 broadened the classifier to also recognize VPC-SC / Org Policy / Billing-disabled / Consumer-invalid. Existing test surface + the historical contract ("non-None ⇒ API-disabled") is preserved via this shim that returns the service name only when the new classifier yields category ``api_disabled``; every other "structural skip" category surfaces through :func:`_classify_gcp_403_category`. Direct callers were migrated; tests covering the legacy contract keep using this shim. """ result = _classify_gcp_403_category(exc) if result is None: return None category, detail = result if category == "api_disabled": return detail or "unknown" return None def _classify_gcp_403_category( exc: HttpError, ) -> tuple[str, str] | None: """Classify a GCP 403 into a structured (category, detail) skip bucket, or None if it should stay a real ``partial_failures`` leg. GCP returns HTTP 403 for at least SIX semantically distinct cases: 1. ``IAM_PERMISSION_DENIED`` — real permission gap. The SA legitimately lacks the IAM binding. RETURN None (admin must fix IAM; surfacing as partial_failure pages the failure-notifier). 2. ``SERVICE_DISABLED`` / ``accessNotConfigured`` — the API itself is not enabled on this project. RETURN ``("api_disabled", "")``. (Round-25 baseline.) 3. VPC Service Controls — a perimeter rule blocks the call. Body carries ``details[*].@type = "type.googleapis.com/google.rpc.PreconditionFailure"`` with ``violations[*].type = "VPC_SERVICE_CONTROLS"``, OR an ``ErrorInfo`` with ``reason="SECURITY_POLICY_VIOLATED"`` and ``metadata.violationReason="VPC_SERVICE_CONTROLS"``. RETURN ``("vpc_sc", "")``. Round-26 addition. 4. Org Policy constraint — body carries ``ErrorInfo`` with ``reason="ORG_POLICY_CONSTRAINT_VIOLATED"`` or a PreconditionFailure with constraint-shaped ``type``. RETURN ``("org_policy", "")``. 5. ``BILLING_DISABLED`` — project has billing turned off. Most GCP APIs 403 the same way as if the API were disabled. RETURN ``("billing_disabled", "")``. 6. ``CONSUMER_INVALID`` — project lifecycle weird (delete pending, suspended, legal-hold). RETURN ``("consumer_invalid", "")``. Precedence rule: if IAM_PERMISSION_DENIED appears ANYWHERE in ``details``, the function returns None unconditionally — even when other "structural" markers also appear. Silently masking a real permission gap is the failure mode we will not accept. Safe defaults: any parse failure / unexpected shape returns None (= treat as real 403). We will surface the bare ``http_403`` to the admin with the redacted ``error.message`` snippet attached (via :func:`_gcp_403_leg` + :func:`_gcp_403_error_message`) rather than risk hiding a real auth issue behind an over-eager skip. Round-31 (2026-05-18): three signal layers, checked in order — a. Modern ``error.details[*]`` ErrorInfo (the original Round-26 logic; covers all five categories with rich detail). b. Legacy ``error.errors[*].reason`` (Compute V1 / DNS V1 old body shape; covers api_disabled / billing_disabled / consumer_invalid — the only three categories that have stable legacy reason names). c. Top-level ``error.message`` strict pattern match (last resort; ONLY billing-disabled, gated by a real-permission-text exclusion to prevent silent IAM-denial downgrade). """ error = _gcp_error_body(exc) if error is None: return None # First pass: scan for the IAM kill-switch and collect all # ErrorInfo / PreconditionFailure markers we recognize. We don't # return on first match — we walk the whole list so the IAM denial # in any position dominates. iam_denied = False api_disabled_service: str | None = None vpc_sc_service: str | None = None org_policy_constraint: str | None = None billing_disabled_service: str | None = None consumer_invalid_service: str | None = None def _service_from(md: Any) -> str | None: if not isinstance(md, dict): return None svc = md.get("service") if isinstance(svc, str) and svc: cleaned = _GCP_SAFE_TOKEN_RE.sub("", svc)[:_GCP_SAFE_TOKEN_MAX] return cleaned or None return None def _violation_reason(md: Any) -> str | None: if not isinstance(md, dict): return None v = md.get("violationReason") if isinstance(v, str) and v: return v.upper() return None details = error.get("details") if isinstance(details, list): for det in details: if not isinstance(det, dict): continue det_type = det.get("@type") reason = det.get("reason") # ErrorInfo (google.rpc.ErrorInfo) — primary discriminator. if reason == "IAM_PERMISSION_DENIED": iam_denied = True continue if reason == "SERVICE_DISABLED": svc = _service_from(det.get("metadata")) or "unknown" if api_disabled_service is None: api_disabled_service = svc continue if reason == "VPC_SERVICE_CONTROLS": # Some GCP services short-circuit to ``reason="VPC_SERVICE_CONTROLS"`` # directly with no violationReason. Treat as VPC-SC. svc = _service_from(det.get("metadata")) or "unknown" if vpc_sc_service is None: vpc_sc_service = svc continue if reason == "SECURITY_POLICY_VIOLATED": # Round-26 review P0-1 (2026-05-18): require a POSITIVE # ``metadata.violationReason == "VPC_SERVICE_CONTROLS"`` # match before classifying as vpc_sc. Pre-P0 the branch # accepted "any violationReason that isn't explicitly # non-VPC-SC", which silently swallowed real Cloud Armor # / BeyondCorp policy denials whose bodies carry # ``SECURITY_POLICY_VIOLATED`` with NO violationReason # at all — exactly the "mask a real permission gap" # failure mode the Round-26 fix is meant to prevent. vr = _violation_reason(det.get("metadata")) if vr != "VPC_SERVICE_CONTROLS": # Real Cloud Armor / BeyondCorp / unknown security # policy — stays a real 403 so the admin investigates. continue svc = _service_from(det.get("metadata")) or "unknown" if vpc_sc_service is None: vpc_sc_service = svc continue if reason == "ORG_POLICY_CONSTRAINT_VIOLATED": md = det.get("metadata") if isinstance(det.get("metadata"), dict) else {} constraint = md.get("constraint") if isinstance(md, dict) else None if isinstance(constraint, str) and constraint: cleaned = _GCP_SAFE_TOKEN_RE.sub("", constraint)[:_GCP_SAFE_TOKEN_MAX] if org_policy_constraint is None: org_policy_constraint = cleaned or "unknown" elif org_policy_constraint is None: org_policy_constraint = "unknown" continue if reason == "BILLING_DISABLED": svc = _service_from(det.get("metadata")) or "unknown" if billing_disabled_service is None: billing_disabled_service = svc continue if reason == "CONSUMER_INVALID": svc = _service_from(det.get("metadata")) or "unknown" if consumer_invalid_service is None: consumer_invalid_service = svc continue # PreconditionFailure (google.rpc.PreconditionFailure) — # carries violations[*].type. VPC-SC and Org Policy both # surface here on some APIs (Cloud DNS, IAM Policy) in # addition to the ErrorInfo above. if det_type and "PreconditionFailure" in str(det_type): violations = det.get("violations") if isinstance(violations, list): for vio in violations: if not isinstance(vio, dict): continue vio_type = str(vio.get("type") or "") if vio_type == "VPC_SERVICE_CONTROLS": if vpc_sc_service is None: vpc_sc_service = "unknown" elif vio_type.startswith("constraints/"): cleaned = _GCP_SAFE_TOKEN_RE.sub( "", vio_type, )[:_GCP_SAFE_TOKEN_MAX] if org_policy_constraint is None: org_policy_constraint = cleaned or "unknown" # IAM denial dominates — silently masking it would defeat the whole # purpose of the failure-notifier. Return None so the leg stays in # partial_failures and the admin gets paged to fix the IAM binding. if iam_denied: return None # Precedence among structural skips: api_disabled > billing_disabled # > vpc_sc > org_policy > consumer_invalid. Most-specific-actionable # first: "enable the API" is a clearer admin task than "billing is # off"; VPC-SC and org policy are perimeter-level configurations. if api_disabled_service is not None: return ("api_disabled", api_disabled_service) if billing_disabled_service is not None: return ("billing_disabled", billing_disabled_service) if vpc_sc_service is not None: return ("vpc_sc", vpc_sc_service) if org_policy_constraint is not None: return ("org_policy", org_policy_constraint) if consumer_invalid_service is not None: return ("consumer_invalid", consumer_invalid_service) # Legacy shape: ``error.errors[*].reason``. Older GCP REST APIs # (Compute V1 + DNS V1 on some projects) still return the pre-gRPC # body shape with a flat ``errors`` array instead of ``details``. # Round-31 (2026-05-18) extended this block from the original # ``accessNotConfigured``-only check to mirror the modern classifier: # walk the whole list, accumulate every recognized reason, apply # the same precedence afterwards. Direct ``return`` on first hit # would re-introduce ordering bugs if a future GCP API ever attaches # multiple legacy reasons to the same error. legacy_api_disabled = False legacy_billing_disabled = False legacy_consumer_invalid = False errors_block = error.get("errors") if isinstance(errors_block, list): for err in errors_block: if not isinstance(err, dict): continue err_reason = err.get("reason") if not isinstance(err_reason, str): continue if err_reason == "accessNotConfigured": legacy_api_disabled = True elif err_reason in ("billingNotEnabled", "billingDisabled"): # GCP Compute V1 emits ``billingNotEnabled``; Cloud # Functions / older Datastore endpoints emit # ``billingDisabled``. Treat both as the same category. legacy_billing_disabled = True elif err_reason in ("consumerInvalid", "accountDisabled"): # Legacy synonym for the project being suspended / # in delete pending. Modern shape uses CONSUMER_INVALID. legacy_consumer_invalid = True if legacy_api_disabled: return ("api_disabled", "unknown") if legacy_billing_disabled: return ("billing_disabled", "unknown") if legacy_consumer_invalid: return ("consumer_invalid", "unknown") # Final fallback: top-level ``error.message`` pattern match for the # canonical GCP Compute V1 "billing not enabled" phrasing. Only # reached when neither the modern ``details[]`` nor the legacy # ``errors[]`` shape carried a structured marker — a body GCP can # return when the project is brand new / billing was just removed # and the API hasn't populated the error envelope yet. # # Strict false-positive guard: a real IAM denial whose message # references "enable billing" (as a remediation suggestion) must # NOT silently downgrade to a skip. The exclusion regex requires # the message to NOT contain any of the canonical "real permission # denied" phrasings before we accept the billing classification. # If both patterns match (extremely unlikely but possible), we # return None so the leg stays a real partial_failure — surfacing # a real IAM gap as a skip is the failure mode we will not accept. message = error.get("message") if isinstance(message, str) and message: if _GCP_BILLING_NOT_ENABLED_RE.search(message): if not _GCP_REAL_PERMISSION_RE.search(message): return ("billing_disabled", "unknown") return None def _gcp_skip_leg_reason(category: str, detail: str | None) -> str: """Format a ``skipped_legs[].reason`` string for one of the structural-skip categories returned by :func:`_classify_gcp_403_category`. Shape: ``":"`` where detail is sanitized and capped by the classifier. Empty / missing detail falls back to ``"unknown"`` so the reason is never the bare category name. """ safe_detail = detail if detail else "unknown" return f"{category}:{safe_detail}" def _gcp_403_error_message(exc: HttpError) -> str | None: """Return a credential-safe ``error.message`` snippet from a GCP error body, or None if absent / malformed. Round-26 hardening: the pre-fix ``reason: "http_403"`` leg gave the admin zero information about what GCP actually rejected the request for. Production incident 2026-05-18: ``project:tidb-376913`` was flagging http_403 for >24h with the admin unable to diagnose without SSH-ing the worker logs (the diagnostic friction memory ``feedback_gcp_403_audit_log_source_of_truth`` describes — but now addressed in-driver). The returned snippet lands in a separate ``error_message`` leg field so: * ``reason`` stays a bare token (``http_403``) for group-by-reason in :func:`_summarize_leg_failures` (one ``http_403 × N`` row, not one row per project). * The dedup signature space (``_STRUCTURAL_AUTH_REASON_RE`` in sync.py) and the all-legs-auth-failure degrade keep matching on the bare reason. * Admins see the actionable text directly when reading the ``capability_results.partial_failures`` JSON, and the failure- notifier picks the most-common ``error_message`` as a sample when rendering the summary line. Returns the redacted, length-bounded message, or None if no ``error.message`` could be parsed. """ error = _gcp_error_body(exc) if error is None: return None message = error.get("message") if not isinstance(message, str) or not message.strip(): return None # Redact known credential-shaped substrings BEFORE truncating — # truncation alone wouldn't help if the leaked credential prefix is # within the first 160 chars (which it usually is). redacted = redact_provider_error(message, max_len=_GCP_LEG_MESSAGE_MAX) if not redacted: return None return redacted def _gcp_403_leg( project_id: str, capability: str, exc: HttpError, *, bare_reason: str, zone_name: str | None = None, ) -> dict[str, Any]: """Build a ``partial_failures`` leg dict for a GCP 4xx error. Centralizes the leg shape so every 403 / 404 branch (servers / DNS managedZones / DNS rrsets / GCS buckets) emits the same fields in the same order. The optional ``error_message`` field is only set when GCP's body carried a parseable ``error.message`` — keeps rows tight on the legacy/transport-error 4xx that have no body. """ leg_id = ( f"project:{project_id}:zone:{zone_name}" if zone_name is not None else f"project:{project_id}" ) entry: dict[str, Any] = { "leg": leg_id, "capability": capability, "reason": bare_reason, } message = _gcp_403_error_message(exc) if message: entry["error_message"] = message return entry def _server_from_instance( inst: dict[str, Any], project_id: str, zone_key: str, ) -> CloudServer: # Zone key from aggregatedList looks like ``zones/us-central1-a``. zone = zone_key.split("/", 1)[-1] region = zone.rsplit("-", 1)[0] if "-" in zone else zone # GCP ``TERMINATED`` is its own state — the instance was killed by # GCP infra (preemption, host-error, lifecycle teardown), distinct # from a user-initiated ``STOPPED``. Surfacing both as "stopped" # hides the diagnostic signal admins need ("did GCP kill it or did # someone stop it"). AWS terminated instances are dropped during # sync (they vanish from the API after a while); GCP terminated # instances STAY listed so we keep them with the right semantic. status_map = { "RUNNING": "running", "STOPPED": "stopped", "TERMINATED": "terminated", "SUSPENDED": "stopped", "SUSPENDING": "stopped", "STOPPING": "stopped", "STAGING": "running", "PROVISIONING": "running", "REPAIRING": "running", } raw_status = str(inst.get("status") or "").upper() status = status_map.get(raw_status, raw_status.lower()) public_ips: list[str] = [] private_ips: list[str] = [] for nic in inst.get("networkInterfaces") or []: if nic.get("networkIP"): private_ips.append(str(nic["networkIP"])) for cfg in nic.get("accessConfigs") or []: nat = cfg.get("natIP") if nat: public_ips.append(str(nat)) # machineType URL looks like # ``https://.../zones/us-central1-a/machineTypes/n2-standard-4``. machine_type_url = str(inst.get("machineType") or "") machine_type = machine_type_url.rsplit("/", 1)[-1] # vCPU / memory aren't inline in the instances response — parse from # machine type name as a best-effort (``n2-standard-4`` → 4 vCPU). vcpu = _guess_vcpu_from_machine_type(machine_type) memory_mb = None # requires a second call to machineTypes.get tags_raw = (inst.get("labels") or {}) tags = {str(k): str(v) for k, v in tags_raw.items()} disks = inst.get("disks") or [] image_id = None os_type: str | None = None for disk in disks: if disk.get("boot"): # licenses[] contains URLs like ``.../windows-cloud/licenses/windows-2019`` licenses = [str(lic) for lic in (disk.get("licenses") or [])] if any("windows" in lic.lower() for lic in licenses): os_type = "windows" elif licenses: os_type = "linux" image_id = str(disk.get("source") or "").rsplit("/", 1)[-1] or None break raw = _json_safe(inst) raw["_project_id"] = project_id return CloudServer( instance_id=str(inst.get("id") or inst.get("name") or ""), name=str(inst.get("name") or inst.get("id") or ""), status=status, region=region, zone=zone, vcpu=vcpu, memory_mb=memory_mb, public_ips=public_ips, private_ips=private_ips, image_id=image_id, os_type=os_type, tags=tags, raw=raw, ) def _guess_vcpu_from_machine_type(machine_type: str) -> int | None: """Parse vCPU count from names like ``n2-standard-4``, ``e2-highmem-2``. Returns None for shared-core types (``e2-micro`` / ``f1-micro``) and anything we can't confidently parse — the sync job stores None and the UI shows a placeholder. """ if not machine_type or "-" not in machine_type: return None tail = machine_type.rsplit("-", 1)[-1] try: return int(tail) except ValueError: return None def _bucket_to_dict(bucket: dict[str, Any], project_id: str) -> dict[str, Any]: iam_cfg = bucket.get("iamConfiguration") or {} pap = iam_cfg.get("publicAccessPrevention") or "inherited" ubla = iam_cfg.get("uniformBucketLevelAccess") or {} access_type = ( "public" if pap == "inherited" and not ubla.get("enabled") else "private" ) return { "bucket_name": str(bucket.get("name") or ""), "region": str(bucket.get("location") or "").lower() or None, "storage_class": str(bucket.get("storageClass") or "") or None, "access_type": access_type, "size_bytes": None, # bucket size needs a separate usage metrics call "object_count": None, "status": "active", "owner_raw": (bucket.get("owner") or {}).get("entity"), "raw": _json_safe({**bucket, "_project_id": project_id}), } def _json_safe(obj: Any) -> Any: """Recursively convert Google API responses to JSON-serializable values. Google discovery responses are already plain dict/list/str/int, but some fields are ISO-format datetime strings and a few endpoints return ``datetime`` / ``date`` objects directly. Normalise them. """ if isinstance(obj, datetime): return obj.isoformat() if isinstance(obj, date): return obj.isoformat() if isinstance(obj, dict): return {k: _json_safe(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [_json_safe(v) for v in obj] if isinstance(obj, (bytes, bytearray)): return obj.decode("utf-8", errors="replace") return obj