Peer matrix: good and malicious scripted peers per bus

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

examples/README.md

@@ -4,11 +4,65 @@
 | --- | --- |
 | `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 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. 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.

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()

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)

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] }

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])))

examples/peers/eth/malicious_frames.py

@@ -0,0 +1,92 @@
+# 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). 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
+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)

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] }