1# Copyright 2013-2019 Donald Stufft and individual contributors
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14from __future__ import annotations
15
16from typing import TYPE_CHECKING, NoReturn
17
18from nacl import exceptions as exc
19from nacl._sodium import ffi, lib
20from nacl.exceptions import ensure
21
22if TYPE_CHECKING:
23 from typing_extensions import Self
24
25crypto_generichash_BYTES: int = lib.crypto_generichash_blake2b_bytes()
26crypto_generichash_BYTES_MIN: int = lib.crypto_generichash_blake2b_bytes_min()
27crypto_generichash_BYTES_MAX: int = lib.crypto_generichash_blake2b_bytes_max()
28crypto_generichash_KEYBYTES: int = lib.crypto_generichash_blake2b_keybytes()
29crypto_generichash_KEYBYTES_MIN: int = (
30 lib.crypto_generichash_blake2b_keybytes_min()
31)
32crypto_generichash_KEYBYTES_MAX: int = (
33 lib.crypto_generichash_blake2b_keybytes_max()
34)
35crypto_generichash_SALTBYTES: int = lib.crypto_generichash_blake2b_saltbytes()
36crypto_generichash_PERSONALBYTES: int = (
37 lib.crypto_generichash_blake2b_personalbytes()
38)
39crypto_generichash_STATEBYTES: int = lib.crypto_generichash_statebytes()
40
41_OVERLONG = "{0} length greater than {1} bytes"
42_TOOBIG = "{0} greater than {1}"
43
44
45def _checkparams(
46 digest_size: int, key: bytes, salt: bytes, person: bytes
47) -> None:
48 """Check hash parameters"""
49 ensure(
50 isinstance(key, bytes),
51 "Key must be a bytes sequence",
52 raising=exc.TypeError,
53 )
54
55 ensure(
56 isinstance(salt, bytes),
57 "Salt must be a bytes sequence",
58 raising=exc.TypeError,
59 )
60
61 ensure(
62 isinstance(person, bytes),
63 "Person must be a bytes sequence",
64 raising=exc.TypeError,
65 )
66
67 ensure(
68 isinstance(digest_size, int),
69 "Digest size must be an integer number",
70 raising=exc.TypeError,
71 )
72
73 ensure(
74 digest_size <= crypto_generichash_BYTES_MAX,
75 _TOOBIG.format("Digest_size", crypto_generichash_BYTES_MAX),
76 raising=exc.ValueError,
77 )
78
79 ensure(
80 len(key) <= crypto_generichash_KEYBYTES_MAX,
81 _OVERLONG.format("Key", crypto_generichash_KEYBYTES_MAX),
82 raising=exc.ValueError,
83 )
84
85 ensure(
86 len(salt) <= crypto_generichash_SALTBYTES,
87 _OVERLONG.format("Salt", crypto_generichash_SALTBYTES),
88 raising=exc.ValueError,
89 )
90
91 ensure(
92 len(person) <= crypto_generichash_PERSONALBYTES,
93 _OVERLONG.format("Person", crypto_generichash_PERSONALBYTES),
94 raising=exc.ValueError,
95 )
96
97
98def generichash_blake2b_salt_personal(
99 data: bytes,
100 digest_size: int = crypto_generichash_BYTES,
101 key: bytes = b"",
102 salt: bytes = b"",
103 person: bytes = b"",
104) -> bytes:
105 """One shot hash interface
106
107 :param data: the input data to the hash function
108 :type data: bytes
109 :param digest_size: must be at most
110 :py:data:`.crypto_generichash_BYTES_MAX`;
111 the default digest size is
112 :py:data:`.crypto_generichash_BYTES`
113 :type digest_size: int
114 :param key: must be at most
115 :py:data:`.crypto_generichash_KEYBYTES_MAX` long
116 :type key: bytes
117 :param salt: must be at most
118 :py:data:`.crypto_generichash_SALTBYTES` long;
119 will be zero-padded if needed
120 :type salt: bytes
121 :param person: must be at most
122 :py:data:`.crypto_generichash_PERSONALBYTES` long:
123 will be zero-padded if needed
124 :type person: bytes
125 :return: digest_size long digest
126 :rtype: bytes
127 """
128
129 _checkparams(digest_size, key, salt, person)
130
131 ensure(
132 isinstance(data, bytes),
133 "Input data must be a bytes sequence",
134 raising=exc.TypeError,
135 )
136
137 digest = ffi.new("unsigned char[]", digest_size)
138
139 # both _salt and _personal must be zero-padded to the correct length
140 _salt = ffi.new("unsigned char []", crypto_generichash_SALTBYTES)
141 _person = ffi.new("unsigned char []", crypto_generichash_PERSONALBYTES)
142
143 ffi.memmove(_salt, salt, len(salt))
144 ffi.memmove(_person, person, len(person))
145
146 rc = lib.crypto_generichash_blake2b_salt_personal(
147 digest, digest_size, data, len(data), key, len(key), _salt, _person
148 )
149 ensure(rc == 0, "Unexpected failure", raising=exc.RuntimeError)
150
151 return ffi.buffer(digest, digest_size)[:]
152
153
154class Blake2State:
155 """
156 Python-level wrapper for the crypto_generichash_blake2b state buffer
157 """
158
159 __slots__ = ["_statebuf", "digest_size"]
160
161 def __init__(self, digest_size: int):
162 self._statebuf = ffi.new(
163 "unsigned char[]", crypto_generichash_STATEBYTES
164 )
165 self.digest_size = digest_size
166
167 def __reduce__(self) -> NoReturn:
168 """
169 Raise the same exception as hashlib's blake implementation
170 on copy.copy()
171 """
172 raise TypeError(f"can't pickle {self.__class__.__name__} objects")
173
174 def copy(self) -> Self:
175 _st = self.__class__(self.digest_size)
176 ffi.memmove(
177 _st._statebuf, self._statebuf, crypto_generichash_STATEBYTES
178 )
179 return _st
180
181
182def generichash_blake2b_init(
183 key: bytes = b"",
184 salt: bytes = b"",
185 person: bytes = b"",
186 digest_size: int = crypto_generichash_BYTES,
187) -> Blake2State:
188 """
189 Create a new initialized blake2b hash state
190
191 :param key: must be at most
192 :py:data:`.crypto_generichash_KEYBYTES_MAX` long
193 :type key: bytes
194 :param salt: must be at most
195 :py:data:`.crypto_generichash_SALTBYTES` long;
196 will be zero-padded if needed
197 :type salt: bytes
198 :param person: must be at most
199 :py:data:`.crypto_generichash_PERSONALBYTES` long:
200 will be zero-padded if needed
201 :type person: bytes
202 :param digest_size: must be at most
203 :py:data:`.crypto_generichash_BYTES_MAX`;
204 the default digest size is
205 :py:data:`.crypto_generichash_BYTES`
206 :type digest_size: int
207 :return: a initialized :py:class:`.Blake2State`
208 :rtype: object
209 """
210
211 _checkparams(digest_size, key, salt, person)
212
213 state = Blake2State(digest_size)
214
215 # both _salt and _personal must be zero-padded to the correct length
216 _salt = ffi.new("unsigned char []", crypto_generichash_SALTBYTES)
217 _person = ffi.new("unsigned char []", crypto_generichash_PERSONALBYTES)
218
219 ffi.memmove(_salt, salt, len(salt))
220 ffi.memmove(_person, person, len(person))
221
222 rc = lib.crypto_generichash_blake2b_init_salt_personal(
223 state._statebuf, key, len(key), digest_size, _salt, _person
224 )
225 ensure(rc == 0, "Unexpected failure", raising=exc.RuntimeError)
226
227 return state
228
229
230def generichash_blake2b_update(state: Blake2State, data: bytes) -> None:
231 """Update the blake2b hash state
232
233 :param state: a initialized Blake2bState object as returned from
234 :py:func:`.crypto_generichash_blake2b_init`
235 :type state: :py:class:`.Blake2State`
236 :param data:
237 :type data: bytes
238 """
239
240 ensure(
241 isinstance(state, Blake2State),
242 "State must be a Blake2State object",
243 raising=exc.TypeError,
244 )
245
246 ensure(
247 isinstance(data, bytes),
248 "Input data must be a bytes sequence",
249 raising=exc.TypeError,
250 )
251
252 rc = lib.crypto_generichash_blake2b_update(
253 state._statebuf, data, len(data)
254 )
255 ensure(rc == 0, "Unexpected failure", raising=exc.RuntimeError)
256
257
258def generichash_blake2b_final(state: Blake2State) -> bytes:
259 """Finalize the blake2b hash state and return the digest.
260
261 :param state: a initialized Blake2bState object as returned from
262 :py:func:`.crypto_generichash_blake2b_init`
263 :type state: :py:class:`.Blake2State`
264 :return: the blake2 digest of the passed-in data stream
265 :rtype: bytes
266 """
267
268 ensure(
269 isinstance(state, Blake2State),
270 "State must be a Blake2State object",
271 raising=exc.TypeError,
272 )
273
274 _digest = ffi.new("unsigned char[]", crypto_generichash_BYTES_MAX)
275 rc = lib.crypto_generichash_blake2b_final(
276 state._statebuf, _digest, state.digest_size
277 )
278
279 ensure(rc == 0, "Unexpected failure", raising=exc.RuntimeError)
280 return ffi.buffer(_digest, state.digest_size)[:]