openwebrx-clone/owrx/receiverid.py

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())