Mixins¶
blake2signer.mixins
¶
Mixins: abstract classes that provide certain restricted functionality.
They work as a building block for other classes.
CompressorMixin (Mixin, ABC)
¶
Compressor mixin.
Adds compressing capabilities to a subclass.
Source code in blake2signer/mixins.py
class CompressorMixin(Mixin, ABC):
"""Compressor mixin.
Adds compressing capabilities to a subclass.
"""
def __init__(
self,
*args: typing.Any,
compressor: typing.Type[CompressorInterface] = ZlibCompressor,
compression_flag: typing.Union[str, bytes] = b'.',
compression_ratio: typing.Union[int, float] = 5.0,
**kwargs: typing.Any,
) -> None:
"""Add compressing capabilities.
Args:
*args: Additional positional arguments.
Keyword Args:
compressor (optional): Compressor class to use (defaults to a Zlib
compressor).
compression_flag (optional): Character to mark the payload as compressed.
It must not belong to the encoder alphabet and be ASCII (defaults
to ".").
compression_ratio (optional): Desired minimal compression ratio, between
0 and below 100 (defaults to 5). It is used to calculate when
to consider a payload sufficiently compressed to detect detrimental
compression. By default, if compression achieves less than 5% of
size reduction, it is considered detrimental.
**kwargs: Additional keyword only arguments.
Returns:
None.
"""
self._compressor = compressor()
personalisation = self._force_bytes(kwargs.get('personalisation', b''))
personalisation += self._compressor.__class__.__name__.encode()
kwargs['personalisation'] = personalisation
self._compression_flag: bytes = self._validate_comp_flag(compression_flag)
self._compression_ratio: float = self._validate_comp_ratio(compression_ratio)
super().__init__(*args, **kwargs)
def _validate_comp_flag(self, flag: typing.Union[str, bytes]) -> bytes:
"""Validate the compression flag value and return it clean."""
if not flag:
raise InvalidOptionError('the compression flag character must have a value')
if not flag.isascii():
raise InvalidOptionError('the compression flag character must be ASCII')
return self._force_bytes(flag)
@staticmethod
def _validate_comp_ratio(ratio_: float) -> float:
"""Validate the compression ratio value and return it clean."""
ratio = float(ratio_)
if 0.0 <= ratio < 100.0:
return ratio
raise InvalidOptionError('the compression ratio must be between 0 and 99')
def _add_compression_flag(self, data: bytes) -> bytes:
"""Add the compression flag to given data."""
return self._compression_flag + data # prevents zip bombs
def _is_compressed(self, data: bytes) -> bool:
"""Return True if given data is compressed, checking the compression flag."""
return data.startswith(self._compression_flag, 0, len(self._compression_flag))
def _remove_compression_flag(self, data: bytes) -> bytes:
"""Remove the compression flag from given data."""
return data[len(self._compression_flag):]
def _remove_compression_flag_if_compressed(
self,
data: bytes,
) -> typing.Tuple[bytes, bool]:
"""Remove the compression flag from given data if it is compressed.
Args:
data: Data to process.
Returns:
A tuple of given data without the flag, and a boolean indicating
if it is compressed or not.
"""
if self._is_compressed(data):
return self._remove_compression_flag(data), True
return data, False
def _is_significantly_compressed(
self,
data_size: int,
compressed_size: int,
) -> bool:
"""Return True if the compressed size is significantly lower than data size."""
return compressed_size < (data_size * (1 - (self._compression_ratio / 100)))
def _compress(
self,
data: bytes,
*,
level: typing.Optional[int] = None,
force: bool = False,
) -> typing.Tuple[bytes, bool]:
"""Compress given data if convenient or forced, otherwise do nothing.
A check is done to verify if compressed data is significantly smaller than
given data and if not then it returns given data as-is, unless compression
is forced.
Args:
data: Data to compress.
Keyword Args:
level (optional): Compression level wanted from 1 (least compressed)
to 9 (most compressed) or None for the default.
force (optional): Force compression without checking if convenient.
Returns:
A tuple containing data, and a flag indicating if data is compressed
(True) or not.
Raises
CompressionError: Data can't be compressed.
"""
compression_level = self._compressor.get_compression_level(level)
try:
compressed = self._compressor.compress(data, level=compression_level)
except Exception as exc:
raise CompressionError('data can not be compressed') from exc
if force or self._is_significantly_compressed(len(data), len(compressed)):
return compressed, True
# Compression isn't reducing size so do nothing.
return data, False
def _decompress(self, data: bytes) -> bytes:
"""Decompress given data.
Raises:
DecompressionError: Data can't be decompressed.
"""
try:
return self._compressor.decompress(data)
except Exception as exc:
raise DecompressionError('data can not be decompressed') from exc
__init__(self, *args, *, compressor=<class 'blake2signer.compressors.ZlibCompressor'>, compression_flag=b'.', compression_ratio=5.0, **kwargs)
special
¶
Add compressing capabilities.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
*args |
Any |
Additional positional arguments. |
() |
Keyword arguments:
Name | Type | Description |
---|---|---|
compressor |
optional |
Compressor class to use (defaults to a Zlib compressor). |
compression_flag |
optional |
Character to mark the payload as compressed. It must not belong to the encoder alphabet and be ASCII (defaults to "."). |
compression_ratio |
optional |
Desired minimal compression ratio, between 0 and below 100 (defaults to 5). It is used to calculate when to consider a payload sufficiently compressed to detect detrimental compression. By default, if compression achieves less than 5% of size reduction, it is considered detrimental. |
**kwargs |
Any |
Additional keyword only arguments. |
Returns:
Type | Description |
---|---|
None |
None. |
Source code in blake2signer/mixins.py
def __init__(
self,
*args: typing.Any,
compressor: typing.Type[CompressorInterface] = ZlibCompressor,
compression_flag: typing.Union[str, bytes] = b'.',
compression_ratio: typing.Union[int, float] = 5.0,
**kwargs: typing.Any,
) -> None:
"""Add compressing capabilities.
Args:
*args: Additional positional arguments.
Keyword Args:
compressor (optional): Compressor class to use (defaults to a Zlib
compressor).
compression_flag (optional): Character to mark the payload as compressed.
It must not belong to the encoder alphabet and be ASCII (defaults
to ".").
compression_ratio (optional): Desired minimal compression ratio, between
0 and below 100 (defaults to 5). It is used to calculate when
to consider a payload sufficiently compressed to detect detrimental
compression. By default, if compression achieves less than 5% of
size reduction, it is considered detrimental.
**kwargs: Additional keyword only arguments.
Returns:
None.
"""
self._compressor = compressor()
personalisation = self._force_bytes(kwargs.get('personalisation', b''))
personalisation += self._compressor.__class__.__name__.encode()
kwargs['personalisation'] = personalisation
self._compression_flag: bytes = self._validate_comp_flag(compression_flag)
self._compression_ratio: float = self._validate_comp_ratio(compression_ratio)
super().__init__(*args, **kwargs)
EncoderMixin (Mixin, ABC)
¶
Encoder mixin.
Adds encoding capabilities to a subclass.
Source code in blake2signer/mixins.py
class EncoderMixin(Mixin, ABC):
"""Encoder mixin.
Adds encoding capabilities to a subclass.
"""
def __init__(
self,
*args: typing.Any,
encoder: typing.Type[EncoderInterface] = B64URLEncoder,
**kwargs: typing.Any,
) -> None:
"""Add encoding capabilities.
Args:
*args: Additional positional arguments.
Keyword Args:
encoder (optional): Encoder class to use (defaults to a Base64 URL
safe encoder).
**kwargs: Additional keyword only arguments.
Returns:
None.
"""
self._encoder = self._validate_encoder(encoder)
personalisation = self._force_bytes(kwargs.get('personalisation', b''))
personalisation += self._encoder.__class__.__name__.encode()
kwargs['personalisation'] = personalisation
super().__init__(*args, **kwargs)
@staticmethod
def _validate_encoder(
encoder_class: typing.Type[EncoderInterface],
) -> EncoderInterface:
"""Validate the separator value and return it clean."""
encoder = encoder_class()
if not encoder.alphabet:
raise InvalidOptionError('the encoder alphabet must have a value')
if not encoder.alphabet.isascii():
raise InvalidOptionError('the encoder alphabet must be ASCII')
return encoder
def _encode(self, data: bytes) -> bytes:
"""Encode given data.
Raises:
EncodeError: Data can't be encoded.
"""
try:
return self._encoder.encode(data)
except Exception as exc:
raise EncodeError('data can not be encoded') from exc
def _decode(self, data: bytes) -> bytes:
"""Decode given encoded data.
Raises:
DecodeError: Data can't be decoded.
"""
try:
return self._encoder.decode(data)
except Exception as exc:
raise DecodeError('data can not be decoded') from exc
__init__(self, *args, *, encoder=<class 'blake2signer.encoders.B64URLEncoder'>, **kwargs)
special
¶
Add encoding capabilities.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
*args |
Any |
Additional positional arguments. |
() |
Keyword arguments:
Name | Type | Description |
---|---|---|
encoder |
optional |
Encoder class to use (defaults to a Base64 URL safe encoder). |
**kwargs |
Any |
Additional keyword only arguments. |
Returns:
Type | Description |
---|---|
None |
None. |
Source code in blake2signer/mixins.py
def __init__(
self,
*args: typing.Any,
encoder: typing.Type[EncoderInterface] = B64URLEncoder,
**kwargs: typing.Any,
) -> None:
"""Add encoding capabilities.
Args:
*args: Additional positional arguments.
Keyword Args:
encoder (optional): Encoder class to use (defaults to a Base64 URL
safe encoder).
**kwargs: Additional keyword only arguments.
Returns:
None.
"""
self._encoder = self._validate_encoder(encoder)
personalisation = self._force_bytes(kwargs.get('personalisation', b''))
personalisation += self._encoder.__class__.__name__.encode()
kwargs['personalisation'] = personalisation
super().__init__(*args, **kwargs)
Mixin (ABC)
¶
Base class for a Blake2Signer mixin.
Source code in blake2signer/mixins.py
class Mixin(ABC):
"""Base class for a Blake2Signer mixin."""
@staticmethod
def _force_bytes(value: typing.Any) -> bytes:
"""Force given value into bytes.
Args:
value: Value to convert to bytes.
Returns:
Converted value into bytes.
Raises:
ConversionError: Can't force value into bytes.
"""
try:
return force_bytes(value)
except Exception as exc:
raise ConversionError('value can not be converted to bytes') from exc
SerializerMixin (Mixin, ABC)
¶
Serializer mixin.
Adds serializing capabilities to a subclass.
Source code in blake2signer/mixins.py
class SerializerMixin(Mixin, ABC):
"""Serializer mixin.
Adds serializing capabilities to a subclass.
"""
def __init__(
self,
*args: typing.Any,
serializer: typing.Type[SerializerInterface] = JSONSerializer,
**kwargs: typing.Any,
) -> None:
"""Add serializing capabilities.
Args:
*args: Additional positional arguments.
Keyword Args:
serializer (optional): Serializer class to use (defaults to a JSON
serializer).
**kwargs: Additional keyword only arguments.
Returns:
None.
"""
self._serializer = serializer()
personalisation = self._force_bytes(kwargs.get('personalisation', b''))
personalisation += self._serializer.__class__.__name__.encode()
kwargs['personalisation'] = personalisation
super().__init__(*args, **kwargs)
def _serialize(self, data: typing.Any, **kwargs: typing.Any) -> bytes:
"""Serialize given data. Additional kwargs are passed to the serializer.
Raises:
SerializationError: Data can't be serialized.
"""
try:
return self._serializer.serialize(data, **kwargs)
except Exception as exc:
raise SerializationError('data can not be serialized') from exc
def _unserialize(self, data: bytes) -> typing.Any:
"""Unserialize given data.
Raises:
UnserializationError: Data can't be unserialized.
"""
try:
return self._serializer.unserialize(data)
except Exception as exc:
raise UnserializationError('data can not be unserialized') from exc
__init__(self, *args, *, serializer=<class 'blake2signer.serializers.JSONSerializer'>, **kwargs)
special
¶
Add serializing capabilities.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
*args |
Any |
Additional positional arguments. |
() |
Keyword arguments:
Name | Type | Description |
---|---|---|
serializer |
optional |
Serializer class to use (defaults to a JSON serializer). |
**kwargs |
Any |
Additional keyword only arguments. |
Returns:
Type | Description |
---|---|
None |
None. |
Source code in blake2signer/mixins.py
def __init__(
self,
*args: typing.Any,
serializer: typing.Type[SerializerInterface] = JSONSerializer,
**kwargs: typing.Any,
) -> None:
"""Add serializing capabilities.
Args:
*args: Additional positional arguments.
Keyword Args:
serializer (optional): Serializer class to use (defaults to a JSON
serializer).
**kwargs: Additional keyword only arguments.
Returns:
None.
"""
self._serializer = serializer()
personalisation = self._force_bytes(kwargs.get('personalisation', b''))
personalisation += self._serializer.__class__.__name__.encode()
kwargs['personalisation'] = personalisation
super().__init__(*args, **kwargs)