From 65d42229ada18301446f48ba9ac64087885f40fc Mon Sep 17 00:00:00 2001 From: Cobra Date: Tue, 14 Apr 2026 12:46:44 -0400 Subject: [PATCH] Fix P1/P2 audit findings from #671 review P1 Fixes: - Signal count enrollment logic: Changed from broken signal_count increment to tracking distinct signal types (BLE vs WiFi) using a set. Device enrolls when len(signal_types) >= 2, ensuring multi-source correlation. - DNS mDNS pointer endianness: Added bounds check to prevent out-of-bounds reads when following DNS compression pointers. Checks pointer_offset < offset and pointer_offset < len(payload) before recursing. - Nested RLock fragility: Refactored enrollment callback to not acquire lock (caller _ingest_signal holds it). Renamed _on_device_enrolled() to _fire_enrollment_callback() and removed lock acquisition. P2 Fixes: - BLE Handoff parsing: Implemented full HCI packet parsing to extract Apple Company ID (0x004C), Handoff message type (0x0C), and sequence number (bytes 4-5, big-endian). Calls _ingest_signal() with handoff_seq parameter. - DNS record count overflow: Capped total_records at 1000 to prevent unbounded loop DoS on crafted mDNS packets. - device_store unbounded growth: Added simple eviction when store exceeds 500 entries - evicts 100 oldest by first_seen timestamp. No LRU needed for MVP. All 40 existing tests continue to pass. --- net_alerter/net_alerter.py | 149 ++++++++++++++++++++++++++++--------- 1 file changed, 114 insertions(+), 35 deletions(-) diff --git a/net_alerter/net_alerter.py b/net_alerter/net_alerter.py index ef5217f..91489d1 100644 --- a/net_alerter/net_alerter.py +++ b/net_alerter/net_alerter.py @@ -156,7 +156,7 @@ def _create_identity_record(identity_id: str) -> dict: 'wifi_macs': set(), 'ble_macs': set(), 'last_seen': {}, # {mac: timestamp} - 'signal_count': 0, # Count of distinct signal source types (BLE, WiFi/DHCP/ARP/mDNS) + 'signal_types': set(), # Set of distinct signal categories: {'ble'} or {'wifi'} or both 'enrolled': False, 'dhcp_fingerprint': "", # DHCP Option 55 fingerprint (hex string) 'handoff_seq': None, # Last BLE Handoff sequence number seen @@ -164,19 +164,19 @@ def _create_identity_record(identity_id: str) -> dict: } -def _on_device_enrolled(identity_id: str) -> None: +def _fire_enrollment_callback(identity_id: str) -> None: """Called when a device meets enrollment criteria (2+ signal types). Logs enrollment, sends no alert (silent enrollment). + Caller must hold device_store_lock. """ - with device_store_lock: - if identity_id not in device_store: - return - dev = device_store[identity_id] - if not dev['enrolled']: - dev['enrolled'] = True - all_macs = dev['wifi_macs'] | dev['ble_macs'] - logging.info(f"Device enrolled (multi-source identity {identity_id}): {len(all_macs)} MAC(s), signal_count={dev['signal_count']}") + if identity_id not in device_store: + return + dev = device_store[identity_id] + if not dev['enrolled']: + dev['enrolled'] = True + all_macs = dev['wifi_macs'] | dev['ble_macs'] + logging.info(f"Device enrolled (multi-source identity {identity_id}): {len(all_macs)} MAC(s), signal_types={dev['signal_types']}") def _correlate_or_create_identity(mac: str, signal_type: str, dhcp_fp: str = "", handoff_seq: int = None) -> str: @@ -193,6 +193,9 @@ def _correlate_or_create_identity(mac: str, signal_type: str, dhcp_fp: str = "", Returns the identity_id (stable key). """ with device_store_lock: + # Determine signal category: 'ble' or 'wifi' (for dhcp/arp/mdns) + signal_category = 'ble' if signal_type == 'ble' else 'wifi' + # Strategy 1: Match by BLE Handoff sequence (most stable) if handoff_seq is not None: for iid, dev in device_store.items(): @@ -200,6 +203,7 @@ def _correlate_or_create_identity(mac: str, signal_type: str, dhcp_fp: str = "", # Found matching handoff anchor — add this MAC dev['ble_macs'].add(mac) dev['last_seen'][mac] = time.time() + dev['signal_types'].add('ble') logging.debug(f"BLE MAC rotation detected: {mac} linked to identity {iid} (seq {handoff_seq})") return iid @@ -214,9 +218,8 @@ def _correlate_or_create_identity(mac: str, signal_type: str, dhcp_fp: str = "", dev['last_seen'][mac] = time.time() if old_macs: logging.debug(f"MAC rotation (DHCP fingerprint match): {old_macs} → {mac}, identity {iid}") - # Increment signal count if this is first WiFi signal for this identity - if not old_macs and 'dhcp' not in signal_type: - dev['signal_count'] += 1 + # Add wifi signal type if not already present + dev['signal_types'].add('wifi') return iid # Strategy 3: Create new provisional identity @@ -236,12 +239,12 @@ def _correlate_or_create_identity(mac: str, signal_type: str, dhcp_fp: str = "", dev['ble_macs'].add(mac) if handoff_seq is not None: dev['handoff_seq'] = handoff_seq - dev['signal_count'] = 1 + dev['signal_types'].add('ble') else: dev['wifi_macs'].add(mac) if dhcp_fp: dev['dhcp_fingerprint'] = dhcp_fp - dev['signal_count'] = 1 + dev['signal_types'].add('wifi') dev['last_seen'][mac] = time.time() device_store[identity_id] = dev @@ -265,12 +268,16 @@ def _ingest_signal(mac: str, signal_type: str, dhcp_fp: str = "", handoff_seq: i dev = device_store[identity_id] dev['last_seen'][mac] = time.time() - # Enrollment check: if we have 2+ signal types and not yet enrolled - has_ble = bool(dev['ble_macs']) - has_wifi_like = bool(dev['wifi_macs']) + # Enrollment check: if we have 2+ distinct signal types and not yet enrolled + if not dev['enrolled'] and len(dev['signal_types']) >= 2: + _fire_enrollment_callback(identity_id) - if not dev['enrolled'] and has_ble and has_wifi_like: - _on_device_enrolled(identity_id) + # Eviction: if device_store exceeds 500 entries, evict 100 oldest by first_seen + if len(device_store) > 500: + sorted_ids = sorted(device_store.keys(), key=lambda iid: device_store[iid]['first_seen']) + for iid in sorted_ids[:100]: + del device_store[iid] + logging.debug(f"Evicted 100 oldest device identities; store size now {len(device_store)}") return identity_id @@ -1163,24 +1170,89 @@ def _parse_hci_event(data: bytes) -> None: def _parse_hcidump_line(line: bytes) -> None: - """Parse a line from hcidump output to extract Apple Continuity Handoff data.""" - try: - # hcidump -R outputs hex lines, e.g.: - # > 04 3E 2B 02 01 00 19 ... - # This requires hex parsing from the subprocess output - # Simplified: convert to hex string and look for Apple Company ID and Handoff type - hex_str = line.hex() + """Parse a line from hcidump output to extract Apple Continuity Handoff data. - # Look for Apple Company ID (0x004C in little-endian: 4C 00) - if '4c00' not in hex_str.lower(): + hcidump -R outputs hex dumps like: + > 04 3E 2B 02 01 00 19 00 01 02 03 04 05 06 07 08 ... + + Extracts BLE MAC and Handoff sequence number, then calls _ingest_signal(). + Returns None silently if parsing fails (incomplete packet). + """ + try: + # Parse hcidump hex output: skip leading ">" or "<" and split on whitespace + hex_str = line.decode('utf-8', errors='replace').strip() + if not hex_str or hex_str[0] not in '><': return - # Parse AD structure to find Handoff message type (0x0C) - # This is a simplified heuristic; full parsing would be more complex - # For MVP, we note that proper BLE parsing requires full HCI frame decode + # Extract hex bytes + hex_parts = hex_str[1:].split() + data = bytes([int(h, 16) for h in hex_parts if h]) + + if len(data) < 7: + return # Too short for HCI LE advertising report + + # Check for LE Meta Event (HCI packet type 04, HCI event 3E) + if data[0] != 0x04 or data[1] != 0x3E: + return + + # Skip HCI event header: type(1) event(1) len(1) subevt(1) = 4 bytes + if len(data) < 12: # Need at least event header + minimal adv report + return + + # HCI_LE_Meta_Event (0x3E), subevent is at offset 3 + subevent = data[3] + if subevent != 0x02: # LE Advertising Report subevent + return + + # LE Advertising Report structure: subevent(1) num_reports(1) event_type(1) addr_type(1) addr(6) len(1) data(variable) + if len(data) < 13: # Minimum for header + MAC + return + + num_reports = data[4] + if num_reports < 1: + return + + event_type = data[5] + addr_type = data[6] + + # Extract BLE MAC (6 bytes, little-endian on wire) + ble_mac_raw = data[7:13] + ble_mac = ':'.join(f'{b:02x}' for b in reversed(ble_mac_raw)) + + # Advertisement data length + if len(data) < 14: + return + ad_data_len = data[13] + if len(data) < 14 + ad_data_len: + return # Incomplete packet + + # Parse advertisement data for Apple Company ID (0x004C little-endian = 4C 00) + ad_data = data[14:14 + ad_data_len] + offset = 0 + + while offset < len(ad_data) - 1: + ad_len = ad_data[offset] + if ad_len == 0 or offset + 1 + ad_len > len(ad_data): + break + + ad_type = ad_data[offset + 1] + ad_payload = ad_data[offset + 2:offset + 1 + ad_len] + + # Look for Manufacturer Specific Data (type 0xFF) with Apple Company ID (0x004C) + if ad_type == 0xFF and len(ad_payload) >= 2: + company_id = struct.unpack('= BLE_HANDOFF_SEQ_OFFSET + 2: + # Check for Handoff message type (0x0C) + if ad_payload[2] == BLE_HANDOFF_MSG_TYPE: + # Extract sequence number from bytes 4-5 (big-endian) + seq = struct.unpack('!H', ad_payload[BLE_HANDOFF_SEQ_OFFSET:BLE_HANDOFF_SEQ_OFFSET + 2])[0] + _ingest_signal(ble_mac, 'ble', handoff_seq=seq) + return + + offset += 1 + ad_len except Exception: - pass + return # Silently fail on malformed packets def dhcp_sniffer(iface: str) -> None: @@ -1232,7 +1304,10 @@ def _parse_mdns_for_names(payload: bytes, mac: str) -> None: offset += 4 # QTYPE + QCLASS # Parse answers/authority/additional records - total_records = ancount + struct.unpack('!H', payload[8:10])[0] + struct.unpack('!H', payload[10:12])[0] + nscount = struct.unpack('!H', payload[8:10])[0] + arcount = struct.unpack('!H', payload[10:12])[0] + # Cap total records to prevent unbounded loop on crafted packets + total_records = min(ancount + nscount + arcount, 1000) for _ in range(total_records): if offset is None or offset + 10 > len(payload): break @@ -1307,7 +1382,10 @@ def _extract_dns_name(payload: bytes, offset: int) -> str: elif (length & 0xC0) == 0xC0: # Pointer if offset >= len(payload): break - pointer_offset = struct.unpack('!H', bytes([length & 0x3F]) + bytes([payload[offset]]))[0] + pointer_offset = struct.unpack('!H', bytes([(length & 0x3F), payload[offset]]))[0] + # Bounds check before recursing: pointer must not point backwards or beyond payload + if pointer_offset >= offset or pointer_offset >= len(payload): + break offset = pointer_offset continue @@ -1422,6 +1500,7 @@ def parse_frame(data: bytes) -> None: elif opt == 50 and length == 4: # Requested IP requested_ip = socket.inet_ntoa(val) elif opt == 55: # Parameter Request List (Option 55) + # Bounds check already done above; safe to use val dhcp_option55 = val i += 2 + length