Fuzzing & Robustness Testing

"""CAN bus fuzzer for ECU robustness testing."""
import can, random, time
from dataclasses import dataclass
from typing import List

@dataclass
class FuzzResult:
    can_id: int
    data:   bytes
    ecu_response: str  # "normal", "silent", "error_frame", "reset"

class CanFuzzer:
    def __init__(self, tx_bus: can.BusABC, rx_bus: can.BusABC, db):
        self.tx = tx_bus
        self.rx = rx_bus
        self.db = db
        self.results: List[FuzzResult] = []

    def fuzz_known_message(self, can_id: int, n_tests: int = 100):
        """Fuzz a known CAN ID with random data."""
        for _ in range(n_tests):
            dlc = random.randint(0, 8)
            data = bytes([random.randint(0, 255) for _ in range(dlc)])
            self.tx.send(can.Message(
                arbitration_id=can_id,
                data=data, is_extended_id=False))
            time.sleep(0.005)
            response = self._check_ecu_alive()
            self.results.append(FuzzResult(can_id, data, response))

    def fuzz_unknown_ids(self, n_tests: int = 500):
        """Send frames with IDs not in the DBC."""
        known_ids = {m.frame_id for m in self.db.messages}
        for _ in range(n_tests):
            can_id = random.randint(0, 0x7FF)
            if can_id not in known_ids:
                data = bytes([random.randint(0,255) for _ in range(8)])
                self.tx.send(can.Message(
                    arbitration_id=can_id, data=data))
                time.sleep(0.005)

    def _check_ecu_alive(self) -> str:
        """Check ECU heartbeat to detect crash or reset."""
        msg = self.rx.recv(timeout=0.1)
        if msg is None: return "silent"
        if msg.is_error_frame: return "error_frame"
        return "normal"