Vulnerability Scanning & Fuzzing

#!/usr/bin/env python3
# Vehicle network attack-surface discovery
import subprocess, socket, struct

def scan_doip_auth(vehicle_ip: str, port=13400) -> bool:
    # Check if DoIP accepts unauthenticated RoutingActivation -- critical finding if True
    try:
        s = socket.create_connection((vehicle_ip, port), timeout=3)
        req = struct.pack(">HHIB", 0xFFFE, 0x0005, 1, 0x00)
        s.sendall(req); resp = s.recv(256); s.close()
        return len(resp) > 0  # positive response = unauthenticated routing accepted
    except Exception:
        return False  # connection refused = likely requires TLS

def nmap_vehicle_ports(vehicle_ip: str) -> str:
    # Enumerate open ports: DoIP, SOME/IP-SD, optional debug services
    result = subprocess.run(
        ["nmap", "-sV", "-p", "13400,30490,443,22,8080,9090",
         "--script", "banner,tls-nextprotoneg", vehicle_ip],
        capture_output=True, text=True
    )
    return result.stdout

# Threat: open debug port (SSH/22 on vehicle Ethernet = Critical finding)
# Threat: DoIP without TLS = Critical finding (TS-09 countermeasure ineffective)
# result = scan_doip_auth("192.168.1.100")
# print("DoIP unauthenticated routing:", result)

#!/usr/bin/env python3
# UDS fuzzer: crash detection via ECU heartbeat monitoring
import can, random, time, struct

HEARTBEAT_ID      = 0x3C5    # ECU sends this ID every 100 ms when healthy
HEARTBEAT_TIMEOUT = 0.5      # 500 ms silence = ECU crash

def heartbeat_alive(bus, timeout=HEARTBEAT_TIMEOUT) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        msg = bus.recv(timeout=0.05)
        if msg and msg.arbitration_id == HEARTBEAT_ID: return True
    return False

def fuzz_uds(iface="vcan0", iterations=5000) -> list:
    bus = can.interface.Bus(iface, bustype="socketcan")
    crashes = []

    for i in range(iterations):
        sid = random.randint(0x00, 0xFF)
        # Mix: random, boundary values, known-dangerous lengths
        length = random.choice([0, 1, 7, 8, 255, 4095, random.randint(0, 65535)])
        payload = bytes([sid]) + random.randbytes(min(length, 7))
        frame = can.Message(arbitration_id=0x7DF, data=payload[:8], is_extended_id=False)
        bus.send(frame)
        time.sleep(0.005)

        if i % 100 == 0 and not heartbeat_alive(bus):
            crashes.append({"iteration": i, "payload": payload.hex()})
            print(f"CRASH at iteration {i}: payload={payload.hex()}")

    print(f"Complete: {iterations} iterations, {len(crashes)} crashes")
    return crashes

#!/usr/bin/env python3
# Safety-signal targeted CAN fuzzer (from DBC analysis)
import can, random, time

SAFETY_IDS = {
    0x200: "EngTorqueCmd",
    0x201: "BrakePressure",
    0x300: "SteeringAngle",
}

def fuzz_safety_signals(iface="vcan0", duration_s=60) -> list:
    bus = can.interface.Bus(iface, bustype="socketcan")
    anomalies = []
    end = time.time() + duration_s

    while time.time() < end:
        can_id = random.choice(list(SAFETY_IDS.keys()))
        # Structured payloads: minimum, maximum, random, alternating
        for payload in [bytes(8), bytes([0xFF]*8), random.randbytes(8),
                        bytes([0x00,0xFF]*4)]:
            frame = can.Message(arbitration_id=can_id, data=payload, is_extended_id=False)
            bus.send(frame)
            time.sleep(0.001)

            # Check for unexpected actuator responses
            resp = bus.recv(timeout=0.005)
            if resp and resp.arbitration_id not in SAFETY_IDS:
                anomalies.append({
                    "injected_id": hex(can_id),
                    "response_id": hex(resp.arbitration_id),
                    "response_data": resp.data.hex(),
                })

    print(f"Anomalies detected: {len(anomalies)}")
    return anomalies

#!/usr/bin/env python3
from boofuzz import Session, Target, SocketConnection
from boofuzz import s_initialize, s_block, s_word, s_dword, s_bytes

def configure_someip_fuzzer(target_ip="192.168.1.100", port=30490):
    target  = Target(connection=SocketConnection(target_ip, port, proto="udp"))
    session = Session(target=target)

    s_initialize("SOMEIP_Request")
    with s_block("Header"):
        s_word(0x0123, name="ServiceID",   fuzzable=True)
        s_word(0x8001, name="MethodID",    fuzzable=True)
        s_dword(0x0008, name="Length",     fuzzable=True)   # include off-by-one values
        s_word(0x0001, name="ClientID",    fuzzable=False)
        s_word(0x0001, name="SessionID",   fuzzable=False)
        s_bytes(b"\x01\x01\x01\x00",  name="Ctrl",      fuzzable=False)
    with s_block("Payload"):
        s_bytes(b"\x00"*16, name="Data",  fuzzable=True)

    session.connect(s_get("SOMEIP_Request"))

    # Crash detection: monitor target availability between mutations
    # session.fuzz()  # start campaign; crashes saved to boofuzz-results/
    return session

session = configure_someip_fuzzer()
print("Boofuzz SOME/IP session configured: run session.fuzz() to start campaign")
print("Each crash generates a minimised reproducer + call stack in boofuzz-results/")