diff --git a/bluelinktoken.py b/bluelinktoken.py index 252bcfe..e9f8c08 100755 --- a/bluelinktoken.py +++ b/bluelinktoken.py @@ -65,11 +65,15 @@ # ── Headless Login ──────────────────────────────────────── -def headless_login(brand_key, username, password): +def headless_login(brand_key, username, password, timeout=30): """ Headless login using curl_cffi (Android TLS fingerprint). No browser needed. Works for EU Kia and EU Hyundai. + `timeout` (seconds) applies to every individual HTTP call. The Bluelink + backend can be slow on the first call after a quiet period; the CLI + wraps this function in a retry loop for that case (see `main`). + Flow: 1. GET authorize page (get session cookies) 2. GET /auth/api/v1/accounts/certs (RSA public key) @@ -98,12 +102,12 @@ def headless_login(brand_key, username, password): auth_url = (f"{host}/auth/api/v2/user/oauth2/authorize" f"?response_type=code&client_id={client_id}" f"&redirect_uri={redirect_uri}&lang=de&state=ccsp&country=de") - s.get(auth_url, allow_redirects=True) + s.get(auth_url, allow_redirects=True, timeout=timeout) print(f" ✅ Session established") # Step 2: Get RSA public key print(f"[2/4] Fetching RSA public key...") - resp = s.get(f"{host}/auth/api/v1/accounts/certs") + resp = s.get(f"{host}/auth/api/v1/accounts/certs", timeout=timeout) if resp.status_code != 200: print(f" ❌ Certs endpoint returned {resp.status_code}") sys.exit(1) @@ -131,7 +135,7 @@ def headless_login(brand_key, username, password): "connector_session_key": "", "kid": kid, "_csrf": "", - }, allow_redirects=False) + }, allow_redirects=False, timeout=timeout) if resp.status_code != 302: print(f" ❌ Signin returned HTTP {resp.status_code}") @@ -156,7 +160,7 @@ def headless_login(brand_key, username, password): "redirect_uri": redirect_uri, "client_id": client_id, "client_secret": cfg["client_secret"], - }) + }, timeout=timeout) if resp.status_code != 200: print(f" ❌ Token exchange failed: HTTP {resp.status_code}") @@ -273,6 +277,17 @@ def main(): help="headless (default, no browser) or browser (Selenium)") parser.add_argument("--username", help="Email/username (headless mode)") parser.add_argument("--password", help="Password (headless mode)") + parser.add_argument("--timeout", type=int, default=30, + help="Per-request HTTP timeout in seconds " + "(headless mode, default: 30). Bump to 60+ on " + "very slow networks; tune lower if you'd rather " + "have auth errors surface faster.") + parser.add_argument("--retries", type=int, default=1, + help="Retry the full headless flow this many times " + "on timeout/connection errors. The Bluelink " + "backend often needs a warm-up on the first " + "call. Default: 1 (= one retry after a failure, " + "two attempts total). Use 0 to disable retries.") args = parser.parse_args() if args.mode == "headless": @@ -281,7 +296,32 @@ def main(): print(" Example: python3 bluelinktoken.py --brand kia --username you@email.com --password yourpass") print(" Or use --mode browser for manual login.") sys.exit(1) - headless_login(args.brand, args.username, args.password) + + # Build the set of exception classes worth retrying. We only want to + # retry on transient transport errors (timeout / connection), never + # on auth failures or HTTP-status problems, which are raised as + # sys.exit inside headless_login already. + try: + from curl_cffi import requests as _cc_req + retry_excs = (_cc_req.exceptions.RequestException, OSError) + except (ImportError, AttributeError): + retry_excs = (OSError,) + + for attempt in range(args.retries + 1): + try: + headless_login(args.brand, args.username, args.password, + timeout=args.timeout) + break + except retry_excs as e: + if attempt < args.retries: + print(f"\n⏳ Attempt {attempt + 1} failed " + f"({type(e).__name__}: {e}).") + print(f" The Bluelink backend often needs a warm-up on " + f"the first call — retrying in 5s…\n") + time.sleep(5) + else: + print(f"\n❌ All {args.retries + 1} attempts failed.") + raise else: browser_login(args.brand)