Radek
d121b0f0f6
test_entropy.py covers the HMAC-DRBG (determinism, reseed, length,
reseed limit), health checks (random passes, frozen/all-ones/short/
long-run rejected), a bit-balance smoke test on DRBG output, and the
API endpoints including auth gating. All 16 tests pass without a
camera (producer is stubbed).
Switch api.py from the deprecated @app.on_event('startup') to the
lifespan async context manager. Add pytest to requirements.txt.
Generated by Mistral Vibe.
Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
125 lines
3.4 KiB
Python
125 lines
3.4 KiB
Python
"""FastAPI entropy service.
|
|
|
|
Runs the camera entropy producer in a background thread and serves
|
|
the latest generated blob over HTTP. Designed for container deployment:
|
|
the producer and API share one process, so there is no FIFO to manage
|
|
and no cross-container IPC.
|
|
|
|
Endpoints:
|
|
GET /healthz -> liveness probe
|
|
GET /entropy -> { hex, bits, ts } (requires X-API-Key if configured)
|
|
GET / -> service info
|
|
"""
|
|
import os
|
|
import threading
|
|
import time
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import FastAPI, Header, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
import entropy
|
|
|
|
CAMERA_URL = os.environ.get("ENTROPY_CAMERA_URL", "http://192.168.0.200/mjpg/video.mjpg")
|
|
BITS = int(os.environ.get("ENTROPY_BITS", "256"))
|
|
INTERVAL = float(os.environ.get("ENTROPY_INTERVAL", "1.0"))
|
|
NO_HEALTH = os.environ.get("ENTROPY_NO_HEALTH", "") == "1"
|
|
# Simple shared-secret auth. Empty = no auth (for the first iteration).
|
|
API_KEY = os.environ.get("ENTROPY_API_KEY", "")
|
|
|
|
NUM_BYTES = BITS // 8
|
|
|
|
|
|
class LatestBlob:
|
|
"""Thread-safe holder for the most recent entropy blob."""
|
|
|
|
def __init__(self):
|
|
self._lock = threading.Lock()
|
|
self._blob = None
|
|
self._ts = None
|
|
self._ok = False
|
|
|
|
def set(self, blob):
|
|
with self._lock:
|
|
self._blob = blob
|
|
self._ts = time.time()
|
|
self._ok = True
|
|
|
|
def get(self):
|
|
with self._lock:
|
|
return self._blob, self._ts, self._ok
|
|
|
|
def mark_failed(self):
|
|
with self._lock:
|
|
self._ok = False
|
|
|
|
|
|
latest = LatestBlob()
|
|
|
|
|
|
def producer_loop():
|
|
"""Continuously generate entropy blobs and stash the latest one."""
|
|
while True:
|
|
try:
|
|
frame = entropy.fetch_frame_opencv(CAMERA_URL)
|
|
raw = entropy.frame_raw_bytes(frame)
|
|
if not NO_HEALTH:
|
|
from health import check as health_check
|
|
health_check(raw)
|
|
digest = entropy.HMACDRBG(raw).generate(NUM_BYTES)
|
|
latest.set(digest)
|
|
except Exception as e:
|
|
print(f"[producer] error: {e}", flush=True)
|
|
latest.mark_failed()
|
|
time.sleep(INTERVAL)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app):
|
|
t = threading.Thread(target=producer_loop, daemon=True)
|
|
t.start()
|
|
yield
|
|
|
|
|
|
app = FastAPI(title="Entropy-RNG Service", version="1.0", lifespan=lifespan)
|
|
|
|
|
|
class EntropyResponse(BaseModel):
|
|
hex: str
|
|
bits: int
|
|
ts: str
|
|
|
|
|
|
@app.get("/")
|
|
def root():
|
|
blob, ts, ok = latest.get()
|
|
return {
|
|
"service": "entropy-rng",
|
|
"bits": BITS,
|
|
"healthy": ok,
|
|
"last_update": datetime.fromtimestamp(ts, tz=timezone.utc).isoformat() if ts else None,
|
|
}
|
|
|
|
|
|
@app.get("/healthz")
|
|
def healthz():
|
|
blob, ts, ok = latest.get()
|
|
if not ok or blob is None:
|
|
raise HTTPException(status_code=503, detail="no entropy available")
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/entropy")
|
|
def get_entropy(x_api_key: str | None = Header(default=None)):
|
|
if API_KEY and x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="invalid or missing API key")
|
|
blob, ts, ok = latest.get()
|
|
if blob is None:
|
|
raise HTTPException(status_code=503, detail="no entropy available yet")
|
|
return EntropyResponse(
|
|
hex=blob.hex(),
|
|
bits=BITS,
|
|
ts=datetime.fromtimestamp(ts, tz=timezone.utc).isoformat(),
|
|
)
|