From 5ea4dd388eddcc13ca4b33041076d01227c699fc Mon Sep 17 00:00:00 2001 From: malmert Date: Sun, 8 Mar 2026 16:54:28 +0100 Subject: [PATCH] tshark protocol restriction lifted, frontedn details for packet --- backend/src/utilities/database.py | 48 +- backend/src/utilities/tshark_manager.py | 637 ++++++++++++++--------- frontend/src/components/PacketViewer.tsx | 82 ++- 3 files changed, 481 insertions(+), 286 deletions(-) diff --git a/backend/src/utilities/database.py b/backend/src/utilities/database.py index e7920e8..95e165a 100644 --- a/backend/src/utilities/database.py +++ b/backend/src/utilities/database.py @@ -271,11 +271,12 @@ class DatabasePool: self, *, iface: str, + eth_type_raw: Optional[int], src_ip: str, dst_ip: str, src_port: int, dst_port: int, - protocol: int, + protocol: Optional[int], length: int, observed_at_ms: int, enrichment: Dict[str, Any], @@ -297,35 +298,39 @@ class DatabasePool: UPDATE packets SET updated_at = NOW(), - app_protocol = COALESCE(packets.app_protocol, $10), - app_master_protocol = COALESCE(packets.app_master_protocol, $11), - app_category = COALESCE(packets.app_category, $12), - app_confidence = COALESCE(packets.app_confidence, $13), - app_hostname = COALESCE(packets.app_hostname, $14), - app_is_encrypted = COALESCE(packets.app_is_encrypted, $15), + app_protocol = COALESCE(packets.app_protocol, $11), + app_master_protocol = COALESCE(packets.app_master_protocol, $12), + app_category = COALESCE(packets.app_category, $13), + app_confidence = COALESCE(packets.app_confidence, $14), + app_hostname = COALESCE(packets.app_hostname, $15), + app_is_encrypted = COALESCE(packets.app_is_encrypted, $16), dpi_metadata = CASE - WHEN $16::jsonb IS NULL THEN packets.dpi_metadata - WHEN packets.dpi_metadata IS NULL THEN $16::jsonb - ELSE packets.dpi_metadata || $16::jsonb + WHEN $17::jsonb IS NULL THEN packets.dpi_metadata + WHEN packets.dpi_metadata IS NULL THEN $17::jsonb + ELSE packets.dpi_metadata || $17::jsonb END, capture_sources = ( SELECT ARRAY( SELECT DISTINCT source FROM unnest( COALESCE(packets.capture_sources, ARRAY[]::text[]) || - COALESCE($17::text[], ARRAY[]::text[]) + COALESCE($18::text[], ARRAY[]::text[]) ) AS source ) ) WHERE - ip_proto_raw = $1 - AND (capture_iface = $2 OR ingress_if = $2 OR egress_if = $2) - AND src_ip = $3::inet - AND dst_ip = $4::inet - AND src_port = $5 - AND dst_port = $6 - AND length = $7 - AND timestamp BETWEEN $8 AND $9 + ( + ($1::int IS NOT NULL AND ip_proto_raw = $1) + OR + ($1::int IS NULL AND $2::int IS NOT NULL AND eth_type_raw = $2) + ) + AND (capture_iface = $3 OR ingress_if = $3 OR egress_if = $3) + AND src_ip = $4::inet + AND dst_ip = $5::inet + AND COALESCE(src_port, 0) = $6 + AND COALESCE(dst_port, 0) = $7 + AND length = $8 + AND timestamp BETWEEN $9 AND $10 AND ( packets.app_protocol IS NULL OR packets.app_master_protocol IS NULL @@ -333,12 +338,13 @@ class DatabasePool: OR packets.app_confidence IS NULL OR packets.app_hostname IS NULL OR packets.app_is_encrypted IS NULL - OR ($16::jsonb IS NOT NULL) - OR (COALESCE(array_length($17::text[], 1), 0) > 0) + OR ($17::jsonb IS NOT NULL) + OR (COALESCE(array_length($18::text[], 1), 0) > 0) ) RETURNING * """, protocol, + eth_type_raw, iface, src_ip, dst_ip, diff --git a/backend/src/utilities/tshark_manager.py b/backend/src/utilities/tshark_manager.py index 448febf..1316229 100644 --- a/backend/src/utilities/tshark_manager.py +++ b/backend/src/utilities/tshark_manager.py @@ -3,73 +3,38 @@ 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 IP, IPv6, TCP, UDP # type: ignore +from scapy.all import ARP, IP, IPv6, TCP, UDP # type: ignore from src.config import settings logger = logging.getLogger("tshark_manager") -_FIELDS: List[str] = [ - "frame.time_epoch", - "frame.interface_name", - "frame.len", - "_ws.col.Protocol", - "_ws.col.Info", - "ip.src", - "ipv6.src", - "ip.dst", - "ipv6.dst", - "ip.proto", - "ipv6.nxt", - "tcp.stream", - "udp.stream", - "tcp.srcport", - "udp.srcport", - "tcp.dstport", - "udp.dstport", - "tcp.flags", - "tcp.seq_raw", - "tcp.ack_raw", - "tcp.len", - "tcp.analysis.retransmission", - "tcp.analysis.fast_retransmission", - "tcp.analysis.spurious_retransmission", - "tcp.analysis.keep_alive", - "tcp.analysis.keep_alive_ack", - "tcp.analysis.duplicate_ack", - "arp.opcode", - "icmp.type", - "icmp.code", - "icmpv6.type", - "icmpv6.code", - "frame.protocols", - "http.request.method", - "http.request.uri", - "http.host", - "http.user_agent", - "http.response.code", - "http.response.phrase", - "http.server", - "http.content_type", - "tls.handshake.extensions_server_name", - "tls.handshake.version", - "dns.flags.response", - "dns.qry.name", - "dns.qry.type", - "dns.resp.name", - "dns.a", - "dns.aaaa", - "dns.cname", -] +_GENERIC_PROTOCOLS = { + "", + "frame", + "eth", + "ethertype", + "ip", + "ipv6", + "tcp", + "udp", + "data", + "data_text_lines", + "text", + "text_lines", + "geninfo", +} def _safe_text(value: Any) -> Optional[str]: @@ -84,7 +49,7 @@ def _safe_int(value: Any) -> Optional[int]: text = _safe_text(value) if text is None: return None - return int(text) + return int(text, 0) except Exception: return None @@ -110,6 +75,267 @@ def _safe_bool_flag(value: Any) -> Optional[bool]: 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 @@ -164,161 +390,95 @@ def _tcp_flag_names(flags_value: Optional[int]) -> List[str]: return names -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 _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 _packet_signature( - iface: str, - protocol: int, - src_ip: str, - dst_ip: str, - src_port: int, - dst_port: int, - length: int, -) -> Tuple[str, int, str, str, int, int, int]: - return (str(iface or ""), int(protocol), 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, str, str, int, int, int]]: - if IP in pkt: - ip_layer = pkt[IP] - 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] - 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 "") - else: +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 - 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: - return None - return _packet_signature(iface, 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 _parse_line(line: str, fallback_iface: str) -> Optional[Dict[str, Any]]: - parts = line.rstrip("\n").split("\t") - if len(parts) < len(_FIELDS): - parts.extend([""] * (len(_FIELDS) - len(parts))) - row = dict(zip(_FIELDS, parts)) - - timestamp = _safe_float(row["frame.time_epoch"]) - length = _safe_int(row["frame.len"]) - protocol_col = _safe_text(row["_ws.col.Protocol"]) - info_col = _safe_text(row["_ws.col.Info"]) - protocol = _safe_int(row["ip.proto"]) or _safe_int(row["ipv6.nxt"]) - tcp_stream = _safe_int(row["tcp.stream"]) - udp_stream = _safe_int(row["udp.stream"]) - src_ip = _safe_text(row["ip.src"]) or _safe_text(row["ipv6.src"]) - dst_ip = _safe_text(row["ip.dst"]) or _safe_text(row["ipv6.dst"]) - src_port = _safe_int(row["tcp.srcport"]) or _safe_int(row["udp.srcport"]) or 0 - dst_port = _safe_int(row["tcp.dstport"]) or _safe_int(row["udp.dstport"]) or 0 - iface = _safe_text(row["frame.interface_name"]) or fallback_iface - - if timestamp is None or length is None or protocol is None or src_ip is None or dst_ip is 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 - observed_at_ms = int(timestamp * 1000) 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": tcp_stream, - "udp_stream": udp_stream, - "tcp_flags": _safe_int(row["tcp.flags"]), - "tcp_seq_raw": _safe_int(row["tcp.seq_raw"]), - "tcp_ack_raw": _safe_int(row["tcp.ack_raw"]), - "tcp_len": _safe_int(row["tcp.len"]) or 0, - "tcp_retransmission": _safe_bool_flag(row["tcp.analysis.retransmission"]), - "tcp_fast_retransmission": _safe_bool_flag(row["tcp.analysis.fast_retransmission"]), - "tcp_spurious_retransmission": _safe_bool_flag(row["tcp.analysis.spurious_retransmission"]), - "tcp_keep_alive": _safe_bool_flag(row["tcp.analysis.keep_alive"]), - "tcp_keep_alive_ack": _safe_bool_flag(row["tcp.analysis.keep_alive_ack"]), - "tcp_duplicate_ack": _safe_bool_flag(row["tcp.analysis.duplicate_ack"]), - "arp_opcode": _safe_int(row["arp.opcode"]), - "icmp_type": _safe_int(row["icmp.type"]) or _safe_int(row["icmpv6.type"]), - "icmp_code": _safe_int(row["icmp.code"]) or _safe_int(row["icmpv6.code"]), - "protocol_col": protocol_col, + "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": _safe_text(row["frame.protocols"]), - "http": _jsonable( - { - "method": _safe_text(row["http.request.method"]), - "uri": _safe_text(row["http.request.uri"]), - "host": _safe_text(row["http.host"]), - "user_agent": _safe_text(row["http.user_agent"]), - "response_code": _safe_int(row["http.response.code"]), - "response_phrase": _safe_text(row["http.response.phrase"]), - "server": _safe_text(row["http.server"]), - "content_type": _safe_text(row["http.content_type"]), - } - ), - "tls": _jsonable( - { - "server_name": _safe_text(row["tls.handshake.extensions_server_name"]), - "handshake_version": _safe_text(row["tls.handshake.version"]), - } - ), - "dns": _jsonable( - { - "is_response": _safe_bool_flag(row["dns.flags.response"]), - "query_name": _safe_text(row["dns.qry.name"]), - "query_type": _safe_text(row["dns.qry.type"]), - "response_name": _safe_text(row["dns.resp.name"]), - "a": _safe_text(row["dns.a"]), - "aaaa": _safe_text(row["dns.aaaa"]), - "cname": _safe_text(row["dns.cname"]), - } - ), + "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_col = _safe_text(event.get("protocol_col")) - protocols = str(event.get("frame_protocols") or "") + 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 {}) @@ -338,19 +498,20 @@ def _derive_flow_context(event: Dict[str, Any]) -> Dict[str, Any]: app_category = "Infrastructure" app_hostname = dns_meta.get("query_name") or dns_meta.get("response_name") app_is_encrypted = False - elif tls_meta or "quic" in protocols.lower(): - app_protocol = "QUIC" if "quic" in protocols.lower() and not tls_meta else "TLS" + elif tls_meta: + app_protocol = "TLS" app_category = "Encrypted" app_hostname = tls_meta.get("server_name") app_is_encrypted = True - elif protocol_col: - app_protocol = protocol_col + else: + app_protocol = _safe_text(event.get("protocol_col")) - if app_protocol == "TCP": + normalized = str(app_protocol or "").upper() + if normalized == "TCP": app_category = app_category or "Transport" - elif app_protocol == "UDP": + elif normalized == "UDP": app_category = app_category or "Transport" - elif app_protocol in {"ARP", "ICMP", "ICMPV6", "IP", "IPV6"}: + 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" @@ -367,14 +528,9 @@ def _derive_flow_context(event: Dict[str, Any]) -> Dict[str, Any]: def _build_enrichment(event: Dict[str, Any], flow_context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - http_meta = dict(event.get("http") or {}) - tls_meta = dict(event.get("tls") or {}) - dns_meta = dict(event.get("dns") or {}) - protocols = str(event.get("frame_protocols") or "") - protocol_col = _safe_text(event.get("protocol_col")) - info_col = _safe_text(event.get("info_col")) 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") @@ -394,10 +550,12 @@ def _build_enrichment(event: Dict[str, Any], flow_context: Optional[Dict[str, An tshark_meta = _jsonable( { "observed_at_ms": event.get("observed_at_ms"), - "protocol": protocol_col, - "info": info_col, + "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, @@ -408,6 +566,7 @@ def _build_enrichment(event: Dict[str, Any], flow_context: Optional[Dict[str, An dpi_metadata = _jsonable( { "tshark": tshark_meta, + "layers": event.get("layers"), "tcp": _jsonable( { "stream": event.get("tcp_stream"), @@ -428,9 +587,9 @@ def _build_enrichment(event: Dict[str, Any], flow_context: Optional[Dict[str, An "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": http_meta, - "tls": tls_meta, - "dns": dns_meta, + "http": dict(event.get("http") or {}), + "tls": dict(event.get("tls") or {}), + "dns": dict(event.get("dns") or {}), } ) @@ -473,7 +632,7 @@ def _build_stream_enrichment(flow_context: Dict[str, Any]) -> Dict[str, Any]: def _is_generic_stream_protocol(protocol_name: Optional[str]) -> bool: - return str(protocol_name or "").upper() in {"", "TCP", "UDP", "IP", "IPV6", "ETH"} + return str(protocol_name or "").replace("-", "_").replace(".", "_").lower() in _GENERIC_PROTOCOLS class TsharkManager: @@ -482,7 +641,7 @@ class TsharkManager: def __init__(self) -> None: self._enabled = settings.tshark_enabled self._workers: Dict[str, Dict[str, Any]] = {} - self._cache: Dict[Tuple[str, int, str, str, int, int, int], List[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] = { @@ -587,23 +746,13 @@ class TsharkManager: "-i", iface, "-T", - "fields", - "-E", - "header=n", - "-E", - "separator=\t", - "-E", - "quote=n", - "-E", - "occurrence=f", + "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]) - for field in _FIELDS: - cmd.extend(["-e", field]) logger.info("Starting tshark worker for %s", iface) try: @@ -675,17 +824,31 @@ class TsharkManager: with self._lock: self._stats["raw_lines_total"] += 1 - event = _parse_line(text, iface) - if event is None: + 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["protocol"]), + 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"]), @@ -702,7 +865,7 @@ class TsharkManager: 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({k: v for k, v in derived_context.items() if v is not None}) + 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, []) @@ -710,20 +873,18 @@ class TsharkManager: 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"), - "tcp_packet_type": _tcp_flags_summary(_safe_int(event.get("tcp_flags")), int(event.get("tcp_len") or 0)), "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"), - "http": event.get("http"), - "tls": event.get("tls"), - "dns": event.get("dns"), + "protocol_stack": event.get("protocol_stack"), } self._purge_cache_locked(now_ms=int(event["observed_at_ms"])) @@ -752,11 +913,12 @@ class TsharkManager: 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=int(event["protocol"]), + protocol=_safe_int(event.get("protocol")), length=int(event["length"]), observed_at_ms=int(event["observed_at_ms"]), enrichment=enrichment, @@ -765,15 +927,6 @@ class TsharkManager: if updated: with self._lock: self._stats["backfill_updates_total"] += updated - logger.debug( - "Backfilled tshark metadata for %s packets on %s %s:%s -> %s:%s", - updated, - event["iface"], - event["src_ip"], - event["src_port"], - event["dst_ip"], - event["dst_port"], - ) stream_key = _stream_key(event) if stream_key is None or not flow_context: @@ -799,7 +952,7 @@ class TsharkManager: stream_id = int(stream_key[2]) stream_updated = await web_db.backfill_stream_metadata( iface=str(event["iface"]), - protocol=int(event["protocol"]), + protocol=_safe_int(event.get("protocol")) or 0, stream_kind=stream_kind, stream_id=stream_id, observed_at_ms=int(event["observed_at_ms"]), @@ -833,7 +986,7 @@ class TsharkManager: now_ms = int(time.time() * 1000) expiry_ms = int(settings.tshark_cache_ttl_seconds * 1000) - stale_keys: List[Tuple[str, int, str, str, int, int, int]] = [] + stale_keys: List[Tuple[str, int, int, str, str, int, int, int]] = [] for key, entries in self._cache.items(): fresh_entries = [ event diff --git a/frontend/src/components/PacketViewer.tsx b/frontend/src/components/PacketViewer.tsx index f053331..44553e3 100644 --- a/frontend/src/components/PacketViewer.tsx +++ b/frontend/src/components/PacketViewer.tsx @@ -1,5 +1,5 @@ // src/components/PacketViewer.tsx -import { DeleteOutlined, DownloadOutlined, SyncOutlined } from '@ant-design/icons'; +import { DeleteOutlined, DownloadOutlined, EyeOutlined, SyncOutlined } from '@ant-design/icons'; import { Button, Card, @@ -47,6 +47,15 @@ function base64ToBlob(b64: string) { return new Blob([arr.buffer], { type: 'application/octet-stream' }); } +function formatJson(value: unknown) { + if (value == null) return '(no tshark data)'; + try { + return JSON.stringify(value, null, 2); + } catch { + return '(failed to format tshark data)'; + } +} + /** * Improved deterministic color generation * @@ -133,6 +142,7 @@ export default function PacketViewer(): ReactElement { const wsRef = useRef(null); const [hexModalOpen, setHexModalOpen] = useState(false); const [rawHex, setRawHex] = useState(null); + const [rawTshark, setRawTshark] = useState(null); const [rawB64ForDownload, setRawB64ForDownload] = useState(null); const [selectedPacket, setSelectedPacket] = useState(null); const queuedDuringPause = useRef([]); @@ -277,6 +287,7 @@ export default function PacketViewer(): ReactElement { return; } setRawHex(base64ToHex(row.raw_b64)); + setRawTshark(formatJson(row.dpi_metadata ?? null)); setRawB64ForDownload(row.raw_b64 ?? null); setSelectedPacket(row); setHexModalOpen(true); @@ -457,28 +468,34 @@ export default function PacketViewer(): ReactElement { { title: 'Actions', key: 'actions', - width: 180, + width: 96, render: (_: any, rec: PacketRow) => ( - - {rec.raw_b64 && ( + + icon={} + onClick={() => openRawModal(rec)} + aria-label="View raw and tshark data" + /> + + {rec.raw_b64 && ( + + + @@ -558,7 +585,16 @@ export default function PacketViewer(): ReactElement { } width={900} > - 
{rawHex}
+ +
+ Raw bytes + 
{rawHex}
+
+
+ TShark data + 
{rawTshark}
+
+ );