tshark protocol restriction lifted, frontedn details for packet

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