Fix core RNG: seed HMAC-DRBG from camera frame hash

The previous version computed a SHA-256 of the frame into a variable
named `entropy` and then discarded it, drawing instead from the OS
CSPRNG via secrets.randbits(). The camera contributed nothing.

Replace with a NIST SP 800-90A HMAC-DRBG seeded from the frame hash so
the camera genuinely contributes entropy. Add --interval for loop
rate control, --out for writing raw bytes to a FIFO, mutual exclusion
of --loop/--single, and bit-multiple validation.

Generated by Mistral Vibe.
Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
Radek
2026-08-31 09:41:12 +01:00
parent a2dedb502e
commit 239672ca63
+88 -21
View File
@@ -1,13 +1,57 @@
#!/usr/bin/env python3
import cv2
import hashlib
import secrets
import time
"""Camera-based entropy source.
Fetches MJPEG frames from a network camera, hashes them, and feeds the
hash into a NIST SP 800-90A HMAC-DRBG. Output is the DRBG's generated
bytes, so the camera genuinely contributes entropy (unlike the previous
version which computed a hash and then discarded it).
"""
import argparse
import hashlib
import hmac
import sys
import time
import cv2
sys.stdout.reconfigure(line_buffering=True)
class HMACDRBG:
"""NIST SP 800-90A HMAC-DRBG using SHA-256.
Seeded with camera frame entropy. Supports reseed so each frame
mixes fresh entropy into the internal state.
"""
def __init__(self, entropy, personalization=b""):
self.K = b"\x00" * 32
self.V = b"\x01" * 32
self.reseed_counter = 0
self._update(entropy + personalization)
def _update(self, provided_data):
self.K = hmac.new(self.K, self.V + b"\x00" + provided_data, hashlib.sha256).digest()
self.V = hmac.new(self.K, self.V, hashlib.sha256).digest()
self.K = hmac.new(self.K, self.V + b"\x01" + provided_data, hashlib.sha256).digest()
self.V = hmac.new(self.K, self.V, hashlib.sha256).digest()
def reseed(self, entropy):
self._update(entropy)
self.reseed_counter = 0
def generate(self, num_bytes):
if self.reseed_counter > 10000:
raise RuntimeError("DRBG requires reseed before generating more output.")
self.reseed_counter += 1
out = b""
while len(out) < num_bytes:
self.V = hmac.new(self.K, self.V, hashlib.sha256).digest()
out += self.V
self._update(b"")
return out[:num_bytes]
def fetch_frame_opencv(url):
cap = cv2.VideoCapture(url)
ret, frame = cap.read()
@@ -16,11 +60,20 @@ def fetch_frame_opencv(url):
raise ValueError("Failed to fetch frame.")
return frame
def generate_random_numbers(frame, bit_lengths):
_, img_encoded = cv2.imencode('.jpg', frame)
img_bytes = img_encoded.tobytes()
entropy = hashlib.sha256(img_bytes).digest()
return {length: secrets.randbits(length) for length in bit_lengths}
def frame_entropy(frame):
"""Hash a captured frame to a 32-byte digest."""
_, img_encoded = cv2.imencode(".jpg", frame)
return hashlib.sha256(img_encoded.tobytes()).digest()
def emit(blob, out_path):
if out_path:
with open(out_path, "ab") as f:
f.write(blob)
else:
sys.stdout.write(blob.hex() + "\n")
def main():
parser = argparse.ArgumentParser(description="Generate random numbers from camera entropy.")
@@ -28,31 +81,45 @@ def main():
parser.add_argument("--single", type=int, default=1, help="Generate N random numbers and exit (default: 1)")
parser.add_argument("--bit", type=int, default=256, help="Bit length of each random number (default: 256)")
parser.add_argument("--loop", action="store_true", help="Run forever, generating numbers every second")
parser.add_argument("--interval", type=float, default=1.0, help="Seconds between iterations in --loop (default: 1.0)")
parser.add_argument("--out", help="Append raw bytes to this path (e.g. a FIFO) instead of printing hex")
args = parser.parse_args()
if args.bit <= 0 or args.bit % 8 != 0:
print("Error: --bit must be a positive multiple of 8", file=sys.stderr)
return 2
num_bytes = args.bit // 8
if args.loop and args.single != 1:
print("Error: --loop and --single are mutually exclusive", file=sys.stderr)
return 2
def produce():
frame = fetch_frame_opencv(args.url)
entropy = frame_entropy(frame)
drbg = HMACDRBG(entropy)
return drbg.generate(num_bytes)
if args.loop:
while True:
try:
frame = fetch_frame_opencv(args.url)
random_numbers = generate_random_numbers(frame, [args.bit])
for number in random_numbers.values():
print(f"{number:x}")
blob = produce()
emit(blob, args.out)
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
time.sleep(1) # Wait before retrying
time.sleep(args.interval)
else:
for i in range(args.single):
try:
frame = fetch_frame_opencv(args.url)
random_numbers = generate_random_numbers(frame, [args.bit])
for number in random_numbers.values():
print(f"{number:x}")
blob = produce()
emit(blob, args.out)
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
return 1
if args.single > 1 and i < args.single - 1:
time.sleep(1) # Small delay between frames
time.sleep(args.interval)
return 0
if __name__ == "__main__":
import sys
main()
sys.exit(main())