AntiCoco: serveur MCP HelloFresh sans noix de coco
- Auth Playwright (login local, session persistee, capture du bearer token) - Client httpx vers l'API interne (endpoints via discover_api.py) - Filtre d'exclusion insensible aux accents (coco & co) - Serveur FastMCP (streamable-http) + outils hf_* - Docker + compose pour deploiement homelab
This commit is contained in:
204
hellofresh/auth.py
Normal file
204
hellofresh/auth.py
Normal file
@@ -0,0 +1,204 @@
|
||||
"""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 json
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
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"
|
||||
|
||||
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-menu"
|
||||
ACCOUNT_PAGE = f"{BASE_URL}/my-account"
|
||||
|
||||
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 _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:
|
||||
"""Pas connecté = un champ mot de passe est visible (page de login).
|
||||
|
||||
Détection volontairement indépendante de la locale (sélecteur CSS, pas de texte).
|
||||
"""
|
||||
try:
|
||||
return page.locator('input[type="password"]').count() == 0
|
||||
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):
|
||||
return pw.chromium.launch_persistent_context(
|
||||
user_data_dir=str(PROFILE_DIR),
|
||||
headless=_headless(),
|
||||
locale="fr-FR",
|
||||
viewport={"width": 1280, "height": 900},
|
||||
)
|
||||
|
||||
|
||||
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_token(force: bool = False) -> dict:
|
||||
"""Capture (ou relit depuis le cache) le bearer token et les hôtes gateway observés.
|
||||
|
||||
Retourne {"token": str, "gateways": [str], "captured_at": float}.
|
||||
"""
|
||||
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)
|
||||
observed = {"token": None, "gateways": set()}
|
||||
|
||||
with sync_playwright() as pw:
|
||||
ctx = _open_context(pw)
|
||||
try:
|
||||
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)
|
||||
auth = req.headers.get("authorization") or req.headers.get("Authorization")
|
||||
if auth and auth.lower().startswith("bearer "):
|
||||
observed["token"] = auth.split(" ", 1)[1].strip()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
page.on("request", on_request)
|
||||
page.goto(MENU_PAGE, wait_until="networkidle", timeout=45000)
|
||||
page.wait_for_timeout(3000)
|
||||
|
||||
if not _is_logged_in(page):
|
||||
raise RuntimeError(
|
||||
"Non connecté lors de la capture du token. Lancer ensure_logged_in() d'abord."
|
||||
)
|
||||
if not observed["token"]:
|
||||
raise RuntimeError(
|
||||
"Aucun bearer token capturé. Vérifier MENU_PAGE / le pattern gateway, "
|
||||
"ou rejouer tools/discover_api.py."
|
||||
)
|
||||
|
||||
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
|
||||
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 get_token(force: bool = False) -> str:
|
||||
"""Renvoie un bearer token valide, en s'assurant d'être connecté au préalable."""
|
||||
ensure_logged_in()
|
||||
return capture_token(force=force)["token"]
|
||||
|
||||
|
||||
def auth_status() -> dict:
|
||||
"""État de connexion sans ouvrir de fenêtre si un token en cache est encore frais."""
|
||||
cached = _read_token_cache()
|
||||
if cached and (time.time() - cached.get("captured_at", 0)) < TOKEN_TTL_S:
|
||||
age = int(time.time() - cached["captured_at"])
|
||||
return {"logged_in": True, "source": "cache", "token_age_s": age, "gateways": cached.get("gateways", [])}
|
||||
try:
|
||||
ok = ensure_logged_in()
|
||||
return {"logged_in": bool(ok), "source": "browser"}
|
||||
except Exception as e:
|
||||
return {"logged_in": False, "error": str(e)}
|
||||
Reference in New Issue
Block a user