"""Cloudflare :class:`AccountIntegrationProvider` — scoped API tokens. There is no general-purpose federated identity option for Cloudflare; the federated-first equivalent is **scoped API tokens** (the global API key is deprecated and intentionally rejected by this provider). Onboarding flow (Tavily research §1, §2): 1. ``bootstrap_artifacts`` — surface the token-template UI deep-link plus the scoped permission groups for the requested template (``dns_edit`` | ``tunnel`` | ``access`` | ``worker`` | ``r2``). The customer creates the token in the Cloudflare dashboard and pastes it back into step 5 — AlphaSwarm never sees an unscoped token. 2. ``validate_identity`` — ``GET /user/tokens/verify`` with the supplied token. 3. ``validate_permissions`` — inspect the verified token's ``policies`` against the baseline scopes for the chosen template. 4. ``enumerate_resources`` — list accounts and zones visible to the token. 5. ``connect`` — persist the token (encrypted at rest by the :class:`IntegrationCredentialStore`) plus the resolved account/zone metadata. The token NEVER appears in audit payloads, response bodies, or log lines — only metadata (token actor, expiry, scopes) crosses the BFF boundary. """ from __future__ import annotations import logging from typing import Any import httpx from alphaswarm_admin.providers.base import ( AccountIntegrationProvider, BootstrapArtifacts, EnumerationResult, IdentityProbe, IntegrationHealth, IntegrationKind, IntegrationProviderError, IntegrationRecord, IntegrationStatus, PermissionPreview, now, ) from alphaswarm_admin.providers.cloud_common import 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__) _CF_API_BASE: str = "https://api.cloudflare.com/client/v4" _CF_TIMEOUT: float = 10.0 # Permission groups per template — used by validate_permissions to # surface "missing scope" warnings. The exact group ids are stable # Cloudflare values documented in the API reference. _CF_TEMPLATE_REQUIRED_GROUPS: dict[str, tuple[str, ...]] = { "dns_edit": ("zone.dns_records.edit", "zone.zone.read"), "tunnel": ("account.tunnel.edit", "account.tunnel.read"), "access": ("account.access.edit", "account.access.read"), "worker": ("account.worker.scripts.edit", "account.worker.scripts.read"), "r2": ("account.r2.edit", "account.r2.read"), } class CloudCloudflareProvider(AccountIntegrationProvider): """Cloudflare scoped-API-token onboarding provider.""" kind = IntegrationKind.CLOUD_CLOUDFLARE def __init__( self, *, store: IntegrationCredentialStore | None = None, settings: AdminSettings | None = None, http: httpx.AsyncClient | None = None, ) -> None: self._store = store or get_integration_store() self._settings = settings or get_settings() self._http = http self._owns_http = http is None # ------------------------------------------------------------------ # Bootstrap — surface the template + dashboard deep-link # ------------------------------------------------------------------ async def bootstrap_artifacts( self, org_id: str, body: dict[str, Any], ) -> BootstrapArtifacts: template = str( body.get("template") or self._settings.cloud_cloudflare_token_template ).strip() if template not in _CF_TEMPLATE_REQUIRED_GROUPS: raise IntegrationProviderError( f"unknown Cloudflare token template {template!r}; " f"valid values: {sorted(_CF_TEMPLATE_REQUIRED_GROUPS.keys())}", code="invalid_payload", status_code=422, ) required = _CF_TEMPLATE_REQUIRED_GROUPS[template] return BootstrapArtifacts( cloud_kind=self.kind, auth_method="scoped_api_token", external_id=None, templates={}, cli_commands={}, iac_links={ "create_token_ui": "https://dash.cloudflare.com/profile/api-tokens", }, metadata={ "template": template, "required_permission_groups": list(required), "deprecated_global_api_key": ( "Global API key is deprecated; the wizard accepts " "scoped tokens only." ), }, next_step=( "Create a scoped API token in the Cloudflare dashboard " "matching the template above, then return to step 3 " "with the token to validate it." ), ) # ------------------------------------------------------------------ # Identity probe — /user/tokens/verify # ------------------------------------------------------------------ async def validate_identity( self, org_id: str, body: dict[str, Any], ) -> IdentityProbe: token = str(body.get("token") or "").strip() if not token: return IdentityProbe(ok=False, error="token required") try: payload = await self._cf_get(token, "/user/tokens/verify") except IntegrationProviderError as exc: return IdentityProbe(ok=False, error=str(exc)) result = payload.get("result") if isinstance(payload, dict) else None if not isinstance(result, dict): return IdentityProbe( ok=False, error="Cloudflare /user/tokens/verify returned no result", ) if str(result.get("status") or "").lower() != "active": return IdentityProbe( ok=False, error=f"token status: {result.get('status') or 'unknown'}", ) return IdentityProbe( ok=True, identity={ "token_id": str(result.get("id") or ""), "token_status": str(result.get("status") or ""), "expires_on": str(result.get("expires_on") or ""), "not_before": str(result.get("not_before") or ""), }, ) # ------------------------------------------------------------------ # Permission preview — inspect token policies # ------------------------------------------------------------------ async def validate_permissions( self, org_id: str, body: dict[str, Any], ) -> PermissionPreview: token = str(body.get("token") or "").strip() if not token: return PermissionPreview(error="token required") template = str( body.get("template") or self._settings.cloud_cloudflare_token_template ).strip() required = _CF_TEMPLATE_REQUIRED_GROUPS.get(template, ()) try: payload = await self._cf_get(token, "/user/tokens/verify") except IntegrationProviderError as exc: return PermissionPreview(error=str(exc)) result = payload.get("result") if isinstance(payload, dict) else None if not isinstance(result, dict): return PermissionPreview(error="Cloudflare token verify returned no result") # The verify response carries a "policies" list with # permission_groups when the token was minted via the API; for # dashboard-minted tokens the policies block is missing, so the # absence of policies is NOT a failure — we just report # "unknown scope". Customers can still proceed. policies = result.get("policies") or [] groups: list[str] = [] for policy in policies: if not isinstance(policy, dict): continue for grp in policy.get("permission_groups") or []: name = "" if isinstance(grp, dict): name = str(grp.get("name") or "") if name: groups.append(name) missing = tuple(g for g in required if g not in groups) return PermissionPreview( allowed=tuple(g for g in required if g in groups), denied=tuple(), missing_required=missing, metadata={"template": template, "verified_scopes": groups}, ) # ------------------------------------------------------------------ # Enumerate — accounts + zones # ------------------------------------------------------------------ async def enumerate_resources( self, org_id: str, body: dict[str, Any], ) -> EnumerationResult: token = str(body.get("token") or "").strip() if not token: return EnumerationResult(error="token required") accounts: list[dict[str, Any]] = [] zones: list[dict[str, Any]] = [] try: accounts_payload = await self._cf_get(token, "/accounts") for acct in (accounts_payload or {}).get("result") or []: if isinstance(acct, dict): accounts.append( {"id": str(acct.get("id") or ""), "name": str(acct.get("name") or "")} ) except IntegrationProviderError as exc: return EnumerationResult(error=str(exc)) try: zones_payload = await self._cf_get(token, "/zones?per_page=50") for zone in (zones_payload or {}).get("result") or []: if isinstance(zone, dict): zones.append( { "id": str(zone.get("id") or ""), "name": str(zone.get("name") or ""), "account_id": str( (zone.get("account") or {}).get("id") or "" ), } ) except IntegrationProviderError: # Zone-list permission is template-specific; ignore. logger.debug("Cloudflare zones listing failed (non-fatal)", exc_info=True) return EnumerationResult(resources={"accounts": accounts, "zones": zones}) # ------------------------------------------------------------------ # Connect — persist the token + metadata # ------------------------------------------------------------------ async def connect( self, org_id: str, body: dict[str, Any], *, bearer: str | None = None, ) -> IntegrationRecord: token = str(body.get("token") or "").strip() if not token: raise IntegrationProviderError( "token required", code="invalid_payload", status_code=422, ) 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, ) account_id = str(body.get("account_id") or "").strip() zone_id = str(body.get("zone_id") or "").strip() template = str( body.get("template") or self._settings.cloud_cloudflare_token_template ).strip() namespace = str(body.get("namespace") or account_id or zone_id or "cloudflare") row = self._store.put( org_id=org_id, kind=self.kind.value, namespace=namespace, token=token, metadata={ "auth_method": "scoped_api_token", "token_id": str(probe.identity.get("token_id") or ""), "expires_on": str(probe.identity.get("expires_on") or ""), "template": template, "account_id": account_id, "zone_id": zone_id, }, ) return row_to_record(row, self.kind, status_override=IntegrationStatus.HEALTHY) async def health(self, record: IntegrationRecord) -> IntegrationHealth: token = self._store.reveal_token(record.org_id, self.kind.value) if not token: return IntegrationHealth( status=IntegrationStatus.DISCONNECTED, checked_at=now(), error="credential not found", ) probe = await self.validate_identity(record.org_id, {"token": token}) 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={"token_id": str(probe.identity.get("token_id") or "")}, ) 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] # ------------------------------------------------------------------ # HTTP plumbing # ------------------------------------------------------------------ async def _client(self) -> httpx.AsyncClient: if self._http is None: self._http = httpx.AsyncClient( base_url=_CF_API_BASE, timeout=_CF_TIMEOUT, headers={"Accept": "application/json"}, ) return self._http async def close(self) -> None: if self._owns_http and self._http is not None: await self._http.aclose() self._http = None async def _cf_get(self, token: str, path: str) -> dict[str, Any]: client = await self._client() try: response = await client.get( path, headers={"Authorization": f"Bearer {token}"}, ) except httpx.HTTPError as exc: raise IntegrationProviderError( f"Cloudflare unreachable: {exc}", code="upstream_unreachable", status_code=502, ) from exc if response.status_code in (401, 403): raise IntegrationProviderError( "Cloudflare token rejected", code="unauthorized", status_code=401, ) if response.status_code >= 400: raise IntegrationProviderError( f"Cloudflare GET {path} returned {response.status_code}", code="upstream_5xx", status_code=response.status_code, ) try: payload = response.json() except ValueError as exc: raise IntegrationProviderError( f"Cloudflare GET {path} returned non-JSON", code="invalid_response", status_code=502, ) from exc if isinstance(payload, dict) and payload.get("success") is False: errors = payload.get("errors") or [] message = "; ".join( str((e or {}).get("message") or "") for e in errors if isinstance(e, dict) ) or "request failed" raise IntegrationProviderError( f"Cloudflare API error: {message}", code="upstream_4xx", status_code=response.status_code, ) return payload if isinstance(payload, dict) else {} __all__ = ["CloudCloudflareProvider"]