"""GCP cloud driver — scope lifecycle + IAM grants + CMDB sync bridge. One integration per GCP organization. Credentials are a service account JSON. Project discovery is recursively bounded to the configured organization with strongly-consistent ``projects.list(parent=...)`` and ``folders.list(parent=...)`` calls. The service account therefore needs the corresponding list permissions on the organization and every folder; missing authority fails the snapshot closed instead of falling back to credential-wide ``projects.search``. CMDB resource sync (servers / DNS / buckets) routes through ``app.domains.cmdb.providers.gcp.GcpProvider`` (via ``_CloudDriverBase``), so enabling an ``integration.capabilities=['servers', 'dns_records', 'object_storage']`` automatically schedules per-capability sync jobs. Grant / revoke semantics: * ``scope_type=gcp_project`` * ``external_id`` = GCP project id (not the auto-generated number). * Grants = IAM policy bindings. GCP IAM uses READ-MODIFY-WRITE on the whole project policy; we serialize via the ``etag`` optimistic- concurrency field (409/412 → re-read and retry, up to 3 times). * ``revoke_account`` sweeps every visible project and removes the member from every binding — the "kill switch" for a leaver. Optional features: * ``billing_bq_table`` (config) — enables daily cost rollups. Requires ``google-cloud-bigquery`` to be pip-installed AND the SA to have BigQuery Data Viewer on the dataset. Without this, billing daily returns empty (the cmdb.billing_daily_sync job tolerates it). * ``admin_impersonation_email`` (config) — enables Admin SDK calls (Directory API). Used by ``get_last_login`` and principal email validation. Requires the SA to have domain-wide delegation granted in the Workspace admin console, and the target email must be a super-admin (or have the User Management Admin role). Without it, ``get_last_login`` returns None and principal validation relies on GCP's own "does not exist" error surfaced from setIamPolicy. """ from __future__ import annotations import asyncio import dataclasses import json import re from datetime import 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 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 from app.core.safe_logging import get_safe_logger from app.domains.cmdb.providers.exceptions import redact_provider_error from app.domains.cmdb.providers.gcp import ( GcpProvider, ) from app.domains.cmdb.providers.gcp import ( fetch_gcp_billed_project_ids as _fetch_gcp_billed_project_ids, ) from app.domains.cmdb.providers.gcp import ( is_gcp_org_root_unbilled as _is_gcp_org_root_unbilled, ) from app.domains.cmdb.providers.gcp import ( is_gcp_personal_sandbox_project as _is_gcp_personal_sandbox_project, ) from app.domains.cmdb.providers.gcp import ( is_gcp_system_project as _is_gcp_system_project, ) from app.domains.cmdb.providers.gcp import ( list_gcp_organization_projects as _list_gcp_organization_projects, ) from app.domains.cmdb.providers.gcp import ( normalize_gcp_billing_account_name as _normalize_gcp_billing_account_name, ) from app.integrations.driver_base import ( COST_BUCKET_ADJUSTMENT, COST_BUCKET_COMMITMENT_DISCOUNT, COST_BUCKET_CONTRACT_DISCOUNT, COST_BUCKET_CREDIT, COST_BUCKET_SUPPORT, COST_BUCKET_USAGE, ClearBlockersResult, CloseBlocker, ClosePreflight, CloseScopeResult, CloudAccessContext, CloudAccountRevokeContext, CloudDailyCost, CloudGrantResult, CloudPrincipalNotFoundError, CloudRoleInfo, CloudScopeInfo, CloudScopePartitionStatus, CloudScopeRef, CostFact, DiscoveredGrant, ProvisioningResult, ResourceCreatorInfo, ScopeClosureState, ) from app.integrations.drivers._cloud_shared import _CloudDriverBase from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) logger = get_safe_logger(__name__) _GCP_HTTP_TIMEOUT_S = 30 _GCP_READ_RETRIES = 3 # GCP ``credits[].type`` → 财务瀑布桶。5 月真实账单实测枚举: # PROMOTION(−$31.6 万,零线元凶)/ FEE_UTILIZATION_OFFSET(−$17.7 万, # 花费型承诺抵扣)/ SUBSCRIPTION_BENEFIT / DISCOUNT。全部属于公司层面 # 折扣(无项目自挣的 SUD),研发口径一律不参与——此映射只决定财务 # 瀑布的展示分桶。未知新类型默认 COST_BUCKET_CREDIT(白名单思路: # 进保密桶,绝不进 usage)。 _GCP_CREDIT_TYPE_BUCKETS: dict[str, str] = { "FEE_UTILIZATION_OFFSET": COST_BUCKET_COMMITMENT_DISCOUNT, "COMMITTED_USAGE_DISCOUNT": COST_BUCKET_COMMITMENT_DISCOUNT, "COMMITTED_USAGE_DISCOUNT_DOLLAR_BASE": COST_BUCKET_COMMITMENT_DISCOUNT, "SUSTAINED_USAGE_DISCOUNT": COST_BUCKET_CONTRACT_DISCOUNT, "DISCOUNT": COST_BUCKET_CONTRACT_DISCOUNT, "RESELLER_MARGIN": COST_BUCKET_CONTRACT_DISCOUNT, "PROMOTION": COST_BUCKET_CREDIT, "SUBSCRIPTION_BENEFIT": COST_BUCKET_CREDIT, "FREE_TIER": COST_BUCKET_CREDIT, } # Validates BigQuery table name components (project_id, dataset_id, table_id). # GCP's naming rules: letters, digits, underscores, dots, hyphens; 1-1024 chars. _BQ_SAFE_NAME_RE = re.compile(r'^[a-zA-Z0-9_.\-]{1,1024}$') def _validate_bq_name(value: str, label: str) -> None: """Raise ValueError if ``value`` is not a safe BQ name component.""" if not _BQ_SAFE_NAME_RE.match(value): raise ValueError( f"GCP BigQuery {label} {value!r} contains unsafe characters; " "expected only [a-zA-Z0-9_.-] (1-1024 chars)", ) def _gcp_extract_error_detail(exc: BaseException) -> tuple[str, str]: """Pull ``(message, reason)`` from a googleapiclient ``HttpError``. ``HttpError.content`` carries the JSON body GCP returned. The canonical error shape is ``{"error": {"message": ..., "details": [{"@type": "...", "reason": "...", ...}]}}``. The ``reason`` field on the first detail entry is the symbolic dispatch key we want (e.g. ``LIEN``, ``BILLING_DISABLED``). Returns empty strings when the body is missing, not JSON, or doesn't carry the expected shape — callers degrade to the bare ``str(exc)`` form. """ content = getattr(exc, "content", None) if not content: return "", "" if isinstance(content, bytes): try: content = content.decode("utf-8", errors="replace") except (UnicodeDecodeError, AttributeError): return "", "" if not isinstance(content, str): return "", "" try: payload = json.loads(content) except ValueError: # ``json.JSONDecodeError`` is a subclass of ``ValueError``. return "", "" if not isinstance(payload, dict): return "", "" err = payload.get("error") if not isinstance(err, dict): return "", "" message = str(err.get("message") or "").strip() reason = "" details = err.get("details") if isinstance(details, list): # Prefer the most-specific signal in this order: # 1. ``reason`` on an ErrorInfo entry (canonical GCP enum) # 2. ``violations[].type`` on a PreconditionFailure entry # (carries the symbolic constraint name e.g. ``LIEN``) # 3. ``@type`` (least specific — just the proto type URL) for entry in details: if isinstance(entry, dict) and entry.get("reason"): reason = str(entry["reason"]).strip() break if not reason: for entry in details: if not isinstance(entry, dict): continue violations = entry.get("violations") if isinstance(violations, list): for v in violations: if isinstance(v, dict) and v.get("type"): reason = str(v["type"]).strip() break if reason: break if not reason: for entry in details: if isinstance(entry, dict) and entry.get("@type"): reason = str(entry["@type"]).strip() break return message, reason # Base scopes for IAM / scope lifecycle work. _BASE_SCOPES = [ "https://www.googleapis.com/auth/cloud-platform", "https://www.googleapis.com/auth/cloud-billing.readonly", ] # Extra scope required for the Admin SDK Directory API (last-login + # principal email validation). Only requested when impersonation is # configured — otherwise the extra scope would block SA creation. _ADMIN_SCOPE = "https://www.googleapis.com/auth/admin.directory.user.readonly" # Fallback used if migration 054 hasn't seeded the platform_setting. _GCP_FALLBACK: list[dict[str, Any]] = [ {"role_key": "roles/owner", "display_name": "Owner"}, {"role_key": "roles/editor", "display_name": "Editor"}, {"role_key": "roles/viewer", "display_name": "Viewer"}, {"role_key": "roles/browser", "display_name": "Browser"}, ] # GCP predefined roles that live ONLY at the billing-account or # organization level — they are NOT valid principals for a project IAM # policy, so ``projects/{id}:setIamPolicy`` rejects them with HTTP 400 # "Role roles/billing.viewer is not supported for this resource." # (``roles/billing.projectManager`` is deliberately absent: it *is* # project-grantable.) These cannot be granted through Helix's # project-scope model — billing visibility is a billing-account grant. # We both hide them from the request dropdown (``_list_roles_sync``) # and refuse them at the grant chokepoint (``grant_access``) so a # stale ``platform_settings`` entry or a hand-typed role still fails # fast with an actionable message instead of a raw Google 400. _PROJECT_UNSUPPORTED_ROLES: frozenset[str] = frozenset({ "roles/billing.viewer", "roles/billing.user", "roles/billing.admin", "roles/billing.creator", "roles/billing.costsManager", }) # Org / folder-level predefined roles that also reject on a project IAM # policy (same 400 class as billing.*). Folded into the project # ungrantable set so the governance classifier hides them and the grant # chokepoint can fail fast. ``roles/billing.projectManager`` is NOT here # — it is project-grantable. _GCP_ORG_FOLDER_LEVEL_ROLES: frozenset[str] = frozenset({ "roles/resourcemanager.organizationAdmin", "roles/resourcemanager.organizationViewer", "roles/resourcemanager.folderAdmin", "roles/resourcemanager.folderEditor", "roles/resourcemanager.folderViewer", "roles/resourcemanager.folderIamAdmin", "roles/resourcemanager.folderCreator", "roles/resourcemanager.folderMover", "roles/orgpolicy.policyAdmin", "roles/orgpolicy.policyViewer", }) # Project-level roles that are owner-equivalent or grant full IAM-write # (privilege escalation) — restricted tier. _GCP_RESTRICTED_ROLES: frozenset[str] = frozenset({ "roles/editor", "roles/iam.securityAdmin", "roles/iam.roleAdmin", "roles/iam.serviceAccountKeyAdmin", "roles/resourcemanager.projectIamAdmin", }) # Broad-write / act-as roles that aren't named "admin" — elevated tier. _GCP_ELEVATED_ROLES: frozenset[str] = frozenset({ "roles/iam.serviceAccountUser", "roles/iam.serviceAccountTokenCreator", "roles/iam.workloadIdentityUser", }) # Role catalog for a ``gcp_billing_account`` scope. These bind on the # billing account (cloudbilling billingAccounts:setIamPolicy), NOT a # project. ``billing.viewer`` (read-only console billing visibility) is # the promoted default — it's exactly the "I want to see the bill" # request. ``billing.user``/``costsManager`` are offered but not # promoted. ``billing.admin`` is deliberately absent (admin keyword → # restricted; grant it via the admin path, not self-service). _GCP_BILLING_ACCOUNT_ROLE_CATALOG: list[dict[str, Any]] = [ { "role_key": "roles/billing.viewer", "display_name": "Billing Account Viewer", "description": "查看账单账号 + 发票 + 成本报表(只读)。", "promoted": True, }, { "role_key": "roles/billing.user", "display_name": "Billing Account User", "description": "把项目关联到此账单账号。", "promoted": False, }, { "role_key": "roles/billing.costsManager", "display_name": "Billing Account Costs Manager", "description": "查看/管理预算与成本信息。", "promoted": False, }, ] # GCP system-project filter (``sys-*``) lives in ``providers/gcp`` to # avoid a circular import — the provider module is the lower-level # dependency. Re-exported above via ``is_gcp_system_project``; the # leading-underscore alias preserves the original local name used # throughout this driver's grants / billing paths. @register_driver class GcpCloudDriver(_CloudDriverBase): provider = "gcp" display_name = "Google Cloud Platform" description = ( "GCP Organization 下多项目管理。Service account JSON 需要 " "resourcemanager.projects.get / setIamPolicy,可选 Admin SDK " "域内委派用于登录时间查询与外部账号校验。" ) # Align with what the driver + provider actually implement. supported_capabilities = frozenset({ "servers", "dns_records", "object_storage", "user_revoke", }) capabilities = ["servers", "dns_records", "object_storage", "user_revoke"] account_mode = "sso" # Google IAM primitive role ``roles/owner`` = full project control # incl. IAM. ``roles/resourcemanager.projectOwner`` is the same # role presented under its long name; we include both since either # form may surface in a binding. owner_role_keys = frozenset({ "roles/owner", "roles/resourcemanager.projectOwner", }) # Role governance (see app/integrations/cloud_role_governance.py). restricted_role_keys = _GCP_RESTRICTED_ROLES elevated_role_keys = _GCP_ELEVATED_ROLES @classmethod def ungrantable_role_keys(cls, scope_type: str) -> frozenset[str]: if scope_type == "gcp_project": return _PROJECT_UNSUPPORTED_ROLES | _GCP_ORG_FOLDER_LEVEL_ROLES return frozenset() # Bridge to the CMDB CloudProvider (used by _CloudDriverBase for # list_servers / list_dns_records). provider_cls = GcpProvider config_schema = [ IntegrationFieldInfo( key="organization_id", label="Organization ID", is_secret=False, required=True, placeholder="123456789012", ), IntegrationFieldInfo( key="service_account_json", label="Service Account JSON", is_secret=True, required=True, placeholder='{"type":"service_account",...}', ), IntegrationFieldInfo( key="resource_inventory_enabled", label="启用 Cloud Asset 全资源目录", is_secret=False, required=False, placeholder="false", default="false", ), IntegrationFieldInfo( key="billing_bq_table", label="Billing BigQuery Export Table (optional)", is_secret=False, required=False, placeholder="my-project.billing_dataset.gcp_billing_export_v1_XXX", ), IntegrationFieldInfo( key="admin_impersonation_email", label="Admin Impersonation Email (optional, for last-login tracking)", is_secret=False, required=False, placeholder="admin@yourcompany.com", ), IntegrationFieldInfo( key="billing_scope_mode", label="账单 Scope 权威模式", is_secret=False, required=True, options=[ { "value": "projects_only", "label": "仅项目（不管理账单账号 IAM）", }, { "value": "allowlist", "label": "显式账单账号白名单", }, ], hint=( "必须显式选择；不再把服务账号可见的所有账单账号" "当作本 organization 的租户边界。" ), ), IntegrationFieldInfo( key="billing_account_allowlist", label="账单账号白名单", is_secret=False, required=False, is_textarea=True, visible_when={"billing_scope_mode": "allowlist"}, placeholder=( "billingAccounts/AAAAAA-BBBBBB-CCCCCC\n" "billingAccounts/DDDDDD-EEEEEE-FFFFFF" ), hint=( "每行或逗号分隔；每个账号都会通过 parent 链验证" "必须归属已配置 organization。" ), ), # Parent resource for new projects created via ``create_scope``. # Format: ``organizations/{numeric-org-id}`` or # ``folders/{numeric-folder-id}``. Required for the auto-create # path to work; the SA also needs # ``resourcemanager.projects.create`` on the parent. Without # this field, ``create_scope`` raises a clear error pointing # the admin at the integration settings instead of failing # late inside the Resource Manager API. IntegrationFieldInfo( key="create_scope_parent", label="新建项目的父资源 (organizations/123 或 folders/456)", is_secret=False, required=False, placeholder="organizations/123456789012", ), # Optional billing account to attach to newly-created projects. # Format: ``billingAccounts/XXXXXX-XXXXXX-XXXXXX``. If set, the # driver attaches it post-create via the cloudbilling API; the # SA needs ``billing.resourceAssociations.create`` on this # billing account. Most production GCP orgs require this — a # project without billing can't actually run anything. IntegrationFieldInfo( key="create_scope_billing_account", label="新建项目挂载的计费账号 (可选, billingAccounts/XXX-XXX-XXX)", is_secret=False, required=False, placeholder="billingAccounts/XXXXXX-XXXXXX-XXXXXX", ), ] # ── Auth helpers ────────────────────────────────────────── def _credentials(self, *, with_admin_scope: bool = False) -> Any: if not _GOOGLE_AVAILABLE: raise RuntimeError("google-api-python-client not installed") raw = str(self.config.get("service_account_json") or "").strip() if not raw: raise RuntimeError("service_account_json not configured") info = json.loads(raw) required = ("type", "client_email", "private_key", "project_id") missing = [f for f in required if not info.get(f)] if missing: raise ValueError(f"service_account_json is missing fields: {missing}") scopes = list(_BASE_SCOPES) if with_admin_scope: scopes.append(_ADMIN_SCOPE) creds = service_account.Credentials.from_service_account_info( # type: ignore[union-attr] info, scopes=scopes, ) if with_admin_scope: subject = str(self.config.get("admin_impersonation_email") or "").strip() if not subject: raise RuntimeError( "admin_impersonation_email not configured — required for " "Admin SDK calls", ) creds = creds.with_subject(subject) return creds def _svc(self, service: str, version: str, *, admin: bool = False) -> Any: credentials = self._credentials(with_admin_scope=admin) authorized_http = google_auth_httplib2.AuthorizedHttp( # type: ignore[union-attr] credentials, http=httplib2.Http(timeout=_GCP_HTTP_TIMEOUT_S), # type: ignore[union-attr] ) return build( # type: ignore[misc] service, version, http=authorized_http, cache_discovery=False, ) @staticmethod def _execute_read(request: Any) -> Any: """Execute an idempotent read with bounded SDK transient retries.""" return request.execute(num_retries=_GCP_READ_RETRIES) @staticmethod def _execute_write(request: Any) -> Any: """Execute a mutation once; callers reconcile unknown outcomes.""" return request.execute(num_retries=0) def _has_admin_impersonation(self) -> bool: return bool(str(self.config.get("admin_impersonation_email") or "").strip()) # ── Health probe ────────────────────────────────────────── async def test_connection(self) -> IntegrationTestResult: try: scopes = await self.list_scopes() detail = { "dry_run": True, "discovered_scopes": [ { "scope_type": scope.scope_type, "external_id": scope.external_id, } for scope in sorted( scopes, key=lambda item: (item.scope_type, item.external_id), ) ], "scope_partitions": { scope_type: { "complete": state.complete, "error": state.error, } for scope_type, state in sorted( self.scope_snapshot_partitions().items(), ) }, } incomplete = self._incomplete_scope_snapshot_error() if incomplete is not None: return IntegrationTestResult( ok=False, message=incomplete, detail=detail, ) msg = f"OK · {len(scopes)} scope(s) visible" if self._has_admin_impersonation(): # Try an Admin SDK call to catch DWD misconfig early. try: await asyncio.to_thread(self._probe_admin_sdk_sync) msg += " · Admin SDK impersonation OK" except Exception as exc: # noqa: BLE001 error = redact_provider_error(exc, max_len=240) msg += f" · Admin SDK impersonation FAILED: {error}" return IntegrationTestResult( ok=False, message=msg, detail=detail, ) return IntegrationTestResult(ok=True, message=msg, detail=detail) except Exception as e: # noqa: BLE001 return IntegrationTestResult( ok=False, message=redact_provider_error(e, max_len=300), ) def _probe_admin_sdk_sync(self) -> None: admin = self._svc("admin", "directory_v1", admin=True) self._execute_read( admin.users().list(customer="my_customer", maxResults=1), ) # ── Scope / role catalog ───────────────────────────────── async def list_scopes(self) -> list[CloudScopeInfo]: org_id = str(self.config.get("organization_id") or "").strip() if not org_id: raise RuntimeError("organization_id not configured") self._scope_snapshot_partitions = {} try: scopes = await asyncio.to_thread(self._list_projects_sync, org_id) except Exception as exc: # noqa: BLE001 self._scope_snapshot_partitions["gcp_project"] = ( CloudScopePartitionStatus( complete=False, error=redact_provider_error(str(exc), max_len=300), ) ) # Projects are required; there is no safe projects-absent policy. # Fail before the optional billing sibling can make a partial # snapshot look authoritative. raise self._scope_snapshot_partitions["gcp_project"] = ( CloudScopePartitionStatus(complete=True) ) # Also enumerate billing accounts as their own grantable scopes — # GCP billing roles live at the billing-account level, so a user # who wants console billing visibility requests access on one of # these (not on a project). This is an INDEPENDENT authoritative # partition: a billing API failure may leave projects usable, but it # must be surfaced explicitly so the lifecycle service can fence the # billing-account close sweep (#2282). try: billing_scopes = await asyncio.to_thread( self._list_billing_accounts_sync, ) except Exception as exc: # noqa: BLE001 error = redact_provider_error(str(exc), max_len=300) self._scope_snapshot_partitions["gcp_billing_account"] = ( CloudScopePartitionStatus(complete=False, error=error) ) logger.warning( "GCP billing scope allowlist incomplete; partition " "is non-authoritative and must not be swept: %s", error, ) else: scopes.extend(billing_scopes) self._scope_snapshot_partitions["gcp_billing_account"] = ( CloudScopePartitionStatus(complete=True) ) return scopes def scope_snapshot_partitions(self) -> dict[str, CloudScopePartitionStatus]: return dict(getattr(self, "_scope_snapshot_partitions", {})) def _incomplete_scope_snapshot_error(self) -> str | None: incomplete = { scope_type: state for scope_type, state in self.scope_snapshot_partitions().items() if not state.complete } if not incomplete: return None reasons = "; ".join( f"{scope_type}: {state.error or 'incomplete'}" for scope_type, state in sorted(incomplete.items()) ) return f"GCP scope snapshot incomplete — {reasons}" def _list_billing_accounts_sync(self) -> list[CloudScopeInfo]: mode, allowed_names = self._billing_scope_policy() if mode == "projects_only": return [] billing = self._svc("cloudbilling", "v1") organization = "organizations/" + str( self.config.get("organization_id") or "", ).strip() account_cache: dict[str, dict[str, Any]] = {} out: list[CloudScopeInfo] = [] for name in allowed_names: acct = self._get_billing_account_sync( billing, name, account_cache=account_cache, ) self._assert_billing_account_in_organization( billing, acct, organization=organization, account_cache=account_cache, ) if acct.get("open") is False: continue out.append(CloudScopeInfo( scope_type="gcp_billing_account", external_id=name, display_name=acct.get("displayName") or name, status="active", raw={ "open": acct.get("open"), "name": name, "parent": acct.get("parent"), "organization_id": organization.removeprefix( "organizations/", ), "tenant_boundary_verified": True, }, )) return out def _billing_scope_policy(self) -> tuple[str, list[str]]: raw_mode = str(self.config.get("billing_scope_mode") or "").strip().lower() raw_allowlist = self.config.get("billing_account_allowlist") configured_create_account = self.config.get( "create_scope_billing_account", ) if not raw_mode and (raw_allowlist or configured_create_account): # Backward-compatible inference for integrations that already # configured the one account used by create_scope. The boundary # is still explicit: only that account is admitted. raw_mode = "allowlist" if raw_mode not in {"projects_only", "allowlist"}: raise RuntimeError( "billing_scope_mode must explicitly be projects_only or " "allowlist; refusing credential-wide billing discovery" ) if raw_mode == "projects_only": if raw_allowlist: raise RuntimeError( "billing_account_allowlist must be empty when " "billing_scope_mode=projects_only" ) return raw_mode, [] raw_items: list[object] if isinstance(raw_allowlist, (list, tuple, set)): raw_items = list(raw_allowlist) else: raw_items = re.split(r"[,;\s]+", str(raw_allowlist or "")) if configured_create_account: raw_items.append(configured_create_account) names: list[str] = [] seen: set[str] = set() for raw in raw_items: normalized = _normalize_gcp_billing_account_name( str(raw or "").strip(), ) if normalized is None: continue if not re.fullmatch(r"billingAccounts/[A-Za-z0-9-]{3,64}", normalized): raise RuntimeError( f"invalid GCP billing account resource name: {normalized!r}" ) if normalized not in seen: seen.add(normalized) names.append(normalized) if not names: raise RuntimeError( "billing_scope_mode=allowlist requires at least one explicit " "billing account" ) return raw_mode, names def _get_billing_account_sync( self, billing: Any, name: str, *, account_cache: dict[str, dict[str, Any]], ) -> dict[str, Any]: cached = account_cache.get(name) if cached is not None: return cached response = self._execute_read( billing.billingAccounts().get(name=name), ) if not isinstance(response, dict): raise RuntimeError( f"GCP billingAccounts.get({name}) returned a non-object; " "snapshot is incomplete" ) returned_name = str(response.get("name") or "").strip() if returned_name != name: raise RuntimeError( f"GCP billingAccounts.get({name}) returned identity " f"{returned_name!r}; snapshot is incomplete" ) account_cache[name] = response return response def _assert_billing_account_in_organization( self, billing: Any, account: dict[str, Any], *, organization: str, account_cache: dict[str, dict[str, Any]], ) -> None: current = account seen: set[str] = set() for _depth in range(20): current_name = str(current.get("name") or "").strip() if current_name in seen: raise RuntimeError( "GCP billing account parent cycle detected; " "snapshot is incomplete" ) seen.add(current_name) parent = str(current.get("parent") or "").strip() if parent == organization: return if parent.startswith("organizations/"): raise RuntimeError( f"GCP billing account {account.get('name')!r} belongs to " f"{parent}, not configured {organization}" ) if not re.fullmatch(r"billingAccounts/[A-Za-z0-9-]{3,64}", parent): raise RuntimeError( f"GCP billing account {current_name!r} has no verifiable " "organization parent; snapshot is incomplete" ) current = self._get_billing_account_sync( billing, parent, account_cache=account_cache, ) raise RuntimeError( "GCP billing account parent depth exceeded; snapshot is incomplete" ) def _list_projects_sync(self, org_id: str) -> list[CloudScopeInfo]: # Tenant authority comes from recursive parent-bounded list calls, # never credential-wide projects.search. See #2446. # GCP ``sys-*`` system projects (Apigee / AppEngine / Firebase / # Workspace) are intentionally skipped — see # ``_is_gcp_system_project`` docstring. Listing them as scopes # creates ``cmdb_cloud_scopes`` rows that downstream sync handlers # (grants_sync / resource_sync) would later 403 on every cron # tick. Already-existing scope rows for sys-* projects get # auto-closed on the next ``cmdb.scopes_sync`` run (orphan sweep # marks them ``status='closed'``) so no manual cleanup is needed. # Personal / sandbox projects (Gemini auto-create, Console default # names) are likewise skipped — see # ``_is_gcp_personal_sandbox_project``. 2026-05-16 incident: a # single sync tick added 1028 personal projects, ballooning # cmdb_cloud_scopes from 58 → 1086 rows. # # Billing-account gate (PR 2026-05-19): when the integration has # ``create_scope_billing_account`` set, org-root projects (parent # = organizations/*) MUST be linked to that billing account. # This catches the ~111 unfilterable personal sandboxes that # PR #389's name-pattern filter could not (random employee # names, hackathon codenames, etc.). Folder-hosted projects # are fast-path-accepted — the folder hierarchy IS the corp # structure. See ``_is_gcp_org_root_unbilled`` for the # predicate and ``_fetch_billed_project_ids`` for the # cloudbilling API call. billed_project_ids = self._fetch_billed_project_ids() crm = self._svc("cloudresourcemanager", "v3") out: list[CloudScopeInfo] = [] projects = _list_gcp_organization_projects(crm, org_id) 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(CloudScopeInfo( scope_type="gcp_project", external_id=project_id, display_name=proj.get("displayName"), status="active", raw={ "name": proj.get("name"), "project_number": proj.get("projectNumber"), "project_id": proj.get("projectId"), "parent": proj.get("parent"), "labels": proj.get("labels") or {}, "organization_id": org_id, "tenant_boundary_verified": True, }, )) 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 disabled (legacy behavior, kept for orgs that haven't centralized billing). Returns the set of projectIds with billing-enabled links to the billing account otherwise. Raises on any cloudbilling API failure — no silent fallback. The whole point of the gate is to fail loud rather than silently re-introducing the personal-sandbox flood the moment the SA loses billing perms. Per memory ``handler_status_must_raise``. """ billing_account = _normalize_gcp_billing_account_name( self.config.get("create_scope_billing_account"), ) if billing_account is None: return None cloudbilling = self._svc("cloudbilling", "v1") return _fetch_gcp_billed_project_ids(cloudbilling, billing_account) async def list_scope_roles( self, scope: CloudScopeRef, ) -> list[CloudRoleInfo]: # Billing-account scopes get the billing-account role catalog # (billing.viewer etc.) — NOT the project predefined roles. if scope.scope_type == "gcp_billing_account": return [ CloudRoleInfo( role_key=r["role_key"], display_name=r["display_name"], description=r.get("description"), is_custom=False, source="gcp_billing_predefined", scope=scope, is_promoted=bool(r.get("promoted")), ) for r in _GCP_BILLING_ACCOUNT_ROLE_CATALOG ] from app.integrations.drivers._cloud_defaults import load_default_roles defaults = await load_default_roles( "cloud.gcp.default_roles", _GCP_FALLBACK, ) return await asyncio.to_thread(self._list_roles_sync, scope, defaults) def _list_roles_sync( self, scope: CloudScopeRef, defaults: list[dict[str, Any]], ) -> list[CloudRoleInfo]: iam = self._svc("iam", "v1") out: list[CloudRoleInfo] = [] ungrantable = self.ungrantable_role_keys(scope.scope_type) # Admin-curated predefined roles (platform_setting). for entry in defaults: key = str(entry.get("role_key") or "").strip() if not key: continue # Drop billing-account/org/folder-level roles a stale # platform_setting may still carry — they can't bind at this # scope so offering them is a dead end. if key in ungrantable: continue out.append(CloudRoleInfo( role_key=key, display_name=str(entry.get("display_name") or key), description=entry.get("description"), is_custom=False, source="gcp_predefined", scope=scope, is_promoted=True, # curated default — surfaced in the modal )) # Project-level custom roles — always promoted (an admin chose # to define them, so they are by definition relevant). try: req = iam.projects().roles().list(parent=f"projects/{scope.external_id}") while req is not None: resp = self._execute_read(req) for r in resp.get("roles") or []: out.append(CloudRoleInfo( role_key=r["name"], display_name=r.get("title") or r["name"], description=r.get("description"), is_custom=True, source="gcp_custom", scope=scope, raw={"stage": r.get("stage")}, is_promoted=True, )) req = iam.projects().roles().list_next(req, resp) except HttpError as exc: # Custom roles require iam.roles.list on the project. The # most common case is a read-only SA that's intentionally # scoped to predefined roles only — log the gap so admins # see why custom roles are missing instead of silent empty. logger.warning( "GCP list custom roles for project %s failed: %s", scope.external_id, exc, ) return out # ── Scope creation: cloudresourcemanager.projects.create ── async def create_scope( self, *, name: str, description: str | None, owner_email: str, budget_usd_monthly: float, metadata: dict[str, Any] | None = None, idempotency_key: str | None = None, # noqa: ARG002 ) -> CloudScopeInfo: """GCP ``cloudresourcemanager.projects.create`` — production path for the self-service "申请新 GCP 项目" flow. Requires the integration's SA to have ``resourcemanager.projects.create`` on the configured parent (``organizations/{id}`` or ``folders/{id}``). Optional: ``create_scope_billing_account`` attaches a billing account post-create — without it the project is created but not billable until an admin manually links one. ``project_id`` (the immutable, dotted form) is derived from the ``name`` slug + a 4-char suffix to keep the global GCP namespace from collision (project ids must be globally unique across all of Google's customers and can't be reused after deletion). The display name is the human-readable string. ``projects.create`` is **async**: returns an Operation immediately and the actual project lands 5-30s later. We poll the operation up to 90s, then either return the :class:`CloudScopeInfo` (success) or raise (timeout / error). On timeout, the request lands in ``failed`` and an admin can manually-bind via the ``manual-provision`` endpoint after verifying the project landed. """ if not _GOOGLE_AVAILABLE: raise RuntimeError( "google-api-python-client not installed — GCP " "create_scope cannot run on this worker.", ) parent = str( self.config.get("create_scope_parent") or "", ).strip() if not parent: raise RuntimeError( "GCP integration is missing create_scope_parent — " "configure the parent organization or folder ID in " "the integration settings before approving a project " "request, then retry. ``cloudresourcemanager.projects." "create`` requires a parent.", ) if not re.fullmatch(r"(?:organizations|folders)/[0-9]+", parent): raise RuntimeError( f"create_scope_parent must look like 'organizations/' " f"or 'folders/', got {parent!r}", ) organization_id = str( self.config.get("organization_id") or "", ).strip() if not organization_id.isdigit(): raise RuntimeError( "organization_id must be numeric before creating a GCP project" ) billing_account = str( self.config.get("create_scope_billing_account") or "", ).strip() or None return await asyncio.to_thread( self._create_project_sync, name, description, owner_email, budget_usd_monthly, metadata, parent, billing_account, ) def _create_project_sync( self, name: str, description: str | None, owner_email: str, budget_usd_monthly: float, metadata: dict[str, Any] | None, parent: str, billing_account: str | None, ) -> CloudScopeInfo: import secrets import time as _time # Project ID rules: 6-30 chars, [a-z][a-z0-9-], must start with # a letter, can't end with dash, can't contain "google" or # restricted strings. Lowercase + dash-only normalization # covers every case the user's name might break. base_slug = "".join( c if (c.isalnum() or c == "-") else "-" for c in name.strip().lower() ).strip("-") or "helix" while "--" in base_slug: base_slug = base_slug.replace("--", "-") # Reserve room for a 5-char suffix (1 dash + 4 chars). base_slug = base_slug[:25].rstrip("-") if not base_slug or not base_slug[0].isalpha(): base_slug = f"helix-{base_slug}".rstrip("-")[:25] # 4 random hex chars = 65k combinations; keeps the project_id # human-recognizable while making collisions effectively zero # at our request volume. suffix = secrets.token_hex(2) project_id = f"{base_slug}-{suffix}"[:30] crm = self._svc("cloudresourcemanager", "v3") organization = "organizations/" + str( self.config.get("organization_id") or "", ).strip() self._assert_resource_parent_in_organization( crm, parent, organization=organization, ) billing = None if billing_account: normalized_billing_account = _normalize_gcp_billing_account_name( billing_account, ) if ( normalized_billing_account is None or not re.fullmatch( r"billingAccounts/[A-Za-z0-9-]{3,64}", normalized_billing_account, ) ): raise RuntimeError( "create_scope_billing_account is not a canonical GCP " "billing account resource name" ) billing_account = normalized_billing_account billing = self._svc("cloudbilling", "v1") account_cache: dict[str, dict[str, Any]] = {} account = self._get_billing_account_sync( billing, billing_account, account_cache=account_cache, ) self._assert_billing_account_in_organization( billing, account, organization=organization, account_cache=account_cache, ) body: dict[str, Any] = { "projectId": project_id, "displayName": name[:30], # GCP caps displayName at 30 "parent": parent, "labels": { "managed-by": "helix", }, } try: op = self._execute_write(crm.projects().create(body=body)) except HttpError as exc: raise RuntimeError( f"projects.create rejected: {exc}", ) from exc op_name = str(op.get("name") or "").strip() if not op_name: raise RuntimeError( f"projects.create returned no operation name: {op!r}", ) # Poll the operation up to 90s. Each poll is cheap; we want # snappy feedback when projects land in <10s (typical) without # holding the request thread for hours when GCP is slow. deadline = _time.monotonic() + 90 last_state: dict[str, Any] = {} while _time.monotonic() < deadline: try: last_state = self._execute_read( crm.operations().get(name=op_name), ) except HttpError as exc: raise RuntimeError( f"operations.get({op_name}) failed: {exc}", ) from exc if last_state.get("done"): break _time.sleep(3) if not last_state.get("done"): raise RuntimeError( f"projects.create timed out after 90s " f"(operation={op_name}). Check `gcloud projects " f"describe {project_id}` and bind manually if it " f"eventually succeeds.", ) err = last_state.get("error") if err: raise RuntimeError( f"projects.create operation failed: {err.get('message') or err}", ) # Optional: attach billing account. Failure here is a real # error (project exists but can't run anything), so we surface # it. Admin can rerun after fixing IAM or attach manually. if billing_account: try: assert billing is not None self._execute_write( billing.projects().updateBillingInfo( name=f"projects/{project_id}", body={"billingAccountName": billing_account}, ), ) except HttpError as exc: raise RuntimeError( f"projects.create succeeded but billing attach to " f"{billing_account} failed: {exc}. Project " f"{project_id} exists; attach billing manually.", ) from exc return CloudScopeInfo( scope_type="gcp_project", external_id=project_id, display_name=name, status="active", raw={ "parent": parent, "billing_account": billing_account, "description": description, "metadata": metadata or {}, "budget_usd_monthly": budget_usd_monthly, "operation_name": op_name, "organization_id": organization.removeprefix("organizations/"), "tenant_boundary_verified": True, }, ) def _assert_resource_parent_in_organization( self, crm: Any, parent: str, *, organization: str, ) -> None: """Verify an organization/folder create parent before any mutation.""" if parent == organization: return if parent.startswith("organizations/"): raise RuntimeError( f"create_scope_parent {parent!r} is outside configured " f"{organization!r}" ) current = parent seen: set[str] = set() for _depth in range(100): if current in seen: raise RuntimeError( "create_scope_parent folder ancestry contains a cycle" ) seen.add(current) response = self._execute_read(crm.folders().get(name=current)) if not isinstance(response, dict): raise RuntimeError( f"folders.get({current}) returned a non-object response" ) returned_name = str(response.get("name") or "").strip() if returned_name != current: raise RuntimeError( f"folders.get({current}) returned identity " f"{returned_name!r}" ) ancestor = str(response.get("parent") or "").strip() if ancestor == organization: return if ancestor.startswith("organizations/"): raise RuntimeError( f"create_scope_parent {parent!r} belongs to {ancestor!r}, " f"not configured {organization!r}" ) if not re.fullmatch(r"folders/[0-9]+", ancestor): raise RuntimeError( f"folder {current!r} has no verifiable organization parent" ) current = ancestor raise RuntimeError( "create_scope_parent folder ancestry depth exceeded" ) # ── CMDB list overrides ─────────────────────────────────── # # GCP has no concept directly comparable to AWS Route 53 Registered # Domains or Aliyun registered domains, so list_domains stays empty # (overriding _CloudDriverBase which would raise CapabilityNotSupported # through the provider). async def list_domains(self) -> list[dict[str, Any]]: return [] async def list_load_balancers(self) -> list[dict[str, Any]]: return [] async def list_cdn_domains(self) -> list[dict[str, Any]]: return [] # ── Grant / revoke (READ-MODIFY-WRITE with etag) ────────── async def grant_access( self, ctx: CloudAccessContext, ) -> CloudGrantResult: # Billing-account / org / folder-level roles cannot bind on a # project IAM policy — GCP rejects them with a 400 "not supported # for this resource". Refuse before the API round-trip (the full # ungrantable set, not just billing.*, so org/folder roles also # fail fast) with an actionable message rather than a raw Google # error — last line of defence behind the request/approve gates. if ctx.role_key in self.ungrantable_role_keys(ctx.scope.scope_type): return CloudGrantResult( ok=False, error_message=( f"角色 {ctx.role_key} 是 GCP 账单账号/组织/文件夹级角色," f"不能授予在 {ctx.scope.external_id} 上(GCP setIamPolicy " f"会拒绝)。请改用项目级角色(如 roles/viewer);账单" f"可见性需在 GCP 账单账号上单独授权。" ), ) # Best-effort pre-validation: if the SA can talk to Admin SDK, # verify the principal exists in the Workspace before even # attempting setIamPolicy. Saves a round-trip on typos + gives # a cleaner error message. if self._has_admin_impersonation(): exists = await asyncio.to_thread( self._principal_exists_sync, ctx.principal_email, ) if exists is False: # Raise CloudPrincipalNotFoundError so the lifecycle service # surfaces a 409 "principal not in IdP" — Helix's SCIM / # onboarding docs point at this. raise CloudPrincipalNotFoundError( f"{ctx.principal_email!r} not found in Google Workspace " f"(via Admin SDK Directory)" ) # exists is None → admin SDK unavailable, fall through. if ctx.scope.scope_type == "gcp_billing_account": return await asyncio.to_thread( self._modify_billing_policy_sync, ctx, add=True, ) return await asyncio.to_thread(self._modify_policy_sync, ctx, add=True) async def revoke_access( self, ctx: CloudAccessContext, ) -> CloudGrantResult: if ctx.scope.scope_type == "gcp_billing_account": return await asyncio.to_thread( self._modify_billing_policy_sync, ctx, add=False, ) return await asyncio.to_thread(self._modify_policy_sync, ctx, add=False) def _principal_exists_sync(self, email: str) -> bool | None: """Return True/False if we could check, None if Admin SDK unavailable.""" try: admin = self._svc("admin", "directory_v1", admin=True) except Exception as exc: # noqa: BLE001 logger.info( "GCP admin SDK unavailable for principal check (%s): %s", email, exc, ) return None try: self._execute_read(admin.users().get(userKey=email)) return True except HttpError as exc: if exc.resp.status == 404: return False # 403 = impersonation fails → treat as "unknown" logger.info("GCP admin users.get non-404 error: %s", exc) return None @staticmethod def _policy_error_names_member(e: HttpError, member: str) -> bool: """True when a setIamPolicy 400 carries a structured ``ProjectIamPolicyError`` detail that names THIS call's member. cloudresourcemanager v3 rejects a binding for a deleted / never-provisioned Google account with a generic top-level message ("Request contains an invalid argument.") — the only machine-readable signal is in ``error.details``:: {"@type": ".../google.cloudresourcemanager.v1.ProjectIamPolicyError", "role": "roles/owner", "member": "user:gone@example.com", ...} The member match is REQUIRED: setIamPolicy validates the whole policy, so a detail naming some *other* stale member means the policy has a pre-existing problem — not that our principal is missing — and must stay a generic failure. """ try: payload = json.loads((e.content or b"").decode("utf-8", "replace")) except (ValueError, AttributeError): return False details = (payload.get("error") or {}).get("details") \ if isinstance(payload, dict) else None if not isinstance(details, list): return False member_lc = member.lower() for entry in details: if not isinstance(entry, dict): continue if not str(entry.get("@type") or "").endswith("ProjectIamPolicyError"): continue if str(entry.get("member") or "").lower() == member_lc: return True return False def _rmw_iam_member( self, *, ctx: CloudAccessContext, add: bool, get_policy, set_policy, ref_prefix: str, detail_extra: dict[str, Any], ) -> CloudGrantResult: """Generic IAM read-modify-write-with-etag for a single ``user:`` binding. Shared by the project (cloudresourcemanager) and billing-account (cloudbilling) paths; ``get_policy``/``set_policy`` close over the provider-specific resource. ``external_ref`` = ``::::`` (revoke-able).""" member = f"user:{ctx.principal_email}" last_err: str | None = None for attempt in range(3): try: policy = get_policy() except Exception as e: # noqa: BLE001 return CloudGrantResult( ok=False, error_message=redact_provider_error(e), ) bindings = policy.get("bindings") or [] target: dict[str, Any] | None = None for b in bindings: if b.get("role") == ctx.role_key: target = b break already = target is not None and member in (target.get("members") or []) if add: if already: return CloudGrantResult( ok=True, already_exists=True, external_ref=f"{ref_prefix}::{ctx.role_key}::{member}", detail=dict(detail_extra), ) if target is None: bindings.append({"role": ctx.role_key, "members": [member]}) else: target.setdefault("members", []).append(member) else: if not already: return CloudGrantResult(ok=True, already_absent=True) target["members"] = [m for m in target.get("members") or [] if m != member] if not target["members"]: bindings.remove(target) policy["bindings"] = bindings try: set_policy(policy) return CloudGrantResult( ok=True, external_ref=f"{ref_prefix}::{ctx.role_key}::{member}", detail={**detail_extra, "attempts": attempt + 1}, ) except Exception as e: # noqa: BLE001 # 409 / ABORTED indicates etag mismatch — re-read and retry. status = getattr(getattr(e, "resp", None), "status", None) if status in (409, 412) and attempt < 2: last_err = redact_provider_error(e) continue if status in (409, 412): return CloudGrantResult( ok=False, error_message=( "GCP setIamPolicy etag retries exhausted: " f"{redact_provider_error(e)}" ), ) # 400 with "not a valid principal" → email unmapped to # a real Google account (external contractor, shared inbox). content = getattr(e, "content", None) or b"" if isinstance(e, HttpError) and ( b"does not exist" in content or b"not a valid" in content ): raise CloudPrincipalNotFoundError( f"GCP IAM rejected principal {ctx.principal_email!r}: {e}" ) from e # Deleted Google accounts fail differently: cloudresourcemanager # v3 returns 400 "Request contains an invalid argument." with a # structured ProjectIamPolicyError detail naming the member — # neither substring above appears (prod: gcp-storage-datarepl-dev # / weiwan@, 2026-07 — the generic classification kept the grant # PENDING and the backfill cron retried setIamPolicy every 10 min). if ( isinstance(e, HttpError) and status == 400 and self._policy_error_names_member( e, member, ) ): raise CloudPrincipalNotFoundError( f"GCP IAM rejected principal {ctx.principal_email!r} — " f"the Google account does not exist (deleted or never " f"provisioned; setIamPolicy returned a " f"ProjectIamPolicyError for this member). Restore / " f"create the Google account, or point the grant at a " f"valid principal. Configuring admin_impersonation_email " f"on the integration enables the Admin SDK pre-check " f"that catches this before the IAM round-trip. ({e})" ) from e # A definitive 4xx (other than timeout/throttle) says Google # rejected the mutation before accepting it. Preserve the # ordinary failure contract; only transport/408/429/5xx have # an ambiguous commit outcome that requires reconciliation. if ( isinstance(e, HttpError) and isinstance(status, int) and status < 500 and status not in (408, 429) ): return CloudGrantResult( ok=False, error_message=redact_provider_error(e), ) # A transport/5xx error can arrive after Google committed the # policy. Never replay that write blindly. Re-read the policy # through the bounded idempotent-read path and turn the result # into success only when the desired state is authoritative. write_error = redact_provider_error(e) try: reconciled_policy = get_policy() except Exception as reconcile_exc: # noqa: BLE001 return CloudGrantResult( ok=False, error_message=( "GCP setIamPolicy outcome unknown; reconciliation " "read failed: " f"{redact_provider_error(reconcile_exc)}" ), detail={ **detail_extra, "write_outcome": "unknown", "write_error": write_error, }, ) observed = self._policy_role_has_member( reconciled_policy, role_key=ctx.role_key, member=member, ) if observed is add: return CloudGrantResult( ok=True, external_ref=( f"{ref_prefix}::{ctx.role_key}::{member}" ), detail={ **detail_extra, "attempts": attempt + 1, "write_outcome": "reconciled_applied", "write_error": write_error, }, ) return CloudGrantResult( ok=False, error_message=( "GCP setIamPolicy failed and reconciliation confirmed " "the desired IAM state was not applied" ), detail={ **detail_extra, "write_outcome": "confirmed_not_applied", "write_error": write_error, }, ) return CloudGrantResult( ok=False, error_message=f"GCP setIamPolicy retries exhausted: {last_err}", ) @staticmethod def _policy_role_has_member( policy: dict[str, Any], *, role_key: str, member: str, ) -> bool: return any( binding.get("role") == role_key and member in (binding.get("members") or []) for binding in (policy.get("bindings") or []) if isinstance(binding, dict) ) def _modify_policy_sync( self, ctx: CloudAccessContext, *, add: bool, ) -> CloudGrantResult: crm = self._svc("cloudresourcemanager", "v3") project = ctx.scope.external_id return self._rmw_iam_member( ctx=ctx, add=add, get_policy=lambda: self._execute_read( crm.projects().getIamPolicy( resource=f"projects/{project}", body={"options": {"requestedPolicyVersion": 3}}, ), ), set_policy=lambda policy: self._execute_write( crm.projects().setIamPolicy( resource=f"projects/{project}", body={"policy": policy}, ), ), ref_prefix=project, detail_extra={"project": project}, ) def _modify_billing_policy_sync( self, ctx: CloudAccessContext, *, add: bool, ) -> CloudGrantResult: """Bind/unbind a billing role on a billing ACCOUNT (not a project) via the Cloud Billing API. ``external_id`` is the full resource name ``billingAccounts/XXXX-XXXX-XXXX``.""" billing = self._svc("cloudbilling", "v1") ba = ctx.scope.external_id # "billingAccounts/XXXX-XXXX-XXXX" return self._rmw_iam_member( ctx=ctx, add=add, get_policy=lambda: self._execute_read( billing.billingAccounts().getIamPolicy( resource=ba, # requestedPolicyVersion=3 so a read-modify-write preserves # conditional (v3) bindings; without it setIamPolicy would # silently drop any condition-bearing binding on the billing # account (#1654). NOTE cloudbilling v1 getIamPolicy is a GET # — the version is the query param options.requestedPolicyVersion # (Python kwarg options_requestedPolicyVersion), NOT a request # body. #1654 copied the project path's body form and broke it # ("unexpected keyword argument body"); project getIamPolicy is # a POST that does take body, this one does not. options_requestedPolicyVersion=3, ), ), set_policy=lambda policy: self._execute_write( billing.billingAccounts().setIamPolicy( resource=ba, body={"policy": policy}, ), ), ref_prefix=ba, detail_extra={"billing_account": ba}, ) # ── Reverse grants discovery (cloud → Helix) ───────────────── async def list_scope_grants( self, scope: CloudScopeRef, ) -> list[DiscoveredGrant]: """Enumerate every ``user:`` binding on the project's IAM policy. GCP represents principals inline as ``"user:alice@example.com"`` (or ``serviceAccount:``, ``group:``, ``domain:``, etc.) — no separate lookup needed. We filter to ``user:`` only; groups and service accounts don't map to a Helix person. GCP's policy doesn't expose per-binding assignment ids, so ``external_ref`` composes ``::::`` — the same shape ``grant_access`` returns, allowing a later revoke to target the exact (role, member) pair.""" if scope.scope_type == "gcp_billing_account": return await asyncio.to_thread(self._list_billing_grants_sync, scope) return await asyncio.to_thread( self._list_scope_grants_sync, scope, ) def _list_billing_grants_sync( self, scope: CloudScopeRef, ) -> list[DiscoveredGrant]: """Enumerate ``user:`` bindings on a billing ACCOUNT IAM policy (cloudbilling), mirroring the project variant. 403 on a billing account the SA can't read → raise (same drift-sweep safety as the project path); 404 → empty.""" billing = self._svc("cloudbilling", "v1") ba = scope.external_id try: policy = self._execute_read( billing.billingAccounts().getIamPolicy( resource=ba, # v3 so conditional bindings are enumerated (not silently # collapsed to their non-conditional form) during discovery. # cloudbilling v1 getIamPolicy is a GET — pass the version as # the options_requestedPolicyVersion query kwarg, NOT body= # (body= raises "unexpected keyword argument body"; #1654). options_requestedPolicyVersion=3, ), ) except HttpError as exc: status = getattr(getattr(exc, "resp", None), "status", None) if status == 404: logger.info( "GCP list_scope_grants: 404 on billing account %s — skip", ba, ) return [] raise out: list[DiscoveredGrant] = [] for binding in policy.get("bindings") or []: role_key = str(binding.get("role") or "") if not role_key: continue for member in binding.get("members") or []: member_s = str(member) if not member_s.startswith("user:"): continue email = member_s[len("user:"):].strip().lower() if not email: continue out.append(DiscoveredGrant( principal_email=email, principal_type="user", role_key=role_key, role_display_name=None, external_ref=f"{ba}::{role_key}::{member_s}", scope=scope, granted_at=None, raw={ "billing_account": ba, "role": role_key, "member": member_s, "condition": binding.get("condition"), }, )) return out def _list_scope_grants_sync( self, scope: CloudScopeRef, ) -> list[DiscoveredGrant]: project = scope.external_id # Defense-in-depth: ``list_scopes`` already filters ``sys-*`` # projects, but a scope row created by an earlier driver version # (before that filter landed) may still live in # ``cmdb_cloud_scopes`` until the next ``scopes_sync`` orphan # sweep closes it. Skip the API call so the in-between window # doesn't emit a daily 403 noise floor. Empty list is the # correct response: a system project has no user grants # discoverable through the project IAM surface, by design. # Bail BEFORE the ``_svc`` build — both saves the SA-credential # decode for system projects and lets unit tests exercise the # filter without faking google-auth. if _is_gcp_system_project(project): return [] # Same defense-in-depth for employee-owned personal / sandbox # projects (Gemini ``gen-lang-client-*`` auto-create pattern). # We only have the projectId here (not displayName), so this # bail short-circuits the ``gen-lang-client-`` prefix family — # the "My First Project" / "Untitled project" families have # random projectIds and rely on the upstream scope-listing # filter to prevent their rows being created in the first place. # 2026-05-16 incident: 179 ``gen-lang-client-*`` projects were # mass-imported as scopes; the row-cleanup script soft-deleted # them, but a future driver-side regression could resurface them. if _is_gcp_personal_sandbox_project({"projectId": project}): return [] crm = self._svc("cloudresourcemanager", "v3") try: policy = self._execute_read( crm.projects().getIamPolicy( resource=f"projects/{project}", body={"options": {"requestedPolicyVersion": 3}}, ), ) except HttpError as exc: status = getattr(getattr(exc, "resp", None), "status", None) # 404 = project deleted/gone — legitimately empty. # 403 = SA lost ``resourcemanager.projects.getIamPolicy`` (or the # binding was revoked) — returning [] would feed the daily # grants-sync drift sweep an empty cloud-side roster and it # would mass-revoke every Helix-tracked GCP grant on this # project, capped only by the 50% drift floor (small projects # bypass it entirely). AWS and Aliyun already raise on auth # errors here for the same reason — see aws_cloud.py # equivalent and round-15 PR #216 history. if status == 404: logger.info( "GCP list_scope_grants: 404 on project %s — skip", project, ) return [] raise out: list[DiscoveredGrant] = [] for binding in policy.get("bindings") or []: role_key = str(binding.get("role") or "") if not role_key: continue for member in binding.get("members") or []: member_s = str(member) if not member_s.startswith("user:"): # serviceAccount / group / domain / deleted: — # don't resolve to a Helix person. Skip. continue email = member_s[len("user:"):].strip().lower() if not email: continue out.append(DiscoveredGrant( principal_email=email, principal_type="user", role_key=role_key, role_display_name=None, external_ref=f"{project}::{role_key}::{member_s}", scope=scope, granted_at=None, # GCP policy doesn't surface this raw={ "project": project, "role": role_key, "member": member_s, "condition": binding.get("condition"), }, )) return out # ── Last login (via Admin SDK Directory) ────────────────── async def get_last_login( self, ctx: CloudAccessContext, ) -> datetime | None: if not self._has_admin_impersonation(): return None return await asyncio.to_thread( self._get_last_login_sync, ctx.principal_email, ) def _get_last_login_sync(self, email: str) -> datetime | None: try: admin = self._svc("admin", "directory_v1", admin=True) except Exception as exc: # noqa: BLE001 logger.info("GCP admin SDK unavailable for get_last_login: %s", exc) return None try: user = self._execute_read( admin.users().get( userKey=email, projection="basic", ), ) except HttpError as exc: if exc.resp.status == 404: return None logger.info("GCP users.get failed for %s: %s", email, exc) return None raw = user.get("lastLoginTime") if not raw: return None # Workspace returns RFC3339, e.g. "2026-03-09T12:34:56.000Z". # The sentinel "1970-01-01T00:00:00.000Z" means the user has never # signed in — normalize to None so the inactivity sweep doesn't # flag brand-new never-used accounts. try: parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) except ValueError: logger.info("Unexpected lastLoginTime format: %s", raw) return None if parsed.year <= 1970: return None return parsed # ── Account revocation (full sweep) ─────────────────────── async def revoke_account( self, ctx: CloudAccountRevokeContext, ) -> ProvisioningResult: """Remove ``user:`` from every binding in every visible project. ``account_id`` must be the email (Google principals are identified by email in IAM). Folder / org-level bindings are NOT swept — those require elevated permissions most SAs don't have, and they're typically for admin groups, not individual employees. The lifecycle layer surfaces the projects touched so an admin can spot any leftover. """ if not ctx.account_id or "@" not in ctx.account_id: return ProvisioningResult( ok=False, error_message=( f"GCP revoke_account requires an email as account_id; " f"got {ctx.account_id!r}" ), ) try: scopes = await self.list_scopes() except Exception as exc: # noqa: BLE001 # Return a structured failure so the lifecycle caller records its # normal IAM-cleanup failure audit. Letting provider pagination / # permission errors escape here produced a generic 500 with no # cleanup audit even though no IAM call had been attempted. error = redact_provider_error(str(exc), max_len=300) return ProvisioningResult( ok=False, error_message=( "GCP revoke_account refused: scope enumeration failed — " f"{error}. No IAM changes were attempted." ), ) incomplete = self._incomplete_scope_snapshot_error() if incomplete is not None: return ProvisioningResult( ok=False, error_message=( f"GCP revoke_account refused: {incomplete}. " "No IAM changes were attempted." ), ) # Round-22 audit hardening: if list_scopes() returns # materially fewer logical scopes than the prior sync recorded for # this integration (e.g. SA lost ``resourcemanager.projects. # list`` mid-flight, IAM-propagation race per memory "GCP 403 # — audit log is source of truth"), the per-project revoke # loop will quietly succeed on the visible subset and the # leaver still holds access on the invisible projects. Cross- # check the active, previously-synced logical identity count against # ``cmdb_cloud_scopes`` and refuse the # revoke when today's scan saw <90% of the previously-recorded # project count for this integration — caller's lifecycle # service can then surface an actionable error and the leaver # stays blocked-on-Helix until SRE resolves the discovery # drift. try: from sqlalchemy import func, select from app.core.database import async_session_factory from app.domains.cmdb.models import CmdbCloudScope async with async_session_factory() as drift_session: prior_scope_keys = ( select( CmdbCloudScope.scope_type, CmdbCloudScope.external_id, ) .where( CmdbCloudScope.integration_id == str(self.integration_id), CmdbCloudScope.status == "active", CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.last_seen_sync_job_id.is_not(None), ) .group_by( CmdbCloudScope.scope_type, CmdbCloudScope.external_id, ) .subquery() ) prior_q = await drift_session.execute( select(func.count()).select_from(prior_scope_keys) ) prior_count = int(prior_q.scalar() or 0) today_scope_keys = { (scope.scope_type, scope.external_id) for scope in scopes } today_count = len(today_scope_keys) # Only refuse when (a) Helix knows about >5 prior scopes # (otherwise small fleets have noisy ratios) and (b) # today's count is a real drop, not a same-or-grow snapshot. if prior_count >= 5 and today_count < (prior_count * 0.9): return ProvisioningResult( ok=False, error_message=( f"GCP revoke_account refused: list_scopes() saw " f"{today_count} unique active scopes but the prior " f"Helix sync recorded {prior_count} (drop > 10%). " f"This likely " f"means the SA's ``resourcemanager.projects.list`` " f"permission was just propagated-out or a parent " f"folder's IAM is in flux. Re-run after the org-" f"level IAM stabilises (≤15 min); revoking against " f"a truncated project list would silently leave " f"the leaver with access on the invisible projects." ), raw_response={ "scopes_today": today_count, "scopes_prior": prior_count, "scope_types_today": { scope_type: sum( 1 for candidate_type, _ in today_scope_keys if candidate_type == scope_type ) for scope_type in sorted({ scope_type for scope_type, _ in today_scope_keys }) }, }, ) except Exception: # noqa: BLE001 # If the drift check itself fails (DB hiccup), fall # through — we'd rather attempt the revoke than block on # a defense-in-depth check that can't run. The per- # project loop still surfaces individual failures. logger.exception( "GCP revoke_account: project-count drift check failed; " "proceeding with revoke against %d projects from current " "list_scopes()", len(scopes), ) result = await asyncio.to_thread( self._revoke_account_sync, ctx.account_id, scopes, ) return result def _revoke_account_sync( self, email: str, scopes: list[CloudScopeInfo], ) -> ProvisioningResult: crm = self._svc("cloudresourcemanager", "v3") member = f"user:{email}" touched: list[str] = [] failed: list[tuple[str, str]] = [] billing = None # lazily built only if a billing scope is present for scope in scopes: ext = scope.external_id try: # #1642: dispatch by scope_type. A gcp_billing_account scope's # external_id is ``billingAccounts/…`` — running the project # getIamPolicy against it 404/errored, so the leaver's billing # IAM was never removed and the whole revoke returned 502. if scope.scope_type == "gcp_billing_account": if billing is None: billing = self._svc("cloudbilling", "v1") self._remove_member_from_billing_account(billing, ext, member) else: self._remove_member_from_project(crm, ext, member) touched.append(ext) except _NoChangeNeeded: continue except Exception as exc: # noqa: BLE001 logger.warning( "GCP revoke_account failed on scope %s: %s", ext, exc, ) failed.append((ext, str(exc))) detail = { "projects_modified": touched, "projects_failed": failed, "projects_scanned": len(scopes), } if failed: return ProvisioningResult( ok=False, external_ref=email, error_message=( f"GCP revoke_account touched {len(touched)}/{len(scopes)} " f"project(s); {len(failed)} failed" ), raw_response=detail, ) return ProvisioningResult( ok=True, external_ref=email, raw_response=detail, ) def _remove_member_from_project( self, crm: Any, project: str, member: str, ) -> None: self._remove_member_via_policy_rmw( member=member, resource_label=f"projects/{project}", get_policy=lambda: self._execute_read( crm.projects().getIamPolicy( resource=f"projects/{project}", body={"options": {"requestedPolicyVersion": 3}}, ), ), set_policy=lambda policy: self._execute_write( crm.projects().setIamPolicy( resource=f"projects/{project}", body={"policy": policy}, ), ), ) def _remove_member_from_billing_account( self, billing: Any, billing_account: str, member: str, ) -> None: """#1642: billing-account IAM member removal. The revoke_account sweep previously ran the PROJECT getIamPolicy against a ``billingAccounts/…`` external_id, so a leaver kept their billing- account IAM and every offboarding falsely returned 502. Uses the cloudbilling API with the same proven per-binding removal + etag retry logic.""" self._remove_member_via_policy_rmw( member=member, resource_label=billing_account, get_policy=lambda: self._execute_read( billing.billingAccounts().getIamPolicy( resource=billing_account, # cloudbilling v1 getIamPolicy is a GET — version is the # options_requestedPolicyVersion query kwarg, not body= (#1654). options_requestedPolicyVersion=3, ), ), set_policy=lambda policy: self._execute_write( billing.billingAccounts().setIamPolicy( resource=billing_account, body={"policy": policy}, ), ), ) def _remove_member_via_policy_rmw( self, *, member: str, resource_label: str, get_policy: Any, set_policy: Any, ) -> None: # Bumped from 3 → 5 retries with capped backoff. Concurrent admin # edits on the same policy under onboarding bursts make 412 etag # conflicts realistic; 3 attempts can race-lose. max_attempts = 5 for attempt in range(max_attempts): policy = get_policy() bindings = policy.get("bindings") or [] policy_changed = False new_bindings: list[dict[str, Any]] = [] # CRITICAL: track per-binding whether WE removed members from # it. The previous policy-wide ``changed`` flag dropped every # subsequent originally-empty binding once we'd removed from # any binding (the ``elif not changed: append`` branch was # only taken before the first removal). That silently # rewrote IAM by deleting unrelated empty conditional # bindings — production would log a successful # setIamPolicy while bindings disappeared. for b in bindings: original_members = list(b.get("members") or []) we_removed_here = False if member in original_members: members = [m for m in original_members if m != member] we_removed_here = True policy_changed = True else: members = original_members if members: # Binding still has members — keep with the new list. b["members"] = members new_bindings.append(b) elif we_removed_here: # We emptied this binding by removing the leaver — drop it. pass else: # Binding was already empty (legacy, conditional # placeholders, etc.) — preserve it untouched. new_bindings.append(b) if not policy_changed: raise _NoChangeNeeded() policy["bindings"] = new_bindings try: set_policy(policy) return except Exception as e: # noqa: BLE001 status = getattr(getattr(e, "resp", None), "status", None) if status in (409, 412) and attempt < max_attempts - 1: # Capped exponential backoff so we don't spin under contention. import time as _time _time.sleep(min(0.2 * (2 ** attempt), 2.0)) continue write_error = redact_provider_error(e) try: reconciled_policy = get_policy() except Exception as reconcile_exc: # noqa: BLE001 raise RuntimeError( f"setIamPolicy on {resource_label} outcome unknown; " "reconciliation read failed: " f"{redact_provider_error(reconcile_exc)}; " f"write error: {write_error}" ) from reconcile_exc member_absent = all( member not in (binding.get("members") or []) for binding in (reconciled_policy.get("bindings") or []) if isinstance(binding, dict) ) if member_absent: logger.warning( "GCP setIamPolicy on %s raised but reconciliation " "confirmed member absent; treating revoke as applied: %s", resource_label, write_error, ) return raise RuntimeError( f"setIamPolicy on {resource_label} failed and " "reconciliation confirmed the member is still present: " f"{write_error}" ) from e raise RuntimeError( f"setIamPolicy on {resource_label} exhausted etag retries", ) # ── Scope-level closure (admin "下线") ───────────────────── async def close_scope_on_provider( self, scope: CloudScopeRef, *, iam_account_ids: list[str] | None = None, ) -> CloseScopeResult: """Real cloud-side scope closure. For ``gcp_project`` → ``cloudresourcemanager.projects.delete``. GCP marks the project ``DELETE_REQUESTED`` immediately; it is permanently removed after a 30-day grace period (during which an Owner can ``undelete`` via the GCP console). Idempotent: a project already in ``DELETE_REQUESTED`` / ``DELETE_IN_PROGRESS`` / 404 is treated as ``already_closed=True``. Anything else raises ``NotImplementedError``. GCP has no single-tenant IAM scopes — every GCP scope in Helix is a project. ``iam_account_ids`` is accepted for protocol symmetry with sibling drivers but ignored here. """ del iam_account_ids # unused; project-only model if not _GOOGLE_AVAILABLE: return CloseScopeResult( ok=False, error_code="google_libs_missing", error_message="google-api-python-client not installed", ) if scope.scope_type != "gcp_project": raise NotImplementedError( f"GCP driver cannot close scope_type={scope.scope_type!r}", ) return await asyncio.to_thread( self._delete_project_sync, scope.external_id, ) # ── Close pre-flight + blocker cleanup ───────────────────── # # GCP's ``projects.delete`` precondition family is narrower than # Azure's: a project full of resources deletes fine (everything # inside goes with it), so the only blocker Helix can both SEE and # CLEAR is a **lien** — the delete-protection marker Cloud Billing, # Anthos and Terraform's ``google_resource_manager_lien`` attach to # projects. It is also the one that produced the worst version of # the old behaviour: the admin got a red modal telling them to run # ``gcloud alpha resource-manager liens list``, which is Helix # handing back a two-API-call chore. # # Billing attachment (``billing_attached``) is deliberately NOT # cleaned automatically. Unlinking a billing account is a finance # action with its own audit trail and its own blast radius (it # instantly stops every billable service in the project, and the # unlink is NOT undone by cancelling the decommission), so it is # reported as a non-removable blocker and escalated to a human. async def preflight_close_scope( self, scope: CloudScopeRef, ) -> ClosePreflight: """List the liens standing between this project and deleted.""" if scope.scope_type != "gcp_project": return ClosePreflight(supported=False) if not _GOOGLE_AVAILABLE: return ClosePreflight( supported=True, checked=False, cleanup_supported=True, error_code="sdk_missing", error_message="google-api-python-client not installed", ) project_id = (scope.external_id or "").strip() if not project_id: return ClosePreflight( supported=True, checked=False, error_code="missing_project_id", error_message="GCP project id is empty", ) return await asyncio.to_thread(self._preflight_close_sync, project_id) def _preflight_close_sync(self, project_id: str) -> ClosePreflight: crm = self._svc("cloudresourcemanager", "v3") liens: list[dict[str, Any]] = [] page_token: str | None = None try: for _ in range(20): resp = self._execute_read(crm.liens().list( parent=f"projects/{project_id}", pageSize=100, pageToken=page_token, )) liens.extend( r for r in (resp.get("liens") or []) if isinstance(r, dict) ) page_token = str(resp.get("nextPageToken") or "") or None if not page_token: break except HttpError as exc: detail_text, _reason = _gcp_extract_error_detail(exc) return ClosePreflight( supported=True, checked=False, cleanup_supported=True, error_code="preflight_read_failed", error_message=( "GCP liens.list: " + redact_provider_error( detail_text or str(exc), max_len=250, ) ), ) except Exception as exc: # noqa: BLE001 return ClosePreflight( supported=True, checked=False, cleanup_supported=True, error_code="preflight_read_failed", error_message=redact_provider_error(str(exc), max_len=250), ) blockers: list[CloseBlocker] = [] for lien in liens: name = str(lien.get("name") or "").strip() if not name: continue origin = str(lien.get("origin") or "").strip() reason = str(lien.get("reason") or "").strip() blockers.append(CloseBlocker( key=name, kind="lien", # ``liens/p1234abcd`` is not a name a human recognises; # the origin ("google.com/billing", a Terraform module # path) is what tells the admin who put it there and # therefore whether removing it is safe. name=origin or name, detail=redact_provider_error(reason, max_len=200) or None, removable=True, )) return ClosePreflight( supported=True, checked=True, cleanup_supported=True, blockers=blockers, detail={"lien_count": len(blockers)}, ) async def clear_close_blockers( self, scope: CloudScopeRef, *, keys: list[str], ) -> ClearBlockersResult: """Delete the confirmed liens so ``projects.delete`` can proceed. Synchronous at the provider — unlike an Azure resource-group delete there is nothing to wait for, so ``in_progress`` stays ``False`` and the caller can retry the close immediately. """ if scope.scope_type != "gcp_project": return ClearBlockersResult( ok=False, error_code="cleanup_unsupported", error_message=( f"GCP driver cannot clean scope_type={scope.scope_type!r}" ), ) if not _GOOGLE_AVAILABLE: return ClearBlockersResult( ok=False, error_code="sdk_missing", error_message="google-api-python-client not installed", ) # Liens are named ``liens/{id}`` — a global namespace with no # project segment, so the cross-scope guard Azure applies to ARM # ids is not available here. Guard on shape instead: anything # that is not a lien name cannot be deleted by this method, and # silently ignoring it would let a caller believe a blocker was # cleared when nothing happened. bad = [k for k in keys if not str(k or "").startswith("liens/")] if bad: return ClearBlockersResult( ok=False, error_code="key_shape_mismatch", error_message=( f"{len(bad)} key(s) are not GCP lien names — refusing" ), detail={"bad_keys": bad[:5]}, ) return await asyncio.to_thread(self._clear_liens_sync, list(keys)) def _clear_liens_sync(self, keys: list[str]) -> ClearBlockersResult: crm = self._svc("cloudresourcemanager", "v3") completed: list[str] = [] failed: list[dict[str, str]] = [] for key in keys: try: self._execute_write(crm.liens().delete(name=key)) completed.append(key) except HttpError as exc: resp = getattr(exc, "resp", None) code = resp.status if resp is not None else 0 if code == 404: # Already gone — idempotent, not a failure. completed.append(key) continue detail_text, _reason = _gcp_extract_error_detail(exc) failed.append({ "key": key, "error": f"{code}: " + redact_provider_error( detail_text or str(exc), max_len=200, ), }) except Exception as exc: # noqa: BLE001 failed.append({ "key": key, "error": redact_provider_error(str(exc), max_len=200), }) return ClearBlockersResult( ok=bool(completed) or not failed, completed=completed, failed=failed, in_progress=False, error_code="cleanup_partial" if failed else None, error_message=( f"{len(failed)} lien(s) could not be deleted" if failed else None ), detail={"liens": len(keys)}, ) # ── Post-close reconciliation (read-only) ────────────────── async def get_scope_closure_state( self, scope: CloudScopeRef, ) -> ScopeClosureState: """Re-read GCP's view of a project Helix already closed. ``projects.delete`` sets ``DELETE_REQUESTED`` and GCP purges the project ~30 days later. **``projects.undelete`` is available to any Owner for that entire window and tells Helix nothing** — an ``ACTIVE`` project on a scope Helix marked closed is exactly that, and it is the reason this read exists for GCP. """ if scope.scope_type != "gcp_project": raise NotImplementedError( "GCP driver has no post-close state for " f"scope_type={scope.scope_type!r}", ) if not _GOOGLE_AVAILABLE: return ScopeClosureState( disposition="unknown", error_message="google-api-python-client not installed", ) return await asyncio.to_thread( self._closure_state_sync, (scope.external_id or "").strip(), ) def _closure_state_sync(self, project_id: str) -> ScopeClosureState: crm = self._svc("cloudresourcemanager", "v3") try: describe = self._execute_read( crm.projects().get(name=f"projects/{project_id}"), ) except HttpError as exc: resp = getattr(exc, "resp", None) code = resp.status if resp is not None else 0 if code == 404: return ScopeClosureState( disposition="closed", provider_state="not_found", detail={"reason": "Project no longer visible to this SA"}, ) return ScopeClosureState( disposition="unknown", error_message=( "GCP projects.get: " + redact_provider_error(str(exc), max_len=250) ), ) except Exception as exc: # noqa: BLE001 return ScopeClosureState( disposition="unknown", error_message=redact_provider_error(str(exc), max_len=250), ) state = str(describe.get("state") or "").upper() detail = {"gcp_state": state} if state in {"DELETE_REQUESTED", "DELETE_IN_PROGRESS"}: return ScopeClosureState( disposition="pending", provider_state=state.lower(), detail=detail, ) if state == "ACTIVE": return ScopeClosureState( disposition="alive", provider_state="active", detail=detail, error_message=( "GCP reports this project as ACTIVE, but Helix has it " "marked closed and has already disabled its grants. " "The most likely cause is a projects.undelete during " "the 30-day window." ), ) return ScopeClosureState( disposition="unknown", detail=detail, error_message=f"GCP returned an unrecognised project state {state!r}", ) def _delete_project_sync(self, project_id: str) -> CloseScopeResult: crm = self._svc("cloudresourcemanager", "v3") try: describe = self._execute_read( crm.projects().get(name=f"projects/{project_id}"), ) except HttpError as exc: status = getattr(exc, "resp", None) code = status.status if status is not None else 0 if code == 404: return CloseScopeResult( ok=True, already_closed=True, provider_state="not_found", detail={"reason": "Project not visible to this SA"}, ) if code == 403: return CloseScopeResult( ok=False, error_code="access_denied", error_message=( "GCP denied projects.get on this project. The SA " "needs Browser (or higher) plus " "resourcemanager.projects.delete to close it." ), ) return CloseScopeResult( ok=False, error_code="describe_failed", error_message=( "GCP projects.get: " + redact_provider_error(str(exc), max_len=300) ), ) existing_state = str(describe.get("state") or "").upper() if existing_state in {"DELETE_REQUESTED", "DELETE_IN_PROGRESS"}: return CloseScopeResult( ok=True, already_closed=True, provider_state=existing_state.lower(), detail={"prior_state": existing_state}, ) try: op = self._execute_write( crm.projects().delete(name=f"projects/{project_id}"), ) except HttpError as exc: status = getattr(exc, "resp", None) code = status.status if status is not None else 0 if code == 404: return CloseScopeResult( ok=True, already_closed=True, provider_state="not_found", detail={"reason": "Project disappeared during delete"}, ) if code == 403: return CloseScopeResult( ok=False, error_code="access_denied", error_message=( "GCP denied projects.delete. The SA needs " "resourcemanager.projects.delete (typically the " "Project Owner role at the project) — Browser is " "not sufficient." ), ) if code in (400, 412): # 400 / 412 are GCP's precondition-failed responses. # Common production causes: # * project has an unresolved lien # (``orgPolicy.projectsLien``) — usually a billing- # anchor lien created by Cloud Billing or Anthos. # * project is still attached to a billing account. # * project is the host of a Shared VPC. # The exception body carries the symbolic reason; surface # it AND a remediation hint so the admin doesn't have to # guess. raw_detail_text, raw_detail_reason = ( _gcp_extract_error_detail(exc) ) # Route every piece of provider-originated text through # the shared credential redactor before it lands in a # user-facing message OR in audit detail. Defense-in- # depth: GCP error bodies don't normally echo creds, but # the same hygiene rule the other drivers apply # (`_safe_body` in azure_cloud.py, `_acs3_call` in # aliyun_cloud.py) must hold here too. Round-52 security # review caught the asymmetry. detail_text = redact_provider_error( raw_detail_text, max_len=400, ) if raw_detail_text else "" detail_reason = redact_provider_error( raw_detail_reason, max_len=120, ) if raw_detail_reason else "" hint = ( " — common causes: (1) project has an unresolved " "lien (run `gcloud alpha resource-manager liens " "list --project=` and remove what you find), " "(2) project is still attached to a billing " "account (unlink via `gcloud beta billing " "projects unlink `), or (3) project is a " "Shared VPC host (detach service projects first)." ) # Dispatch on the symbolic ``detail_reason`` enum # (canonical GCP ErrorInfo.reason / PreconditionFailure # violation.type), NOT on a substring of the free-text # message. The pre-Round-52 form # ``"billing" in detail_text.lower()`` mis-routed lien # messages that mentioned the word "billing" (e.g. # "lien from Cloud Billing"). Keep the message-substring # fallback ONLY when no symbolic reason was present. _BILLING_REASONS = { "BILLING_DISABLED", "BILLING_ACCOUNT_LINKED", "BILLING_REQUIRED", } upper_reason = detail_reason.upper() if detail_reason else "" error_code = "precondition_failed" if upper_reason and "LIEN" in upper_reason: error_code = "liens_exist" elif upper_reason and upper_reason in _BILLING_REASONS: error_code = "billing_attached" elif ( not upper_reason and detail_text and "billing" in detail_text.lower() ): error_code = "billing_attached" # Cap the user-facing text so a verbose GCP message # cannot blow up the admin-modal toast or balloon the # audit-detail JSON column. `redact_provider_error` # already caps to max_len, but trim again here for the # message slot. trimmed_text = ( detail_text[:300] if detail_text else redact_provider_error(str(exc), max_len=300) ) return CloseScopeResult( ok=False, error_code=error_code, error_message=( f"GCP projects.delete rejected: {trimmed_text}" + hint ), detail={ "http_status": code, "gcp_reason": detail_reason, "gcp_message": detail_text, }, ) if code == 429: return CloseScopeResult( ok=False, error_code="throttled", error_message=( "GCP projects.delete throttled (HTTP 429). Wait " "a few minutes and retry." ), ) if code >= 500: return CloseScopeResult( ok=False, error_code="transient_failure", error_message=( f"GCP projects.delete returned {code} — " "server-side failure, safe to retry." ), ) return CloseScopeResult( ok=False, error_code="delete_failed", error_message=( "GCP projects.delete: " + redact_provider_error(str(exc), max_len=300) ), ) return CloseScopeResult( ok=True, provider_state="delete_requested", detail={ "grace_period_days": 30, "operation_name": op.get("name") if isinstance(op, dict) else None, "note": ( "Project moved to DELETE_REQUESTED; GCP permanently " "deletes it 30 days after this call. An Owner can " "undelete via the GCP console during the window." ), }, ) # ── Daily billing (BigQuery export) ───────────────────── async def billing_daily_precheck(self) -> str | None: # GCP's Cloud Billing REST API only exposes month-to-date totals — # useless for per-day rollups — so BigQuery export is the only # path for daily cost. Surface the missing config instead of # silently returning upserted=0. if not str(self.config.get("billing_bq_table") or "").strip(): return ( "billing_bq_table not configured — GCP per-day cost needs " "a BigQuery billing export; set it on the integration to " "enable billing_daily_sync." ) try: import google.cloud.bigquery # noqa: F401 except ImportError: return ( "google-cloud-bigquery is not installed — billing_daily_sync " "cannot query the BigQuery export on this runtime." ) return None async def list_billing_daily( self, start_date: date, end_date: date, *, scope: CloudScopeRef | None = None, include_owner_breakdown: bool = False, bq_billing: dict[str, Any] | None = None, ) -> list[CloudDailyCost]: # ``bq_billing`` is AWS-driver-specific (the prod main AWS Org # reads its CUR from BigQuery rather than Cost Explorer; the # cost-service layer pre-resolves the SA + dataset and threads # it through to every driver). GCP's billing is native BQ, so # the kwarg is structurally identical to what we'd compute # locally — accept it for signature compatibility with the base # class and the service-layer fanout, then ignore. del bq_billing raw = str(self.config.get("billing_bq_table") or "").strip() if not raw: # The lifecycle service short-circuits via # ``billing_daily_precheck`` so this branch is only hit on # direct callers that bypass the normal sync entry-point # (mostly accidental / debug). Raise instead of returning # ``[]`` so the misuse is loud — silently returning an # empty list would let spike-detection run on stale 7-day # averages without telling anyone the missing data was # missing on purpose. raise RuntimeError( "billing_bq_table is not configured on this GCP " "integration — call billing_daily_precheck() first to " "detect this gracefully.", ) # Multi-billing-account orgs ship a comma- or whitespace-separated # list of fully-qualified export tables (one per billing account). # Pre-fix this driver interpolated the raw string into ``FROM # `{table}` `` directly, which made BigQuery treat ``a,b,c`` as one # invalid table id. Same parser/validator the cost-facts path uses # (PR #232) — required parity now that both paths share the same # config field. tables = GcpProvider._split_bq_tables(raw) if not tables: raise RuntimeError( f"billing_bq_table parsed empty from value {raw!r}", ) try: GcpProvider._validate_bq_table_shapes(tables) except ValueError as exc: raise RuntimeError(f"GCP {exc}") from exc return await asyncio.to_thread( self._list_billing_daily_sync, start_date, end_date, scope, tables, ) def _list_billing_daily_sync( self, start: date, end: date, scope: CloudScopeRef | None, tables: list[str], ) -> list[CloudDailyCost]: # ``billing_daily_precheck`` is supposed to gate this method, so a # missing dependency here is a configuration drift, not a normal # path. Raise so the scheduled job records the error against the # integration and the admin sees it in scheduled_job_runs. from google.cloud import bigquery # type: ignore # Round-42 audit: pin BQ Client to the billing-export's project # (extracted from the first configured table) rather than the # SA's home project. The ``list_cost_facts`` twin (line 1452) # already does this; daily billing was the holdout. When the # SA's home project is different from the billing-export project # and the SA project doesn't have BigQuery API enabled, jobs # hard-fail with "BigQuery API has not been used in project ...". # Tables are already shape-validated by ``_validate_bq_table_shapes`` # in the async wrapper before reaching this method. bq_project = tables[0].split(".", 1)[0] if tables else None client = bigquery.Client( credentials=self._credentials(), project=bq_project, ) where_extra = "" # ``_PARTITIONTIME`` is the BQ ingestion-time pseudo-column on the # billing export table. Without a partition predicate the query # becomes a FULL TABLE SCAN — fast on small tenants, but on the # multi-TB exports we hit (PingCAP prod) the job either bills 100s # of GB or trips the project's "require partition filter" policy # and 400s. We then keep the precise ``DATE(usage_start_time)`` # filter for the actual returned rows. # Window padding MUST match ``list_cost_facts`` (start-3d / end+8d): # this query now nets the per-row ``credits`` array into total_cost # (see SELECT below), and GCP back-dates SUD/CUD credit lines days # after the original usage_start_time while their ``_PARTITIONTIME`` # (ingestion time) lands on the credit-write day. A ±1d window would # match the credit row's date but prune its later partition, so the # daily path would under-net late-arriving credits and diverge # (higher) from both the cost-facts surface and the GCP invoice — # which can falsely trip spike-detection / budget-threshold alerts # in ``billing_alerts.py``. See parity with the monthly fix in # ``providers/gcp.py`` (commit 988e7ff). from datetime import timedelta partition_lo = (start - timedelta(days=3)).isoformat() partition_hi = (end + timedelta(days=8)).isoformat() params: list[Any] = [ bigquery.ScalarQueryParameter("start", "DATE", start), bigquery.ScalarQueryParameter("end", "DATE", end), bigquery.ScalarQueryParameter( "partition_lo", "STRING", partition_lo, ), bigquery.ScalarQueryParameter( "partition_hi", "STRING", partition_hi, ), ] if scope is not None: where_extra = "AND project.id = @project_id" params.append(bigquery.ScalarQueryParameter( "project_id", "STRING", scope.external_id, )) # Multi-table UNION ALL — orgs with >1 billing account get one # export table per account. Each per-table SELECT pre-aggregates # to bound row count; the outer SELECT merges duplicates that # can transiently appear during a billing-account migration. # Tables are interpolated raw because ``_validate_bq_table_shapes`` # ran in the async wrapper. The daily-billing query references no # detailed-export-only fields (no ``resource.*``), so the standard/ # detailed flavour split that ``list_cost_facts`` does is not # required here — both flavours expose the columns this query # touches. # Currency stays in BOTH inner and outer GROUP BY — without it, # an org with multiple billing accounts in different currencies # (e.g. one USD-billed account + one CNY-billed account) would # have ``ANY_VALUE(currency)`` arbitrarily pick one tag while # ``SUM(total_cost)`` blended raw amounts across currencies, # silently producing mixed-unit totals downstream. Round-2 # audit caught this. With currency in the GROUP BY each # (day × project × service × currency) row stays atomic and # the downstream ingestion service decides USD attribution per # row from the ``currency`` column. # Validate table name components before interpolation into SQL. for _t in tables: for _part in _t.split("."): _validate_bq_name(_part, "table component") def _daily_sql(with_list: bool) -> str: # 双口径:``total_cost_list`` = 业务折前成本(研发口径,与云 # 管理团队 2026-07-27 用 5 月真实账单核验定稿的白名单): # cost_type='regular' AND service ∉ {Support, Invoice} # AND cost_at_list IS NOT NULL → SUM(cost_at_list)。 # 关键:**必须用 cost_at_list(牌价列),不能用 cost** —— # cost 已折入合同价(5 月实测整体 −21%),给研发即泄露折扣率; # credits 完全不参与(混有 FCUD Fee Offset / 合同 Promotion / # Support Credit,统一扣除会把公司优惠重新混入);Marketplace # (THIRD_PARTY_RESELLER)计入;三种 consumption_model 都保留; # Spot/抢占式在 cost_at_list 里本就是官方 Spot SKU 价;正常 # 用量的负数冲正保留(不加 >0 条件)。 # ``with_list=False`` 是 cost_at_list 列缺失(老导出 schema) # 的降级路径——list 落 NULL(=口径未接入),绝不伪造 0。 list_col = ( """, SUM(CASE WHEN cost_type = 'regular' AND service.description NOT IN ('Support', 'Invoice') AND cost_at_list IS NOT NULL THEN cost_at_list END) AS total_cost_list""" if with_list else "" ) per_table_sql = " UNION ALL ".join( f""" SELECT DATE(usage_start_time) AS day, project.id AS project_id, service.description AS service_name, -- Net-of-credits: ``cost`` is GCP's pre-credit list amount; -- SUD/CUD discounts + committed-use credits + promotions live -- in the per-row ``credits`` array (negative amounts). Net the -- unnested credit sum into total_cost so spike-detection / -- reconciliation see true cost, matching ``list_cost_facts``. IFNULL(SUM(cost), 0) + IFNULL(SUM((SELECT SUM(c.amount) FROM UNNEST(credits) c)), 0) AS total_cost, currency{list_col} FROM `{t}` WHERE _PARTITIONTIME >= TIMESTAMP(@partition_lo) AND _PARTITIONTIME < TIMESTAMP(@partition_hi) AND DATE(usage_start_time) BETWEEN @start AND @end {where_extra} GROUP BY day, project_id, service_name, currency """ for t in tables ) if len(tables) == 1: # Single-table case: keep the original query shape (with # ``ORDER BY day`` for stable row order in tests / debug). return per_table_sql + "\nORDER BY day" outer_list = ( ",\n SUM(total_cost_list) AS total_cost_list" if with_list else "" ) return f""" SELECT day, project_id, service_name, currency, IFNULL(SUM(total_cost), 0) AS total_cost{outer_list} FROM ( {per_table_sql} ) GROUP BY day, project_id, service_name, currency ORDER BY day """ query = _daily_sql(with_list=True) # Round-42 audit: mirror the cost-facts twin (line 1650-1665). # Without ``maximum_bytes_billed`` a malformed partition predicate # on a multi-TB export bills hundreds of GB per run; without # ``job_timeout_ms`` a stuck plan can hang the worker past the # per-job budget; without ``client.close()`` in the finally block, # per-scope iteration on a 50-project tenant leaks 50 gRPC # channels per cron run. Same numbers as cost-facts. job_cfg = bigquery.QueryJobConfig( query_parameters=params, maximum_bytes_billed=50_000_000_000, job_timeout_ms=300_000, ) # Raise on failure — the lifecycle service catches and records # the error per-scope so other scopes still sync. Returning ``[]`` # here would mask BQ outages and leave spike-detection running on # stale 7-day averages that don't include the missing days. # ``cost_at_list`` 缺列(极老的导出 schema)→ 单次降级重试: # 发票口径照常产出,list 度量落 None(=口径未接入),并 WARN # 一次供运维升级导出 schema。其它错误原样上抛。 with_list = True try: try: rows = list(client.query(query, job_config=job_cfg).result()) except Exception as exc: # noqa: BLE001 if "cost_at_list" not in str(exc): raise logger.warning( "GCP billing_daily: export schema lacks cost_at_list " "(%s) — list-cost measure degrades to NULL. Upgrade " "the billing export to the schema that includes " "cost_at_list to enable the engineering cost view.", exc, ) with_list = False rows = list(client.query( _daily_sql(with_list=False), job_config=job_cfg, ).result()) finally: try: client.close() except Exception: # noqa: BLE001 pass # Key on ``(day, project_id, currency)`` — Round-3 audit caught # that the previous ``(day, project_id)`` key re-collapsed the # currency dimension that the SQL ``GROUP BY day, project_id, # service_name, currency`` was carrying, blending native amounts # across currencies into one bucket and leaving an arbitrary # ``ANY_VALUE``-style currency tag. With currency in the key, # multi-currency multi-billing-account orgs get one ``CloudDaily # Cost`` per (day × project × currency) and the downstream # consumer can attribute USD correctly per row. by_key: dict[tuple, dict[str, Any]] = {} for r in rows: day = r["day"] if isinstance(r["day"], date) else date.fromisoformat(str(r["day"])) # GCP's billing export writes ``project.id IS NULL`` for org- # level adjustments (credits, refunds, invoice-level promotions). # Coerce to empty string so the tuple key stays sortable. Note: # the upstream reconciliation pass at ``lifecycle_service.py`` # filters via ``if acct: ...`` (empty string is falsy), so # these org-level rows are **dropped from reconciliation** # (no scope match) — same net effect as the pre-fix path # where the TypeError aborted the whole pass. The point of # this coercion is sortability + crash-safety, NOT routing # org-level cost into a NULL-scope bucket (that's a separate # design conversation). Production incident 2026-05-16: # ``sorted(by_key.items())`` raised ``TypeError: '<' not # supported between instances of 'NoneType' and 'str'`` and # daily reconciliation has been silently failing. row_project_id = r["project_id"] or "" if scope is None or row_project_id == scope.external_id: currency = str(r.get("currency") or "USD") key = (day, row_project_id, currency) entry = by_key.setdefault(key, { "total": 0.0, "list": 0.0, "services": {}, }) entry["total"] += float(r["total_cost"] or 0) if with_list: # 白名单 CASE 在 SQL 里已做行级隔离;组级 NULL # (该 service 组无白名单行,如 Support/Invoice) # 记 0 —— 该项目当日确实没有可计入的业务用量。 entry["list"] += float(r.get("total_cost_list") or 0) entry["services"][r["service_name"]] = ( entry["services"].get(r["service_name"], 0.0) + float(r["total_cost"] or 0) ) return [ CloudDailyCost( date=day, scope=CloudScopeRef( scope_type="gcp_project", external_id=project_id, ), total_cost_usd=entry["total"], # GCP 的牌价就是 cost_at_list 本身 → public 与 list 同值 # (「研发自主节省」对 GCP 恒为 0:Spot/阶梯已体现在 # SKU 牌价里)。缺列降级时两者皆 None。 total_cost_usd_list=( float(entry["list"]) if with_list else None ), total_cost_usd_public=( float(entry["list"]) if with_list else None ), currency=currency, cost_by_service=entry["services"], ) for (day, project_id, currency), entry in sorted(by_key.items()) ] # ── Line-item cost facts (BigQuery export) ─────────────────────── # # GCP's Cloud Billing export to BigQuery is the canonical FinOps # source — one row per (usage_start × resource × SKU). Schema # ref: https://cloud.google.com/billing/docs/how-to/export-data-bigquery-tables # The driver queries the same export the daily-billing path uses # but with full per-resource breakdown: resource.name, sku.id, # sku.description, location.region, plus labels/system_labels # for tag-based allocation. async def list_cost_facts( self, start_date: date, end_date: date, *, scope: CloudScopeRef | None = None, cny_per_usd: float | None = None, principal_tag_keys: list[str] | None = None, bq_billing: dict[str, Any] | None = None, ) -> list[CostFact]: # ``bq_billing`` — see ``list_billing_daily`` rationale. Accept # and ignore (GCP reads its own ``billing_bq_table`` from # integration config; cross-cloud BQ routing is AWS-only). del bq_billing # ``principal_tag_keys`` is AWS-specific. GCP's BigQuery # billing export carries ``identity.user_email`` per row # which would be the closest analogue, but plumbing it # through is out of scope here — accept and ignore. del principal_tag_keys # GCP exports give us ``cost`` (native billing currency) + # ``currency_conversion_rate`` (USD → local at billing time); # the SQL below derives USD via ``SAFE_DIVIDE(cost, rate)``. # The CNY→USD platform rate is irrelevant for GCP's USD # derivation — we accept the kwarg only to match the base- # class signature so the cost-ingestion service can pass it # uniformly to every driver. del cny_per_usd raw = str(self.config.get("billing_bq_table") or "").strip() if not raw: raise RuntimeError( "billing_bq_table is not configured on this GCP " "integration. Set it to a fully-qualified BigQuery " "table id (project.dataset.gcp_billing_export_v1_X). " "Multiple billing accounts: comma-separate (e.g. " "``proj.ds.t1,proj.ds.t2``).", ) # Same parser/validator the daily-billing path uses — the # config string can carry a comma-separated list when an # org has multiple billing accounts (one export table each). # Pre-fix this driver interpolated the raw string into # ``FROM `{table}` `` directly, which made BigQuery treat # ``a,b,c`` as one invalid table id. tables = GcpProvider._split_bq_tables(raw) if not tables: raise RuntimeError( f"billing_bq_table parsed empty from value {raw!r}", ) try: GcpProvider._validate_bq_table_shapes(tables) except ValueError as exc: raise RuntimeError(f"GCP {exc}") from exc return await asyncio.to_thread( self._list_cost_facts_sync, start_date, end_date, scope, tables, ) def _list_cost_facts_sync( self, start: date, end: date, scope: CloudScopeRef | None, tables: list[str], ) -> list[CostFact]: from google.cloud import bigquery # type: ignore # Pin the BQ Client to the billing-export's project (extracted # from the first configured table) rather than the SA's home # project. Pre-fix the client fell back to the SA credentials' # project — when an admin created the SA in a different project # than the billing-export, BQ billed/billed jobs against the # SA project (often without BigQuery API enabled, hard-failing # the job). Tables are already shape-validated by # ``_validate_bq_table_shapes`` before reaching this method. bq_project = tables[0].split(".", 1)[0] if tables else None client = bigquery.Client( credentials=self._credentials(), project=bq_project, ) # Partition window padding: GCP back-dates SUD/CUD credit lines # 30-45 days after the original usage_start_time, but their # ``_PARTITIONTIME`` (ingestion time) lands on the credit-write # day. A nightly cost-facts run with ±1d padding misses every # back-dated credit row whose usage_start_time falls in the # query window but whose _PARTITIONTIME is days later. Widen to # start-3d / end+7d for cost-facts (the monthly path uses ±45d # because it's once-a-month and can afford the wider scan; this # daily path takes the smaller hit of scanning ~10 extra # partitions per run for correctness on credit lines). partition_lo = (start - timedelta(days=3)).isoformat() partition_hi = (end + timedelta(days=8)).isoformat() params: list[Any] = [ bigquery.ScalarQueryParameter("start", "DATE", start), bigquery.ScalarQueryParameter("end", "DATE", end), bigquery.ScalarQueryParameter( "partition_lo", "STRING", partition_lo, ), bigquery.ScalarQueryParameter( "partition_hi", "STRING", partition_hi, ), ] where_extra = "" if scope is not None: where_extra = "AND project.id = @project_id" params.append(bigquery.ScalarQueryParameter( "project_id", "STRING", scope.external_id, )) # Aggregating to (day × resource × sku) keeps the row count # bounded — without the GROUP BY the export can return millions # of micro-transactions per day for big projects, which blows # past the 100MB job result limit. # # USD conversion: GCP's BigQuery billing export does NOT carry # a ``cost_in_usd`` column (that's AWS CUR; an earlier driver # version assumed parity and crashed every cost-facts sync on # 2026-05-10 with ``400 Unrecognized name: cost_in_usd`` across # ``gcp-qa-perf-testing-dev`` / ``gcp-serverless-engine-dev`` / # ``gcp-pm-preview``). The export carries ``cost`` (native # billing currency), ``currency``, and ``currency_conversion_rate`` # — the rate that converts USD → local at billing time # (``local = USD × rate``, so ``USD = local / rate``). For # USD-billed accounts the rate is 1.0 and USD == cost; for # foreign-currency accounts the rate is the historical FX. # ``SAFE_DIVIDE`` returns NULL on a 0 / NULL rate so the row # contributes nothing to the USD sum (the handler tags such # rows with ``amount_usd=0`` + the native currency below) # rather than crashing the whole batch on a bad export row. # # Net-of-credits: ``cost`` is the PRE-credit (gross, list-price) # amount. GCP applies SUD/CUD discounts, committed-use credits, # and promotions as negative ``credits[].amount`` entries inline # on the same row (NOT as separate rows like AWS's # SavingsPlanNegation), so we unnest + sum them into # ``credit_native`` / ``credit_usd`` and the row builder emits # ``amount_native`` / ``amount_usd`` as net = gross + credits. # The gross figure is preserved in ``CostFact.raw`` for invoice # reconciliation. # # Multi-table UNION ALL: orgs with >1 billing account get one # export table per account, all under the same dataset (or # different datasets / projects entirely). Each per-table # SELECT pre-aggregates to bound row count; the outer GROUP # BY merges duplicates that can appear during a billing- # account migration (transient overlap of the same project # across two exports). Tables are interpolated raw because # ``_validate_bq_table_shapes`` was already called by the # async wrapper — refuses anything outside # ``[A-Za-z0-9_-].[A-Za-z0-9_].[A-Za-z0-9_]``. # # Resource-column degradation: the ``resource`` STRUCT only # exists on the *detailed* export (table prefix # ``gcp_billing_export_resource_v1_``). Tenants that only # enabled the standard export hit ``400 Unrecognized name: # resource`` and every per-scope sync failed. We now substitute # empty-string literals on standard tables so the sync still # produces (day × project × service × sku × region) rows; the # per-resource breakdown is simply absent for those rows, which # matches the schema's actual capability. Detection lives on # ``GcpProvider._is_detailed_export``. # GROUP BY notes — every dimension that downstream uses as part # of the cost-fact natural key (or for analytics dis-aggregation) # must appear in BOTH the inner per-table GROUP BY AND the outer # multi-table aggregator's GROUP BY. Two dimensions matter for # correctness and were missing pre-2026-05-10: # # * ``sku_id`` — the cost-fact natural key uses ``usage_type`` # which the row builder (further down) maps to ``sku_id``, # NOT ``sku_description``. ``sku.description`` collapses # cross-region (e.g. "N1 Predefined Instance Core" appears # on every regional SKU); two SKUs sharing one description # would have collapsed to one ``fact_key`` and the upsert # (ON DUPLICATE KEY) would silently overwrite, producing a # cross-region cost undercount on standard-export tenants. # ``sku_description`` is intentionally NOT in GROUP BY: GCP # occasionally renames an SKU mid-window (e.g. "N1 Predefined # Instance Core" → "N1 Predefined Instance Core (Iowa)" for # marketing reasons). Including the description in GROUP BY # would split one SKU's daily total into two rows across the # rename day. ``ANY_VALUE(sku_description)`` keeps the most # recent label without inflating row count. # * ``currency`` — the outer aggregator's ``ANY_VALUE(currency)`` # on a multi-billing-account org with mixed currencies (one # USD billing account + one CNY billing account during a # migration) blended ``native_cost``/``usd_cost`` across # currencies under one arbitrary tag. Adding ``currency`` to # BOTH GROUP BYs keeps each (day × resource × sku × currency) # atomic so analytics by-currency dis-aggregation works and # the row builder doesn't see mis-tagged rows. Mirrors the # daily-billing path's same fix from PR #234. # # Partition predicate: GCP billing exports are exclusively # ingestion-time-partitioned (the column-partitioned variant # doesn't exist for ``gcp_billing_export_*``), so a strict # ``_PARTITIONTIME >= ... AND < ...`` is correct and lets BQ # prune partitions. An earlier defensive ``IS NULL OR`` # variant defeated partition pruning across the whole table, # blowing past the 50 GB ``maximum_bytes_billed`` cap on # multi-year exports. The ±3d / +7d window padding is enough # to capture GCP's back-dated SUD/CUD credit lines (which back- # date 30-45 days but cluster within a 7-day tail of the # original usage_start_time). # # Labels: ``ARRAY_AGG(STRUCT(labels) LIMIT 1)`` was non- # deterministic — BQ may pick a row with NULL/empty labels even # when the same group has populated labels on another row, and # tag-based cost-allocation rules (matching ``fact.tags``) then # turn into a coin flip. Bias toward populated labels via # ``ORDER BY ARRAY_LENGTH(labels) DESC, TO_JSON_STRING(labels) # DESC LIMIT 1``: first key picks the most-populated row, second # key breaks ties deterministically (lexicographic-by-JSON so # repeated cron runs land on the same struct). def _select_for_table(t: str, *, with_list: bool) -> str: # Validate each component of the fully-qualified table name # to prevent SQL injection via user-supplied config values. parts = t.split(".") for part in parts: _validate_bq_name(part, "table component") if GcpProvider._is_detailed_export(t): resource_name_expr = "IFNULL(resource.name, '')" resource_global_expr = "IFNULL(resource.global_name, '')" else: resource_name_expr = "''" resource_global_expr = "''" # 双口径:业务折前成本(list)= cost_at_list 白名单(与 # billing_daily 的 ``_daily_sql`` 完全同一套规则;禁用 cost — # 已折入合同价;credits 完全不参与)。``with_list=False`` = # cost_at_list 缺列降级,list 度量落 NULL。 list_cols = ( """, SUM(CASE WHEN cost_type = 'regular' AND service.description NOT IN ('Support', 'Invoice') AND cost_at_list IS NOT NULL THEN cost_at_list END) AS list_native, SUM(CASE WHEN cost_type = 'regular' AND service.description NOT IN ('Support', 'Invoice') AND cost_at_list IS NOT NULL THEN SAFE_DIVIDE(cost_at_list, currency_conversion_rate) END) AS list_usd""" if with_list else "" ) return f""" SELECT DATE(usage_start_time) AS day, project.id AS project_id, project.name AS project_name, service.description AS service_name, ANY_VALUE(sku.description) AS sku_description, sku.id AS sku_id, {resource_name_expr} AS resource_name, {resource_global_expr} AS resource_global, IFNULL(location.region, '') AS region, IFNULL(currency, '') AS currency, IFNULL(usage.unit, '') AS usage_unit, SUM(cost) AS native_cost, SUM(SAFE_DIVIDE(cost, currency_conversion_rate)) AS usd_cost, SUM((SELECT SUM(c.amount) FROM UNNEST(credits) c)) AS credit_native, SUM(SAFE_DIVIDE( (SELECT SUM(c.amount) FROM UNNEST(credits) c), currency_conversion_rate)) AS credit_usd, SUM(usage.amount) AS usage_amount, ARRAY_AGG(STRUCT(labels) ORDER BY ARRAY_LENGTH(labels) DESC, TO_JSON_STRING(labels) DESC LIMIT 1) AS labels_agg{list_cols} FROM `{t}` WHERE _PARTITIONTIME >= TIMESTAMP(@partition_lo) AND _PARTITIONTIME < TIMESTAMP(@partition_hi) AND DATE(usage_start_time) BETWEEN @start AND @end {where_extra} GROUP BY day, project_id, project_name, service_name, sku_id, resource_name, resource_global, region, currency, usage_unit """ def _facts_sql(*, with_list: bool) -> str: per_table_sql = " UNION ALL ".join( _select_for_table(t, with_list=with_list) for t in tables ) if len(tables) == 1: # Single-table case: skip the outer wrapper entirely so # the explain plan stays identical to the pre-fix shape # for the common (single-billing-account) tenant. return per_table_sql # Outer GROUP BY mirrors the inner SELECT's grain — currency # and usage_unit must be GROUP BY keys (not ANY_VALUE) so a # multi-billing-account org with mixed currencies during a # migration doesn't blend native amounts under one arbitrary # currency tag. Same fix as the daily-billing path. outer_list = ( """, SUM(list_native) AS list_native, SUM(list_usd) AS list_usd""" if with_list else "" ) return f""" SELECT day, project_id, project_name, service_name, ANY_VALUE(sku_description) AS sku_description, sku_id, resource_name, resource_global, region, currency, usage_unit, SUM(native_cost) AS native_cost, SUM(usd_cost) AS usd_cost, SUM(credit_native) AS credit_native, SUM(credit_usd) AS credit_usd, SUM(usage_amount) AS usage_amount, ANY_VALUE(labels_agg) AS labels_agg{outer_list} FROM ( {per_table_sql} ) GROUP BY day, project_id, project_name, service_name, sku_id, resource_name, resource_global, region, currency, usage_unit """ query = _facts_sql(with_list=True) # Defensive cost / runtime caps: a misconfigured tenant with a # year-old window or a corrupted partition predicate could # otherwise issue a runaway BigQuery scan billed to the host # project. 50 GB max bytes is generous for a 7-day cost-facts # window (typical scan is <500 MB) and 5-minute timeout matches # the worker's per-job budget. Both numbers are from the same # ballpark the monthly path already enforces by virtue of its # narrower BillingDate filter. job_cfg = bigquery.QueryJobConfig( query_parameters=params, maximum_bytes_billed=50_000_000_000, job_timeout_ms=300_000, ) # ``cost_at_list`` 缺列(老导出 schema)→ 单次降级重试,list 度量 # 落 NULL;其它错误原样上抛(与 billing_daily 同一策略)。 # credits 明细(第二个查询)与 cost_at_list 无关,不受降级影响。 with_list = True try: try: rows = list(client.query(query, job_config=job_cfg).result()) except Exception as exc: # noqa: BLE001 if "cost_at_list" not in str(exc): raise logger.warning( "GCP cost_facts: export schema lacks cost_at_list " "(%s) — list-cost measure degrades to NULL. Upgrade " "the billing export schema to enable the engineering " "cost view.", exc, ) with_list = False rows = list(client.query( _facts_sql(with_list=False), job_config=job_cfg, ).result()) # ── credits 明细(独立负数 fact 行)──────────────────── # 发票口径的结构化拆分:usage 行改存税前毛额(cost,见下), # credits 按 (day × project × type) 聚合成独立负数行 —— # 每日合计不变(cost + credits),但财务瀑布能按 credit 类型 # (PROMOTION / FEE_UTILIZATION_OFFSET / …)分桶,研发口径 # 读侧天然不受影响(credit 行的 list 恒 NULL)。 credit_sql = " UNION ALL ".join( f""" SELECT DATE(usage_start_time) AS day, project.id AS project_id, ANY_VALUE(project.name) AS project_name, c.type AS credit_type, IFNULL(currency, '') AS currency, SUM(c.amount) AS credit_native, SUM(SAFE_DIVIDE(c.amount, currency_conversion_rate)) AS credit_usd FROM `{t}`, UNNEST(credits) AS c WHERE _PARTITIONTIME >= TIMESTAMP(@partition_lo) AND _PARTITIONTIME < TIMESTAMP(@partition_hi) AND DATE(usage_start_time) BETWEEN @start AND @end {where_extra} GROUP BY day, project_id, credit_type, currency """ for t in tables ) if len(tables) > 1: credit_sql = f""" SELECT day, project_id, ANY_VALUE(project_name) AS project_name, credit_type, currency, SUM(credit_native) AS credit_native, SUM(credit_usd) AS credit_usd FROM ({credit_sql}) GROUP BY day, project_id, credit_type, currency """ credit_rows = list(client.query( credit_sql, job_config=job_cfg, ).result()) finally: # Release the gRPC channel so a per-scope iteration on a # 50-project tenant doesn't leak 50 zombie channels per cron # run. The monthly-billing path already does this; the # cost-facts driver path was the holdout. try: client.close() except Exception: # noqa: BLE001 pass out: list[CostFact] = [] integration_id = self.integration_id for r in rows: # Guard against malformed BQ rows (schema drift, accidental # column rename). Skip the row rather than crashing the batch. if not r.get("day") or not r.get("project_id"): logger.warning( "GCP cost-facts: skipping row missing required fields " "(day=%r project_id=%r)", r.get("day"), r.get("project_id"), ) continue # ``native_cost`` / ``usd_cost`` are the PRE-credit (gross, # list-price) sums — GCP's ``cost`` column is pre-credit. We # resolve gross first, then net the credit lines below so the # emitted ``amount_native`` / ``amount_usd`` are true net cost # and the gross stays inspectable in ``raw``. gross_native = float(r["native_cost"] or 0) # Currency: post-Round-2 fix the SQL emits ``IFNULL(currency, # '')`` so a NULL-currency row arrives as empty string. We # do NOT default empty to "USD" — that was a silent silent- # data-corruption hazard (a row genuinely missing currency # could have its native amount in any currency, and treating # it as USD would silently inflate USD totals). Instead, # treat empty/unknown currency as "UNKNOWN" and route to # ``amount_usd=0``; the row still persists for native- # currency analytics, and the operator sees an explicit # "UNKNOWN" tag in dashboards instead of a falsified "USD". currency_raw = str(r.get("currency") or "").strip().upper() currency = currency_raw or "UNKNOWN" # ``usd_cost`` is the SAFE_DIVIDE(cost, currency_conversion_rate) # aggregate from the SQL above (GCP exports don't carry # a native ``cost_in_usd`` column — see the schema note # at the query). NULL means every row in the group had a # 0 / NULL conversion rate. If the billing currency is # USD we fall through to ``native`` (rate-1.0 equivalence); # otherwise (foreign + bad rate, OR unknown currency) we # leave amount_usd at 0 so the row persists for native- # currency analytics but is excluded from USD aggregates. usd_raw = r["usd_cost"] if usd_raw is not None: gross_usd = float(usd_raw) elif currency == "USD": gross_usd = gross_native else: gross_usd = 0.0 # 2026-07 成本口径改造:usage 行改存 **税前毛额**(cost 列, # PRE-credit)。credits 不再折进本行 —— 它们由上面的第二个 # 查询按 (day × project × type) 聚合成独立负数 fact 行(见 # 循环末尾),发票口径的每日合计不变(= cost + credits), # 但财务瀑布获得 credit 类型分桶,行级金额不再被摊派的 # PROMOTION / FCUD 抵扣冲小(此前 5 月赠金把项目日成本冲成 # 精确 $0 的根因)。原 credit 聚合列保留在本查询里仅供 # ``raw`` 审计。 credit_native = float(r.get("credit_native") or 0) credit_usd_raw = r.get("credit_usd") if credit_usd_raw is not None: credit_usd = float(credit_usd_raw) elif currency == "USD": credit_usd = credit_native else: credit_usd = 0.0 native = gross_native usd = gross_usd # Skip true-zero rows (no gross cost — e.g. a pure-credit # carrier row whose monetary content now lives in the # synthesized credit facts). if native == 0 and usd == 0: continue day = ( r["day"] if isinstance(r["day"], date) else date.fromisoformat(str(r["day"])) ) project_id = r["project_id"] scope_ref = scope if scope is not None else CloudScopeRef( scope_type="gcp_project", external_id=project_id, ) # Labels — BQ export carries them as STRUCT # repeated. ANY_VALUE caps to one row per (day, resource); # we crack the struct here. tag_map: dict[str, str] = {} labels_agg = r.get("labels_agg") or [] if labels_agg: first = labels_agg[0] # BQ structs deserialize as dicts; use isinstance so a # malformed row (string, int, empty list) is skipped # cleanly rather than silently returning wrong data via # getattr (strings have a .get attr via dict-like check). labels_field = first.get("labels") if isinstance( first, dict ) else None if labels_field: for lab in (labels_field if isinstance(labels_field, list) else []): if not isinstance(lab, dict): continue k = lab.get("key") v = lab.get("value") if k: tag_map[str(k)] = str(v) if v is not None else "" resource_name = ( str(r.get("resource_name") or r.get("resource_global") or "") or None ) sku_id = str(r.get("sku_id") or "").strip() sku_description = str(r.get("sku_description") or "").strip() region_str = str(r.get("region") or "").strip() # ``usage_type`` is a natural-key component of ``compute_fact_key`` # (date × integration × scope × resource_id × usage_type). # Pre-fix the driver set ``usage_type=sku_description`` — # but SKU descriptions are NOT unique cross-region (e.g. # "N1 Predefined Instance Core running Americas" appears # on every regional N1 SKU). On standard exports # (``resource_id=None``), two regional SKUs sharing one # description collapsed to one ``fact_key`` and the # ON DUPLICATE KEY upsert silently overwrote — silent # cross-region cost undercount on every standard-export # tenant. Fix uses the SKU ID (globally unique per GCP) # plus the region as the ``usage_type`` string so the # natural key disambiguates correctly. The human-readable # description is preserved in ``resource_type`` (already # surfaced by the dashboard / CSV export). For rows # missing sku_id (extremely rare; fall through to # description-only — better than nothing). if sku_id: usage_type = ( f"{sku_id}@{region_str}" if region_str else sku_id ) else: usage_type = sku_description or None # 折扣前(原价): GCP's pre-credit ``cost`` IS the list price; # SUD/CUD/committed-use/promotion credits are the discount, so # gross = pre-credit, net = gross + credits (credits ≤ 0). Surface # it on ``amount_usd_gross`` so the AI-spending / cost dashboards # render 原价 → 优惠 → 实付 for Vertex the same way Bailian / # AWS Bedrock / Azure Foundry already do. Same honesty guard as # the Azure driver: list ≥ net always, so a gross that came out # below net (a positive "credit" adjustment, FX skew on a bad # rate) is untrustworthy → NULL rather than a wrong, too-low # gross; also meaningless on zero/negative net → NULL there too. gross_usd_value: float | None = ( gross_usd if (usd > 0 and gross_usd >= usd - 1e-6) else None ) gross_native_value: float | None = ( gross_native if (gross_usd_value is not None and gross_native >= native - 1e-6) else None ) # 双口径:list = cost_at_list 白名单(SQL CASE 已做行级 # 隔离;Support / Invoice / 非 regular 组落 NULL)。GCP 的 # 牌价就是 cost_at_list → public 与 list 同值。缺列降级 # (with_list=False)时恒 None。USD 回退规则与 usd_cost 同。 service_str = str(r["service_name"] or "Other") or "Other" list_native_value: float | None = None list_usd_value: float | None = None if with_list: _ln = r.get("list_native") if _ln is not None: list_native_value = float(_ln) _lu = r.get("list_usd") if _lu is not None: list_usd_value = float(_lu) elif currency == "USD": list_usd_value = list_native_value else: list_usd_value = 0.0 if service_str == "Support": bucket = COST_BUCKET_SUPPORT elif service_str == "Invoice": bucket = COST_BUCKET_ADJUSTMENT else: bucket = COST_BUCKET_USAGE out.append(CostFact( date=day, integration_id=integration_id, service=service_str, amount_usd=usd, amount_native=native, amount_usd_gross=gross_usd_value, amount_native_gross=gross_native_value, amount_usd_list=list_usd_value, amount_native_list=list_native_value, amount_usd_public=list_usd_value, cost_bucket=bucket, currency=currency, scope=scope_ref, resource_group=str(r.get("project_name") or project_id), resource_id=resource_name, resource_name=( resource_name.rsplit("/", 1)[-1] if resource_name else None ), resource_type=sku_description or None, region=region_str or None, usage_type=usage_type, tags=tag_map, usage_quantity=( float(r["usage_amount"]) if r.get("usage_amount") is not None else None ), usage_unit=str(r.get("usage_unit") or "") or None, raw={ "sku_id": sku_id, "sku_description": sku_description, "project_id": project_id, # Pre-credit (gross) amounts + the group's inline # credit sums (audit only — the monetary credit rows # are synthesized separately below, per type). "gross_native": gross_native, "gross_usd": gross_usd, "credit_native": credit_native, "credit_usd": credit_usd, }, )) # ── 合成 credit facts(独立负数行,按类型分桶)────────────── # 发票口径合计 = usage 行(税前毛额) + credit 行 = cost + credits # —— 与切换前的净额口径逐日一致。fact_key 天然唯一: # resource_id = ``gcp-project::credit:``,usage_type = # credit type。研发口径不受影响(list 恒 NULL)。 for cr in credit_rows: if not cr.get("day") or cr.get("credit_native") is None: continue c_native = float(cr["credit_native"] or 0) c_currency = ( str(cr.get("currency") or "").strip().upper() or "UNKNOWN" ) c_usd_raw = cr.get("credit_usd") if c_usd_raw is not None: c_usd = float(c_usd_raw) elif c_currency == "USD": c_usd = c_native else: c_usd = 0.0 if c_native == 0 and c_usd == 0: continue c_day = ( cr["day"] if isinstance(cr["day"], date) else date.fromisoformat(str(cr["day"])) ) c_project_id = cr.get("project_id") or "" if scope is not None and c_project_id != scope.external_id: continue c_scope = scope if scope is not None else CloudScopeRef( scope_type="gcp_project", external_id=c_project_id, ) c_type = str(cr.get("credit_type") or "CREDIT_TYPE_UNSPECIFIED") out.append(CostFact( date=c_day, integration_id=integration_id, service="Credits", amount_usd=c_usd, amount_native=c_native, cost_bucket=_GCP_CREDIT_TYPE_BUCKETS.get( c_type, COST_BUCKET_CREDIT, ), currency=c_currency, scope=c_scope, resource_group=str(cr.get("project_name") or c_project_id), resource_id=f"gcp-project:{c_project_id}:credit:{c_type}", resource_name=c_type, resource_type="gcp-credit", usage_type=c_type, raw={"credit_type": c_type, "project_id": c_project_id}, )) return out # ── Resource-creator lookup via Cloud Audit Logs ──────────── # # Cloud Audit Logs queryable via the Logging API # (``logging.googleapis.com/v2/entries:list``). The # ``activity`` log captures Admin Activity events — including # every create/insert call — and the default ``_Default`` log # bucket retains them for **400 days**, far beyond AWS/Aliyun/ # Azure's 90-day window, so most prod resources are still in # range. # # Resource-id formats stored in ``cmdb_cost_facts.resource_id`` # by the GCP driver are full GCP resource paths: # # * Compute Engine: # ``//compute.googleapis.com/projects/

/zones//instances/`` # * GCS bucket: ``gs://`` # * BigQuery table: # ``//bigquery.googleapis.com/projects/

/datasets//tables/`` # # Audit-log filter syntax matches ``protoPayload.resourceName`` # directly, so we just pass the resource_id through (stripping # the leading ``//`` if present — the audit-log resourceName # uses the same path WITHOUT the host prefix sometimes). # # The principal is at # ``protoPayload.authenticationInfo.principalEmail``. _GCP_CREATE_METHOD_PATTERNS = ( # Each entry is a regex fragment ``methodName=~`` match in # the Logging filter. Order doesn't matter — we OR them # together server-side. Keep narrow so non-create events # (Update / Delete / Get) don't pollute results. r"\.insert$", # compute, sql, etc. r"\.create$", # storage, bigquery dataset, pubsub, … r"CreateTable$", r"CreateInstance$", r"CreateCluster$", r"CreateFunction$", r"CreateService$", r"CreateTopic$", r"CreateSubscription$", r"CreateBucket$", r"CreateDataset$", ) # GCP project ID charset: lowercase letter start, then # letters / digits / hyphens, 6-30 chars. Required to validate # the extracted project before interpolating into the Logging # filter string (defence-in-depth — the source resource_id is # owned by the cost-facts ingest path, but a future driver # bug or a hostile config could land malformed values). _GCP_PROJECT_ID_RE = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$") @classmethod def _gcp_extract_project_from_resource_id( cls, resource_id: str, ) -> str | None: """Pull the GCP project id from a resource_id so the audit- log filter can target the right project's ``_Default`` log. Returns ``None`` when the resource_id doesn't embed a project (rare — billing-export rows may use synthetic ids) OR when the extracted token fails the project-id regex (corrupt data — never interpolate unvalidated tokens into the Logging filter string). """ if not resource_id: return None # ``//compute.googleapis.com/projects/X/...`` and # ``//bigquery.googleapis.com/projects/X/...`` shapes. marker = "/projects/" if marker in resource_id: tail = resource_id.split(marker, 1)[1] # Project id ends at next `/` or end of string. candidate = tail.split("/", 1)[0] if candidate and cls._GCP_PROJECT_ID_RE.match(candidate): return candidate return None return None async def get_resource_creator( self, resource_id: str, *, resource_type: str | None = None, scope: CloudScopeRef | None = None, ) -> ResourceCreatorInfo | None: """Resolve "who created this resource" via Cloud Audit Logs. Returns the EARLIEST matching create event for the resource within the audit-log retention window (400 days default). """ from datetime import UTC from datetime import datetime as _dt from datetime import timedelta as _td if not resource_id: return None if resource_id.startswith("gcp-aggregated:"): # Synthetic id from the billing-export driver — no # cloud-side resource to attribute. return None # Determine project for the audit-log filter. project = self._gcp_extract_project_from_resource_id(resource_id) if project is None and scope is not None: candidate = scope.external_id # Same project-id regex as the extractor — scope # external_id originates from the integration config and # should be well-formed, but defence-in-depth. if candidate and self._GCP_PROJECT_ID_RE.match(candidate): project = candidate if not project: return None end_time = _dt.now(UTC) # 400 days — GCP _Default retention. We don't artificially # narrow because the older history is one of GCP's real # advantages over AWS for this use case. start_time = end_time - _td(days=400) return await asyncio.to_thread( self._gcp_audit_lookup_sync, resource_id, project, start_time, end_time, ) def _gcp_audit_lookup_sync( self, resource_id: str, project: str, start_time: datetime, end_time: datetime, ) -> ResourceCreatorInfo | None: """Synchronous Cloud Logging ``entries.list`` call. Filter targets: * ``logName="projects/

/logs/cloudaudit.googleapis.com%2Factivity"`` — Admin Activity log (every create / delete / IAM change). NOT ``data_access`` (read-side audit, off by default) or ``system_event`` (GCP-internal events). * ``timestamp >= ...`` for the window. * ``protoPayload.methodName=~"(\\.insert|\\.create|...)$"`` — narrows to create events. * ``protoPayload.resourceName=""`` — pin to the specific resource. Order ``timestamp ASC`` so the EARLIEST event is first — that's the original creation event. """ from googleapiclient.errors import HttpError as _HttpError svc = build( # type: ignore[misc] "logging", "v2", credentials=self._credentials(), cache_discovery=False, ) # Escape double-quotes in resource_id for the Logging # filter string (rare but possible in user-named buckets / # datasets). The filter uses ``=`` exact match so # backslash-escape is the only special. safe_rid = resource_id.replace("\\", "\\\\").replace('"', '\\"') method_pattern = "|".join( p.replace("\\", "\\\\") for p in self._GCP_CREATE_METHOD_PATTERNS ) filter_str = ( f'logName="projects/{project}/logs/' 'cloudaudit.googleapis.com%2Factivity" AND ' f'timestamp>="{start_time.isoformat()}" AND ' f'timestamp<="{end_time.isoformat()}" AND ' f'protoPayload.resourceName="{safe_rid}" AND ' f'protoPayload.methodName=~"({method_pattern})"' ) body = { "resourceNames": [f"projects/{project}"], "filter": filter_str, "orderBy": "timestamp asc", "pageSize": 10, } try: resp = self._execute_read(svc.entries().list(body=body)) except _HttpError: # 403 on Logging API typically means the SA lacks the # ``logging.privateLogEntries.list`` permission (need # ``roles/logging.privateLogViewer`` for activity log). # Re-raise so the service layer marks the row as # ``error`` and the operator can fix scopes. raise except Exception: raise entries = resp.get("entries") or [] if not entries: return None # Earliest by virtue of orderBy=timestamp asc. entry = entries[0] return self._gcp_audit_entry_to_creator_info(entry) @classmethod def _gcp_audit_entry_to_creator_info( cls, entry: dict[str, Any], ) -> ResourceCreatorInfo: """Parse one Cloud Audit Log entry → ResourceCreatorInfo.""" proto = entry.get("protoPayload") or {} auth = proto.get("authenticationInfo") or {} email = auth.get("principalEmail") if not isinstance(email, str) or "@" not in email: email = None # ``principalSubject`` is the canonical principal identifier # for service accounts + workforce identity pools (more # stable across renames than email). subject = ( auth.get("principalSubject") or auth.get("serviceAccountKeyName") or email or "" ) principal_id = str(subject) display_name: str | None = None if email: local = email.split("@", 1)[0] display_name = local.replace(".", " ").title() or None timestamp = entry.get("timestamp") created_at_cloud: datetime | None = None if isinstance(timestamp, str): try: from datetime import datetime as _dt # Logging API ISO-8601 includes ``Z`` → ``+00:00``. created_at_cloud = _dt.fromisoformat( timestamp.replace("Z", "+00:00"), ) except ValueError: created_at_cloud = None _RAW_CAP_BYTES = 16 * 1024 raw_to_store: dict[str, Any] = { "log_name": entry.get("logName"), "method_name": proto.get("methodName"), "resource_name": proto.get("resourceName"), "timestamp": timestamp, "principal_email": email, "principal_subject": auth.get("principalSubject"), "service_name": proto.get("serviceName"), } # Cap on size. import json as _json if len(_json.dumps(raw_to_store, default=str)) > _RAW_CAP_BYTES: # Drop the most verbose fields if we're over cap. raw_to_store.pop("principal_subject", None) return ResourceCreatorInfo( principal_id=principal_id, email=email, display_name=display_name, created_at_cloud=created_at_cloud, source="gcp_audit", raw=raw_to_store, ) class _NoChangeNeeded(Exception): """Internal signal: project policy didn't contain the member.""" # Capture unused-import warning for optional deps / re-exports kept for # external callers (audit + notification builders import these). _ = (service_account, build, HttpError, timedelta, dataclasses)