223 lines
7.8 KiB
Python
223 lines
7.8 KiB
Python
# API Key Based Authorization
|
|
#
|
|
# Copyright (C) 2020 Eric Callahan <arksine.code@gmail.com>
|
|
#
|
|
# This file may be distributed under the terms of the GNU GPLv3 license
|
|
import base64
|
|
import uuid
|
|
import os
|
|
import time
|
|
import ipaddress
|
|
import re
|
|
import logging
|
|
from tornado.ioloop import IOLoop, PeriodicCallback
|
|
from utils import ServerError
|
|
|
|
TOKEN_TIMEOUT = 5
|
|
CONNECTION_TIMEOUT = 3600
|
|
PRUNE_CHECK_TIME = 300 * 1000
|
|
|
|
class Authorization:
|
|
def __init__(self, config):
|
|
self.server = config.get_server()
|
|
database = self.server.lookup_component('database')
|
|
database.register_local_namespace('authorization', forbidden=True)
|
|
self.auth_db = database.wrap_namespace('authorization')
|
|
self.api_key = self.auth_db.get('api_key', None)
|
|
if self.api_key is None:
|
|
self.api_key = uuid.uuid4().hex
|
|
self.auth_db['api_key'] = self.api_key
|
|
self.trusted_connections = {}
|
|
self.access_tokens = {}
|
|
|
|
# Get allowed cors domains
|
|
self.cors_domains = []
|
|
cors_cfg = config.get('cors_domains', "").strip()
|
|
cds = [d.strip() for d in cors_cfg.split('\n')if d.strip()]
|
|
for domain in cds:
|
|
bad_match = re.search(r"^.+\.[^:]*\*", domain)
|
|
if bad_match is not None:
|
|
raise config.error(
|
|
f"Unsafe CORS Domain '{domain}'. Wildcards are not"
|
|
" permitted in the top level domain.")
|
|
self.cors_domains.append(
|
|
domain.replace(".", "\\.").replace("*", ".*"))
|
|
|
|
# Get Trusted Clients
|
|
self.trusted_ips = []
|
|
self.trusted_ranges = []
|
|
trusted_clients = config.get('trusted_clients', "")
|
|
trusted_clients = [c.strip() for c in trusted_clients.split('\n')
|
|
if c.strip()]
|
|
for ip in trusted_clients:
|
|
# Check IP address
|
|
try:
|
|
tc = ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
tc = None
|
|
if tc is None:
|
|
# Check ip network
|
|
try:
|
|
tc = ipaddress.ip_network(ip)
|
|
except ValueError:
|
|
raise ServerError(
|
|
f"Invalid option in trusted_clients: {ip}")
|
|
self.trusted_ranges.append(tc)
|
|
else:
|
|
self.trusted_ips.append(tc)
|
|
|
|
t_clients = "\n".join(
|
|
[str(ip) for ip in self.trusted_ips] +
|
|
[str(rng) for rng in self.trusted_ranges])
|
|
c_domains = "\n".join(self.cors_domains)
|
|
|
|
logging.info(
|
|
f"Authorization Configuration Loaded\n"
|
|
f"Trusted Clients:\n{t_clients}\n"
|
|
f"CORS Domains:\n{c_domains}")
|
|
|
|
self.prune_handler = PeriodicCallback(
|
|
self._prune_conn_handler, PRUNE_CHECK_TIME)
|
|
self.prune_handler.start()
|
|
|
|
# Register Authorization Endpoints
|
|
self.server.register_endpoint(
|
|
"/access/api_key", ['GET', 'POST'],
|
|
self._handle_apikey_request, protocol=['http'])
|
|
self.server.register_endpoint(
|
|
"/access/oneshot_token", ['GET'],
|
|
self._handle_token_request, protocol=['http'])
|
|
|
|
async def _handle_apikey_request(self, web_request):
|
|
action = web_request.get_action()
|
|
if action.upper() == 'POST':
|
|
self.api_key = uuid.uuid4().hex
|
|
self.auth_db['api_key'] = self.api_key
|
|
return self.api_key
|
|
|
|
async def _handle_token_request(self, web_request):
|
|
return self.get_access_token()
|
|
|
|
def _check_authorized_ip(self, ip):
|
|
if ip in self.trusted_ips:
|
|
return True
|
|
for rng in self.trusted_ranges:
|
|
if ip in rng:
|
|
return True
|
|
return False
|
|
|
|
def _prune_conn_handler(self):
|
|
cur_time = time.time()
|
|
expired_conns = []
|
|
for ip, access_time in self.trusted_connections.items():
|
|
if cur_time - access_time > CONNECTION_TIMEOUT:
|
|
expired_conns.append(ip)
|
|
for ip in expired_conns:
|
|
self.trusted_connections.pop(ip, None)
|
|
logging.info(
|
|
f"Trusted Connection Expired, IP: {ip}")
|
|
|
|
def _token_expire_handler(self, token):
|
|
self.access_tokens.pop(token, None)
|
|
|
|
def get_access_token(self):
|
|
token = base64.b32encode(os.urandom(20)).decode()
|
|
ioloop = IOLoop.current()
|
|
self.access_tokens[token] = ioloop.call_later(
|
|
TOKEN_TIMEOUT, self._token_expire_handler, token)
|
|
return token
|
|
|
|
def _check_trusted_connection(self, ip):
|
|
if ip is not None:
|
|
if ip in self.trusted_connections:
|
|
self.trusted_connections[ip] = time.time()
|
|
return True
|
|
elif self._check_authorized_ip(ip):
|
|
logging.info(
|
|
f"Trusted Connection Detected, IP: {ip}")
|
|
self.trusted_connections[ip] = time.time()
|
|
return True
|
|
return False
|
|
|
|
def _check_access_token(self, token):
|
|
if token in self.access_tokens:
|
|
token_handler = self.access_tokens.pop(token, None)
|
|
IOLoop.current().remove_timeout(token_handler)
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def check_authorized(self, request):
|
|
# Check if IP is trusted
|
|
try:
|
|
ip = ipaddress.ip_address(request.remote_ip)
|
|
except ValueError:
|
|
logging.exception(
|
|
f"Unable to Create IP Address {request.remote_ip}")
|
|
ip = None
|
|
if self._check_trusted_connection(ip):
|
|
return True
|
|
|
|
# Check API Key Header
|
|
key = request.headers.get("X-Api-Key")
|
|
if key and key == self.api_key:
|
|
return True
|
|
|
|
# Check one-shot access token
|
|
token = request.arguments.get('token', [b""])[0].decode()
|
|
if self._check_access_token(token):
|
|
return True
|
|
return False
|
|
|
|
def check_cors(self, origin, request=None):
|
|
if origin is None or not self.cors_domains:
|
|
return False
|
|
for regex in self.cors_domains:
|
|
match = re.match(regex, origin)
|
|
if match is not None:
|
|
if match.group() == origin:
|
|
logging.debug(f"CORS Pattern Matched, origin: {origin} "
|
|
f" | pattern: {regex}")
|
|
self._set_cors_headers(origin, request)
|
|
return True
|
|
else:
|
|
logging.debug(f"Partial Cors Match: {match.group()}")
|
|
else:
|
|
# Check to see if the origin contains an IP that matches a
|
|
# current trusted connection
|
|
match = re.search(r"^https?://([^/:]+)", origin)
|
|
if match is not None:
|
|
ip = match.group(1)
|
|
try:
|
|
ipaddr = ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
if self._check_authorized_ip(ipaddr):
|
|
logging.debug(
|
|
f"Cors request matched trusted IP: {ip}")
|
|
self._set_cors_headers(origin, request)
|
|
return True
|
|
logging.debug(f"No CORS match for origin: {origin}\n"
|
|
f"Patterns: {self.cors_domains}")
|
|
return False
|
|
|
|
def _set_cors_headers(self, origin, request):
|
|
if request is None:
|
|
return
|
|
request.set_header("Access-Control-Allow-Origin", origin)
|
|
request.set_header(
|
|
"Access-Control-Allow-Methods",
|
|
"GET, POST, PUT, DELETE, OPTIONS")
|
|
request.set_header(
|
|
"Access-Control-Allow-Headers",
|
|
"Origin, Accept, Content-Type, X-Requested-With, "
|
|
"X-CRSF-Token")
|
|
|
|
def close(self):
|
|
self.prune_handler.stop()
|
|
|
|
|
|
def load_component(config):
|
|
return Authorization(config)
|