From 1d9e1e043ef2f0e948ede901260843afcb89728a Mon Sep 17 00:00:00 2001 From: Ahnaf Shahriar Date: Sun, 11 Oct 2026 02:29:42 -0700 Subject: [PATCH 1/2] Add good and malicious scripted peers for UART, I2C, SPI, CAN, Ethernet Moves the three existing peers into examples/peers//, adds a per-bus scenario, a matrix README, and examples/tests/test_peer_mocks.py which drives every peer through its callbacks with a mock ctx (no firmware, no sim). Co-Authored-By: Claude Sonnet 5.5 --- docs/writing-tests.md | 2 +- examples/README.md | 40 +- examples/peers/can/good_dronecan.py | 69 ++++ examples/peers/can/malicious_echo.py | 33 ++ examples/peers/can/scenario.yaml | 18 + examples/peers/eth/good_arp_udp.py | 59 +++ examples/peers/eth/malicious_frames.py | 91 +++++ examples/peers/eth/scenario.yaml | 18 + .../good_icp20100.py} | 0 examples/peers/i2c/good_ina228.py | 99 +++++ examples/peers/i2c/malicious_tf02.py | 65 +++ examples/peers/i2c/scenario.yaml | 15 + examples/peers/scenario.yaml | 6 +- .../good_icm42688.py} | 0 examples/peers/spi/malicious_icm42688.py | 60 +++ examples/peers/spi/scenario.yaml | 15 + examples/peers/uart/good_crsf.py | 40 ++ .../peers/{gps_peer.py => uart/good_gps.py} | 0 examples/peers/uart/malicious_uart.py | 78 ++++ examples/peers/uart/scenario.yaml | 18 + examples/tests/test_peer_mocks.py | 375 ++++++++++++++++++ examples/tests/test_peers.py | 4 +- 22 files changed, 1094 insertions(+), 11 deletions(-) create mode 100644 examples/peers/can/good_dronecan.py create mode 100644 examples/peers/can/malicious_echo.py create mode 100644 examples/peers/can/scenario.yaml create mode 100644 examples/peers/eth/good_arp_udp.py create mode 100644 examples/peers/eth/malicious_frames.py create mode 100644 examples/peers/eth/scenario.yaml rename examples/peers/{icp20100_peer.py => i2c/good_icp20100.py} (100%) create mode 100644 examples/peers/i2c/good_ina228.py create mode 100644 examples/peers/i2c/malicious_tf02.py create mode 100644 examples/peers/i2c/scenario.yaml rename examples/peers/{icm42688_peer.py => spi/good_icm42688.py} (100%) create mode 100644 examples/peers/spi/malicious_icm42688.py create mode 100644 examples/peers/spi/scenario.yaml create mode 100644 examples/peers/uart/good_crsf.py rename examples/peers/{gps_peer.py => uart/good_gps.py} (100%) create mode 100644 examples/peers/uart/malicious_uart.py create mode 100644 examples/peers/uart/scenario.yaml create mode 100644 examples/tests/test_peer_mocks.py diff --git a/docs/writing-tests.md b/docs/writing-tests.md index 0912087..a92da82 100644 --- a/docs/writing-tests.md +++ b/docs/writing-tests.md @@ -61,7 +61,7 @@ def run(): ```python def test_gps_fix_reaches_mavlink(run): msgs, _ = run - assert last(msgs, GPS_RAW_INT)[28] >= 3 # gps_peer reports a 3D fix + assert last(msgs, GPS_RAW_INT)[28] >= 3 # good_gps reports a 3D fix def test_baro_is_read_over_i2c(run): _, bus = run diff --git a/examples/README.md b/examples/README.md index 1e2dd37..2947ecc 100644 --- a/examples/README.md +++ b/examples/README.md @@ -4,11 +4,41 @@ | --- | --- | | `step_and_peek.py` | Plain script, no pytest: boot a board, `expect()` a line, read memory, step time. `python step_and_peek.py fw.elf STM32F401RE` | | `parallel_sweep.py` | Run the same firmware N times in parallel and collect when each reaches a line. | -| `peers/gps_peer.py` | u-blox UBX GPS on a UART: ACKs CFG-VALSET, streams NAV-PVT at the rate the firmware programs. | -| `peers/icm42688_peer.py` | TDK ICM-42688-P IMU on SPI: registers, FIFO, ODR from the firmware's config. | -| `peers/icp20100_peer.py` | TDK ICP-20100 barometer on I2C: register file and FIFO. | -| `peers/scenario.yaml` | Wires the three peers to a machine. Needs your own `firmware.elf`; see the header. | -| `tests/test_peers.py` | pytest suite over those peers: one shared run per module, strict `xfail`. See `docs/writing-tests.md`. | +| `tests/test_peers.py` | pytest suite over the three-peer `peers/scenario.yaml`: one shared run per module, strict `xfail`. Needs your firmware; see `docs/writing-tests.md`. | +| `tests/test_peer_mocks.py` | Self-test of every peer below with a mock `ctx`: no firmware, no simulator. `pytest examples/tests/test_peer_mocks.py` | + +## Scripted peers: one good and one malicious per bus + +A good peer follows the part's spec; a malicious one injects faults so the +firmware's validation, timeouts and recovery are exercised. Each file's header +says what it models, which datasheet or RFC it follows and, for malicious +ones, the exact faults and what a robust firmware should do. Each bus directory +has a `scenario.yaml` that runs the good and the malicious peer as two machines +against your own `firmware.elf`. + +| Bus | Good peer | Malicious peer | +| --- | --- | --- | +| UART | `peers/uart/good_gps.py`: u-blox UBX GPS, ACKs CFG-VALSET, streams NAV-PVT at the rate the firmware programs. `peers/uart/good_crsf.py`: CRSF RC frames at 250 Hz, crc8 DVB-S2, 420 kbaud. | `peers/uart/malicious_uart.py`: CRSF-style link with bad CRC, truncated frames, garbage, bad length, a late then missing reply, then permanent silence. | +| I2C | `peers/i2c/good_icp20100.py`: TDK ICP-20100 barometer, register file and FIFO. `peers/i2c/good_ina228.py`: TI INA228 power monitor, conversion time from the firmware's ADC_CONFIG and SHUNT_CAL. | `peers/i2c/malicious_tf02.py`: TF02-Pro-style LiDAR with a wrong ID, bad checksum, bad header, out-of-range and stuck readings. | +| SPI | `peers/spi/good_icm42688.py`: TDK ICM-42688-P IMU, registers, FIFO, ODR from the firmware's config. | `peers/spi/malicious_icm42688.py`: wrong WHO_AM_I, FIFO that reports a count but returns empty packets, stuck gyro, bit-flipped accel. | +| CAN | `peers/can/good_dronecan.py`: DroneCAN anonymous node-ID allocation, then 1 Hz NodeStatus. | `peers/can/malicious_echo.py`: echoes each frame with a flipped last byte, wrong ID, short DLC, long DLC, then once clean. | +| Ethernet | `peers/eth/good_arp_udp.py`: answers ARP, echoes UDP on port 7 with correct IPv4 and UDP checksums. | `peers/eth/malicious_frames.py`: bad FCS, runt, foreign MAC, bad IPv4 checksum, bad UDP checksum, between good frames. | + +`peers/scenario.yaml` wires the GPS, IMU and barometer peers to one machine +(used by `tests/test_peers.py`). + +Limits to know about: + +- I2C: scripted slaves cannot yet NACK a data byte or stretch SCL mid-transfer + , so the I2C malicious peer does not inject either. + The CLI guide describes an `on_address` hook that returns False + to NACK the address; these examples do not use or test it. +- Ethernet: the bad-FCS frame uses `ctx.send_wire`, present in the pyrite + engine's peer only; the malicious peer skips that frame (with a warning) where + it is missing. The addresses in `malicious_frames.py` (`DUT_MAC`, `DUT_IP`) + are constants to set to your firmware's. +- A peer's `ctx` has a single timer, so peers that need several delays count + ticks of one periodic timer. No firmware is shipped. The peers hold their expected values as constants at the top of each file; the tests read those same numbers. diff --git a/examples/peers/can/good_dronecan.py b/examples/peers/can/good_dronecan.py new file mode 100644 index 0000000..b7a4bf2 --- /dev/null +++ b/examples/peers/can/good_dronecan.py @@ -0,0 +1,69 @@ +# DroneCAN (UAVCAN v0) node as a scripted CAN peer: anonymous dynamic node-ID +# allocation, then 1 Hz NodeStatus heartbeats from the allocated ID. +# Spec: DroneCAN / UAVCAN v0.9 specification: uavcan.protocol.dynamic_node_id +# .Allocation (data type ID 1) and uavcan.protocol.NodeStatus (ID 341). +# message id = prio<<24 | type<<8 | src (src 1..127) +# anonymous = prio<<24 | discriminator<<10 | (type&3)<<8 | 0 +# tail byte = start<<7 | end<<6 | toggle<<5 | transfer_id (5 bits) +# Allocation is a 3-stage handshake: the node sends 6, 6 and 4 bytes of its +# 16-byte unique ID, each request carrying the preferred node ID in bits 7..1 +# of byte 0 (bit 0 = "first part"); the allocator answers with the unique-ID +# bytes it has so far, and with the full 16 once it grants an ID. +# The allocator is expected to be the firmware at node ID 1. +NODESTATUS_ID = 341 +ALLOCATION_ID = 1 +PRIO_LOW = 24 +ALLOCATOR = 1 +UNIQUE_ID = bytes(range(0x50, 0x60)) +PREFERRED = 42 +STAGES = (UNIQUE_ID[0:6], UNIQUE_ID[6:12], UNIQUE_ID[12:16]) +BOOT_DELAY_US = 3_000_000 # give the firmware time to bring its CAN controller up +RETRY_US = 1_000_000 + + +def msg_id(dtype, src, prio=PRIO_LOW): + return prio << 24 | dtype << 8 | src + + +def anon_id(dtype, discr=0x1234, prio=PRIO_LOW): + return prio << 24 | discr << 10 | (dtype & 3) << 8 + + +class Node: + def __init__(self, ctx): + self.ctx = ctx + self.node_id = self.stage = self.tid = self.uptime = 0 + ctx.schedule_oneshot(BOOT_DELAY_US) + + def tail(self): + t = 0xC0 | (self.tid & 0x1F) # single-frame transfer: start|end, toggle 0 + self.tid += 1 + return t + + def on_timer(self): + if self.node_id: + self.uptime += 1 + # uptime u32 LE, health/mode/sub_mode byte (0 = OK/OPERATIONAL), vendor-specific u16 + payload = self.uptime.to_bytes(4, "little") + b"\x00" + b"\x00\x00" + self.ctx.send(msg_id(NODESTATUS_ID, self.node_id), payload + bytes([self.tail()]), extended=True) + return + self.stage = 0 + self._send_stage() + self.ctx.schedule_oneshot(RETRY_US) # retry if the allocator stays silent + + def _send_stage(self): + head = bytes([(PREFERRED << 1) | (1 if self.stage == 0 else 0)]) + self.ctx.send(anon_id(ALLOCATION_ID), head + STAGES[self.stage] + bytes([self.tail()]), extended=True) + + def on_frame(self, frame): + cid, data = frame["id"], bytes(frame["data"]) + src, dtype = cid & 0x7F, (cid >> 8) & 0xFFFF + if src != ALLOCATOR or dtype != ALLOCATION_ID or not data or not data[-1] & 0x80: + return + allocated = (data[0] if data[-1] & 0x40 else data[2]) >> 1 # first frame of a multi-frame transfer has 2 CRC bytes + if allocated: + self.node_id, self.uptime = allocated, 0 + self.ctx.schedule_periodic(1_000_000) + elif self.stage < 2: + self.stage += 1 + self._send_stage() diff --git a/examples/peers/can/malicious_echo.py b/examples/peers/can/malicious_echo.py new file mode 100644 index 0000000..c072bb6 --- /dev/null +++ b/examples/peers/can/malicious_echo.py @@ -0,0 +1,33 @@ +# Hostile CAN echo peer. Echoes every frame the firmware sends, but corrupts +# it in a rotating way so the firmware's validation of replies is exercised. +# A well-behaved peer never does this; a scripted one does it deterministically. +# echo 0 (mod 5) last data byte flipped (^0xFF) -> payload compare must fail +# echo 1 identifier + 1 -> reply must be rejected as +# from the wrong ID +# echo 2 DLC shortened by one byte -> length check must fail +# echo 3 one extra payload byte (DLC+1) -> length check must fail +# (classic CAN: capped at 8) +# echo 4 echoed unchanged -> the control: accepted +# Frames with an empty payload are echoed unchanged. A robust firmware checks +# ID, DLC and payload of every reply and treats a mismatch as a bus error, +# not as data. +class Node: + def __init__(self, ctx): + self.ctx = ctx + self.n = 0 + + def on_frame(self, frame): + cid, data, ext = frame["id"], bytearray(frame["data"]), bool(frame.get("extended")) + mode = self.n % 5 + self.n += 1 + if data: + if mode == 0: + data[-1] ^= 0xFF + elif mode == 2: + del data[-1] + elif mode == 3 and len(data) < 8: + data.append(0xA5) + if mode == 1: + cid += 1 + self.ctx.info(f"echo #{self.n} mode {mode}: id=0x{cid:x} len={len(data)}") + self.ctx.send(cid, bytes(data), extended=ext) diff --git a/examples/peers/can/scenario.yaml b/examples/peers/can/scenario.yaml new file mode 100644 index 0000000..9659395 --- /dev/null +++ b/examples/peers/can/scenario.yaml @@ -0,0 +1,18 @@ +# CAN: the good and the malicious peer on the same wiring, as two machines running +# YOUR firmware (not shipped). Run from this directory: +# sim --scenario scenario.yaml --timeout 12 +# Change mcu, bus, pins and baud to match your firmware. +machines: + good: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: node, type: can-node, script: good_dronecan.py } + malicious: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: node, type: can-node, script: malicious_echo.py } +media: + - { type: can, connect: [good.fdcan1, good.node] } + - { type: can, connect: [malicious.fdcan1, malicious.node] } diff --git a/examples/peers/eth/good_arp_udp.py b/examples/peers/eth/good_arp_udp.py new file mode 100644 index 0000000..ab7f200 --- /dev/null +++ b/examples/peers/eth/good_arp_udp.py @@ -0,0 +1,59 @@ +# Minimal Ethernet host as a scripted eth-peer: answers ARP requests for its +# IP and echoes UDP datagrams sent to it, with correct checksums. +# Specs: RFC 826 (ARP), RFC 791 (IPv4 header checksum, 16-bit one's +# complement), RFC 768 (UDP, checksum over the pseudo-header; a computed 0 is +# sent as 0xFFFF), RFC 862 (echo). `frame` handed to on_frame has `bytes` +# (without FCS); ctx.send takes bytes from the destination MAC on and the +# medium appends the FCS. The peer's MAC is the eth-peer's `mac:`. +import struct + +PEER_IP = bytes((192, 168, 1, 1)) +UDP_PORT = 7 + + +def inet_sum(data): + if len(data) % 2: + data += b"\0" + s = sum(struct.unpack(f">{len(data) // 2}H", data)) + while s >> 16: + s = (s & 0xFFFF) + (s >> 16) + return s + + +def udp_checksum(src, dst, udp): + s = inet_sum(src + dst + struct.pack(">BBH", 0, 17, len(udp)) + udp) + return ((~s) & 0xFFFF) or 0xFFFF + + +def ipv4_udp(src_ip, dst_ip, sport, dport, payload): + udp = struct.pack(">HHHH", sport, dport, 8 + len(payload), 0) + payload + udp = udp[:6] + struct.pack(">H", udp_checksum(src_ip, dst_ip, udp)) + udp[8:] + hdr = struct.pack(">BBHHHBBH", 0x45, 0, 20 + len(udp), 0, 0x4000, 64, 17, 0) + src_ip + dst_ip + return hdr[:10] + struct.pack(">H", (~inet_sum(hdr)) & 0xFFFF) + hdr[12:] + udp + + +def eth(dst, src, ethertype, body): + f = dst + src + struct.pack(">H", ethertype) + body + return f + bytes(max(0, 60 - len(f))) # pad to the 60-byte minimum (without FCS) + + +class Peer: + def __init__(self, ctx): + self.ctx = ctx + self.mac = bytes(int(b, 16) for b in ctx.mac.split(":")) + + def on_frame(self, frame): + f = bytes(frame["bytes"]) + if len(f) < 42: + return + src_mac, ethertype = f[6:12], f[12:14] + if ethertype == b"\x08\x06" and f[20:22] == b"\x00\x01" and f[38:42] == PEER_IP: # ARP request for us + reply = struct.pack(">HHBBH", 1, 0x0800, 6, 4, 2) + self.mac + PEER_IP + f[22:28] + f[28:32] + self.ctx.send(eth(src_mac, self.mac, 0x0806, reply)) + elif ethertype == b"\x08\x00" and f[14] == 0x45 and f[23] == 17 and f[30:34] == PEER_IP: + if inet_sum(f[14:34]) != 0xFFFF: # IPv4 header checksum must verify + return + sport, dport, ulen = struct.unpack(">HHH", f[34:40]) + if dport != UDP_PORT or ulen < 8: + return + self.ctx.send(eth(src_mac, self.mac, 0x0800, ipv4_udp(PEER_IP, f[26:30], dport, sport, f[42:34 + ulen]))) diff --git a/examples/peers/eth/malicious_frames.py b/examples/peers/eth/malicious_frames.py new file mode 100644 index 0000000..ee8f940 --- /dev/null +++ b/examples/peers/eth/malicious_frames.py @@ -0,0 +1,91 @@ +# Fault-injecting Ethernet peer: after BOOT_US it sends one frame per +# millisecond to the DUT, a fixed sequence of good frames and faults. Each +# frame is an IPv4/UDP datagram to DUT_IP:UDP_PORT whose payload's first byte +# is the frame's marker, so a test can tell which ones got through. +# 1 valid -> must be delivered +# 2 bad FCS (CRC-32 inverted) -> MAC must drop it (CRC error) +# 3 runt: 30 bytes + FCS -> MAC must drop it (under 64 on the wire) +# 4 foreign destination MAC -> address filter must drop it +# 5 bad IPv4 header checksum -> IP stack must drop it +# 6 bad UDP checksum -> UDP layer must drop it +# 7 valid -> must be delivered (the stack recovered) +# Expected: markers 1 and 7 reach the application, 2-6 do not, and the +# firmware does not hang, leak buffers or crash on the bad ones. +# Frame 2 uses ctx.send_wire (we supply the FCS), which the pyrite engine's +# scripted peer has; the C# peer's ctx has no send_wire, so there frame 2 is +# skipped with a warning. +# Spec: IEEE 802.3 (FCS = CRC-32, sent least-significant byte first), RFC 791, +# RFC 768. +import struct +import zlib + +DUT_MAC = bytes.fromhex("020000000001") # set to the MAC the firmware programs +OTHER_MAC = bytes.fromhex("020000000077") +DUT_IP = bytes((192, 168, 1, 10)) # set to the IP the firmware uses +PEER_IP = bytes((192, 168, 1, 1)) +UDP_PORT = 9000 +BOOT_US = 1_500_000 + + +def inet_sum(data): + if len(data) % 2: + data += b"\0" + s = sum(struct.unpack(f">{len(data) // 2}H", data)) + while s >> 16: + s = (s & 0xFFFF) + (s >> 16) + return s + + +def datagram(marker, bad_ip=False, bad_udp=False): + payload = bytes([marker]) + b"simantic" + bytes(8) + udp = struct.pack(">HHHH", 40000, UDP_PORT, 8 + len(payload), 0) + payload + ck = ((~inet_sum(PEER_IP + DUT_IP + struct.pack(">BBH", 0, 17, len(udp)) + udp)) & 0xFFFF) or 0xFFFF + udp = udp[:6] + struct.pack(">H", ck ^ (0x1111 if bad_udp else 0)) + udp[8:] + hdr = struct.pack(">BBHHHBBH", 0x45, 0, 20 + len(udp), marker, 0x4000, 64, 17, 0) + PEER_IP + DUT_IP + ip_ck = ((~inet_sum(hdr)) & 0xFFFF) ^ (0x1111 if bad_ip else 0) + return hdr[:10] + struct.pack(">H", ip_ck) + hdr[12:] + udp + + +def eth(dst, src, body): + f = dst + src + b"\x08\x00" + body + return f + bytes(max(0, 60 - len(f))) + + +def fcs(data): + return zlib.crc32(data).to_bytes(4, "little") + + +class Peer: + def __init__(self, ctx): + self.ctx = ctx + self.mac = bytes(int(b, 16) for b in ctx.mac.split(":")) + self.step = 0 + ctx.schedule_oneshot(BOOT_US) + + def on_frame(self, frame): + pass + + def on_timer(self): + c, m = self.ctx, self.mac + if self.step == 0: + c.send(eth(DUT_MAC, m, datagram(1))) + elif self.step == 1: + f = eth(DUT_MAC, m, datagram(2)) + if hasattr(c, "send_wire"): + c.send_wire(f + bytes(b ^ 0xFF for b in fcs(f))) + else: + c.warning("send_wire unavailable on this engine: bad-FCS frame skipped") + elif self.step == 2: + c.send((DUT_MAC + m + b"\x08\x00" + datagram(3))[:30]) + elif self.step == 3: + c.send(eth(OTHER_MAC, m, datagram(4))) + elif self.step == 4: + c.send(eth(DUT_MAC, m, datagram(5, bad_ip=True))) + elif self.step == 5: + c.send(eth(DUT_MAC, m, datagram(6, bad_udp=True))) + elif self.step == 6: + c.send(eth(DUT_MAC, m, datagram(7))) + else: + return + self.step += 1 + c.schedule_oneshot(1000) diff --git a/examples/peers/eth/scenario.yaml b/examples/peers/eth/scenario.yaml new file mode 100644 index 0000000..8e2585d --- /dev/null +++ b/examples/peers/eth/scenario.yaml @@ -0,0 +1,18 @@ +# Ethernet: the good and the malicious peer on the same wiring, as two machines running +# YOUR firmware (not shipped). Run from this directory: +# sim --scenario scenario.yaml --timeout 12 +# Change mcu, bus, pins and baud to match your firmware. +machines: + good: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: host, type: eth-peer, mac: "02:00:00:00:00:99", script: good_arp_udp.py } + malicious: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: host, type: eth-peer, mac: "02:00:00:00:00:99", script: malicious_frames.py } +media: + - { type: eth, connect: [good.ethernet, good.host] } + - { type: eth, connect: [malicious.ethernet, malicious.host] } diff --git a/examples/peers/icp20100_peer.py b/examples/peers/i2c/good_icp20100.py similarity index 100% rename from examples/peers/icp20100_peer.py rename to examples/peers/i2c/good_icp20100.py diff --git a/examples/peers/i2c/good_ina228.py b/examples/peers/i2c/good_ina228.py new file mode 100644 index 0000000..996750a --- /dev/null +++ b/examples/peers/i2c/good_ina228.py @@ -0,0 +1,99 @@ +# TI INA228 power monitor as a scripted I2C peer (7-bit 0x45 on A0/A1 = GND). +# Source: TI INA228 datasheet SLYS021A (rev. May 2022). +# Register file with an 8-bit pointer. A write arrives as [reg] then [data...] +# in one or two on_write calls and on_finish at STOP: accumulate, byte 0 is +# the pointer. +# +# Timing follows what the firmware programs: the conversion cycle is the sum of +# the enabled inputs' conversion times x averaging, from ADC_CONFIG; CURRENT +# and POWER scale from the SHUNT_CAL it writes (CURRENT_LSB, Eq. 2). Results +# read 0 until the first cycle completes; reading DIAG_ALRT clears CNVRF. +# ENERGY/CHARGE accumulators are modelled for time since conversions started. +PACK_V = 16.8 # V, bus voltage +LOAD_A = 12.0 # A, current through the shunt +DIE_C = 35.0 # degC +R_SHUNT = 0.0005 # ohm; the firmware's shunt value + +CT_US = (50, 84, 150, 280, 540, 1052, 2074, 4120) # per VBUSCT/VSHCT/VTCT code: SLYS021A Table 7-6 +AVG_N = (1, 4, 16, 64, 128, 256, 512, 1024) # per AVG code: Table 7-6 +CONVDLY_STEP_US = 2000 # CONFIG.CONVDLY step: Table 7-5 + +CONFIG, ADC_CONFIG, SHUNT_CAL, VBUS, DIETEMP, CURRENT, POWER, DIAG_ALRT = 0, 1, 2, 5, 6, 7, 8, 11 +RESET = {CONFIG: 0x0000, ADC_CONFIG: 0xFB68, SHUNT_CAL: 0x1000} # SLYS021A 7.6.1.1-3 + + +def be(v, n): + return (int(v) & ((1 << (8 * n)) - 1)).to_bytes(n, "big") + + +class Slave: + def __init__(self, ctx): + self.ctx = ctx + self.ptr = 0 + self.buf = b"" + self._reset() + + def _reset(self): + self.regs = dict(RESET) + self.t0 = self.ctx.now_us + self.seen = 0 + + def _cycle_us(self): + a = self.regs[ADC_CONFIG] + ct = (CT_US[(a >> 9) & 7], CT_US[(a >> 6) & 7], CT_US[(a >> 3) & 7]) # bus, shunt, temp + return sum(t for bit, t in zip((1, 2, 4), ct) if (a >> 12) & bit) * AVG_N[a & 7] + + def _cycles(self): + mode, cyc = self.regs[ADC_CONFIG] >> 12, self._cycle_us() + if mode & 7 == 0 or cyc == 0: + return 0 + start = self.t0 + ((self.regs[CONFIG] >> 6) & 0xFF) * CONVDLY_STEP_US + n = max(0, (self.ctx.now_us - start) // cyc) + return n if mode & 8 else min(n, 1) # triggered modes convert once + + def _current_lsb(self): + adcrange = (self.regs[CONFIG] >> 4) & 1 + return (self.regs[SHUNT_CAL] & 0x7FFF) / (13107.2e6 * R_SHUNT * (4 if adcrange else 1)) # Eq. 2 + + def _value(self, reg): + n = self._cycles() + if reg == DIAG_ALRT: + cnvrf = n > self.seen + self.seen = n + return be(0x0001 | cnvrf << 1, 2) # MEMSTAT=1, CNVRF + if reg in RESET: + return be(self.regs[reg], 2) + lsb = self._current_lsb() + if n == 0 or (reg in (CURRENT, POWER) and lsb == 0): + return bytes(2 if reg == DIETEMP else 3) + if reg == VBUS: + return be(int(PACK_V / 195.3125e-6) << 4, 3) # 195.3125 uV/LSB, 20 bit left-aligned + if reg == DIETEMP: + return be(int(DIE_C / 7.8125e-3), 2) # 7.8125 mdegC/LSB + if reg == CURRENT: + return be(int(LOAD_A / lsb) << 4, 3) # Eq. 4 + if reg == POWER: + return be(PACK_V * LOAD_A / (3.2 * lsb), 3) # Eq. 5 + return bytes(2) + + def on_write(self, data): + self.buf += bytes(data) + if not self.buf: + return + self.ptr = self.buf[0] + if len(self.buf) < 3 or self.ptr not in RESET: + return + v = int.from_bytes(self.buf[1:3], "big") + if self.ptr == CONFIG and v & 0x8000: # RST + self._reset() + return + if self.ptr == ADC_CONFIG: # writing MODE restarts conversions + self.t0, self.seen = self.ctx.now_us, 0 + self.regs[self.ptr] = v & 0x7FFF if self.ptr == CONFIG else v + + def on_finish(self): + self.buf = b"" + + def on_read(self, count): + self.buf = b"" + return (self._value(self.ptr) + bytes(count))[:count] diff --git a/examples/peers/i2c/malicious_tf02.py b/examples/peers/i2c/malicious_tf02.py new file mode 100644 index 0000000..977010d --- /dev/null +++ b/examples/peers/i2c/malicious_tf02.py @@ -0,0 +1,65 @@ +# Fault-injecting TF02-Pro-style LiDAR on I2C (7-bit 0x10). +# Good-case protocol: Benewake TF02-Pro User Manual (REV 01/04/2024): command +# frame 5A len id ... cksum (cksum = low byte of the sum of all prior bytes); +# "trigger data" (5A 05 00 01 60) is answered with a 9-byte frame +# 59 59 distL distH strL strH tempL tempH cksum; the reply to the last command +# is served on the next read. +# +# Faults (the phase is the number of trigger commands the firmware has sent, +# modulo 5, so each poll gets the next fault): +# 0 valid frame, 1250 cm -> accept +# 1 valid frame but checksum + 1 -> reject (bad checksum) +# 2 header 59 58 -> reject (bad header) +# 3 distance 0xFFFF, strength 0 -> out of range / no return: ignore +# 4 the identical valid frame again -> stuck sensor: the same value for +# every following poll in this phase; detect "never changes" +# Version query (5A 04 01 5F) replies with a frame whose ID byte is wrong +# (0xEE instead of 0x01), so a probe that checks the device identity must fail. +# I2C limits: peers cannot yet NACK a data byte or stretch SCL mid-transfer, +# so none of that is injected here. +# A robust firmware checks header and checksum, range-checks distance and +# strength, and flags a value that never changes. +GOOD_CM = 1250 +TEMP_RAW = (25 + 256) * 8 + + +def cksum(b): + return sum(b) & 0xFF + + +def data_frame(dist, strength, head=b"\x59\x59", bump=0): + body = head + dist.to_bytes(2, "little") + strength.to_bytes(2, "little") + TEMP_RAW.to_bytes(2, "little") + return body + bytes([(cksum(body) + bump) & 0xFF]) + + +def phase_frame(phase): + if phase == 0: + return data_frame(GOOD_CM, 3000) + if phase == 1: + return data_frame(GOOD_CM, 3000, bump=1) + if phase == 2: + return data_frame(GOOD_CM, 3000, head=b"\x59\x58") + if phase == 3: + return data_frame(0xFFFF, 0) + return data_frame(GOOD_CM, 3000) # stuck: identical frame every time + + +class Slave: + def __init__(self, ctx): + self.ctx = ctx + self.reply = b"" + self.triggers = 0 + + def on_write(self, data): + data = bytes(data) + if len(data) < 3 or data[0] != 0x5A: + return + if data[2] == 0x01: # version query: wrong ID byte + body = bytes([0x5A, 0x07, 0xEE, 0x02, 0x03, 0x01]) + self.reply = body + bytes([cksum(body)]) + elif data[2] == 0x00: # trigger + self.reply = phase_frame(self.triggers % 5) + self.triggers += 1 + + def on_read(self, count): + return (self.reply + bytes(count))[:count] diff --git a/examples/peers/i2c/scenario.yaml b/examples/peers/i2c/scenario.yaml new file mode 100644 index 0000000..306b692 --- /dev/null +++ b/examples/peers/i2c/scenario.yaml @@ -0,0 +1,15 @@ +# I2C: the good and the malicious peer on the same wiring, as two machines running +# YOUR firmware (not shipped). Run from this directory: +# sim --scenario scenario.yaml --timeout 12 +# Change mcu, bus, pins and baud to match your firmware. +machines: + good: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: pwr, type: i2c-device, bus: i2c1, address: 0x45, script: good_ina228.py } + malicious: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: lidar, type: i2c-device, bus: i2c3, address: 0x10, script: malicious_tf02.py } diff --git a/examples/peers/scenario.yaml b/examples/peers/scenario.yaml index cd61924..371f512 100644 --- a/examples/peers/scenario.yaml +++ b/examples/peers/scenario.yaml @@ -8,9 +8,9 @@ machines: mcu: STM32H753IITX elf: firmware.elf parts: - - { name: gps, type: uart-device, baud: 230400, script: gps_peer.py } - - { name: imu, type: spi-device, bus: spi1, cs: PC4, script: icm42688_peer.py } - - { name: baro, type: i2c-device, bus: i2c2, address: 0x63, script: icp20100_peer.py } + - { name: gps, type: uart-device, baud: 230400, script: uart/good_gps.py } + - { name: imu, type: spi-device, bus: spi1, cs: PC4, script: spi/good_icm42688.py } + - { name: baro, type: i2c-device, bus: i2c2, address: 0x63, script: i2c/good_icp20100.py } media: - type: uart connect: [dut.usart6, dut.gps] diff --git a/examples/peers/icm42688_peer.py b/examples/peers/spi/good_icm42688.py similarity index 100% rename from examples/peers/icm42688_peer.py rename to examples/peers/spi/good_icm42688.py diff --git a/examples/peers/spi/malicious_icm42688.py b/examples/peers/spi/malicious_icm42688.py new file mode 100644 index 0000000..7bb84d8 --- /dev/null +++ b/examples/peers/spi/malicious_icm42688.py @@ -0,0 +1,60 @@ +# Fault-injecting ICM-42688-P-like SPI peer. Follows the register framing of +# good_icm42688.py (TDK DS-000347 rev 1.2: beat 0 = register byte, bit7 = +# read; beats 1.. = data) but is deliberately small and self-contained. +# +# Faults, all deterministic: +# WHO_AM_I (0x75) reads 0x42 instead of 0x47 -> the driver must +# refuse to bring the part up. +# FIFO_COUNTH/L always reports FIFO_COUNT = 4 records, but FIFO_DATA +# returns empty packets (header 0x80 = FIFO empty, +# DS-000347 6.2) -> the driver must not treat them as +# samples; it must stop reading and not loop forever. +# GYRO_DATA_X1 (0x25) returns the stuck value 0x1234 on all three axes, +# every read -> a plausibility/"never changes" check. +# ACCEL_DATA_X1 (0x1F) returns a real-looking sample with bit 14 of each +# axis flipped on every second read -> a plausibility +# or consecutive-sample check. +# A robust driver checks WHO_AM_I, validates FIFO headers, range-checks and +# change-checks samples, and has a bounded read loop. +WHO_AM_I, FIFO_COUNTH, FIFO_DATA, GYRO_DATA_X1, ACCEL_DATA_X1 = 0x75, 0x2E, 0x30, 0x25, 0x1F +BAD_WHO_AM_I = 0x42 +FIFO_COUNT = 4 +STUCK_GYRO = 0x1234 +ACCEL_G_LSB = 2048 # +-16 g: 2048 LSB/g; stationary, 1 g on Z + + +class Slave: + def __init__(self, ctx): + self.ctx = ctx + self.beat, self.reg, self.read, self.out = 0, 0, False, b"" + self.accel_reads = 0 + + def _value(self, reg): + if reg == WHO_AM_I: + return bytes([BAD_WHO_AM_I]) + if reg == FIFO_COUNTH: + return FIFO_COUNT.to_bytes(2, "big") + if reg == GYRO_DATA_X1: + return STUCK_GYRO.to_bytes(2, "big") * 3 + if reg == ACCEL_DATA_X1: + self.accel_reads += 1 + z = ACCEL_G_LSB ^ (0x4000 if self.accel_reads % 2 == 0 else 0) + return (0).to_bytes(2, "big") * 2 + z.to_bytes(2, "big") + return b"\x00" + + def on_transmit(self, b): + i = self.beat + self.beat += 1 + if i == 0: + self.read, self.reg = bool(b & 0x80), b & 0x7F + if self.read: + self.out = self._value(self.reg) + return 0x00 + if not self.read: + return 0x00 + if self.reg == FIFO_DATA: + return 0x80 if (i - 1) % 16 == 0 else 0x00 # empty packet + return self.out[i - 1] if i - 1 < len(self.out) else 0x00 + + def on_finish(self): + self.beat = 0 diff --git a/examples/peers/spi/scenario.yaml b/examples/peers/spi/scenario.yaml new file mode 100644 index 0000000..597e210 --- /dev/null +++ b/examples/peers/spi/scenario.yaml @@ -0,0 +1,15 @@ +# SPI: the good and the malicious peer on the same wiring, as two machines running +# YOUR firmware (not shipped). Run from this directory: +# sim --scenario scenario.yaml --timeout 12 +# Change mcu, bus, pins and baud to match your firmware. +machines: + good: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: imu, type: spi-device, bus: spi1, cs: PC4, script: good_icm42688.py } + malicious: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: imu, type: spi-device, bus: spi1, cs: PC4, script: malicious_icm42688.py } diff --git a/examples/peers/uart/good_crsf.py b/examples/peers/uart/good_crsf.py new file mode 100644 index 0000000..cf36d09 --- /dev/null +++ b/examples/peers/uart/good_crsf.py @@ -0,0 +1,40 @@ +# TBS Crossfire / ExpressLRS receiver as a scripted UART peer (420000 8N1). +# Spec: CRSF protocol (Team BlackSheep / ExpressLRS CRSF documentation): +# frame = address(0xC8 flight controller) | len | type | payload | crc8 +# len = type + payload + crc bytes; crc8 = poly 0xD5 (DVB-S2), init 0, +# over type + payload. +# RC_CHANNELS_PACKED (type 0x16): 16 channels x 11 bits, little-endian +# bit-packed in 22 bytes; 172 = 988 us, 992 = 1500 us, 1811 = 2012 us. +# Behaviour: streams one RC frame every 4 ms (250 Hz link rate). Sticks +# centred, throttle low, channel 5 high. The firmware never talks to it. +CHANNELS = [992, 992, 172, 992, 1811, 172, 172, 172, 992, 992, 992, 992, 992, 992, 992, 992] +PERIOD_US = 4000 + + +def crc8(data): + crc = 0 + for b in data: + crc ^= b + for _ in range(8): + crc = ((crc << 1) ^ 0xD5) & 0xFF if crc & 0x80 else (crc << 1) & 0xFF + return crc + + +def rc_frame(ch): + bits = 0 + for i, v in enumerate(ch): + bits |= (v & 0x7FF) << (11 * i) + body = b"\x16" + bits.to_bytes(22, "little") + return bytes([0xC8, len(body) + 1]) + body + bytes([crc8(body)]) + + +class Peer: + def __init__(self, ctx): + self.ctx = ctx + ctx.schedule_periodic(PERIOD_US) + + def on_byte(self, b): + pass + + def on_timer(self): + self.ctx.send(rc_frame(CHANNELS)) diff --git a/examples/peers/gps_peer.py b/examples/peers/uart/good_gps.py similarity index 100% rename from examples/peers/gps_peer.py rename to examples/peers/uart/good_gps.py diff --git a/examples/peers/uart/malicious_uart.py b/examples/peers/uart/malicious_uart.py new file mode 100644 index 0000000..d5eae09 --- /dev/null +++ b/examples/peers/uart/malicious_uart.py @@ -0,0 +1,78 @@ +# Fault-injecting UART peer: a CRSF-style link that misbehaves on a schedule. +# Models a flaky receiver or a noisy wire, so firmware frame validation and +# receive timeouts get exercised. Framing is the CRSF one (see good_crsf.py): +# 0xC8 | len | type | payload | crc8 (poly 0xD5). +# +# Fault schedule, one frame per 4 ms tick (cycle repeats every 6 ticks): +# tick 0 valid frame -> must be accepted +# tick 1 valid frame, crc byte inverted -> must be rejected (CRC mismatch) +# tick 2 valid frame cut after 10 bytes -> must not be accepted; the +# parser must resync on the next +# 0xC8 and not wait for 26 bytes +# tick 3 8 bytes of 0xFF/0x00 garbage -> must be skipped byte-wise +# tick 4 frame with len field 0xFF -> must be rejected (CRSF max is 64) +# tick 5 silence -> link-loss timeout/failsafe +# After SILENCE_AFTER_TICKS ticks (2 s of virtual time) the peer stops sending +# for good: the firmware should declare link loss and fail safe. +# Request/reply: a '?' byte from the firmware is answered with '!' LATE_TICKS +# ticks later (300 ms, virtual time) for odd-numbered requests and never for +# even-numbered ones, so the firmware's reply timeout and retry are exercised. +# (The peer ctx has one timer, so delays are counted in ticks of it.) +# The sequence is the same every run (no randomness). +# A robust firmware validates CRC and length before using any field, never +# blocks on a partial frame, and times out when no valid frame arrives. +PERIOD_US = 4000 +SILENCE_AFTER_TICKS = 500 +LATE_TICKS = 75 + + +def crc8(data): + crc = 0 + for b in data: + crc ^= b + for _ in range(8): + crc = ((crc << 1) ^ 0xD5) & 0xFF if crc & 0x80 else (crc << 1) & 0xFF + return crc + + +def good_frame(): + body = b"\x16" + bytes([0x55]) * 22 + return bytes([0xC8, len(body) + 1]) + body + bytes([crc8(body)]) + + +def faulty(tick): + f = good_frame() + if tick == 0: + return f + if tick == 1: + return f[:-1] + bytes([f[-1] ^ 0xFF]) + if tick == 2: + return f[:10] + if tick == 3: + return bytes([0xFF, 0x00] * 4) + if tick == 4: + return bytes([0xC8, 0xFF]) + f[2:] + return b"" + + +class Peer: + def __init__(self, ctx): + self.ctx, self.tick, self.requests, self.reply_at = ctx, 0, 0, None + ctx.schedule_periodic(PERIOD_US) + + def on_byte(self, b): + if b == 0x3F: + self.requests += 1 + if self.requests % 2: + self.reply_at = self.tick + LATE_TICKS + + def on_timer(self): + if self.tick == self.reply_at: + self.ctx.send(b"!") + if self.tick >= SILENCE_AFTER_TICKS: + self.ctx.cancel_timer() + return + out = faulty(self.tick % 6) + self.tick += 1 + if out: + self.ctx.send(out) diff --git a/examples/peers/uart/scenario.yaml b/examples/peers/uart/scenario.yaml new file mode 100644 index 0000000..441e154 --- /dev/null +++ b/examples/peers/uart/scenario.yaml @@ -0,0 +1,18 @@ +# UART: the good and the malicious peer on the same wiring, as two machines running +# YOUR firmware (not shipped). Run from this directory: +# sim --scenario scenario.yaml --timeout 12 +# Change mcu, bus, pins and baud to match your firmware. +machines: + good: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: rc, type: uart-device, baud: 420000, script: good_crsf.py } + malicious: + mcu: STM32H753IITX + elf: firmware.elf + parts: + - { name: rc, type: uart-device, baud: 420000, script: malicious_uart.py } +media: + - { type: uart, connect: [good.uart4, good.rc] } + - { type: uart, connect: [malicious.uart4, malicious.rc] } diff --git a/examples/tests/test_peer_mocks.py b/examples/tests/test_peer_mocks.py new file mode 100644 index 0000000..0660236 --- /dev/null +++ b/examples/tests/test_peer_mocks.py @@ -0,0 +1,375 @@ +"""Peer self-tests: no firmware, no simulator. + +Each peer under ../peers is loaded by path and driven through the same callback +sequence the engine uses, with a mock `ctx`. Good peers are checked against +known-answer vectors computed here from the specs (not by calling the peer's +own helpers); malicious peers are checked to really produce their fault. + + pytest examples/tests/test_peer_mocks.py +""" +import importlib.util +import math +import struct +import zlib +from pathlib import Path + +PEERS = Path(__file__).resolve().parent.parent / "peers" + + +class Ctx: + mac = "02:00:00:00:00:99" + + def __init__(self): + self.now_us = 0 + self.sent = [] # (args, kwargs) of ctx.send + self.wire = [] # ctx.send_wire payloads + self.order = [] # "send"/"wire", in call order + self.timers = [] # ("periodic"|"oneshot"|"cancel", us) + + def send(self, *a, **k): + self.sent.append((a, k)) + self.order.append("send") + + def send_wire(self, data): + self.wire.append(bytes(data)) + self.order.append("wire") + + def schedule_periodic(self, us): + self.timers.append(("periodic", us)) + + def schedule_oneshot(self, us): + self.timers.append(("oneshot", us)) + + def cancel_timer(self): + self.timers.append(("cancel", 0)) + + def log(self, *a): + pass + + info = warning = error = debug = noisy = log + + +def load(rel, cls): + spec = importlib.util.spec_from_file_location(Path(rel).stem, PEERS / rel) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + ctx = Ctx() + return getattr(mod, cls)(ctx), ctx, mod + + +def payloads(ctx): + return [bytes(a[0]) for a, _ in ctx.sent] + + +# ---- independent reference implementations (from the specs) ---------------- + +def crc8_dvbs2(data): + table = [] + for i in range(256): + c = i + for _ in range(8): + c = ((c << 1) ^ 0xD5) & 0xFF if c & 0x80 else (c << 1) & 0xFF + table.append(c) + crc = 0 + for b in data: + crc = table[crc ^ b] + return crc + + +def fletcher8(data): + a = b = 0 + for x in data: + a = (a + x) % 256 + b = (b + a) % 256 + return bytes((a, b)) + + +def ones_complement_sum(data): + data = data + b"\0" * (len(data) % 2) + s = 0 + for i in range(0, len(data), 2): + s += data[i] << 8 | data[i + 1] + while s > 0xFFFF: + s = (s >> 16) + (s & 0xFFFF) + return s + + +def udp_ok(ip_frame): + """True if the IPv4 header and UDP checksums in an Ethernet frame verify.""" + ip, udp = ip_frame[14:34], ip_frame[34:] + pseudo = ip[12:20] + bytes((0, 17)) + udp[4:6] + return ones_complement_sum(ip) == 0xFFFF and ones_complement_sum(pseudo + udp) == 0xFFFF + + +# ---- UART ------------------------------------------------------------------ + +def test_gps_acks_valset_and_streams_pvt_at_programmed_rate(): + p, ctx, _ = load("uart/good_gps.py", "Peer") + + def valset(key, val): + body = bytes((0x06, 0x8A)) + (4 + 4 + 4).to_bytes(2, "little") + bytes((0, 1, 0, 0)) + struct.pack("> (11 * i)) & 0x7FF for i in range(16)] == mod.CHANNELS + + +def test_malicious_uart_faults_in_order(): + p, ctx, mod = load("uart/malicious_uart.py", "Peer") + for _ in range(6): + p.on_timer() + ok, badcrc, trunc, junk, badlen = payloads(ctx) # tick 5 is silence: nothing sent + assert len(payloads(ctx)) == 5 + assert ok[25] == crc8_dvbs2(ok[2:25]) + assert badcrc[25] != crc8_dvbs2(badcrc[2:25]) and len(badcrc) == 26 + assert len(trunc) == 10 and trunc[0] == 0xC8 and trunc == ok[:10] + assert junk == bytes([0xFF, 0x00] * 4) + assert badlen[1] == 0xFF and len(badlen) == 26 + + +def test_malicious_uart_late_then_never_reply_and_link_loss(): + p, ctx, _ = load("uart/malicious_uart.py", "Peer") + p.on_byte(0x3F) # request 1: answered late + for _ in range(75): + p.on_timer() + assert b"!" not in payloads(ctx) + p.on_timer() + assert b"!" in payloads(ctx) + n = len(payloads(ctx)) + p.on_byte(0x3F) # request 2: never answered + for _ in range(100): + p.on_timer() + assert payloads(ctx)[n:].count(b"!") == 0 + for _ in range(600): + p.on_timer() + assert ("cancel", 0) in ctx.timers # the link goes silent for good + + +# ---- I2C ------------------------------------------------------------------- + +def read_reg(s, reg, n): + s.on_write(bytes([reg])) + return s.on_read(n) + + +def test_ina228_known_answers_follow_the_firmware_config(): + s, ctx, _ = load("i2c/good_ina228.py", "Slave") + assert read_reg(s, 0x01, 2) == b"\xfb\x68" # ADC_CONFIG reset: SLYS021A 7.6.1.2 + s.on_write(bytes([0x02, 0x10, 0x00])) # SHUNT_CAL = 0x1000 + s.on_finish() + s.on_write(bytes([0x01, 0xFB, 0x68])) # (re)start continuous conversions + s.on_finish() + assert read_reg(s, 0x05, 3) == b"\x00\x00\x00" # nothing converted yet + ctx.now_us = 3156 - 1 # 3 x 1052 us, AVG = 1: one us short + assert read_reg(s, 0x0B, 2) == b"\x00\x01" + ctx.now_us = 3156 + assert read_reg(s, 0x0B, 2) == b"\x00\x03" # CNVRF set + assert read_reg(s, 0x0B, 2) == b"\x00\x01" # cleared by the read + assert read_reg(s, 0x05, 3) == (round(16.8 / 195.3125e-6) << 4).to_bytes(3, "big") # 86016 << 4 = 0x150000 + lsb = 0x1000 / (13107.2e6 * 0.0005) # Eq. 2 = 625 uA + assert read_reg(s, 0x07, 3) == (round(12.0 / lsb) << 4).to_bytes(3, "big") # 19200 << 4 = 0x04B000 + assert read_reg(s, 0x08, 3) == round(16.8 * 12.0 / (3.2 * lsb)).to_bytes(3, "big") # Eq. 5 + + +def test_icp20100_pressure_pair_known_answer(): + s, ctx, _ = load("i2c/good_icp20100.py", "Slave") + assert read_reg(s, 0xD3, 1) == b"\xb2" + s.on_write(bytes([0xC0, 0x08])) # MEAS_MODE = 1, mode 0 = 25 Hz + s.on_finish() + ctx.now_us = 40_000 + p, t = 99614, -12098 # (100.4-70)*2^17/40, (22-25)*2^18/65, truncated + exp = bytes((p & 0xFF, p >> 8 & 0xFF, p >> 16, t & 0xFF, t >> 8 & 0xFF, (t >> 16) & 0x0F)) + assert read_reg(s, 0xFA, 6) == exp + + +def tf_frame(dist, strength, head=b"\x59\x59"): + body = head + struct.pack("6h", ax, 0, az, 0, 0, 0) + b"\x00" + struct.pack(">H", 30_000) + + +def test_malicious_icm42688_faults(): + s, ctx, _ = load("spi/malicious_icm42688.py", "Slave") + assert spi(s, [0xF5, 0])[1] == 0x42 # wrong WHO_AM_I + assert spi(s, [0xAE, 0, 0])[1:] == b"\x00\x04" # claims 4 records + pkt = spi(s, [0xB0] + [0] * 16)[1:] + assert pkt[0] == 0x80 and pkt[1:] == bytes(15) # ...but every packet is "FIFO empty" + g1, g2 = spi(s, [0xA5] + [0] * 6)[1:], spi(s, [0xA5] + [0] * 6)[1:] + assert g1 == g2 == b"\x12\x34" * 3 # gyro stuck + a1, a2 = spi(s, [0x9F] + [0] * 6)[1:], spi(s, [0x9F] + [0] * 6)[1:] + assert struct.unpack(">3h", a1)[2] == 2048 and struct.unpack(">3h", a2)[2] == 2048 ^ 0x4000 # bit 14 flipped + + +# ---- CAN ------------------------------------------------------------------- + +def test_dronecan_allocation_then_nodestatus(): + n, ctx, _ = load("can/good_dronecan.py", "Node") + assert ctx.timers == [("oneshot", 3_000_000)] + n.on_timer() # stage 1 request: anonymous + (a, k), = ctx.sent + uid = bytes(range(0x50, 0x60)) + assert a[0] == (24 << 24 | 0x1234 << 10 | 1 << 8) and k["extended"] + assert a[1] == bytes([42 << 1 | 1]) + uid[:6] + b"\xc0" + alloc = 24 << 24 | 1 << 8 | 1 # allocator = node 1 + n.on_frame({"id": alloc, "data": bytes([0]) + uid[:6] + b"\xc0", "extended": True}) + assert ctx.sent[-1][0][1] == bytes([42 << 1]) + uid[6:12] + b"\xc1" + n.on_frame({"id": alloc, "data": bytes([0]) + uid[:12] + b"\xc1", "extended": True}) + assert ctx.sent[-1][0][1] == bytes([42 << 1]) + uid[12:] + b"\xc2" + n.on_frame({"id": alloc, "data": bytes([42 << 1]) + uid + b"\xc3", "extended": True}) # grant + assert ("periodic", 1_000_000) in ctx.timers + n.on_timer() + (cid, data), k = ctx.sent[-1] + assert cid == 0x1801552A # prio 24, NodeStatus 341, node 42 + assert data == struct.pack("HHBBH", 1, 0x0800, 6, 4, 1) + + DUT_MAC + DUT_IP + bytes(6) + PEER_IP) + p.on_frame({"bytes": req}) + exp = (DUT_MAC + PEER_MAC + b"\x08\x06" + struct.pack(">HHBBH", 1, 0x0800, 6, 4, 2) + + PEER_MAC + PEER_IP + DUT_MAC + DUT_IP) + assert payloads(ctx)[0] == exp + bytes(60 - len(exp)) + + +def udp_request(payload, bad_ip=False): + udp = struct.pack(">HHHH", 5000, 7, 8 + len(payload), 0) + payload + pseudo = DUT_IP + PEER_IP + bytes((0, 17)) + udp[4:6] + udp = udp[:6] + struct.pack(">H", (~ones_complement_sum(pseudo + udp)) & 0xFFFF) + udp[8:] + ip = struct.pack(">BBHHHBBH", 0x45, 0, 20 + len(udp), 1, 0, 64, 17, 0) + DUT_IP + PEER_IP + ck = (~ones_complement_sum(ip)) & 0xFFFF + ip = ip[:10] + struct.pack(">H", ck ^ bad_ip) + ip[12:] + f = PEER_MAC + DUT_MAC + b"\x08\x00" + ip + udp + return f + bytes(max(0, 60 - len(f))) + + +def test_udp_echo_has_valid_checksums_and_swapped_endpoints(): + p, ctx, _ = load("eth/good_arp_udp.py", "Peer") + p.on_frame({"bytes": udp_request(b"hello")}) + r = payloads(ctx)[0] + assert r[:6] == DUT_MAC and r[6:12] == PEER_MAC and r[12:14] == b"\x08\x00" + assert r[26:30] == PEER_IP and r[30:34] == DUT_IP + assert struct.unpack(">HH", r[34:38]) == (7, 5000) + assert r[42:47] == b"hello" and udp_ok(r[:14 + 20 + 8 + 5]) + + +def test_udp_echo_ignores_a_bad_ip_checksum(): + p, ctx, _ = load("eth/good_arp_udp.py", "Peer") + p.on_frame({"bytes": udp_request(b"hello", bad_ip=0x1111)}) + assert ctx.sent == [] + + +def test_malicious_eth_sequence(): + p, ctx, _ = load("eth/malicious_frames.py", "Peer") + for _ in range(7): + p.on_timer() + assert ctx.order == ["send", "wire", "send", "send", "send", "send", "send"] + s = payloads(ctx) + good1, runt, foreign, bad_ip, bad_udp, good7 = s + wire, = ctx.wire + assert wire[-4:] != zlib.crc32(wire[:-4]).to_bytes(4, "little") # FCS is wrong... + assert wire[-4:] == bytes(b ^ 0xFF for b in zlib.crc32(wire[:-4]).to_bytes(4, "little")) # ...inverted CRC-32 + assert len(runt) == 30 + assert foreign[:6] == bytes.fromhex("020000000077") and udp_ok(foreign[:14 + 20 + 8 + 17]) + assert not udp_ok(bad_ip[:14 + 20 + 8 + 17]) and ones_complement_sum(bad_ip[14:34]) != 0xFFFF + assert ones_complement_sum(bad_udp[14:34]) == 0xFFFF and not udp_ok(bad_udp[:14 + 20 + 8 + 17]) + for f, marker in ((good1, 1), (good7, 7)): + assert f[:6] == DUT_MAC and f[42] == marker and udp_ok(f[:14 + 20 + 8 + 17]) + + +def test_scenarios_parse_and_reference_existing_scripts(): + import yaml + for sc in PEERS.rglob("scenario.yaml"): + doc = yaml.safe_load(sc.read_text()) + for m in doc["machines"].values(): + for part in m["parts"]: + assert (sc.parent / part["script"]).is_file(), (sc, part) diff --git a/examples/tests/test_peers.py b/examples/tests/test_peers.py index eb8eec2..ed3f475 100644 --- a/examples/tests/test_peers.py +++ b/examples/tests/test_peers.py @@ -65,14 +65,14 @@ def test_mavlink_stream_has_the_expected_messages(run): def test_gps_fix_reaches_mavlink(run): msgs, _ = run - assert last(msgs, GPS_RAW_INT)[28] >= 3 # fix_type: gps_peer reports a 3D fix + assert last(msgs, GPS_RAW_INT)[28] >= 3 # fix_type: good_gps reports a 3D fix def test_imu_is_polled_and_attitude_follows(run): msgs, bus = run assert bus["imu"] > 100 pitch_deg = struct.unpack_from(" Date: Sun, 11 Oct 2026 02:41:20 -0700 Subject: [PATCH 2/2] Fix peers found by end-to-end runs; add Verified table good_gps re-arms its stream when the rate arrives after NAV-PVT enable, good_ina228 serves ENERGY/CHARGE, malicious_tf02 NACKs the first address phases, send_wire caveat corrected for sim 0.6.6. Co-Authored-By: Claude Sonnet 5.5 --- examples/README.md | 42 +++++++++++++++++++----- examples/peers/eth/malicious_frames.py | 7 ++-- examples/peers/i2c/good_ina228.py | 18 ++++++---- examples/peers/i2c/malicious_tf02.py | 14 ++++++-- examples/peers/spi/malicious_icm42688.py | 4 +++ examples/peers/uart/good_gps.py | 14 +++++--- examples/tests/test_peer_mocks.py | 25 ++++++++++++++ 7 files changed, 99 insertions(+), 25 deletions(-) diff --git a/examples/README.md b/examples/README.md index 2947ecc..858a00a 100644 --- a/examples/README.md +++ b/examples/README.md @@ -29,16 +29,40 @@ against your own `firmware.elf`. Limits to know about: -- I2C: scripted slaves cannot yet NACK a data byte or stretch SCL mid-transfer - , so the I2C malicious peer does not inject either. - The CLI guide describes an `on_address` hook that returns False - to NACK the address; these examples do not use or test it. -- Ethernet: the bad-FCS frame uses `ctx.send_wire`, present in the pyrite - engine's peer only; the malicious peer skips that frame (with a warning) where - it is missing. The addresses in `malicious_frames.py` (`DUT_MAC`, `DUT_IP`) - are constants to set to your firmware's. +- I2C: scripted slaves cannot yet NACK a data byte or stretch SCL + mid-transfer, so the I2C malicious peer injects neither. It does NACK the + first three address phases via `on_address` returning False, which works in + sim 0.6.6 (the firmware under test retried and recovered). +- Ethernet: the bad-FCS frame needs `ctx.send_wire`, which sim 0.6.6 does not + have; `malicious_frames.py` skips that frame with a warning there. The + addresses in it (`DUT_MAC`, `DUT_IP`) are constants to set to your firmware's. + In 0.6.6 an `eth-peer` must be a `parts:` entry; declaring + `Network.ScriptedEthernetPeer` in an overlay fails registration. - A peer's `ctx` has a single timer, so peers that need several delays count - ticks of one periodic timer. + ticks of one periodic timer. The UART peer ctx has no `now_us`. + +## Verified (local runs, no firmware shipped) + +Firmware class: an STM32H753 flight-controller build with GPS on two UARTs, +two ICM-42688-P on SPI1, INA228, ICP-20100 and TF02-Pro on I2C, CRSF on UART4, +DroneCAN on FDCAN1 and MAVLink on USART1, 12 s virtual, `sim` 0.6.6 (C# engine). +The firmware's own reference peers were swapped for each example on its bus. + +| Example | Result | +| --- | --- | +| `uart/good_gps.py` | Run. 34 `GPS_RAW_INT`, fix 3D, 11 sats, same count as the reference peer (a first version streamed at 1 Hz because the firmware enables NAV-PVT before programming the rate; fixed). | +| `uart/good_crsf.py` | Run. `RC_CHANNELS` carries 1500/1500/1000/1500/2011 us, CRC-clean MAVLink. | +| `i2c/good_ina228.py` | Run. `BATTERY_STATUS` identical to the reference peer (a first version lacked ENERGY/CHARGE; fixed). | +| `i2c/good_icp20100.py` | Run. Baro polled (about 7250 I2C transactions), run completes. | +| `spi/good_icm42688.py` | Run on both IMUs. `ATTITUDE` pitch -10.47 deg, same as reference. | +| `can/good_dronecan.py` | Run. 29 CAN frames, both directions, same as the reference. | +| `uart/malicious_uart.py` | Run. Completes. The firmware's RC output stayed at its failsafe values (1500/1500/1000/1500/1000...) for the whole run. With only truncated, garbage or bad-length faults mixed with valid frames it accepted the valid ones; with a bad-CRC frame in the stream it never did. Cause not investigated. | +| `i2c/malicious_tf02.py` | Run. 724 faulty polls served; `DISTANCE_SENSOR` stayed 1250 cm (the 65535 frame never surfaced). The firmware never sent the version query, so the wrong-ID reply was not exercised. Address NACK works and the firmware recovers. | +| `spi/malicious_icm42688.py` | Run on each IMU. On the IMU the firmware probes with WHO_AM_I the wrong ID disabled it: no `RAW_IMU`, pitch 0. The firmware never reads WHO_AM_I on the other IMU; there it read the FIFO count 4 and empty packets repeatedly without crashing. | +| `can/malicious_echo.py` | Run. Completes; 22 CAN frames, the firmware's own frames were echoed corrupted (last byte, ID, DLC) and it kept running. | +| `eth/malicious_frames.py` | Run against a prebuilt Zephyr raw-frame listener on NUCLEO-H753ZI: completes, but the listener only sees EtherType 0x88B5, so none of its IP-level faults were observable. A scratch variant (0x88B5, marker byte) showed sim 0.6.6 delivering the runt and the foreign-MAC frame to the application; bad-FCS skipped (no `send_wire`). | +| `eth/good_arp_udp.py` | Not run: no firmware with an IP stack is available. Mock test only. | +| `tests/test_peers.py` | Run with that firmware on the installed 0.6.9 engine: 4 pass, `test_baro_reaches_scaled_pressure` XPASS(strict) because this build does not have the uninitialised-struct bug the xfail documents. | No firmware is shipped. The peers hold their expected values as constants at the top of each file; the tests read those same numbers. diff --git a/examples/peers/eth/malicious_frames.py b/examples/peers/eth/malicious_frames.py index ee8f940..74f6c46 100644 --- a/examples/peers/eth/malicious_frames.py +++ b/examples/peers/eth/malicious_frames.py @@ -11,9 +11,10 @@ # 7 valid -> must be delivered (the stack recovered) # Expected: markers 1 and 7 reach the application, 2-6 do not, and the # firmware does not hang, leak buffers or crash on the bad ones. -# Frame 2 uses ctx.send_wire (we supply the FCS), which the pyrite engine's -# scripted peer has; the C# peer's ctx has no send_wire, so there frame 2 is -# skipped with a warning. +# Frame 2 uses ctx.send_wire (we supply the FCS). sim 0.6.6's ctx has no +# send_wire (observed: "'_Ctx' object has no attribute 'send_wire'"), so there +# frame 2 is skipped with a warning and the run continues; an engine whose ctx +# has send_wire (the pyrite engine's peer documents it) sends it. # Spec: IEEE 802.3 (FCS = CRC-32, sent least-significant byte first), RFC 791, # RFC 768. import struct diff --git a/examples/peers/i2c/good_ina228.py b/examples/peers/i2c/good_ina228.py index 996750a..edd5446 100644 --- a/examples/peers/i2c/good_ina228.py +++ b/examples/peers/i2c/good_ina228.py @@ -8,7 +8,8 @@ # the enabled inputs' conversion times x averaging, from ADC_CONFIG; CURRENT # and POWER scale from the SHUNT_CAL it writes (CURRENT_LSB, Eq. 2). Results # read 0 until the first cycle completes; reading DIAG_ALRT clears CNVRF. -# ENERGY/CHARGE accumulators are modelled for time since conversions started. +# ENERGY/CHARGE (40 bit, Eq. 6-7) accumulate from the last ADC_CONFIG write; +# RSTACC is not modelled. PACK_V = 16.8 # V, bus voltage LOAD_A = 12.0 # A, current through the shunt DIE_C = 35.0 # degC @@ -18,7 +19,7 @@ AVG_N = (1, 4, 16, 64, 128, 256, 512, 1024) # per AVG code: Table 7-6 CONVDLY_STEP_US = 2000 # CONFIG.CONVDLY step: Table 7-5 -CONFIG, ADC_CONFIG, SHUNT_CAL, VBUS, DIETEMP, CURRENT, POWER, DIAG_ALRT = 0, 1, 2, 5, 6, 7, 8, 11 +CONFIG, ADC_CONFIG, SHUNT_CAL, VBUS, DIETEMP, CURRENT, POWER, ENERGY, CHARGE, DIAG_ALRT = 0, 1, 2, 5, 6, 7, 8, 9, 10, 11 RESET = {CONFIG: 0x0000, ADC_CONFIG: 0xFB68, SHUNT_CAL: 0x1000} # SLYS021A 7.6.1.1-3 @@ -64,16 +65,21 @@ def _value(self, reg): if reg in RESET: return be(self.regs[reg], 2) lsb = self._current_lsb() - if n == 0 or (reg in (CURRENT, POWER) and lsb == 0): - return bytes(2 if reg == DIETEMP else 3) + if n == 0 or (reg in (CURRENT, POWER, ENERGY, CHARGE) and lsb == 0): + return bytes({DIETEMP: 2, ENERGY: 5, CHARGE: 5}.get(reg, 3)) + t_s = n * self._cycle_us() * 1e-6 # time since conversions started if reg == VBUS: - return be(int(PACK_V / 195.3125e-6) << 4, 3) # 195.3125 uV/LSB, 20 bit left-aligned + return be(round(PACK_V / 195.3125e-6) << 4, 3) # 195.3125 uV/LSB, 20 bit left-aligned if reg == DIETEMP: return be(int(DIE_C / 7.8125e-3), 2) # 7.8125 mdegC/LSB if reg == CURRENT: - return be(int(LOAD_A / lsb) << 4, 3) # Eq. 4 + return be(round(LOAD_A / lsb) << 4, 3) # Eq. 4 if reg == POWER: return be(PACK_V * LOAD_A / (3.2 * lsb), 3) # Eq. 5 + if reg == ENERGY: + return be(PACK_V * LOAD_A * t_s / (16 * 3.2 * lsb), 5) # Eq. 6, 40 bit + if reg == CHARGE: + return be(LOAD_A * t_s / lsb, 5) # Eq. 7, 40 bit return bytes(2) def on_write(self, data): diff --git a/examples/peers/i2c/malicious_tf02.py b/examples/peers/i2c/malicious_tf02.py index 977010d..629df4b 100644 --- a/examples/peers/i2c/malicious_tf02.py +++ b/examples/peers/i2c/malicious_tf02.py @@ -15,11 +15,14 @@ # every following poll in this phase; detect "never changes" # Version query (5A 04 01 5F) replies with a frame whose ID byte is wrong # (0xEE instead of 0x01), so a probe that checks the device identity must fail. -# I2C limits: peers cannot yet NACK a data byte or stretch SCL mid-transfer, -# so none of that is injected here. +# Address NACK: on_address returns False for the first NACK_FIRST address +# phases (the part is "not there yet"), then ACKs. This hook worked in sim 0.6.6 +# (see ../../README.md). Peers cannot yet NACK a data byte or stretch SCL +# mid-transfer, so none of that is injected here. # A robust firmware checks header and checksum, range-checks distance and # strength, and flags a value that never changes. GOOD_CM = 1250 +NACK_FIRST = 3 TEMP_RAW = (25 + 256) * 8 @@ -49,17 +52,24 @@ def __init__(self, ctx): self.ctx = ctx self.reply = b"" self.triggers = 0 + self.addr_phases = 0 def on_write(self, data): data = bytes(data) if len(data) < 3 or data[0] != 0x5A: return if data[2] == 0x01: # version query: wrong ID byte + self.ctx.info("version query: replying with wrong ID 0xEE") body = bytes([0x5A, 0x07, 0xEE, 0x02, 0x03, 0x01]) self.reply = body + bytes([cksum(body)]) elif data[2] == 0x00: # trigger self.reply = phase_frame(self.triggers % 5) + self.ctx.info(f"trigger #{self.triggers}: fault phase {self.triggers % 5}") self.triggers += 1 + def on_address(self, addr, is_read): + self.addr_phases += 1 + return self.addr_phases > NACK_FIRST + def on_read(self, count): return (self.reply + bytes(count))[:count] diff --git a/examples/peers/spi/malicious_icm42688.py b/examples/peers/spi/malicious_icm42688.py index 7bb84d8..889465e 100644 --- a/examples/peers/spi/malicious_icm42688.py +++ b/examples/peers/spi/malicious_icm42688.py @@ -28,6 +28,7 @@ def __init__(self, ctx): self.ctx = ctx self.beat, self.reg, self.read, self.out = 0, 0, False, b"" self.accel_reads = 0 + self.logged = set() def _value(self, reg): if reg == WHO_AM_I: @@ -49,6 +50,9 @@ def on_transmit(self, b): self.read, self.reg = bool(b & 0x80), b & 0x7F if self.read: self.out = self._value(self.reg) + if self.reg in (WHO_AM_I, FIFO_COUNTH) and self.reg not in self.logged: + self.logged.add(self.reg) + self.ctx.info(f"read reg 0x{self.reg:02x}: injecting fault") return 0x00 if not self.read: return 0x00 diff --git a/examples/peers/uart/good_gps.py b/examples/peers/uart/good_gps.py index d0c9831..894d64d 100644 --- a/examples/peers/uart/good_gps.py +++ b/examples/peers/uart/good_gps.py @@ -44,7 +44,8 @@ def __init__(self, ctx): self.ctx, self.rx = ctx, bytearray() self.itow_ms = 0 self.rate_meas = self.rate_nav = None - self.pvt_on = self.streaming = False + self.pvt_on = False + self.period_us = None def on_byte(self, b): self.rx.append(b) @@ -80,13 +81,16 @@ def _frame(self, raw): self._maybe_stream() def _maybe_stream(self): - if self.streaming or not self.pvt_on: - return - self.streaming = True + # The firmware enables NAV-PVT before it programs the rate (observed + # on a real driver), so re-arm the timer whenever the period changes. meas = self.rate_meas or DEFAULT_RATE_MEAS_MS nav = self.rate_nav or DEFAULT_RATE_NAV + period = meas * nav * 1000 + if not self.pvt_on or period == self.period_us: + return + self.period_us = period self.ctx.info(f"NAV-PVT every {meas * nav} ms (as programmed by the firmware)") - self.ctx.schedule_periodic(meas * nav * 1000) + self.ctx.schedule_periodic(period) def on_timer(self): self.itow_ms += self.rate_meas or DEFAULT_RATE_MEAS_MS diff --git a/examples/tests/test_peer_mocks.py b/examples/tests/test_peer_mocks.py index 0660236..b8fa8a4 100644 --- a/examples/tests/test_peer_mocks.py +++ b/examples/tests/test_peer_mocks.py @@ -123,6 +123,19 @@ def valset(key, val): assert struct.unpack_from("