1import time
  2import zlib
  3from typing import Any
  4
  5from plain.signing import (
  6    JSONSerializer,
  7    SignatureExpired,
  8    Signer,
  9    b62_decode,
 10    b62_encode,
 11    b64_decode,
 12    b64_encode,
 13)
 14
 15
 16class ExpiringSigner:
 17    """A signer with an embedded expiration (vs max age unsign).
 18
 19    Uses composition rather than inheritance since the interface
 20    intentionally differs from Signer (requires expires_in parameter).
 21    """
 22
 23    def __init__(
 24        self,
 25        *,
 26        key: str | None = None,
 27        sep: str = ":",
 28        salt: str | None = None,
 29        algorithm: str = "sha256",
 30        fallback_keys: list[str] | None = None,
 31    ) -> None:
 32        # Compute default salt here to preserve backwards compatibility.
 33        # When ExpiringSigner inherited from Signer, the default salt was
 34        # "plain.loginlink.signing.ExpiringSigner". Now that we use composition,
 35        # we must set it explicitly rather than letting Signer compute its own.
 36        if salt is None:
 37            salt = f"{self.__class__.__module__}.{self.__class__.__name__}"
 38        self._signer = Signer(
 39            key=key,
 40            sep=sep,
 41            salt=salt,
 42            algorithm=algorithm,
 43            fallback_keys=fallback_keys,
 44        )
 45
 46    @property
 47    def sep(self) -> str:
 48        return self._signer.sep
 49
 50    def sign(self, value: str, expires_in: int) -> str:
 51        timestamp = b62_encode(int(time.time() + expires_in))
 52        value = f"{value}{self.sep}{timestamp}"
 53        return self._signer.sign(value)
 54
 55    def unsign(self, signed_value: str) -> str:
 56        """
 57        Retrieve original value and check the expiration hasn't passed.
 58        """
 59        result = self._signer.unsign(signed_value)
 60        value, timestamp = result.rsplit(self.sep, 1)
 61        ts = b62_decode(timestamp)
 62        if ts < time.time():
 63            raise SignatureExpired("Signature expired")
 64        return value
 65
 66    def sign_object(
 67        self,
 68        obj: Any,
 69        *,
 70        expires_in: int,
 71        serializer: type = JSONSerializer,
 72        compress: bool = False,
 73    ) -> str:
 74        """
 75        Return URL-safe, hmac signed base64 compressed JSON string.
 76
 77        If compress is True (not the default), check if compressing using zlib
 78        can save some space. Prepend a '.' to signify compression. This is
 79        included in the signature, to protect against zip bombs.
 80
 81        The serializer is expected to return a bytestring.
 82        """
 83        data = serializer().dumps(obj)
 84        # Flag for if it's been compressed or not.
 85        is_compressed = False
 86
 87        if compress:
 88            # Avoid zlib dependency unless compress is being used.
 89            compressed = zlib.compress(data)
 90            if len(compressed) < (len(data) - 1):
 91                data = compressed
 92                is_compressed = True
 93        base64d = b64_encode(data).decode()
 94        if is_compressed:
 95            base64d = "." + base64d
 96        return self.sign(base64d, expires_in)
 97
 98    def unsign_object(self, signed_obj: str, serializer: type = JSONSerializer) -> Any:
 99        # Signer.unsign() returns str but base64 and zlib compression operate
100        # on bytes.
101        base64d = self.unsign(signed_obj).encode()
102        decompress = base64d[:1] == b"."
103        if decompress:
104            # It's compressed; uncompress it first.
105            base64d = base64d[1:]
106        data = b64_decode(base64d)
107        if decompress:
108            data = zlib.decompress(data)
109        return serializer().loads(data)