54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
from collections import OrderedDict
|
|
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}$")
|
|
with open('/run/secrets/monzo-api_client-id') as file: client_id = file.read().rstrip()
|
|
with open('/run/secrets/monzo-api_client-secret') as file: client_secret = file.read().rstrip()
|
|
with open('/run/secrets/monzo-api_callback-uri') as file: callback_uri = file.read().rstrip()
|
|
|
|
app = FastAPI(root_path='/monzo')
|
|
|
|
@app.get('/')
|
|
def read_root(): return {'version': 'v0.0.10'}
|
|
|
|
|
|
@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() }
|