Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2013-2018 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
16from typing import Union, cast
18from nacl import exceptions as exc
19from nacl._sodium import ffi, lib
20from nacl.exceptions import ensure
22crypto_secretstream_xchacha20poly1305_ABYTES: int = (
23 lib.crypto_secretstream_xchacha20poly1305_abytes()
24)
25crypto_secretstream_xchacha20poly1305_HEADERBYTES: int = (
26 lib.crypto_secretstream_xchacha20poly1305_headerbytes()
27)
28crypto_secretstream_xchacha20poly1305_KEYBYTES: int = (
29 lib.crypto_secretstream_xchacha20poly1305_keybytes()
30)
31crypto_secretstream_xchacha20poly1305_MESSAGEBYTES_MAX: int = (
32 lib.crypto_secretstream_xchacha20poly1305_messagebytes_max()
33)
34crypto_secretstream_xchacha20poly1305_STATEBYTES: int = (
35 lib.crypto_secretstream_xchacha20poly1305_statebytes()
36)
39crypto_secretstream_xchacha20poly1305_TAG_MESSAGE: int = (
40 lib.crypto_secretstream_xchacha20poly1305_tag_message()
41)
42crypto_secretstream_xchacha20poly1305_TAG_PUSH: int = (
43 lib.crypto_secretstream_xchacha20poly1305_tag_push()
44)
45crypto_secretstream_xchacha20poly1305_TAG_REKEY: int = (
46 lib.crypto_secretstream_xchacha20poly1305_tag_rekey()
47)
48crypto_secretstream_xchacha20poly1305_TAG_FINAL: int = (
49 lib.crypto_secretstream_xchacha20poly1305_tag_final()
50)
53def crypto_secretstream_xchacha20poly1305_keygen() -> bytes:
54 """
55 Generate a key for use with
56 :func:`.crypto_secretstream_xchacha20poly1305_init_push`.
58 """
59 keybuf = ffi.new(
60 "unsigned char[]",
61 crypto_secretstream_xchacha20poly1305_KEYBYTES,
62 )
63 lib.crypto_secretstream_xchacha20poly1305_keygen(keybuf)
64 return ffi.buffer(keybuf)[:]
67class crypto_secretstream_xchacha20poly1305_state:
68 """
69 An object wrapping the crypto_secretstream_xchacha20poly1305 state.
71 """
73 __slots__ = ["rawbuf", "statebuf", "tagbuf"]
75 def __init__(self) -> None:
76 """Initialize a clean state object."""
77 ByteString = Union[bytes, bytearray, memoryview]
78 self.statebuf: ByteString = ffi.new(
79 "unsigned char[]",
80 crypto_secretstream_xchacha20poly1305_STATEBYTES,
81 )
83 self.rawbuf: ByteString | None = None
84 self.tagbuf: ByteString | None = None
87def crypto_secretstream_xchacha20poly1305_init_push(
88 state: crypto_secretstream_xchacha20poly1305_state, key: bytes
89) -> bytes:
90 """
91 Initialize a crypto_secretstream_xchacha20poly1305 encryption buffer.
93 :param state: a secretstream state object
94 :type state: crypto_secretstream_xchacha20poly1305_state
95 :param key: must be
96 :data:`.crypto_secretstream_xchacha20poly1305_KEYBYTES` long
97 :type key: bytes
98 :return: header
99 :rtype: bytes
101 """
102 ensure(
103 isinstance(state, crypto_secretstream_xchacha20poly1305_state),
104 "State must be a crypto_secretstream_xchacha20poly1305_state object",
105 raising=exc.TypeError,
106 )
107 ensure(
108 isinstance(key, bytes),
109 "Key must be a bytes sequence",
110 raising=exc.TypeError,
111 )
112 ensure(
113 len(key) == crypto_secretstream_xchacha20poly1305_KEYBYTES,
114 "Invalid key length",
115 raising=exc.ValueError,
116 )
118 headerbuf = ffi.new(
119 "unsigned char []",
120 crypto_secretstream_xchacha20poly1305_HEADERBYTES,
121 )
123 rc = lib.crypto_secretstream_xchacha20poly1305_init_push(
124 state.statebuf, headerbuf, key
125 )
126 ensure(rc == 0, "Unexpected failure", raising=exc.RuntimeError)
128 return ffi.buffer(headerbuf)[:]
131def crypto_secretstream_xchacha20poly1305_push(
132 state: crypto_secretstream_xchacha20poly1305_state,
133 m: bytes,
134 ad: bytes | None = None,
135 tag: int = crypto_secretstream_xchacha20poly1305_TAG_MESSAGE,
136) -> bytes:
137 """
138 Add an encrypted message to the secret stream.
140 :param state: a secretstream state object
141 :type state: crypto_secretstream_xchacha20poly1305_state
142 :param m: the message to encrypt, the maximum length of an individual
143 message is
144 :data:`.crypto_secretstream_xchacha20poly1305_MESSAGEBYTES_MAX`.
145 :type m: bytes
146 :param ad: additional data to include in the authentication tag
147 :type ad: bytes or None
148 :param tag: the message tag, usually
149 :data:`.crypto_secretstream_xchacha20poly1305_TAG_MESSAGE` or
150 :data:`.crypto_secretstream_xchacha20poly1305_TAG_FINAL`.
151 :type tag: int
152 :return: ciphertext
153 :rtype: bytes
155 """
156 ensure(
157 isinstance(state, crypto_secretstream_xchacha20poly1305_state),
158 "State must be a crypto_secretstream_xchacha20poly1305_state object",
159 raising=exc.TypeError,
160 )
161 ensure(isinstance(m, bytes), "Message is not bytes", raising=exc.TypeError)
162 ensure(
163 len(m) <= crypto_secretstream_xchacha20poly1305_MESSAGEBYTES_MAX,
164 "Message is too long",
165 raising=exc.ValueError,
166 )
167 ensure(
168 ad is None or isinstance(ad, bytes),
169 "Additional data must be bytes or None",
170 raising=exc.TypeError,
171 )
173 clen = len(m) + crypto_secretstream_xchacha20poly1305_ABYTES
174 if state.rawbuf is None or len(state.rawbuf) < clen:
175 state.rawbuf = ffi.new("unsigned char[]", clen)
177 if ad is None:
178 ad = ffi.NULL
179 adlen = 0
180 else:
181 adlen = len(ad)
183 rc = lib.crypto_secretstream_xchacha20poly1305_push(
184 state.statebuf,
185 state.rawbuf,
186 ffi.NULL,
187 m,
188 len(m),
189 ad,
190 adlen,
191 tag,
192 )
193 ensure(rc == 0, "Unexpected failure", raising=exc.RuntimeError)
195 return ffi.buffer(state.rawbuf, clen)[:]
198def crypto_secretstream_xchacha20poly1305_init_pull(
199 state: crypto_secretstream_xchacha20poly1305_state,
200 header: bytes,
201 key: bytes,
202) -> None:
203 """
204 Initialize a crypto_secretstream_xchacha20poly1305 decryption buffer.
206 :param state: a secretstream state object
207 :type state: crypto_secretstream_xchacha20poly1305_state
208 :param header: must be
209 :data:`.crypto_secretstream_xchacha20poly1305_HEADERBYTES` long
210 :type header: bytes
211 :param key: must be
212 :data:`.crypto_secretstream_xchacha20poly1305_KEYBYTES` long
213 :type key: bytes
215 """
216 ensure(
217 isinstance(state, crypto_secretstream_xchacha20poly1305_state),
218 "State must be a crypto_secretstream_xchacha20poly1305_state object",
219 raising=exc.TypeError,
220 )
221 ensure(
222 isinstance(header, bytes),
223 "Header must be a bytes sequence",
224 raising=exc.TypeError,
225 )
226 ensure(
227 len(header) == crypto_secretstream_xchacha20poly1305_HEADERBYTES,
228 "Invalid header length",
229 raising=exc.ValueError,
230 )
231 ensure(
232 isinstance(key, bytes),
233 "Key must be a bytes sequence",
234 raising=exc.TypeError,
235 )
236 ensure(
237 len(key) == crypto_secretstream_xchacha20poly1305_KEYBYTES,
238 "Invalid key length",
239 raising=exc.ValueError,
240 )
242 if state.tagbuf is None:
243 state.tagbuf = ffi.new("unsigned char *")
245 rc = lib.crypto_secretstream_xchacha20poly1305_init_pull(
246 state.statebuf, header, key
247 )
248 ensure(rc == 0, "Unexpected failure", raising=exc.RuntimeError)
251def crypto_secretstream_xchacha20poly1305_pull(
252 state: crypto_secretstream_xchacha20poly1305_state,
253 c: bytes,
254 ad: bytes | None = None,
255) -> tuple[bytes, int]:
256 """
257 Read a decrypted message from the secret stream.
259 :param state: a secretstream state object
260 :type state: crypto_secretstream_xchacha20poly1305_state
261 :param c: the ciphertext to decrypt, the maximum length of an individual
262 ciphertext is
263 :data:`.crypto_secretstream_xchacha20poly1305_MESSAGEBYTES_MAX` +
264 :data:`.crypto_secretstream_xchacha20poly1305_ABYTES`.
265 :type c: bytes
266 :param ad: additional data to include in the authentication tag
267 :type ad: bytes or None
268 :return: (message, tag)
269 :rtype: (bytes, int)
271 """
272 ensure(
273 isinstance(state, crypto_secretstream_xchacha20poly1305_state),
274 "State must be a crypto_secretstream_xchacha20poly1305_state object",
275 raising=exc.TypeError,
276 )
277 ensure(
278 state.tagbuf is not None,
279 (
280 "State must be initialized using "
281 "crypto_secretstream_xchacha20poly1305_init_pull"
282 ),
283 raising=exc.ValueError,
284 )
285 ensure(
286 isinstance(c, bytes),
287 "Ciphertext is not bytes",
288 raising=exc.TypeError,
289 )
290 ensure(
291 len(c) >= crypto_secretstream_xchacha20poly1305_ABYTES,
292 "Ciphertext is too short",
293 raising=exc.ValueError,
294 )
295 ensure(
296 len(c)
297 <= (
298 crypto_secretstream_xchacha20poly1305_MESSAGEBYTES_MAX
299 + crypto_secretstream_xchacha20poly1305_ABYTES
300 ),
301 "Ciphertext is too long",
302 raising=exc.ValueError,
303 )
304 ensure(
305 ad is None or isinstance(ad, bytes),
306 "Additional data must be bytes or None",
307 raising=exc.TypeError,
308 )
310 mlen = len(c) - crypto_secretstream_xchacha20poly1305_ABYTES
311 if state.rawbuf is None or len(state.rawbuf) < mlen:
312 state.rawbuf = ffi.new("unsigned char[]", mlen)
314 if ad is None:
315 ad = ffi.NULL
316 adlen = 0
317 else:
318 adlen = len(ad)
320 rc = lib.crypto_secretstream_xchacha20poly1305_pull(
321 state.statebuf,
322 state.rawbuf,
323 ffi.NULL,
324 state.tagbuf,
325 c,
326 len(c),
327 ad,
328 adlen,
329 )
330 ensure(rc == 0, "Unexpected failure", raising=exc.RuntimeError)
332 # Cast safety: we `ensure` above that `state.tagbuf is not None`.
333 return (
334 ffi.buffer(state.rawbuf, mlen)[:],
335 int(cast(bytes, state.tagbuf)[0]),
336 )
339def crypto_secretstream_xchacha20poly1305_rekey(
340 state: crypto_secretstream_xchacha20poly1305_state,
341) -> None:
342 """
343 Explicitly change the encryption key in the stream.
345 Normally the stream is re-keyed as needed or an explicit ``tag`` of
346 :data:`.crypto_secretstream_xchacha20poly1305_TAG_REKEY` is added to a
347 message to ensure forward secrecy, but this method can be used instead
348 if the re-keying is controlled without adding the tag.
350 :param state: a secretstream state object
351 :type state: crypto_secretstream_xchacha20poly1305_state
353 """
354 ensure(
355 isinstance(state, crypto_secretstream_xchacha20poly1305_state),
356 "State must be a crypto_secretstream_xchacha20poly1305_state object",
357 raising=exc.TypeError,
358 )
359 lib.crypto_secretstream_xchacha20poly1305_rekey(state.statebuf)