flowdeck/app/services/oauth_connectors.py

"""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]}