#!/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!")