Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 46 additions & 6 deletions bluelinktoken.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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}")
Expand All @@ -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}")
Expand Down Expand Up @@ -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":
Expand All @@ -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)

Expand Down