99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
import re
|
|
import logging
|
|
import hashlib
|
|
import hmac
|
|
from datetime import datetime, timezone
|
|
from owrx.config import Config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
keyRegex = re.compile("^([a-zA-Z]+)-([0-9a-f]{32})-([0-9a-f]{64})$")
|
|
keyChallengeRegex = re.compile("^([a-zA-Z]+)-([0-9a-f]{32})-([0-9a-f]{32})$")
|
|
headerRegex = re.compile("^ReceiverId (.*)$")
|
|
|
|
|
|
class KeyException(Exception):
|
|
pass
|
|
|
|
|
|
class Key(object):
|
|
def __init__(self, keyString):
|
|
matches = keyRegex.match(keyString)
|
|
if not matches:
|
|
raise KeyException("invalid key format")
|
|
self.source = matches.group(1)
|
|
self.id = matches.group(2)
|
|
self.secret = matches.group(3)
|
|
|
|
|
|
class KeyChallenge(object):
|
|
def __init__(self, challengeString):
|
|
matches = keyChallengeRegex.match(challengeString)
|
|
if not matches:
|
|
raise KeyException("invalid key challenge format")
|
|
self.source = matches.group(1)
|
|
self.id = matches.group(2)
|
|
self.challenge = matches.group(3)
|
|
|
|
|
|
class KeyResponse(object):
|
|
def __init__(self, source, id, time, signature):
|
|
self.source = source
|
|
self.id = id
|
|
self.time = time
|
|
self.signature = signature
|
|
|
|
def __str__(self):
|
|
return "{source}-{id}-{time}-{signature}".format(
|
|
source=self.source,
|
|
id=self.id,
|
|
time=self.time,
|
|
signature=self.signature,
|
|
)
|
|
|
|
|
|
class ReceiverId(object):
|
|
@staticmethod
|
|
def getResponseHeader(requestHeader):
|
|
matches = headerRegex.match(requestHeader)
|
|
if not matches:
|
|
raise KeyException("invalid authorization header")
|
|
challenges = [KeyChallenge(i) for i in matches.group(1).split(",")]
|
|
|
|
def signChallenge(challenge):
|
|
key = ReceiverId.findKey(challenge)
|
|
if key is None:
|
|
return
|
|
return ReceiverId.signChallenge(challenge, key)
|
|
|
|
responses = [signChallenge(c) for c in challenges]
|
|
return ",".join(str(r) for r in responses if r is not None)
|
|
|
|
@staticmethod
|
|
def findKey(challenge):
|
|
def parseKey(keyString):
|
|
try:
|
|
return Key(keyString)
|
|
except KeyException as e:
|
|
logger.error(e)
|
|
|
|
config = Config.get()
|
|
if "receiver_keys" not in config or config["receiver_keys"] is None:
|
|
return None
|
|
keys = [parseKey(keyString) for keyString in config["receiver_keys"]]
|
|
keys = [key for key in keys if key is not None]
|
|
matching_keys = [key for key in keys if key.source == challenge.source and key.id == challenge.id]
|
|
if matching_keys:
|
|
return matching_keys[0]
|
|
return None
|
|
|
|
@staticmethod
|
|
def signChallenge(challenge, key):
|
|
now = datetime.utcnow().replace(microsecond=0, tzinfo=timezone.utc)
|
|
now_bytes = int(now.timestamp()).to_bytes(4, byteorder="big")
|
|
m = hmac.new(bytes.fromhex(key.secret), digestmod=hashlib.sha256)
|
|
m.update(bytes.fromhex(challenge.challenge))
|
|
m.update(now_bytes)
|
|
return KeyResponse(challenge.source, challenge.id, now_bytes.hex(), m.hexdigest())
|