Fix login 403 by persisting captured SLAS cookies

asdabot/auth.py

@@ -1,13 +1,16 @@
 """Authentication — token management and automatic refresh."""
 
-import time
+from __future__ import annotations
 
-import httpx
+import base64
+import json
+import time
 
 from asdabot.config import SFCC_ORG, SFCC_PROXY_BASE, load_account, save_account
 
 SLAS_CLIENT_ID = "e68ca36d-6516-4704-b705-06b74f85ef2e"
 TOKEN_REFRESH_URL = f"{SFCC_PROXY_BASE}/shopper/auth/v1/organizations/{SFCC_ORG}/oauth2/token"
+_DEFAULT_REFRESH_TTL = 90 * 86400
 
 
 class AuthError(Exception):
@@ -22,56 +25,119 @@ def require_account() -> dict:
     return account
 
 
-def refresh_tokens() -> dict:
-    """Refresh SLAS tokens. Returns updated account."""
-    account = require_account()
-    tokens = account.get("tokens", {})
+def decode_jwt_payload(token: str) -> dict:
+    """Decode a JWT payload without verifying the signature."""
+    raw = token.removeprefix("Bearer ").strip()
+    parts = raw.split(".")
+    if len(parts) != 3:
+        raise AuthError("Access token is not a JWT")
+    try:
+        return json.loads(base64.urlsafe_b64decode(parts[1] + "=="))
+    except (ValueError, json.JSONDecodeError) as exc:
+        raise AuthError("Access token payload is not valid JSON") from exc
+
+
+def parse_isb(isb: str) -> dict[str, str]:
+    """Split a SLAS `isb` claim (`key:value::key:value`)."""
+    fields: dict[str, str] = {}
+    for chunk in isb.split("::"):
+        if ":" in chunk:
+            key, _, value = chunk.partition(":")
+            fields[key] = value
+    return fields
+
+
+def usid_from_sub(sub: str) -> str:
+    fields = parse_isb(sub) if "::" in sub else {}
+    if fields.get("usid"):
+        return fields["usid"]
+    if sub:
+        return sub.rsplit(":", 1)[-1]
+    return ""
+
+
+def tokens_from_access_jwt(
+    access_token: str,
+    refresh_token: str,
+    *,
+    adb2c_token: str = "",
+    refresh_expires_in: float | None = None,
+) -> dict:
+    """Build the stored token map from cookies / a SLAS token response."""
+    access = access_token.removeprefix("Bearer ").strip()
+    payload = decode_jwt_payload(access)
+    isb = str(payload.get("isb") or "")
+    if "upn:Guest" in isb:
+        raise AuthError("Guest session. Run 'asdabot auth login' and sign in.")
+    fields = parse_isb(isb)
+    now = time.time()
+    refresh_ttl = float(refresh_expires_in) if refresh_expires_in else _DEFAULT_REFRESH_TTL
+    return {
+        "slas_auth": f"Bearer {access}",
+        "slas_refresh": refresh_token,
+        "customer_id": fields.get("rcid") or fields.get("gcid") or "",
+        "usid": usid_from_sub(str(payload.get("sub") or "")),
+        "adb2c_auth": adb2c_token,
+        "expires_at": float(payload.get("exp") or 0),
+        "refresh_expires_at": now + refresh_ttl,
+    }
 
-    refresh_token = tokens.get("slas_refresh")
-    if not refresh_token:
-        raise AuthError("No refresh token. Run 'asdabot auth login' first.")
 
-    resp = httpx.post(
-        TOKEN_REFRESH_URL,
-        headers={
-            "content-type": "application/x-www-form-urlencoded",
-            "user-agent": "Mozilla/5.0",
-            "origin": "https://www.asda.com",
-            "referer": "https://www.asda.com/",
-        },
-        data={
-            "grant_type": "refresh_token",
-            "refresh_token": refresh_token,
-            "client_id": SLAS_CLIENT_ID,
-        },
+def apply_token_response(account: dict, data: dict) -> dict:
+    """Write a SLAS /token JSON body onto the account and persist it."""
+    extra = tokens_from_access_jwt(
+        data["access_token"],
+        data["refresh_token"],
+        adb2c_token=str(data.get("idp_access_token") or ""),
+        refresh_expires_in=data.get("refresh_token_expires_in"),
     )
-    if resp.status_code >= 400:
-        raise AuthError(
-            f"ASDA rejected the session (HTTP {resp.status_code}). "
-            "Run 'asdabot auth login' to start a new one."
-        )
-    data = resp.json()
-
-    account["tokens"] = {
-        "slas_auth": f"Bearer {data['access_token']}",
-        "slas_refresh": data["refresh_token"],
-        "customer_id": data["customer_id"],
-        "usid": data["usid"],
-        "adb2c_auth": data.get("idp_access_token", ""),
-        "expires_at": time.time() + data["expires_in"],
-        "refresh_expires_at": time.time() + data["refresh_token_expires_in"],
-    }
+    if data.get("customer_id"):
+        extra["customer_id"] = data["customer_id"]
+    if data.get("usid"):
+        extra["usid"] = data["usid"]
+    if data.get("expires_in"):
+        extra["expires_at"] = time.time() + float(data["expires_in"])
+    account["tokens"] = extra
     save_account(account)
     return account
 
 
+def describe_http_failure(status: int, body: str) -> str:
+    """Explain a failed token HTTP response without dumping a Cloudflare page."""
+    text = body or ""
+    lowered = text.lower()
+    cloudflare = (
+        "<html" in lowered or "cloudflare" in lowered or "attention required" in lowered
+    )
+    extra = " Cloudflare blocked the request." if cloudflare else ""
+    detail = ""
+    if text and not cloudflare:
+        detail = f" {text[:160].replace(chr(10), ' ')}"
+    return (
+        f"ASDA rejected the session (HTTP {status}).{extra} "
+        f"Run 'asdabot auth login' to start a new one.{detail}"
+    )
+
+
+def refresh_tokens() -> dict:
+    """Refresh SLAS tokens via the existing Chrome profile. Returns updated account."""
+    account = require_account()
+    refresh_token = account.get("tokens", {}).get("slas_refresh")
+    if not refresh_token:
+        raise AuthError("No refresh token. Run 'asdabot auth login' first.")
+    from asdabot.browser import refresh_tokens_in_browser
+
+    data = refresh_tokens_in_browser(refresh_token)
+    return apply_token_response(account, data)
+
+
 def ensure_valid_tokens() -> dict:
     """Return valid account, refreshing tokens if expired."""
     account = require_account()
     tokens = account.get("tokens", {})
-    if time.time() > (tokens.get("expires_at", 0) - 60):
-        account = refresh_tokens()
-    return account
+    if tokens.get("slas_auth") and time.time() <= (tokens.get("expires_at", 0) - 60):
+        return account
+    return refresh_tokens()
 
 
 def get_slas_bearer_token() -> str:

asdabot/browser.py

@@ -3,11 +3,16 @@
 The user logs in through their own Chromium-based browser (dedicated
 profile) launched with no automation flags — Cloudflare Turnstile rejects
 logins while a CDP debugging port is open. Afterwards an invisible headless
-instance reads the session cookies straight off the profile's disk; only
-the SLAS refresh token matters, since the token refresh flow mints all
-other credentials from it.
+instance reads the session cookies straight off the profile's disk.
+
+The access JWT is already in SLAS.AUTH_TOKEN after login. We persist that
+together with the refresh token instead of exchanging the refresh token
+over a bare HTTP client (www.asda.com is Cloudflare-fronted and returns 403).
+Later refreshes reuse this same Chrome profile so the request stays in-browser.
 """
 
+from __future__ import annotations
+
 import base64
 import contextlib
 import json
@@ -22,7 +27,13 @@
 import httpx
 from websockets.sync.client import connect
 
-from asdabot.auth import refresh_tokens
+from asdabot.auth import (
+    SLAS_CLIENT_ID,
+    TOKEN_REFRESH_URL,
+    AuthError,
+    describe_http_failure,
+    tokens_from_access_jwt,
+)
 from asdabot.config import (
     ASDA_BASE,
     CHROME_PROFILE_DIR,
@@ -34,7 +45,7 @@
     save_account,
 )
 
-SESSION_COOKIE_NAMES = ("SLAS.AUTH_TOKEN", "SLAS.REFRESH_TOKEN")
+SESSION_COOKIE_NAMES = ("SLAS.AUTH_TOKEN", "SLAS.REFRESH_TOKEN", "ADB2C.AUTH_TOKEN")
 
 MAC_BROWSERS = [
     "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
@@ -89,9 +100,12 @@ def __init__(self, ws_url: str):
         self.ws = connect(ws_url, max_size=16 * 1024 * 1024)
         self.msg_id = 0
 
-    def call(self, method: str, params: dict | None = None) -> dict:
+    def call(self, method: str, params: dict | None = None, session_id: str | None = None) -> dict:
         self.msg_id += 1
-        self.ws.send(json.dumps({"id": self.msg_id, "method": method, "params": params or {}}))
+        payload: dict = {"id": self.msg_id, "method": method, "params": params or {}}
+        if session_id:
+            payload["sessionId"] = session_id
+        self.ws.send(json.dumps(payload))
         while True:
             msg = json.loads(self.ws.recv(timeout=30))
             if msg.get("id") == self.msg_id:
@@ -138,8 +152,8 @@ def _shutdown(proc: subprocess.Popen):
     time.sleep(1)
 
 
-def _session_cookies() -> dict[str, str]:
-    """Read the ASDA session cookies off the login profile via an invisible browser."""
+def _connect_headless() -> tuple[subprocess.Popen, CDP]:
+    """Start headless Chrome on the login profile and attach over CDP."""
     port_file = CHROME_PROFILE_DIR / "DevToolsActivePort"
     port_file.unlink(missing_ok=True)
 
@@ -153,7 +167,12 @@ def _session_cookies() -> dict[str, str]:
         raise RuntimeError("Browser did not expose a DevTools port.")
 
     port, ws_path = port_file.read_text().splitlines()[:2]
-    browser = CDP(f"ws://127.0.0.1:{port}{ws_path}")
+    return proc, CDP(f"ws://127.0.0.1:{port}{ws_path}")
+
+
+def _session_cookies() -> dict[str, str]:
+    """Read the ASDA session cookies off the login profile via an invisible browser."""
+    proc, browser = _connect_headless()
     try:
         raw = browser.call("Storage.getCookies").get("cookies", [])
         browser.call("Browser.close")
@@ -169,6 +188,61 @@ def _session_cookies() -> dict[str, str]:
     }
 
 
+def refresh_tokens_in_browser(refresh_token: str) -> dict:
+    """POST the SLAS refresh from the dedicated Chrome profile (same cookies as login)."""
+    proc, browser = _connect_headless()
+    try:
+        created = browser.call("Target.createTarget", {"url": ASDA_BASE})
+        attached = browser.call(
+            "Target.attachToTarget",
+            {"targetId": created["targetId"], "flatten": True},
+        )
+        session = attached["sessionId"]
+        browser.call("Page.enable", session_id=session)
+        browser.call("Runtime.enable", session_id=session)
+        time.sleep(2)
+        script = (
+            "(async () => {"
+            f" const p = {json.dumps({'url': TOKEN_REFRESH_URL, 'refresh_token': refresh_token, 'client_id': SLAS_CLIENT_ID})};"
+            " const resp = await fetch(p.url, {"
+            "  method: 'POST',"
+            "  headers: {"
+            "    'content-type': 'application/x-www-form-urlencoded',"
+            "    origin: 'https://www.asda.com',"
+            "    referer: 'https://www.asda.com/',"
+            "  },"
+            "  credentials: 'include',"
+            "  body: new URLSearchParams({"
+            "    grant_type: 'refresh_token',"
+            "    refresh_token: p.refresh_token,"
+            "    client_id: p.client_id,"
+            "  }).toString(),"
+            " });"
+            " return { status: resp.status, text: await resp.text() };"
+            "})()"
+        )
+        result = browser.call(
+            "Runtime.evaluate",
+            {"expression": script, "awaitPromise": True, "returnByValue": True},
+            session_id=session,
+        )
+        value = (result.get("result") or {}).get("value") or {}
+        status = int(value.get("status") or 0)
+        body = str(value.get("text") or "")
+        if status >= 400 or status == 0:
+            raise AuthError(describe_http_failure(status or 0, body))
+        try:
+            return json.loads(body)
+        except json.JSONDecodeError as exc:
+            raise AuthError(describe_http_failure(status, body)) from exc
+    finally:
+        with contextlib.suppress(Exception):
+            browser.call("Browser.close")
+        browser.close()
+        with contextlib.suppress(Exception):
+            proc.wait(timeout=10)
+
+
 def _is_logged_in(cookies: dict[str, str]) -> bool:
     """True once the SLAS token belongs to a real customer.
 
@@ -241,15 +315,22 @@ def browser_login() -> dict:
         )
 
     account = load_account() or {}
-    account["tokens"] = {"slas_refresh": cookies["SLAS.REFRESH_TOKEN"]}
+    account["tokens"] = tokens_from_access_jwt(
+        cookies["SLAS.AUTH_TOKEN"],
+        cookies["SLAS.REFRESH_TOKEN"],
+        adb2c_token=cookies.get("ADB2C.AUTH_TOKEN", ""),
+    )
     save_account(account)
-    account = refresh_tokens()  # mints all other tokens and proves the session works
 
-    try:
-        profile = _fetch_profile(account["tokens"]["adb2c_auth"])
-        account["store_id"], account["address"] = _extract_profile(profile)
-        save_account(account)
-    except Exception as e:
-        print(f"Warning: couldn't fetch your delivery address ({e}). Keeping existing details.")
+    adb2c = account["tokens"].get("adb2c_auth")
+    if adb2c:
+        try:
+            profile = _fetch_profile(adb2c)
+            account["store_id"], account["address"] = _extract_profile(profile)
+            save_account(account)
+        except Exception as e:
+            print(f"Warning: couldn't fetch your delivery address ({e}). Keeping existing details.")
+    else:
+        print("Warning: no ADB2C token in the session; delivery address may be missing.")
 
     return account

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "asdabot"
-version = "1.0.2"
+version = "1.0.3"
 description = "ASDA grocery shopping from the terminal — search, basket, slots; payment stays in your browser"
 readme = "README.md"
 license = "MIT"
@@ -71,3 +71,4 @@ ignore = [
 
 [tool.ruff.lint.per-file-ignores]
 "asdabot/cli.py" = ["T201", "PLC0415", "B008"]  # lazy imports, typer uses defaults
+"asdabot/auth.py" = ["PLC0415"]  # lazy import of browser to avoid a cycle