diff --git a/nebulaforge/robustness.py b/nebulaforge/robustness.py new file mode 100644 index 0000000..8897c7f --- /dev/null +++ b/nebulaforge/robustness.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +"""SEU fuzzing harness: deterministic bitflip injection for tensors. + +This small utility injects single-event-upset-style bit flips into +numpy arrays (viewed as raw bytes). It is deterministic given a seed +and returns a small bundle describing the flips for reproducible +counterexample bundles used in CI and forensics. + +Functions: +- inject_bitflips(arr, rate=1e-3, seed=None): returns (flipped, bundle) +- detect_corruption(arr, bundle): quick checksum-based detector + +The implementation is intentionally small and pure-Python to keep tests +fast and dependency-free (only numpy used, which the project already +depends on). +""" + +from dataclasses import dataclass, asdict +import zlib +import hashlib +from typing import Dict, List, Optional, Tuple + +import numpy as np + + +@dataclass +class SEUBundle: + seed: Optional[int] + rate: float + flips: List[int] + original_crc32: int + + +def _crc32_bytes(b: bytes) -> int: + return zlib.crc32(b) & 0xFFFFFFFF + + +def inject_bitflips( + arr: np.ndarray, + rate: float = 1e-3, + seed: Optional[int] = None, + max_flips: Optional[int] = None, +) -> Tuple[np.ndarray, SEUBundle]: + """Inject deterministic bit flips into a numpy array's bytes. + + - arr: input numpy array (will not be modified; a copy is returned) + - rate: probability of flipping each bit (0..1). Small values recommended. + - seed: optional RNG seed for determinism. If None, non-deterministic. + - max_flips: optional cap on number of flips applied. + + Returns (flipped_array, bundle) where bundle records seed, rate, + list of flipped bit indices (as bit offsets into the raw buffer), + and the CRC32 of the original bytes for detection. + """ + if rate < 0.0 or rate > 1.0: + raise ValueError("rate must be in [0,1]") + + rnd = np.random.RandomState(seed) + + orig_bytes = arr.tobytes() + orig_crc = _crc32_bytes(orig_bytes) + + # Represent data as a mutable bytearray so we can flip bits + b = bytearray(orig_bytes) + n_bits = len(b) * 8 + + # decide which bits to flip + # draw bernoulli per-bit; to remain efficient for small arrays we sample indices + flip_mask = rnd.rand(n_bits) < rate + idxs = np.nonzero(flip_mask)[0].tolist() + + if max_flips is not None and len(idxs) > max_flips: + idxs = idxs[:max_flips] + + # apply flips: for each bit index, flip the corresponding bit in the bytearray + for bit_idx in idxs: + byte_idx = bit_idx // 8 + bit_in_byte = bit_idx % 8 + b[byte_idx] ^= 1 << bit_in_byte + + flipped = np.frombuffer(bytes(b), dtype=np.uint8).view(arr.dtype).reshape(arr.shape).copy() + + bundle = SEUBundle(seed=seed, rate=rate, flips=idxs, original_crc32=orig_crc) + return flipped, bundle + + +def detect_corruption(arr: np.ndarray, bundle: SEUBundle) -> bool: + """Return True if the array's bytes differ from the bundle's recorded CRC32.""" + current_crc = _crc32_bytes(arr.tobytes()) + return current_crc != bundle.original_crc32 + + +def reproduce_flips(arr: np.ndarray, bundle: SEUBundle) -> np.ndarray: + """Reproduce the flips recorded in bundle on a copy of arr. + + This is useful for deterministic replay: it applies the listed bit + offsets to a fresh copy of the original array. + """ + b = bytearray(arr.tobytes()) + for bit_idx in bundle.flips: + byte_idx = bit_idx // 8 + bit_in_byte = bit_idx % 8 + b[byte_idx] ^= 1 << bit_in_byte + out = np.frombuffer(bytes(b), dtype=np.uint8).view(arr.dtype).reshape(arr.shape).copy() + return out diff --git a/tests/test_robustness.py b/tests/test_robustness.py new file mode 100644 index 0000000..ee16972 --- /dev/null +++ b/tests/test_robustness.py @@ -0,0 +1,27 @@ +import numpy as np +from nebulaforge.robustness import inject_bitflips, detect_corruption, reproduce_flips + + +def test_inject_deterministic_and_detect(): + a = np.arange(16, dtype=np.float32) + # use a noticeable rate so we flip some bits + flipped1, bundle = inject_bitflips(a, rate=0.01, seed=42) + + # detector should say corruption relative to original + assert detect_corruption(flipped1, bundle) + + # reproducing flips on original should yield identical bytes + reproduced = reproduce_flips(a, bundle) + assert reproduced.tobytes() == flipped1.tobytes() + + +def test_reproducible_with_same_seed(): + a = np.linspace(-1.0, 1.0, num=64, dtype=np.float32) + f1, b1 = inject_bitflips(a, rate=0.005, seed=123) + f2, b2 = inject_bitflips(a, rate=0.005, seed=123) + + # bundles should be identical for same seed + assert b1.seed == b2.seed + assert b1.flips == b2.flips + # flipped arrays equal + assert f1.tobytes() == f2.tobytes()