"""Manage optional tshark packet enrichment workers and matching.""" from __future__ import annotations import asyncio import json import logging import os import signal import subprocess import threading import time from datetime import datetime, timezone from typing import Any, Dict, Iterable, List, Optional, Tuple import src.shared_objects as shared_objects from scapy.all import ARP, IP, IPv6, TCP, UDP # type: ignore from src.config import settings logger = logging.getLogger("tshark_manager") _GENERIC_PROTOCOLS = { "", "frame", "eth", "ethertype", "ip", "ipv6", "tcp", "udp", "data", "data_text_lines", "text", "text_lines", "geninfo", } def _safe_text(value: Any) -> Optional[str]: if value is None: return None text = str(value).strip() return text if text else None def _safe_int(value: Any) -> Optional[int]: try: text = _safe_text(value) if text is None: return None return int(text, 0) except Exception: return None def _safe_float(value: Any) -> Optional[float]: try: text = _safe_text(value) if text is None: return None return float(text) except Exception: return None def _safe_bool_flag(value: Any) -> Optional[bool]: text = _safe_text(value) if text is None: return None if text in {"1", "true", "True"}: return True if text in {"0", "false", "False"}: return False return None def _coerce_json_value(value: Any) -> Any: if isinstance(value, dict): out: Dict[str, Any] = {} for key, child in value.items(): coerced = _coerce_json_value(child) if coerced in (None, "", [], {}): continue out[str(key)] = coerced return out if isinstance(value, list): out = [_coerce_json_value(child) for child in value] out = [child for child in out if child not in (None, "", [], {})] if not out: return None if len(out) == 1: return out[0] return out return value def _iter_nodes(value: Any) -> Iterable[Tuple[Optional[str], Any]]: if isinstance(value, dict): for key, child in value.items(): yield key, child yield from _iter_nodes(child) elif isinstance(value, list): for child in value: yield None, child yield from _iter_nodes(child) def _key_matches_suffix(key: str, suffixes: Tuple[str, ...]) -> bool: normalized = key.lower().replace(".", "_") return any(normalized == suffix or normalized.endswith(f"_{suffix}") for suffix in suffixes) def _find_value(value: Any, *suffixes: str) -> Any: normalized_suffixes = tuple(suffix.lower().replace(".", "_") for suffix in suffixes) if isinstance(value, dict): for key, child in _iter_nodes(value): if key is None or not _key_matches_suffix(key, normalized_suffixes): continue coerced = _coerce_json_value(child) if coerced not in (None, "", [], {}): return coerced return None def _parse_time_ms(value: Any) -> Optional[int]: float_value = _safe_float(value) if float_value is not None: if float_value > 1_000_000_000_000: return int(float_value) return int(float_value * 1000) text = _safe_text(value) if text is None: return None try: return int(datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp() * 1000) except Exception: return None def _jsonable(metadata: Dict[str, Any]) -> Dict[str, Any]: out: Dict[str, Any] = {} for key, value in metadata.items(): if value in (None, "", [], {}): continue out[key] = value return out def _packet_signature( iface: str, eth_type_raw: int, protocol: Optional[int], src_ip: str, dst_ip: str, src_port: int, dst_port: int, length: int, ) -> Tuple[str, int, int, str, str, int, int, int]: return ( str(iface or ""), int(eth_type_raw or 0), int(protocol if protocol is not None else -1), str(src_ip), str(dst_ip), int(src_port), int(dst_port), int(length), ) def _signature_from_packet(pkt: Any, iface: str) -> Optional[Tuple[str, int, int, str, str, int, int, int]]: eth_type_raw: Optional[int] = None protocol: Optional[int] = None src_ip: Optional[str] = None dst_ip: Optional[str] = None if IP in pkt: ip_layer = pkt[IP] eth_type_raw = 0x0800 protocol = int(getattr(ip_layer, "proto", 0) or 0) src_ip = str(getattr(ip_layer, "src", "") or "") dst_ip = str(getattr(ip_layer, "dst", "") or "") elif IPv6 in pkt: ip_layer = pkt[IPv6] eth_type_raw = 0x86DD protocol = int(getattr(ip_layer, "nh", 0) or 0) src_ip = str(getattr(ip_layer, "src", "") or "") dst_ip = str(getattr(ip_layer, "dst", "") or "") elif ARP in pkt: arp_layer = pkt[ARP] eth_type_raw = 0x0806 src_ip = str(getattr(arp_layer, "psrc", "") or "") dst_ip = str(getattr(arp_layer, "pdst", "") or "") else: return None src_port = 0 dst_port = 0 if TCP in pkt: src_port = int(getattr(pkt[TCP], "sport", 0) or 0) dst_port = int(getattr(pkt[TCP], "dport", 0) or 0) elif UDP in pkt: src_port = int(getattr(pkt[UDP], "sport", 0) or 0) dst_port = int(getattr(pkt[UDP], "dport", 0) or 0) if not src_ip or not dst_ip or eth_type_raw is None: return None return _packet_signature(iface, eth_type_raw, protocol, src_ip, dst_ip, src_port, dst_port, len(pkt)) def _packet_observed_at_ms(pkt: Any) -> int: try: packet_time = float(getattr(pkt, "time", 0.0) or 0.0) if packet_time > 0: return int(packet_time * 1000) except Exception: pass return int(time.time() * 1000) def _stream_key(event: Dict[str, Any]) -> Optional[Tuple[str, str, int]]: iface = str(event.get("iface") or "") tcp_stream = event.get("tcp_stream") if tcp_stream is not None: return (iface, "tcp", int(tcp_stream)) udp_stream = event.get("udp_stream") if udp_stream is not None: return (iface, "udp", int(udp_stream)) return None def _protocol_stack(layers: Dict[str, Any]) -> List[str]: frame_layer = layers.get("frame") raw_protocols = _find_value(frame_layer, "frame_protocols", "protocols") text = _safe_text(raw_protocols) if text: return [part.strip() for part in text.split(":") if part.strip()] return [str(key) for key in layers.keys() if key] def _display_protocol_name(token: Optional[str]) -> Optional[str]: text = _safe_text(token) if text is None: return None normalized = text.replace("-", "_").replace(".", "_").lower() mapping = { "http": "HTTP", "http2": "HTTP2", "dns": "DNS", "tls": "TLS", "ssl": "TLS", "quic": "QUIC", "tcp": "TCP", "udp": "UDP", "arp": "ARP", "icmp": "ICMP", "icmpv6": "ICMPV6", "ip": "IP", "ipv6": "IPv6", } return mapping.get(normalized, text.upper()) def _meaningful_protocol(protocol_stack: List[str], layers: Dict[str, Any]) -> Optional[str]: for token in reversed(protocol_stack): normalized = token.replace("-", "_").replace(".", "_").lower() if normalized not in _GENERIC_PROTOCOLS: return _display_protocol_name(token) for token in reversed(list(layers.keys())): normalized = str(token).replace("-", "_").replace(".", "_").lower() if normalized not in _GENERIC_PROTOCOLS: return _display_protocol_name(str(token)) if protocol_stack: return _display_protocol_name(protocol_stack[-1]) return None def _eth_type_from_protocol_stack(protocol_stack: List[str]) -> Optional[int]: normalized = {token.replace("-", "_").replace(".", "_").lower() for token in protocol_stack} if "arp" in normalized: return 0x0806 if "ipv6" in normalized: return 0x86DD if "ip" in normalized: return 0x0800 return None def _http_metadata(layers: Dict[str, Any]) -> Dict[str, Any]: http_layer = layers.get("http") if not isinstance(http_layer, dict): return {} return _jsonable( { "method": _safe_text(_find_value(http_layer, "request_method", "method")), "uri": _safe_text(_find_value(http_layer, "request_uri", "uri", "full_uri")), "host": _safe_text(_find_value(http_layer, "host")), "user_agent": _safe_text(_find_value(http_layer, "user_agent")), "response_code": _safe_int(_find_value(http_layer, "response_code", "code")), "response_phrase": _safe_text(_find_value(http_layer, "response_phrase", "phrase")), "server": _safe_text(_find_value(http_layer, "server")), "content_type": _safe_text(_find_value(http_layer, "content_type")), } ) def _tls_metadata(layers: Dict[str, Any]) -> Dict[str, Any]: tls_layer = layers.get("tls") or layers.get("ssl") if not isinstance(tls_layer, dict): return {} return _jsonable( { "server_name": _safe_text(_find_value(tls_layer, "handshake_extensions_server_name", "server_name")), "handshake_version": _safe_text(_find_value(tls_layer, "handshake_version", "record_version", "version")), "alpn": _safe_text(_find_value(tls_layer, "handshake_extensions_alpn_str", "alpn_str")), } ) def _dns_metadata(layers: Dict[str, Any]) -> Dict[str, Any]: dns_layer = layers.get("dns") if not isinstance(dns_layer, dict): return {} return _jsonable( { "is_response": _safe_bool_flag(_find_value(dns_layer, "flags_response", "response")), "query_name": _safe_text(_find_value(dns_layer, "qry_name", "qry_name_len")), "query_type": _safe_text(_find_value(dns_layer, "qry_type")), "response_name": _safe_text(_find_value(dns_layer, "resp_name")), "a": _coerce_json_value(_find_value(dns_layer, "a")), "aaaa": _coerce_json_value(_find_value(dns_layer, "aaaa")), "cname": _coerce_json_value(_find_value(dns_layer, "cname")), } ) def _tcp_flags_summary(flags_value: Optional[int], payload_len: int) -> Optional[str]: if flags_value is None: return None syn = bool(flags_value & 0x02) ack = bool(flags_value & 0x10) fin = bool(flags_value & 0x01) rst = bool(flags_value & 0x04) psh = bool(flags_value & 0x08) urg = bool(flags_value & 0x20) if syn and ack: return "SYN-ACK" if syn: return "SYN" if rst and ack: return "RST-ACK" if rst: return "RST" if fin and ack: return "FIN-ACK" if fin: return "FIN" if psh and ack: return "PSH-ACK" if psh: return "PSH" if ack and payload_len == 0: return "ACK" if urg and ack: return "URG-ACK" return None def _tcp_flag_names(flags_value: Optional[int]) -> List[str]: if flags_value is None: return [] names: List[str] = [] for bit, name in ( (0x01, "FIN"), (0x02, "SYN"), (0x04, "RST"), (0x08, "PSH"), (0x10, "ACK"), (0x20, "URG"), (0x40, "ECE"), (0x80, "CWR"), ): if flags_value & bit: names.append(name) return names def _extract_event(record: Dict[str, Any], fallback_iface: str) -> Optional[Dict[str, Any]]: layers_raw = record.get("layers") if not isinstance(layers_raw, dict): return None layers = _coerce_json_value(layers_raw) if not isinstance(layers, dict): return None protocol_stack = _protocol_stack(layers) frame_layer = layers.get("frame") ip_layer = layers.get("ip") ipv6_layer = layers.get("ipv6") tcp_layer = layers.get("tcp") udp_layer = layers.get("udp") observed_at_ms = ( _parse_time_ms(record.get("timestamp")) or _parse_time_ms(_find_value(frame_layer, "frame_time_epoch", "time_epoch", "time")) or int(time.time() * 1000) ) iface = _safe_text(_find_value(frame_layer, "frame_interface_name", "interface_name")) or fallback_iface length = _safe_int(_find_value(frame_layer, "frame_len", "len")) eth_type_raw = _eth_type_from_protocol_stack(protocol_stack) protocol = _safe_int(_find_value(ip_layer, "ip_proto", "proto")) if protocol is None: protocol = _safe_int(_find_value(ipv6_layer, "ipv6_nxt", "nxt")) src_ip = ( _safe_text(_find_value(ip_layer, "ip_src", "src")) or _safe_text(_find_value(ipv6_layer, "ipv6_src", "src")) or _safe_text(_find_value(layers.get("arp"), "src_proto_ipv4", "src_proto_ipv6")) ) dst_ip = ( _safe_text(_find_value(ip_layer, "ip_dst", "dst")) or _safe_text(_find_value(ipv6_layer, "ipv6_dst", "dst")) or _safe_text(_find_value(layers.get("arp"), "dst_proto_ipv4", "dst_proto_ipv6")) ) src_port = _safe_int(_find_value(tcp_layer, "srcport")) or _safe_int(_find_value(udp_layer, "srcport")) or 0 dst_port = _safe_int(_find_value(tcp_layer, "dstport")) or _safe_int(_find_value(udp_layer, "dstport")) or 0 tcp_flags = _safe_int(_find_value(tcp_layer, "flags")) tcp_len = _safe_int(_find_value(tcp_layer, "len")) or 0 display_protocol = ( _safe_text(_find_value(frame_layer, "col_protocol", "protocol")) or _meaningful_protocol(protocol_stack, layers) ) info_col = _safe_text(_find_value(frame_layer, "col_info", "info")) if length is None or src_ip is None or dst_ip is None: return None return { "iface": iface, "observed_at_ms": observed_at_ms, "length": length, "eth_type_raw": eth_type_raw, "protocol": protocol, "src_ip": src_ip, "dst_ip": dst_ip, "src_port": src_port, "dst_port": dst_port, "tcp_stream": _safe_int(_find_value(tcp_layer, "stream")), "udp_stream": _safe_int(_find_value(udp_layer, "stream")), "tcp_flags": tcp_flags, "tcp_seq_raw": _safe_int(_find_value(tcp_layer, "seq_raw")), "tcp_ack_raw": _safe_int(_find_value(tcp_layer, "ack_raw")), "tcp_len": tcp_len, "tcp_retransmission": _safe_bool_flag(_find_value(tcp_layer, "analysis_retransmission")), "tcp_fast_retransmission": _safe_bool_flag(_find_value(tcp_layer, "analysis_fast_retransmission")), "tcp_spurious_retransmission": _safe_bool_flag(_find_value(tcp_layer, "analysis_spurious_retransmission")), "tcp_keep_alive": _safe_bool_flag(_find_value(tcp_layer, "analysis_keep_alive")), "tcp_keep_alive_ack": _safe_bool_flag(_find_value(tcp_layer, "analysis_keep_alive_ack")), "tcp_duplicate_ack": _safe_bool_flag(_find_value(tcp_layer, "analysis_duplicate_ack")), "arp_opcode": _safe_int(_find_value(layers.get("arp"), "opcode")), "icmp_type": _safe_int(_find_value(layers.get("icmp"), "type")) or _safe_int(_find_value(layers.get("icmpv6"), "type")), "icmp_code": _safe_int(_find_value(layers.get("icmp"), "code")) or _safe_int(_find_value(layers.get("icmpv6"), "code")), "protocol_col": display_protocol, "info_col": info_col, "frame_protocols": ":".join(protocol_stack) if protocol_stack else None, "protocol_stack": protocol_stack, "http": _http_metadata(layers), "tls": _tls_metadata(layers), "dns": _dns_metadata(layers), "layers": layers, } def _derive_flow_context(event: Dict[str, Any]) -> Dict[str, Any]: protocol_stack = list(event.get("protocol_stack") or []) http_meta = dict(event.get("http") or {}) tls_meta = dict(event.get("tls") or {}) dns_meta = dict(event.get("dns") or {}) app_protocol: Optional[str] = None app_category: Optional[str] = None app_hostname: Optional[str] = None app_is_encrypted: Optional[bool] = None if http_meta: app_protocol = "HTTP" app_category = "Web" app_hostname = http_meta.get("host") app_is_encrypted = False elif dns_meta: app_protocol = "DNS" app_category = "Infrastructure" app_hostname = dns_meta.get("query_name") or dns_meta.get("response_name") app_is_encrypted = False elif tls_meta: app_protocol = "TLS" app_category = "Encrypted" app_hostname = tls_meta.get("server_name") app_is_encrypted = True else: app_protocol = _safe_text(event.get("protocol_col")) normalized = str(app_protocol or "").upper() if normalized == "TCP": app_category = app_category or "Transport" elif normalized == "UDP": app_category = app_category or "Transport" elif normalized in {"ARP", "ICMP", "ICMPV6", "IP", "IPV6"}: app_category = app_category or "Network" elif app_protocol and app_category is None: app_category = "Protocol" return { "app_protocol": app_protocol, "app_master_protocol": app_protocol, "app_category": app_category, "app_confidence": "high" if app_protocol else None, "app_hostname": app_hostname, "app_is_encrypted": app_is_encrypted, "last_seen_ms": int(event.get("observed_at_ms") or 0), } def _build_enrichment(event: Dict[str, Any], flow_context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: flow_context = dict(flow_context or {}) derived_context = _derive_flow_context(event) protocol_stack = list(event.get("protocol_stack") or []) app_protocol = derived_context.get("app_protocol") or flow_context.get("app_protocol") app_category = derived_context.get("app_category") or flow_context.get("app_category") app_hostname = derived_context.get("app_hostname") or flow_context.get("app_hostname") app_is_encrypted = ( derived_context.get("app_is_encrypted") if derived_context.get("app_is_encrypted") is not None else flow_context.get("app_is_encrypted") ) app_confidence = derived_context.get("app_confidence") or flow_context.get("app_confidence") tcp_flags = _safe_int(event.get("tcp_flags")) tcp_len = int(event.get("tcp_len") or 0) tcp_packet_type = _tcp_flags_summary(tcp_flags, tcp_len) tcp_flag_names = _tcp_flag_names(tcp_flags) tshark_meta = _jsonable( { "observed_at_ms": event.get("observed_at_ms"), "protocol": event.get("protocol_col"), "info": event.get("info_col"), "frame_protocols": event.get("frame_protocols"), "protocol_stack": protocol_stack, "length": event.get("length"), "eth_type_raw": event.get("eth_type_raw"), "tcp_stream": event.get("tcp_stream"), "udp_stream": event.get("udp_stream"), "tcp_packet_type": tcp_packet_type, "tcp_flag_names": tcp_flag_names, } ) dpi_metadata = _jsonable( { "tshark": tshark_meta, "layers": event.get("layers"), "tcp": _jsonable( { "stream": event.get("tcp_stream"), "flags": tcp_flags, "packet_type": tcp_packet_type, "flag_names": tcp_flag_names, "seq_raw": event.get("tcp_seq_raw"), "ack_raw": event.get("tcp_ack_raw"), "payload_len": tcp_len, "retransmission": event.get("tcp_retransmission"), "fast_retransmission": event.get("tcp_fast_retransmission"), "spurious_retransmission": event.get("tcp_spurious_retransmission"), "keep_alive": event.get("tcp_keep_alive"), "keep_alive_ack": event.get("tcp_keep_alive_ack"), "duplicate_ack": event.get("tcp_duplicate_ack"), } ), "udp": _jsonable({"stream": event.get("udp_stream")}), "icmp": _jsonable({"type": event.get("icmp_type"), "code": event.get("icmp_code")}), "arp": _jsonable({"opcode": event.get("arp_opcode")}), "http": dict(event.get("http") or {}), "tls": dict(event.get("tls") or {}), "dns": dict(event.get("dns") or {}), } ) return { "app_protocol": app_protocol, "app_master_protocol": app_protocol, "app_category": app_category, "app_confidence": app_confidence, "app_hostname": app_hostname, "app_is_encrypted": app_is_encrypted, "dpi_metadata": dpi_metadata or None, "capture_sources": ["tshark"], } def _has_useful_enrichment(enrichment: Dict[str, Any]) -> bool: if enrichment.get("app_protocol") is not None: return True if enrichment.get("app_hostname") is not None: return True dpi_metadata = enrichment.get("dpi_metadata") or {} if not isinstance(dpi_metadata, dict): return False for value in dpi_metadata.values(): if isinstance(value, dict) and value: return True return bool(dpi_metadata) def _build_stream_enrichment(flow_context: Dict[str, Any]) -> Dict[str, Any]: return { "app_protocol": flow_context.get("app_protocol"), "app_master_protocol": flow_context.get("app_master_protocol"), "app_category": flow_context.get("app_category"), "app_confidence": flow_context.get("app_confidence"), "app_hostname": flow_context.get("app_hostname"), "app_is_encrypted": flow_context.get("app_is_encrypted"), "capture_sources": ["tshark"], } def _is_generic_stream_protocol(protocol_name: Optional[str]) -> bool: return str(protocol_name or "").replace("-", "_").replace(".", "_").lower() in _GENERIC_PROTOCOLS class TsharkManager: """Own long-lived tshark subprocesses and recent packet metadata cache.""" def __init__(self) -> None: self._enabled = settings.tshark_enabled self._workers: Dict[str, Dict[str, Any]] = {} self._cache: Dict[Tuple[str, int, int, str, str, int, int, int], List[Dict[str, Any]]] = {} self._flow_context: Dict[Tuple[str, str, int], Dict[str, Any]] = {} self._last_error_by_iface: Dict[str, str] = {} self._stats: Dict[str, Any] = { "raw_lines_total": 0, "events_total": 0, "events_useful": 0, "parse_failures": 0, "lookup_hits": 0, "backfill_updates_total": 0, "last_event_by_iface": {}, "last_unparsed_line_by_iface": {}, } self._lock = threading.Lock() @property def enabled(self) -> bool: return self._enabled def update_interfaces(self, interfaces: Iterable[str]) -> None: if not self._enabled: return targets = {iface.strip() for iface in interfaces if iface and iface.strip()} with self._lock: current = set(self._workers.keys()) for iface in sorted(current - targets): self._stop_worker_locked(iface) for iface in sorted(targets - current): self._start_worker_locked(iface) self._purge_cache_locked() def stop(self) -> None: with self._lock: for iface in list(self._workers.keys()): self._stop_worker_locked(iface) self._cache.clear() self._flow_context.clear() def lookup_packet(self, pkt: Any, iface: Optional[str]) -> Dict[str, Any]: if not self._enabled or not iface: return {} signature = _signature_from_packet(pkt, iface) if signature is None: return {} packet_time_ms = _packet_observed_at_ms(pkt) with self._lock: self._purge_cache_locked(now_ms=packet_time_ms) entries = self._cache.get(signature) if not entries: return {} window_ms = settings.tshark_match_window_ms best_index: Optional[int] = None best_delta: Optional[int] = None for index, event in enumerate(entries): delta = abs(int(event.get("observed_at_ms") or 0) - packet_time_ms) if delta > window_ms: continue if best_delta is None or delta < best_delta: best_index = index best_delta = delta if best_index is None: return {} event = entries.pop(best_index) if not entries: self._cache.pop(signature, None) self._stats["lookup_hits"] += 1 stream_key = _stream_key(event) flow_context = dict(self._flow_context.get(stream_key) or {}) if stream_key is not None else None return _build_enrichment(event, flow_context=flow_context) def get_debug_snapshot(self) -> Dict[str, Any]: with self._lock: self._purge_cache_locked() return { "enabled": self._enabled, "workers": { iface: { "running": bool(worker.get("process") and worker["process"].poll() is None), "thread_alive": bool(worker.get("reader") and worker["reader"].is_alive()), "last_error": self._last_error_by_iface.get(iface), } for iface, worker in self._workers.items() }, "cache_entries": sum(len(entries) for entries in self._cache.values()), "flow_context_entries": len(self._flow_context), "last_error_by_iface": dict(self._last_error_by_iface), "stats": dict(self._stats), } def _start_worker_locked(self, iface: str) -> None: cmd = [ "tshark", "-l", "-n", "-Q", "-i", iface, "-T", "ek", ] if settings.tshark_try_heuristic_first: cmd.extend(["-o", "tcp.try_heuristic_first:true"]) cmd.extend(["-o", "udp.try_heuristic_first:true"]) if settings.tshark_display_filter: cmd.extend(["-Y", settings.tshark_display_filter]) logger.info("Starting tshark worker for %s", iface) try: process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, env=os.environ.copy(), start_new_session=True, ) except Exception: logger.exception("Failed to start tshark worker for %s", iface) self._last_error_by_iface[iface] = "failed_to_start_process" return reader = threading.Thread( target=self._read_loop, args=(iface, process), daemon=True, name=f"tshark-reader-{iface}", ) self._workers[iface] = { "process": process, "reader": reader, } reader.start() def _stop_worker_locked(self, iface: str) -> None: worker = self._workers.pop(iface, None) if worker is None: return process = worker.get("process") reader = worker.get("reader") if process is not None and process.poll() is None: try: os.killpg(os.getpgid(process.pid), signal.SIGTERM) process.wait(timeout=settings.tshark_process_stop_timeout_seconds) except subprocess.TimeoutExpired: try: os.killpg(os.getpgid(process.pid), signal.SIGKILL) except ProcessLookupError: pass except Exception: logger.exception("Failed to stop tshark worker for %s cleanly", iface) if reader is not None and reader.is_alive(): reader.join(timeout=settings.tshark_reader_join_timeout_seconds) stale_keys = [key for key in self._cache.keys() if key[0] == iface] for key in stale_keys: self._cache.pop(key, None) stale_flow_keys = [key for key in self._flow_context.keys() if key[0] == iface] for key in stale_flow_keys: self._flow_context.pop(key, None) def _read_loop(self, iface: str, process: subprocess.Popen[str]) -> None: stdout = process.stdout if stdout is None: return for line in stdout: text = line.rstrip("\n") if not text: continue with self._lock: self._stats["raw_lines_total"] += 1 try: payload = json.loads(text) except json.JSONDecodeError: with self._lock: self._stats["parse_failures"] += 1 self._stats["last_unparsed_line_by_iface"][iface] = text logger.debug("Ignoring unparsable tshark line on %s: %s", iface, text) continue if not isinstance(payload, dict): continue if "index" in payload and "layers" not in payload: continue event = _extract_event(payload, iface) if event is None: with self._lock: self._stats["parse_failures"] += 1 self._stats["last_unparsed_line_by_iface"][iface] = text continue signature = _packet_signature( str(event["iface"]), int(event.get("eth_type_raw") or 0), _safe_int(event.get("protocol")), str(event["src_ip"]), str(event["dst_ip"]), int(event["src_port"]), int(event["dst_port"]), int(event["length"]), ) stream_key = _stream_key(event) flow_context: Optional[Dict[str, Any]] = None with self._lock: if stream_key is not None: flow_context = dict(self._flow_context.get(stream_key) or {}) derived_context = _derive_flow_context(event) if stream_key is not None and derived_context.get("app_protocol") is not None: merged_context = dict(flow_context or {}) merged_context.update({key: value for key, value in derived_context.items() if value is not None}) self._flow_context[stream_key] = merged_context flow_context = merged_context entries = self._cache.setdefault(signature, []) entries.append(event) self._stats["events_total"] += 1 self._stats["last_event_by_iface"][iface] = { "observed_at_ms": event.get("observed_at_ms"), "eth_type_raw": event.get("eth_type_raw"), "tcp_stream": event.get("tcp_stream"), "udp_stream": event.get("udp_stream"), "protocol_col": event.get("protocol_col"), "info_col": event.get("info_col"), "protocol": event.get("protocol"), "src_ip": event.get("src_ip"), "dst_ip": event.get("dst_ip"), "src_port": event.get("src_port"), "dst_port": event.get("dst_port"), "frame_protocols": event.get("frame_protocols"), "protocol_stack": event.get("protocol_stack"), } self._purge_cache_locked(now_ms=int(event["observed_at_ms"])) enrichment = _build_enrichment(event, flow_context=flow_context) if _has_useful_enrichment(enrichment): with self._lock: self._stats["events_useful"] += 1 self._schedule_backfill(event, enrichment, flow_context=flow_context or derived_context) rc = process.poll() if rc not in (0, None): logger.warning("tshark worker for %s exited with code %s", iface, rc) def _schedule_backfill( self, event: Dict[str, Any], enrichment: Dict[str, Any], *, flow_context: Optional[Dict[str, Any]] = None, ) -> None: web_db = getattr(shared_objects, "db", None) web_loop = getattr(shared_objects, "web_loop", None) if web_db is None or web_loop is None: return async def _backfill() -> None: updated = await web_db.backfill_packet_metadata( iface=str(event["iface"]), eth_type_raw=_safe_int(event.get("eth_type_raw")), src_ip=str(event["src_ip"]), dst_ip=str(event["dst_ip"]), src_port=int(event["src_port"]), dst_port=int(event["dst_port"]), protocol=_safe_int(event.get("protocol")), length=int(event["length"]), observed_at_ms=int(event["observed_at_ms"]), enrichment=enrichment, window_ms=settings.tshark_match_window_ms, ) if updated: with self._lock: self._stats["backfill_updates_total"] += updated stream_key = _stream_key(event) if stream_key is None or not flow_context: return stream_enrichment = _build_stream_enrichment(flow_context) if _is_generic_stream_protocol(stream_enrichment.get("app_protocol")) and stream_enrichment.get("app_hostname") is None: return if not any( stream_enrichment.get(key) is not None for key in ( "app_protocol", "app_master_protocol", "app_category", "app_confidence", "app_hostname", "app_is_encrypted", ) ): return stream_kind = str(stream_key[1]) stream_id = int(stream_key[2]) stream_updated = await web_db.backfill_stream_metadata( iface=str(event["iface"]), protocol=_safe_int(event.get("protocol")) or 0, stream_kind=stream_kind, stream_id=stream_id, observed_at_ms=int(event["observed_at_ms"]), enrichment=stream_enrichment, window_ms=max(settings.tshark_match_window_ms, 10_000), ) if stream_updated: logger.debug( "Backfilled tshark stream context for %s packets on %s %s.stream=%s", stream_updated, event["iface"], stream_kind, stream_id, ) try: future = asyncio.run_coroutine_threadsafe(_backfill(), web_loop) future.add_done_callback(self._log_backfill_result) except Exception: logger.exception("Failed to schedule tshark metadata backfill") @staticmethod def _log_backfill_result(future: Any) -> None: try: future.result() except Exception: logger.exception("tshark metadata backfill task failed") def _purge_cache_locked(self, now_ms: Optional[int] = None) -> None: if now_ms is None: now_ms = int(time.time() * 1000) expiry_ms = int(settings.tshark_cache_ttl_seconds * 1000) stale_keys: List[Tuple[str, int, int, str, str, int, int, int]] = [] for key, entries in self._cache.items(): fresh_entries = [ event for event in entries if now_ms - int(event.get("observed_at_ms") or 0) <= expiry_ms ] if fresh_entries: self._cache[key] = fresh_entries else: stale_keys.append(key) for key in stale_keys: self._cache.pop(key, None) stale_flow_keys = [] for key, payload in self._flow_context.items(): if now_ms - int(payload.get("last_seen_ms") or 0) > expiry_ms: stale_flow_keys.append(key) for key in stale_flow_keys: self._flow_context.pop(key, None) tshark_manager = TsharkManager()