mazda/mazda_webservice2.py

import base64
import os

import requests
from flask import Flask, redirect, request, session

app = Flask(__name__)
app.secret_key = os.urandom(24)

# Mazda OAuth2 configuration
CLIENT_ID = "YOUR_CLIENT_ID"
CLIENT_SECRET = "YOUR_CLIENT_SECRET"
AUTHORIZATION_BASE_URL = "https://mappsacc.mazdaeur.com/oauth/authorize"
TOKEN_URL = "https://mappsacc.mazdaeur.com/oauth/token"
REDIRECT_URI = "http://localhost:5000/callback"


@app.route("/")
def home():
    return '<a href="/login">Login with Mazda</a>'


@app.route("/login")
def login():
    state = os.urandom(8).hex()
    session["oauth_state"] = state
    auth_url = (
        f"{AUTHORIZATION_BASE_URL}?response_type=code"
        f"&client_id={CLIENT_ID}"
        f"&redirect_uri={REDIRECT_URI}"
        f"&state={state}"
    )
    return redirect(auth_url)


@app.route("/callback")
def callback():
    code = request.args.get("code")
    state = request.args.get("state")

    if state != session.get("oauth_state"):
        return "State mismatch. Possible CSRF attack.", 400

    # Exchange code for token
    auth = base64.b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
    headers = {"Authorization": f"Basic {auth}", "Content-Type": "application/x-www-form-urlencoded"}
    data = {"grant_type": "authorization_code", "code": code, "redirect_uri": REDIRECT_URI}

    response = requests.post(TOKEN_URL, headers=headers, data=data)
    token_data = response.json()

    session["access_token"] = token_data.get("access_token")
    session["refresh_token"] = token_data.get("refresh_token")

    return f"Access Token: {session['access_token']}<br>Refresh Token: {session['refresh_token']}"


@app.route("/refresh")
def refresh_token():
    refresh_token = session.get("refresh_token")
    if not refresh_token:
        return "No refresh token available.", 400

    auth = base64.b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
    headers = {"Authorization": f"Basic {auth}", "Content-Type": "application/x-www-form-urlencoded"}
    data = {"grant_type": "refresh_token", "refresh_token": refresh_token}

    response = requests.post(TOKEN_URL, headers=headers, data=data)
    token_data = response.json()

    session["access_token"] = token_data.get("access_token")
    session["refresh_token"] = token_data.get("refresh_token")

    return f"New Access Token: {session['access_token']}"


if __name__ == "__main__":
    app.run(
        host="0.0.0.0",
        port="5000",
        ssl_context=(
            "config/cert/global-cube.com.crt",
            "config/cert/global-cube.com.key",
            # "config/cert/intermediate.crt",
        ),
        debug=True,
    )