XF/pluto
XF
/
pluto
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Handle invalid JSON in Revolut

This commit is contained in:
Mark Veidemanis 2021-12-23 22:57:51 +00:00
parent 04c20b6b69
commit b6fe0b433d
Signed by: m
GPG Key ID: 5ACFCEED46C0904F
1 changed files with 18 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
handler/revolut.py
View File

@ -3,6 +3,7 @@ from twisted.logger import Logger
# Other library imports
from json import dumps
from json.decoder import JSONDecodeError
import requests
# Project imports
@ -16,23 +17,27 @@ class Revolut(object):
def __init__(self):
self.log = Logger("revolut")
self.token = None
def get_new_token(self):
headers = {"Content-Type": "application/x-www-form-urlencoded"}
data = {
"grant_type": "refresh_token",
"refresh_token": settings.refresh_token,
"client_id": settings.client_id,
"refresh_token": settings.Revolut.RefreshToken,
"client_id": settings.Revolut.ClientID,
"client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
"client_assertion": settings.jwt,
"client_assertion": settings.Revolut.JWT,
}
r = requests.post(f"{settings.api_base}/auth/token", data=data, headers=headers)
parsed = r.json()
r = requests.post(f"{settings.Revolut.Base}/auth/token", data=data, headers=headers)
try:
parsed = r.json()
except JSONDecodeError:
return False
if r.status_code == 200:
if "access_token" in parsed.keys():
settings.access_token = parsed["access_token"]
if len(settings.access_token) == len(settings.refresh_token):
self.log.info("Refreshed access token: {access_token}", access_token=settings.access_token)
self.token = parsed["access_token"]
if len(self.token) == len(settings.Revolut.RefreshToken):
self.log.info("Refreshed access token: {access_token}", access_token=self.token)
return True
else:
self.log.error(f"Token refresh didn't contain access token: {parsed}", parsed=parsed)
@ -42,12 +47,12 @@ class Revolut(object):
return False
def setup_webhook(self):
self.log.info("Setting up webhook: {url}", url=settings.webhook_url)
headers = {"Authorization": f"Bearer {settings.access_token}"}
data = {"url": settings.webhook_url}
r = requests.post(f"{settings.api_base}/webhook", data=dumps(data), headers=headers)
self.log.info("Setting up webhook: {url}", url=settings.Revolut.WebhookURL)
headers = {"Authorization": f"Bearer {self.token}"}
data = {"url": settings.Revolut.WebhookURL}
r = requests.post(f"{settings.Revolut.Base}/webhook", data=dumps(data), headers=headers)
if r.status_code == 204:
self.log.info("Set up webhook: {url}", url=settings.webhook_url)
self.log.info("Set up webhook: {url}", url=settings.Revolut.WebhookURL)
return dumps({"success": True})
else:
parsed = r.json()