56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
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()
|
|
|
|
@app.get("/")
|
|
def read_root(): return {"version": "v0.0.8"}
|
|
|
|
|
|
@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() }
|