"""GCP :class:`AccountIntegrationProvider` — Workload Identity Federation. Federated-first onboarding flow (Tavily research §1, §2): 1. ``bootstrap_artifacts`` — emit a Workload Identity Pool + Provider config (issuer URI, attribute mapping, allowed audiences) plus a ``gcloud iam workload-identity-pools providers create-oidc`` snippet. Also emits the ``iam.serviceAccounts.add-iam-policy-binding`` command to grant ``iam.workloadIdentityUser`` on the AlphaSwarm service account. 2. ``validate_identity`` — exchange a federated token for an impersonated service-account access token (via ``iamcredentials.generateAccessToken``) and call ``cloudresourcemanager.projects.get`` to confirm the project is reachable. 3. ``validate_permissions`` — ``testIamPermissions`` on the project for the wizard's baseline permission list. 4. ``enumerate_resources`` — list projects + billing accounts visible to the impersonated SA. 5. ``connect`` — persist ``{project_id, pool_id, provider_id, service_account_email}`` under ``CredentialKey('cloud_gcp', f'org:{org_id}')``. """ from __future__ import annotations import logging from typing import Any from alphaswarm_admin.providers.base import ( AccountIntegrationProvider, BootstrapArtifacts, EnumerationResult, IdentityProbe, IntegrationHealth, IntegrationKind, IntegrationProviderError, IntegrationRecord, IntegrationStatus, PermissionPreview, now, ) from alphaswarm_admin.providers.cloud_common import require_sdk, row_to_record from alphaswarm_admin.services.integration_store import ( IntegrationCredentialStore, get_integration_store, ) from alphaswarm_admin.settings import AdminSettings, get_settings logger = logging.getLogger(__name__) # testIamPermissions baseline — keep aligned with what the wizard's # downstream tooling actually needs (project reads + IAM introspection). _GCP_BASELINE_PERMISSIONS: tuple[str, ...] = ( "resourcemanager.projects.get", "iam.serviceAccounts.actAs", ) class CloudGcpProvider(AccountIntegrationProvider): """GCP Workload Identity Federation onboarding provider.""" kind = IntegrationKind.CLOUD_GCP def __init__( self, *, store: IntegrationCredentialStore | None = None, settings: AdminSettings | None = None, impersonation_factory: Any = None, crm_factory: Any = None, billing_factory: Any = None, ) -> None: self._store = store or get_integration_store() self._settings = settings or get_settings() # tests inject (sa_email) -> {"ok": bool, "identity": {...}, "error": str | None} self._impersonation_factory = impersonation_factory # tests inject (project_id) -> {"project": {...}, "permissions": [...]} self._crm_factory = crm_factory # tests inject () -> {"projects": [...], "billing_accounts": [...]} self._billing_factory = billing_factory # ------------------------------------------------------------------ # Bootstrap # ------------------------------------------------------------------ async def bootstrap_artifacts( self, org_id: str, body: dict[str, Any], ) -> BootstrapArtifacts: audience = self._settings.cloud_gcp_wif_audience.strip() sa_email = self._settings.cloud_gcp_wif_service_account_email.strip() if not audience or not sa_email: raise IntegrationProviderError( "ALPHASWARM_ADMIN_GCP_WIF_AUDIENCE / " "ALPHASWARM_ADMIN_GCP_WIF_SERVICE_ACCOUNT_EMAIL must be configured " "before the GCP wizard can emit a WIF template", code="provider_misconfigured", status_code=503, ) pool_id = str(body.get("pool_id") or f"alphaswarm-broker-{org_id}").strip() provider_id = str(body.get("provider_id") or "alphaswarm-oidc").strip() issuer_uri = str(body.get("issuer_uri") or "https://alpha-swarm.ai/oidc/").strip() project_id = str(body.get("project_id") or "").strip() attribute_mapping = { "google.subject": "assertion.sub", "attribute.org_id": f'"{org_id}"', } pool_config = { "workloadIdentityPool": pool_id, "providerId": provider_id, "issuerUri": issuer_uri, "allowedAudiences": [audience], "attributeMapping": attribute_mapping, } cli_commands: dict[str, str] = {} if project_id: cli_commands["pool_create"] = ( f'gcloud iam workload-identity-pools create {pool_id} ' f'--project={project_id} --location=global ' f'--display-name="AlphaSwarm broker {org_id}"' ) cli_commands["provider_create"] = ( f'gcloud iam workload-identity-pools providers create-oidc ' f'{provider_id} --project={project_id} --location=global ' f'--workload-identity-pool={pool_id} ' f'--issuer-uri={issuer_uri} ' f'--allowed-audiences="{audience}" ' f"--attribute-mapping=\"google.subject=assertion.sub\"" ) cli_commands["sa_binding"] = ( f'gcloud iam service-accounts add-iam-policy-binding ' f'{sa_email} ' f'--project={project_id} ' f'--role=roles/iam.workloadIdentityUser ' f'--member="principalSet://iam.googleapis.com/projects/' f'PROJECT_NUMBER/locations/global/workloadIdentityPools/{pool_id}/*"' ) return BootstrapArtifacts( cloud_kind=self.kind, auth_method="workload_identity_federation", external_id=None, templates={"wif_provider_config.json": _compact_json(pool_config)}, cli_commands=cli_commands, iac_links={ "gcp_console_wif": "https://console.cloud.google.com/iam-admin/workload-identity-pools", }, metadata={ "audience": audience, "service_account_email": sa_email, "pool_id": pool_id, "provider_id": provider_id, "baseline_permissions": list(_GCP_BASELINE_PERMISSIONS), }, next_step=( "Apply the pool, provider, and SA binding on the customer's " "GCP project, note the resulting project id, then return to " "step 3 to validate the identity." ), ) # ------------------------------------------------------------------ # Identity probe # ------------------------------------------------------------------ async def validate_identity( self, org_id: str, body: dict[str, Any], ) -> IdentityProbe: project_id = str(body.get("project_id") or "").strip() if not project_id: return IdentityProbe(ok=False, error="project_id required") sa_email = ( str(body.get("service_account_email") or "").strip() or self._settings.cloud_gcp_wif_service_account_email.strip() ) if self._impersonation_factory is not None: try: result = self._impersonation_factory(sa_email) except Exception as exc: # noqa: BLE001 return IdentityProbe(ok=False, error=str(exc)) return IdentityProbe( ok=bool((result or {}).get("ok")), error=(result or {}).get("error"), identity={ "project_id": project_id, "service_account": sa_email, **((result or {}).get("identity") or {}), }, ) google_auth = require_sdk( self.kind, extras_name="cloud-gcp", import_target="google.auth", ) try: credentials, _ = google_auth.default( # type: ignore[attr-defined] scopes=["https://www.googleapis.com/auth/cloud-platform"], ) credentials.refresh(google_auth.transport.requests.Request()) # type: ignore[attr-defined] except Exception as exc: # noqa: BLE001 return IdentityProbe(ok=False, error=f"google.auth.default failed: {exc}") return IdentityProbe( ok=True, identity={ "project_id": project_id, "service_account": sa_email, "token_type": "impersonated_access_token", }, ) # ------------------------------------------------------------------ # Permission preview — testIamPermissions # ------------------------------------------------------------------ async def validate_permissions( self, org_id: str, body: dict[str, Any], ) -> PermissionPreview: project_id = str(body.get("project_id") or "").strip() if not project_id: return PermissionPreview(error="project_id required") required = tuple( str(p) for p in (body.get("permissions") or _GCP_BASELINE_PERMISSIONS) ) if self._crm_factory is not None: try: result = self._crm_factory(project_id) except Exception as exc: # noqa: BLE001 return PermissionPreview(error=str(exc)) granted = tuple(str(p) for p in (result or {}).get("permissions") or ()) denied = tuple(p for p in required if p not in granted) return PermissionPreview( allowed=tuple(p for p in required if p in granted), denied=denied, missing_required=denied, metadata={"project_id": project_id, "granted": list(granted)}, ) discovery = require_sdk( self.kind, extras_name="cloud-gcp", import_target="googleapiclient.discovery", ) try: crm = discovery.build("cloudresourcemanager", "v1", cache_discovery=False) response = crm.projects().testIamPermissions( resource=project_id, body={"permissions": list(required)} ).execute() except Exception as exc: # noqa: BLE001 return PermissionPreview(error=f"testIamPermissions failed: {exc}") granted = tuple(str(p) for p in response.get("permissions") or ()) denied = tuple(p for p in required if p not in granted) return PermissionPreview( allowed=tuple(p for p in required if p in granted), denied=denied, missing_required=denied, metadata={"project_id": project_id}, ) # ------------------------------------------------------------------ # Enumerate # ------------------------------------------------------------------ async def enumerate_resources( self, org_id: str, body: dict[str, Any], ) -> EnumerationResult: if self._billing_factory is not None: try: result = self._billing_factory() except Exception as exc: # noqa: BLE001 return EnumerationResult(error=str(exc)) return EnumerationResult( resources={ "projects": list((result or {}).get("projects") or []), "billing_accounts": list((result or {}).get("billing_accounts") or []), }, ) discovery = require_sdk( self.kind, extras_name="cloud-gcp", import_target="googleapiclient.discovery", ) try: crm = discovery.build("cloudresourcemanager", "v1", cache_discovery=False) projects = [ { "id": str(p.get("projectId") or ""), "name": str(p.get("name") or ""), "number": str(p.get("projectNumber") or ""), } for p in (crm.projects().list().execute().get("projects") or []) ] except Exception as exc: # noqa: BLE001 return EnumerationResult(error=f"projects.list failed: {exc}") billing_accounts: list[dict[str, Any]] = [] try: billing = discovery.build("cloudbilling", "v1", cache_discovery=False) billing_accounts = [ { "id": str(b.get("name") or "").rsplit("/", 1)[-1], "display_name": str(b.get("displayName") or ""), "open": bool(b.get("open")), } for b in (billing.billingAccounts().list().execute().get("billingAccounts") or []) ] except Exception: # noqa: BLE001 — billing surface is optional logger.debug("billingAccounts.list failed (non-fatal)", exc_info=True) return EnumerationResult( resources={"projects": projects, "billing_accounts": billing_accounts}, ) # ------------------------------------------------------------------ # Connect # ------------------------------------------------------------------ async def connect( self, org_id: str, body: dict[str, Any], *, bearer: str | None = None, ) -> IntegrationRecord: project_id = str(body.get("project_id") or "").strip() if not project_id: raise IntegrationProviderError( "project_id required", code="invalid_payload", status_code=422, ) sa_email = ( str(body.get("service_account_email") or "").strip() or self._settings.cloud_gcp_wif_service_account_email.strip() ) probe = await self.validate_identity(org_id, body) if not probe.ok: raise IntegrationProviderError( probe.error or "identity probe failed", code="unauthorized", status_code=401, ) namespace = str(body.get("namespace") or project_id) pool_id = str(body.get("pool_id") or "") provider_id = str(body.get("provider_id") or "") row = self._store.put( org_id=org_id, kind=self.kind.value, namespace=namespace, token=project_id, # placeholder — no long-lived secret metadata={ "auth_method": "workload_identity_federation", "project_id": project_id, "service_account_email": sa_email, "pool_id": pool_id, "provider_id": provider_id, "audience": self._settings.cloud_gcp_wif_audience, }, ) return row_to_record(row, self.kind, status_override=IntegrationStatus.HEALTHY) async def health(self, record: IntegrationRecord) -> IntegrationHealth: project_id = str(record.metadata.get("project_id") or "") sa_email = str(record.metadata.get("service_account_email") or "") if not project_id: return IntegrationHealth( status=IntegrationStatus.DISCONNECTED, checked_at=now(), error="project_id missing on persisted record", ) probe = await self.validate_identity( record.org_id, {"project_id": project_id, "service_account_email": sa_email}, ) next_status = IntegrationStatus.HEALTHY if probe.ok else IntegrationStatus.DEGRADED self._store.update_health( org_id=record.org_id, kind=self.kind.value, status=next_status.value, error=probe.error, ) return IntegrationHealth( status=next_status, checked_at=now(), error=probe.error, metadata={"project_id": project_id}, ) async def disconnect( self, record: IntegrationRecord, *, bearer: str | None = None, ) -> None: self._store.delete(org_id=record.org_id, kind=self.kind.value) async def list_for_org(self, org_id: str) -> list[IntegrationRecord]: rows = [ row for row in self._store.list_for_org(org_id) if row.kind == self.kind.value ] return [row_to_record(row, self.kind) for row in rows] def _compact_json(payload: dict[str, Any]) -> str: import json return json.dumps(payload, separators=(", ", ": "), sort_keys=True) __all__ = ["CloudGcpProvider"]