Auth autonome pur HTTP via /gw/refresh (sans navigateur)

Le refresh du token passe désormais par POST /gw/refresh (l'endpoint que la
SPA appelle) au lieu d'un navigateur headless : pur httpx, refresh_token rotaté
persisté dans token.json, fenêtre 60j remise à zéro à chaque refresh. Lock
single-flight pour la rotation. get_token()/auth_status() tentent /gw/refresh
avant le filet Playwright. Homelab allumé = authentifié indéfiniment, sans re-sync.

hellofresh/auth.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 import base64
 import json
 import os
+import threading
 import time
 import urllib.parse
 from pathlib import Path
@@ -30,6 +31,20 @@ TOKEN_CACHE = SESSION_DIR / "token.json"
 STATE_PATH = SESSION_DIR / "storage_state.json"
 
 BASE_URL = "https://www.hellofresh.fr"
+# Passerelle de renouvellement de HelloFresh : `POST {BASE_URL}/gw/refresh` avec
+# `{"refresh_token": ...}` renvoie un bundle frais (access_token JWT + nouveau
+# refresh_token rotaté + refresh_expires_in ~60 j). C'est ce que la SPA appelle ;
+# côté serveur HelloFresh c'est wrappé sur Auth0 (le client_secret reste chez eux,
+# d'où l'impossibilité d'appeler Auth0 en direct). Pur HTTP, pas de navigateur,
+# hors du parcours web protégé par l'anti-bot. Chaque refresh remet la fenêtre à
+# 60 j : un homelab allumé reste authentifié indéfiniment.
+REFRESH_URL = f"{BASE_URL}/gw/refresh"
+
+# Sérialise les renouvellements : les outils MCP tournent dans un worker thread
+# (anyio.to_thread) → appels concurrents possibles. La rotation du refresh_token
+# rendrait fragile un double appel ; le lock sérialise (avec re-test du cache frais
+# après acquisition).
+_refresh_lock = threading.Lock()
 # Page qui déclenche des appels gateway authentifiés (menu de la semaine).
 MENU_PAGE = f"{BASE_URL}/my-account/deliveries/menu"
 ACCOUNT_PAGE = f"{BASE_URL}/my-account"
@@ -274,12 +289,88 @@ def token_from_storage_state() -> str | None:
     return None
 
 
+def _save_token(access_token: str, refresh_token: str | None = None,
+                gateways: list[str] | None = None) -> dict:
+    """Met à jour `.session/token.json` (access token + refresh_token rotatif).
+
+    On préserve les `gateways`/`refresh_token` déjà connus si non fournis, pour ne
+    pas perdre le refresh_token courant lors d'un simple rafraîchissement d'access.
+    """
+    prev = _read_token_cache() or {}
+    result = {
+        "token": access_token,
+        "gateways": gateways if gateways is not None else prev.get("gateways", [f"{BASE_URL}/gw"]),
+        "captured_at": time.time(),
+    }
+    rt = refresh_token or prev.get("refresh_token")
+    if rt:
+        result["refresh_token"] = rt
+    SESSION_DIR.mkdir(parents=True, exist_ok=True)
+    TOKEN_CACHE.write_text(json.dumps(result, indent=2), encoding="utf-8")
+    return result
+
+
+def _refresh_token_from_storage_state() -> str | None:
+    """Extrait le refresh_token du cookie `apiV2Auth` de storage_state.json.
+
+    Sert de bootstrap : c'est le refresh_token déjà présent dans la session capturée,
+    donc aucun nouveau login local n'est requis pour migrer vers le refresh HTTP.
+    """
+    if not STATE_PATH.exists():
+        return None
+    try:
+        state = json.loads(STATE_PATH.read_text(encoding="utf-8"))
+        for c in state.get("cookies", []):
+            if c.get("name") == "apiV2Auth":
+                data = json.loads(urllib.parse.unquote(c["value"]))
+                return data.get("refresh_token") or None
+    except Exception:
+        return None
+    return None
+
+
+def _get_refresh_token() -> str | None:
+    """Refresh token courant : cache token.json (rotations) puis bootstrap cookie."""
+    cached = _read_token_cache()
+    if cached and cached.get("refresh_token"):
+        return cached["refresh_token"]
+    return _refresh_token_from_storage_state()
+
+
+def _refresh_session() -> str | None:
+    """Renouvelle la session via `POST /gw/refresh` (pur HTTP, sans navigateur).
+
+    Persiste le bundle frais : access_token + nouveau refresh_token (rotaté). Renvoie
+    l'access token, ou None si pas de refresh_token disponible / échec réseau ou HTTP.
+    """
+    rt = _get_refresh_token()
+    if not rt:
+        return None
+    try:
+        resp = httpx.post(REFRESH_URL, json={"refresh_token": rt}, timeout=20.0,
+                          headers={"Accept": "application/json"})
+    except Exception as e:
+        print(f"[auth] /gw/refresh réseau KO: {e}")
+        return None
+    if resp.status_code != 200:
+        print(f"[auth] /gw/refresh → HTTP {resp.status_code} ({resp.text[:160]})")
+        return None
+    data = resp.json()
+    access = data.get("access_token")
+    if not access:
+        return None
+    _save_token(access, refresh_token=data.get("refresh_token"))
+    return access
+
+
 def get_token(force: bool = False) -> str:
     """Renvoie un bearer token valide, par ordre de préférence :
 
     1. token en cache encore frais ;
     2. cookie `apiV2Auth` de la session synchronisée (sans navigateur) ;
-    3. capture via navigateur (login si besoin) — impossible en headless si session morte.
+    3. refresh via `/gw/refresh` (pur HTTP, sans navigateur) ;
+    4. capture via navigateur (filet de secours : auto-login HF_EMAIL/HF_PASSWORD
+       ou login interactif) — impossible en headless si la session est morte.
     """
     if not force:
         cached = _read_token_cache()
@@ -288,8 +379,17 @@ def get_token(force: bool = False) -> str:
         tok = token_from_storage_state()
         if tok:
             return tok
-    # Refresh nécessaire : si on a une session synchronisée, le navigateur (même headless)
-    # la rafraîchit ; sinon login interactif via le profil persistant.
+    # Renouvellement nécessaire. On privilégie le HTTP pur, sérialisé pour ne pas
+    # brûler un refresh_token rotatif via deux appels concurrents.
+    with _refresh_lock:
+        cached = _read_token_cache()  # un autre thread a pu rafraîchir entre-temps
+        if cached and (time.time() - cached.get("captured_at", 0)) < TOKEN_TTL_S:
+            return cached["token"]
+        tok = _refresh_session()
+        if tok:
+            return tok
+    # Dernier recours : navigateur. Session synchronisée → refresh headless ;
+    # sinon login interactif via le profil persistant (inutile en headless homelab).
     if STATE_PATH.exists():
         return capture_token(force=force)["token"]
     ensure_logged_in()
@@ -331,8 +431,14 @@ def auth_status() -> dict:
     tok = token_from_storage_state()
     if tok and _token_works(tok):
         return {"logged_in": True, "source": "storage_state"}
+    # Récupération autonome SANS navigateur : refresh via /gw/refresh.
+    with _refresh_lock:
+        tok = _refresh_session()
+        if tok and _token_works(tok):
+            return {"logged_in": True, "source": "gw_refresh"}
     return {
         "logged_in": False,
         "error": "Session HelloFresh absente ou expirée (aucun token valide accepté par l'API). "
-                 "Refaire le login en local (ANTICOCO_HEADLESS=0) puis re-sync .session/.",
+                 "Refaire le login en local (ANTICOCO_HEADLESS=0, auto-login via HF_EMAIL/"
+                 "HF_PASSWORD ou manuel) puis re-sync .session/.",
     }