redis/auth/token.py

from abc import ABC, abstractmethod
from datetime import datetime, timezone

from redis.auth.err import InvalidTokenSchemaErr


class TokenInterface(ABC):
    @abstractmethod
    def is_expired(self) -> bool:
        pass

    @abstractmethod
    def ttl(self) -> float:
        pass

    @abstractmethod
    def try_get(self, key: str) -> str:
        pass

    @abstractmethod
    def get_value(self) -> str:
        pass

    @abstractmethod
    def get_expires_at_ms(self) -> float:
        pass

    @abstractmethod
    def get_received_at_ms(self) -> float:
        pass


class TokenResponse:
    def __init__(self, token: TokenInterface):
        self._token = token

    def get_token(self) -> TokenInterface:
        return self._token

    def get_ttl_ms(self) -> float:
        return self._token.get_expires_at_ms() - self._token.get_received_at_ms()


class SimpleToken(TokenInterface):
    def __init__(
        self, value: str, expires_at_ms: float, received_at_ms: float, claims: dict
    ) -> None:
        self.value = value
        self.expires_at = expires_at_ms
        self.received_at = received_at_ms
        self.claims = claims

    def ttl(self) -> float:
        if self.expires_at == -1:
            return -1

        return self.expires_at - (datetime.now(timezone.utc).timestamp() * 1000)

    def is_expired(self) -> bool:
        if self.expires_at == -1:
            return False

        return self.ttl() <= 0

    def try_get(self, key: str) -> str:
        return self.claims.get(key)

    def get_value(self) -> str:
        return self.value

    def get_expires_at_ms(self) -> float:
        return self.expires_at

    def get_received_at_ms(self) -> float:
        return self.received_at


class JWToken(TokenInterface):
    REQUIRED_FIELDS = {"exp"}

    def __init__(self, token: str):
        try:
            import jwt
        except ImportError as ie:
            raise ImportError(
                f"The PyJWT library is required for {self.__class__.__name__}.",
            ) from ie
        self._value = token
        self._decoded = jwt.decode(
            self._value,
            options={"verify_signature": False},
            algorithms=[jwt.get_unverified_header(self._value).get("alg")],
        )
        self._validate_token()

    def is_expired(self) -> bool:
        exp = self._decoded["exp"]
        if exp == -1:
            return False

        return (
            self._decoded["exp"] * 1000 <= datetime.now(timezone.utc).timestamp() * 1000
        )

    def ttl(self) -> float:
        exp = self._decoded["exp"]
        if exp == -1:
            return -1

        return (
            self._decoded["exp"] * 1000 - datetime.now(timezone.utc).timestamp() * 1000
        )

    def try_get(self, key: str) -> str:
        return self._decoded.get(key)

    def get_value(self) -> str:
        return self._value

    def get_expires_at_ms(self) -> float:
        return float(self._decoded["exp"] * 1000)

    def get_received_at_ms(self) -> float:
        return datetime.now(timezone.utc).timestamp() * 1000

    def _validate_token(self):
        actual_fields = {x for x in self._decoded.keys()}

        if len(self.REQUIRED_FIELDS - actual_fields) != 0:
            raise InvalidTokenSchemaErr(self.REQUIRED_FIELDS - actual_fields)