"""Azure :class:`AccountIntegrationProvider` — Workload Identity Federation. Federated-first onboarding flow (Tavily research §1, §2): 1. ``bootstrap_artifacts`` — emit a federated-credential JSON skeleton (``issuer``, ``subject``, ``audiences``) keyed to the AlphaSwarm Entra app's ``object_id`` plus a ``az ad app federated-credential create`` snippet. Customer applies this on their tenant. 2. ``validate_identity`` — assemble a ``ClientAssertionCredential`` against the federated credential and call the ``Microsoft Graph /me`` (or equivalent) endpoint to verify the token issuance. 3. ``validate_permissions`` — list role assignments scoped to the supplied subscription and flag whether ``Reader`` (or the ``required_roles`` list passed by the wizard) is granted. 4. ``enumerate_resources`` — list subscriptions + tenants visible to the federated principal. 5. ``connect`` — persist ``{subscription_id, tenant_id, federated_credential_id}`` under ``CredentialKey('cloud_azure', 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__) _AZURE_BASELINE_ROLES: tuple[str, ...] = ("Reader",) class CloudAzureProvider(AccountIntegrationProvider): """Azure Workload Identity Federation onboarding provider.""" kind = IntegrationKind.CLOUD_AZURE def __init__( self, *, store: IntegrationCredentialStore | None = None, settings: AdminSettings | None = None, identity_factory: Any = None, authorization_factory: Any = None, resource_factory: Any = None, ) -> None: self._store = store or get_integration_store() self._settings = settings or get_settings() self._identity_factory = identity_factory # tests inject (tenant_id) -> probe-result self._authorization_factory = authorization_factory # tests inject (subscription_id) -> mgmt client self._resource_factory = resource_factory # tests inject (tenant_id) -> SubscriptionClient # ------------------------------------------------------------------ # Bootstrap # ------------------------------------------------------------------ async def bootstrap_artifacts( self, org_id: str, body: dict[str, Any], ) -> BootstrapArtifacts: client_id = self._settings.cloud_azure_app_client_id.strip() object_id = self._settings.cloud_azure_app_object_id.strip() if not client_id or not object_id: raise IntegrationProviderError( "ALPHASWARM_ADMIN_AZURE_APP_CLIENT_ID / ALPHASWARM_ADMIN_AZURE_APP_OBJECT_ID " "must be configured before the Azure wizard can emit a " "federated-credential template", code="provider_misconfigured", status_code=503, ) audience = self._settings.cloud_azure_audience.strip() or "api://AzureADTokenExchange" # The wizard caller supplies the issuer + subject claim values # they will assert from their IdP; we generate sensible defaults # but never invent the issuer (that's customer-controlled). issuer = str(body.get("issuer") or "https://token.actions.githubusercontent.com").strip() subject = str(body.get("subject") or f"repo:alphaswarm-broker/{org_id}:ref:refs/heads/main").strip() cred_name = str(body.get("credential_name") or f"alphaswarm-{org_id}").strip() federated_credential = { "name": cred_name, "issuer": issuer, "subject": subject, "audiences": [audience], "description": f"AlphaSwarm broker federation for org {org_id}", } cli_command = ( f'az ad app federated-credential create ' f'--id {object_id} ' f"--parameters '{_compact_json(federated_credential)}'" ) portal_url = ( f"https://portal.azure.com/#view/Microsoft_AAD_RegisteredApps" f"/ApplicationMenuBlade/~/Credentials/appId/{client_id}" ) return BootstrapArtifacts( cloud_kind=self.kind, auth_method="workload_identity_federation", external_id=None, templates={"federated_credential.json": _compact_json(federated_credential)}, cli_commands={"federated_credential_create": cli_command}, iac_links={"azure_portal_app_credentials": portal_url}, metadata={ "app_client_id": client_id, "app_object_id": object_id, "audience": audience, "baseline_roles": list(_AZURE_BASELINE_ROLES), }, next_step=( "Apply the federated credential on the customer's Entra app, " "grant the chosen subscription scope (default: Reader), then " "return to step 3 to validate the identity." ), ) # ------------------------------------------------------------------ # Identity probe # ------------------------------------------------------------------ async def validate_identity( self, org_id: str, body: dict[str, Any], ) -> IdentityProbe: tenant_id = str(body.get("tenant_id") or "").strip() if not tenant_id: return IdentityProbe( ok=False, error="tenant_id required", ) subscription_id = str(body.get("subscription_id") or "").strip() if self._identity_factory is not None: try: result = self._identity_factory(tenant_id, subscription_id) if not isinstance(result, dict): return IdentityProbe(ok=False, error="test override returned non-dict") return IdentityProbe( ok=bool(result.get("ok")), error=result.get("error"), identity={ "tenant_id": tenant_id, "subscription_id": subscription_id, "app_client_id": self._settings.cloud_azure_app_client_id, **(result.get("identity") or {}), }, ) except Exception as exc: # noqa: BLE001 return IdentityProbe(ok=False, error=str(exc)) # Real path — acquire a token via the federated credential. We # never log the token; only the resulting claim metadata is # surfaced back to the wizard. azure_identity = require_sdk( self.kind, extras_name="cloud-azure", import_target="azure.identity", ) try: credential = azure_identity.DefaultAzureCredential( exclude_developer_cli_credential=False, ) scope = "https://management.azure.com/.default" token = credential.get_token(scope) except Exception as exc: # noqa: BLE001 return IdentityProbe(ok=False, error=f"Azure identity acquisition failed: {exc}") if not getattr(token, "token", None): return IdentityProbe(ok=False, error="Azure identity returned empty token") return IdentityProbe( ok=True, identity={ "tenant_id": tenant_id, "subscription_id": subscription_id, "app_client_id": self._settings.cloud_azure_app_client_id, "scope": scope, }, ) # ------------------------------------------------------------------ # Permission preview # ------------------------------------------------------------------ async def validate_permissions( self, org_id: str, body: dict[str, Any], ) -> PermissionPreview: subscription_id = str(body.get("subscription_id") or "").strip() if not subscription_id: return PermissionPreview(error="subscription_id required") required_roles = tuple( str(r) for r in (body.get("required_roles") or _AZURE_BASELINE_ROLES) ) if self._authorization_factory is not None: try: result = self._authorization_factory(subscription_id) except Exception as exc: # noqa: BLE001 return PermissionPreview(error=str(exc)) assigned = tuple(str(r) for r in (result or {}).get("assigned_roles") or ()) allowed = tuple(role for role in required_roles if role in assigned) missing = tuple(role for role in required_roles if role not in assigned) return PermissionPreview( allowed=allowed, denied=tuple(), missing_required=missing, metadata={"subscription_id": subscription_id, "assigned_roles": list(assigned)}, ) # Real path — list role assignments scoped to the subscription. mgmt = require_sdk( self.kind, extras_name="cloud-azure", import_target="azure.mgmt.authorization", ) identity = require_sdk( self.kind, extras_name="cloud-azure", import_target="azure.identity", ) try: client = mgmt.AuthorizationManagementClient( credential=identity.DefaultAzureCredential(), subscription_id=subscription_id, ) assignments = list( client.role_assignments.list_for_subscription(filter="atScope()") ) except Exception as exc: # noqa: BLE001 return PermissionPreview(error=f"role_assignments.list_for_subscription failed: {exc}") assigned_role_ids = { str(getattr(a, "role_definition_id", "")).rsplit("/", 1)[-1] for a in assignments } # We cannot resolve role-definition names without an extra # API call per assignment; treat presence of ANY of the # configured baseline ids/aliases as sufficient. allowed = tuple(role for role in required_roles if role in assigned_role_ids) missing = tuple(role for role in required_roles if role not in assigned_role_ids) return PermissionPreview( allowed=allowed, missing_required=missing, metadata={"subscription_id": subscription_id}, ) # ------------------------------------------------------------------ # Enumerate # ------------------------------------------------------------------ async def enumerate_resources( self, org_id: str, body: dict[str, Any], ) -> EnumerationResult: tenant_id = str(body.get("tenant_id") or "").strip() if self._resource_factory is not None: try: result = self._resource_factory(tenant_id or None) except Exception as exc: # noqa: BLE001 return EnumerationResult(error=str(exc)) return EnumerationResult( resources={ "tenants": list((result or {}).get("tenants") or []), "subscriptions": list((result or {}).get("subscriptions") or []), }, ) mgmt_resource = require_sdk( self.kind, extras_name="cloud-azure", import_target="azure.mgmt.resource", ) identity = require_sdk( self.kind, extras_name="cloud-azure", import_target="azure.identity", ) try: client = mgmt_resource.SubscriptionClient(credential=identity.DefaultAzureCredential()) subscriptions = [ { "id": str(getattr(s, "subscription_id", "")), "display_name": str(getattr(s, "display_name", "")), "state": str(getattr(s, "state", "")), } for s in client.subscriptions.list() ] tenants = [ { "id": str(getattr(t, "tenant_id", "")), "default_domain": str(getattr(t, "default_domain", "")), } for t in client.tenants.list() ] except Exception as exc: # noqa: BLE001 return EnumerationResult(error=f"Azure subscription enumeration failed: {exc}") return EnumerationResult(resources={"subscriptions": subscriptions, "tenants": tenants}) # ------------------------------------------------------------------ # Connect # ------------------------------------------------------------------ async def connect( self, org_id: str, body: dict[str, Any], *, bearer: str | None = None, ) -> IntegrationRecord: tenant_id = str(body.get("tenant_id") or "").strip() subscription_id = str(body.get("subscription_id") or "").strip() if not tenant_id or not subscription_id: raise IntegrationProviderError( "tenant_id and subscription_id are 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, ) namespace = str(body.get("namespace") or subscription_id) federated_credential_id = str(body.get("federated_credential_id") or "") row = self._store.put( org_id=org_id, kind=self.kind.value, namespace=namespace, token=subscription_id, # placeholder — no long-lived secret metadata={ "auth_method": "workload_identity_federation", "tenant_id": tenant_id, "subscription_id": subscription_id, "app_client_id": self._settings.cloud_azure_app_client_id, "federated_credential_id": federated_credential_id, }, ) return row_to_record(row, self.kind, status_override=IntegrationStatus.HEALTHY) async def health(self, record: IntegrationRecord) -> IntegrationHealth: tenant_id = str(record.metadata.get("tenant_id") or "") subscription_id = str(record.metadata.get("subscription_id") or "") if not tenant_id or not subscription_id: return IntegrationHealth( status=IntegrationStatus.DISCONNECTED, checked_at=now(), error="tenant_id / subscription_id missing on persisted record", ) probe = await self.validate_identity( record.org_id, {"tenant_id": tenant_id, "subscription_id": subscription_id}, ) 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={"subscription_id": subscription_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__ = ["CloudAzureProvider"]