HacKanCuBa/blake2signer.py

## blake2signer.py
# ---
# DEPRECATED BY: https://gitlab.com/hackancuba/blake2signer
# ---
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
#
# by HacKan (https://hackan.net), 2020.
# This software is provided as-is. You are free to use, share, modify
# and share modifications under the terms of that license, even with
# proprietary code. Attribution is not required to share but is
# appreciated.
"""Blake2Signer: use Blake2 in keyed hashing mode to sign and verify data.

DEPRECATED BY: https://gitlab.com/hackancuba/blake2signer

The goal of this module is to provide a simple way to securely sign data and the
main use case is to sign cookies. There are much better packages for other or
more general use cases. This module uses Blake2 in keyed hashing mode. For more
information regarding this see:
https://docs.python.org/3/library/hashlib.html#blake2

This module provides three classes:

- Signer: a high-level signer class that handles data serialization, compression
  and encoding along with signing and timestamped signing.
- Blake2Signer: a low-level signer class that simply signs and verifies data as
  bytes.
- Blake2TimestampSigner: a low-level signer class that simply signs and verifies
  timestamped data as bytes.

You can't mix and match signers, and that's or purpose: if you need anything more
complex consider using "itsdangerous", Django's signer, "pypaseto", "pyjwt" or
others like those.
This means that unsigning a stream signed by Blake2Signer using
Blake2TimestampSigner may result in corrupt data and/or an error checking the
timestamp (considering that the key is the same for both), and the same goes for
the other way around.
When using Signer directly, you need to know the settings to sign/unsign from
beforehand: again, those are not stored in the stream. I.e.: signing some
compressed data but unsigning without compression may result in a DecodeError
exception or in invalid data.

I'm not a cryptoexpert, so there are some things that remain to be confirmed:

- If an attacker can control some part (or all) of the input data, is it possible
  for them to guess the secret key given a huge amount of attempts? (asuming the
  key is long enough to prevent bruteforcing in the first place, which it should
  since I set the minimum key size to 128b).
  > I think it is not possible but I would like an expert answer.

- I always asume that no attacker can influence the instantiation of the classes,
  thus they can't change any setting. If someone would break all of the given
  recomendations and somehow manage to get attacker-controlled data to class
  instantiation, which settings an attacker may change to break the security of
  this implementation and guess the secret key? This is more of an excercise but
  a fun one.
  > I think that `Signer` class is the the best target that allows more room to
  play since it deals with many layers: serialization, compression, encoding...

If you think on something else, have questions, concerns or great ideas please
feel free to contact me. If you use this code in your app I would greatly appreciate
a "thumbs up" and to share your use case and implementation :)

by HacKan (https://hackan.net), 2020.
Ref: https://gist.github.com/HacKanCuBa/b93864a1ed41746b3d75f80eb09de109

Changelog:
- 2020-10-04: use compact JSON encoding.
- 2020-09-15: change composition order b/c its easier to work with positive slices
and it's kinda a convention to have salt at the begining rather than at the end
(incentive from thread https://twitter.com/HacKanCuBa/status/1305611525344956416).
- 2020-09-14: add basic tests (run with `python -m unittest blake2signer` or your
preferred runner). Fix digest and key size check.
- 2020-09-13: relicense with MPL 2.0, derive person in Signer.
- 2020-09-12: initial release.

Examples:
>>> secret = b'ZnVja3RoZXBvbGljZQ'
>>> data = [{'a': 'b'}, 1] * 10000  # big data structure
>>> len(data)  # 20000

>>> signer = Signer(secret)
>>> signed = signer.sign(data)
>>> len(signed)  # 200043
>>> unsigned = signer.unsign(signed)
>>> data == unsigned  # True

>>> signer = Signer(secret, max_age=timedelta(weeks=1), use_compression=True)
>>> signed = signer.sign(data)
>>> len(signed)  # 492
>>> unsigned = signer.unsign(signed)
>>> data == unsigned  # True

>>> data = b'facundo castro presente'
>>> signer = Blake2Signer(secret)
>>> signed = signer.sign(data)
>>> len(signed)  # 103
>>> unsigned = signer.unsign(signed)
>>> data == unsigned  # True
>>> signer = Blake2TimestampSigner(secret)
>>> signed = signer.sign(data)
>>> len(signed)  # 107
>>> unsigned = signer.unsign(signed, max_age=10)
>>> data == unsigned  # True

>>> # Let's mix and match, what could go wrong? spoiler: everything!
>>> data = b'#ACAB'
>>> signer = Blake2Signer(secret)
>>> t_signer = Blake2TimestampSigner(secret)
>>> t_signer.unsign(signer.sign(data), max_age=1)  # ExpiredSignatureError
>>> # We see an exception because since the signature is OK the timestamped signer
>>> # is considering the 4 bytes `b'ACAB'` as a timestamp which gives us
>>> # 2004-09-11T15:17:38, way in the past. Is this an issue with the signer? NO.
>>> # As stated before, one must be careful of NOT mixing and matching things.
>>> signer.unsign(t_signer.sign(data))  # b'...#ACAB'
>>> # This time we don't even get an exception because all is OK for the signer,
>>> # but the recovered data is wrong! It contains the timestamp from the timestamp
>>> # signer.
>>> # When using different signers for different things, its a good idea to use
>>> # the personalisation parameter which prevents these situations:
>>> signer = Blake2Signer(secret, person=b'1234')
>>> signer.unsign(signer.sign(data))  # b'#ACAB'
>>> signer.unsign(t_signer.sign(data))  # InvalidSignatureError
>>> # Even though the key/secret is the same, the personalisation parameter changes
>>> # the hashing output thus changing the signature. This is very useful to
>>> # prevent these situations and should be implemented whenever used. It doesn't
>>> # have to be random nor secret nor too long, it just needs to be unique for
>>> # the usage. There's a limit to its size but the Signer class derives the value
>>> # so it has no limit in it.

The moral of the story is: always sign and unsign using the exact same signer with
the exact same parameters (there aren't many anyway), and use the personalisation
parameter whenever you can.
"""

import base64
import json
import typing
from abc import ABC
from abc import abstractmethod
from dataclasses import dataclass
from datetime import timedelta
from enum import Enum
from hashlib import blake2b
from hashlib import blake2s
from secrets import compare_digest
from secrets import token_bytes
from time import time
from unittest import TestCase
from zlib import compress
from zlib import decompress
from zlib import error as zlib_error


def b64encode(data: bytes) -> bytes:
    """Encode data as Base 64 URL safe, stripping padding."""
    return base64.urlsafe_b64encode(data).rstrip(b'=')


def b64decode(data: typing.Union[bytes, str]) -> bytes:
    """Decode data encoded as Base 64 URL safe without padding."""
    if isinstance(data, str):
        data = data.encode()
    return base64.urlsafe_b64decode(data + b'=' * (len(data) % 4))


class SignerError(Exception):
    """Base exception for errors."""


class InvalidOptionError(SignerError):
    """Invalid options error."""


class DecodeError(SignerError):
    """Decode error."""


class InvalidSignatureError(DecodeError):
    """Invalid signature error."""


class ExpiredSignatureError(InvalidSignatureError):
    """Expired signature error."""


class EncodeError(SignerError):
    """Encode error."""


@dataclass(frozen=True)
class SignedDataParts:
    """Parts of a signed data container."""

    data: bytes
    salt: bytes
    signature: bytes


@dataclass(frozen=True)
class TimestampedDataParts:
    """Parts of a timestamped data container."""

    data: bytes
    timestamp: bytes


class Blake2Options(typing.TypedDict):
    """Blake2 options."""

    fanout: int
    depth: int
    leaf_size: int
    node_offset: int
    node_depth: int
    inner_size: int
    last_node: bool


class Blake2ExtendedOptions(Blake2Options):
    """Blake2 extended options."""

    key: bytes
    digest_size: int
    person: bytes


class Hashers(Enum):
    """Hasher selection options."""

    blake2b = 'blake2b'
    blake2s = 'blake2s'


class Blake2SignerBase(ABC):
    """Base class for a signer based on Blake2 in keyed hashing mode."""

    __slots__ = (
        '_hasher',
        '_hasher_options',
    )

    MIN_KEY_SIZE: int = 16
    MIN_DIGEST_SIZE: int = 8

    def __init__(
            self,
            key: bytes,
            *,
            hasher: Hashers = Hashers.blake2b,
            digest_size: typing.Optional[int] = None,
            person: bytes = b'',
            fanout: int = 1,
            depth: int = 1,
            leaf_size: int = 0,
            node_offset: int = 0,
            node_depth: int = 0,
            inner_size: int = 0,
            last_node: bool = False,
    ) -> None:
        """Sign and verify signed data using Blake2 in keyed hashing mode.

        :param key: Secret key for signing and verifying signed data. The minimum
                    key size is enforced to 16 bytes, and the maximum depends on
                    the chosen hasher.
        :param hasher: [optional] Hash function to use: blake2b (default) or blake2s.
        :param digest_size: [optional] Size of output signature (digest) in bytes
                            (defaults to maximum digest size of chosen function).
                            Bear in mind that a small digest size increases the
                            risk collision. I.e. if we used 1 byte as digest size,
                            an attacker would be able to correctly sign any payload
                            in around ~128 attempts or less without needing to
                            know the secret key. For this reason the minimum size
                            is enforced to 8 bytes.
        :param person: [optional] personalisation string to force the hash function
                       to produce different digests for the same input.

        :raise InvalidOptionError: A parameter is out of bounds.
        """
        self._hasher: typing.Union[typing.Type[blake2b], typing.Type[blake2s]]
        if hasher is Hashers.blake2b:
            self._hasher = blake2b
        else:
            self._hasher = blake2s

        if not (self.MIN_KEY_SIZE <= len(key) <= self._hasher.MAX_KEY_SIZE):
            raise InvalidOptionError(
                f'key length should be between {self.MIN_KEY_SIZE} and '
                f'{self._hasher.MAX_KEY_SIZE}',
            )

        if digest_size is None:
            digest_size = self._hasher.MAX_DIGEST_SIZE
        elif not (self.MIN_DIGEST_SIZE <= digest_size <= self._hasher.MAX_DIGEST_SIZE):
            raise InvalidOptionError(
                f'digest_size should be between {self.MIN_DIGEST_SIZE} and '
                f'{self._hasher.MAX_DIGEST_SIZE}',
            )

        self._hasher_options: Blake2ExtendedOptions = Blake2ExtendedOptions(
            key=key,
            person=person,
            digest_size=digest_size,
            fanout=fanout,
            depth=depth,
            leaf_size=leaf_size,
            node_offset=node_offset,
            node_depth=node_depth,
            inner_size=inner_size,
            last_node=last_node,
        )

        self._check_hasher_options()

    def _check_hasher_options(self) -> None:
        """Check hasher options to be valid."""
        try:
            self._hasher(**self._hasher_options)
        except ValueError as exc:
            raise InvalidOptionError(exc) from exc

    @property
    def salt_size(self) -> int:
        """Get the salt size."""
        return self._hasher.SALT_SIZE

    @property
    def signature_size(self) -> int:
        """Get the signature size/length."""
        return self._hasher_options['digest_size']

    def _generate_salt(self) -> bytes:
        """Generate a cryptographically secure pseudorandom salt."""
        return token_bytes(self.salt_size)

    @staticmethod
    def _compose(*, salt: bytes, signature: bytes, data: bytes) -> bytes:
        """Compose signed data parts into a single stream."""
        return salt + signature + data

    def _decompose(self, signed_data: bytes) -> SignedDataParts:
        """Decompose a signed data stream into its parts.

        :raise DecodeError: Invalid signed data.
        """
        if len(signed_data) < (self.salt_size + self.signature_size):
            raise DecodeError('signed data is too short')

        salt = signed_data[:self.salt_size]
        signature = signed_data[self.salt_size:self.salt_size + self.signature_size]
        data = signed_data[self.salt_size + self.signature_size:]

        return SignedDataParts(data=data, salt=salt, signature=signature)

    def _sign(self, *, salt: bytes, data: bytes) -> bytes:
        """Sign given data using salt and all of the hasher options."""
        signed_data = self._hasher(
            data,
            salt=salt,
            **self._hasher_options,
        ).digest()

        return signed_data

    def _verify(self, parts: SignedDataParts) -> bool:
        """Verify a signature.

        :return: True if the signature is correct, False otherwise.
        """
        good_signature = self._sign(salt=parts.salt, data=parts.data)

        return compare_digest(good_signature, parts.signature)

    def _unsign(self, signed_data: bytes) -> bytes:
        """Verify a signed stream and recover original data.

        :param signed_data: Signed data to unsign.

        :raise DecodeError: Signed data is not valid or it can't be decoded.
        :raise InvalidSignatureError: Signed data has invalid signature.

        :return: Original data.
        """
        parts = self._decompose(signed_data)
        if not self._verify(parts):
            raise InvalidSignatureError('signature is not valid')

        return parts.data

    @abstractmethod
    def sign(self, data: bytes) -> bytes:
        """Sign given data and produce a stream."""


class Blake2Signer(Blake2SignerBase):
    """Blake2 in keyed hashing mode for signing data."""

    def sign(self, data: bytes) -> bytes:
        """Sign given data and produce a stream composed of it, salt and signature.

        Note that given data is _not_ encrypted, only signed. To recover data from
        it, while validating the signature, use `unsign`.

        The salt is a cryptographically secure pseudorandom string generated for
        this signature only.

        The total length of the resulting stream can be calculated as:
        len(data) + salt_size + signature_size.

        :return: A signed stream composed of data + salt + signature.
        """
        salt = self._generate_salt()
        signature = self._sign(salt=salt, data=data)

        return self._compose(salt=salt, signature=signature, data=data)

    def unsign(self, signed_data: bytes) -> bytes:
        """Verify a signed stream and recover original data.

        :param signed_data: Signed data to unsign.

        :raise DecodeError: Signed data is not valid or it can't be decoded.
        :raise InvalidSignatureError: Signed data has invalid signature.

        :return: Original data.
        """
        return self._unsign(signed_data)


class Blake2TimestampSigner(Blake2SignerBase):
    """Blake2 in keyed hashing mode for signing data with timestamp."""

    @property
    def timestamp_size(self) -> int:
        """Get the timestamp value size in bytes."""
        return 4  # Good enough until 2038

    @property
    def timestamp(self) -> bytes:
        """Get the encoded timestamp value."""
        timestamp = int(time())  # its easier to encode an integer
        try:
            return timestamp.to_bytes(self.timestamp_size, 'big', signed=False)
        except OverflowError:  # This will happen in ~2038-01-19
            raise NotImplementedError(
                'can not represent this timestamp in bytes: this library is '
                'too old and needs to be updated!',
            )

    @staticmethod
    def _decode_timestamp(encoded_timestamp: bytes) -> int:
        """Decode an encoded timestamp which should have been validated."""
        timestamp = int.from_bytes(encoded_timestamp, 'big', signed=False)

        return timestamp

    def _add_timestamp(self, data: bytes) -> bytes:
        """Add timestamp value to given data."""
        return self.timestamp + data

    def _split_timestamp(self, timestamped_data: bytes) -> TimestampedDataParts:
        """Split data + timestamp value.

        :raise DecodeError: Invalid timestamped data.
        """
        if len(timestamped_data) < self.timestamp_size:
            raise DecodeError('timestamped data is too short')

        timestamp = timestamped_data[:self.timestamp_size]
        data = timestamped_data[self.timestamp_size:]

        return TimestampedDataParts(data=data, timestamp=timestamp)

    def sign(self, data: bytes) -> bytes:
        """Sign given data and produce a stream of it, timestamp, salt and signature.

        Note that given data is _not_ encrypted, only signed. To recover data from
        it, while validating the signature, use `unsign`.

        The salt is a cryptographically secure pseudorandom string generated for
        this signature only.

        The total length of the resulting stream can be calculated as:
        len(data) + timestamp_size + salt_size + signature_size.

        :return: A signed stream composed of data + timestamp + salt + signature.
        """
        salt = self._generate_salt()
        data_to_sign = self._add_timestamp(data)
        signature = self._sign(salt=salt, data=data_to_sign)
        return self._compose(salt=salt, signature=signature, data=data_to_sign)

    def unsign(
            self,
            signed_data: bytes,
            *,
            max_age: typing.Union[int, float, timedelta],
    ) -> bytes:
        """Verify a signed stream with timestamp and recover original data.

        :param signed_data: Signed data to unsign.
        :param max_age: Ensure the signature is not older than this time in seconds.

        :raise DecodeError: Signed data is not valid or it can't be decoded.
        :raise InvalidSignatureError: Signed data has invalid signature.
        :raise ExpiredSignatureError: Signed data has expired.

        :return: Original data.
        """
        data = self._unsign(signed_data)

        data_parts = self._split_timestamp(data)

        if isinstance(max_age, timedelta):
            ttl = max_age.total_seconds()
        else:
            ttl = float(max_age)

        now = time()
        timestamp = self._decode_timestamp(data_parts.timestamp)
        age = timestamp + ttl
        if age < now:
            raise ExpiredSignatureError('signed data has expired')

        return data_parts.data


class Signer:
    """Blake2 in keyed hashing mode for signing (optionally timestamped) data.

    It can handle data serialization, compression and encoding.
    """

    MIN_SECRET_SIZE: int = Blake2SignerBase.MIN_KEY_SIZE
    DEFAULT_DIGEST_SIZE: int = 16  # 16 bytes is good security/size tradeoff

    __slots__ = (
        '_hasher',
        '_max_age',
        '_signer',
        '_use_compression',
    )

    def __init__(
            self,
            secret: bytes,
            *,
            max_age: typing.Union[None, int, float, timedelta] = None,
            use_compression: bool = False,
            person: bytes = b'',
            hasher: Hashers = Hashers.blake2b,
            digest_size: int = DEFAULT_DIGEST_SIZE,
            **kwargs: Blake2Options,
    ) -> None:
        """Sign and verify signed data using Blake2 in keyed hashing mode.

        Setting max_age will produce a timestamped signed stream. Using compression
        may help reducing the size of resulting stream.

        Note that configuration parameters like max_age and use_compression are
        NOT saved in the signed stream so you need to specify them to sign and
        unsign.

        This class is intended to be used to sign and verify cookies or similar.
        It sets sane defaults such as a signature size of 16 bytes, key derivation
        from given secret and the use of a personalisation string.

        It is not supposed to cover all corner cases and be ultimately flexible
        so if you are in need of that please consider using "itsdangerous",
        Django's signer, "pypaseto", "pyjwt" or others like those.

        :param secret: Secret value which will be derived using blake2 to
                       produce the signing key. The minimum secret size is
                       enforced to 16 bytes and there is no maximum since the key
                       will be derived to the maximum supported size.
        :param max_age: [optional] Use a timestamp signer instead of a regular
                        one to ensure that the signature is not older than this
                        time in seconds.
        :param use_compression: [optional] Compress data after serializing it and
                                decompress it before unserializing. For low entropy
                                payloads such as human readable text, it's beneficial
                                from around ~30bytes, and detrimental if smaller.
                                For high entropy payloads like pseudorandom text,
                                it's beneficial from around ~300bytes and detrimental
                                if lower than ~100bytes. You should choose to enable
                                it based on your knowledge of the average payload
                                size and type.
        :param person: [optional] Set the personalisation string to force the hash
                       function to produce different digests for the same input,
                       which will be derived using blake2 to ensure it fits the
                       hasher, so there's no practical size limit. It defaults
                       to the class name.
        :param hasher: [optional] Hash function to use: blake2b (default) or blake2s.
        :param digest_size: [optional] Size of output signature (digest) in bytes
                            (defaults to 16 bytes).
        :param kwargs: [optional] Other options for blake2 functions.

        :raise InvalidOptionError: A parameter is out of bounds.
        """
        self._hasher: typing.Union[typing.Type[blake2b], typing.Type[blake2s]]
        if hasher is Hashers.blake2b:
            self._hasher = blake2b
        else:
            self._hasher = blake2s

        if len(secret) < self.MIN_SECRET_SIZE:
            raise InvalidOptionError(
                f'secret should be longer than {self.MIN_SECRET_SIZE} bytes',
            )

        self._use_compression: bool = use_compression

        # read more about personalisation in the hashlib docs:
        # https://docs.python.org/3/library/hashlib.html#personalisation
        if not person:
            person = self.__class__.__name__.encode()
        person = self.derive_person(person)

        # mypy issue: https://github.com/python/mypy/issues/8890
        signer_options: Blake2ExtendedOptions = Blake2ExtendedOptions(  # type: ignore
            key=self.derive_key(secret, person=person),
            digest_size=digest_size,
            person=person,
            **kwargs,
        )

        self._max_age: typing.Union[int, float, timedelta]
        self._signer: typing.Union[Blake2Signer, Blake2TimestampSigner]
        if max_age is None:
            self._signer = Blake2Signer(hasher=hasher, **signer_options)
            self._max_age = 0
        else:
            self._signer = Blake2TimestampSigner(hasher=hasher, **signer_options)
            self._max_age = max_age

    def derive_person(self, person: bytes) -> bytes:
        """Derive given personalisation value to ensure it fits the hasher correctly."""
        return self._hasher(person, digest_size=self._hasher.PERSON_SIZE).digest()

    def derive_key(self, secret: bytes, *, person: bytes = b'') -> bytes:
        """Derive given secret to ensure it fits correctly as the hasher key."""
        return self._hasher(
            secret,
            person=person,
            digest_size=self._hasher.MAX_KEY_SIZE,
        ).digest()

    @staticmethod
    def _serialize(data: typing.Any) -> bytes:
        """Serialize given data to JSON."""
        try:
            return json.dumps(data, separators=(',', ':')).encode()  # compact encoding
        except TypeError as exc:
            raise EncodeError(exc) from exc

    @staticmethod
    def _unserialize(data: bytes) -> typing.Any:
        try:
            return json.loads(data)
        except ValueError:
            raise DecodeError('data can not be unserialized')

    @staticmethod
    def _compress(data: bytes) -> bytes:
        # Default level is 6 currently but 5 usually performs better with
        # little compression tradeoff.
        try:
            return compress(data, level=5)
        except zlib_error as exc:
            raise EncodeError(exc) from exc

    @staticmethod
    def _decompress(data: bytes) -> bytes:
        try:
            return decompress(data)
        except zlib_error:
            raise DecodeError('data can not be decompressed')

    @staticmethod
    def _encode(data: bytes) -> str:
        try:
            return b64encode(data).decode()
        except (ValueError, TypeError) as exc:
            raise EncodeError(exc) from exc

    @staticmethod
    def _decode(data: str) -> bytes:
        try:
            return b64decode(data)
        except (ValueError, TypeError):
            raise DecodeError('invalid base64 data')

    def sign(self, data: typing.Any) -> str:
        """Sign after encoding given data and produce a stream of it.

        Data will be serialized to JSON and optionally compressed before being
        signed. This means that the original data type may not be recoverable so
        if you are using some custom class you will need to load it from basic
        types.

        If `max_age` was specified then the stream will be timestamped.
        The stream is also salted by a cryptographically secure pseudorandom
        string generated for this signature only.

        The resulting stream is base64 encoded.

        Note that given data is _not_ encrypted, only signed. To recover data from
        it, while validating the signature (and timestamp if any), use `unsign`.

        The full flow is as follows, where optional actions are marked between brackets:
        data -> serialize -> [compress] -> [timestamp] -> sign -> encode

        :param data: Any JSON encodable object.

        :raise EncodeError: Data could not be encoded.

        :return: A base64 encoded, signed and optionally timestamped stream of data.
        """
        # If you are modifying this lib, please don't put configuration parameters
        # into the stream! Unless you want vulnerabilities, because that's how you
        # get vulnerabilities. If you need more flexibility, consider using
        # "itsdangerous", "pypaseto" or "pyjwt".
        serialized = self._serialize(data)
        compressed = self._compress(serialized) if self._use_compression else serialized
        signed = self._signer.sign(compressed)
        return self._encode(signed)

    def unsign(self, signed_data: str) -> typing.Any:
        """Verify a signed stream and recover original data.

        If `max_age` was specified then it will be ensured that the signature is
        not older than this time in seconds.

        Important note: if signed data was timestamped but `max_age` was not
        specified then an error will occur. The same goes for the other way
        around: if it wasn't timestamped but `max_age` is now set. And the same
        for compression. So you need to know these parameters from beforehand:
        they won't live in the signed stream!

        The full flow is as follows, where optional actions are marked between brackets:
        data -> decode -> check sig -> [check timestamp] -> [decompress] -> unserialize

        :param signed_data: Signed data to unsign.

        :raise DecodeError: Signed data is not valid or it can't be decoded.
        :raise InvalidSignatureError: Signed data has invalid signature.
        :raise ExpiredSignatureError: Signed data has expired.

        :return: Unserialized data.
        """
        decoded = self._decode(signed_data)

        if isinstance(self._signer, Blake2Signer):
            unsigned = self._signer.unsign(decoded)
        else:
            unsigned = self._signer.unsign(decoded, max_age=self._max_age)

        decompressed = self._decompress(unsigned) if self._use_compression else unsigned
        unserizalized = self._unserialize(decompressed)

        return unserizalized


# You may want to move the tests to its own module

class Blake2SignerTests(TestCase):
    """Test Blake2Signer class."""

    def setUp(self) -> None:
        """Set up test cases."""
        self.key = b'0123456789012345'
        self.data = b'datadata'

    def test_initialisation_defaults(self) -> None:
        """Test correct class defaults initialisation."""
        signer = Blake2Signer(self.key)
        self.assertIsInstance(signer, Blake2Signer)

        signer = Blake2Signer(self.key, hasher=Hashers.blake2s)
        self.assertIsInstance(signer, Blake2Signer)

    def test_initialisation_all_options(self) -> None:
        """Test correct class initialisation with all options."""
        # ToDo

    def test_sign(self) -> None:
        """Test signing is correct."""
        signer = Blake2Signer(self.key)
        signed = signer.sign(self.data)
        self.assertIsInstance(signed, bytes)
        expected_size = len(self.data) + signer.salt_size + signer.signature_size
        self.assertEqual(len(signed), expected_size)

    def test_unsign(self) -> None:
        """Test unsigning is correct."""
        signer = Blake2Signer(self.key)
        signed = signer.sign(self.data)
        unsigned = signer.unsign(signed)
        self.assertEqual(unsigned, self.data)


class Blake2SignerErrorTests(TestCase):
    """Test Blake2Signer class for errors."""

    def setUp(self) -> None:
        """Set up test cases."""
        self.key = b'0123456789012345'
        self.data = b'datadata'

    def test_key_too_short(self) -> None:
        """Test key too short."""
        with self.assertRaises(InvalidOptionError):
            Blake2Signer(b'12345678')

    def test_key_too_long(self) -> None:
        """Test key too long."""
        with self.assertRaises(InvalidOptionError):
            Blake2Signer(b'0' * 65)

        with self.assertRaises(InvalidOptionError):
            Blake2Signer(b'0' * 33, hasher=Hashers.blake2s)

    def test_digest_too_small(self) -> None:
        """Test digest too small."""
        with self.assertRaises(InvalidOptionError):
            Blake2Signer(self.key, digest_size=4)

        with self.assertRaises(InvalidOptionError):
            Blake2Signer(self.key, digest_size=4, hasher=Hashers.blake2s)

    def test_digest_too_large(self) -> None:
        """Test digest too large."""
        with self.assertRaises(InvalidOptionError):
            Blake2Signer(self.key, digest_size=65)

        with self.assertRaises(InvalidOptionError):
            Blake2Signer(self.key, digest_size=33, hasher=Hashers.blake2s)

    def test_wrong_options(self) -> None:
        """Test parameters out of bounds."""
        with self.assertRaises(InvalidOptionError):
            Blake2Signer(self.key, person=b'0' * 17)

        with self.assertRaises(InvalidOptionError):
            Blake2Signer(self.key, fanout=256)

    def test_unsign_wrong_data(self) -> None:
        """Test unsign with wrong data."""
        signer = Blake2Signer(self.key)
        with self.assertRaises(DecodeError, msg='signed data is too short'):
            signer.unsign(b'12345678')

    def test_unsign_invalid_signature(self) -> None:
        """Test unsign with invalid signature."""
        signer = Blake2Signer(self.key)
        with self.assertRaises(InvalidSignatureError):
            signer.unsign(b'0' * (signer.salt_size + signer.signature_size))


class Blake2TimestampSignerTests(TestCase):
    """Test Blake2TimestampSigner class."""

    def setUp(self) -> None:
        """Set up test cases."""
        self.key = b'0123456789012345'
        self.data = b'datadata'

    def test_initialisation_defaults(self) -> None:
        """Test correct class defaults initialisation."""
        signer = Blake2TimestampSigner(self.key)
        self.assertIsInstance(signer, Blake2TimestampSigner)

        signer = Blake2TimestampSigner(self.key, hasher=Hashers.blake2s)
        self.assertIsInstance(signer, Blake2TimestampSigner)

    def test_sign(self) -> None:
        """Test signing is correct."""
        signer = Blake2TimestampSigner(self.key)
        signed = signer.sign(self.data)
        self.assertIsInstance(signed, bytes)
        added_size = signer.timestamp_size + signer.salt_size + signer.signature_size
        self.assertEqual(
            len(signed),
            len(self.data) + added_size,
        )

    def test_unsign(self) -> None:
        """Test unsigning is correct."""
        signer = Blake2TimestampSigner(self.key)
        signed = signer.sign(self.data)
        unsigned = signer.unsign(signed, max_age=timedelta(seconds=1))
        self.assertEqual(unsigned, self.data)


class Blake2TimestampSignerErrorTests(TestCase):
    """Test Blake2TimestampSigner class for errors."""

    def setUp(self) -> None:
        """Set up test cases."""
        self.key = b'0123456789012345'
        self.data = b'datadata'

    def test_unsign_timestamp_expired(self) -> None:
        """Test unsigning with timestamp is correct."""
        from time import sleep

        timeout = 0.00001
        signer = Blake2TimestampSigner(self.key)
        signed = signer.sign(self.data)
        sleep(timeout)
        with self.assertRaises(ExpiredSignatureError):
            signer.unsign(signed, max_age=timeout)

    def test_unsign_wrong_data(self) -> None:
        """Test unsing wrong data."""
        # To test this I need a valid signature w/ wrong timestamp
        signed = Blake2Signer(self.key).sign(b'0')
        with self.assertRaises(DecodeError):
            Blake2TimestampSigner(self.key).unsign(signed, max_age=1)


class SignerTests(TestCase):
    """Test Signer class."""

    def setUp(self) -> None:
        """Set up test cases."""
        self.secret = b'0123456789012345'
        self.data = 'datadata'

    def test_initialisation_defaults(self) -> None:
        """Test correct class defaults initialisation."""
        signer = Signer(self.secret)
        self.assertIsInstance(signer, Signer)

        signer = Signer(self.secret, hasher=Hashers.blake2s)
        self.assertIsInstance(signer, Signer)

    def test_initialisation_timestamp(self) -> None:
        """Test correct class defaults with timestamping initialisation."""
        # ToDo

    def test_initialisation_all_options(self) -> None:
        """Test correct class initialisation with all options."""
        # ToDo

    def test_sign(self) -> None:
        """Test signing is correct."""
        signer = Signer(self.secret)
        signed = signer.sign(self.data)
        self.assertIsInstance(signed, str)
        self.assertEqual(len(signed), 56)

    def test_sign_timestamp(self) -> None:
        """Test signing with timestamp is correct."""
        signer = Signer(self.secret, max_age=1)
        signed = signer.sign(self.data)
        self.assertIsInstance(signed, str)
        self.assertEqual(len(signed), 62)

    def test_sign_timestamp_compression(self) -> None:
        """Test signing with timestamp and compression is correct."""
        # ToDo

    def test_sign_other_options(self) -> None:
        """Test signing with other options is correct."""
        # ToDo

    def test_unsign(self) -> None:
        """Test unsigning is correct."""
        signer = Signer(self.secret)
        signed = signer.sign(self.data)
        unsigned = signer.unsign(signed)
        self.assertEqual(unsigned, self.data)

    def test_unsign_timestamp(self) -> None:
        """Test unsigning with timestamp is correct."""
        signer = Signer(self.secret, max_age=1)
        signed = signer.sign(self.data)
        unsigned = signer.unsign(signed)
        self.assertEqual(unsigned, self.data)

    def test_unsign_timestamp_compression(self) -> None:
        """Test unsigning with timestamp and compression is correct."""
        # ToDo

    def test_unsign_other_options(self) -> None:
        """Test unsigning with other options is correct."""
        # ToDo


class SignerErrorTests(TestCase):
    """Test Signer class for errors."""

    def setUp(self) -> None:
        """Set up test cases."""
        self.secret = b'0123456789012345'
        self.data = 'datadata'

    def test_secret_too_short(self) -> None:
        """Test parameters out of bounds."""
        with self.assertRaises(InvalidOptionError):
            Signer(b'12345678')

    def test_signature_too_short(self) -> None:
        """Test parameters out of bounds."""
        with self.assertRaises(InvalidOptionError):
            Signer(b'01234567890123456789', digest_size=4)

    def test_unsign_timestamp_expired(self) -> None:
        """Test unsigning with timestamp is correct."""
        from time import sleep

        timeout = 0.00001
        signer = Signer(self.secret, max_age=timeout)
        signed = signer.sign(self.data)
        sleep(timeout)
        with self.assertRaises(ExpiredSignatureError):
            signer.unsign(signed)

    def test_sign_wrong_data(self) -> None:
        """Test sign wrong data."""
        signer = Signer(self.secret)
        with self.assertRaises(EncodeError):
            signer.sign(b'datadata')  # any non JSON encodable type

    def test_unsign_wrong_data(self) -> None:
        """Test unsign wrong data."""
        signer = Signer(self.secret)
        with self.assertRaises(DecodeError):
            # noinspection PyTypeChecker
            signer.unsign(1234)  # type: ignore

    def test_unsign_decompression_error(self) -> None:
        """Test unsign wrong data causing decompression error."""
        # I need valid signed uncompressed data
        signed = Signer(self.secret, use_compression=False).sign(self.data)
        with self.assertRaises(DecodeError):
            Signer(self.secret, use_compression=True).unsign(signed)

    def test_unsign_unserialization_error(self) -> None:
        """Test unsign wrong data causing unserialization error."""
        # I need valid signed non JSON encodable data
        signed = Signer(self.secret, use_compression=True).sign(self.data)
        with self.assertRaises(DecodeError):
            Signer(self.secret).unsign(signed)