| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112 |
- from Crypto.Util._raw_api import (VoidPointer, SmartPointer,
- create_string_buffer,
- get_raw_buffer, c_size_t,
- c_uint8_ptr, c_ubyte)
- from Crypto.Util.number import long_to_bytes
- from Crypto.Util.py3compat import bchr
- from .keccak import _raw_keccak_lib
- class TurboSHAKE(object):
- """A TurboSHAKE hash object.
- Do not instantiate directly.
- Use the :func:`new` function.
- """
- def __init__(self, capacity, domain_separation, data):
- state = VoidPointer()
- result = _raw_keccak_lib.keccak_init(state.address_of(),
- c_size_t(capacity),
- c_ubyte(12)) # Reduced number of rounds
- if result:
- raise ValueError("Error %d while instantiating TurboSHAKE"
- % result)
- self._state = SmartPointer(state.get(), _raw_keccak_lib.keccak_destroy)
- self._is_squeezing = False
- self._capacity = capacity
- self._domain = domain_separation
- if data:
- self.update(data)
- def update(self, data):
- """Continue hashing of a message by consuming the next chunk of data.
- Args:
- data (byte string/byte array/memoryview): The next chunk of the message being hashed.
- """
- if self._is_squeezing:
- raise TypeError("You cannot call 'update' after the first 'read'")
- result = _raw_keccak_lib.keccak_absorb(self._state.get(),
- c_uint8_ptr(data),
- c_size_t(len(data)))
- if result:
- raise ValueError("Error %d while updating TurboSHAKE state"
- % result)
- return self
- def read(self, length):
- """
- Compute the next piece of XOF output.
- .. note::
- You cannot use :meth:`update` anymore after the first call to
- :meth:`read`.
- Args:
- length (integer): the amount of bytes this method must return
- :return: the next piece of XOF output (of the given length)
- :rtype: byte string
- """
- self._is_squeezing = True
- bfr = create_string_buffer(length)
- result = _raw_keccak_lib.keccak_squeeze(self._state.get(),
- bfr,
- c_size_t(length),
- c_ubyte(self._domain))
- if result:
- raise ValueError("Error %d while extracting from TurboSHAKE"
- % result)
- return get_raw_buffer(bfr)
- def new(self, data=None):
- return type(self)(self._capacity, self._domain, data)
- def _reset(self):
- result = _raw_keccak_lib.keccak_reset(self._state.get())
- if result:
- raise ValueError("Error %d while resetting TurboSHAKE state"
- % result)
- self._is_squeezing = False
- def new(**kwargs):
- """Create a new TurboSHAKE128 object.
- Args:
- domain (integer):
- Optional - A domain separation byte, between 0x01 and 0x7F.
- The default value is 0x1F.
- data (bytes/bytearray/memoryview):
- Optional - The very first chunk of the message to hash.
- It is equivalent to an early call to :meth:`update`.
- :Return: A :class:`TurboSHAKE` object
- """
- domain_separation = kwargs.get('domain', 0x1F)
- if not (0x01 <= domain_separation <= 0x7F):
- raise ValueError("Incorrect domain separation value (%d)" %
- domain_separation)
- data = kwargs.get('data')
- return TurboSHAKE(32, domain_separation, data=data)
|