diff --git a/asdabot/auth.py b/asdabot/auth.py index 916f3a6..9a0c756 100644 --- a/asdabot/auth.py +++ b/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 = ( + " 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: diff --git a/asdabot/browser.py b/asdabot/browser.py index c443d9e..f5edb2c 100644 --- a/asdabot/browser.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index ffaf6ac..1a04f8d 100644 --- a/pyproject.toml +++ b/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 diff --git a/tests/test_login_logic.py b/tests/test_login_logic.py index e4541cf..e55d9d2 100644 --- a/tests/test_login_logic.py +++ b/tests/test_login_logic.py @@ -10,11 +10,13 @@ import pytest +from asdabot.auth import AuthError, describe_http_failure, tokens_from_access_jwt from asdabot.browser import _extract_profile, _is_logged_in -def make_jwt(isb: str) -> str: - payload = base64.urlsafe_b64encode(json.dumps({"isb": isb}).encode()).decode().rstrip("=") +def make_jwt(isb: str, **extra) -> str: + body = {"isb": isb, **extra} + payload = base64.urlsafe_b64encode(json.dumps(body).encode()).decode().rstrip("=") return f"header.{payload}.signature" @@ -73,3 +75,32 @@ def test_extract_profile_maps_the_default_address(): def test_extract_profile_refuses_an_account_without_an_address(): with pytest.raises(LookupError): _extract_profile({"profile": {}, "addresses": []}) + + +def test_tokens_from_registered_jwt_without_httpx(): + token = make_jwt( + "uido:azure_adb2c-signin-bjgs_prd::upn:1234::uidn:Mark::rcid:cust-99", + sub="cc-slas::ecom::usid:shopper-1", + exp=1_900_000_000, + ) + stored = tokens_from_access_jwt(token, "refresh-xyz", adb2c_token="adb2c-1") + assert stored["slas_auth"] == f"Bearer {token}" + assert stored["slas_refresh"] == "refresh-xyz" + assert stored["customer_id"] == "cust-99" + assert stored["usid"] == "shopper-1" + assert stored["adb2c_auth"] == "adb2c-1" + assert stored["expires_at"] == 1_900_000_000 + + +def test_tokens_from_guest_jwt_rejected(): + token = make_jwt("uido:slas::upn:Guest::uidn:Guest User::gcid:abc::chid:ASD") + with pytest.raises(AuthError, match="Guest"): + tokens_from_access_jwt(token, "refresh-xyz") + + +def test_describe_http_failure_flags_cloudflare_html(): + msg = describe_http_failure(403, "Attention Required! cloudflare") + assert "HTTP 403" in msg + assert "Cloudflare" in msg + assert "auth login" in msg + assert "" not in msg