diff --git a/CHANGELOG.md b/CHANGELOG.md index 94c861a..73f1127 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,13 @@ This repository continues the `overflowdigital` `cuid2` package (`b02c8ca`); ver ## [Unreleased] +### Changed + +- Faster ID generation and cold-start imports via single-draw `getrandbits` MSB fixed-point entropy and letter generation (`< 2^-64` bias), leading-digit base36 SHA3-512 hash encoding, no per-instance lock for the default `Cuid()` and `create_id()` generator (its `random.SystemRandom` and counter are already thread-safe), reduced reference-count contention on free-threaded Python (`3.14t` and `3.15t`), removed `inspect`, `string`, and `secrets` imports, and a deferred `socket` import. +- `random_generator` is typed as `random.Random` instead of `_random.Random`, and random bits are drawn with its `.getrandbits(k)`. Generators without `.getrandbits(k)`, or whose class defines `.random()` but not `.getrandbits(k)` (such as `random.Random` subclasses following the custom-generator pattern in the `random` module documentation), keep one `.random()` call per character and produce the same IDs as in 2.2.0. + +**Upgrade note:** IDs from a seeded generator that provides `.getrandbits(k)` (for example `functools.partial(random.Random, 42)`) differ from 2.2.0 for the same seed, so update any stored expected IDs in tests. Test doubles that create attributes on demand (such as `unittest.mock.Mock` or `MagicMock`) and stub only `.random` must stub `.getrandbits` instead. + ## [2.2.0] - 2026-10-30 ### Added diff --git a/README.md b/README.md index d63ad83..d7f1629 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ by [Parallel Drive](https://github.com/paralleldrive) to Python 3. ## What is CUID2? -* Secure: It's not feasible to guess the next ID (unguessability derives from the `secrets.SystemRandom` CSPRNG salt; per [RFC 9562 §8](https://www.rfc-editor.org/rfc/rfc9562.html#section-8), IDs are identifiers, not capability tokens or secrets). +* Secure: It's not feasible to guess the next ID (unguessability derives from the `random.SystemRandom` CSPRNG salt; per [RFC 9562 §8](https://www.rfc-editor.org/rfc/rfc9562.html#section-8), IDs are identifiers, not capability tokens or secrets). * Collision resistant: It's extremely unlikely to generate the same ID twice. * Horizontally scalable: Generate IDs on multiple machines without coordination. * Offline-compatible: Generate IDs without a network connection. @@ -41,7 +41,7 @@ For more information on the improvements of CUID2 over CUID, see the * **123.6-Bit Default Keyspace (`length=24`)**: 1 leading lowercase letter ($26$) followed by 23 base36 characters ($36^{23}$) yields $26 \times 36^{23} \approx \mathbf{1.61 \times 10^{37}}$ possible IDs ($\approx \mathbf{123.6\text{ bits}}$ of entropy)—slightly more than UUIDv4's 122 random bits in a 33% shorter string (24 vs. 36 chars). Reaching a 50% birthday-bound collision probability requires $\sqrt{2 \ln(2) \times 26 \times 36^{23}} \approx \mathbf{4.7 \times 10^{18}}$ IDs (generating **149 million IDs/sec continuously for 1,000 years**). * **Up to 165.0 Bits (`BIG_LENGTH = 32`)**: At `length=32`, the keyspace expands to $26 \times 36^{31} \approx \mathbf{4.54 \times 10^{49}}$ IDs ($\approx \mathbf{165.0\text{ bits}}$; 50% collision bound at $\approx 7.9 \times 10^{24}$ IDs). * **Histogram De-Biasing**: Because $2^{512} \approx 1.34 \times 10^{154}$ is not an exact power of 36, the most-significant base36 digit of a 512-bit SHA3-512 digest is biased toward low digits; CUID2 drops that leading hash character (`[1:]`) and prepends an independent uniform `[a-z]` letter so every character position has a uniform distribution. -* **Thread, Pre-Fork, & Container Snapshot Resilience**: `Cuid`, `create_id`, `cuid_wrapper`, and `create_counter` are thread-safe (including on free-threaded Python, tested on `3.14t` and `3.15t`). Each generator records the process ID (`os.getpid()`) that built its state; the first ID generated in a forked child process (`multiprocessing`, Gunicorn, Celery prefork, and uWSGI workers, including uWSGI's default mode, which skips Python's `os.register_at_fork` hooks) rebuilds that state with a fresh lock, RNG state, random initial counter (`INITIAL_COUNT_MAX = 476,782,367`, giving $\approx 25,700$ concurrent hosts before a 50% chance of two hosts starting at the same counter value), and child-PID fingerprints (`os.getpid()`, `socket.gethostname()`, and environment variable keys). For cloned VM/container snapshots where no `fork()` syscall occurs, per-call `secrets.SystemRandom` (`os.urandom`) salt and `time.time_ns()` provide defense-in-depth against state collisions. +* **Thread, Pre-Fork, & Container Snapshot Resilience**: `Cuid`, `create_id`, `cuid_wrapper`, and `create_counter` are thread-safe (including on free-threaded Python, tested on `3.14t` and `3.15t`). Each generator records the process ID (`os.getpid()`) that built its state; the first ID generated in a forked child process (`multiprocessing`, Gunicorn, Celery prefork, and uWSGI workers, including uWSGI's default mode, which skips Python's `os.register_at_fork` hooks) rebuilds that state with a fresh RNG state, lock (when `random_generator` returns something other than a `random.SystemRandom` instance or `counter` is not the default; the default `random.SystemRandom` and counter are already thread-safe), random initial counter (`INITIAL_COUNT_MAX = 476,782,367`, giving $\approx 25,700$ concurrent hosts before a 50% chance of two hosts starting at the same counter value), and child-PID fingerprints (`os.getpid()`, `socket.gethostname()`, and environment variable keys). For cloned VM/container snapshots where no `fork()` syscall occurs, per-call `random.SystemRandom` (`os.urandom`) salt and `time.time_ns()` provide defense-in-depth against state collisions. * **Polyglot ORM Compatibility (Prisma & Drizzle)**: Generates 24-character base36 primary keys compatible with TypeScript/Node.js services using Prisma (`@default(cuid(2))` / `@onCreate(cuid(2))`) or Drizzle (`@paralleldrive/cuid2`), and replaces the deprecated CUIDv1 (`cuid`) format. | Property | CUID2 (`cuid2`) | UUIDv7 (`uuid.uuid7`) | UUIDv4 (`uuid.uuid4`) | ULID (`python-ulid`) | NanoID (`nanoid`) | diff --git a/local/tests/test_cli.py b/local/tests/test_cli.py index 08a5c00..e82d69f 100644 --- a/local/tests/test_cli.py +++ b/local/tests/test_cli.py @@ -224,7 +224,7 @@ def test_cli_custom_fingerprint( capsys: pytest.CaptureFixture[str], ) -> None: """Tests that -f/--fingerprint is hashed verbatim (like upstream) and rejects empty strings.""" - hash_spy = mocker.spy(cuid2.utils, "create_hash") + hash_spy = mocker.spy(cuid2.utils, "_create_hash_slice") cli_main(["-f", "host-a", "-n", "2"]) lines = capsys.readouterr().out.splitlines() assert len(lines) == 2 diff --git a/local/tests/test_cuid.py b/local/tests/test_cuid.py index 98baa59..ad0663f 100644 --- a/local/tests/test_cuid.py +++ b/local/tests/test_cuid.py @@ -5,12 +5,14 @@ import os import pickle import random +import re +import subprocess +import sys import threading import time import warnings from functools import partial -from math import floor -from secrets import SystemRandom +from random import SystemRandom from timeit import repeat from typing import TYPE_CHECKING @@ -27,6 +29,7 @@ create_id, cuid_wrapper, is_cuid, + utils, ) if TYPE_CHECKING: @@ -47,8 +50,31 @@ def test_generate_performance(self: TestCuid) -> None: repeat=10, number=50_000, ) - # 50k iterations of cuid.generate() should take less than 2 seconds - assert min(timing_result) < 2 + # 50k iterations of cuid.generate() should take less than 0.7 seconds + assert min(timing_result) < 0.7 + + # Tests that a cold `import cuid2` stays fast (best of 5 fresh interpreters). + def test_import_time(self: TestCuid) -> None: + pattern = re.compile(r"^import time:\s+\d+ \|\s+(\d+) \| cuid2$", re.MULTILINE) + cumulative_us: list[int] = [] + for _ in range(5): + stderr = subprocess.run( + [sys.executable, "-I", "-X", "importtime", "-c", "import cuid2"], + capture_output=True, + text=True, + check=True, + ).stderr + match = pattern.search(stderr) + assert match is not None, stderr + cumulative_us.append(int(match.group(1))) + # A cold `import cuid2` (including its standard library imports) should take less than 10 ms locally. Shared + # CI runners (CI=true) are slower and noisier: 25 ms on Linux, or 45 ms on macOS and Windows, whose runners + # load modules up to about twice as slowly. + if os.environ.get("CI", "").lower() == "true": + limit_us = 25_000 if sys.platform.startswith("linux") else 45_000 + else: + limit_us = 10_000 + assert min(cumulative_us) < limit_us # Tests that generate() returns a string of length DEFAULT_LENGTH. def test_generate_default_length(self: TestCuid) -> None: @@ -133,61 +159,17 @@ def test_generate_unique_ids(self: TestCuid) -> None: cuid = Cuid() assert cuid.generate() != cuid.generate() - # Tests that generate() works when random_generator returns same values. - def test_generate_same_random_values(self: TestCuid, mocker: Mock) -> None: - mock_random_class = mocker.Mock() - mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 - mock_random_class.return_value = mock_random - - mocker.patch("cuid2.utils.create_entropy", return_value="mocked_entropy") - mocker.patch("cuid2.utils.create_letter", return_value="l") - mocker.patch("cuid2.utils.create_hash", return_value="mocked_hash") - - # Initialize Cuid with mocked random generator - cuid = Cuid(random_generator=mock_random_class) - - # Generate two Cuids and assert they are the same - cuid1 = cuid.generate() - cuid2 = cuid.generate() - assert cuid1 == cuid2 - # Tests that the constructor sets default values if not provided. def test_constructor_default_values(self: TestCuid) -> None: # Initialize Cuid with default values cuid = Cuid() # Assert that the values are set to their default values - assert isinstance(cuid._state.random, SystemRandom) + assert type(cuid._state.random) is SystemRandom assert callable(cuid._state.counter) assert cuid._length == DEFAULT_LENGTH assert isinstance(cuid._state.fingerprint, str) - # Tests that the random_generator and fingerprint functions are called correctly when mocked. - def test_mock_random_generator_and_fingerprint(self: TestCuid, mocker: Mock) -> None: - mock_random_class = mocker.Mock() - mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 - mock_random_class.return_value = mock_random - - mock_fingerprint = mocker.Mock() - mock_fingerprint.return_value = "mocked_fingerprint" - - mocker.patch("cuid2.utils.create_entropy", return_value="mocked_entropy") - mocker.patch("cuid2.utils.create_letter", return_value="l") - mocker.patch("cuid2.utils.create_hash", return_value="mocked_hash") - - # Initialize Cuid with mocked random generator and fingerprint functions - cuid = Cuid(random_generator=mock_random_class, fingerprint=mock_fingerprint) - - # Generate a Cuid and assert that the mocked functions were called correctly - cuid_str = cuid.generate() - assert cuid_str == "locked_hash" # first letter is "l" from mocked create_letter function - assert cuid._state.fingerprint == "mocked_fingerprint" - assert ( - cuid._state.counter() == 238391185 - ) # counter is statically assigned since mocked random always returns the same value - # Tests that generate() returns a string of specified length. def test_generate_specified_length(self: TestCuid) -> None: cuid = Cuid(length=10) @@ -199,7 +181,7 @@ def test_constructor_initializes_members(self: TestCuid, mocker: Mock) -> None: # Arrange mock_random_class = mocker.Mock() mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) mock_random_class.return_value = mock_random mock_counter = mocker.Mock() mock_fingerprint = mocker.Mock() @@ -209,7 +191,8 @@ def test_constructor_initializes_members(self: TestCuid, mocker: Mock) -> None: # Assert assert cuid._state.random == mock_random_class() - assert cuid._state.counter == mock_counter(floor(cuid._state.random.random() * INITIAL_COUNT_MAX)) + mock_counter.assert_called_once_with(INITIAL_COUNT_MAX // 2) + assert cuid._state.counter == mock_counter.return_value assert cuid._length == DEFAULT_LENGTH assert cuid._state.fingerprint == mock_fingerprint(random_generator=cuid._state.random) @@ -218,13 +201,12 @@ def test_generate_uses_base36_encoding(self: TestCuid, mocker: Mock) -> None: # Arrange mock_random_class = mocker.Mock() mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) mock_random_class.return_value = mock_random mock_counter = mocker.Mock() mock_counter.return_value = lambda: 123456789 mock_fingerprint = mocker.Mock(return_value="abcdefg") mocker.patch("time.time_ns", return_value=1627584000000000000) # 2021-07-30 00:00:00 UTC - mocker.patch("cuid2.utils.create_letter", return_value="l") cuid = Cuid(random_generator=mock_random_class, counter=mock_counter, fingerprint=mock_fingerprint) @@ -232,9 +214,10 @@ def test_generate_uses_base36_encoding(self: TestCuid, mocker: Mock) -> None: result = cuid.generate() # Assert - assert result.startswith("l") + assert result.startswith("n") assert len(result) == DEFAULT_LENGTH - assert result == "l9j3ikop1bi8tcvzme3x3yv7" + assert result == "n8xjuve9pq2189v81u8xb7rq" + assert cuid.generate() == result def test_generate_thread_safety_custom_random(self: TestCuid) -> None: """Tests that Cuid.generate() serializes calls to a non-reentrant custom random generator across threads.""" @@ -244,14 +227,14 @@ def __init__(self: _NonReentrantRandom, seed: int) -> None: super().__init__(seed) self._busy = False - def random(self: _NonReentrantRandom) -> float: + def getrandbits(self: _NonReentrantRandom, k: int) -> int: if self._busy: msg = "Concurrent access to non-reentrant random generator" raise RuntimeError(msg) self._busy = True try: time.sleep(0) - return super().random() + return super().getrandbits(k) finally: self._busy = False @@ -291,11 +274,11 @@ def __init__(self: _ReentrantRandom, seed: int) -> None: self.entered = False self.nested: list[str] = [] - def random(self: _ReentrantRandom) -> float: + def getrandbits(self: _ReentrantRandom, k: int) -> int: if self.cuid is not None and not self.entered: self.entered = True self.nested.append(self.cuid.generate()) - return super().random() + return super().getrandbits(k) rng = _ReentrantRandom(1) cuid = Cuid(random_generator=lambda: rng) @@ -312,6 +295,87 @@ def random(self: _ReentrantRandom) -> float: assert is_cuid(rng.nested[0]) assert outer[0] != rng.nested[0] + def test_module_level_random_seed_does_not_affect_default_cuid( + self: TestCuid, + mocker: Mock, + ) -> None: + mocker.patch("time.time_ns", return_value=1627584000000000000) + + module_state = random.getstate() + try: + # Positive control: an instance wired to module-level `random` collides when `random.seed()` is reset. + random.seed(12345) + vulnerable_1 = Cuid(random_generator=lambda: random).generate() # type: ignore[arg-type, return-value] + random.seed(12345) + vulnerable_2 = Cuid(random_generator=lambda: random).generate() # type: ignore[arg-type, return-value] + assert vulnerable_1 == vulnerable_2 + + # Default Cuid() uses random.SystemRandom (os.urandom) and is unaffected by random.seed(). + random.seed(12345) + cuid1 = Cuid() + random.seed(12345) + cuid2 = Cuid() + assert cuid1._state.fingerprint != cuid2._state.fingerprint + assert cuid1.generate() != cuid2.generate() + finally: + random.setstate(module_state) + + def test_default_generator_takes_no_instance_lock(self: TestCuid) -> None: + """Tests that the thread-safe defaults (SystemRandom, create_counter) skip the per-instance lock.""" + assert Cuid()._state.lock is None + create_id() + default_generator = cuid2.generator._default_generator + assert default_generator is not None + assert default_generator._state.lock is None + + @pytest.mark.parametrize( + "kwargs", + [ + {"random_generator": random.Random}, + {"random_generator": type("_SystemRandomSubclass", (SystemRandom,), {})}, + {"counter": partial(utils.create_counter)}, + ], + ids=["custom-random", "systemrandom-subclass", "custom-counter"], + ) + def test_custom_generator_or_counter_keeps_instance_lock(self: TestCuid, kwargs: dict[str, Any]) -> None: + """Tests that any non-default random generator or counter keeps the re-entrant instance lock.""" + lock = Cuid(**kwargs)._state.lock + assert lock is not None + assert type(lock) is type(threading.RLock()) + + def test_default_generator_concurrent_ids_are_unique(self: TestCuid) -> None: + """Tests that a shared default Cuid() draws distinct, consecutive counter values across threads lock-free.""" + cuid = Cuid() + assert cuid._state.lock is None + real_counter = cuid._state.counter + counts: list[int] = [] + + def recording_counter() -> int: + value = real_counter() + counts.append(value) + return value + + cuid._state = cuid._state._replace(counter=recording_counter) + num_threads = 8 + ids_per_thread = 250 + barrier = threading.Barrier(num_threads) + buckets: list[list[str]] = [[] for _ in range(num_threads)] + + def worker(index: int) -> None: + barrier.wait() + buckets[index] = [cuid.generate() for _ in range(ids_per_thread)] + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(num_threads)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + generated = [cid for bucket in buckets for cid in bucket] + assert len(generated) == num_threads * ids_per_thread + assert len(set(generated)) == len(generated) + assert sorted(counts) == list(range(min(counts), min(counts) + len(generated))) + def test_deepcopy_returns_independent_generator(self: TestCuid) -> None: """Tests that deepcopy builds fresh state instead of cloning the lock, RNG stream, or counter.""" cuid = Cuid(random_generator=random.Random, length=10) @@ -689,9 +753,11 @@ def test_create_id_invalid_length( invalid_length: int, expected_match: str, ) -> None: - """Tests that create_id() raises ValueError when length is out of bounds.""" + """Tests that create_id() raises ValueError when length is out of bounds, before building the default.""" + assert cuid2.generator._default_generator is None with pytest.raises(ValueError, match=expected_match): create_id(length=invalid_length) + assert cuid2.generator._default_generator is None def test_create_id_concurrent_first_call_singleton(self: TestCreateId, mocker: Mock) -> None: """Tests that concurrent first-time create_id() calls initialize _default_generator once.""" @@ -853,7 +919,7 @@ def _fork_child_worker(custom_cuid: Cuid, conn: Connection[Any, Any]) -> None: class TestForkSafety: def test_generate_rebuilds_state_when_pid_changes(self: TestForkSafety, mocker: Mock) -> None: """Tests that generate() replaces the lock, RNG, counter, and fingerprint once per new process ID.""" - cuid = Cuid() + cuid = Cuid(random_generator=random.Random) old_state = cuid._state assert old_state.pid == os.getpid() @@ -869,6 +935,25 @@ def test_generate_rebuilds_state_when_pid_changes(self: TestForkSafety, mocker: assert is_cuid(cuid.generate()) assert cuid._state is new_state + def test_default_generator_rebuilds_lock_free_state_when_pid_changes( + self: TestForkSafety, + mocker: Mock, + ) -> None: + """Tests that the default Cuid() rebuilds its RNG, counter, and fingerprint per process with no lock.""" + cuid = Cuid() + old_state = cuid._state + assert old_state.lock is None + + mocker.patch("os.getpid", return_value=old_state.pid + 1) + assert is_cuid(cuid.generate()) + + new_state = cuid._state + assert new_state.pid == old_state.pid + 1 + assert new_state.lock is None + assert new_state.random is not old_state.random + assert new_state.counter is not old_state.counter + assert new_state.fingerprint != old_state.fingerprint + def test_pid_change_reseeds_shared_random_and_tolerates_seedless_rng( self: TestForkSafety, mocker: Mock, @@ -879,13 +964,17 @@ def test_pid_change_reseeds_shared_random_and_tolerates_seedless_rng( unreseeded_rng = random.Random() # noqa: S311 unreseeded_rng.setstate(shared_rng.getstate()) # Draws made by the rebuild (counter + fingerprint entropy) and by one generate() (letter + salt). - for _ in range(1 + BIG_LENGTH + 1 + DEFAULT_LENGTH): - _ = unreseeded_rng.random() + for draw_bits in ( + utils._COUNTER_DRAW_BITS, + (36**BIG_LENGTH).bit_length() + 64, + (26 * 36**DEFAULT_LENGTH).bit_length() + 64, + ): + _ = unreseeded_rng.getrandbits(draw_bits) seed_spy = mocker.spy(shared_rng, "seed") class _SeedlessRng: - def random(self: _SeedlessRng) -> float: - return 0.25 + def getrandbits(self: _SeedlessRng, k: int) -> int: + return 1 << (k - 2) seedless = _SeedlessRng() cuid_seedless = Cuid(random_generator=lambda: seedless) # type: ignore[arg-type, return-value] @@ -898,7 +987,7 @@ def random(self: _SeedlessRng) -> float: assert cuid_shared._state.random is shared_rng assert shared_rng.getstate() != unreseeded_rng.getstate() assert cuid_seedless._state.pid == os.getpid() - assert cuid_seedless._state.random.random() == 0.25 + assert cuid_seedless._state.random.getrandbits(4) == 4 def test_rebuild_failure_propagates_without_affecting_other_instances( self: TestForkSafety, @@ -936,8 +1025,9 @@ def test_pid_change_never_acquires_inherited_locks( ) -> None: """Tests that a new process never waits on instance or default-generator locks held at fork time.""" monkeypatch.setattr(cuid2.generator, "_default_generator", None) - cuid = Cuid() + cuid = Cuid(random_generator=random.Random) inherited_instance_lock = cuid._state.lock + assert inherited_instance_lock is not None inherited_process_lock = cuid2.generator._process_lock(os.getpid()) child_pid = os.getpid() + 1 holding = threading.Event() @@ -1035,12 +1125,14 @@ def test_posix_fork_children_reinitialize_despite_held_locks( shared_rng = random.Random(12345) # noqa: S311 custom_cuid = Cuid(random_generator=lambda: shared_rng) parent_fingerprint = custom_cuid._state.fingerprint + inherited_instance_lock = custom_cuid._state.lock + assert inherited_instance_lock is not None inherited_process_lock = cuid2.generator._process_lock(os.getpid()) holding = threading.Event() release = threading.Event() def hold_locks() -> None: - with custom_cuid._state.lock, inherited_process_lock: + with inherited_instance_lock, inherited_process_lock: holding.set() release.wait() diff --git a/local/tests/test_fuzz.py b/local/tests/test_fuzz.py index ff7c6ac..8bf10f2 100644 --- a/local/tests/test_fuzz.py +++ b/local/tests/test_fuzz.py @@ -1,10 +1,12 @@ from __future__ import annotations +import math import random from typing import TYPE_CHECKING, Protocol, cast +from unittest.mock import Mock, patch import pytest -from hypothesis import given +from hypothesis import example, given from hypothesis import strategies as st from cuid2 import Cuid, create_id, cuid_wrapper, is_cuid, utils @@ -151,6 +153,68 @@ def test_fuzz_cuid_deprecated_and_invalid_lengths(deprecated_length: int, invali cuid_wrapper(length=invalid_length) +@given( + digest_int=st.integers(min_value=0, max_value=(1 << 512) - 1), + length=st.integers(min_value=1, max_value=MAXIMUM_LENGTH), +) +@example(digest_int=0, length=1) +@example(digest_int=35, length=BIG_LENGTH) +@example(digest_int=36, length=BIG_LENGTH) +@example(digest_int=36**97 - 1, length=MAXIMUM_LENGTH) +@example(digest_int=36**98 - 1, length=MAXIMUM_LENGTH - 1) +@example(digest_int=36**98, length=1) +@example(digest_int=36**98, length=MAXIMUM_LENGTH - 1) +@example(digest_int=36**98, length=MAXIMUM_LENGTH) +@example(digest_int=36**99 - 1, length=MAXIMUM_LENGTH - 1) +@example(digest_int=36**99, length=1) +@example(digest_int=36**99, length=MAXIMUM_LENGTH - 1) +@example(digest_int=36**99, length=MAXIMUM_LENGTH) +@example(digest_int=(1 << 512) - 1, length=MAXIMUM_LENGTH - 1) +@example(digest_int=(1 << 512) - 1, length=MAXIMUM_LENGTH) +def test_fuzz_create_hash_slice_arbitrary_digest(digest_int: int, length: int) -> None: + mock_hash = Mock() + mock_hash.digest.return_value = digest_int.to_bytes(64, byteorder="big") + with patch.object(utils, "sha3_512", return_value=mock_hash): + full_hash = utils.create_hash("fuzz") + assert utils._create_hash_slice("fuzz", 1, length) == full_hash[1:length] + assert utils._create_hash_slice("fuzz", 0, length) == full_hash[:length] + assert utils._create_hash_slice("fuzz", 0, BIG_LENGTH) == full_hash[:BIG_LENGTH] + + +@given( + data=st.text(max_size=200), + length=st.integers(min_value=1, max_value=MAXIMUM_LENGTH), + u53=st.integers(min_value=0, max_value=(1 << 53) - 1), + dyadic_j=st.integers(min_value=0, max_value=(1 << 20) - 1), +) +def test_fuzz_create_hash_slice_and_scale_randbits( + data: str, + length: int, + u53: int, + dyadic_j: int, +) -> None: + assert utils._create_hash_slice(data, 1, length) == utils.create_hash(data)[1:length] + + modulus = BASE36_RADIX**length + draw_bits = modulus.bit_length() + utils._OVERSAMPLE_BITS + assert 1 << draw_bits >= (1 << 64) * modulus + fused_modulus = len(ASCII_LOWERCASE) * modulus + assert 1 << (fused_modulus.bit_length() + utils._OVERSAMPLE_BITS) >= (1 << 64) * fused_modulus + + high_bits = draw_bits - 64 + x_high = u53 & ((1 << high_bits) - 1) + low_zero = utils._scale_randbits(x_high << 64, modulus, draw_bits) + low_ones = utils._scale_randbits((x_high << 64) | ((1 << 64) - 1), modulus, draw_bits) + assert 0 <= low_ones - low_zero <= 1 + + k_count = utils._COUNTER_DRAW_BITS + assert utils._scale_randbits(u53 << (k_count - 53), INITIAL_COUNT_MAX, k_count) == ((u53 * INITIAL_COUNT_MAX) >> 53) + dyadic_u53 = dyadic_j << (53 - 20) + assert utils._scale_randbits(dyadic_u53 << (k_count - 53), INITIAL_COUNT_MAX, k_count) == math.floor( + (dyadic_u53 / 2**53) * INITIAL_COUNT_MAX, + ) + + _FUZZ_TARGETS: tuple[_FuzzTarget, ...] = cast( "tuple[_FuzzTarget, ...]", ( @@ -161,6 +225,8 @@ def test_fuzz_cuid_deprecated_and_invalid_lengths(deprecated_length: int, invali test_fuzz_utils_primitives, test_fuzz_cuid_spec_lengths, test_fuzz_cuid_deprecated_and_invalid_lengths, + test_fuzz_create_hash_slice_arbitrary_digest, + test_fuzz_create_hash_slice_and_scale_randbits, ), ) diff --git a/local/tests/test_utils.py b/local/tests/test_utils.py index a2a78e9..3c97f19 100644 --- a/local/tests/test_utils.py +++ b/local/tests/test_utils.py @@ -1,20 +1,55 @@ from __future__ import annotations +import math import os +import random import string import threading import time +from types import SimpleNamespace from typing import TYPE_CHECKING import pytest -from cuid2 import utils +from cuid2 import Cuid, utils if TYPE_CHECKING: from collections.abc import Callable from unittest.mock import Mock +class _LowBitStuckRandom(random.Random): + """Test PRNG whose lowest 64 bits are permanently stuck at zero.""" + + def getrandbits(self: _LowBitStuckRandom, k: int) -> int: + return super().getrandbits(k) & ~((1 << 64) - 1) + + +class _RandomOnlyRandom(random.Random): + """xorshift64 core generator plugged in the way the `random` docs describe: random() and state, no getrandbits. + + Its overridden seed() never seeds the inherited Mersenne Twister, so the inherited getrandbits() ignores this + generator. + """ + + def seed(self: _RandomOnlyRandom, a: object = None, version: int = 2) -> None: # noqa: ARG002 + self._xorshift = a if isinstance(a, int) and a > 0 else 0x9E3779B97F4A7C15 + + def random(self: _RandomOnlyRandom) -> float: + x = self._xorshift + x ^= (x << 13) & 0xFFFFFFFFFFFFFFFF + x ^= x >> 7 + x ^= (x << 17) & 0xFFFFFFFFFFFFFFFF + self._xorshift = x + return (x >> 11) / 9007199254740992.0 + + def getstate(self: _RandomOnlyRandom) -> tuple[int]: + return (self._xorshift,) + + def setstate(self: _RandomOnlyRandom, state: tuple[int]) -> None: # type: ignore[override] + (self._xorshift,) = state + + class TestBase36Encode: # Tests that the function returns "0" when given the number 0 def test_zero_returns_0(self: TestBase36Encode) -> None: @@ -62,20 +97,29 @@ class TestCreateLetter: # Tests that create_letter function returns a lowercase letter def test_happy_path_create_letter(self: TestCreateLetter, mocker: Mock) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 - assert utils.create_letter(random_generator=mock_random) in string.ascii_lowercase + mock_random.getrandbits.return_value = 1 << (utils._LETTER_DRAW_BITS - 1) + result = utils.create_letter(random_generator=mock_random) + assert result == "n" + assert result in string.ascii_lowercase + mock_random.getrandbits.assert_called_once_with(utils._LETTER_DRAW_BITS) # Tests that create_letter function returns a lowercase letter when random_generator returns 0 def test_edge_case_create_letter_zero(self: TestCreateLetter, mocker: Mock) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0 - assert utils.create_letter(random_generator=mock_random) in string.ascii_lowercase + mock_random.getrandbits.return_value = 0 + result = utils.create_letter(random_generator=mock_random) + assert result == "a" + assert result in string.ascii_lowercase + mock_random.getrandbits.assert_called_once_with(utils._LETTER_DRAW_BITS) - # Tests that create_letter function returns a lowercase letter when random_generator returns 1 + # Tests that create_letter function returns a lowercase letter when random_generator returns max bits def test_edge_case_create_letter_one(self: TestCreateLetter, mocker: Mock) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0.999 - assert utils.create_letter(random_generator=mock_random) in string.ascii_lowercase + mock_random.getrandbits.return_value = (1 << utils._LETTER_DRAW_BITS) - 1 + result = utils.create_letter(random_generator=mock_random) + assert result == "z" + assert result in string.ascii_lowercase + mock_random.getrandbits.assert_called_once_with(utils._LETTER_DRAW_BITS) class TestCreateHash: @@ -136,94 +180,104 @@ class TestCreateEntropy: def test_create_entropy_happy(self: TestCreateEntropy, mocker: Mock) -> None: # Happy path test mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) result = utils.create_entropy(mock_random, length=4) assert len(result) == 4 assert all(c in string.digits + string.ascii_lowercase for c in result) + assert mock_random.getrandbits.call_count == 1 # Tests that create_entropy defaults to length 4 when length is omitted def test_create_entropy_default_length(self: TestCreateEntropy, mocker: Mock) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) result = utils.create_entropy(mock_random) assert len(result) == 4 assert all(c in string.digits + string.ascii_lowercase for c in result) + assert mock_random.getrandbits.call_count == 1 # Tests that create_entropy raises a ValueError when given a valid instance of Random and a length of 0 def test_create_entropy_zero_length(self: TestCreateEntropy, mocker: Mock) -> None: # Edge case test with length = 0 mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) with pytest.raises(ValueError, match=r"Cannot create entropy without a length >= 1\."): utils.create_entropy(mock_random, length=0) + assert mock_random.getrandbits.call_count == 0 # Tests that create_entropy raises a ValueError when given a valid instance of Random and a negative length def test_create_entropy_negative_length(self: TestCreateEntropy, mocker: Mock) -> None: # Edge case test with negative length mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) with pytest.raises(ValueError, match=r"Cannot create entropy without a length >= 1\."): utils.create_entropy(mock_random, length=-1) + assert mock_random.getrandbits.call_count == 0 - # Tests that create_entropy raises a TypeError when given an invalid instance of Random + # Tests that create_entropy raises an AttributeError when given an invalid instance of Random def test_create_entropy_invalid_random_generator(self: TestCreateEntropy) -> None: # Edge case test with invalid random generator with pytest.raises(AttributeError): utils.create_entropy("not a random generator") # type: ignore[arg-type] # Tests that create_entropy returns a string of expected length when given a - # valid instance of Random and a length equal to the maximum possible length (36) + # valid instance of Random and a length of 36 (one character per base36 digit) def test_create_entropy_length_max(self: TestCreateEntropy, mocker: Mock) -> None: - # Happy path test with length = maximum possible length (36) + # Happy path test with length = 36 mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) result = utils.create_entropy(mock_random, length=36) assert len(result) == 36 assert all(c in string.digits + string.ascii_lowercase for c in result) + assert mock_random.getrandbits.call_count == 1 # Tests that create_entropy returns a string of length 1 when given a valid instance of Random and a length of 1 def test_create_entropy_length_one(self: TestCreateEntropy, mocker: Mock) -> None: # Happy path test with length = 1 mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) result = utils.create_entropy(mock_random, length=1) assert len(result) == 1 assert all(c in string.digits + string.ascii_lowercase for c in result) + assert mock_random.getrandbits.call_count == 1 # Tests that create_entropy returns a string of expected length when given a valid instance of - # Random and a length greater than the maximum possible length (36) + # Random and a length greater than len(_POWERS_OF_36) - 1 (100) def test_create_entropy_length_greater_than_max(self: TestCreateEntropy, mocker: Mock) -> None: - # Edge case test with length > maximum possible length (36) + # Edge case test with length > 100 so both arms of length < len(_POWERS_OF_36) are covered mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 - result = utils.create_entropy(mock_random, length=40) - assert len(result) == 40 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) + result = utils.create_entropy(mock_random, length=101) + assert len(result) == 101 assert all(c in string.digits + string.ascii_lowercase for c in result) + assert mock_random.getrandbits.call_count == 1 # Tests that create_entropy returns a string of expected length when given a valid instance of - # Random and a length between 1 and the maximum possible length (36) + # Random and a length of 10 def test_create_entropy_length_between_1_and_max(self: TestCreateEntropy, mocker: Mock) -> None: - # Happy path test with length between 1 and maximum possible length (36) + # Happy path test with length = 10 mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) result = utils.create_entropy(mock_random, length=10) assert len(result) == 10 assert all(c in string.digits + string.ascii_lowercase for c in result) + assert mock_random.getrandbits.call_count == 1 # Tests that create_entropy indexes BASE36_ALPHABET directly without calling base36_encode def test_create_entropy_does_not_call_base36_encode(self: TestCreateEntropy, mocker: Mock) -> None: spy = mocker.spy(utils, "base36_encode") mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + draw_bits = (36**4).bit_length() + 64 + mock_random.getrandbits.return_value = ((int("iiii", 36) << draw_bits) + 36**4 - 1) // 36**4 assert utils.create_entropy(mock_random, length=4) == "iiii" assert spy.call_count == 0 + assert mock_random.getrandbits.call_count == 1 class TestCreateFingerprint: # Tests that create_fingerprint returns a string of length BIG_LENGTH when valid fingerprint data is provided def test_create_fingerprint_with_valid_data(self: TestCreateFingerprint, mocker: Mock) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) fingerprint_data = "test_data" expected_length = utils.BIG_LENGTH @@ -234,7 +288,7 @@ def test_create_fingerprint_with_valid_data(self: TestCreateFingerprint, mocker: # Tests that create_fingerprint generates valid fingerprint data when no fingerprint data is provided def test_create_fingerprint_with_empty_fingerprint_data(self: TestCreateFingerprint, mocker: Mock) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) expected_length = utils.BIG_LENGTH result = utils.create_fingerprint(random_generator=mock_random) @@ -244,7 +298,7 @@ def test_create_fingerprint_with_empty_fingerprint_data(self: TestCreateFingerpr # Tests that create_fingerprint generates valid fingerprint data when the hostname is empty def test_create_fingerprint_with_empty_hostname(self: TestCreateFingerprint, mocker: Mock) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) mocker.patch("socket.gethostname", return_value="") expected_length = utils.BIG_LENGTH @@ -255,7 +309,7 @@ def test_create_fingerprint_with_empty_hostname(self: TestCreateFingerprint, moc # Tests that create_fingerprint generates valid fingerprint data when the env variables are empty def test_create_fingerprint_with_empty_env_variables(self: TestCreateFingerprint, mocker: Mock) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) mocker.patch.dict(os.environ, {}) expected_length = utils.BIG_LENGTH @@ -266,7 +320,8 @@ def test_create_fingerprint_with_empty_env_variables(self: TestCreateFingerprint # Tests that create_fingerprint ignores invalid data types are provided as input def test_create_fingerprint_with_invalid_data_types(self: TestCreateFingerprint, mocker: Mock) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + draw_bits = (36**32).bit_length() + 64 + mock_random.getrandbits.return_value = ((int("i" * 32, 36) << draw_bits) + 36**32 - 1) // 36**32 non_string_data: int = 123 fingerprint = utils.create_fingerprint( random_generator=mock_random, @@ -297,7 +352,7 @@ def test_create_fingerprint_with_empty_hostname_and_empty_env_variables( mocker: Mock, ) -> None: mock_random = mocker.Mock() - mock_random.random.return_value = 0.5 + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) # Mock os.getpid() and socket.gethostname() to return empty strings mocker.patch("os.getpid", return_value="") mocker.patch("socket.gethostname", return_value="") @@ -446,3 +501,156 @@ def __add__(self: _ReentrantInt, other: int) -> _ReentrantInt: assert not thread.is_alive(), "re-entrant counter call deadlocked on the counter lock" assert len(outer) == 1 assert nested[0] >= 1 + + +class TestFixedPointScalingAndHashSlice: + @pytest.mark.parametrize("modulus", [26, 36, utils.INITIAL_COUNT_MAX, 36**24, 26 * 36**24]) + def test_scale_randbits_endpoints_and_midpoint( + self: TestFixedPointScalingAndHashSlice, + modulus: int, + ) -> None: + k = modulus.bit_length() + utils._OVERSAMPLE_BITS + assert 1 << k >= (1 << 64) * modulus + assert utils._scale_randbits(0, modulus, k) == 0 + assert utils._scale_randbits(1 << (k - 1), modulus, k) == modulus // 2 + assert utils._scale_randbits((1 << k) - 1, modulus, k) == modulus - 1 + + def test_decode_letter_and_entropy_endpoints(self: TestFixedPointScalingAndHashSlice) -> None: + k_fused = (len(utils.ASCII_LOWERCASE) * 36**24).bit_length() + utils._OVERSAMPLE_BITS + + def decode_draw(bits: int) -> tuple[str, str]: + return utils._letter_and_entropy_via_getrandbits(lambda k: bits if k == k_fused else -1, 24) + + assert decode_draw(0) == ("a", "0" * 24) + assert decode_draw(1 << (k_fused - 1))[0] == "n" + assert decode_draw((1 << k_fused) - 1) == ("z", "z" * 24) + + def test_low_bit_stuck_random_immunity(self: TestFixedPointScalingAndHashSlice) -> None: + rng = _LowBitStuckRandom(42) + assert rng.getrandbits(utils._LETTER_DRAW_BITS) % 4 == 0 + + assert {utils.create_letter(rng) for _ in range(500)} == set(utils.ASCII_LOWERCASE) + + rng_entropy = _LowBitStuckRandom(42) + entropy_draws = [utils.create_entropy(rng_entropy, 24) for _ in range(500)] + for position in range(24): + assert {draw[position] for draw in entropy_draws} == set(utils.BASE36_ALPHABET), position + + rng_counter = _LowBitStuckRandom(42) + assert {utils._create_initial_count(rng_counter) % 2 for _ in range(500)} == {0, 1} + + cuid = Cuid(random_generator=lambda: _LowBitStuckRandom(42)) + assert {cuid.generate()[0] for _ in range(500)} == set(utils.ASCII_LOWERCASE) + + @pytest.mark.parametrize("boundary_int", [0, 36**98 - 1, 36**98, 36**99, (1 << 512) - 1]) + @pytest.mark.parametrize("length", [1, utils.BIG_LENGTH, utils.MAXIMUM_LENGTH]) + def test_create_hash_slice_branches( + self: TestFixedPointScalingAndHashSlice, + mocker: Mock, + boundary_int: int, + length: int, + ) -> None: + mock_hash = mocker.Mock() + mock_hash.digest.return_value = boundary_int.to_bytes(64, byteorder="big") + mocker.patch("cuid2.utils.sha3_512", return_value=mock_hash) + + full_hash = utils.create_hash("boundary") + assert utils._create_hash_slice("boundary", 0, length) == full_hash[:length] + assert utils._create_hash_slice("boundary", 1, length) == full_hash[1:length] + + +class _LowResolutionRandom(random.Random): + """random()-only generator whose floats have 8 bits of resolution (a valid, if coarse, random() contract).""" + + def random(self: _LowResolutionRandom) -> float: + return math.floor(super().random() * 256) / 256 + + +def _per_character_letter(draw: Callable[[], float]) -> str: + """Reference letter draw: one floor(random() * 26) call.""" + return utils.ASCII_LOWERCASE[math.floor(draw() * len(utils.ASCII_LOWERCASE))] + + +def _per_character_salt(draw: Callable[[], float], length: int) -> str: + """Reference salt draw: one floor(random() * 36) call per character, in order.""" + return "".join(utils.BASE36_ALPHABET[math.floor(draw() * utils.BASE36_RADIX)] for _ in range(length)) + + +class TestRandomOnlyGenerators: + def test_uses_random_only_follows_nearest_definition(self: TestRandomOnlyGenerators, mocker: Mock) -> None: + class _InheritsRandomOnly(_RandomOnlyRandom): + pass + + class _AddsGetrandbits(_RandomOnlyRandom): + def getrandbits(self: _AddsGetrandbits, k: int) -> int: + return k + + assert utils._uses_random_only(_RandomOnlyRandom()) + assert utils._uses_random_only(_InheritsRandomOnly()) + duck_typed = SimpleNamespace(random=random.Random(1).random) # noqa: S311 + assert utils._uses_random_only(duck_typed) # type: ignore[arg-type] + assert utils._uses_random_only(mocker.Mock(spec=["random", "seed"])) + assert not utils._uses_random_only(_AddsGetrandbits()) + assert not utils._uses_random_only(random.Random()) # noqa: S311 + assert not utils._uses_random_only(random.SystemRandom()) + assert not utils._uses_random_only(_LowBitStuckRandom()) + assert not utils._uses_random_only(mocker.Mock()) + + def test_mock_with_getrandbits_never_calls_random(self: TestRandomOnlyGenerators, mocker: Mock) -> None: + mock_random = mocker.Mock() + mock_random.getrandbits.side_effect = lambda k: 1 << (k - 1) + assert len(Cuid(random_generator=lambda: mock_random).generate()) == utils.DEFAULT_LENGTH + mock_random.random.assert_not_called() + + @pytest.mark.parametrize("rng_type", [_RandomOnlyRandom, _LowResolutionRandom]) + @pytest.mark.parametrize("length", [1, 4, 24, utils.MAXIMUM_LENGTH]) + def test_random_only_draws_match_per_character_reference( + self: TestRandomOnlyGenerators, + rng_type: type[random.Random], + length: int, + ) -> None: + for seed in range(1, 51): + reference = rng_type(seed).random + expected = (_per_character_letter(reference), _per_character_salt(reference, length)) + assert utils._letter_and_entropy_via_random(rng_type(seed), length) == expected + + assert utils.create_letter(rng_type(seed)) == _per_character_letter(rng_type(seed).random) + assert utils.create_entropy(rng_type(seed), length) == _per_character_salt(rng_type(seed).random, length) + expected_count = math.floor(rng_type(seed).random() * utils.INITIAL_COUNT_MAX) + assert utils._create_initial_count(rng_type(seed)) == expected_count + + def test_out_of_range_random_fails_or_wraps_per_character(self: TestRandomOnlyGenerators, mocker: Mock) -> None: + # 1.0 and negative random() values index the alphabet per character: IndexError, or a wrap within that + # character only (never a carry or borrow into neighbouring characters). + rng = mocker.Mock(spec=["random"]) + rng.random.side_effect = [0.5, 0.5, 1.0, 0.1, 0.2] # initial counter, letter, then the salt characters + cuid = Cuid(random_generator=lambda: rng, length=4, fingerprint=lambda random_generator: "fp") # noqa: ARG005 + with pytest.raises(IndexError): + cuid.generate() + rng.random.side_effect = [0.0, -0.01, 0.2, 0.3] + assert utils._letter_and_entropy_via_random(rng, 3) == ("a", "z7a") + + def test_low_resolution_random_keeps_full_salt_entropy(self: TestRandomOnlyGenerators) -> None: + rng = _LowResolutionRandom(3) + salts = [utils.create_entropy(rng, 24) for _ in range(2000)] + assert len(set(salts)) == len(salts) + for position in range(24): + assert {salt[position] for salt in salts} == set(utils.BASE36_ALPHABET), position + + def test_cuid_with_random_only_generator_matches_per_character_reference( + self: TestRandomOnlyGenerators, + mocker: Mock, + ) -> None: + time_ns = 1627584000000000000 + mocker.patch("time.time_ns", return_value=time_ns) + cuid = Cuid(random_generator=lambda: _RandomOnlyRandom(7), fingerprint=lambda random_generator: "fp") # noqa: ARG005 + + reference = _RandomOnlyRandom(7).random + start_count = math.floor(reference() * utils.INITIAL_COUNT_MAX) + letter = _per_character_letter(reference) + salt = _per_character_salt(reference, 24) + hash_input = utils.base36_encode(time_ns) + salt + utils.base36_encode(start_count + 1) + "fp" + assert cuid.generate() == letter + utils.create_hash(hash_input)[1:24] + + varied = Cuid(random_generator=_RandomOnlyRandom) + assert {varied.generate()[0] for _ in range(500)} == set(utils.ASCII_LOWERCASE) diff --git a/src/cuid2/cli.py b/src/cuid2/cli.py index 7353e22..bb9e8b1 100644 --- a/src/cuid2/cli.py +++ b/src/cuid2/cli.py @@ -15,9 +15,9 @@ __all__ = ["main"] if TYPE_CHECKING: - from _random import Random from argparse import Namespace from collections.abc import Iterable, Iterator, Sequence + from random import Random _SLUG_LENGTH: int = 5 diff --git a/src/cuid2/generator.py b/src/cuid2/generator.py index e1d9dbb..02a8352 100644 --- a/src/cuid2/generator.py +++ b/src/cuid2/generator.py @@ -2,7 +2,6 @@ from __future__ import annotations -import inspect import os import re import threading @@ -11,9 +10,8 @@ from contextlib import suppress from copy import copy, deepcopy from functools import partial -from math import floor -from secrets import SystemRandom -from typing import TYPE_CHECKING, ClassVar, Final, NamedTuple, Protocol +from random import SystemRandom # secrets.SystemRandom re-exports random.SystemRandom (os.urandom CSPRNG) +from typing import TYPE_CHECKING, NamedTuple from cuid2 import utils from cuid2.utils import BIG_LENGTH, DEFAULT_LENGTH, INITIAL_COUNT_MAX, MAXIMUM_LENGTH, MIN_LENGTH @@ -31,8 +29,9 @@ ] if TYPE_CHECKING: - from _random import Random from collections.abc import Callable + from random import Random + from typing import ClassVar, Final, Protocol class FingerprintCallable(Protocol): """Protocol for fingerprint generator callables.""" @@ -82,16 +81,11 @@ def _validate_length(length: int) -> None: msg = "Length must never exceed 98 characters." raise ValueError(msg) if length < MIN_LENGTH or length > BIG_LENGTH: - stacklevel: int = 1 - frame = inspect.currentframe() - while frame is not None and frame.f_code.co_filename == __file__: - stacklevel += 1 - frame = frame.f_back warnings.warn( f"Length {length} outside the CUID2 specification range ({MIN_LENGTH}..{BIG_LENGTH}) " "is deprecated and will raise ValueError in v3.0.0.", DeprecationWarning, - stacklevel=stacklevel, + stacklevel=3, ) @@ -128,8 +122,9 @@ class _GeneratorState(NamedTuple): """Per-process generator state, replaced as a whole when a forked child process first generates an ID.""" pid: int - lock: threading.RLock + lock: threading.RLock | None random: Random + draw: Callable[[int], tuple[str, str]] counter: Callable[[], int] fingerprint: str @@ -165,8 +160,12 @@ def __init__( Parameters ---------- random_generator : Callable[[], "Random"], default=SystemRandom - Factory callable returning the random number generator instance. Defaults to `secrets.SystemRandom`. - Called again in a forked child process (see Notes). + Factory callable returning the random number generator instance. Defaults to `random.SystemRandom` + (`os.urandom`). Random bits are drawn with the instance's `.getrandbits(k)`. A generator without + `.getrandbits(k)`, or whose class defines `.random()` but not `.getrandbits(k)` (such as a `random.Random` + subclass overriding only `.random()`, the custom generator pattern described in the `random` module + documentation), is drawn from with one `.random()` call per character. Called again in a forked child + process (see Notes). counter : Callable[[int], Callable[[], int]], default=utils.create_counter The `counter` parameter is a callable that creates a counter returning an incremented value each time it is called. The `create_counter` function from the `utils` module is used by default. Called again with a @@ -192,22 +191,25 @@ def __init__( Notes ----- - `generate()` may be called concurrently from multiple threads, including on free-threaded Python. Draws from - the random generator and counter are serialized per instance by a re-entrant lock, so a signal handler or - finalizer that generates an ID on the same thread does not deadlock. + `generate()` may be called concurrently from multiple threads, including on free-threaded Python. When + `random_generator` returns a `random.SystemRandom` instance and `counter` is the default, both of which are + thread-safe on their own, no instance lock is taken. Otherwise, draws from the random generator and counter + are serialized per instance by a re-entrant lock, so a signal handler or finalizer that generates an ID on the + same thread does not deadlock. Each instance records the process ID that built its state. When `generate()` first runs in a different process, such as any `os.fork()` child (including pre-fork servers that fork without running Python's fork - hooks), it builds a new lock, random generator, counter, and fingerprint once, by calling `random_generator`, - `counter`, and `fingerprint` again before generating. If `random_generator` returns the same object as before - the fork, its `seed()` method is called to reseed it from OS entropy; if it has no `seed()` method or `seed()` - raises `NotImplementedError` or `TypeError`, that generator keeps the parent's stream, and IDs still differ by - timestamp and, with the default `fingerprint`, by process ID. Fixed-seed factories (such as - `functools.partial(random.Random, 42)`) give every process the same stream and are intended for deterministic - single-process tests. + hooks), it builds a new random generator, counter, fingerprint, and (when one is used) lock once, by + calling `random_generator`, `counter`, and `fingerprint` again before generating. If `random_generator` + returns the same object as before the fork, its `seed()` method is called to reseed it from OS entropy; if it + has no `seed()` method or `seed()` raises `NotImplementedError` or `TypeError`, that generator keeps the + parent's stream, and IDs still differ by timestamp and, with the default `fingerprint`, by process ID. + Fixed-seed factories (such as `functools.partial(random.Random, 42)`) give every process the same stream and + are intended for deterministic single-process tests. `copy.deepcopy()` returns an independent generator: the three factories are shared by reference, other - attributes are deep-copied, and the lock, random generator, counter, and fingerprint are built anew. + attributes are deep-copied, and the random generator, counter, fingerprint, and (when one is used) lock are + built anew. """ _validate_length(length) @@ -245,11 +247,19 @@ def _create_state(self: Cuid, previous_random: Random | None) -> _GeneratorState if random_generator is previous_random: with suppress(AttributeError, NotImplementedError, TypeError): random_generator.seed() + # SystemRandom draws are stateless os.urandom reads and create_counter locks internally, so the defaults + # need no instance lock; any other generator or counter may keep unsynchronized state. + needs_lock: bool = ( + type(random_generator) is not SystemRandom or self._counter_factory is not utils.create_counter + ) return _GeneratorState( pid=os.getpid(), - lock=threading.RLock(), + lock=threading.RLock() if needs_lock else None, random=random_generator, - counter=self._counter_factory(floor(random_generator.random() * INITIAL_COUNT_MAX)), + draw=utils._letter_and_entropy_source(random_generator), # noqa: SLF001 # pyright: ignore[reportPrivateUsage] + counter=self._counter_factory( + utils._create_initial_count(random_generator), # noqa: SLF001 # pyright: ignore[reportPrivateUsage] + ), fingerprint=self._fingerprint_factory(random_generator=random_generator), ) @@ -283,18 +293,28 @@ def generate(self: Cuid, length: int | None = None) -> str: """ if length is not None: _validate_length(length) - resolved_length: int = self._length if length is None else length + return self._generate(self._length if length is None else length) + def _generate(self: Cuid, resolved_length: int) -> str: state: _GeneratorState = self._current_state() - with state.lock: - first_letter: str = utils.create_letter(random_generator=state.random) - base36_count: str = utils.base36_encode(state.counter()) - salt: str = utils.create_entropy(length=resolved_length, random_generator=state.random) - + lock: threading.RLock | None = state.lock + if lock is None: + first_letter, salt = state.draw(resolved_length) + count = state.counter() + else: + with lock: + first_letter, salt = state.draw(resolved_length) + count = state.counter() + + base36_count: str = utils.base36_encode(count) base36_time: str = utils.base36_encode(time.time_ns()) hash_input: str = base36_time + salt + base36_count + state.fingerprint - return first_letter + utils.create_hash(hash_input)[1:resolved_length] + return first_letter + utils._create_hash_slice( # noqa: SLF001 # pyright: ignore[reportPrivateUsage] + hash_input, + 1, + resolved_length, + ) is_cuid: ClassVar[IsCuidCallable] = staticmethod(is_cuid) @@ -328,6 +348,8 @@ def create_id(length: int | None = None) -> str: If the length parameter is outside `MIN_LENGTH`..`BIG_LENGTH` (2..32 characters). """ global _default_generator # noqa: PLW0603 + if length is not None: + _validate_length(length) generator = _default_generator if generator is None: with _process_lock(os.getpid()): @@ -335,7 +357,7 @@ def create_id(length: int | None = None) -> str: if generator is None: generator = Cuid() _default_generator = generator - return generator.generate(length=length) + return generator._generate(DEFAULT_LENGTH if length is None else length) # noqa: SLF001 # pyright: ignore[reportPrivateUsage] def cuid_wrapper(length: int = DEFAULT_LENGTH) -> Callable[[], str]: diff --git a/src/cuid2/utils.py b/src/cuid2/utils.py index 2e0cae1..0c45e1f 100644 --- a/src/cuid2/utils.py +++ b/src/cuid2/utils.py @@ -3,12 +3,16 @@ from __future__ import annotations import os -import socket -import string import threading +from functools import partial from hashlib import sha3_512 from math import floor -from typing import TYPE_CHECKING, Final +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable + from random import Random + from typing import Final # ~25.7k hosts before 50% chance of initial counter collision # with a remaining counter range of 9.0e+15 in JavaScript. @@ -18,15 +22,94 @@ BIG_LENGTH: Final[int] = 32 MAXIMUM_LENGTH: Final[int] = 98 -ASCII_LOWERCASE: Final[str] = string.ascii_lowercase +ASCII_LOWERCASE: Final[str] = "abcdefghijklmnopqrstuvwxyz" ASCII_LOWERCASE_LEN: Final[int] = len(ASCII_LOWERCASE) -BASE36_ALPHABET: Final[str] = string.digits + string.ascii_lowercase +BASE36_ALPHABET: Final[str] = "0123456789abcdefghijklmnopqrstuvwxyz" BASE36_RADIX: Final[int] = len(BASE36_ALPHABET) +_OVERSAMPLE_BITS: Final[int] = 64 +_COUNTER_DRAW_BITS: Final[int] = INITIAL_COUNT_MAX.bit_length() + _OVERSAMPLE_BITS +_LETTER_DRAW_BITS: Final[int] = ASCII_LOWERCASE_LEN.bit_length() + _OVERSAMPLE_BITS +_POWERS_OF_36: Final[tuple[int, ...]] = tuple(BASE36_RADIX**i for i in range(101)) +_HASH_MIN_99_DIGITS: Final[int] = _POWERS_OF_36[98] +_HASH_MIN_100_DIGITS: Final[int] = _POWERS_OF_36[99] + + +def _uses_random_only(random_generator: Random) -> bool: + # Generators without getrandbits(), and random.Random subclasses whose nearest definition is random() rather + # than getrandbits() (mirroring random.Random.__init_subclass__), supply their own core generator through + # random() alone; they are drawn from with one floor(random() * n) call per character, never through the + # inherited getrandbits(), so random() of any resolution keeps its full per-character entropy. + if getattr(random_generator, "getrandbits", None) is None: + return True + for klass in type(random_generator).__mro__: + namespace = vars(klass) + if "getrandbits" in namespace: + return False + if "random" in namespace: + return True + return False + + +def _scale_randbits(bits: int, upper_bound: int, draw_bits: int) -> int: + return (bits * upper_bound) >> draw_bits + + +def _create_initial_count(random_generator: Random) -> int: # pyright: ignore[reportUnusedFunction] + if _uses_random_only(random_generator): + return floor(random_generator.random() * INITIAL_COUNT_MAX) + return _scale_randbits( + random_generator.getrandbits(_COUNTER_DRAW_BITS), + INITIAL_COUNT_MAX, + _COUNTER_DRAW_BITS, + ) + + +def _entropy_from_scaled(scaled: int, length: int) -> str: + alphabet: str = BASE36_ALPHABET + chars: list[str] = [] + for _ in range(length): + scaled, mod = divmod(scaled, BASE36_RADIX) + chars.append(alphabet[mod]) + return "".join(chars) + + +def _letter_via_random(random_generator: Random) -> str: + return ASCII_LOWERCASE[floor(random_generator.random() * ASCII_LOWERCASE_LEN)] -if TYPE_CHECKING: - from _random import Random - from collections.abc import Callable + +def _entropy_via_random(random_generator: Random, length: int) -> str: + return "".join([BASE36_ALPHABET[floor(random_generator.random() * BASE36_RADIX)] for _ in range(length)]) + + +def _letter_and_entropy_via_getrandbits(getrandbits: Callable[[int], int], length: int) -> tuple[str, str]: + entropy_modulus: int = _POWERS_OF_36[length] + modulus: int = ASCII_LOWERCASE_LEN * entropy_modulus + draw_bits: int = modulus.bit_length() + _OVERSAMPLE_BITS + letter_idx, entropy_scaled = divmod(_scale_randbits(getrandbits(draw_bits), modulus, draw_bits), entropy_modulus) + return ASCII_LOWERCASE[letter_idx], _entropy_from_scaled(entropy_scaled, length) + + +def _letter_and_entropy_via_random(random_generator: Random, length: int) -> tuple[str, str]: + return _letter_via_random(random_generator), _entropy_via_random(random_generator, length) + + +def _letter_and_entropy_source( # pyright: ignore[reportUnusedFunction] + random_generator: Random, +) -> Callable[[int], tuple[str, str]]: + if _uses_random_only(random_generator): + return partial(_letter_and_entropy_via_random, random_generator) + return partial(_letter_and_entropy_via_getrandbits, random_generator.getrandbits) + + +def _create_hash_slice(data: str, start: int, stop: int) -> str: + hashed_int: int = int.from_bytes(sha3_512(data.encode()).digest(), byteorder="big") + raw_start: int = start + 1 + raw_stop: int = stop + 1 + if hashed_int >= _HASH_MIN_99_DIGITS and stop < MAXIMUM_LENGTH: + total_digits: int = 100 if hashed_int >= _HASH_MIN_100_DIGITS else 99 + return base36_encode(hashed_int // _POWERS_OF_36[total_digits - raw_stop])[raw_start:raw_stop] + return base36_encode(hashed_int)[raw_start:raw_stop] def create_counter(count: int) -> Callable[[], int]: @@ -61,8 +144,10 @@ def create_fingerprint(random_generator: Random, fingerprint_data: str | None = Parameters ---------- - random_generator : "Random" - Used as the base generator to generate some entropy. + random_generator : random.Random + Random number generator used to generate fingerprint entropy. Its `.getrandbits(k)` is used; a generator + without `.getrandbits(k)`, or whose class defines `.random()` but not `.getrandbits(k)` (such as a + `random.Random` subclass overriding only `.random()`), is drawn from with one `.random()` call per character. fingerprint_data : str, optional An optional parameter that contains data to be included in the fingerprint. If it is not provided, the function combines process ID, hostname, and environment variables @@ -75,6 +160,8 @@ def create_fingerprint(random_generator: Random, fingerprint_data: str | None = The length of the returned string is trimmed to the constant `BIG_LENGTH` (32 characters). """ if not fingerprint_data: + import socket # noqa: PLC0415 + process_id: int = os.getpid() hostname: str = socket.gethostname() env_variables: str = "".join(os.environ.keys()) @@ -82,7 +169,7 @@ def create_fingerprint(random_generator: Random, fingerprint_data: str | None = fingerprint_data = str(process_id) + hostname + env_variables fingerprint: str = str(fingerprint_data) + create_entropy(random_generator, BIG_LENGTH) - return create_hash(fingerprint)[0:BIG_LENGTH] + return _create_hash_slice(fingerprint, 0, BIG_LENGTH) def create_entropy(random_generator: Random, length: int = 4) -> str: @@ -90,8 +177,10 @@ def create_entropy(random_generator: Random, length: int = 4) -> str: Parameters ---------- - random_generator : "Random" - Used as the base generator to generate a random string. + random_generator : random.Random + Random number generator used to generate a random string. Its `.getrandbits(k)` is used; a generator + without `.getrandbits(k)`, or whose class defines `.random()` but not `.getrandbits(k)` (such as a + `random.Random` subclass overriding only `.random()`), is drawn from with one `.random()` call per character. length : int, default=4 The length parameter is an integer that specifies the length of the entropy string to be generated. The default value is 4, but it can be set to any positive integer value. @@ -111,7 +200,13 @@ def create_entropy(random_generator: Random, length: int = 4) -> str: msg = "Cannot create entropy without a length >= 1." raise ValueError(msg) - return "".join([BASE36_ALPHABET[floor(random_generator.random() * BASE36_RADIX)] for _ in range(length)]) + if _uses_random_only(random_generator): + return _entropy_via_random(random_generator, length) + + modulus: int = _POWERS_OF_36[length] if length < len(_POWERS_OF_36) else BASE36_RADIX**length + draw_bits: int = modulus.bit_length() + _OVERSAMPLE_BITS + scaled: int = _scale_randbits(random_generator.getrandbits(draw_bits), modulus, draw_bits) + return _entropy_from_scaled(scaled, length) def create_hash(data: str = "") -> str: @@ -141,15 +236,25 @@ def create_letter(random_generator: Random) -> str: Parameters ---------- - random_generator : "Random" - Used as the base generator to generate a random letter. + random_generator : random.Random + Random number generator used to generate a random letter. Its `.getrandbits(k)` is used; a generator + without `.getrandbits(k)`, or whose class defines `.random()` but not `.getrandbits(k)` (such as a + `random.Random` subclass overriding only `.random()`), is drawn from with one `.random()` call. Returns ------- str a randomly generated lowercase letter from the English alphabet. """ - return ASCII_LOWERCASE[floor(random_generator.random() * ASCII_LOWERCASE_LEN)] + if _uses_random_only(random_generator): + return _letter_via_random(random_generator) + return ASCII_LOWERCASE[ + _scale_randbits( + random_generator.getrandbits(_LETTER_DRAW_BITS), + ASCII_LOWERCASE_LEN, + _LETTER_DRAW_BITS, + ) + ] def base36_encode(number: int) -> str: @@ -179,10 +284,11 @@ def base36_encode(number: int) -> str: if number == 0: return "0" + alphabet: str = BASE36_ALPHABET encoded_chars: list[str] = [] while number != 0: number, mod = divmod(number, BASE36_RADIX) - encoded_chars.append(BASE36_ALPHABET[mod]) + encoded_chars.append(alphabet[mod]) encoded_chars.reverse() return "".join(encoded_chars)