From 9c6eaaa7b26853097ac19678a9dca6ab3f266aed Mon Sep 17 00:00:00 2001 From: malmert Date: Sun, 12 Apr 2026 18:27:41 +0200 Subject: [PATCH] try new correlation af packet bridge mode --- backend/src/Models/packets.py | 33 +++++++++- backend/src/network_sniffer.py | 43 +++++++++++- backend/src/utilities/database.py | 17 ++++- backend/src/utilities/packet_tracker.py | 65 ++++++++++++++++++- .../src/components/PacketInspectModal.tsx | 11 ++++ frontend/src/components/PacketViewer.tsx | 40 +++++++++++- frontend/src/types/packets.ts | 14 ++++ setup_database.sh | 2 + 8 files changed, 214 insertions(+), 11 deletions(-) diff --git a/backend/src/Models/packets.py b/backend/src/Models/packets.py index 84bf173..115ef10 100644 --- a/backend/src/Models/packets.py +++ b/backend/src/Models/packets.py @@ -1,11 +1,26 @@ """Pydantic model for packet rows returned by the backend.""" from datetime import datetime -from typing import Optional, Union +from typing import Literal, Optional, Union from pydantic import BaseModel, ConfigDict, Field, IPvAnyAddress +class PacketObservationModel(BaseModel): + """One raw-capture or telemetry observation that contributed to a packet row.""" + + observation_type: Literal["capture", "telemetry"] + source: str + iface: Optional[str] = None + timestamp: Optional[str] = None + event_type: Optional[str] = None + capture_mode: Optional[str] = None + capture_session_id: Optional[str] = None + session_label: Optional[str] = None + session_kind: Optional[str] = None + reason: Optional[str] = None + + class PacketDBModel(BaseModel): """Normalized packet representation used across DB and API layers.""" @@ -48,6 +63,18 @@ class PacketDBModel(BaseModel): "dpi_metadata": {"method": "GET"}, "capture_metadata": {"capture_mode": "tc_ingress", "packet_id": "123456"}, "telemetry_metadata": {"event_type": "egress", "iface": "eth1", "packet_id": "123456"}, + "capture_observations": [ + { + "observation_type": "capture", + "source": "af_packet", + "iface": "eth0", + "timestamp": "2026-03-05T12:34:56.789000+00:00", + "capture_mode": "af_packet", + "capture_session_id": "session-1", + "session_label": "br0", + "session_kind": "bridge", + } + ], "verdict": "accept", "verdict_reason": "egress-observed", "verdict_confidence": "high", @@ -92,6 +119,10 @@ class PacketDBModel(BaseModel): dpi_metadata: Optional[dict] = Field(None, description="Raw metadata from DPI/flow enrichment.") capture_metadata: Optional[dict] = Field(None, description="Raw-capture metadata from the bridge tc ingress exporter.") telemetry_metadata: Optional[dict] = Field(None, description="Kernel telemetry details from eBPF collector.") + capture_observations: Optional[list[PacketObservationModel]] = Field( + None, + description="Ordered list of raw-capture and telemetry observations merged into this packet row.", + ) verdict: Optional[str] = None verdict_reason: Optional[str] = None verdict_confidence: Optional[str] = None diff --git a/backend/src/network_sniffer.py b/backend/src/network_sniffer.py index 5d5159b..3c807b8 100644 --- a/backend/src/network_sniffer.py +++ b/backend/src/network_sniffer.py @@ -98,6 +98,7 @@ class PacketInfo(TypedDict, total=False): app_risk_score: Optional[int] dpi_metadata: Optional[Dict[str, Any]] capture_metadata: Optional[Dict[str, Any]] + capture_observation: Optional[Dict[str, Any]] ip_id: Optional[int] icmp_type: Optional[int] icmp_code: Optional[int] @@ -203,6 +204,34 @@ def _merge_enrichment(pkt_info: PacketInfo, enrichment: Dict[str, Any]) -> None: pkt_info[key] = value +def _build_capture_observation( + *, + pkt_iface: str, + pkt_timestamp: datetime, + capture_source: str, + capture_session_id: Optional[str], + capture_metadata: Optional[Dict[str, Any]], +) -> Dict[str, Any]: + session = sessions.get(capture_session_id) if capture_session_id else None + session_is_bridge = bool(session and session.get("is_bridge")) + session_label = session.get("label") if isinstance(session, dict) else None + session_capture_mode = ( + session.get("capture_mode") + if isinstance(session, dict) and session.get("capture_mode") + else ("tc_ebpf" if capture_metadata else "af_packet") + ) + return { + "observation_type": "capture", + "source": capture_source, + "iface": pkt_iface, + "timestamp": pkt_timestamp.isoformat(), + "capture_mode": session_capture_mode, + "capture_session_id": capture_session_id, + "session_label": session_label, + "session_kind": "bridge" if session_is_bridge else "interface", + } + + def parse_packet( pkt, bridge_label: str, @@ -219,14 +248,17 @@ def parse_packet( logger.debug("Packet captured on %s (bridge_label %s)", pkt_iface, bridge_label) + pkt_timestamp = _packet_timestamp(pkt) + capture_source = "tc_ingress_raw" if capture_metadata else "af_packet" + pkt_info: PacketInfo = { - "timestamp": _packet_timestamp(pkt), + "timestamp": pkt_timestamp, "capture_session_id": capture_session_id, "iface": pkt_iface, "capture_iface": None, "length": len(pkt), "raw": bytes(pkt), - "capture_source": "tc_ingress_raw" if capture_metadata else "af_packet", + "capture_source": capture_source, "packet_id": capture_metadata.get("packet_id") if capture_metadata else None, "skb_mark": capture_metadata.get("skb_mark") if capture_metadata else None, "src_mac": None, @@ -250,6 +282,13 @@ def parse_packet( "app_risk_score": None, "dpi_metadata": None, "capture_metadata": capture_metadata, + "capture_observation": _build_capture_observation( + pkt_iface=pkt_iface, + pkt_timestamp=pkt_timestamp, + capture_source=capture_source, + capture_session_id=capture_session_id, + capture_metadata=capture_metadata, + ), "ip_id": None, "icmp_type": None, "icmp_code": None, diff --git a/backend/src/utilities/database.py b/backend/src/utilities/database.py index 39181d8..5f4ffcb 100644 --- a/backend/src/utilities/database.py +++ b/backend/src/utilities/database.py @@ -124,14 +124,14 @@ def _serialize_row_for_broadcast(row: Dict[str, Any]) -> Dict[str, Any]: def _normalize_json_fields(payload: Dict[str, Any]) -> None: - for key in ("dpi_metadata", "capture_metadata", "telemetry_metadata"): + for key in ("dpi_metadata", "capture_metadata", "telemetry_metadata", "capture_observations"): value = payload.get(key) if isinstance(value, str): try: parsed = json.loads(value) except json.JSONDecodeError: continue - if isinstance(parsed, dict): + if isinstance(parsed, dict) or (key == "capture_observations" and isinstance(parsed, list)): payload[key] = parsed @@ -227,6 +227,13 @@ class DatabasePool: min_size=self._min_size, max_size=self._max_size, ) + async with self._pool.acquire() as conn: + await conn.execute( + """ + ALTER TABLE IF EXISTS packets + ADD COLUMN IF NOT EXISTS capture_observations JSONB + """ + ) logger.info("DB pool initialized") except Exception: logger.exception("Failed to create DB pool") @@ -259,6 +266,7 @@ class DatabasePool: dpi_metadata = pkt_info.get("dpi_metadata") telemetry_metadata = pkt_info.get("telemetry_metadata") + capture_observations = pkt_info.get("capture_observations") try: async with self._pool.acquire() as conn: @@ -302,10 +310,11 @@ class DatabasePool: dpi_metadata, capture_metadata, telemetry_metadata, + capture_observations, raw ) VALUES( $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20, - $21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31,$32,$33,$34,$35::jsonb,$36::jsonb,$37::jsonb,$38 + $21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31,$32,$33,$34,$35::jsonb,$36::jsonb,$37::jsonb,$38::jsonb,$39 ) ON CONFLICT (correlation_key) DO UPDATE SET updated_at = NOW(), @@ -357,6 +366,7 @@ class DatabasePool: dpi_metadata = COALESCE(EXCLUDED.dpi_metadata, packets.dpi_metadata), capture_metadata = COALESCE(EXCLUDED.capture_metadata, packets.capture_metadata), telemetry_metadata = COALESCE(EXCLUDED.telemetry_metadata, packets.telemetry_metadata), + capture_observations = COALESCE(EXCLUDED.capture_observations, packets.capture_observations), raw = COALESCE(EXCLUDED.raw, packets.raw) RETURNING * """, @@ -397,6 +407,7 @@ class DatabasePool: json.dumps(dpi_metadata) if dpi_metadata is not None else None, json.dumps(pkt_info.get("capture_metadata")) if pkt_info.get("capture_metadata") is not None else None, json.dumps(telemetry_metadata) if telemetry_metadata is not None else None, + json.dumps(capture_observations) if capture_observations is not None else None, pkt_info.get("raw"), ) except Exception: diff --git a/backend/src/utilities/packet_tracker.py b/backend/src/utilities/packet_tracker.py index fe2ba41..80f2ac4 100644 --- a/backend/src/utilities/packet_tracker.py +++ b/backend/src/utilities/packet_tracker.py @@ -23,6 +23,21 @@ def _utcnow() -> datetime: return datetime.now(timezone.utc) +def _observation_signature(observation: Dict[str, Any]) -> tuple[Any, ...]: + return ( + observation.get("observation_type"), + observation.get("source"), + observation.get("iface"), + observation.get("timestamp"), + observation.get("event_type"), + observation.get("capture_mode"), + observation.get("capture_session_id"), + observation.get("session_label"), + observation.get("session_kind"), + observation.get("reason"), + ) + + class PacketTracker: """Deduplicate packet observations and persist one upserted row per packet.""" @@ -73,6 +88,11 @@ class PacketTracker: telemetry_metadata = payload.get("telemetry_metadata") or {} if isinstance(telemetry_metadata, dict): related_ifaces.add(telemetry_metadata.get("iface")) + capture_observations = payload.get("capture_observations") or [] + if isinstance(capture_observations, list): + for observation in capture_observations: + if isinstance(observation, dict): + related_ifaces.add(observation.get("iface")) if related_ifaces & targets: self._entries.pop(correlation_key, None) @@ -124,6 +144,18 @@ class PacketTracker: payload["telemetry_metadata"] = event payload["last_observed_at"] = now_ts self._add_capture_source(payload, "telemetry") + self._add_capture_observation( + payload, + { + "observation_type": "telemetry", + "source": "telemetry", + "iface": event.get("iface"), + "timestamp": _utcnow().isoformat(), + "event_type": event.get("event_type"), + "capture_mode": "tc_ebpf", + "reason": event.get("reason"), + }, + ) for key, value in event.items(): if value is None or key in {"event_type", "reason", "reason_code", "iface", "packet_uid", "correlation_key"}: continue @@ -189,6 +221,7 @@ class PacketTracker: "capture_sources": [], "capture_metadata": None, "telemetry_metadata": None, + "capture_observations": [], }, "persisted": False, "dirty": True, @@ -236,12 +269,33 @@ class PacketTracker: if source not in capture_sources: capture_sources.append(source) + def _add_capture_observation(self, payload: Dict[str, Any], observation: Optional[Dict[str, Any]]) -> bool: + if not isinstance(observation, dict): + return False + + observations = payload.setdefault("capture_observations", []) + if not isinstance(observations, list): + observations = [] + payload["capture_observations"] = observations + + normalized = {key: value for key, value in observation.items() if value is not None} + signature = _observation_signature(normalized) + for existing in observations: + if isinstance(existing, dict) and _observation_signature(existing) == signature: + return False + + observations.append(normalized) + observations.sort(key=lambda item: str(item.get("timestamp") or "")) + return True + def _merge_packet_info(self, entry: Dict[str, Any], pkt_info: Dict[str, Any], now_ts: float) -> None: payload = entry["payload"] changed = False self._add_capture_source(payload, pkt_info.get("capture_source") or "af_packet") + if self._add_capture_observation(payload, pkt_info.get("capture_observation")): + changed = True for key, value in pkt_info.items(): - if key in {"iface", "capture_source"}: + if key in {"iface", "capture_source", "capture_observation"}: continue if value is None: continue @@ -266,7 +320,14 @@ class PacketTracker: payload["capture_iface"] = iface changed = True - if iface and not pkt_info.get("capture_metadata") and not payload.get("ingress_if"): + capture_observation = pkt_info.get("capture_observation") or {} + is_bridge_af_packet = ( + isinstance(capture_observation, dict) + and capture_observation.get("session_kind") == "bridge" + and capture_observation.get("capture_mode") == "af_packet" + ) + + if iface and not pkt_info.get("capture_metadata") and not payload.get("ingress_if") and not is_bridge_af_packet: payload["ingress_if"] = iface payload["ingress_seen_at"] = _utcnow() changed = True diff --git a/frontend/src/components/PacketInspectModal.tsx b/frontend/src/components/PacketInspectModal.tsx index ca12825..7b45c9c 100644 --- a/frontend/src/components/PacketInspectModal.tsx +++ b/frontend/src/components/PacketInspectModal.tsx @@ -149,6 +149,7 @@ function buildReducedMetadata(packet: PacketRow | null) { dpi_metadata: dpi, capture_metadata: packet.capture_metadata ?? null, telemetry_metadata: packet.telemetry_metadata ?? null, + capture_observations: packet.capture_observations ?? null, }; } @@ -499,6 +500,7 @@ export default function PacketInspectModal({ const reducedMetadata = useMemo(() => buildReducedMetadata(packet), [packet]); const fullPacketJson = useMemo(() => formatJson(packet), [packet]); const decodedPayload = useMemo(() => getDecodedPayload(packet), [packet]); + const captureObservationCount = Array.isArray(packet?.capture_observations) ? packet.capture_observations.length : 0; const downloadRaw = () => { if (!packet?.raw_b64) return; @@ -548,11 +550,20 @@ export default function PacketInspectModal({ {packet?.app_protocol ?? packet?.app_master_protocol ?? '-'} {packet?.app_hostname ?? '-'} {packet?.verdict ?? '-'} + {captureObservationCount || '-'} {packet?.src_ip ? `${packet.src_ip}${packet?.src_port ? `:${packet.src_port}` : ''}` : '-'} {packet?.dst_ip ? `${packet.dst_ip}${packet?.dst_port ? `:${packet.dst_port}` : ''}` : '-'} {[packet?.ingress_if, packet?.egress_if].filter(Boolean).join(' -> ') || '-'} {packet?.capture_sources?.join(', ') || '-'} + {captureObservationCount > 0 ? ( +
+ Capture observations + 
+                      {formatJson(packet?.capture_observations)}
+
+
+ ) : null}
Reduced metadata 
{formatJson(reducedMetadata)}
diff --git a/frontend/src/components/PacketViewer.tsx b/frontend/src/components/PacketViewer.tsx index 593353b..76ff30c 100644 --- a/frontend/src/components/PacketViewer.tsx +++ b/frontend/src/components/PacketViewer.tsx @@ -296,6 +296,23 @@ function getStringList(value: unknown): string[] { .filter(Boolean); } +function getObservationIfaces(packet: PacketRow): string[] { + const observations = Array.isArray(packet.capture_observations) ? packet.capture_observations : []; + const seen = new Set(); + for (const observation of observations) { + const iface = observation?.iface; + if (iface) seen.add(String(iface)); + } + return Array.from(seen); +} + +function getObservationSummary(packet: PacketRow) { + const observations = Array.isArray(packet.capture_observations) ? packet.capture_observations : []; + const captureCount = observations.filter((observation) => observation?.observation_type === 'capture').length; + const telemetryCount = observations.filter((observation) => observation?.observation_type === 'telemetry').length; + return { total: observations.length, captureCount, telemetryCount }; +} + function isLikelyText(bytes: Uint8Array) { if (bytes.length === 0) return false; let printable = 0; @@ -1181,9 +1198,9 @@ export default function PacketViewer(): ReactElement { render: (val: any) => {formatTimestamp(val)}, }, { - title: 'Path', + title: 'Path / Seen On', key: 'path', - width: 150, + width: 190, render: (_: any, rec: PacketRow) => ( {[rec.ingress_if, rec.egress_if].filter(Boolean).length > 0 ? ( @@ -1204,7 +1221,24 @@ export default function PacketViewer(): ReactElement { )} ) : ( - - + (() => { + const observationIfaces = getObservationIfaces(rec); + const summary = getObservationSummary(rec); + return observationIfaces.length > 0 ? ( + <> + {observationIfaces.map((iface) => ( + + {iface} + + ))} + + {summary.total} obs + + + ) : ( + - + ); + })() )} ), diff --git a/frontend/src/types/packets.ts b/frontend/src/types/packets.ts index 4efb5a6..c3cd71a 100644 --- a/frontend/src/types/packets.ts +++ b/frontend/src/types/packets.ts @@ -1,3 +1,16 @@ +export interface PacketObservationRow { + observation_type: 'capture' | 'telemetry'; + source: string; + iface?: string | null; + timestamp?: string | null; + event_type?: string | null; + capture_mode?: string | null; + capture_session_id?: string | null; + session_label?: string | null; + session_kind?: string | null; + reason?: string | null; +} + export interface PacketRow { id?: number | string; timestamp?: string; @@ -36,6 +49,7 @@ export interface PacketRow { dpi_metadata?: Record | null; capture_metadata?: Record | null; telemetry_metadata?: Record | null; + capture_observations?: PacketObservationRow[] | null; verdict?: string | null; verdict_reason?: string | null; verdict_confidence?: string | null; diff --git a/setup_database.sh b/setup_database.sh index 94a04b2..39a45f4 100755 --- a/setup_database.sh +++ b/setup_database.sh @@ -95,10 +95,12 @@ CREATE TABLE IF NOT EXISTS packets ( dpi_metadata JSONB, capture_metadata JSONB, telemetry_metadata JSONB, + capture_observations JSONB, -- Full packet dump raw BYTEA ); +ALTER TABLE packets ADD COLUMN IF NOT EXISTS capture_observations JSONB; CREATE UNIQUE INDEX IF NOT EXISTS idx_packets_correlation_key ON packets(correlation_key); CREATE UNIQUE INDEX IF NOT EXISTS idx_packets_packet_uid ON packets(packet_uid); CREATE INDEX IF NOT EXISTS idx_packets_flow_id ON packets(flow_id);