Skip to content

Mixins

blake2signer.mixins

Mixins: abstract classes that provide certain restricted functionality.

They work as a building block for other classes.

CompressorMixin

Bases: Mixin, ABC

Compressor mixin.

Adds compressing capabilities to a subclass.

Source code in blake2signer/mixins.py
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
class CompressorMixin(Mixin, ABC):
    """Compressor mixin.

    Adds compressing capabilities to a subclass.
    """

    # ToDo: D417 is a false positive, see https://github.com/PyCQA/pydocstyle/issues/514
    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:  # noqa: D417
        """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.
        """
        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.

        Args:
            flag: compression flag to validate.

        Returns:
            Validated compression flag as bytes.

        Raises:
            InvalidOptionError: the compression flag is not valid.
        """
        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.

        Args:
            ratio: compression ratio to validate.

        Returns:
            Validated compression ratio as float.

        Raises:
            InvalidOptionError: the compression ratio is out of bounds.
        """
        if 0.0 <= ratio < 100.0:
            return float(ratio)

        raise InvalidOptionError('the compression ratio must be between 0 and less than 100')

    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.

        Args:
            data: compressed data to decompress.

        Returns:
             Original 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__(*args, compressor=ZlibCompressor, compression_flag=b'.', compression_ratio=5.0, **kwargs)

Add compressing capabilities.

Parameters:

Name Type Description Default
*args typing.Any

Additional positional arguments.

()

Other Parameters:

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 typing.Any

Additional keyword only arguments.

Source code in blake2signer/mixins.py
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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:  # noqa: D417
    """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.
    """
    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

Bases: Mixin, ABC

Encoder mixin.

Adds encoding capabilities to a subclass.

Source code in blake2signer/mixins.py
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
class EncoderMixin(Mixin, ABC):
    """Encoder mixin.

    Adds encoding capabilities to a subclass.
    """

    # ToDo: D417 is a false positive, see https://github.com/PyCQA/pydocstyle/issues/514
    def __init__(
        self,
        *args: typing.Any,
        encoder: typing.Type[EncoderInterface] = B64URLEncoder,
        **kwargs: typing.Any,
    ) -> None:  # noqa: D417
        """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.
        """
        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 encoder characteristics and return it clean.

        Args:
            encoder_class: encoder class to validate.

        Returns:
            Validated encoder instance.

        Raises:
            InvalidOptionError: the encoder alphabet is empty or is not ASCII.
        """
        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.

        Args:
            data: data to encode.

        Returns:
            Encoded 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.

        Args:
            data: encoded data to decode.

        Returns:
            Original 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__(*args, encoder=B64URLEncoder, **kwargs)

Add encoding capabilities.

Parameters:

Name Type Description Default
*args typing.Any

Additional positional arguments.

()

Other Parameters:

Name Type Description
encoder optional

Encoder class to use (defaults to a Base64 URL safe encoder).

**kwargs typing.Any

Additional keyword only arguments.

Source code in blake2signer/mixins.py
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def __init__(
    self,
    *args: typing.Any,
    encoder: typing.Type[EncoderInterface] = B64URLEncoder,
    **kwargs: typing.Any,
) -> None:  # noqa: D417
    """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.
    """
    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

Bases: ABC

Base class for a Blake2Signer mixin.

Source code in blake2signer/mixins.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class Mixin(ABC):  # pylint: disable=R0903  # noqa: B024  # false positives
    """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

Bases: Mixin, ABC

Serializer mixin.

Adds serializing capabilities to a subclass.

Source code in blake2signer/mixins.py
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class SerializerMixin(Mixin, ABC):
    """Serializer mixin.

    Adds serializing capabilities to a subclass.
    """

    # ToDo: D417 is a false positive, see https://github.com/PyCQA/pydocstyle/issues/514
    def __init__(
        self,
        *args: typing.Any,
        serializer: typing.Type[SerializerInterface] = JSONSerializer,
        **kwargs: typing.Any,
    ) -> None:  # noqa: D417
        """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.
        """
        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.

        Args:
            data: data to serialize.

        Keyword Args:
          **kwargs: Additional keyword only arguments for the serializer.

        Returns:
            Serialized data.

        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.

        Args:
            data: serialized data to unserialize.

        Returns:
            Original 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__(*args, serializer=JSONSerializer, **kwargs)

Add serializing capabilities.

Parameters:

Name Type Description Default
*args typing.Any

Additional positional arguments.

()

Other Parameters:

Name Type Description
serializer optional

Serializer class to use (defaults to a JSON serializer).

**kwargs typing.Any

Additional keyword only arguments.

Source code in blake2signer/mixins.py
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    *args: typing.Any,
    serializer: typing.Type[SerializerInterface] = JSONSerializer,
    **kwargs: typing.Any,
) -> None:  # noqa: D417
    """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.
    """
    self._serializer = serializer()

    personalisation = self._force_bytes(kwargs.get('personalisation', b''))
    personalisation += self._serializer.__class__.__name__.encode()
    kwargs['personalisation'] = personalisation

    super().__init__(*args, **kwargs)