make authorization functions

backtesting.py

@@ -1,5 +1,4 @@
 import backtrader as bt
-import numpy as np
 from datetime import datetime
 from importlib import import_module
 

utils.py

@@ -1,4 +1,4 @@
-import os
+import jwt, os
 from pymongo.server_api import ServerApi
 from pymongo import MongoClient
 from dotenv import load_dotenv
@@ -10,3 +10,17 @@
 
 def get_algorithms():
 	return [algorithm['name'] for algorithm in algorithms.find({'owner': {'$not': {'$type': 'object'}}})]
+
+def authorize(encoded):
+	if encoded.startswith('Bearer'):
+		encoded = encoded[7:]
+
+	return jwt.decode(encoded, os.environ['JWT_SECRET'], algorithms=['HS256'])
+
+def authorize_server(encoded):
+	decoded = authorize(encoded)
+
+	if not decoded['server']:
+		raise Exception('Client Token')
+
+	return decoded

views/internal_checker.py

@@ -1,10 +1,10 @@
-import jwt, os
+import os
 from redis import from_url
 from price import get_prices
 from ipaddress import ip_address
-from flask import request, jsonify
+from flask import request
 from importlib import import_module
-from utils import get_algorithms
+from utils import authorize_server, get_algorithms
 
 redis = from_url(os.environ['REDIS_URI'])
 last_checked_point = 0
@@ -20,18 +20,11 @@ def internal_checker():
 	if not ip_address(request.remote_addr).is_private:
 		return 'Forbidden', 403
 
-	jwt_encoded = request.headers.get('Authorization')
-	if not jwt_encoded:
-		return 'Bad Request', 400
-
 	try:
-		jwt_decoded = jwt.decode(jwt_encoded, os.environ['JWT_SECRET'], algorithms=['HS256'])
+		authorize_server(request.headers.get('Authorization'))
 	except Exception:
 		return 'Unauthorized', 401
 
-	if jwt_decoded['event'] != 'auth':
-		return 'Unauthorized', 401
-
 	prices, timestamps, last_complete_point = get_prices()
 	new_datapoint = False
 	if last_complete_point > last_checked_point:
@@ -51,4 +44,4 @@ def internal_checker():
 		redis.hset('signals', mapping=signals)
 		redis.hset('strengths', mapping=strengths)
 
-	return jsonify({'algorithms': algorithms, 'new_datapoint': new_datapoint})
+	return {'algorithms': algorithms, 'new_datapoint': new_datapoint}

views/update_interval.py

@@ -2,26 +2,20 @@
 from flask import request
 from ipaddress import ip_address
 from price import set_default_interval
+from utils import authorize_server
 
 def update_interval():
 	if not ip_address(request.remote_addr).is_private:
 		return 'Forbidden', 403
 
-	jwt_encoded = request.headers.get('Authorization')
-	if not jwt_encoded:
-		return 'Bad Request', 400
-
 	if 'interval' not in request.json:
 		return 'Bad Request', 400
 
 	try:
-		jwt_decoded = jwt.decode(jwt_encoded, os.environ['JWT_SECRET'], algorithms=['HS256'])
+		authorize_server(request.headers.get('Authorization'))
 	except:
 		return 'Unauthorized', 401
 
-	if jwt_decoded['event'] != 'auth':
-		return 'Unauthorized', 401
-
 	try:
 		new_interval = set_default_interval(request.json.interval)
 	except Exception as error:

views/worth.py

@@ -1,4 +1,4 @@
-import jwt, os, io, utils
+import io, utils
 import matplotlib.pyplot as plt
 import numpy as np
 from flask import Response, request
@@ -11,14 +11,10 @@
 bots = utils.client['database']['bots']
 
 def worth(bot_id):
-	jwt_encoded = request.headers.get('Authorization')
-	if not jwt_encoded:
-		return 'Bad Request', 400
-
 	try:
 		bot = bots.find_one({'_id': ObjectId(bot_id)})
-		jwt_decoded = jwt.decode(jwt_encoded, os.environ['JWT_SECRET'], algorithms=['HS256'])
-		if jwt_decoded['_id'] != str(bot['owner']):
+		decoded = utils.authorize(request.headers.get('Authorization'))
+		if decoded['_id'] != str(bot['owner']):
 			raise 'Token Mismatch'
 	except Exception:
 		return 'Unauthorized', 401