from collections import OrderedDict from dotenv import load_dotenv from fastapi import FastAPI, Query, Request, Response, status from fastapi.responses import RedirectResponse from uuid import uuid4 from urllib.parse import urlencode from typing import Annotated, Union import requests import os UUIDPattern = "[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}" UUIDQuery = Query(min_length=36, max_length=36, pattern=f"^{UUIDPattern}$") load_dotenv() client_id = os.environ['MONZO_CLIENT_ID'] client_secret = os.environ['MONZO_CLIENT_SECRET'] callback_uri = os.environ['MONZO_CALLBACK_URI'] app = FastAPI(root_path='/monzo') @app.get('/') def read_root(): return {'version': 'v0.0.9'} @app.get('/redirect', response_class=RedirectResponse) def read_redirect(res: Response): state = uuid4() # TODO: store state in a cookie to check it later query = urlencode(OrderedDict( client_id=client_id, redirect_uri=callback_uri, state=state, response_type="code", )) res.set_cookie(key="monzo-api:state", value=state) return f"https://auth.monzo.com/?{query}" @app.get('/callback') def read_callback(code: str, state: Annotated[str, UUIDQuery], req: Request, res: Response): stored_state = req.cookies['monzo-api:state'] if (state != stored_state): res.status_code = status.HTTP_400_BAD_REQUEST return { 'error': True, 'data': f"Callback state '{state}' does not match stored state '{stored_state}'." } # TODO: check the state with the user's cookie # TODO: check that the code is a valid jwt data = { 'grant_type': 'authorization_code', 'client_id': client_id, 'client_secret': client_secret, 'redirect_uri': callback_uri, 'code': code, } token = requests.post("https://api.monzo.com/oauth2/token", data=data) return { 'error': False, 'data': token.json() }