Загрузить файлы в «venv/Lib/site-packages/itsdangerous»

venv/Lib/site-packages/itsdangerous/serializer.py

@@ -0,0 +1,406 @@
+from __future__ import annotations
+
+import collections.abc as cabc
+import json
+import typing as t
+
+from .encoding import want_bytes
+from .exc import BadPayload
+from .exc import BadSignature
+from .signer import _make_keys_list
+from .signer import Signer
+
+if t.TYPE_CHECKING:
+    import typing_extensions as te
+
+    # This should be either be str or bytes. To avoid having to specify the
+    # bound type, it falls back to a union if structural matching fails.
+    _TSerialized = te.TypeVar(
+        "_TSerialized", bound=t.Union[str, bytes], default=t.Union[str, bytes]
+    )
+else:
+    # Still available at runtime on Python < 3.13, but without the default.
+    _TSerialized = t.TypeVar("_TSerialized", bound=t.Union[str, bytes])
+
+
+class _PDataSerializer(t.Protocol[_TSerialized]):
+    def loads(self, payload: _TSerialized, /) -> t.Any: ...
+    # A signature with additional arguments is not handled correctly by type
+    # checkers right now, so an overload is used below for serializers that
+    # don't match this strict protocol.
+    def dumps(self, obj: t.Any, /) -> _TSerialized: ...
+
+
+# Use TypeIs once it's available in typing_extensions or 3.13.
+def is_text_serializer(
+    serializer: _PDataSerializer[t.Any],
+) -> te.TypeGuard[_PDataSerializer[str]]:
+    """Checks whether a serializer generates text or binary."""
+    return isinstance(serializer.dumps({}), str)
+
+
+class Serializer(t.Generic[_TSerialized]):
+    """A serializer wraps a :class:`~itsdangerous.signer.Signer` to
+    enable serializing and securely signing data other than bytes. It
+    can unsign to verify that the data hasn't been changed.
+
+    The serializer provides :meth:`dumps` and :meth:`loads`, similar to
+    :mod:`json`, and by default uses :mod:`json` internally to serialize
+    the data to bytes.
+
+    The secret key should be a random string of ``bytes`` and should not
+    be saved to code or version control. Different salts should be used
+    to distinguish signing in different contexts. See :doc:`/concepts`
+    for information about the security of the secret key and salt.
+
+    :param secret_key: The secret key to sign and verify with. Can be a
+        list of keys, oldest to newest, to support key rotation.
+    :param salt: Extra key to combine with ``secret_key`` to distinguish
+        signatures in different contexts.
+    :param serializer: An object that provides ``dumps`` and ``loads``
+        methods for serializing data to a string. Defaults to
+        :attr:`default_serializer`, which defaults to :mod:`json`.
+    :param serializer_kwargs: Keyword arguments to pass when calling
+        ``serializer.dumps``.
+    :param signer: A ``Signer`` class to instantiate when signing data.
+        Defaults to :attr:`default_signer`, which defaults to
+        :class:`~itsdangerous.signer.Signer`.
+    :param signer_kwargs: Keyword arguments to pass when instantiating
+        the ``Signer`` class.
+    :param fallback_signers: List of signer parameters to try when
+        unsigning with the default signer fails. Each item can be a dict
+        of ``signer_kwargs``, a ``Signer`` class, or a tuple of
+        ``(signer, signer_kwargs)``. Defaults to
+        :attr:`default_fallback_signers`.
+
+    .. versionchanged:: 2.0
+        Added support for key rotation by passing a list to
+        ``secret_key``.
+
+    .. versionchanged:: 2.0
+        Removed the default SHA-512 fallback signer from
+        ``default_fallback_signers``.
+
+    .. versionchanged:: 1.1
+        Added support for ``fallback_signers`` and configured a default
+        SHA-512 fallback. This fallback is for users who used the yanked
+        1.0.0 release which defaulted to SHA-512.
+
+    .. versionchanged:: 0.14
+        The ``signer`` and ``signer_kwargs`` parameters were added to
+        the constructor.
+    """
+
+    #: The default serialization module to use to serialize data to a
+    #: string internally. The default is :mod:`json`, but can be changed
+    #: to any object that provides ``dumps`` and ``loads`` methods.
+    default_serializer: _PDataSerializer[t.Any] = json
+
+    #: The default ``Signer`` class to instantiate when signing data.
+    #: The default is :class:`itsdangerous.signer.Signer`.
+    default_signer: type[Signer] = Signer
+
+    #: The default fallback signers to try when unsigning fails.
+    default_fallback_signers: list[
+        dict[str, t.Any] | tuple[type[Signer], dict[str, t.Any]] | type[Signer]
+    ] = []
+
+    # Serializer[str] if no data serializer is provided, or if it returns str.
+    @t.overload
+    def __init__(
+        self: Serializer[str],
+        secret_key: str | bytes | cabc.Iterable[str] | cabc.Iterable[bytes],
+        salt: str | bytes | None = b"itsdangerous",
+        serializer: None | _PDataSerializer[str] = None,
+        serializer_kwargs: dict[str, t.Any] | None = None,
+        signer: type[Signer] | None = None,
+        signer_kwargs: dict[str, t.Any] | None = None,
+        fallback_signers: list[
+            dict[str, t.Any] | tuple[type[Signer], dict[str, t.Any]] | type[Signer]
+        ]
+        | None = None,
+    ): ...
+
+    # Serializer[bytes] with a bytes data serializer positional argument.
+    @t.overload
+    def __init__(
+        self: Serializer[bytes],
+        secret_key: str | bytes | cabc.Iterable[str] | cabc.Iterable[bytes],
+        salt: str | bytes | None,
+        serializer: _PDataSerializer[bytes],
+        serializer_kwargs: dict[str, t.Any] | None = None,
+        signer: type[Signer] | None = None,
+        signer_kwargs: dict[str, t.Any] | None = None,
+        fallback_signers: list[
+            dict[str, t.Any] | tuple[type[Signer], dict[str, t.Any]] | type[Signer]
+        ]
+        | None = None,
+    ): ...
+
+    # Serializer[bytes] with a bytes data serializer keyword argument.
+    @t.overload
+    def __init__(
+        self: Serializer[bytes],
+        secret_key: str | bytes | cabc.Iterable[str] | cabc.Iterable[bytes],
+        salt: str | bytes | None = b"itsdangerous",
+        *,
+        serializer: _PDataSerializer[bytes],
+        serializer_kwargs: dict[str, t.Any] | None = None,
+        signer: type[Signer] | None = None,
+        signer_kwargs: dict[str, t.Any] | None = None,
+        fallback_signers: list[
+            dict[str, t.Any] | tuple[type[Signer], dict[str, t.Any]] | type[Signer]
+        ]
+        | None = None,
+    ): ...
+
+    # Fall back with a positional argument. If the strict signature of
+    # _PDataSerializer doesn't match, fall back to a union, requiring the user
+    # to specify the type.
+    @t.overload
+    def __init__(
+        self,
+        secret_key: str | bytes | cabc.Iterable[str] | cabc.Iterable[bytes],
+        salt: str | bytes | None,
+        serializer: t.Any,
+        serializer_kwargs: dict[str, t.Any] | None = None,
+        signer: type[Signer] | None = None,
+        signer_kwargs: dict[str, t.Any] | None = None,
+        fallback_signers: list[
+            dict[str, t.Any] | tuple[type[Signer], dict[str, t.Any]] | type[Signer]
+        ]
+        | None = None,
+    ): ...
+
+    # Fall back with a keyword argument.
+    @t.overload
+    def __init__(
+        self,
+        secret_key: str | bytes | cabc.Iterable[str] | cabc.Iterable[bytes],
+        salt: str | bytes | None = b"itsdangerous",
+        *,
+        serializer: t.Any,
+        serializer_kwargs: dict[str, t.Any] | None = None,
+        signer: type[Signer] | None = None,
+        signer_kwargs: dict[str, t.Any] | None = None,
+        fallback_signers: list[
+            dict[str, t.Any] | tuple[type[Signer], dict[str, t.Any]] | type[Signer]
+        ]
+        | None = None,
+    ): ...
+
+    def __init__(
+        self,
+        secret_key: str | bytes | cabc.Iterable[str] | cabc.Iterable[bytes],
+        salt: str | bytes | None = b"itsdangerous",
+        serializer: t.Any | None = None,
+        serializer_kwargs: dict[str, t.Any] | None = None,
+        signer: type[Signer] | None = None,
+        signer_kwargs: dict[str, t.Any] | None = None,
+        fallback_signers: list[
+            dict[str, t.Any] | tuple[type[Signer], dict[str, t.Any]] | type[Signer]
+        ]
+        | None = None,
+    ):
+        #: The list of secret keys to try for verifying signatures, from
+        #: oldest to newest. The newest (last) key is used for signing.
+        #:
+        #: This allows a key rotation system to keep a list of allowed
+        #: keys and remove expired ones.
+        self.secret_keys: list[bytes] = _make_keys_list(secret_key)
+
+        if salt is not None:
+            salt = want_bytes(salt)
+            # if salt is None then the signer's default is used
+
+        self.salt = salt
+
+        if serializer is None:
+            serializer = self.default_serializer
+
+        self.serializer: _PDataSerializer[_TSerialized] = serializer
+        self.is_text_serializer: bool = is_text_serializer(serializer)
+
+        if signer is None:
+            signer = self.default_signer
+
+        self.signer: type[Signer] = signer
+        self.signer_kwargs: dict[str, t.Any] = signer_kwargs or {}
+
+        if fallback_signers is None:
+            fallback_signers = list(self.default_fallback_signers)
+
+        self.fallback_signers: list[
+            dict[str, t.Any] | tuple[type[Signer], dict[str, t.Any]] | type[Signer]
+        ] = fallback_signers
+        self.serializer_kwargs: dict[str, t.Any] = serializer_kwargs or {}
+
+    @property
+    def secret_key(self) -> bytes:
+        """The newest (last) entry in the :attr:`secret_keys` list. This
+        is for compatibility from before key rotation support was added.
+        """
+        return self.secret_keys[-1]
+
+    def load_payload(
+        self, payload: bytes, serializer: _PDataSerializer[t.Any] | None = None
+    ) -> t.Any:
+        """Loads the encoded object. This function raises
+        :class:`.BadPayload` if the payload is not valid. The
+        ``serializer`` parameter can be used to override the serializer
+        stored on the class. The encoded ``payload`` should always be
+        bytes.
+        """
+        if serializer is None:
+            use_serializer = self.serializer
+            is_text = self.is_text_serializer
+        else:
+            use_serializer = serializer
+            is_text = is_text_serializer(serializer)
+
+        try:
+            if is_text:
+                return use_serializer.loads(payload.decode("utf-8"))  # type: ignore[arg-type]
+
+            return use_serializer.loads(payload)  # type: ignore[arg-type]
+        except Exception as e:
+            raise BadPayload(
+                "Could not load the payload because an exception"
+                " occurred on unserializing the data.",
+                original_error=e,
+            ) from e
+
+    def dump_payload(self, obj: t.Any) -> bytes:
+        """Dumps the encoded object. The return value is always bytes.
+        If the internal serializer returns text, the value will be
+        encoded as UTF-8.
+        """
+        return want_bytes(self.serializer.dumps(obj, **self.serializer_kwargs))
+
+    def make_signer(self, salt: str | bytes | None = None) -> Signer:
+        """Creates a new instance of the signer to be used. The default
+        implementation uses the :class:`.Signer` base class.
+        """
+        if salt is None:
+            salt = self.salt
+
+        return self.signer(self.secret_keys, salt=salt, **self.signer_kwargs)
+
+    def iter_unsigners(self, salt: str | bytes | None = None) -> cabc.Iterator[Signer]:
+        """Iterates over all signers to be tried for unsigning. Starts
+        with the configured signer, then constructs each signer
+        specified in ``fallback_signers``.
+        """
+        if salt is None:
+            salt = self.salt
+
+        yield self.make_signer(salt)
+
+        for fallback in self.fallback_signers:
+            if isinstance(fallback, dict):
+                kwargs = fallback
+                fallback = self.signer
+            elif isinstance(fallback, tuple):
+                fallback, kwargs = fallback
+            else:
+                kwargs = self.signer_kwargs
+
+            for secret_key in self.secret_keys:
+                yield fallback(secret_key, salt=salt, **kwargs)
+
+    def dumps(self, obj: t.Any, salt: str | bytes | None = None) -> _TSerialized:
+        """Returns a signed string serialized with the internal
+        serializer. The return value can be either a byte or unicode
+        string depending on the format of the internal serializer.
+        """
+        payload = want_bytes(self.dump_payload(obj))
+        rv = self.make_signer(salt).sign(payload)
+
+        if self.is_text_serializer:
+            return rv.decode("utf-8")  # type: ignore[return-value]
+
+        return rv  # type: ignore[return-value]
+
+    def dump(self, obj: t.Any, f: t.IO[t.Any], salt: str | bytes | None = None) -> None:
+        """Like :meth:`dumps` but dumps into a file. The file handle has
+        to be compatible with what the internal serializer expects.
+        """
+        f.write(self.dumps(obj, salt))
+
+    def loads(
+        self, s: str | bytes, salt: str | bytes | None = None, **kwargs: t.Any
+    ) -> t.Any:
+        """Reverse of :meth:`dumps`. Raises :exc:`.BadSignature` if the
+        signature validation fails.
+        """
+        s = want_bytes(s)
+        last_exception = None
+
+        for signer in self.iter_unsigners(salt):
+            try:
+                return self.load_payload(signer.unsign(s))
+            except BadSignature as err:
+                last_exception = err
+
+        raise t.cast(BadSignature, last_exception)
+
+    def load(self, f: t.IO[t.Any], salt: str | bytes | None = None) -> t.Any:
+        """Like :meth:`loads` but loads from a file."""
+        return self.loads(f.read(), salt)
+
+    def loads_unsafe(
+        self, s: str | bytes, salt: str | bytes | None = None
+    ) -> tuple[bool, t.Any]:
+        """Like :meth:`loads` but without verifying the signature. This
+        is potentially very dangerous to use depending on how your
+        serializer works. The return value is ``(signature_valid,
+        payload)`` instead of just the payload. The first item will be a
+        boolean that indicates if the signature is valid. This function
+        never fails.
+
+        Use it for debugging only and if you know that your serializer
+        module is not exploitable (for example, do not use it with a
+        pickle serializer).
+
+        .. versionadded:: 0.15
+        """
+        return self._loads_unsafe_impl(s, salt)
+
+    def _loads_unsafe_impl(
+        self,
+        s: str | bytes,
+        salt: str | bytes | None,
+        load_kwargs: dict[str, t.Any] | None = None,
+        load_payload_kwargs: dict[str, t.Any] | None = None,
+    ) -> tuple[bool, t.Any]:
+        """Low level helper function to implement :meth:`loads_unsafe`
+        in serializer subclasses.
+        """
+        if load_kwargs is None:
+            load_kwargs = {}
+
+        try:
+            return True, self.loads(s, salt=salt, **load_kwargs)
+        except BadSignature as e:
+            if e.payload is None:
+                return False, None
+
+            if load_payload_kwargs is None:
+                load_payload_kwargs = {}
+
+            try:
+                return (
+                    False,
+                    self.load_payload(e.payload, **load_payload_kwargs),
+                )
+            except BadPayload:
+                return False, None
+
+    def load_unsafe(
+        self, f: t.IO[t.Any], salt: str | bytes | None = None
+    ) -> tuple[bool, t.Any]:
+        """Like :meth:`loads_unsafe` but loads from a file.
+
+        .. versionadded:: 0.15
+        """
+        return self.loads_unsafe(f.read(), salt=salt)

venv/Lib/site-packages/itsdangerous/signer.py

@@ -0,0 +1,266 @@
+from __future__ import annotations
+
+import collections.abc as cabc
+import hashlib
+import hmac
+import typing as t
+
+from .encoding import _base64_alphabet
+from .encoding import base64_decode
+from .encoding import base64_encode
+from .encoding import want_bytes
+from .exc import BadSignature
+
+
+class SigningAlgorithm:
+    """Subclasses must implement :meth:`get_signature` to provide
+    signature generation functionality.
+    """
+
+    def get_signature(self, key: bytes, value: bytes) -> bytes:
+        """Returns the signature for the given key and value."""
+        raise NotImplementedError()
+
+    def verify_signature(self, key: bytes, value: bytes, sig: bytes) -> bool:
+        """Verifies the given signature matches the expected
+        signature.
+        """
+        return hmac.compare_digest(sig, self.get_signature(key, value))
+
+
+class NoneAlgorithm(SigningAlgorithm):
+    """Provides an algorithm that does not perform any signing and
+    returns an empty signature.
+    """
+
+    def get_signature(self, key: bytes, value: bytes) -> bytes:
+        return b""
+
+
+def _lazy_sha1(string: bytes = b"") -> t.Any:
+    """Don't access ``hashlib.sha1`` until runtime. FIPS builds may not include
+    SHA-1, in which case the import and use as a default would fail before the
+    developer can configure something else.
+    """
+    return hashlib.sha1(string)
+
+
+class HMACAlgorithm(SigningAlgorithm):
+    """Provides signature generation using HMACs."""
+
+    #: The digest method to use with the MAC algorithm. This defaults to
+    #: SHA1, but can be changed to any other function in the hashlib
+    #: module.
+    default_digest_method: t.Any = staticmethod(_lazy_sha1)
+
+    def __init__(self, digest_method: t.Any = None):
+        if digest_method is None:
+            digest_method = self.default_digest_method
+
+        self.digest_method: t.Any = digest_method
+
+    def get_signature(self, key: bytes, value: bytes) -> bytes:
+        mac = hmac.new(key, msg=value, digestmod=self.digest_method)
+        return mac.digest()
+
+
+def _make_keys_list(
+    secret_key: str | bytes | cabc.Iterable[str] | cabc.Iterable[bytes],
+) -> list[bytes]:
+    if isinstance(secret_key, (str, bytes)):
+        return [want_bytes(secret_key)]
+
+    return [want_bytes(s) for s in secret_key]  # pyright: ignore
+
+
+class Signer:
+    """A signer securely signs bytes, then unsigns them to verify that
+    the value hasn't been changed.
+
+    The secret key should be a random string of ``bytes`` and should not
+    be saved to code or version control. Different salts should be used
+    to distinguish signing in different contexts. See :doc:`/concepts`
+    for information about the security of the secret key and salt.
+
+    :param secret_key: The secret key to sign and verify with. Can be a
+        list of keys, oldest to newest, to support key rotation.
+    :param salt: Extra key to combine with ``secret_key`` to distinguish
+        signatures in different contexts.
+    :param sep: Separator between the signature and value.
+    :param key_derivation: How to derive the signing key from the secret
+        key and salt. Possible values are ``concat``, ``django-concat``,
+        or ``hmac``. Defaults to :attr:`default_key_derivation`, which
+        defaults to ``django-concat``.
+    :param digest_method: Hash function to use when generating the HMAC
+        signature. Defaults to :attr:`default_digest_method`, which
+        defaults to :func:`hashlib.sha1`. Note that the security of the
+        hash alone doesn't apply when used intermediately in HMAC.
+    :param algorithm: A :class:`SigningAlgorithm` instance to use
+        instead of building a default :class:`HMACAlgorithm` with the
+        ``digest_method``.
+
+    .. versionchanged:: 2.0
+        Added support for key rotation by passing a list to
+        ``secret_key``.
+
+    .. versionchanged:: 0.18
+        ``algorithm`` was added as an argument to the class constructor.
+
+    .. versionchanged:: 0.14
+        ``key_derivation`` and ``digest_method`` were added as arguments
+        to the class constructor.
+    """
+
+    #: The default digest method to use for the signer. The default is
+    #: :func:`hashlib.sha1`, but can be changed to any :mod:`hashlib` or
+    #: compatible object. Note that the security of the hash alone
+    #: doesn't apply when used intermediately in HMAC.
+    #:
+    #: .. versionadded:: 0.14
+    default_digest_method: t.Any = staticmethod(_lazy_sha1)
+
+    #: The default scheme to use to derive the signing key from the
+    #: secret key and salt. The default is ``django-concat``. Possible
+    #: values are ``concat``, ``django-concat``, and ``hmac``.
+    #:
+    #: .. versionadded:: 0.14
+    default_key_derivation: str = "django-concat"
+
+    def __init__(
+        self,
+        secret_key: str | bytes | cabc.Iterable[str] | cabc.Iterable[bytes],
+        salt: str | bytes | None = b"itsdangerous.Signer",
+        sep: str | bytes = b".",
+        key_derivation: str | None = None,
+        digest_method: t.Any | None = None,
+        algorithm: SigningAlgorithm | None = None,
+    ):
+        #: The list of secret keys to try for verifying signatures, from
+        #: oldest to newest. The newest (last) key is used for signing.
+        #:
+        #: This allows a key rotation system to keep a list of allowed
+        #: keys and remove expired ones.
+        self.secret_keys: list[bytes] = _make_keys_list(secret_key)
+        self.sep: bytes = want_bytes(sep)
+
+        if self.sep in _base64_alphabet:
+            raise ValueError(
+                "The given separator cannot be used because it may be"
+                " contained in the signature itself. ASCII letters,"
+                " digits, and '-_=' must not be used."
+            )
+
+        if salt is not None:
+            salt = want_bytes(salt)
+        else:
+            salt = b"itsdangerous.Signer"
+
+        self.salt = salt
+
+        if key_derivation is None:
+            key_derivation = self.default_key_derivation
+
+        self.key_derivation: str = key_derivation
+
+        if digest_method is None:
+            digest_method = self.default_digest_method
+
+        self.digest_method: t.Any = digest_method
+
+        if algorithm is None:
+            algorithm = HMACAlgorithm(self.digest_method)
+
+        self.algorithm: SigningAlgorithm = algorithm
+
+    @property
+    def secret_key(self) -> bytes:
+        """The newest (last) entry in the :attr:`secret_keys` list. This
+        is for compatibility from before key rotation support was added.
+        """
+        return self.secret_keys[-1]
+
+    def derive_key(self, secret_key: str | bytes | None = None) -> bytes:
+        """This method is called to derive the key. The default key
+        derivation choices can be overridden here. Key derivation is not
+        intended to be used as a security method to make a complex key
+        out of a short password. Instead you should use large random
+        secret keys.
+
+        :param secret_key: A specific secret key to derive from.
+            Defaults to the last item in :attr:`secret_keys`.
+
+        .. versionchanged:: 2.0
+            Added the ``secret_key`` parameter.
+        """
+        if secret_key is None:
+            secret_key = self.secret_keys[-1]
+        else:
+            secret_key = want_bytes(secret_key)
+
+        if self.key_derivation == "concat":
+            return t.cast(bytes, self.digest_method(self.salt + secret_key).digest())
+        elif self.key_derivation == "django-concat":
+            return t.cast(
+                bytes, self.digest_method(self.salt + b"signer" + secret_key).digest()
+            )
+        elif self.key_derivation == "hmac":
+            mac = hmac.new(secret_key, digestmod=self.digest_method)
+            mac.update(self.salt)
+            return mac.digest()
+        elif self.key_derivation == "none":
+            return secret_key
+        else:
+            raise TypeError("Unknown key derivation method")
+
+    def get_signature(self, value: str | bytes) -> bytes:
+        """Returns the signature for the given value."""
+        value = want_bytes(value)
+        key = self.derive_key()
+        sig = self.algorithm.get_signature(key, value)
+        return base64_encode(sig)
+
+    def sign(self, value: str | bytes) -> bytes:
+        """Signs the given string."""
+        value = want_bytes(value)
+        return value + self.sep + self.get_signature(value)
+
+    def verify_signature(self, value: str | bytes, sig: str | bytes) -> bool:
+        """Verifies the signature for the given value."""
+        try:
+            sig = base64_decode(sig)
+        except Exception:
+            return False
+
+        value = want_bytes(value)
+
+        for secret_key in reversed(self.secret_keys):
+            key = self.derive_key(secret_key)
+
+            if self.algorithm.verify_signature(key, value, sig):
+                return True
+
+        return False
+
+    def unsign(self, signed_value: str | bytes) -> bytes:
+        """Unsigns the given string."""
+        signed_value = want_bytes(signed_value)
+
+        if self.sep not in signed_value:
+            raise BadSignature(f"No {self.sep!r} found in value")
+
+        value, sig = signed_value.rsplit(self.sep, 1)
+
+        if self.verify_signature(value, sig):
+            return value
+
+        raise BadSignature(f"Signature {sig!r} does not match", payload=value)
+
+    def validate(self, signed_value: str | bytes) -> bool:
+        """Only validates the given signed value. Returns ``True`` if
+        the signature exists and is valid.
+        """
+        try:
+            self.unsign(signed_value)
+            return True
+        except BadSignature:
+            return False

venv/Lib/site-packages/itsdangerous/timed.py

@@ -0,0 +1,228 @@
+from __future__ import annotations
+
+import collections.abc as cabc
+import time
+import typing as t
+from datetime import datetime
+from datetime import timezone
+
+from .encoding import base64_decode
+from .encoding import base64_encode
+from .encoding import bytes_to_int
+from .encoding import int_to_bytes
+from .encoding import want_bytes
+from .exc import BadSignature
+from .exc import BadTimeSignature
+from .exc import SignatureExpired
+from .serializer import _TSerialized
+from .serializer import Serializer
+from .signer import Signer
+
+
+class TimestampSigner(Signer):
+    """Works like the regular :class:`.Signer` but also records the time
+    of the signing and can be used to expire signatures. The
+    :meth:`unsign` method can raise :exc:`.SignatureExpired` if the
+    unsigning failed because the signature is expired.
+    """
+
+    def get_timestamp(self) -> int:
+        """Returns the current timestamp. The function must return an
+        integer.
+        """
+        return int(time.time())
+
+    def timestamp_to_datetime(self, ts: int) -> datetime:
+        """Convert the timestamp from :meth:`get_timestamp` into an
+        aware :class`datetime.datetime` in UTC.
+
+        .. versionchanged:: 2.0
+            The timestamp is returned as a timezone-aware ``datetime``
+            in UTC rather than a naive ``datetime`` assumed to be UTC.
+        """
+        return datetime.fromtimestamp(ts, tz=timezone.utc)
+
+    def sign(self, value: str | bytes) -> bytes:
+        """Signs the given string and also attaches time information."""
+        value = want_bytes(value)
+        timestamp = base64_encode(int_to_bytes(self.get_timestamp()))
+        sep = want_bytes(self.sep)
+        value = value + sep + timestamp
+        return value + sep + self.get_signature(value)
+
+    # Ignore overlapping signatures check, return_timestamp is the only
+    # parameter that affects the return type.
+
+    @t.overload
+    def unsign(  # type: ignore[overload-overlap]
+        self,
+        signed_value: str | bytes,
+        max_age: int | None = None,
+        return_timestamp: t.Literal[False] = False,
+    ) -> bytes: ...
+
+    @t.overload
+    def unsign(
+        self,
+        signed_value: str | bytes,
+        max_age: int | None = None,
+        return_timestamp: t.Literal[True] = True,
+    ) -> tuple[bytes, datetime]: ...
+
+    def unsign(
+        self,
+        signed_value: str | bytes,
+        max_age: int | None = None,
+        return_timestamp: bool = False,
+    ) -> tuple[bytes, datetime] | bytes:
+        """Works like the regular :meth:`.Signer.unsign` but can also
+        validate the time. See the base docstring of the class for
+        the general behavior. If ``return_timestamp`` is ``True`` the
+        timestamp of the signature will be returned as an aware
+        :class:`datetime.datetime` object in UTC.
+
+        .. versionchanged:: 2.0
+            The timestamp is returned as a timezone-aware ``datetime``
+            in UTC rather than a naive ``datetime`` assumed to be UTC.
+        """
+        try:
+            result = super().unsign(signed_value)
+            sig_error = None
+        except BadSignature as e:
+            sig_error = e
+            result = e.payload or b""
+
+        sep = want_bytes(self.sep)
+
+        # If there is no timestamp in the result there is something
+        # seriously wrong. In case there was a signature error, we raise
+        # that one directly, otherwise we have a weird situation in
+        # which we shouldn't have come except someone uses a time-based
+        # serializer on non-timestamp data, so catch that.
+        if sep not in result:
+            if sig_error:
+                raise sig_error
+
+            raise BadTimeSignature("timestamp missing", payload=result)
+
+        value, ts_bytes = result.rsplit(sep, 1)
+        ts_int: int | None = None
+        ts_dt: datetime | None = None
+
+        try:
+            ts_int = bytes_to_int(base64_decode(ts_bytes))
+        except Exception:
+            pass
+
+        # Signature is *not* okay. Raise a proper error now that we have
+        # split the value and the timestamp.
+        if sig_error is not None:
+            if ts_int is not None:
+                try:
+                    ts_dt = self.timestamp_to_datetime(ts_int)
+                except (ValueError, OSError, OverflowError) as exc:
+                    # Windows raises OSError
+                    # 32-bit raises OverflowError
+                    raise BadTimeSignature(
+                        "Malformed timestamp", payload=value
+                    ) from exc
+
+            raise BadTimeSignature(str(sig_error), payload=value, date_signed=ts_dt)
+
+        # Signature was okay but the timestamp is actually not there or
+        # malformed. Should not happen, but we handle it anyway.
+        if ts_int is None:
+            raise BadTimeSignature("Malformed timestamp", payload=value)
+
+        # Check timestamp is not older than max_age
+        if max_age is not None:
+            age = self.get_timestamp() - ts_int
+
+            if age > max_age:
+                raise SignatureExpired(
+                    f"Signature age {age} > {max_age} seconds",
+                    payload=value,
+                    date_signed=self.timestamp_to_datetime(ts_int),
+                )
+
+            if age < 0:
+                raise SignatureExpired(
+                    f"Signature age {age} < 0 seconds",
+                    payload=value,
+                    date_signed=self.timestamp_to_datetime(ts_int),
+                )
+
+        if return_timestamp:
+            return value, self.timestamp_to_datetime(ts_int)
+
+        return value
+
+    def validate(self, signed_value: str | bytes, max_age: int | None = None) -> bool:
+        """Only validates the given signed value. Returns ``True`` if
+        the signature exists and is valid."""
+        try:
+            self.unsign(signed_value, max_age=max_age)
+            return True
+        except BadSignature:
+            return False
+
+
+class TimedSerializer(Serializer[_TSerialized]):
+    """Uses :class:`TimestampSigner` instead of the default
+    :class:`.Signer`.
+    """
+
+    default_signer: type[TimestampSigner] = TimestampSigner
+
+    def iter_unsigners(
+        self, salt: str | bytes | None = None
+    ) -> cabc.Iterator[TimestampSigner]:
+        return t.cast("cabc.Iterator[TimestampSigner]", super().iter_unsigners(salt))
+
+    # TODO: Signature is incompatible because parameters were added
+    #  before salt.
+
+    def loads(  # type: ignore[override]
+        self,
+        s: str | bytes,
+        max_age: int | None = None,
+        return_timestamp: bool = False,
+        salt: str | bytes | None = None,
+    ) -> t.Any:
+        """Reverse of :meth:`dumps`, raises :exc:`.BadSignature` if the
+        signature validation fails. If a ``max_age`` is provided it will
+        ensure the signature is not older than that time in seconds. In
+        case the signature is outdated, :exc:`.SignatureExpired` is
+        raised. All arguments are forwarded to the signer's
+        :meth:`~TimestampSigner.unsign` method.
+        """
+        s = want_bytes(s)
+        last_exception = None
+
+        for signer in self.iter_unsigners(salt):
+            try:
+                base64d, timestamp = signer.unsign(
+                    s, max_age=max_age, return_timestamp=True
+                )
+                payload = self.load_payload(base64d)
+
+                if return_timestamp:
+                    return payload, timestamp
+
+                return payload
+            except SignatureExpired:
+                # The signature was unsigned successfully but was
+                # expired. Do not try the next signer.
+                raise
+            except BadSignature as err:
+                last_exception = err
+
+        raise t.cast(BadSignature, last_exception)
+
+    def loads_unsafe(  # type: ignore[override]
+        self,
+        s: str | bytes,
+        max_age: int | None = None,
+        salt: str | bytes | None = None,
+    ) -> tuple[bool, t.Any]:
+        return self._loads_unsafe_impl(s, salt, load_kwargs={"max_age": max_age})

venv/Lib/site-packages/itsdangerous/url_safe.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import typing as t
+import zlib
+
+from ._json import _CompactJSON
+from .encoding import base64_decode
+from .encoding import base64_encode
+from .exc import BadPayload
+from .serializer import _PDataSerializer
+from .serializer import Serializer
+from .timed import TimedSerializer
+
+
+class URLSafeSerializerMixin(Serializer[str]):
+    """Mixed in with a regular serializer it will attempt to zlib
+    compress the string to make it shorter if necessary. It will also
+    base64 encode the string so that it can safely be placed in a URL.
+    """
+
+    default_serializer: _PDataSerializer[str] = _CompactJSON
+
+    def load_payload(
+        self,
+        payload: bytes,
+        *args: t.Any,
+        serializer: t.Any | None = None,
+        **kwargs: t.Any,
+    ) -> t.Any:
+        decompress = False
+
+        if payload.startswith(b"."):
+            payload = payload[1:]
+            decompress = True
+
+        try:
+            json = base64_decode(payload)
+        except Exception as e:
+            raise BadPayload(
+                "Could not base64 decode the payload because of an exception",
+                original_error=e,
+            ) from e
+
+        if decompress:
+            try:
+                json = zlib.decompress(json)
+            except Exception as e:
+                raise BadPayload(
+                    "Could not zlib decompress the payload before decoding the payload",
+                    original_error=e,
+                ) from e
+
+        return super().load_payload(json, *args, **kwargs)
+
+    def dump_payload(self, obj: t.Any) -> bytes:
+        json = super().dump_payload(obj)
+        is_compressed = False
+        compressed = zlib.compress(json)
+
+        if len(compressed) < (len(json) - 1):
+            json = compressed
+            is_compressed = True
+
+        base64d = base64_encode(json)
+
+        if is_compressed:
+            base64d = b"." + base64d
+
+        return base64d
+
+
+class URLSafeSerializer(URLSafeSerializerMixin, Serializer[str]):
+    """Works like :class:`.Serializer` but dumps and loads into a URL
+    safe string consisting of the upper and lowercase character of the
+    alphabet as well as ``'_'``, ``'-'`` and ``'.'``.
+    """
+
+
+class URLSafeTimedSerializer(URLSafeSerializerMixin, TimedSerializer[str]):
+    """Works like :class:`.TimedSerializer` but dumps and loads into a
+    URL safe string consisting of the upper and lowercase character of
+    the alphabet as well as ``'_'``, ``'-'`` and ``'.'``.
+    """