Deploy presence daemon to OPi Zero 3 test implant
sensor.service running at 10.0.0.0. ARP + DHCP sensors active. Probe sniffer degraded (no wlan1). BLE degraded (BlueZ too old for or_patterns). Untracked audit files committed for reference.
This commit is contained in:
Cobra
@@ -0,0 +1,245 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test suite for presence_daemon parsers — netlink and DHCP bounds checking.
|
||||
Tests minimal reproductions of bounds checking issues.
|
||||
"""
|
||||
|
||||
import struct
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# We'll test by importing just the functions directly
|
||||
# by reading the source and executing key portions
|
||||
|
||||
|
||||
def parse_netlink_current(data: bytes) -> None:
|
||||
"""Current implementation from presence_daemon.py."""
|
||||
try:
|
||||
offset = 0
|
||||
while offset < len(data):
|
||||
if offset + 16 > len(data):
|
||||
break
|
||||
|
||||
nlmsg_len, nlmsg_type, _, _, _ = struct.unpack_from('=IHHII', data, offset)
|
||||
if nlmsg_len < 16:
|
||||
break
|
||||
|
||||
payload = data[offset + 16:offset + nlmsg_len]
|
||||
# This is the issue: if nlmsg_len > len(data), we silently get a short payload
|
||||
# No guard that offset + nlmsg_len <= len(data)
|
||||
|
||||
offset += (nlmsg_len + 3) & ~3
|
||||
except Exception as e:
|
||||
pass
|
||||
|
||||
|
||||
def parse_ndmsg_current(data: bytes, msg_type: int) -> None:
|
||||
"""Current implementation from presence_daemon.py."""
|
||||
try:
|
||||
if len(data) < 12:
|
||||
return
|
||||
|
||||
state = struct.unpack_from('=H', data, 8)[0]
|
||||
|
||||
mac = None
|
||||
ip = None
|
||||
offset = 12
|
||||
|
||||
while offset + 4 <= len(data):
|
||||
rta_len, rta_type = struct.unpack_from('=HH', data, offset)
|
||||
if rta_len < 4:
|
||||
break
|
||||
|
||||
val = data[offset + 4:offset + rta_len]
|
||||
|
||||
if rta_type == 1 and len(val) == 6: # NDA_LLADDR
|
||||
mac = ':'.join(f'{b:02x}' for b in val)
|
||||
elif rta_type == 2: # NDA_DST
|
||||
if len(val) == 4:
|
||||
pass # would parse IP
|
||||
|
||||
offset += (rta_len + 3) & ~3
|
||||
|
||||
# If we had extracted a MAC, we'd proceed
|
||||
# The issue: rta_len might claim more bytes than available
|
||||
|
||||
except Exception as e:
|
||||
pass
|
||||
|
||||
|
||||
def parse_dhcp_current(data: bytes) -> None:
|
||||
"""Current implementation from presence_daemon.py."""
|
||||
try:
|
||||
if len(data) < 14:
|
||||
return
|
||||
|
||||
eth_type = struct.unpack('!H', data[12:14])[0]
|
||||
if eth_type != 0x0800:
|
||||
return
|
||||
|
||||
if len(data) < 23:
|
||||
return
|
||||
proto = data[23]
|
||||
if proto != 17:
|
||||
return
|
||||
|
||||
if len(data) < 38:
|
||||
return
|
||||
src_port = struct.unpack('!H', data[34:36])[0]
|
||||
dst_port = struct.unpack('!H', data[36:38])[0]
|
||||
if not (src_port in (67, 68) and dst_port in (67, 68)):
|
||||
return
|
||||
|
||||
if len(data) < 42:
|
||||
return
|
||||
dhcp = data[42:]
|
||||
if len(dhcp) < 236:
|
||||
return
|
||||
|
||||
# MAC extraction — offset 28-34
|
||||
mac_bytes = dhcp[28:34]
|
||||
mac = ':'.join(f'{b:02x}' for b in mac_bytes)
|
||||
if mac in ('00:00:00:00:00:00', 'ff:ff:ff:ff:ff:ff'):
|
||||
return
|
||||
|
||||
if len(dhcp) < 240:
|
||||
return
|
||||
|
||||
msg_type = None
|
||||
i = 240
|
||||
while i < len(dhcp):
|
||||
opt = dhcp[i]
|
||||
if opt == 255:
|
||||
break
|
||||
if opt == 0:
|
||||
i += 1
|
||||
continue
|
||||
if i + 1 >= len(dhcp):
|
||||
break
|
||||
|
||||
length = dhcp[i + 1]
|
||||
if i + 2 + length > len(dhcp):
|
||||
break
|
||||
|
||||
val = dhcp[i + 2:i + 2 + length]
|
||||
|
||||
if opt == 53 and length == 1:
|
||||
msg_type = val[0]
|
||||
|
||||
i += 2 + length
|
||||
|
||||
except Exception as e:
|
||||
pass
|
||||
|
||||
|
||||
# Test cases
|
||||
|
||||
def test_parse_netlink_truncated_header():
|
||||
"""Netlink parser should handle buffer shorter than header gracefully."""
|
||||
truncated = b'\x01\x02\x03'
|
||||
# Should not raise
|
||||
parse_netlink_current(truncated)
|
||||
print("✓ test_parse_netlink_truncated_header passed")
|
||||
|
||||
|
||||
def test_parse_netlink_header_claims_more_than_available():
|
||||
"""Netlink parser should handle nlmsg_len > buffer length."""
|
||||
# Build a netlink header that claims 100 bytes but only provide 16
|
||||
nlmsg_len = 100
|
||||
nlmsg_type = 20
|
||||
flags = 0
|
||||
seq = 0
|
||||
pid = 0
|
||||
data = struct.pack('=IHHII', nlmsg_len, nlmsg_type, flags, seq, pid)
|
||||
assert len(data) == 16
|
||||
# Should not raise
|
||||
parse_netlink_current(data)
|
||||
print("✓ test_parse_netlink_header_claims_more_than_available passed")
|
||||
|
||||
|
||||
def test_parse_netlink_offset_overflow():
|
||||
"""Netlink parser should not overflow offset calculation."""
|
||||
# Two messages: first claims huge length, second is valid
|
||||
msg1 = struct.pack('=IHHII', 0xffffffff, 20, 0, 0, 0) # Claims ~4GB
|
||||
msg2 = struct.pack('=IHHII', 16, 20, 0, 1, 0) # Valid header
|
||||
data = msg1 + msg2
|
||||
# Should not raise (might infinite loop with current code)
|
||||
parse_netlink_current(data)
|
||||
print("✓ test_parse_netlink_offset_overflow passed")
|
||||
|
||||
|
||||
def test_parse_ndmsg_rta_claims_more_data():
|
||||
"""ndmsg parser should handle RTA that claims more bytes than available."""
|
||||
# ndmsg header: family, pad, type, flags, state, index (6 bytes with BBBBHH)
|
||||
ndmsg_header = struct.pack('=BBBBHH', 0, 0, 0, 0, 64, 0)
|
||||
|
||||
# RTA header claiming 20 bytes total length
|
||||
rta_len = 20
|
||||
rta_type = 1 # NDA_LLADDR
|
||||
rta_header = struct.pack('=HH', rta_len, rta_type)
|
||||
|
||||
# But only provide 2 bytes of actual RTA data (not 16 bytes as claimed)
|
||||
data = ndmsg_header + rta_header + b'\x01\x02'
|
||||
|
||||
# Should not raise
|
||||
parse_ndmsg_current(data, 20) # RTM_NEWNEIGH
|
||||
print("✓ test_parse_ndmsg_rta_claims_more_data passed")
|
||||
|
||||
|
||||
def test_parse_dhcp_minimal_valid():
|
||||
"""DHCP parser should accept valid minimal DHCP frame."""
|
||||
# Build minimal valid DHCP frame (Eth + IP + UDP + DHCP)
|
||||
eth_dst = b'\xff\xff\xff\xff\xff\xff'
|
||||
eth_src = b'\x00\x11\x22\x33\x44\x55'
|
||||
eth_type = struct.pack('!H', 0x0800)
|
||||
|
||||
# IP header (20 bytes)
|
||||
ip_version_ihl = 0x45
|
||||
ip_dscp_ecn = 0
|
||||
ip_length = struct.pack('!H', 20 + 8 + 260)
|
||||
ip_id = struct.pack('!H', 0)
|
||||
ip_flags_frag = struct.pack('!H', 0x4000)
|
||||
ip_ttl = 64
|
||||
ip_proto = 17 # UDP
|
||||
ip_checksum = struct.pack('!H', 0)
|
||||
ip_src = b'\x00\x00\x00\x00'
|
||||
ip_dst = b'\xff\xff\xff\xff'
|
||||
|
||||
ip_header = (
|
||||
bytes([ip_version_ihl, ip_dscp_ecn]) +
|
||||
ip_length + ip_id + ip_flags_frag +
|
||||
bytes([ip_ttl, ip_proto]) + ip_checksum +
|
||||
ip_src + ip_dst
|
||||
)
|
||||
|
||||
# UDP header (8 bytes)
|
||||
udp_src = struct.pack('!H', 68)
|
||||
udp_dst = struct.pack('!H', 67)
|
||||
udp_length = struct.pack('!H', 8 + 260)
|
||||
udp_checksum = struct.pack('!H', 0)
|
||||
|
||||
udp_header = udp_src + udp_dst + udp_length + udp_checksum
|
||||
|
||||
# DHCP payload (260 bytes)
|
||||
dhcp_body = bytearray(260)
|
||||
dhcp_body[0] = 1 # BOOTREQUEST
|
||||
dhcp_body[4:10] = b'\x12\x34\x56\x78\x9a\xbc'
|
||||
dhcp_body[28:34] = b'\xaa\xbb\xcc\xdd\xee\xff' # chaddr
|
||||
|
||||
# DHCP magic + option 53
|
||||
dhcp_body[240:244] = b'\x63\x82\x53\x63'
|
||||
dhcp_body[244:247] = b'\x35\x01\x01' # option 53, length 1, value 1 (DISCOVER)
|
||||
dhcp_body[247] = 255 # End
|
||||
|
||||
frame = eth_dst + eth_src + eth_type + ip_header + udp_header + bytes(dhcp_body)
|
||||
parse_dhcp_current(frame)
|
||||
print("✓ test_parse_dhcp_minimal_valid passed")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
test_parse_netlink_truncated_header()
|
||||
test_parse_netlink_header_claims_more_than_available()
|
||||
test_parse_netlink_offset_overflow()
|
||||
test_parse_ndmsg_rta_claims_more_data()
|
||||
test_parse_dhcp_minimal_valid()
|
||||
print("\nAll tests passed!")
|
||||
Reference in New Issue
Block a user