distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""RandomSource — cryptographically secure random number generator.
2
3Ported from net.i2p.util.RandomSource.
4Wraps Python's secrets/os.urandom which use the OS CSPRNG.
5"""
6
7import os
8import secrets
9import threading
10
11
12class RandomSource:
13 """Cryptographically secure random source.
14
15 In Java, this extends SecureRandom and adds Fortuna PRNG.
16 In Python, we delegate to the OS CSPRNG via secrets module.
17 """
18
19 _instance: "RandomSource | None" = None
20 _lock = threading.Lock()
21
22 def __init__(self) -> None:
23 pass
24
25 @classmethod
26 def get_instance(cls) -> "RandomSource":
27 if cls._instance is None:
28 with cls._lock:
29 if cls._instance is None:
30 cls._instance = RandomSource()
31 return cls._instance
32
33 def next_int(self, n: int) -> int:
34 """Return random int in [0, n)."""
35 if n <= 0:
36 raise ValueError("bound must be positive")
37 return secrets.randbelow(n)
38
39 def signed_next_int(self) -> int:
40 """Return random int across full 32-bit signed range."""
41 return int.from_bytes(os.urandom(4), "big", signed=True)
42
43 def next_long(self, n: int) -> int:
44 """Return random long in [0, n)."""
45 if n <= 0:
46 raise ValueError("bound must be positive")
47 return secrets.randbelow(n)
48
49 def next_bytes(self, length: int) -> bytes:
50 """Return `length` random bytes."""
51 return os.urandom(length)
52
53 def next_bytes_into(self, buf: bytearray, offset: int = 0, length: int = -1) -> None:
54 """Fill buffer with random bytes."""
55 if length < 0:
56 length = len(buf) - offset
57 rand = os.urandom(length)
58 buf[offset : offset + length] = rand
59
60 def next_boolean(self) -> bool:
61 return bool(secrets.randbelow(2))