"""Provider authorization state machine. Core owns custody and scope truth.""" import base64 import hashlib import hmac import json import secrets from datetime import UTC, datetime, timedelta from urllib.parse import urlencode, urlsplit from uuid import uuid4 from src.config import get_settings from src.core.exceptions import NotFoundError, ValidationError from src.core.security.encryption import decrypt_value, encrypt_value from src.domain.provider_auth.manifests import ( FAMILY_SCOPES, HUBSPOT_FAMILY_SCOPES, MICROSOFT_ADS_FAMILY_SCOPES, SALESFORCE_FAMILY_SCOPES, X_ADS_FAMILY_SCOPES, normalize_scopes, ) from src.domain.provider_auth.models import AuthorizationSession, ProviderConnection, ProviderGrant from src.domain.provider_auth.schemas import AuthorizationCompleted, AuthorizationStarted from src.integrations.google_oauth import GoogleOAuthClient, GoogleOAuthError from src.integrations.hubspot_oauth import HubSpotOAuthClient, HubSpotOAuthError from src.integrations.microsoft_ads_oauth import ( MicrosoftAdsOAuthClient, MicrosoftAdsOAuthError, ) from src.integrations.salesforce_oauth import SalesforceOAuthClient, SalesforceOAuthError from src.integrations.x_ads_oauth import XAdsOAuthClient, XAdsOAuthError OAUTH_CLIENTS = { "google": GoogleOAuthClient, "hubspot": HubSpotOAuthClient, "salesforce": SalesforceOAuthClient, "x_ads": XAdsOAuthClient, "microsoft_ads": MicrosoftAdsOAuthClient, } OAUTH_ERRORS = ( GoogleOAuthError, HubSpotOAuthError, SalesforceOAuthError, XAdsOAuthError, MicrosoftAdsOAuthError, ) OAUTH_FAMILIES = { "google": FAMILY_SCOPES, "hubspot": HUBSPOT_FAMILY_SCOPES, "salesforce": SALESFORCE_FAMILY_SCOPES, "x_ads": X_ADS_FAMILY_SCOPES, "microsoft_ads": MICROSOFT_ADS_FAMILY_SCOPES, } def digest(value): return hashlib.sha256(value.encode()).hexdigest() def safe_return_to(value): # Dashboard-owned destinations only. Generated return targets are separately # authenticated by the identity completion flow, never supplied here. parsed = urlsplit(value) if ( parsed.scheme or parsed.netloc or not value.startswith("/") or value.startswith("//") or "\\" in value or any(ord(c) < 32 for c in value) or parsed.path not in {"/connectors", "/settings/integrations"} ): raise ValidationError("Invalid authorization return destination") return value class ProviderAuthService: def __init__(self, repo, google=None, settings=None, provider="google"): self.repo = repo self.provider = provider self.google = google or GoogleOAuthClient() self.client = google or OAUTH_CLIENTS[provider]() self.settings = settings or get_settings() async def start(self, *, workspace_id, project_id, initiated_by, body): cfg = self.settings prefix = f"{self.provider}_platform_" client_id = getattr(cfg, prefix + "client_id") client_secret = getattr(cfg, prefix + "client_secret") enabled_families = getattr(cfg, prefix + "enabled_families") redirect_uri = getattr(cfg, prefix + "redirect_uri") ttl = getattr(cfg, prefix + "session_ttl_seconds") if not client_id or not client_secret: raise ValidationError( f"{self.provider.title()} platform authorization is not configured" ) if body.grant_family not in enabled_families.split(","): raise ValidationError(f"{self.provider.title()} grant family is not enabled yet") connection = None if body.connection_id: connection = await self.repo.connection(workspace_id, project_id, body.connection_id) if ( not connection or connection.actor_type != "project" or connection.actor_id != str(project_id) ): raise NotFoundError("Provider connection not found") nonce = secrets.token_urlsafe(32) signature = hmac.new(cfg.secret_key.encode(), nonce.encode(), hashlib.sha256).hexdigest() state = nonce + "." + signature verifier = secrets.token_urlsafe(48) challenge = ( base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()) .decode() .rstrip("=") ) now = datetime.now(UTC) row = AuthorizationSession( id=uuid4(), workspace_id=workspace_id, project_id=project_id, actor_type="project", actor_id=str(project_id), initiated_by=str(initiated_by), purpose="authorization", family=body.grant_family, connection_id=connection.id if connection else None, state_hash=digest(state), browser_binding_hash=digest(body.browser_binding), verifier_enc=encrypt_value(verifier), redirect_uri=redirect_uri, return_to=safe_return_to(body.return_to), status="pending", created_at=now, expires_at=now + timedelta(seconds=ttl), ) await self.repo.add(row) families = OAUTH_FAMILIES[self.provider] identity = list(FAMILY_SCOPES["identity"]) if self.provider == "google" else [] scopes = normalize_scopes([*identity, *families[body.grant_family]]) params = dict( client_id=client_id, redirect_uri=row.redirect_uri, response_type="code", scope=" ".join(scopes), state=state, access_type="offline", include_granted_scopes="true", code_challenge=challenge, code_challenge_method="S256", ) if connection: params["login_hint"] = connection.subject else: params["prompt"] = "consent" authorize_base = ( "https://accounts.google.com/o/oauth2/v2/auth" if self.provider == "google" else "https://app.hubspot.com/oauth/authorize" if self.provider == "hubspot" else "https://login.salesforce.com/services/oauth2/authorize" if self.provider == "salesforce" else "https://twitter.com/i/oauth2/authorize" if self.provider == "x_ads" else "https://login.microsoftonline.com/common/oauth2/v2.0/authorize" ) return AuthorizationStarted( authorize_url=authorize_base + "?" + urlencode(params), resume_id=row.id, ) async def complete(self, body): nonce, _, signature = body.state.partition(".") expected = hmac.new( self.settings.secret_key.encode(), nonce.encode(), hashlib.sha256 ).hexdigest() if not hmac.compare_digest(signature, expected): raise ValidationError("Invalid authorization state") session = await self.repo.session_by_state(digest(body.state)) now = datetime.now(UTC) if ( not session or session.status != "pending" or session.expires_at <= now or not hmac.compare_digest(session.browser_binding_hash, digest(body.browser_binding)) ): raise ValidationError("Authorization session is invalid or expired") # Persist consumption BEFORE provider I/O. An uncertain exchange must never # replay the same code; the user can start a fresh session after failure. session.status = "exchanging" await self.repo.db.commit() if body.error or not body.code: session.status = ( "reauthentication_required" if body.error in {"interaction_required", "login_required"} else "denied" ) else: try: token = await self.client.token( grant_type="authorization_code", code=body.code, client_id=getattr(self.settings, f"{self.provider}_platform_client_id"), client_secret=getattr(self.settings, f"{self.provider}_platform_client_secret"), redirect_uri=session.redirect_uri, code_verifier=decrypt_value(session.verifier_enc), ) user = await self.client.userinfo(token) if session.purpose == "identity": context = json.loads(decrypt_value(session.context_enc)) context["identity"] = { k: user[k] for k in ("sub", "email", "name", "picture") if k in user } session.context_enc = encrypt_value(json.dumps(context)) session.status = "completed" # Identity token sets are deliberately discarded here. else: await self._apply_grant(session, token, user, now) except OAUTH_ERRORS as exc: session.status = ( "reauthentication_required" if exc.code in {"invalid_grant", "interaction_required", "login_required"} else "failed" ) session.verifier_enc = "" await self.repo.db.flush() if session.purpose == "identity": completion = secrets.token_urlsafe(32) session.completion_hash = digest(completion) session.expires_at = now + timedelta(minutes=2) context = json.loads(decrypt_value(session.context_enc)) return AuthorizationCompleted( purpose="identity", return_to=context["completion_url"] + "?" + urlencode({"ticket": completion}), resume_id=session.id, status=session.status, ) return AuthorizationCompleted( return_to=session.return_to, resume_id=session.id, status=session.status ) async def _apply_grant(self, session, token, user, now): connection = ( await self.repo.connection( session.workspace_id, session.project_id, session.connection_id ) if session.connection_id else None ) if connection and connection.subject != user["sub"]: session.status = "account_mismatch" return if session.connection_id and not connection: session.status = "failed" return if not connection: connection = await self.repo.by_subject(session, user["sub"]) if not connection: connection = await self.repo.add( ProviderConnection( id=uuid4(), workspace_id=session.workspace_id, project_id=session.project_id, provider=self.provider, actor_type=session.actor_type, actor_id=session.actor_id, subject=user["sub"], status="active", effective_scopes=[], secrets_enc={}, ) ) # Missing scope data is never proof of consent. Only scopes actually returned # by Google contribute to logical grants and execution bindings. granted = set(normalize_scopes(token.get("scope", "").split())) credentials = dict(connection.secrets_enc or {}) for key in ("access_token", "refresh_token"): if token.get(key): credentials[key] = encrypt_value(token[key]) if token.get("instance_url"): credentials["instance_url"] = encrypt_value(token["instance_url"]) if user.get("portal_id"): credentials["portal_id"] = encrypt_value(user["portal_id"]) connection.secrets_enc = credentials connection.expires_at = now + timedelta(seconds=int(token.get("expires_in", 0))) connection.status = "active" grants = await self.repo.grants(connection.id) target = next((g for g in grants if g.family == session.family), None) if target is None: target = ProviderGrant( id=uuid4(), connection_id=connection.id, family=session.family, scopes=[], status="pending", granted_at=now, ) await self.repo.add(target) grants.append(target) for grant in grants: families = OAUTH_FAMILIES[self.provider] required = set(families[grant.family]) grant.scopes = sorted(required & granted) grant.status = "active" if required <= granted else "scope_escalation_pending" grant.granted_at = now grant.revoked_at = None if grant.status == "active" else now connection.effective_scopes = sorted( {scope for grant in grants if grant.status == "active" for scope in grant.scopes} ) session.connection_id = connection.id session.status = target.status if target.status != "active" else "completed"