summaryrefslogtreecommitdiffhomepage
path: root/libs/itsdangerous
diff options
context:
space:
mode:
authormorpheus65535 <[email protected]>2022-01-23 23:07:52 -0500
committermorpheus65535 <[email protected]>2022-01-23 23:07:52 -0500
commit0c3c5a02a75bc61b6bf6e303de20e11741d2afac (patch)
tree30ae1d524ffe5d54172b7a4a8445d90c3461e659 /libs/itsdangerous
parent36bf0d219d0432c20e6314e0ce752b36f4d88e3c (diff)
downloadbazarr-0c3c5a02a75bc61b6bf6e303de20e11741d2afac.tar.gz
bazarr-0c3c5a02a75bc61b6bf6e303de20e11741d2afac.zip
Upgraded vendored Python dependencies to the latest versions and removed the unused dependencies.v1.0.3-beta.16
Diffstat (limited to 'libs/itsdangerous')
-rw-r--r--libs/itsdangerous/__init__.py36
-rw-r--r--libs/itsdangerous/_compat.py46
-rw-r--r--libs/itsdangerous/_json.py34
-rw-r--r--libs/itsdangerous/encoding.py23
-rw-r--r--libs/itsdangerous/exc.py63
-rw-r--r--libs/itsdangerous/jws.py71
-rw-r--r--libs/itsdangerous/py.typed0
-rw-r--r--libs/itsdangerous/serializer.py260
-rw-r--r--libs/itsdangerous/signer.py212
-rw-r--r--libs/itsdangerous/timed.py146
-rw-r--r--libs/itsdangerous/url_safe.py25
11 files changed, 588 insertions, 328 deletions
diff --git a/libs/itsdangerous/__init__.py b/libs/itsdangerous/__init__.py
index 0fcd8c1b1..5010252b4 100644
--- a/libs/itsdangerous/__init__.py
+++ b/libs/itsdangerous/__init__.py
@@ -1,22 +1,22 @@
from ._json import json
-from .encoding import base64_decode
-from .encoding import base64_encode
-from .encoding import want_bytes
-from .exc import BadData
-from .exc import BadHeader
-from .exc import BadPayload
-from .exc import BadSignature
-from .exc import BadTimeSignature
-from .exc import SignatureExpired
+from .encoding import base64_decode as base64_decode
+from .encoding import base64_encode as base64_encode
+from .encoding import want_bytes as want_bytes
+from .exc import BadData as BadData
+from .exc import BadHeader as BadHeader
+from .exc import BadPayload as BadPayload
+from .exc import BadSignature as BadSignature
+from .exc import BadTimeSignature as BadTimeSignature
+from .exc import SignatureExpired as SignatureExpired
from .jws import JSONWebSignatureSerializer
from .jws import TimedJSONWebSignatureSerializer
-from .serializer import Serializer
-from .signer import HMACAlgorithm
-from .signer import NoneAlgorithm
-from .signer import Signer
-from .timed import TimedSerializer
-from .timed import TimestampSigner
-from .url_safe import URLSafeSerializer
-from .url_safe import URLSafeTimedSerializer
+from .serializer import Serializer as Serializer
+from .signer import HMACAlgorithm as HMACAlgorithm
+from .signer import NoneAlgorithm as NoneAlgorithm
+from .signer import Signer as Signer
+from .timed import TimedSerializer as TimedSerializer
+from .timed import TimestampSigner as TimestampSigner
+from .url_safe import URLSafeSerializer as URLSafeSerializer
+from .url_safe import URLSafeTimedSerializer as URLSafeTimedSerializer
-__version__ = "1.1.0"
+__version__ = "2.0.1"
diff --git a/libs/itsdangerous/_compat.py b/libs/itsdangerous/_compat.py
deleted file mode 100644
index 2291bce24..000000000
--- a/libs/itsdangerous/_compat.py
+++ /dev/null
@@ -1,46 +0,0 @@
-import decimal
-import hmac
-import numbers
-import sys
-
-PY2 = sys.version_info[0] == 2
-
-if PY2:
- from itertools import izip
-
- text_type = unicode # noqa: 821
-else:
- izip = zip
- text_type = str
-
-number_types = (numbers.Real, decimal.Decimal)
-
-
-def _constant_time_compare(val1, val2):
- """Return ``True`` if the two strings are equal, ``False``
- otherwise.
-
- The time taken is independent of the number of characters that
- match. Do not use this function for anything else than comparision
- with known length targets.
-
- This is should be implemented in C in order to get it completely
- right.
-
- This is an alias of :func:`hmac.compare_digest` on Python>=2.7,3.3.
- """
- len_eq = len(val1) == len(val2)
- if len_eq:
- result = 0
- left = val1
- else:
- result = 1
- left = val2
- for x, y in izip(bytearray(left), bytearray(val2)):
- result |= x ^ y
- return result == 0
-
-
-# Starting with 2.7/3.3 the standard library has a c-implementation for
-# constant time string compares.
-constant_time_compare = getattr(hmac, "compare_digest", _constant_time_compare)
diff --git a/libs/itsdangerous/_json.py b/libs/itsdangerous/_json.py
index 426b36e9a..9368da2ac 100644
--- a/libs/itsdangerous/_json.py
+++ b/libs/itsdangerous/_json.py
@@ -1,18 +1,34 @@
-try:
- import simplejson as json
-except ImportError:
- import json
+import json as _json
+import typing as _t
+from types import ModuleType
-class _CompactJSON(object):
+class _CompactJSON:
"""Wrapper around json module that strips whitespace."""
@staticmethod
- def loads(payload):
- return json.loads(payload)
+ def loads(payload: _t.Union[str, bytes]) -> _t.Any:
+ return _json.loads(payload)
@staticmethod
- def dumps(obj, **kwargs):
+ def dumps(obj: _t.Any, **kwargs: _t.Any) -> str:
kwargs.setdefault("ensure_ascii", False)
kwargs.setdefault("separators", (",", ":"))
- return json.dumps(obj, **kwargs)
+ return _json.dumps(obj, **kwargs)
+
+
+class DeprecatedJSON(ModuleType):
+ def __getattribute__(self, item: str) -> _t.Any:
+ import warnings
+
+ warnings.warn(
+ "Importing 'itsdangerous.json' is deprecated and will be"
+ " removed in ItsDangerous 2.1. Use Python's 'json' module"
+ " instead.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+ return getattr(_json, item)
+
+
+json = DeprecatedJSON("json")
diff --git a/libs/itsdangerous/encoding.py b/libs/itsdangerous/encoding.py
index 1e28969d2..cacc1c5f2 100644
--- a/libs/itsdangerous/encoding.py
+++ b/libs/itsdangerous/encoding.py
@@ -1,18 +1,23 @@
import base64
import string
import struct
+import typing as _t
-from ._compat import text_type
from .exc import BadData
+_t_str_bytes = _t.Union[str, bytes]
-def want_bytes(s, encoding="utf-8", errors="strict"):
- if isinstance(s, text_type):
+
+def want_bytes(
+ s: _t_str_bytes, encoding: str = "utf-8", errors: str = "strict"
+) -> bytes:
+ if isinstance(s, str):
s = s.encode(encoding, errors)
+
return s
-def base64_encode(string):
+def base64_encode(string: _t_str_bytes) -> bytes:
"""Base64 encode a string of bytes or text. The resulting bytes are
safe to use in URLs.
"""
@@ -20,7 +25,7 @@ def base64_encode(string):
return base64.urlsafe_b64encode(string).rstrip(b"=")
-def base64_decode(string):
+def base64_decode(string: _t_str_bytes) -> bytes:
"""Base64 decode a URL-safe string of bytes or text. The result is
bytes.
"""
@@ -34,16 +39,16 @@ def base64_decode(string):
# The alphabet used by base64.urlsafe_*
-_base64_alphabet = (string.ascii_letters + string.digits + "-_=").encode("ascii")
+_base64_alphabet = f"{string.ascii_letters}{string.digits}-_=".encode("ascii")
_int64_struct = struct.Struct(">Q")
_int_to_bytes = _int64_struct.pack
-_bytes_to_int = _int64_struct.unpack
+_bytes_to_int = _t.cast("_t.Callable[[bytes], _t.Tuple[int]]", _int64_struct.unpack)
-def int_to_bytes(num):
+def int_to_bytes(num: int) -> bytes:
return _int_to_bytes(num).lstrip(b"\x00")
-def bytes_to_int(bytestr):
+def bytes_to_int(bytestr: bytes) -> int:
return _bytes_to_int(bytestr.rjust(8, b"\x00"))[0]
diff --git a/libs/itsdangerous/exc.py b/libs/itsdangerous/exc.py
index 287d6917a..c38a6af52 100644
--- a/libs/itsdangerous/exc.py
+++ b/libs/itsdangerous/exc.py
@@ -1,42 +1,37 @@
-from ._compat import PY2
-from ._compat import text_type
+import typing as _t
+from datetime import datetime
+
+_t_opt_any = _t.Optional[_t.Any]
+_t_opt_exc = _t.Optional[Exception]
class BadData(Exception):
"""Raised if bad data of any sort was encountered. This is the base
- for all exceptions that itsdangerous defines.
+ for all exceptions that ItsDangerous defines.
.. versionadded:: 0.15
"""
- message = None
-
- def __init__(self, message):
- super(BadData, self).__init__(self, message)
+ def __init__(self, message: str):
+ super().__init__(message)
self.message = message
- def __str__(self):
- return text_type(self.message)
-
- if PY2:
- __unicode__ = __str__
-
- def __str__(self):
- return self.__unicode__().encode("utf-8")
+ def __str__(self) -> str:
+ return self.message
class BadSignature(BadData):
"""Raised if a signature does not match."""
- def __init__(self, message, payload=None):
- BadData.__init__(self, message)
+ def __init__(self, message: str, payload: _t_opt_any = None):
+ super().__init__(message)
#: The payload that failed the signature test. In some
#: situations you might still want to inspect this, even if
#: you know it was tampered with.
#:
#: .. versionadded:: 0.14
- self.payload = payload
+ self.payload: _t_opt_any = payload
class BadTimeSignature(BadSignature):
@@ -44,13 +39,21 @@ class BadTimeSignature(BadSignature):
of :class:`BadSignature`.
"""
- def __init__(self, message, payload=None, date_signed=None):
- BadSignature.__init__(self, message, payload)
+ def __init__(
+ self,
+ message: str,
+ payload: _t_opt_any = None,
+ date_signed: _t.Optional[datetime] = None,
+ ):
+ super().__init__(message, payload)
#: If the signature expired this exposes the date of when the
#: signature was created. This can be helpful in order to
#: tell the user how long a link has been gone stale.
#:
+ #: .. versionchanged:: 2.0
+ #: The datetime value is timezone-aware rather than naive.
+ #:
#: .. versionadded:: 0.14
self.date_signed = date_signed
@@ -69,16 +72,22 @@ class BadHeader(BadSignature):
.. versionadded:: 0.24
"""
- def __init__(self, message, payload=None, header=None, original_error=None):
- BadSignature.__init__(self, message, payload)
+ def __init__(
+ self,
+ message: str,
+ payload: _t_opt_any = None,
+ header: _t_opt_any = None,
+ original_error: _t_opt_exc = None,
+ ):
+ super().__init__(message, payload)
#: If the header is actually available but just malformed it
#: might be stored here.
- self.header = header
+ self.header: _t_opt_any = header
#: If available, the error that indicates why the payload was
#: not valid. This might be ``None``.
- self.original_error = original_error
+ self.original_error: _t_opt_exc = original_error
class BadPayload(BadData):
@@ -90,9 +99,9 @@ class BadPayload(BadData):
.. versionadded:: 0.15
"""
- def __init__(self, message, original_error=None):
- BadData.__init__(self, message)
+ def __init__(self, message: str, original_error: _t_opt_exc = None):
+ super().__init__(message)
#: If available, the error that indicates why the payload was
#: not valid. This might be ``None``.
- self.original_error = original_error
+ self.original_error: _t_opt_exc = original_error
diff --git a/libs/itsdangerous/jws.py b/libs/itsdangerous/jws.py
index 92e9ec8b4..2353a30fe 100644
--- a/libs/itsdangerous/jws.py
+++ b/libs/itsdangerous/jws.py
@@ -1,10 +1,12 @@
import hashlib
import time
+import warnings
from datetime import datetime
+from datetime import timezone
+from decimal import Decimal
+from numbers import Real
-from ._compat import number_types
from ._json import _CompactJSON
-from ._json import json
from .encoding import base64_decode
from .encoding import base64_encode
from .encoding import want_bytes
@@ -21,6 +23,10 @@ from .signer import NoneAlgorithm
class JSONWebSignatureSerializer(Serializer):
"""This serializer implements JSON Web Signature (JWS) support. Only
supports the JWS Compact Serialization.
+
+ .. deprecated:: 2.0
+ Will be removed in ItsDangerous 2.1. Use a dedicated library
+ such as authlib.
"""
jws_algorithms = {
@@ -45,25 +51,36 @@ class JSONWebSignatureSerializer(Serializer):
signer_kwargs=None,
algorithm_name=None,
):
- Serializer.__init__(
- self,
- secret_key=secret_key,
+ warnings.warn(
+ "JWS support is deprecated and will be removed in"
+ " ItsDangerous 2.1. Use a dedicated JWS/JWT library such as"
+ " authlib.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+ super().__init__(
+ secret_key,
salt=salt,
serializer=serializer,
serializer_kwargs=serializer_kwargs,
signer=signer,
signer_kwargs=signer_kwargs,
)
+
if algorithm_name is None:
algorithm_name = self.default_algorithm
+
self.algorithm_name = algorithm_name
self.algorithm = self.make_algorithm(algorithm_name)
def load_payload(self, payload, serializer=None, return_header=False):
payload = want_bytes(payload)
+
if b"." not in payload:
raise BadPayload('No "." found in value')
+
base64d_header, base64d_payload = payload.split(b".", 1)
+
try:
json_header = base64_decode(base64d_header)
except Exception as e:
@@ -71,6 +88,7 @@ class JSONWebSignatureSerializer(Serializer):
"Could not base64 decode the header because of an exception",
original_error=e,
)
+
try:
json_payload = base64_decode(base64d_payload)
except Exception as e:
@@ -78,18 +96,23 @@ class JSONWebSignatureSerializer(Serializer):
"Could not base64 decode the payload because of an exception",
original_error=e,
)
+
try:
- header = Serializer.load_payload(self, json_header, serializer=json)
+ header = super().load_payload(json_header, serializer=_CompactJSON)
except BadData as e:
raise BadHeader(
"Could not unserialize header because it was malformed",
original_error=e,
)
+
if not isinstance(header, dict):
raise BadHeader("Header payload is not a JSON object", header=header)
- payload = Serializer.load_payload(self, json_payload, serializer=serializer)
+
+ payload = super().load_payload(json_payload, serializer=serializer)
+
if return_header:
return payload, header
+
return payload
def dump_payload(self, header, obj):
@@ -110,11 +133,14 @@ class JSONWebSignatureSerializer(Serializer):
def make_signer(self, salt=None, algorithm=None):
if salt is None:
salt = self.salt
+
key_derivation = "none" if salt is None else None
+
if algorithm is None:
algorithm = self.algorithm
+
return self.signer(
- self.secret_key,
+ self.secret_keys,
salt=salt,
sep=".",
key_derivation=key_derivation,
@@ -143,10 +169,13 @@ class JSONWebSignatureSerializer(Serializer):
self.make_signer(salt, self.algorithm).unsign(want_bytes(s)),
return_header=True,
)
+
if header.get("alg") != self.algorithm_name:
raise BadHeader("Algorithm mismatch", header=header, payload=payload)
+
if return_header:
return payload, header
+
return payload
def loads_unsafe(self, s, salt=None, return_header=False):
@@ -169,13 +198,15 @@ class TimedJSONWebSignatureSerializer(JSONWebSignatureSerializer):
DEFAULT_EXPIRES_IN = 3600
def __init__(self, secret_key, expires_in=None, **kwargs):
- JSONWebSignatureSerializer.__init__(self, secret_key, **kwargs)
+ super().__init__(secret_key, **kwargs)
+
if expires_in is None:
expires_in = self.DEFAULT_EXPIRES_IN
+
self.expires_in = expires_in
def make_header(self, header_fields):
- header = JSONWebSignatureSerializer.make_header(self, header_fields)
+ header = super().make_header(header_fields)
iat = self.now()
exp = iat + self.expires_in
header["iat"] = iat
@@ -183,18 +214,18 @@ class TimedJSONWebSignatureSerializer(JSONWebSignatureSerializer):
return header
def loads(self, s, salt=None, return_header=False):
- payload, header = JSONWebSignatureSerializer.loads(
- self, s, salt, return_header=True
- )
+ payload, header = super().loads(s, salt, return_header=True)
if "exp" not in header:
raise BadSignature("Missing expiry date", payload=payload)
int_date_error = BadHeader("Expiry date is not an IntDate", payload=payload)
+
try:
header["exp"] = int(header["exp"])
except ValueError:
raise int_date_error
+
if header["exp"] < 0:
raise int_date_error
@@ -207,12 +238,22 @@ class TimedJSONWebSignatureSerializer(JSONWebSignatureSerializer):
if return_header:
return payload, header
+
return payload
def get_issue_date(self, header):
+ """If the header contains the ``iat`` field, return the date the
+ signature was issued, as a timezone-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.
+ """
rv = header.get("iat")
- if isinstance(rv, number_types):
- return datetime.utcfromtimestamp(int(rv))
+
+ if isinstance(rv, (Real, Decimal)):
+ return datetime.fromtimestamp(int(rv), tz=timezone.utc)
def now(self):
return int(time.time())
diff --git a/libs/itsdangerous/py.typed b/libs/itsdangerous/py.typed
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/libs/itsdangerous/py.typed
diff --git a/libs/itsdangerous/serializer.py b/libs/itsdangerous/serializer.py
index 12c20f46d..36a73fbaa 100644
--- a/libs/itsdangerous/serializer.py
+++ b/libs/itsdangerous/serializer.py
@@ -1,108 +1,142 @@
-import hashlib
+import json
+import typing as _t
-from ._compat import text_type
-from ._json import json
from .encoding import want_bytes
from .exc import BadPayload
from .exc import BadSignature
+from .signer import _make_keys_list
from .signer import Signer
+_t_str_bytes = _t.Union[str, bytes]
+_t_opt_str_bytes = _t.Optional[_t_str_bytes]
+_t_kwargs = _t.Dict[str, _t.Any]
+_t_opt_kwargs = _t.Optional[_t_kwargs]
+_t_signer = _t.Type[Signer]
+_t_fallbacks = _t.List[_t.Union[_t_kwargs, _t.Tuple[_t_signer, _t_kwargs], _t_signer]]
+_t_load_unsafe = _t.Tuple[bool, _t.Any]
+_t_secret_key = _t.Union[_t.Iterable[_t_str_bytes], _t_str_bytes]
-def is_text_serializer(serializer):
- """Checks whether a serializer generates text or binary."""
- return isinstance(serializer.dumps({}), text_type)
-
-
-class Serializer(object):
- """This class provides a serialization interface on top of the
- signer. It provides a similar API to json/pickle and other modules
- but is structured differently internally. If you want to change the
- underlying implementation for parsing and loading you have to
- override the :meth:`load_payload` and :meth:`dump_payload`
- functions.
-
- This implementation uses simplejson if available for dumping and
- loading and will fall back to the standard library's json module if
- it's not available.
-
- You do not need to subclass this class in order to switch out or
- customize the :class:`.Signer`. You can instead pass a different
- class to the constructor as well as keyword arguments as a dict that
- should be forwarded.
-
- .. code-block:: python
-
- s = Serializer(signer_kwargs={'key_derivation': 'hmac'})
-
- You may want to upgrade the signing parameters without invalidating
- existing signatures that are in use. Fallback signatures can be
- given that will be tried if unsigning with the current signer fails.
-
- Fallback signers can be defined by providing a list of
- ``fallback_signers``. Each item can be one of the following: a
- signer class (which is instantiated with ``signer_kwargs``,
- ``salt``, and ``secret_key``), a tuple
- ``(signer_class, signer_kwargs)``, or a dict of ``signer_kwargs``.
-
- For example, this is a serializer that signs using SHA-512, but will
- unsign using either SHA-512 or SHA1:
-
- .. code-block:: python
- s = Serializer(
- signer_kwargs={"digest_method": hashlib.sha512},
- fallback_signers=[{"digest_method": hashlib.sha1}]
- )
-
- .. versionchanged:: 0.14:
- The ``signer`` and ``signer_kwargs`` parameters were added to
- the constructor.
-
- .. versionchanged:: 1.1.0:
+def is_text_serializer(serializer: _t.Any) -> bool:
+ """Checks whether a serializer generates text or binary."""
+ return isinstance(serializer.dumps({}), str)
+
+
+class Serializer:
+ """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.
"""
- #: If a serializer module or class is not passed to the constructor
- #: this one is picked up. This currently defaults to :mod:`json`.
- default_serializer = json
+ #: 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: _t.Any = json
- #: The default :class:`Signer` class that is being used by this
- #: serializer.
- #:
- #: .. versionadded:: 0.14
- default_signer = Signer
+ #: The default ``Signer`` class to instantiate when signing data.
+ #: The default is :class:`itsdangerous.signer.Signer`.
+ default_signer: _t_signer = Signer
- #: The default fallback signers.
- default_fallback_signers = [{"digest_method": hashlib.sha512}]
+ #: The default fallback signers to try when unsigning fails.
+ default_fallback_signers: _t_fallbacks = []
def __init__(
self,
- secret_key,
- salt=b"itsdangerous",
- serializer=None,
- serializer_kwargs=None,
- signer=None,
- signer_kwargs=None,
- fallback_signers=None,
+ secret_key: _t_secret_key,
+ salt: _t_opt_str_bytes = b"itsdangerous",
+ serializer: _t.Any = None,
+ serializer_kwargs: _t_opt_kwargs = None,
+ signer: _t.Optional[_t_signer] = None,
+ signer_kwargs: _t_opt_kwargs = None,
+ fallback_signers: _t.Optional[_t_fallbacks] = None,
):
- self.secret_key = want_bytes(secret_key)
- self.salt = want_bytes(salt)
+ #: 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: _t.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 = serializer
- self.is_text_serializer = is_text_serializer(serializer)
+
+ self.serializer: _t.Any = serializer
+ self.is_text_serializer: bool = is_text_serializer(serializer)
+
if signer is None:
signer = self.default_signer
- self.signer = signer
- self.signer_kwargs = signer_kwargs or {}
+
+ self.signer: _t_signer = signer
+ self.signer_kwargs: _t_kwargs = signer_kwargs or {}
+
if fallback_signers is None:
fallback_signers = list(self.default_fallback_signers or ())
- self.fallback_signers = fallback_signers
- self.serializer_kwargs = serializer_kwargs or {}
- def load_payload(self, payload, serializer=None):
+ self.fallback_signers: _t_fallbacks = fallback_signers
+ self.serializer_kwargs: _t_kwargs = 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: _t.Optional[_t.Any] = 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
@@ -114,9 +148,11 @@ class Serializer(object):
is_text = self.is_text_serializer
else:
is_text = is_text_serializer(serializer)
+
try:
if is_text:
- payload = payload.decode("utf-8")
+ return serializer.loads(payload.decode("utf-8"))
+
return serializer.loads(payload)
except Exception as e:
raise BadPayload(
@@ -125,74 +161,87 @@ class Serializer(object):
original_error=e,
)
- def dump_payload(self, obj):
+ 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=None):
+ def make_signer(self, salt: _t_opt_str_bytes = 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_key, salt=salt, **self.signer_kwargs)
- def iter_unsigners(self, salt=None):
+ return self.signer(self.secret_keys, salt=salt, **self.signer_kwargs)
+
+ def iter_unsigners(self, salt: _t_opt_str_bytes = None) -> _t.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 type(fallback) is dict:
+ if isinstance(fallback, dict):
kwargs = fallback
fallback = self.signer
- elif type(fallback) is tuple:
+ elif isinstance(fallback, tuple):
fallback, kwargs = fallback
else:
kwargs = self.signer_kwargs
- yield fallback(self.secret_key, salt=salt, **kwargs)
- def dumps(self, obj, salt=None):
+ for secret_key in self.secret_keys:
+ yield fallback(secret_key, salt=salt, **kwargs)
+
+ def dumps(self, obj: _t.Any, salt: _t_opt_str_bytes = None) -> _t_str_bytes:
"""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:
- rv = rv.decode("utf-8")
+ return rv.decode("utf-8")
+
return rv
- def dump(self, obj, f, salt=None):
+ def dump(self, obj: _t.Any, f: _t.IO, salt: _t_opt_str_bytes = 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, salt=None):
+ def loads(
+ self, s: _t_str_bytes, salt: _t_opt_str_bytes = 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 last_exception
- def load(self, f, salt=None):
+ raise _t.cast(BadSignature, last_exception)
+
+ def load(self, f: _t.IO, salt: _t_opt_str_bytes = None) -> _t.Any:
"""Like :meth:`loads` but loads from a file."""
return self.loads(f.read(), salt)
- def loads_unsafe(self, s, salt=None):
+ def loads_unsafe(
+ self, s: _t_str_bytes, salt: _t_opt_str_bytes = None
+ ) -> _t_load_unsafe:
"""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,
@@ -208,26 +257,39 @@ class Serializer(object):
"""
return self._loads_unsafe_impl(s, salt)
- def _loads_unsafe_impl(self, s, salt, load_kwargs=None, load_payload_kwargs=None):
+ def _loads_unsafe_impl(
+ self,
+ s: _t_str_bytes,
+ salt: _t_opt_str_bytes,
+ load_kwargs: _t_opt_kwargs = None,
+ load_payload_kwargs: _t_opt_kwargs = None,
+ ) -> _t_load_unsafe:
"""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 or {}))
+ 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 or {})),
+ self.load_payload(e.payload, **load_payload_kwargs),
)
except BadPayload:
return False, None
- def load_unsafe(self, f, *args, **kwargs):
+ def load_unsafe(self, f: _t.IO, salt: _t_opt_str_bytes = None) -> _t_load_unsafe:
"""Like :meth:`loads_unsafe` but loads from a file.
.. versionadded:: 0.15
"""
- return self.loads_unsafe(f.read(), *args, **kwargs)
+ return self.loads_unsafe(f.read(), salt=salt)
diff --git a/libs/itsdangerous/signer.py b/libs/itsdangerous/signer.py
index 6bddc0396..aa12005e9 100644
--- a/libs/itsdangerous/signer.py
+++ b/libs/itsdangerous/signer.py
@@ -1,28 +1,32 @@
import hashlib
import hmac
+import typing as _t
-from ._compat import constant_time_compare
from .encoding import _base64_alphabet
from .encoding import base64_decode
from .encoding import base64_encode
from .encoding import want_bytes
from .exc import BadSignature
+_t_str_bytes = _t.Union[str, bytes]
+_t_opt_str_bytes = _t.Optional[_t_str_bytes]
+_t_secret_key = _t.Union[_t.Iterable[_t_str_bytes], _t_str_bytes]
-class SigningAlgorithm(object):
+
+class SigningAlgorithm:
"""Subclasses must implement :meth:`get_signature` to provide
signature generation functionality.
"""
- def get_signature(self, key, value):
+ 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, value, sig):
+ def verify_signature(self, key: bytes, value: bytes, sig: bytes) -> bool:
"""Verifies the given signature matches the expected
signature.
"""
- return constant_time_compare(sig, self.get_signature(key, value))
+ return hmac.compare_digest(sig, self.get_signature(key, value))
class NoneAlgorithm(SigningAlgorithm):
@@ -30,7 +34,7 @@ class NoneAlgorithm(SigningAlgorithm):
returns an empty signature.
"""
- def get_signature(self, key, value):
+ def get_signature(self, key: bytes, value: bytes) -> bytes:
return b""
@@ -40,135 +44,209 @@ class HMACAlgorithm(SigningAlgorithm):
#: 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 = staticmethod(hashlib.sha1)
+ default_digest_method: _t.Any = staticmethod(hashlib.sha1)
- def __init__(self, digest_method=None):
+ def __init__(self, digest_method: _t.Any = None):
if digest_method is None:
digest_method = self.default_digest_method
- self.digest_method = digest_method
- def get_signature(self, key, value):
+ 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()
-class Signer(object):
- """This class can sign and unsign bytes, validating the signature
- provided.
+def _make_keys_list(secret_key: _t_secret_key) -> _t.List[bytes]:
+ if isinstance(secret_key, (str, bytes)):
+ return [want_bytes(secret_key)]
- Salt can be used to namespace the hash, so that a signed string is
- only valid for a given namespace. Leaving this at the default value
- or re-using a salt value across different parts of your application
- where the same signed value in one part can mean something different
- in another part is a security risk.
+ return [want_bytes(s) for s in secret_key]
- See :ref:`the-salt` for an example of what the salt is doing and how
- you can utilize it.
- .. versionadded:: 0.14
- ``key_derivation`` and ``digest_method`` were added as arguments
- to the class constructor.
+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.
- .. versionadded:: 0.18
+ :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 digest method to use for the signer. This defaults to
- #: SHA1 but can be changed to any other function in the hashlib
- #: module.
+ #: 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 = staticmethod(hashlib.sha1)
+ default_digest_method: _t.Any = staticmethod(hashlib.sha1)
- #: Controls how the key is derived. The default is Django-style
- #: concatenation. Possible values are ``concat``, ``django-concat``
- #: and ``hmac``. This is used for deriving a key from the secret key
- #: with an added salt.
+ #: 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 = "django-concat"
+ default_key_derivation: str = "django-concat"
def __init__(
self,
- secret_key,
- salt=None,
- sep=".",
- key_derivation=None,
- digest_method=None,
- algorithm=None,
+ secret_key: _t_secret_key,
+ salt: _t_opt_str_bytes = b"itsdangerous.Signer",
+ sep: _t_str_bytes = b".",
+ key_derivation: _t.Optional[str] = None,
+ digest_method: _t.Optional[_t.Any] = None,
+ algorithm: _t.Optional[SigningAlgorithm] = None,
):
- self.secret_key = want_bytes(secret_key)
- self.sep = want_bytes(sep)
+ #: 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: _t.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. Alphanumeric"
- " characters and `-_=` must not be used."
+ " contained in the signature itself. ASCII letters,"
+ " digits, and '-_=' must not be used."
)
- self.salt = "itsdangerous.Signer" if salt is None else salt
+
+ 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 = key_derivation
+
+ self.key_derivation: str = key_derivation
+
if digest_method is None:
digest_method = self.default_digest_method
- self.digest_method = digest_method
+
+ self.digest_method: _t.Any = digest_method
+
if algorithm is None:
algorithm = HMACAlgorithm(self.digest_method)
- self.algorithm = algorithm
- def derive_key(self):
+ 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: _t_opt_str_bytes = 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.
"""
- salt = want_bytes(self.salt)
+ if secret_key is None:
+ secret_key = self.secret_keys[-1]
+ else:
+ secret_key = want_bytes(secret_key)
+
if self.key_derivation == "concat":
- return self.digest_method(salt + self.secret_key).digest()
+ return _t.cast(bytes, self.digest_method(self.salt + secret_key).digest())
elif self.key_derivation == "django-concat":
- return self.digest_method(salt + b"signer" + self.secret_key).digest()
+ return _t.cast(
+ bytes, self.digest_method(self.salt + b"signer" + secret_key).digest()
+ )
elif self.key_derivation == "hmac":
- mac = hmac.new(self.secret_key, digestmod=self.digest_method)
- mac.update(salt)
+ mac = hmac.new(secret_key, digestmod=self.digest_method)
+ mac.update(self.salt)
return mac.digest()
elif self.key_derivation == "none":
- return self.secret_key
+ return secret_key
else:
raise TypeError("Unknown key derivation method")
- def get_signature(self, value):
+ def get_signature(self, value: _t_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):
+ def sign(self, value: _t_str_bytes) -> bytes:
"""Signs the given string."""
- return want_bytes(value) + want_bytes(self.sep) + self.get_signature(value)
+ value = want_bytes(value)
+ return value + self.sep + self.get_signature(value)
- def verify_signature(self, value, sig):
+ def verify_signature(self, value: _t_str_bytes, sig: _t_str_bytes) -> bool:
"""Verifies the signature for the given value."""
- key = self.derive_key()
try:
sig = base64_decode(sig)
except Exception:
return False
- return self.algorithm.verify_signature(key, value, sig)
- def unsign(self, signed_value):
+ 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: _t_str_bytes) -> bytes:
"""Unsigns the given string."""
signed_value = want_bytes(signed_value)
- sep = want_bytes(self.sep)
- if sep not in signed_value:
- raise BadSignature("No %r found in value" % self.sep)
- value, sig = signed_value.rsplit(sep, 1)
+
+ 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("Signature %r does not match" % sig, payload=value)
- def validate(self, signed_value):
+ raise BadSignature(f"Signature {sig!r} does not match", payload=value)
+
+ def validate(self, signed_value: _t_str_bytes) -> bool:
"""Only validates the given signed value. Returns ``True`` if
the signature exists and is valid.
"""
diff --git a/libs/itsdangerous/timed.py b/libs/itsdangerous/timed.py
index 4c117e419..5ea957f9d 100644
--- a/libs/itsdangerous/timed.py
+++ b/libs/itsdangerous/timed.py
@@ -1,7 +1,9 @@
import time
+import typing
+import typing as _t
from datetime import datetime
+from datetime import timezone
-from ._compat import text_type
from .encoding import base64_decode
from .encoding import base64_encode
from .encoding import bytes_to_int
@@ -13,6 +15,13 @@ from .exc import SignatureExpired
from .serializer import Serializer
from .signer import Signer
+_t_str_bytes = _t.Union[str, bytes]
+_t_opt_str_bytes = _t.Optional[_t_str_bytes]
+_t_opt_int = _t.Optional[int]
+
+if _t.TYPE_CHECKING:
+ import typing_extensions as _te
+
class TimestampSigner(Signer):
"""Works like the regular :class:`.Signer` but also records the time
@@ -21,19 +30,23 @@ class TimestampSigner(Signer):
unsigning failed because the signature is expired.
"""
- def get_timestamp(self):
+ 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):
- """Used to convert the timestamp from :meth:`get_timestamp` into
- a datetime object.
+ 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.utcfromtimestamp(ts)
+ return datetime.fromtimestamp(ts, tz=timezone.utc)
- def sign(self, value):
+ def sign(self, value: _t_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()))
@@ -41,19 +54,50 @@ class TimestampSigner(Signer):
value = value + sep + timestamp
return value + sep + self.get_signature(value)
- def unsign(self, value, max_age=None, return_timestamp=False):
+ # Ignore overlapping signatures check, return_timestamp is the only
+ # parameter that affects the return type.
+
+ @typing.overload
+ def unsign( # type: ignore
+ self,
+ signed_value: _t_str_bytes,
+ max_age: _t_opt_int = None,
+ return_timestamp: "_te.Literal[False]" = False,
+ ) -> bytes:
+ ...
+
+ @typing.overload
+ def unsign(
+ self,
+ signed_value: _t_str_bytes,
+ max_age: _t_opt_int = None,
+ return_timestamp: "_te.Literal[True]" = True,
+ ) -> _t.Tuple[bytes, datetime]:
+ ...
+
+ def unsign(
+ self,
+ signed_value: _t_str_bytes,
+ max_age: _t_opt_int = None,
+ return_timestamp: bool = False,
+ ) -> _t.Union[_t.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 a naive
+ 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 = Signer.unsign(self, value)
+ 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
@@ -64,41 +108,55 @@ class TimestampSigner(Signer):
if sep not in result:
if sig_error:
raise sig_error
+
raise BadTimeSignature("timestamp missing", payload=result)
- value, timestamp = result.rsplit(sep, 1)
+ value, ts_bytes = result.rsplit(sep, 1)
+ ts_int: _t_opt_int = None
+ ts_dt: _t.Optional[datetime] = None
+
try:
- timestamp = bytes_to_int(base64_decode(timestamp))
+ ts_int = bytes_to_int(base64_decode(ts_bytes))
except Exception:
- timestamp = None
+ 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:
- raise BadTimeSignature(
- text_type(sig_error), payload=value, date_signed=timestamp
- )
+ if ts_int is not None:
+ ts_dt = self.timestamp_to_datetime(ts_int)
+
+ 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 timestamp is None:
+ 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() - timestamp
+ age = self.get_timestamp() - ts_int
+
if age > max_age:
raise SignatureExpired(
- "Signature age %s > %s seconds" % (age, max_age),
+ f"Signature age {age} > {max_age} seconds",
payload=value,
- date_signed=self.timestamp_to_datetime(timestamp),
+ 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(timestamp)
+ return value, self.timestamp_to_datetime(ts_int)
+
return value
- def validate(self, signed_value, max_age=None):
+ def validate(self, signed_value: _t_str_bytes, max_age: _t_opt_int = None) -> bool:
"""Only validates the given signed value. Returns ``True`` if
the signature exists and is valid."""
try:
@@ -113,9 +171,23 @@ class TimedSerializer(Serializer):
:class:`.Signer`.
"""
- default_signer = TimestampSigner
+ default_signer: _t.Type[TimestampSigner] = TimestampSigner
+
+ def iter_unsigners(
+ self, salt: _t_opt_str_bytes = None
+ ) -> _t.Iterator[TimestampSigner]:
+ return _t.cast("_t.Iterator[TimestampSigner]", super().iter_unsigners(salt))
+
+ # TODO: Signature is incompatible because parameters were added
+ # before salt.
- def loads(self, s, max_age=None, return_timestamp=False, salt=None):
+ def loads( # type: ignore
+ self,
+ s: _t_str_bytes,
+ max_age: _t_opt_int = None,
+ return_timestamp: bool = False,
+ salt: _t_opt_str_bytes = 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
@@ -125,23 +197,31 @@ class TimedSerializer(Serializer):
"""
s = want_bytes(s)
last_exception = None
+
for signer in self.iter_unsigners(salt):
try:
- base64d, timestamp = signer.unsign(s, max_age, return_timestamp=True)
+ 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
- # If we get a signature expired it means we could read the
- # signature but it's invalid. In that case we do not want to
- # try the next signer.
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 last_exception
- def loads_unsafe(self, s, max_age=None, salt=None):
- load_kwargs = {"max_age": max_age}
- load_payload_kwargs = {}
- return self._loads_unsafe_impl(s, salt, load_kwargs, load_payload_kwargs)
+ raise _t.cast(BadSignature, last_exception)
+
+ def loads_unsafe( # type: ignore
+ self,
+ s: _t_str_bytes,
+ max_age: _t_opt_int = None,
+ salt: _t_opt_str_bytes = None,
+ ) -> _t.Tuple[bool, _t.Any]:
+ return self._loads_unsafe_impl(s, salt, load_kwargs={"max_age": max_age})
diff --git a/libs/itsdangerous/url_safe.py b/libs/itsdangerous/url_safe.py
index fcaa01123..f76fa24f7 100644
--- a/libs/itsdangerous/url_safe.py
+++ b/libs/itsdangerous/url_safe.py
@@ -1,3 +1,4 @@
+import typing as _t
import zlib
from ._json import _CompactJSON
@@ -8,7 +9,7 @@ from .serializer import Serializer
from .timed import TimedSerializer
-class URLSafeSerializerMixin(object):
+class URLSafeSerializerMixin(Serializer):
"""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.
@@ -16,11 +17,19 @@ class URLSafeSerializerMixin(object):
default_serializer = _CompactJSON
- def load_payload(self, payload, *args, **kwargs):
+ def load_payload(
+ self,
+ payload: bytes,
+ *args: _t.Any,
+ serializer: _t.Optional[_t.Any] = 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:
@@ -28,6 +37,7 @@ class URLSafeSerializerMixin(object):
"Could not base64 decode the payload because of an exception",
original_error=e,
)
+
if decompress:
try:
json = zlib.decompress(json)
@@ -36,18 +46,23 @@ class URLSafeSerializerMixin(object):
"Could not zlib decompress the payload before decoding the payload",
original_error=e,
)
- return super(URLSafeSerializerMixin, self).load_payload(json, *args, **kwargs)
- def dump_payload(self, obj):
- json = super(URLSafeSerializerMixin, self).dump_payload(obj)
+ 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