Fix: Reject dead-TRNG output in get_random_bytes (#370) (#372)

### src/rng.py
@@ -2,6 +2,7 @@
 # if os.urandom is available - entropy goes from hardware TRNG
 # in simulator just use /dev/urandom
 import hashlib
+from errors import BaseError
 
 entropy_pool = b"7" * 64
 
@@ -14,16 +15,90 @@ def get_trng_bytes(nbytes):
             return f.read(nbytes)
 
 
-# assuming that entropy_pool has some real entropy
-# we can generate bytes using it as well
-# probably not the best way at the moment,
-# but anything is better than nothing
+class RNGError(BaseError):
+    """Raised when the hardware TRNG output fails a basic sanity check."""
+
+    NAME = "RNG Error"
+
+
+_FP = 1 << 32
+
+
+def _expected_distinct(nbytes):
+    """Expected number of distinct byte values in nbytes of healthy output.
+
+    256 * (1 - (255/256)**nbytes), floored. Computed in fixed point rather
+    than floats so the threshold is identical on every build - MicroPython on
+    the F469 is single precision, CPython in the tests is double.
+    """
+    # (255/256)**nbytes in Q32, by repeated squaring
+    r = _FP
+    b = (255 * _FP) // 256
+    while nbytes:
+        if nbytes & 1:
+            r = (r * b) // _FP
+        b = (b * b) // _FP
+        nbytes >>= 1
+    return (256 * (_FP - r)) // _FP
+
+
+def _looks_dead(data):
+    """Detect a stalled or failed TRNG.
 
+    rng_get() in the STM32 port returns 0 on timeout (ports/stm32/rng.c) and
+    os.urandom() calls it once per byte, so a dead peripheral surfaces as
+    all-zero output - or, more generally, as a single repeated byte. A
+    peripheral that stalls only intermittently surfaces as mostly-repeated
+    output with a few live bytes mixed in, so a plain "all bytes equal" test
+    is not enough. Two checks:
 
+    1. No single byte value may cover more than half the buffer. This is what
+       catches a partial stall - 25 zeros and 7 live bytes in 32 is obviously
+       broken, yet has enough distinct values to pass a counting test.
+    2. The number of distinct values must be at least half of what healthy
+       output gives (_expected_distinct). Scaling with the expectation matters
+       because distinct values saturate at 256: a fixed threshold that is safe
+       for 16 bytes accepts a 75%-dead 1000-byte buffer.
+
+    Both are far from the healthy distribution, so a false rejection is below
+    2^-24 for the shortest checked buffer and below 2^-40 from 16 bytes up
+    (16 bytes is a 12-word seed, 32 a 24-word one).
+
+    This is a liveness check, not a proof of health: no cheap check can
+    distinguish a healthy TRNG from a subtly biased one. It only catches the
+    failure mode where the peripheral stops responding, which is currently
+    silent.
+
+    The check is skipped for very small requests, where a repeated byte can
+    occur legitimately and is not evidence of failure.
+    """
+    n = len(data)
+    if n < 4:
+        return False
+    # iterating bytes yields one int per byte, so counts maps byte value ->
+    # occurrences and top is the largest of those counts
+    counts = {}
+    top = 0
+    for byte in data:
+        c = counts.get(byte, 0) + 1
+        counts[byte] = c
+        if c > top:
+            top = c
+    # below 8 bytes a majority value is plausible in healthy output
+    # (b"\x00\x00\x00\x01" is fine), so only the count check applies there
+    if n >= 8 and top * 2 > n:
+        return True
+    return 2 * len(counts) < _expected_distinct(n)
+
+
+# assuming that entropy_pool has some real entropy
+# we can generate bytes using it as well
 def get_random_bytes(nbytes):
     global entropy_pool
     d = get_trng_bytes(nbytes)
-    feed(d)  # why not?
+    if _looks_dead(d):
+        raise RNGError("TRNG returned no entropy")
+    feed(d)
     # if more than 64 - just do trng
     if nbytes > 64:
         return d
@@ -34,8 +109,6 @@ def get_random_bytes(nbytes):
 
 
 # we hash together entropy pool and data we got
-
-
 def feed(data):
     global entropy_pool
     h = hashlib.sha512(entropy_pool)

### test/tests_native/__init__.py
@@ -1,5 +1,6 @@
 from .test_manifest_inventory import *
 from .test_wallet_manager_parsing import *
+from .test_rng import *
 from .test_wallet_manager_warnings import *
 from .test_change_classification import *
 from .test_transaction_confirmation import *

### test/tests_native/test_rng.py
@@ -0,0 +1,140 @@
+import random
+import sys
+
+if sys.implementation.name != "micropython":
+    from native_support import setup_native_stubs
+
+    setup_native_stubs()
+
+from unittest import TestCase
+
+import rng
+from errors import BaseError
+
+
+class RNGSanityCheckTest(TestCase):
+    def setUp(self):
+        self.original_get_trng_bytes = rng.get_trng_bytes
+        self.original_entropy_pool = rng.entropy_pool
+
+    def tearDown(self):
+        rng.get_trng_bytes = self.original_get_trng_bytes
+        rng.entropy_pool = self.original_entropy_pool
+
+    def test_looks_dead_ignores_short_repeated_buffers(self):
+        self.assertFalse(rng._looks_dead(b""))
+        self.assertFalse(rng._looks_dead(b"\x00"))
+        self.assertFalse(rng._looks_dead(b"\x00\x00\x00"))
+        self.assertFalse(rng._looks_dead(b"\xff\xff\xff"))
+
+    def test_looks_dead_rejects_repeated_buffers_from_four_bytes(self):
+        self.assertTrue(rng._looks_dead(b"\x00\x00\x00\x00"))
+        self.assertTrue(rng._looks_dead(b"\xff\xff\xff\xff"))
+        self.assertTrue(rng._looks_dead(b"\x11" * 32))
+
+    def test_looks_dead_allows_non_repeated_buffers(self):
+        self.assertFalse(rng._looks_dead(b"\x00\x00\x00\x01"))
+        self.assertFalse(rng._looks_dead(bytes(range(32))))
+
+    def test_get_random_bytes_raises_before_feeding_dead_output(self):
+        rng.entropy_pool = b"A" * 64
+        rng.get_trng_bytes = lambda nbytes: b"\x00" * nbytes
+
+        try:
+            rng.get_random_bytes(32)
+        except rng.RNGError:
+            pass
+        else:
+            self.fail("Expected RNGError for repeated TRNG output")
+
+        self.assertEqual(rng.entropy_pool, b"A" * 64)
+
+    def test_get_random_bytes_handles_zero_length_requests(self):
+        # apps/getrandom.py rejects num_bytes < 0 but permits 0, so a host can
+        # reach this: _looks_dead(b"") is False, feed(b"") still advances the
+        # pool, and the caller gets an empty result rather than an error
+        rng.entropy_pool = b"A" * 64
+        rng.get_trng_bytes = lambda nbytes: b""
+
+        self.assertEqual(rng.get_random_bytes(0), b"")
+        self.assertNotEqual(rng.entropy_pool, b"A" * 64)
+
+    def test_get_random_bytes_keeps_one_byte_requests_working(self):
+        rng.get_trng_bytes = lambda nbytes: b"\x00" * nbytes
+        self.assertEqual(len(rng.get_random_bytes(1)), 1)
+
+    def test_get_random_bytes_returns_requested_length_for_live_output(self):
+        rng.get_trng_bytes = lambda nbytes: bytes(range(nbytes))
+        self.assertEqual(len(rng.get_random_bytes(32)), 32)
+
+    def test_looks_dead_rejects_partially_stalled_output(self):
+        # an intermittently stalling peripheral returns mostly-repeated output
+        # with a few live bytes - a plain "all bytes equal" test misses this
+        self.assertTrue(rng._looks_dead(b"\x00" * 31 + b"\x2a"))
+        self.assertTrue(rng._looks_dead(b"\x00" * 26 + bytes(range(1, 7))))
+
+    def test_looks_dead_rejects_majority_stall_that_survives_counting(self):
+        # these have enough distinct values to pass a plain distinct-count
+        # threshold, but are mostly one repeated byte
+        # 16 bytes = 12-word seed entropy: 13 zeros + 3 live bytes
+        self.assertTrue(rng._looks_dead(b"\x00" * 13 + bytes(range(1, 4))))
+        # 32 bytes = 24-word seed entropy: 25 zeros + 7 live bytes
+        self.assertTrue(rng._looks_dead(b"\x00" * 25 + bytes(range(1, 8))))
+        # 96 of 128 bytes stalled
+        self.assertTrue(rng._looks_dead(b"\x00" * 96 + bytes(range(1, 33))))
+
+    def test_looks_dead_rejects_low_variety_without_a_majority_value(self):
+        # no value covers half the buffer, but 8 distinct values in 32 bytes
+        # is ~2^-111 for healthy output (expected is ~30)
+        self.assertTrue(rng._looks_dead(bytes(range(8)) * 4))
+        # 1000 bytes with 40 distinct values - passes any fixed cap of 32
+        self.assertTrue(rng._looks_dead(bytes(range(40)) * 25))
+
+    def test_looks_dead_allows_healthy_long_buffers(self):
+        # distinct byte values saturate at 256, so the threshold cannot grow
+        # with n - 245 distinct values in 1000 bytes is healthy TRNG output
+        data = bytes(range(245)) * 4 + bytes(range(20))
+        self.assertEqual(len(data), 1000)
+        self.assertEqual(len(set(data)), 245)
+        self.assertFalse(rng._looks_dead(data))
+
+    def test_looks_dead_passes_healthy_random_output(self):
+        # guards against a threshold tight enough to reject healthy hardware.
+        # A seeded PRNG rather than os.urandom: the thresholds do have a
+        # non-zero false rejection rate (5.96e-8 at 4 bytes, 1.36e-8 at 8,
+        # 6.43e-13 at 16), which over enough CI runs would eventually flake.
+        # Uniform independent bytes are what the check is specified against,
+        # so a fixed stream tests the same property without the dice roll.
+        prng = random.Random(0x5EEDBEEF)
+        for nbytes in (4, 8, 16, 32, 64, 128, 1000):
+            for _ in range(100):
+                data = bytes(prng.getrandbits(8) for _ in range(nbytes))
+                self.assertFalse(rng._looks_dead(data))
+
+    def test_expected_distinct_matches_the_closed_form(self):
+        for nbytes in (1, 4, 8, 16, 32, 64, 128, 256, 512, 1000):
+            self.assertEqual(
+                rng._expected_distinct(nbytes),
+                int(256 * (1 - (255 / 256) ** nbytes)),
+            )
+
+    def test_get_random_bytes_checks_trng_on_the_raw_path(self):
+        # requests over 64 bytes return TRNG output directly, without mixing
+        # in the entropy pool, so the sanity check is the only defence there
+        rng.get_trng_bytes = lambda nbytes: b"\x00" * nbytes
+        try:
+            rng.get_random_bytes(100)
+        except rng.RNGError:
+            pass
+        else:
+            self.fail("Expected RNGError for dead TRNG output above 64 bytes")
+
+    def test_get_random_bytes_returns_raw_trng_above_64_bytes(self):
+        rng.get_trng_bytes = lambda nbytes: bytes(range(nbytes))
+        self.assertEqual(rng.get_random_bytes(100), bytes(range(100)))
+
+    def test_rng_error_is_a_base_error(self):
+        # BaseError subclasses get a readable alert in specter.py instead of
+        # an "unexpected error" traceback
+        self.assertTrue(issubclass(rng.RNGError, BaseError))
+        self.assertEqual(rng.RNGError.NAME, "RNG Error")