Commit c884c12b authored by Artem Baranovskyi's avatar Artem Baranovskyi
Browse files

Final commit.

parent 81a9e9f3
from collections import defaultdict, deque
from itertools import filterfalse
def unique_everseen(iterable, key=None):
"List unique elements, preserving order. Remember all elements ever seen."
# unique_everseen('AAAABBBCCDAABBB') --> A B C D
# unique_everseen('ABBCcAD', str.lower) --> A B C D
seen = set()
seen_add = seen.add
if key is None:
for element in filterfalse(seen.__contains__, iterable):
seen_add(element)
yield element
else:
for element in iterable:
k = key(element)
if k not in seen:
seen_add(k)
yield element
# copied from more_itertools 8.8
def always_iterable(obj, base_type=(str, bytes)):
"""If *obj* is iterable, return an iterator over its items::
>>> obj = (1, 2, 3)
>>> list(always_iterable(obj))
[1, 2, 3]
If *obj* is not iterable, return a one-item iterable containing *obj*::
>>> obj = 1
>>> list(always_iterable(obj))
[1]
If *obj* is ``None``, return an empty iterable:
>>> obj = None
>>> list(always_iterable(None))
[]
By default, binary and text strings are not considered iterable::
>>> obj = 'foo'
>>> list(always_iterable(obj))
['foo']
If *base_type* is set, objects for which ``isinstance(obj, base_type)``
returns ``True`` won't be considered iterable.
>>> obj = {'a': 1}
>>> list(always_iterable(obj)) # Iterate over the dict's keys
['a']
>>> list(always_iterable(obj, base_type=dict)) # Treat dicts as a unit
[{'a': 1}]
Set *base_type* to ``None`` to avoid any special handling and treat objects
Python considers iterable as iterable:
>>> obj = 'foo'
>>> list(always_iterable(obj, base_type=None))
['f', 'o', 'o']
"""
if obj is None:
return iter(())
if (base_type is not None) and isinstance(obj, base_type):
return iter((obj,))
try:
return iter(obj)
except TypeError:
return iter((obj,))
# Copied from more_itertools 10.3
class bucket:
"""Wrap *iterable* and return an object that buckets the iterable into
child iterables based on a *key* function.
>>> iterable = ['a1', 'b1', 'c1', 'a2', 'b2', 'c2', 'b3']
>>> s = bucket(iterable, key=lambda x: x[0]) # Bucket by 1st character
>>> sorted(list(s)) # Get the keys
['a', 'b', 'c']
>>> a_iterable = s['a']
>>> next(a_iterable)
'a1'
>>> next(a_iterable)
'a2'
>>> list(s['b'])
['b1', 'b2', 'b3']
The original iterable will be advanced and its items will be cached until
they are used by the child iterables. This may require significant storage.
By default, attempting to select a bucket to which no items belong will
exhaust the iterable and cache all values.
If you specify a *validator* function, selected buckets will instead be
checked against it.
>>> from itertools import count
>>> it = count(1, 2) # Infinite sequence of odd numbers
>>> key = lambda x: x % 10 # Bucket by last digit
>>> validator = lambda x: x in {1, 3, 5, 7, 9} # Odd digits only
>>> s = bucket(it, key=key, validator=validator)
>>> 2 in s
False
>>> list(s[2])
[]
"""
def __init__(self, iterable, key, validator=None):
self._it = iter(iterable)
self._key = key
self._cache = defaultdict(deque)
self._validator = validator or (lambda x: True)
def __contains__(self, value):
if not self._validator(value):
return False
try:
item = next(self[value])
except StopIteration:
return False
else:
self._cache[value].appendleft(item)
return True
def _get_values(self, value):
"""
Helper to yield items from the parent iterator that match *value*.
Items that don't match are stored in the local cache as they
are encountered.
"""
while True:
# If we've cached some items that match the target value, emit
# the first one and evict it from the cache.
if self._cache[value]:
yield self._cache[value].popleft()
# Otherwise we need to advance the parent iterator to search for
# a matching item, caching the rest.
else:
while True:
try:
item = next(self._it)
except StopIteration:
return
item_value = self._key(item)
if item_value == value:
yield item
break
elif self._validator(item_value):
self._cache[item_value].append(item)
def __iter__(self):
for item in self._it:
item_value = self._key(item)
if self._validator(item_value):
self._cache[item_value].append(item)
yield from self._cache.keys()
def __getitem__(self, value):
if not self._validator(value):
return iter(())
return self._get_values(value)
from __future__ import annotations
import os
from typing import (
Any,
Dict,
Iterator,
List,
Optional,
Protocol,
TypeVar,
Union,
overload,
)
_T = TypeVar("_T")
class PackageMetadata(Protocol):
def __len__(self) -> int: ... # pragma: no cover
def __contains__(self, item: str) -> bool: ... # pragma: no cover
def __getitem__(self, key: str) -> str: ... # pragma: no cover
def __iter__(self) -> Iterator[str]: ... # pragma: no cover
@overload
def get(
self, name: str, failobj: None = None
) -> Optional[str]: ... # pragma: no cover
@overload
def get(self, name: str, failobj: _T) -> Union[str, _T]: ... # pragma: no cover
# overload per python/importlib_metadata#435
@overload
def get_all(
self, name: str, failobj: None = None
) -> Optional[List[Any]]: ... # pragma: no cover
@overload
def get_all(self, name: str, failobj: _T) -> Union[List[Any], _T]:
"""
Return all values associated with a possibly multi-valued key.
"""
@property
def json(self) -> Dict[str, Union[str, List[str]]]:
"""
A JSON-compatible form of the metadata.
"""
class SimplePath(Protocol):
"""
A minimal subset of pathlib.Path required by Distribution.
"""
def joinpath(
self, other: Union[str, os.PathLike[str]]
) -> SimplePath: ... # pragma: no cover
def __truediv__(
self, other: Union[str, os.PathLike[str]]
) -> SimplePath: ... # pragma: no cover
@property
def parent(self) -> SimplePath: ... # pragma: no cover
def read_text(self, encoding=None) -> str: ... # pragma: no cover
def read_bytes(self) -> bytes: ... # pragma: no cover
def exists(self) -> bool: ... # pragma: no cover
import re
from ._functools import method_cache
# from jaraco.text 3.5
class FoldedCase(str):
"""
A case insensitive string class; behaves just like str
except compares equal when the only variation is case.
>>> s = FoldedCase('hello world')
>>> s == 'Hello World'
True
>>> 'Hello World' == s
True
>>> s != 'Hello World'
False
>>> s.index('O')
4
>>> s.split('O')
['hell', ' w', 'rld']
>>> sorted(map(FoldedCase, ['GAMMA', 'alpha', 'Beta']))
['alpha', 'Beta', 'GAMMA']
Sequence membership is straightforward.
>>> "Hello World" in [s]
True
>>> s in ["Hello World"]
True
You may test for set inclusion, but candidate and elements
must both be folded.
>>> FoldedCase("Hello World") in {s}
True
>>> s in {FoldedCase("Hello World")}
True
String inclusion works as long as the FoldedCase object
is on the right.
>>> "hello" in FoldedCase("Hello World")
True
But not if the FoldedCase object is on the left:
>>> FoldedCase('hello') in 'Hello World'
False
In that case, use in_:
>>> FoldedCase('hello').in_('Hello World')
True
>>> FoldedCase('hello') > FoldedCase('Hello')
False
"""
def __lt__(self, other):
return self.lower() < other.lower()
def __gt__(self, other):
return self.lower() > other.lower()
def __eq__(self, other):
return self.lower() == other.lower()
def __ne__(self, other):
return self.lower() != other.lower()
def __hash__(self):
return hash(self.lower())
def __contains__(self, other):
return super().lower().__contains__(other.lower())
def in_(self, other):
"Does self appear in other?"
return self in FoldedCase(other)
# cache lower since it's likely to be called frequently.
@method_cache
def lower(self):
return super().lower()
def index(self, sub):
return self.lower().index(sub.lower())
def split(self, splitter=' ', maxsplit=0):
pattern = re.compile(re.escape(splitter), re.I)
return pattern.split(self, maxsplit)
import os
import pathlib
import sys
import types
def wrap(path): # pragma: no cover
"""
Workaround for https://github.com/python/cpython/issues/84538
to add backward compatibility for walk_up=True.
An example affected package is dask-labextension, which uses
jupyter-packaging to install JupyterLab javascript files outside
of site-packages.
"""
def relative_to(root, *, walk_up=False):
return pathlib.Path(os.path.relpath(path, root))
return types.SimpleNamespace(relative_to=relative_to)
relative_fix = wrap if sys.version_info < (3, 12) else lambda x: x
"""
Compatibility layer with Python 3.8/3.9
"""
from typing import TYPE_CHECKING, Any, Optional
if TYPE_CHECKING: # pragma: no cover
# Prevent circular imports on runtime.
from .. import Distribution, EntryPoint
else:
Distribution = EntryPoint = Any
def normalized_name(dist: Distribution) -> Optional[str]:
"""
Honor name normalization for distributions that don't provide ``_normalized_name``.
"""
try:
return dist._normalized_name
except AttributeError:
from .. import Prepared # -> delay to prevent circular imports.
return Prepared.normalize(getattr(dist, "name", None) or dist.metadata['Name'])
def ep_matches(ep: EntryPoint, **params) -> bool:
"""
Workaround for ``EntryPoint`` objects without the ``matches`` method.
"""
try:
return ep.matches(**params)
except AttributeError:
from .. import EntryPoint # -> delay to prevent circular imports.
# Reconstruct the EntryPoint object to make sure it is compatible.
return EntryPoint(ep.name, ep.value, ep.group).matches(**params)
import sys
from . import Distribution
def inspect(path):
print("Inspecting", path)
dists = list(Distribution.discover(path=[path]))
if not dists:
return
print("Found", len(dists), "packages:", end=' ')
print(', '.join(dist.name for dist in dists))
def run():
for path in sys.path:
inspect(path)
if __name__ == '__main__':
run()
Copyright 2011 Pallets
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:
1. Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.
3. Neither the name of the copyright holder nor the names of its
contributors may be used to endorse or promote products derived from
this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
Metadata-Version: 2.1
Name: itsdangerous
Version: 2.2.0
Summary: Safely pass data to untrusted environments and back.
Maintainer-email: Pallets <contact@palletsprojects.com>
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: BSD License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Typing :: Typed
Project-URL: Changes, https://itsdangerous.palletsprojects.com/changes/
Project-URL: Chat, https://discord.gg/pallets
Project-URL: Documentation, https://itsdangerous.palletsprojects.com/
Project-URL: Donate, https://palletsprojects.com/donate
Project-URL: Source, https://github.com/pallets/itsdangerous/
# ItsDangerous
... so better sign this
Various helpers to pass data to untrusted environments and to get it
back safe and sound. Data is cryptographically signed to ensure that a
token has not been tampered with.
It's possible to customize how data is serialized. Data is compressed as
needed. A timestamp can be added and verified automatically while
loading a token.
## A Simple Example
Here's how you could generate a token for transmitting a user's id and
name between web requests.
```python
from itsdangerous import URLSafeSerializer
auth_s = URLSafeSerializer("secret key", "auth")
token = auth_s.dumps({"id": 5, "name": "itsdangerous"})
print(token)
# eyJpZCI6NSwibmFtZSI6Iml0c2Rhbmdlcm91cyJ9.6YP6T0BaO67XP--9UzTrmurXSmg
data = auth_s.loads(token)
print(data["name"])
# itsdangerous
```
## Donate
The Pallets organization develops and supports ItsDangerous and other
popular packages. In order to grow the community of contributors and
users, and allow the maintainers to devote more time to the projects,
[please donate today][].
[please donate today]: https://palletsprojects.com/donate
itsdangerous-2.2.0.dist-info/INSTALLER,sha256=zuuue4knoyJ-UwPPXg8fezS7VCrXJQrAP7zeNuwvFQg,4
itsdangerous-2.2.0.dist-info/LICENSE.txt,sha256=Y68JiRtr6K0aQlLtQ68PTvun_JSOIoNnvtfzxa4LCdc,1475
itsdangerous-2.2.0.dist-info/METADATA,sha256=0rk0-1ZwihuU5DnwJVwPWoEI4yWOyCexih3JyZHblhE,1924
itsdangerous-2.2.0.dist-info/RECORD,,
itsdangerous-2.2.0.dist-info/WHEEL,sha256=EZbGkh7Ie4PoZfRQ8I0ZuP9VklN_TvcZ6DSE5Uar4z4,81
itsdangerous/__init__.py,sha256=4SK75sCe29xbRgQE1ZQtMHnKUuZYAf3bSpZOrff1IAY,1427
itsdangerous/__pycache__/__init__.cpython-38.pyc,,
itsdangerous/__pycache__/_json.cpython-38.pyc,,
itsdangerous/__pycache__/encoding.cpython-38.pyc,,
itsdangerous/__pycache__/exc.cpython-38.pyc,,
itsdangerous/__pycache__/serializer.cpython-38.pyc,,
itsdangerous/__pycache__/signer.cpython-38.pyc,,
itsdangerous/__pycache__/timed.cpython-38.pyc,,
itsdangerous/__pycache__/url_safe.cpython-38.pyc,,
itsdangerous/_json.py,sha256=wPQGmge2yZ9328EHKF6gadGeyGYCJQKxtU-iLKE6UnA,473
itsdangerous/encoding.py,sha256=wwTz5q_3zLcaAdunk6_vSoStwGqYWe307Zl_U87aRFM,1409
itsdangerous/exc.py,sha256=Rr3exo0MRFEcPZltwecyK16VV1bE2K9_F1-d-ljcUn4,3201
itsdangerous/py.typed,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
itsdangerous/serializer.py,sha256=PmdwADLqkSyQLZ0jOKAgDsAW4k_H0TlA71Ei3z0C5aI,15601
itsdangerous/signer.py,sha256=YO0CV7NBvHA6j549REHJFUjUojw2pHqwcUpQnU7yNYQ,9647
itsdangerous/timed.py,sha256=6RvDMqNumGMxf0-HlpaZdN9PUQQmRvrQGplKhxuivUs,8083
itsdangerous/url_safe.py,sha256=az4e5fXi_vs-YbWj8YZwn4wiVKfeD--GEKRT5Ueu4P4,2505
Wheel-Version: 1.0
Generator: flit 3.9.0
Root-Is-Purelib: true
Tag: py3-none-any
from __future__ import annotations
import typing as t
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 .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
def __getattr__(name: str) -> t.Any:
if name == "__version__":
import importlib.metadata
import warnings
warnings.warn(
"The '__version__' attribute is deprecated and will be removed in"
" ItsDangerous 2.3. Use feature detection or"
" 'importlib.metadata.version(\"itsdangerous\")' instead.",
DeprecationWarning,
stacklevel=2,
)
return importlib.metadata.version("itsdangerous")
raise AttributeError(name)
from __future__ import annotations
import json as _json
import typing as t
class _CompactJSON:
"""Wrapper around json module that strips whitespace."""
@staticmethod
def loads(payload: str | bytes) -> t.Any:
return _json.loads(payload)
@staticmethod
def dumps(obj: t.Any, **kwargs: t.Any) -> str:
kwargs.setdefault("ensure_ascii", False)
kwargs.setdefault("separators", (",", ":"))
return _json.dumps(obj, **kwargs)
from __future__ import annotations
import base64
import string
import struct
import typing as t
from .exc import BadData
def want_bytes(
s: 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: str | bytes) -> bytes:
"""Base64 encode a string of bytes or text. The resulting bytes are
safe to use in URLs.
"""
string = want_bytes(string)
return base64.urlsafe_b64encode(string).rstrip(b"=")
def base64_decode(string: str | bytes) -> bytes:
"""Base64 decode a URL-safe string of bytes or text. The result is
bytes.
"""
string = want_bytes(string, encoding="ascii", errors="ignore")
string += b"=" * (-len(string) % 4)
try:
return base64.urlsafe_b64decode(string)
except (TypeError, ValueError) as e:
raise BadData("Invalid base64-encoded data") from e
# The alphabet used by base64.urlsafe_*
_base64_alphabet = f"{string.ascii_letters}{string.digits}-_=".encode("ascii")
_int64_struct = struct.Struct(">Q")
_int_to_bytes = _int64_struct.pack
_bytes_to_int = t.cast("t.Callable[[bytes], tuple[int]]", _int64_struct.unpack)
def int_to_bytes(num: int) -> bytes:
return _int_to_bytes(num).lstrip(b"\x00")
def bytes_to_int(bytestr: bytes) -> int:
return _bytes_to_int(bytestr.rjust(8, b"\x00"))[0]
from __future__ import annotations
import typing as t
from datetime import datetime
class BadData(Exception):
"""Raised if bad data of any sort was encountered. This is the base
for all exceptions that ItsDangerous defines.
.. versionadded:: 0.15
"""
def __init__(self, message: str):
super().__init__(message)
self.message = message
def __str__(self) -> str:
return self.message
class BadSignature(BadData):
"""Raised if a signature does not match."""
def __init__(self, message: str, payload: t.Any | None = 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: t.Any | None = payload
class BadTimeSignature(BadSignature):
"""Raised if a time-based signature is invalid. This is a subclass
of :class:`BadSignature`.
"""
def __init__(
self,
message: str,
payload: t.Any | None = None,
date_signed: datetime | None = 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
class SignatureExpired(BadTimeSignature):
"""Raised if a signature timestamp is older than ``max_age``. This
is a subclass of :exc:`BadTimeSignature`.
"""
class BadHeader(BadSignature):
"""Raised if a signed header is invalid in some form. This only
happens for serializers that have a header that goes with the
signature.
.. versionadded:: 0.24
"""
def __init__(
self,
message: str,
payload: t.Any | None = None,
header: t.Any | None = None,
original_error: Exception | None = None,
):
super().__init__(message, payload)
#: If the header is actually available but just malformed it
#: might be stored here.
self.header: t.Any | None = header
#: If available, the error that indicates why the payload was
#: not valid. This might be ``None``.
self.original_error: Exception | None = original_error
class BadPayload(BadData):
"""Raised if a payload is invalid. This could happen if the payload
is loaded despite an invalid signature, or if there is a mismatch
between the serializer and deserializer. The original exception
that occurred during loading is stored on as :attr:`original_error`.
.. versionadded:: 0.15
"""
def __init__(self, message: str, original_error: Exception | None = None):
super().__init__(message)
#: If available, the error that indicates why the payload was
#: not valid. This might be ``None``.
self.original_error: Exception | None = original_error
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)
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
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment