"""Connecteurs OAuth Google / Microsoft 365 (v7.56.0 — phase 6). Flow : ``begin()`` (URL d'autorisation PKCE S256 + state) → callback ``complete()`` (échange du code) → tokens chiffrés Fernet → ``api_get()`` avec refresh automatique. Réutilise : `_encrypt_tokens` / `_decrypt_tokens` (`calendar_sync`, déjà Fernet) et `shared_client` (A42). Les scopes sont **figés en lecture seule**. ponytail : 2 providers décrits par 1 dict (pas de classe par provider) ; les scopes au choix et un écran de connexion dédié arriveront si le besoin se présente — l'état vit dans le catalogue du menu +. """ from __future__ import annotations import base64 import hashlib import logging import secrets import time from urllib.parse import urlencode from app.config import settings from app.db import get_conn from app.services.calendar_sync import _decrypt_tokens, _encrypt_tokens logger = logging.getLogger(__name__) PROVIDERS: dict[str, dict] = { "google": { "name": "Google (Drive · Gmail · Calendar)", "authorize": "https://accounts.google.com/o/oauth2/v2/auth", "token": "https://oauth2.googleapis.com/token", "api": "https://www.googleapis.com", "scopes": [ "openid", "email", "profile", "https://www.googleapis.com/auth/drive.readonly", "https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/calendar.readonly", ], "extra": {"access_type": "offline", "include_granted_scopes": "true"}, "probe_path": "/oauth2/v3/userinfo", }, "ms365": { "name": "Microsoft 365 (Outlook · OneDrive)", "authorize": "https://login.microsoftonline.com/common/oauth2/v2.0/authorize", "token": "https://login.microsoftonline.com/common/oauth2/v2.0/token", "api": "https://graph.microsoft.com/v1.0", "scopes": [ "openid", "email", "profile", "offline_access", "User.Read", "Files.Read", "Mail.Read", "Calendars.Read", "ChannelMessage.Read.All", # Teams (phase 7) — consentement admin ], "extra": {"response_mode": "query", "prompt": "consent"}, "probe_path": "/me", }, } OAUTH_KINDS = tuple(PROVIDERS) # l'access token est renouvelé 60 s avant expiration _REFRESH_SKEW = 60 # ── Config client ─────────────────────────────────────────────────────────── def _client(kind: str) -> tuple[str, str, str]: """(client_id, client_secret, redirect_uri) — lève si non configuré.""" if kind not in PROVIDERS: raise ValueError(f"Connecteur OAuth inconnu: {kind}") if kind == "google": cid, secret = settings.google_client_id, settings.google_client_secret else: cid, secret = settings.ms_client_id, settings.ms_client_secret if not cid or not secret: raise ValueError( f"Connecteur {kind} non configuré (GOOGLE_CLIENT_ID/SECRET ou MS_CLIENT_ID/SECRET)" ) redirect = f"{settings.app_base_url.rstrip('/')}/api/agent/connectors/oauth/{kind}/callback" return cid, secret, redirect def callback_path(kind: str) -> str: return f"/api/agent/connectors/oauth/{kind}/callback" def _pkce_pair() -> tuple[str, str]: verifier = secrets.token_urlsafe(48) digest = hashlib.sha256(verifier.encode("ascii")).digest() return verifier, base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") # ── Flow ──────────────────────────────────────────────────────────────────── def begin(kind: str) -> dict: """URL d'autorisation + state + code_verifier (le route met les cookies).""" cid, _secret, redirect = _client(kind) verifier, challenge = _pkce_pair() state = secrets.token_urlsafe(24) params = { "client_id": cid, "redirect_uri": redirect, "response_type": "code", "scope": " ".join(PROVIDERS[kind]["scopes"]), "state": state, "code_challenge": challenge, "code_challenge_method": "S256", } params.update(PROVIDERS[kind].get("extra", {})) return { "url": f"{PROVIDERS[kind]['authorize']}?{urlencode(params)}", "state": state, "verifier": verifier, "redirect_uri": redirect, } def _normalize(data: dict) -> dict: expires_in = int(data.get("expires_in") or 3600) return { "access_token": str(data.get("access_token") or ""), "refresh_token": str(data.get("refresh_token") or ""), "scope": str(data.get("scope") or ""), "expires_at": int(time.time()) + expires_in, } async def complete(kind: str, code: str, verifier: str) -> dict: """Échange le code contre des tokens (aucun réseau ici hors `_post_form`).""" cid, secret, redirect = _client(kind) data = await _post_form(PROVIDERS[kind]["token"], { "client_id": cid, "client_secret": secret, "grant_type": "authorization_code", "code": code, "redirect_uri": redirect, "code_verifier": verifier, }) tokens = _normalize(data) if not tokens["access_token"]: raise ValueError(str(data.get("error_description") or data.get("error") or "échec de l'échange du code")) return tokens # ── Stockage (chiffré Fernet, comme calendar_sync) ────────────────────────── def save(kind: str, user_id: int, tokens: dict) -> None: with get_conn() as conn: conn.execute( "INSERT INTO connector_tokens (kind, user_id, tokens_enc, updated_at) " "VALUES (?,?,?,CURRENT_TIMESTAMP) " "ON CONFLICT(kind, user_id) DO UPDATE SET " "tokens_enc=excluded.tokens_enc, updated_at=CURRENT_TIMESTAMP", (kind, user_id, _encrypt_tokens(tokens)), ) conn.commit() def tokens(kind: str, user_id: int | None) -> dict: if not user_id: return {} with get_conn() as conn: row = conn.execute( "SELECT tokens_enc FROM connector_tokens WHERE kind=? AND user_id=?", (kind, user_id), ).fetchone() return _decrypt_tokens(row["tokens_enc"]) if row else {} def clear(kind: str, user_id: int) -> bool: with get_conn() as conn: cur = conn.execute( "DELETE FROM connector_tokens WHERE kind=? AND user_id=?", (kind, user_id) ) conn.commit() return cur.rowcount > 0 def is_connected(kind: str, user_id: int | None) -> bool: return bool(tokens(kind, user_id).get("access_token")) # ── Réseau ────────────────────────────────────────────────────────────────── async def _post_form(url: str, data: dict) -> dict: """POST x-www-form-urlencoded vers le token endpoint → dict JSON. Point d'injection des tests (on monkeypatche ``oauth_connectors._post_form``). """ from app.services.http_client import shared_client async with shared_client(timeout=15, headers={"Accept": "application/json"}) as client: resp = await client.post(url, data=data) return resp.json() async def _api_get(url: str, headers: dict) -> tuple[int, str]: """GET d'une API fournisseur — 2ᵉ point d'injection des tests.""" from app.services.http_client import shared_client async with shared_client(timeout=15, headers=headers) as client: resp = await client.get(url) return resp.status_code, (resp.text or "")[:20000] async def access_token(kind: str, user_id: int | None) -> str: """Access token valide, rafraîchi (refresh_token) si nécessaire.""" tok = tokens(kind, user_id) if not tok.get("access_token"): return "" if tok.get("expires_at", 0) > time.time() + _REFRESH_SKEW: return tok["access_token"] if not tok.get("refresh_token"): return "" # expiré sans refresh → à reconnexion try: cid, secret, redirect = _client(kind) data = await _post_form(PROVIDERS[kind]["token"], { "client_id": cid, "client_secret": secret, "grant_type": "refresh_token", "refresh_token": tok["refresh_token"], "redirect_uri": redirect, }) fresh = _normalize(data) if not fresh["access_token"]: raise ValueError(data.get("error_description") or "refresh refusé") # Google ne renvoie parfois pas de nouveau refresh_token : on garde l'ancien fresh["refresh_token"] = fresh["refresh_token"] or tok["refresh_token"] save(kind, int(user_id), fresh) return fresh["access_token"] except Exception as exc: # noqa: BLE001 — jamais d'exception qui casse un run logger.warning("OAuth refresh failed (%s): %s", kind, exc) return "" async def api_get(kind: str, user_id: int | None, path: str = "") -> dict: """GET sur l'API du fournisseur avec le token de l'utilisateur.""" if kind not in PROVIDERS: return {"status": "error", "text": f"Connecteur OAuth inconnu: {kind}"} access = await access_token(kind, user_id) if not access: return {"status": "error", "text": f"{kind} non connecté (lancez la connexion OAuth)"} if "://" in (path or ""): return {"status": "error", "text": "path doit être relatif (pas d'URL absolue)"} # base fixe + chemin : pas d'urljoin (une URL absolue ne peut pas dévier # l'hôte), puis validation SSRF par principe. url = PROVIDERS[kind]["api"].rstrip("/") + "/" + (path or "").lstrip("/") from app.services.importers.url_fetch import _validate_url try: _validate_url(url) except ValueError as exc: return {"status": "error", "text": str(exc)} code, text = await _api_get(url, {"Authorization": f"Bearer {access}"}) if code >= 400: return {"status": "error", "text": f"HTTP {code} — {text[:400]}"} return {"status": "ok", "text": text[:20000]}