e37a27cc1a
- server.py : outils passés en async + déport thread (anyio.to_thread.run_sync). Le SDK mcp 1.27.2 appelle les outils sync directement dans la boucle asyncio, ce qui cassait l'API sync de Playwright. Transport configurable via ANTICOCO_TRANSPORT (défaut streamable-http, stdio pour Claude Code local). - api.py : nouvelle méthode account_info() (client, abonnement, adresse, prochaine livraison) + outil MCP hf_account_info (lecture seule). - auth.py : auth_status() valide désormais le token par un vrai appel API (200 vs 401) au lieu de supposer "token présent = connecté", et n'ouvre plus de navigateur. _is_logged_in() utilise un signal positif (cookie apiV2Auth non expiré) au lieu de l'absence de champ mot de passe. Supprime les faux positifs "connecté" sur session morte (important pour le homelab/Hermes).
339 lines
13 KiB
Python
339 lines
13 KiB
Python
"""Authentification HelloFresh via Playwright + capture du bearer token.
|
|
|
|
Stratégie (cf. plan) :
|
|
- **Login local d'abord** : sur le Mac, fenêtre visible (`headless=False`) → l'utilisateur
|
|
se connecte (captcha/2FA gérés à la main). La session est persistée dans `.session/profile`.
|
|
- **Homelab** : `headless=True`, réutilise la session synchronisée. `HF_EMAIL`/`HF_PASSWORD`
|
|
servent uniquement de fallback de re-login auto si la session a expiré.
|
|
- **Token** : on intercepte l'en-tête `Authorization: Bearer …` envoyé aux hôtes gateway et on
|
|
le met en cache dans `.session/token.json`. `api.py` le réutilise pour les appels httpx.
|
|
|
|
Pattern persistant repris d'`Automood/scraper.py` (`launch_persistent_context`).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import time
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SESSION_DIR = ROOT / ".session"
|
|
PROFILE_DIR = SESSION_DIR / "profile"
|
|
TOKEN_CACHE = SESSION_DIR / "token.json"
|
|
STATE_PATH = SESSION_DIR / "storage_state.json"
|
|
|
|
BASE_URL = "https://www.hellofresh.fr"
|
|
# 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"
|
|
# Endpoint gateway authentifié, léger, utilisé pour VÉRIFIER qu'un token est réellement
|
|
# accepté (200) ou non (401). Vérité de terrain, contre les faux positifs de détection UI.
|
|
VALIDATE_URL = f"{BASE_URL}/gw/api/customers/me/subscriptions"
|
|
|
|
ATTENTE_LOGIN_S = 180 # temps laissé pour un login manuel (captcha / 2FA)
|
|
TOKEN_TTL_S = 30 * 60 # on rafraîchit le token au-delà de 30 min par prudence
|
|
|
|
|
|
def _headless() -> bool:
|
|
return os.environ.get("ANTICOCO_HEADLESS", "1") not in ("0", "false", "False", "")
|
|
|
|
|
|
def _channel() -> str | None:
|
|
"""Canal navigateur : 'chrome' en local (évite les blocages du Chromium bundlé),
|
|
vide/None sur le homelab headless (image Docker = chromium Playwright).
|
|
|
|
Défini par ANTICOCO_BROWSER_CHANNEL ('chrome', 'msedge', ou '' pour chromium).
|
|
"""
|
|
ch = os.environ.get("ANTICOCO_BROWSER_CHANNEL", "")
|
|
return ch or None
|
|
|
|
|
|
def _is_gateway_request(url: str) -> bool:
|
|
return ("/gw/" in url or url.startswith("https://gw.")) and "hellofresh" in url
|
|
|
|
|
|
def _is_logged_in(page) -> bool:
|
|
"""Connecté = cookie d'auth `apiV2Auth` présent avec un access_token non expiré.
|
|
|
|
Signal POSITIF (cookie réel) plutôt que l'absence d'un champ mot de passe : cette
|
|
dernière produisait des faux positifs en headless (page de login non rendue/redirigée,
|
|
bannière cookies) → la session paraissait valide alors qu'elle ne l'était pas.
|
|
"""
|
|
try:
|
|
for c in page.context.cookies():
|
|
if c.get("name") == "apiV2Auth":
|
|
data = json.loads(urllib.parse.unquote(c.get("value", "")))
|
|
tok = data.get("access_token")
|
|
if tok and _jwt_exp(tok) > time.time() + 60:
|
|
return True
|
|
return False
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _auto_login(page) -> bool:
|
|
"""Tente un login automatique avec HF_EMAIL/HF_PASSWORD. Best-effort.
|
|
|
|
Les sélecteurs exacts du formulaire HelloFresh sont à confirmer ; on cible les
|
|
champs standards. En cas d'échec (captcha, sélecteurs changés), renvoie False et
|
|
on retombe sur le login manuel.
|
|
"""
|
|
email = os.environ.get("HF_EMAIL")
|
|
password = os.environ.get("HF_PASSWORD")
|
|
if not email or not password:
|
|
return False
|
|
try:
|
|
page.fill('input[type="email"], input[name="email"], input#email', email, timeout=8000)
|
|
page.fill('input[type="password"], input[name="password"]', password, timeout=8000)
|
|
page.click('button[type="submit"], button[data-test-id="login-submit"]', timeout=8000)
|
|
page.wait_for_timeout(4000)
|
|
return _is_logged_in(page)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _open_context(pw):
|
|
kwargs = dict(
|
|
user_data_dir=str(PROFILE_DIR),
|
|
headless=_headless(),
|
|
locale="fr-FR",
|
|
viewport={"width": 1280, "height": 900},
|
|
)
|
|
channel = _channel()
|
|
if channel:
|
|
kwargs["channel"] = channel
|
|
return pw.chromium.launch_persistent_context(**kwargs)
|
|
|
|
|
|
def ensure_logged_in() -> bool:
|
|
"""Garantit une session connectée dans le profil persistant.
|
|
|
|
- Si déjà connecté : retourne True immédiatement.
|
|
- Sinon, tente l'auto-login (env) ; à défaut attend un login manuel (fenêtre visible).
|
|
Retourne True si la session est établie.
|
|
"""
|
|
SESSION_DIR.mkdir(parents=True, exist_ok=True)
|
|
with sync_playwright() as pw:
|
|
ctx = _open_context(pw)
|
|
try:
|
|
page = ctx.pages[0] if ctx.pages else ctx.new_page()
|
|
page.goto(ACCOUNT_PAGE, wait_until="domcontentloaded", timeout=30000)
|
|
page.wait_for_timeout(2000)
|
|
if _is_logged_in(page):
|
|
return True
|
|
|
|
# Fallback 1 : auto-login si identifiants fournis.
|
|
if _auto_login(page):
|
|
page.wait_for_timeout(2000)
|
|
return True
|
|
|
|
# Fallback 2 : login manuel (uniquement utile en fenêtre visible).
|
|
if _headless():
|
|
raise RuntimeError(
|
|
"Session HelloFresh expirée et auto-login impossible en headless. "
|
|
"Refaire le login en local (ANTICOCO_HEADLESS=0) puis re-sync .session/."
|
|
)
|
|
debut = time.time()
|
|
while time.time() - debut < ATTENTE_LOGIN_S:
|
|
if _is_logged_in(page):
|
|
page.wait_for_timeout(2000)
|
|
return True
|
|
page.wait_for_timeout(2000)
|
|
return False
|
|
finally:
|
|
ctx.close()
|
|
|
|
|
|
def _capture_from_context(ctx, roll_state: bool) -> dict:
|
|
"""Charge le menu dans `ctx` et capture le bearer token (refraîchi par la SPA).
|
|
|
|
`roll_state=True` : ré-exporte storage_state.json après coup, pour faire « rouler »
|
|
la session (le refresh_token tourne ~60 j) sans intervention manuelle.
|
|
"""
|
|
observed = {"token": None, "gateways": set()}
|
|
page = ctx.pages[0] if ctx.pages else ctx.new_page()
|
|
|
|
def on_request(req):
|
|
try:
|
|
if not _is_gateway_request(req.url):
|
|
return
|
|
base = req.url.split("/gw/")[0] + "/gw" if "/gw/" in req.url else req.url
|
|
observed["gateways"].add(base)
|
|
a = req.headers.get("authorization") or req.headers.get("Authorization")
|
|
if a and a.lower().startswith("bearer "):
|
|
observed["token"] = a.split(" ", 1)[1].strip()
|
|
except Exception:
|
|
pass
|
|
|
|
page.on("request", on_request)
|
|
page.goto(MENU_PAGE, wait_until="domcontentloaded", timeout=45000)
|
|
# SPA lourde : on attend que les appels gateway (et un éventuel refresh) partent,
|
|
# plutôt que networkidle qui ne se déclenche jamais.
|
|
for _ in range(15):
|
|
page.wait_for_timeout(1000)
|
|
if observed["token"]:
|
|
page.wait_for_timeout(1000)
|
|
break
|
|
|
|
if not observed["token"]:
|
|
raise RuntimeError(
|
|
"Aucun bearer token capturé (session expirée ? refaire le login + re-sync). "
|
|
)
|
|
|
|
if roll_state:
|
|
try:
|
|
ctx.storage_state(path=str(STATE_PATH))
|
|
except Exception:
|
|
pass
|
|
|
|
result = {
|
|
"token": observed["token"],
|
|
"gateways": sorted(observed["gateways"]),
|
|
"captured_at": time.time(),
|
|
}
|
|
TOKEN_CACHE.write_text(json.dumps(result, indent=2), encoding="utf-8")
|
|
return result
|
|
|
|
|
|
def capture_token(force: bool = False) -> dict:
|
|
"""Renvoie un bearer token frais (et les hôtes gateway), via navigateur.
|
|
|
|
Préfère `storage_state.json` (cookies 60 j) → fonctionne en **headless** sur le homelab :
|
|
la SPA rafraîchit elle-même le token (contourne la protection anti-bot des endpoints
|
|
OAuth bruts). À défaut, retombe sur le profil persistant (login interactif).
|
|
"""
|
|
cached = _read_token_cache()
|
|
if cached and not force and (time.time() - cached.get("captured_at", 0)) < TOKEN_TTL_S:
|
|
return cached
|
|
|
|
SESSION_DIR.mkdir(parents=True, exist_ok=True)
|
|
with sync_playwright() as pw:
|
|
if STATE_PATH.exists():
|
|
launch = {"headless": _headless()}
|
|
if _channel():
|
|
launch["channel"] = _channel()
|
|
browser = pw.chromium.launch(**launch)
|
|
ctx = browser.new_context(storage_state=str(STATE_PATH), locale="fr-FR")
|
|
try:
|
|
return _capture_from_context(ctx, roll_state=True)
|
|
finally:
|
|
ctx.close()
|
|
browser.close()
|
|
ctx = _open_context(pw)
|
|
try:
|
|
return _capture_from_context(ctx, roll_state=False)
|
|
finally:
|
|
ctx.close()
|
|
|
|
|
|
def _read_token_cache() -> dict | None:
|
|
if TOKEN_CACHE.exists():
|
|
try:
|
|
return json.loads(TOKEN_CACHE.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _jwt_exp(token: str) -> float:
|
|
"""Renvoie le timestamp d'expiration d'un JWT (0 si indéterminable)."""
|
|
try:
|
|
payload = token.split(".")[1]
|
|
payload += "=" * (-len(payload) % 4) # padding base64url
|
|
data = json.loads(base64.urlsafe_b64decode(payload))
|
|
return float(data.get("exp", 0))
|
|
except Exception:
|
|
return 0.0
|
|
|
|
|
|
def token_from_storage_state() -> str | None:
|
|
"""Extrait l'access_token du cookie `apiV2Auth` de .session/storage_state.json.
|
|
|
|
Permet de s'authentifier SANS navigateur (utile en headless sur le homelab),
|
|
tant que la session synchronisée n'a pas expiré. Renvoie None si absent/expiré.
|
|
"""
|
|
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"]))
|
|
tok = data.get("access_token")
|
|
if tok and _jwt_exp(tok) > time.time() + 60:
|
|
return tok
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
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.
|
|
"""
|
|
if not force:
|
|
cached = _read_token_cache()
|
|
if cached and (time.time() - cached.get("captured_at", 0)) < TOKEN_TTL_S:
|
|
return cached["token"]
|
|
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.
|
|
if STATE_PATH.exists():
|
|
return capture_token(force=force)["token"]
|
|
ensure_logged_in()
|
|
return capture_token(force=force)["token"]
|
|
|
|
|
|
def _token_works(token: str) -> bool:
|
|
"""Vrai si le token est réellement accepté par l'API gateway (appel léger, 200 vs 401).
|
|
|
|
Vérité de terrain : c'est ce qui empêche de déclarer « connecté » une session morte.
|
|
En cas d'erreur réseau (résultat indéterminable), renvoie False par prudence.
|
|
"""
|
|
try:
|
|
resp = httpx.get(
|
|
VALIDATE_URL,
|
|
params={"country": "FR"},
|
|
headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
|
|
timeout=15.0,
|
|
)
|
|
except Exception:
|
|
return False
|
|
return resp.status_code == 200
|
|
|
|
|
|
def auth_status() -> dict:
|
|
"""État de connexion VÉRIFIÉ par un vrai appel API — pas de faux positif.
|
|
|
|
N'ouvre jamais de navigateur : on prend le meilleur token disponible (cache frais,
|
|
puis cookie storage_state) et on le valide contre l'API. Si aucun token n'est
|
|
exploitable ou s'il est rejeté, renvoie logged_in=False avec une consigne de re-login.
|
|
"""
|
|
cached = _read_token_cache()
|
|
if cached and (time.time() - cached.get("captured_at", 0)) < TOKEN_TTL_S:
|
|
token = cached.get("token")
|
|
if token and _token_works(token):
|
|
age = int(time.time() - cached["captured_at"])
|
|
return {"logged_in": True, "source": "cache", "token_age_s": age,
|
|
"gateways": cached.get("gateways", [])}
|
|
tok = token_from_storage_state()
|
|
if tok and _token_works(tok):
|
|
return {"logged_in": True, "source": "storage_state"}
|
|
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/.",
|
|
}
|