random(): per-thread singleton; random_key/random_nonce use it
Each cipher module's random() now returns a thread-local Random that is created on first use and reused thereafter. Each thread gets its own OS-seeded generator (own key+nonce), so no locking and no cross-thread nonce collisions. random_key() and random_nonce() draw from this CSPRNG instead of secrets.token_bytes, making them as fast as the stream. Changes made in the aegis256x4.py template and regenerated.
This commit is contained in:
1 parent
32954c4eba
commit
9d36d3b81b
9 files changed
+163
-64
No files matched your search
@@ -0,0 +1,56 @@
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from aeg import aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4
|
||||
|
||||
CIPHERS = [aegis128l, aegis128x2, aegis128x4, aegis256, aegis256x2, aegis256x4]
|
||||
|
||||
|
||||
def cipher_id(c):
|
||||
return c.NAME
|
||||
|
||||
|
||||
@pytest.mark.parametrize("c", CIPHERS, ids=cipher_id)
|
||||
def test_random_is_per_thread_singleton(c):
|
||||
assert c.random() is c.random()
|
||||
other = []
|
||||
t = threading.Thread(target=lambda: other.append(c.random()))
|
||||
t.start()
|
||||
t.join()
|
||||
assert other[0] is not c.random()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("c", CIPHERS, ids=cipher_id)
|
||||
def test_random_key_nonce_from_csprng(c):
|
||||
assert len(c.random_key()) == c.KEYBYTES
|
||||
assert len(c.random_nonce()) == c.NONCEBYTES
|
||||
# successive outputs never repeat
|
||||
assert c.random_key() != c.random_key()
|
||||
assert c.random_nonce() != c.random_nonce()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("c", CIPHERS, ids=cipher_id)
|
||||
def test_threads_get_independent_streams(c):
|
||||
results = [None] * 2
|
||||
|
||||
def work(i):
|
||||
results[i] = bytes(c.random_key())
|
||||
|
||||
threads = [threading.Thread(target=work, args=(i,)) for i in range(2)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
assert results[0] != results[1]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("c", CIPHERS, ids=cipher_id)
|
||||
def test_bytes_into_callable(c):
|
||||
rng = c.random()
|
||||
a = rng.bytes(64)
|
||||
buf = bytearray(64)
|
||||
rng.into(buf)
|
||||
b = rng(64)
|
||||
assert len(a) == len(b) == 64
|
||||
assert a != bytes(buf) != b # three consecutive blocks differ
|
||||
Reference in new issue
Block a user