TurboSHAKE128.py 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112
  1. from Crypto.Util._raw_api import (VoidPointer, SmartPointer,
  2. create_string_buffer,
  3. get_raw_buffer, c_size_t,
  4. c_uint8_ptr, c_ubyte)
  5. from Crypto.Util.number import long_to_bytes
  6. from Crypto.Util.py3compat import bchr
  7. from .keccak import _raw_keccak_lib
  8. class TurboSHAKE(object):
  9. """A TurboSHAKE hash object.
  10. Do not instantiate directly.
  11. Use the :func:`new` function.
  12. """
  13. def __init__(self, capacity, domain_separation, data):
  14. state = VoidPointer()
  15. result = _raw_keccak_lib.keccak_init(state.address_of(),
  16. c_size_t(capacity),
  17. c_ubyte(12)) # Reduced number of rounds
  18. if result:
  19. raise ValueError("Error %d while instantiating TurboSHAKE"
  20. % result)
  21. self._state = SmartPointer(state.get(), _raw_keccak_lib.keccak_destroy)
  22. self._is_squeezing = False
  23. self._capacity = capacity
  24. self._domain = domain_separation
  25. if data:
  26. self.update(data)
  27. def update(self, data):
  28. """Continue hashing of a message by consuming the next chunk of data.
  29. Args:
  30. data (byte string/byte array/memoryview): The next chunk of the message being hashed.
  31. """
  32. if self._is_squeezing:
  33. raise TypeError("You cannot call 'update' after the first 'read'")
  34. result = _raw_keccak_lib.keccak_absorb(self._state.get(),
  35. c_uint8_ptr(data),
  36. c_size_t(len(data)))
  37. if result:
  38. raise ValueError("Error %d while updating TurboSHAKE state"
  39. % result)
  40. return self
  41. def read(self, length):
  42. """
  43. Compute the next piece of XOF output.
  44. .. note::
  45. You cannot use :meth:`update` anymore after the first call to
  46. :meth:`read`.
  47. Args:
  48. length (integer): the amount of bytes this method must return
  49. :return: the next piece of XOF output (of the given length)
  50. :rtype: byte string
  51. """
  52. self._is_squeezing = True
  53. bfr = create_string_buffer(length)
  54. result = _raw_keccak_lib.keccak_squeeze(self._state.get(),
  55. bfr,
  56. c_size_t(length),
  57. c_ubyte(self._domain))
  58. if result:
  59. raise ValueError("Error %d while extracting from TurboSHAKE"
  60. % result)
  61. return get_raw_buffer(bfr)
  62. def new(self, data=None):
  63. return type(self)(self._capacity, self._domain, data)
  64. def _reset(self):
  65. result = _raw_keccak_lib.keccak_reset(self._state.get())
  66. if result:
  67. raise ValueError("Error %d while resetting TurboSHAKE state"
  68. % result)
  69. self._is_squeezing = False
  70. def new(**kwargs):
  71. """Create a new TurboSHAKE128 object.
  72. Args:
  73. domain (integer):
  74. Optional - A domain separation byte, between 0x01 and 0x7F.
  75. The default value is 0x1F.
  76. data (bytes/bytearray/memoryview):
  77. Optional - The very first chunk of the message to hash.
  78. It is equivalent to an early call to :meth:`update`.
  79. :Return: A :class:`TurboSHAKE` object
  80. """
  81. domain_separation = kwargs.get('domain', 0x1F)
  82. if not (0x01 <= domain_separation <= 0x7F):
  83. raise ValueError("Incorrect domain separation value (%d)" %
  84. domain_separation)
  85. data = kwargs.get('data')
  86. return TurboSHAKE(32, domain_separation, data=data)