nebulaforge-offline-resilie.../nebulaforge/robustness.py

107 lines
3.5 KiB
Python

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