|
#!/usr/bin/env python3
|
|
"""Write two synthetic Modbus/TCP captures for the Suricata report. Standard library only.
|
|
|
|
python3 make_pcaps.py OUTDIR
|
|
|
|
modbus-midstream-server-first.pcap the capture starts with the server's
|
|
response to a request the capture did not see, then four complete
|
|
request/response pairs. No handshake.
|
|
modbus-midstream-client-first.pcap the control: the same session, but the
|
|
capture starts with a client request. No handshake.
|
|
|
|
Addresses are RFC 5737 documentation addresses. Client 192.0.2.10:49152,
|
|
server 192.0.2.20:502. Every request is Read Holding Registers (FC 3), unit 1,
|
|
address 0, quantity 2; every response carries two registers.
|
|
"""
|
|
import struct, sys
|
|
from pathlib import Path
|
|
|
|
CLIENT, SERVER = (bytes([192, 0, 2, 10]), 49152), (bytes([192, 0, 2, 20]), 502)
|
|
|
|
def checksum(data: bytes) -> int:
|
|
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 & 0xFFFF
|
|
|
|
def frame(src, dst, seq, ack, payload, ts):
|
|
(sip, sport), (dip, dport) = src, dst
|
|
tcp = struct.pack("!HHIIBBHHH", sport, dport, seq, ack, 5 << 4, 0x18, 64240, 0, 0) + payload
|
|
pseudo = sip + dip + struct.pack("!BBH", 0, 6, len(tcp))
|
|
tcp = tcp[:16] + struct.pack("!H", checksum(pseudo + tcp)) + tcp[18:]
|
|
ip = struct.pack("!BBHHHBBH4s4s", 0x45, 0, 20 + len(tcp), 1, 0x4000, 64, 6, 0, sip, dip)
|
|
ip = ip[:10] + struct.pack("!H", checksum(ip)) + ip[12:]
|
|
eth = b"\x02\x00\x00\x00\x00\x02" + b"\x02\x00\x00\x00\x00\x01" + b"\x08\x00"
|
|
pkt = eth + ip + tcp
|
|
return struct.pack("<IIII", int(ts), int((ts % 1) * 1e6), len(pkt), len(pkt)) + pkt
|
|
|
|
def request(tid): # MBAP + FC 3, address 0, quantity 2
|
|
return struct.pack("!HHHB", tid, 0, 6, 1) + struct.pack("!BHH", 3, 0, 2)
|
|
|
|
def response(tid): # MBAP + FC 3, byte count 4, two registers
|
|
return struct.pack("!HHHB", tid, 0, 7, 1) + struct.pack("!BBHH", 3, 4, 0x0011, 0x0022)
|
|
|
|
def session(server_first: bool) -> bytes:
|
|
out = struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 65535, 1)
|
|
cseq, sseq, ts = 1000, 5000, 1700000000.0
|
|
def send(from_client, payload):
|
|
nonlocal cseq, sseq, ts, out
|
|
ts += 0.01
|
|
if from_client:
|
|
out += frame(CLIENT, SERVER, cseq, sseq, payload, ts); cseq += len(payload)
|
|
else:
|
|
out += frame(SERVER, CLIENT, sseq, cseq, payload, ts); sseq += len(payload)
|
|
tid = 1
|
|
if server_first:
|
|
send(False, response(tid)); tid += 1 # the reply to a request before the capture began
|
|
for _ in range(4):
|
|
send(True, request(tid)); send(False, response(tid)); tid += 1
|
|
return out
|
|
|
|
if __name__ == "__main__":
|
|
out = Path(sys.argv[1] if len(sys.argv) > 1 else ".")
|
|
(out / "modbus-midstream-server-first.pcap").write_bytes(session(True))
|
|
(out / "modbus-midstream-client-first.pcap").write_bytes(session(False))
|
|
print("wrote", out / "modbus-midstream-server-first.pcap", "and", out / "modbus-midstream-client-first.pcap")
|