I try to automate the login. I have small success but now I’m stuck, maybe someone knows what my mistake is? It is in python but I’m a openhab user… The fault I get is
Token request failed
Status Code: 500
Response Body: {“fault”:{“faultstring”:“Invalid Access Token”,“detail”:{“errorcode”:“keymanagement.service.invalid_access_token”}}}
import os
import base64
import hashlib
import requests
import time
# =========================================================
# CONFIGURATION
# =========================================================
BMW_DEVICE_CODE_URL = "https://customer.bmwgroup.com/gcdm/oauth/device/code"
BMW_TOKEN_URL = "https://customer.bmwgroup.com/gcdm/oauth/token"
TOKENS_FILE = "bmw_tokens.txt"
# =========================================================
# HELPER FUNCTIONS
# =========================================================
def generate_code_verifier(length: int = 64) -> str:
"""Generate PKCE code_verifier."""
return base64.urlsafe_b64encode(os.urandom(length)).rstrip(b'=').decode('utf-8')
def generate_code_challenge(verifier: str) -> str:
"""Generate S256 code_challenge from code_verifier."""
digest = hashlib.sha256(verifier.encode('ascii')).digest()
return base64.urlsafe_b64encode(digest).rstrip(b'=').decode('utf-8')
def save_tokens(tokens):
"""Save tokens to file for later use."""
with open(TOKENS_FILE, "w") as f:
f.write("BMW ConnectedDrive API Tokens\n")
f.write("=" * 70 + "\n\n")
f.write(f"Access Token: {tokens.get('access_token')}\n")
f.write(f"Refresh Token: {tokens.get('refresh_token')}\n")
f.write(f"ID Token: {tokens.get('id_token')}\n")
f.write(f"Token Type: {tokens.get('token_type')}\n")
f.write(f"Expires In: {tokens.get('expires_in')} seconds\n")
f.write(f"Scope: {tokens.get('scope')}\n")
f.write(f"GCID: {tokens.get('gcid')}\n")
f.write(f"\nGenerated at: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
def load_tokens():
"""Load tokens from file if present."""
if not os.path.exists(TOKENS_FILE):
return None
tokens = {}
with open(TOKENS_FILE, "r") as f:
for line in f:
if ":" in line:
key, value = line.split(":", 1)
tokens[key.strip()] = value.strip()
return tokens
def print_curl(url, headers, data):
print("\n📋 Equivalent cURL (for debugging):")
print("curl --request POST \\")
print(f" '{url}' \\")
for k, v in headers.items():
print(f" -H '{k}: {v}' \\")
for k, v in data.items():
print(f" -d '{k}={v}' \\")
def refresh_access_token(client_id, refresh_token):
"""Refresh access and ID tokens using refresh_token."""
print("\n🔄 Refreshing token using refresh_token...")
payload = {
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id
}
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
resp = requests.post(BMW_TOKEN_URL, data=payload, headers=headers)
if resp.ok:
tokens = resp.json()
save_tokens(tokens)
print("✅ Tokens refreshed and saved.")
return tokens
else:
print("❌ Refresh failed:", resp.text)
return None
# =========================================================
# STEP 1 — GET CLIENT ID & PKCE
# =========================================================
print("🚗 BMW CarData Device Code Flow")
print("=" * 70)
CLIENT_ID = input("Enter your BMW Client ID: ").strip()
if not CLIENT_ID:
print("❌ Client ID is required.")
exit(1)
# Check if refresh token exists
existing_tokens = load_tokens()
if existing_tokens and "Refresh Token" in existing_tokens:
use_refresh = input("Found existing tokens. Refresh? (y/n): ").strip().lower()
if use_refresh == "y":
new_tokens = refresh_access_token(CLIENT_ID, existing_tokens["Refresh Token"])
if new_tokens:
exit(0)
# Generate PKCE
code_verifier = generate_code_verifier()
code_challenge = generate_code_challenge(code_verifier)
print("\n✅ Generated code_verifier and code_challenge")
print("STEP 1 Verifier:", code_verifier)
print("STEP 1 Challenge:", code_challenge)
# =========================================================
# STEP 2 — REQUEST DEVICE CODE
# =========================================================
print("\n" + "=" * 70)
print("🔐 STEP 2: Requesting device & user code")
print("=" * 70)
device_code_payload = {
"client_id": CLIENT_ID,
"response_type": "device_code",
"scope": "authenticate_user openid cardata:api:read cardata:streaming:read",
"code_challenge": code_challenge,
"code_challenge_method": "S256"
}
device_code_headers = {
"Accept": "application/json",
"Content-Type": "application/x-www-form-urlencoded"
}
resp = requests.post(BMW_DEVICE_CODE_URL, data=device_code_payload, headers=device_code_headers)
if resp.status_code != 200:
print("❌ Device Code request failed")
print("Status Code:", resp.status_code)
print("Response Body:", resp.text)
print_curl(BMW_DEVICE_CODE_URL, device_code_headers, device_code_payload)
exit(1)
device_response = resp.json()
user_code = device_response["user_code"]
device_code = device_response["device_code"]
verification_uri = device_response.get("verification_uri_complete") or device_response.get("verification_uri")
interval = device_response.get("interval", 5)
expires_in = device_response.get("expires_in", 300)
print("\n✅ Device Code successfully retrieved!")
print("=" * 70)
print(f"📱 User Code: {user_code}")
print(f"🔐 Device Code: {device_code}")
print(f"🌐 Verification URI: {verification_uri}")
print(f"⏳ Expires In: {expires_in} seconds")
print(f"🔁 Poll Interval: {interval} seconds")
print("=" * 70)
print("\n👉 Please open this URL in your browser to authenticate your BMW account:")
print(f"{verification_uri}")
input("⏳ Press Enter after you finished authorizing the device in the browser...")
print("STEP 2 Verifier (reused):", code_verifier)
# =========================================================
# STEP 3 — EXCHANGE DEVICE CODE FOR TOKENS
# =========================================================
print("\n" + "=" * 70)
print("🔑 STEP 3: Exchanging device code for tokens")
print("=" * 70)
token_payload = {
"client_id": CLIENT_ID,
"device_code": device_code,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
"code_verifier": code_verifier
}
token_headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json"
}
token_resp = requests.post(BMW_TOKEN_URL, data=token_payload, headers=token_headers)
if token_resp.status_code != 200:
print("❌ Token request failed")
print("Status Code:", token_resp.status_code)
print("Response Body:", token_resp.text)
print_curl(BMW_TOKEN_URL, token_headers, token_payload)
exit(1)
tokens = token_resp.json()
print("\n✅ Tokens successfully received!")
print("=" * 70)
print(f"Access Token: {tokens.get('access_token', '')[:50]}...")
print(f"Refresh Token: {tokens.get('refresh_token', '')[:50]}...")
print(f"ID Token: {tokens.get('id_token', '')[:50]}...")
print(f"Token Type: {tokens.get('token_type')}")
print(f"Expires In: {tokens.get('expires_in')} seconds")
print(f"Scope: {tokens.get('scope')}")
print(f"GCID: {tokens.get('gcid')}")
print("=" * 70)
# Save tokens
save_tokens(tokens)
print(f"💾 Tokens saved to {TOKENS_FILE}")
# =========================================================
# STEP 4 — TOKEN USAGE INSTRUCTIONS
# =========================================================
print("\n📚 Token Usage Guide")
print("=" * 70)
print("🔹 Access Token:")
print(" - Use for BMW CarData REST API calls")
print(" - Header: Authorization: Bearer <access_token>")
print(" - Valid for 1 hour")
print()
print("🔹 ID Token:")
print(" - Use for streaming vehicle data")
print(" - Valid for 1 hour")
print()
print("🔹 Refresh Token:")
print(" - Valid for 2 weeks")
print(" - Use to get new tokens without re-authenticating")
print("=" * 70)
# =========================================================
# STEP 5 — TEST API CALL (EXAMPLE)
# =========================================================
print("\n🧪 Testing API Access")
print("=" * 70)
# Extract the access token
access_token = tokens.get('access_token')
# Set up headers with Bearer token
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
# Example: Get vehicles list (adjust the endpoint based on BMW API documentation)
api_url = "https://cardata.bmwgroup.com/api/v1/vehicles" # Replace with actual endpoint
try:
api_response = requests.get(api_url, headers=headers)
if api_response.status_code == 200:
print("✅ API call successful!")
print("Response:", api_response.json())
else:
print(f"❌ API call failed with status {api_response.status_code}")
print("Response:", api_response.text)
except Exception as e:
print(f"❌ Error making API call: {e}")
print("=" * 70)
.
