Cantoche/toqla_auth.py

#!/usr/bin/env python3
"""
Shared authentication helpers for the Toqla/Sodexo API.

Handles OAuth2 PKCE (Proof Key for Code Exchange) with Auth0 session cookies
to silently obtain an access token — no browser interaction needed.

Provide Auth0 session cookies via:
  --cookies "did=...; auth0=..."
or COOKIES env var, or cookies.txt file.

To get your cookies:
  1. Open https://app.toqla.fr and log in.
  2. DevTools (F12) > Network > any request to myid.sodexo.com
  3. Copy the "Cookie" header value (did=...; auth0=...)

The refresh_token obtained from Auth0 is saved to refresh_token.txt and reused
on subsequent runs (no cookies needed) until it expires. This module is shared
by fetch_menus.py and fetch_tickets.py so both scripts reuse the same cached
refresh token / cookies file.
"""

import base64
import hashlib
import logging
import os
import re
import secrets
import sys
from pathlib import Path
from urllib.parse import urlparse, parse_qs, unquote

import requests

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
BASE_URL           = "https://zeus.helium.sodexo.com"
API_KEY            = "94770481-bd9c-4013-96c9-9b85d0759ee2"
AUTH0_DOMAIN       = "https://myid.sodexo.com"
AUTH0_CLIENT_ID    = "cnEsGucenOXKbw8XBJpKhitMB0JUiojz"
AUTH0_REDIRECT_URI  = "https://app.toqla.fr/oidc/callback"
AUTH0_AUDIENCE     = "https://api.pluxee.app/fr/consumer_prd/consumer-bff/api/"
AUTH0_SCOPE        = (
    "openid offline_access profile email phone "
    "https://api.pluxee.app/fr/consumer_prd/consumer-bff/api/scopes/consumer:bff"
)
REFRESH_TOKEN_FILE  = "refresh_token.txt"
COOKIES_FILE        = "cookies.txt"

log = logging.getLogger(__name__)

_COMMON_HEADERS = {
    "accept": "*/*",
    "content-type": "application/x-www-form-urlencoded",
    "origin": "https://app.toqla.fr",
    "referer": "https://app.toqla.fr/",
    "auth0-client": "eyJuYW1lIjoiYXV0aDAtcmVhY3QiLCJ2ZXJzaW9uIjoiMi4xNi4yIn0=",
    "user-agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36"
    ),
}


# ---------------------------------------------------------------------------
# Cookie handling
# ---------------------------------------------------------------------------
def parse_cookie_string(cookie_str: str) -> dict:
    """Parse a cookie header string into a dict {name: value}."""
    cookies = {}
    for part in cookie_str.split(";"):
        part = part.strip()
        if "=" in part:
            k, v = part.split("=", 1)
            cookies[k.strip()] = v.strip()
    return cookies


def load_cookies(args_cookies: str | None, cookies_file: str = COOKIES_FILE) -> dict:
    """Load cookies from args, env var, or a cookies file."""
    # Priority: --cookies arg > COOKIES env > cookies.txt file
    raw = args_cookies or os.environ.get("COOKIES")
    if raw:
        return parse_cookie_string(raw)

    try:
        raw = Path(cookies_file).read_text(encoding="utf-8").strip()
        if raw:
            log.info("Loaded cookies from %s.", cookies_file)
            return parse_cookie_string(raw)
    except FileNotFoundError:
        pass

    return {}


# ---------------------------------------------------------------------------
# PKCE
# ---------------------------------------------------------------------------
def generate_pkce_pair() -> tuple[str, str]:
    """Generate a PKCE code_verifier and code_challenge (S256)."""
    verifier = secrets.token_urlsafe(64)[:128]
    digest = hashlib.sha256(verifier.encode("ascii")).digest()
    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
    return verifier, challenge


# ---------------------------------------------------------------------------
# Token acquisition
# ---------------------------------------------------------------------------
def silent_auth_with_cookies(cookies: dict) -> str:
    """Use Auth0 session cookies + PKCE to silently obtain an access token.

    Flow (same as the browser does with prompt=none):
      1. Generate PKCE code_verifier + code_challenge
      2. GET /authorize with cookies + PKCE (response_mode=web_message)
      3. Extract authorization code from the HTML response
      4. POST /oauth/token with grant_type=authorization_code + code_verifier

    Returns the access_token string.
    Writes the new refresh token back to refresh_token.txt.
    """
    if not cookies:
        log.error(
            "No Auth0 session cookies available. Provide them via "
            "--cookies, COOKIES env var, or cookies.txt file."
        )
        sys.exit(1)

    verifier, challenge = generate_pkce_pair()
    state = secrets.token_urlsafe(16)
    nonce = secrets.token_urlsafe(16)

    # Step 1: GET /authorize with cookies (silent, prompt=none)
    auth_url = (
        f"{AUTH0_DOMAIN}/authorize"
        f"?client_id={AUTH0_CLIENT_ID}"
        f"&scope={AUTH0_SCOPE.replace(' ', '+')}"
        f"&audience={AUTH0_AUDIENCE}"
        f"&redirect_uri={AUTH0_REDIRECT_URI}"
        f"&response_type=code"
        f"&prompt=none"
        f"&response_mode=web_message"
        f"&state={state}"
        f"&nonce={nonce}"
        f"&code_challenge={challenge}"
        f"&code_challenge_method=S256"
        f"&auth0Client=eyJuYW1lIjoiYXV0aDAtcmVhY3QiLCJ2ZXJzaW9uIjoiMi4xNi4yIn0%3D"
    )

    log.info("Silent authorization via Auth0 with session cookies…")
    resp = requests.get(
        auth_url,
        headers={
            "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "accept-language": "en-GB,en;q=0.9,en-US;q=0.8,fr;q=0.7",
            "origin": "https://app.toqla.fr",
            "referer": "https://app.toqla.fr/",
            "auth0-client": "eyJuYW1lIjoiYXV0aDAtcmVhY3QiLCJ2ZXJzaW9uIjoiMi4xNi4yIn0=",
            "user-agent": _COMMON_HEADERS["user-agent"],
        },
        cookies=cookies,
        timeout=30,
        allow_redirects=True,
    )
    resp.raise_for_status()

    # Step 2: Extract the authorization code from the HTML response
    # Auth0 returns an HTML page with the code in a <script> tag:
    #   window.parent.postMessage({type: "authorization_response", response: {code: "...", ...}}, "*")
    # or in an <input name="code" value="...">
    html = resp.text
    print(html)
    code = None

    # Try to find code in postMessage JSON
    m = re.search(r'"code"\s*:\s*"([^"]+)"', html)
    if m:
        code = m.group(1)

    # Fallback: look for input name="code"
    if not code:
        m = re.search(r'<input[^>]*name="code"[^>]*value="([^"]+)"', html)
        if m:
            code = m.group(1)

    # Fallback: look for code in URL fragment (response_mode=web_message uses hash)
    if not code and "#" in resp.url:
        fragment = urlparse(resp.url).fragment
        params = parse_qs(fragment)
        if "code" in params:
            code = params["code"][0]

    if not code:
        # Maybe Auth0 returned an error
        error_m = re.search(r'"error"\s*:\s*"([^"]+)"', html)
        error_desc_m = re.search(r'"error_description"\s*:\s*"([^"]+)"', html)
        if error_m:
            error = error_m.group(1)
            desc = error_desc_m.group(1) if error_desc_m else ""
            log.error("Auth0 error: %s — %s", error, unquote(desc.replace("+", " ")))
        log.error(
            "Could not extract authorization code from Auth0 response. "
            "Your cookies may have expired. Response length: %d chars.",
            len(html),
        )
        sys.exit(1)

    log.info("Authorization code obtained. Exchanging for tokens…")

    # Step 3: POST /oauth/token with authorization_code + code_verifier
    token_resp = requests.post(
        f"{AUTH0_DOMAIN}/oauth/token",
        headers=_COMMON_HEADERS,
        data={
            "client_id": AUTH0_CLIENT_ID,
            "redirect_uri": AUTH0_REDIRECT_URI,
            "response_type": "code",
            "code_verifier": verifier,
            "code": code,
            "grant_type": "authorization_code",
        },
        timeout=30,
    )
    token_resp.raise_for_status()
    body = token_resp.json()

    access_token = body.get("access_token")
    new_refresh  = body.get("refresh_token")
    if not access_token:
        log.error("No access_token in Auth0 response: %s", body)
        sys.exit(1)

    if new_refresh:
        Path(REFRESH_TOKEN_FILE).write_text(new_refresh + "\n", encoding="utf-8")
        log.info("New refresh token saved to %s.", REFRESH_TOKEN_FILE)

    return access_token


def refresh_access_token(cookies: dict | None = None) -> str:
    """Get a fresh access token.

    Strategy:
      1. Try refresh_token.txt (no cookies needed).
      2. If it fails or there's no refresh_token, use PKCE + Auth0 session cookies.
      3. If no cookies either, fail with clear instructions.

    Returns the access_token string on success.
    Writes the new refresh token back to refresh_token.txt.
    """
    # Try refresh_token first
    try:
        rt = Path(REFRESH_TOKEN_FILE).read_text(encoding="utf-8").strip()
    except FileNotFoundError:
        rt = ""

    if rt:
        log.info("Trying refresh token from %s…", REFRESH_TOKEN_FILE)
        resp = requests.post(
            f"{AUTH0_DOMAIN}/oauth/token",
            headers=_COMMON_HEADERS,
            data={
                "client_id": AUTH0_CLIENT_ID,
                "redirect_uri": AUTH0_REDIRECT_URI,
                "response_type": "code",
                "grant_type": "refresh_token",
                "refresh_token": rt,
            },
            timeout=30,
        )

        if resp.status_code == 200:
            body = resp.json()
            access_token = body.get("access_token")
            new_refresh = body.get("refresh_token")
            if access_token:
                if new_refresh:
                    Path(REFRESH_TOKEN_FILE).write_text(new_refresh + "\n", encoding="utf-8")
                    log.info("New refresh token saved to %s.", REFRESH_TOKEN_FILE)
                return access_token

        log.warning("Refresh token failed (HTTP %d) — trying cookie-based auth.", resp.status_code)

    # Fallback: PKCE with cookies
    if cookies:
        return silent_auth_with_cookies(cookies)

    log.error(
        "No valid refresh token and no cookies available.\n"
        "Provide Auth0 session cookies via --cookies, COOKIES env var, or cookies.txt.\n"
        "To get cookies: open https://app.toqla.fr, DevTools > Network > copy Cookie header."
    )
    sys.exit(1)


# ---------------------------------------------------------------------------
# Session
# ---------------------------------------------------------------------------
def build_session(token: str) -> requests.Session:
    """Build a requests.Session pre-configured with the Toqla API auth headers."""
    session = requests.Session()
    session.headers.update({
        "Authorization": f"Bearer {token}",
        "x-api-key": API_KEY,
        "Accept": "application/json",
        "Origin": "https://app.toqla.fr",
        "Referer": "https://app.toqla.fr/",
        "User-Agent": _COMMON_HEADERS["user-agent"],
    })
    return session