|
| 1 | +"""Entropy sources for divination casts (起卦 / 洗牌). |
| 2 | +
|
| 3 | +Three sources, selected per cast: |
| 4 | +
|
| 5 | + * ``seed`` — ``random.Random(seed)``: deterministic, for tests / reproducible casts. |
| 6 | + * ``system`` — ``random.SystemRandom``: OS CSPRNG (default). Cryptographically |
| 7 | + strong pseudo-randomness. |
| 8 | + * ``quantum``— ``QuantumRandom``: physical randomness from quantum vacuum noise |
| 9 | + (ANU QRNG). Falls back to ``os.urandom`` (flagged ``degraded``) |
| 10 | + if the quantum source is unreachable. |
| 11 | +
|
| 12 | +HONEST FRAMING — read this before using ``quantum``: |
| 13 | +A quantum entropy source does NOT make a divination "more accurate" or "more |
| 14 | +true". Hexagram/card outcomes are uniform regardless of whether the bits come |
| 15 | +from a CPU CSPRNG or a photon beam-splitter; accuracy of a reading has no |
| 16 | +physical dependence on the entropy source. ``quantum`` is offered only as a |
| 17 | +*physically-true randomness* option (philosophically meaningful to some |
| 18 | +practitioners), never as an accuracy improvement. Output always labels the |
| 19 | +source so the distinction stays transparent. |
| 20 | +""" |
| 21 | +from __future__ import annotations |
| 22 | + |
| 23 | +import json |
| 24 | +import os |
| 25 | +import random |
| 26 | +import urllib.request |
| 27 | + |
| 28 | +# ANU Quantum Random Number Generator (quantum vacuum fluctuations). |
| 29 | +# Modern authenticated endpoint needs an API key (env ANU_QRNG_API_KEY); |
| 30 | +# the legacy free endpoint is attempted as a fallback. Either may be down — |
| 31 | +# QuantumRandom degrades to os.urandom rather than ever blocking a cast. |
| 32 | +_ANU_AUTH_URL = "https://api.quantumnumbers.anu.edu.au" |
| 33 | +_ANU_LEGACY_URL = "https://qrng.anu.edu.au/API/jsonI.php" |
| 34 | +_BLOCK = 256 # bytes fetched per refill |
| 35 | + |
| 36 | + |
| 37 | +def _fetch_quantum_bytes(n: int, timeout: float = 6.0) -> bytes | None: |
| 38 | + """Return ``n`` quantum-random bytes, or None if unavailable.""" |
| 39 | + key = os.environ.get("ANU_QRNG_API_KEY") |
| 40 | + try: |
| 41 | + if key: |
| 42 | + req = urllib.request.Request( |
| 43 | + f"{_ANU_AUTH_URL}?length={n}&type=uint8", |
| 44 | + headers={"x-api-key": key}, |
| 45 | + ) |
| 46 | + else: |
| 47 | + req = urllib.request.Request( |
| 48 | + f"{_ANU_LEGACY_URL}?length={n}&type=uint8" |
| 49 | + ) |
| 50 | + with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 |
| 51 | + payload = json.loads(resp.read().decode("utf-8")) |
| 52 | + if payload.get("success") and isinstance(payload.get("data"), list): |
| 53 | + data = bytes(int(b) & 0xFF for b in payload["data"]) |
| 54 | + return data[:n] if len(data) >= n else None |
| 55 | + except Exception: |
| 56 | + return None |
| 57 | + return None |
| 58 | + |
| 59 | + |
| 60 | +class QuantumRandom(random.Random): |
| 61 | + """``random.Random`` backed by a quantum entropy pool. |
| 62 | +
|
| 63 | + Consumes bytes from the ANU QRNG; refills on demand. Any fetch failure |
| 64 | + flips ``degraded = True`` and the pool is topped up from ``os.urandom`` |
| 65 | + (still a CSPRNG) so a cast never blocks or crashes. |
| 66 | + """ |
| 67 | + |
| 68 | + def __init__(self) -> None: |
| 69 | + self._pool = bytearray() |
| 70 | + self.degraded = False |
| 71 | + super().__init__() |
| 72 | + |
| 73 | + def _ensure(self, n: int) -> None: |
| 74 | + while len(self._pool) < n: |
| 75 | + qb = _fetch_quantum_bytes(_BLOCK) |
| 76 | + if qb: |
| 77 | + self._pool.extend(qb) |
| 78 | + else: |
| 79 | + self.degraded = True |
| 80 | + self._pool.extend(os.urandom(_BLOCK)) |
| 81 | + |
| 82 | + def _take(self, n: int) -> int: |
| 83 | + self._ensure(n) |
| 84 | + chunk = bytes(self._pool[:n]) |
| 85 | + del self._pool[:n] |
| 86 | + return int.from_bytes(chunk, "big") |
| 87 | + |
| 88 | + def getrandbits(self, k: int) -> int: |
| 89 | + if k <= 0: |
| 90 | + raise ValueError("number of bits must be greater than zero") |
| 91 | + nbytes = (k + 7) // 8 |
| 92 | + value = self._take(nbytes) |
| 93 | + return value >> (nbytes * 8 - k) # trim to exactly k bits |
| 94 | + |
| 95 | + def random(self) -> float: |
| 96 | + # 53-bit mantissa float in [0, 1), per CPython convention. |
| 97 | + return self.getrandbits(53) / (1 << 53) |
| 98 | + |
| 99 | + def seed(self, *args, **kwargs) -> None: # noqa: D401 |
| 100 | + """No-op: entropy is externally sourced, not seedable.""" |
| 101 | + return None |
| 102 | + |
| 103 | + |
| 104 | +def get_rng(seed: int | None = None, source: str = "system"): |
| 105 | + """Return an RNG for a cast. |
| 106 | +
|
| 107 | + seed given -> deterministic random.Random(seed) (source ignored) |
| 108 | + source == quantum -> QuantumRandom (physical, degrades to os.urandom) |
| 109 | + otherwise -> random.SystemRandom (OS CSPRNG, default) |
| 110 | + """ |
| 111 | + if seed is not None: |
| 112 | + return random.Random(seed) |
| 113 | + if source == "quantum": |
| 114 | + return QuantumRandom() |
| 115 | + return random.SystemRandom() |
| 116 | + |
| 117 | + |
| 118 | +def describe(rng, source: str, seed: int | None) -> dict: |
| 119 | + """Structured, honest provenance for the output JSON.""" |
| 120 | + if seed is not None: |
| 121 | + return {"source": "seed", "seed": seed, "reproducible": True, |
| 122 | + "note": "确定性种子, 仅供复现/测试"} |
| 123 | + if source == "quantum": |
| 124 | + degraded = getattr(rng, "degraded", False) |
| 125 | + return { |
| 126 | + "source": "os_urandom_fallback" if degraded else "quantum_vacuum_anu", |
| 127 | + "degraded": degraded, |
| 128 | + "note": ("量子源不可达, 已降级到 OS CSPRNG" if degraded |
| 129 | + else "ANU 量子真空噪声物理真随机; 不影响卦象准确度, 仅熵源差异"), |
| 130 | + } |
| 131 | + return {"source": "system_csprng", |
| 132 | + "note": "OS 加密级伪随机 (random.SystemRandom)"} |
0 commit comments