diff --git a/README.md b/README.md index 532d63a..a6fa919 100644 --- a/README.md +++ b/README.md @@ -128,9 +128,9 @@ Constants (per module): NAME, KEYBYTES, NONCEBYTES, MACBYTES, MACBYTES_LONG, RAT ### CSPRNG (aeg.random) -Seeded once from the OS, then a deterministic AEGIS keystream: each call increments the nonce, so no output block ever repeats. +Seeded once from the OS, then a deterministic AEGIS keystream: each call increments the nonce, so no output block ever repeats. random_key() and random_nonce() draw from it. -- ciph.random() -> aeg.random.Random bound to the cipher module +- ciph.random() -> per-thread aeg.random.Random singleton for the cipher module - rng.bytes(n) -> bytearray, rng.into(buf), rng(n) ### Random-access encrypted files (aeg.raf) diff --git a/src/aeg/aegis128l.py b/src/aeg/aegis128l.py index 6673b05..e280cdc 100644 --- a/src/aeg/aegis128l.py +++ b/src/aeg/aegis128l.py @@ -3,8 +3,8 @@ # DO NOT EDIT OTHER ALGORITHM FILES MANUALLY! import errno -import secrets import sys +import threading from typing import Literal from ._loader import ffi @@ -21,34 +21,41 @@ ALIGNMENT = 64 #: Required alignment for internal structures RATE = 64 #: Byte chunk size in internal processing -def random() -> Random: - """Return a new AEGIS stream Random instance using this cipher module. +_random_tls = threading.local() - Seeded once from the OS, then deterministic: successive calls never - repeat an output block. See aeg.random.Random for the API. + +def random() -> Random: + """Return this module's Random instance (per-thread singleton). + + Each thread gets its own generator on first use, seeded once from the + OS, then deterministic: successive calls never repeat an output block. + No locking is involved. See aeg.random.Random for the API. """ - return Random(sys.modules[__name__]) + rng = getattr(_random_tls, "rng", None) + if rng is None: + rng = _random_tls.rng = Random(sys.modules[__name__]) + return rng def random_key() -> bytearray: """ - Generate a secret key using cryptographically secure random bytes. + Generate a secret key using the module's CSPRNG (see random()). It is recommended to wipe() the key after no longer needed. """ - return bytearray(secrets.token_bytes(KEYBYTES)) + return random().bytes(KEYBYTES) def random_nonce() -> bytearray: """ - Generate a public nonce using cryptographically secure random bytes. + Generate a public nonce using the module's CSPRNG (see random()). Nonces (a number used once) are public data that may be sent together with the ciphertext, but they need to be unique for each use. See also: nonce_increment() can be used to derive sequential nonces. """ - return bytearray(secrets.token_bytes(NONCEBYTES)) + return random().bytes(NONCEBYTES) def _ptr(buf): diff --git a/src/aeg/aegis128x2.py b/src/aeg/aegis128x2.py index c1f8180..6397b3c 100644 --- a/src/aeg/aegis128x2.py +++ b/src/aeg/aegis128x2.py @@ -3,8 +3,8 @@ # DO NOT EDIT OTHER ALGORITHM FILES MANUALLY! import errno -import secrets import sys +import threading from typing import Literal from ._loader import ffi @@ -21,34 +21,41 @@ ALIGNMENT = 64 #: Required alignment for internal structures RATE = 64 #: Byte chunk size in internal processing -def random() -> Random: - """Return a new AEGIS stream Random instance using this cipher module. +_random_tls = threading.local() - Seeded once from the OS, then deterministic: successive calls never - repeat an output block. See aeg.random.Random for the API. + +def random() -> Random: + """Return this module's Random instance (per-thread singleton). + + Each thread gets its own generator on first use, seeded once from the + OS, then deterministic: successive calls never repeat an output block. + No locking is involved. See aeg.random.Random for the API. """ - return Random(sys.modules[__name__]) + rng = getattr(_random_tls, "rng", None) + if rng is None: + rng = _random_tls.rng = Random(sys.modules[__name__]) + return rng def random_key() -> bytearray: """ - Generate a secret key using cryptographically secure random bytes. + Generate a secret key using the module's CSPRNG (see random()). It is recommended to wipe() the key after no longer needed. """ - return bytearray(secrets.token_bytes(KEYBYTES)) + return random().bytes(KEYBYTES) def random_nonce() -> bytearray: """ - Generate a public nonce using cryptographically secure random bytes. + Generate a public nonce using the module's CSPRNG (see random()). Nonces (a number used once) are public data that may be sent together with the ciphertext, but they need to be unique for each use. See also: nonce_increment() can be used to derive sequential nonces. """ - return bytearray(secrets.token_bytes(NONCEBYTES)) + return random().bytes(NONCEBYTES) def _ptr(buf): diff --git a/src/aeg/aegis128x4.py b/src/aeg/aegis128x4.py index 234d136..d8006ab 100644 --- a/src/aeg/aegis128x4.py +++ b/src/aeg/aegis128x4.py @@ -3,8 +3,8 @@ # DO NOT EDIT OTHER ALGORITHM FILES MANUALLY! import errno -import secrets import sys +import threading from typing import Literal from ._loader import ffi @@ -21,34 +21,41 @@ ALIGNMENT = 64 #: Required alignment for internal structures RATE = 64 #: Byte chunk size in internal processing -def random() -> Random: - """Return a new AEGIS stream Random instance using this cipher module. +_random_tls = threading.local() - Seeded once from the OS, then deterministic: successive calls never - repeat an output block. See aeg.random.Random for the API. + +def random() -> Random: + """Return this module's Random instance (per-thread singleton). + + Each thread gets its own generator on first use, seeded once from the + OS, then deterministic: successive calls never repeat an output block. + No locking is involved. See aeg.random.Random for the API. """ - return Random(sys.modules[__name__]) + rng = getattr(_random_tls, "rng", None) + if rng is None: + rng = _random_tls.rng = Random(sys.modules[__name__]) + return rng def random_key() -> bytearray: """ - Generate a secret key using cryptographically secure random bytes. + Generate a secret key using the module's CSPRNG (see random()). It is recommended to wipe() the key after no longer needed. """ - return bytearray(secrets.token_bytes(KEYBYTES)) + return random().bytes(KEYBYTES) def random_nonce() -> bytearray: """ - Generate a public nonce using cryptographically secure random bytes. + Generate a public nonce using the module's CSPRNG (see random()). Nonces (a number used once) are public data that may be sent together with the ciphertext, but they need to be unique for each use. See also: nonce_increment() can be used to derive sequential nonces. """ - return bytearray(secrets.token_bytes(NONCEBYTES)) + return random().bytes(NONCEBYTES) def _ptr(buf): diff --git a/src/aeg/aegis256.py b/src/aeg/aegis256.py index 3b9c6f2..8dd19d1 100644 --- a/src/aeg/aegis256.py +++ b/src/aeg/aegis256.py @@ -3,8 +3,8 @@ # DO NOT EDIT OTHER ALGORITHM FILES MANUALLY! import errno -import secrets import sys +import threading from typing import Literal from ._loader import ffi @@ -21,34 +21,41 @@ ALIGNMENT = 64 #: Required alignment for internal structures RATE = 64 #: Byte chunk size in internal processing -def random() -> Random: - """Return a new AEGIS stream Random instance using this cipher module. +_random_tls = threading.local() - Seeded once from the OS, then deterministic: successive calls never - repeat an output block. See aeg.random.Random for the API. + +def random() -> Random: + """Return this module's Random instance (per-thread singleton). + + Each thread gets its own generator on first use, seeded once from the + OS, then deterministic: successive calls never repeat an output block. + No locking is involved. See aeg.random.Random for the API. """ - return Random(sys.modules[__name__]) + rng = getattr(_random_tls, "rng", None) + if rng is None: + rng = _random_tls.rng = Random(sys.modules[__name__]) + return rng def random_key() -> bytearray: """ - Generate a secret key using cryptographically secure random bytes. + Generate a secret key using the module's CSPRNG (see random()). It is recommended to wipe() the key after no longer needed. """ - return bytearray(secrets.token_bytes(KEYBYTES)) + return random().bytes(KEYBYTES) def random_nonce() -> bytearray: """ - Generate a public nonce using cryptographically secure random bytes. + Generate a public nonce using the module's CSPRNG (see random()). Nonces (a number used once) are public data that may be sent together with the ciphertext, but they need to be unique for each use. See also: nonce_increment() can be used to derive sequential nonces. """ - return bytearray(secrets.token_bytes(NONCEBYTES)) + return random().bytes(NONCEBYTES) def _ptr(buf): diff --git a/src/aeg/aegis256x2.py b/src/aeg/aegis256x2.py index 620992d..c38dfd9 100644 --- a/src/aeg/aegis256x2.py +++ b/src/aeg/aegis256x2.py @@ -3,8 +3,8 @@ # DO NOT EDIT OTHER ALGORITHM FILES MANUALLY! import errno -import secrets import sys +import threading from typing import Literal from ._loader import ffi @@ -21,34 +21,41 @@ ALIGNMENT = 64 #: Required alignment for internal structures RATE = 64 #: Byte chunk size in internal processing -def random() -> Random: - """Return a new AEGIS stream Random instance using this cipher module. +_random_tls = threading.local() - Seeded once from the OS, then deterministic: successive calls never - repeat an output block. See aeg.random.Random for the API. + +def random() -> Random: + """Return this module's Random instance (per-thread singleton). + + Each thread gets its own generator on first use, seeded once from the + OS, then deterministic: successive calls never repeat an output block. + No locking is involved. See aeg.random.Random for the API. """ - return Random(sys.modules[__name__]) + rng = getattr(_random_tls, "rng", None) + if rng is None: + rng = _random_tls.rng = Random(sys.modules[__name__]) + return rng def random_key() -> bytearray: """ - Generate a secret key using cryptographically secure random bytes. + Generate a secret key using the module's CSPRNG (see random()). It is recommended to wipe() the key after no longer needed. """ - return bytearray(secrets.token_bytes(KEYBYTES)) + return random().bytes(KEYBYTES) def random_nonce() -> bytearray: """ - Generate a public nonce using cryptographically secure random bytes. + Generate a public nonce using the module's CSPRNG (see random()). Nonces (a number used once) are public data that may be sent together with the ciphertext, but they need to be unique for each use. See also: nonce_increment() can be used to derive sequential nonces. """ - return bytearray(secrets.token_bytes(NONCEBYTES)) + return random().bytes(NONCEBYTES) def _ptr(buf): diff --git a/src/aeg/aegis256x4.py b/src/aeg/aegis256x4.py index 8126272..540663e 100644 --- a/src/aeg/aegis256x4.py +++ b/src/aeg/aegis256x4.py @@ -3,8 +3,8 @@ # DO NOT EDIT OTHER ALGORITHM FILES MANUALLY! import errno -import secrets import sys +import threading from typing import Literal from ._loader import ffi @@ -21,34 +21,41 @@ ALIGNMENT = 64 #: Required alignment for internal structures RATE = 64 #: Byte chunk size in internal processing -def random() -> Random: - """Return a new AEGIS stream Random instance using this cipher module. +_random_tls = threading.local() - Seeded once from the OS, then deterministic: successive calls never - repeat an output block. See aeg.random.Random for the API. + +def random() -> Random: + """Return this module's Random instance (per-thread singleton). + + Each thread gets its own generator on first use, seeded once from the + OS, then deterministic: successive calls never repeat an output block. + No locking is involved. See aeg.random.Random for the API. """ - return Random(sys.modules[__name__]) + rng = getattr(_random_tls, "rng", None) + if rng is None: + rng = _random_tls.rng = Random(sys.modules[__name__]) + return rng def random_key() -> bytearray: """ - Generate a secret key using cryptographically secure random bytes. + Generate a secret key using the module's CSPRNG (see random()). It is recommended to wipe() the key after no longer needed. """ - return bytearray(secrets.token_bytes(KEYBYTES)) + return random().bytes(KEYBYTES) def random_nonce() -> bytearray: """ - Generate a public nonce using cryptographically secure random bytes. + Generate a public nonce using the module's CSPRNG (see random()). Nonces (a number used once) are public data that may be sent together with the ciphertext, but they need to be unique for each use. See also: nonce_increment() can be used to derive sequential nonces. """ - return bytearray(secrets.token_bytes(NONCEBYTES)) + return random().bytes(NONCEBYTES) def _ptr(buf): diff --git a/src/aeg/random.py b/src/aeg/random.py index 62605e1..1d9e49f 100644 --- a/src/aeg/random.py +++ b/src/aeg/random.py @@ -31,8 +31,9 @@ class Random: Each call produces keystream under the current key/nonce and increments the nonce afterwards, so no output block is ever produced twice. - The primary way to create one is the cipher module's own random() - helper, e.g. ``aeg.cipher("AEGIS-128X2").random()``. + The primary way to get one is the cipher module's own random() helper, + e.g. ``aeg.cipher("AEGIS-128X2").random()``, which returns a per-thread + singleton (each thread gets its own OS-seeded generator; no locking). """ def __init__(self, cipher, *, seed: Buffer | None = None): diff --git a/tests/test_random.py b/tests/test_random.py new file mode 100644 index 0000000..ec1f5ba --- /dev/null +++ b/tests/test_random.py @@ -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