diff --git a/backend/src/Models/packets.py b/backend/src/Models/packets.py index 1975d60..84bf173 100644 --- a/backend/src/Models/packets.py +++ b/backend/src/Models/packets.py @@ -80,6 +80,7 @@ class PacketDBModel(BaseModel): length: Optional[int] = None raw_present: Optional[bool] = Field(None, description="Whether raw packet bytes were captured for this row.") capture_sources: Optional[list[str]] = Field(None, description="Capture sources that contributed to this row.") + flow_id: Optional[str] = Field(None, description="Derived flow identifier from tshark stream metadata, if available.") raw_b64: Optional[str] = Field(None, description="Base64-encoded packet bytes.") app_protocol: Optional[str] = Field(None, description="Detected application protocol.") app_master_protocol: Optional[str] = Field(None, description="Detected application master protocol.") diff --git a/backend/src/utilities/database.py b/backend/src/utilities/database.py index 95e165a..88cb758 100644 --- a/backend/src/utilities/database.py +++ b/backend/src/utilities/database.py @@ -29,6 +29,7 @@ def _db_text(value: Any) -> Any: def _serialize_row_for_broadcast(row: Dict[str, Any]) -> Dict[str, Any]: serialized = dict(row) _normalize_json_fields(serialized) + _attach_derived_fields(serialized) raw_val = serialized.get("raw") if isinstance(raw_val, (bytes, bytearray)): @@ -54,6 +55,42 @@ def _normalize_json_fields(payload: Dict[str, Any]) -> None: payload[key] = parsed +def _derive_flow_id(payload: Dict[str, Any]) -> Optional[str]: + dpi_metadata = payload.get("dpi_metadata") + if not isinstance(dpi_metadata, dict): + return None + + tcp_meta = dpi_metadata.get("tcp") + if isinstance(tcp_meta, dict): + stream = tcp_meta.get("stream") + if stream not in (None, "", []): + return f"tcp:{stream}" + + udp_meta = dpi_metadata.get("udp") + if isinstance(udp_meta, dict): + stream = udp_meta.get("stream") + if stream not in (None, "", []): + return f"udp:{stream}" + + tshark_meta = dpi_metadata.get("tshark") + if isinstance(tshark_meta, dict): + tcp_stream = tshark_meta.get("tcp_stream") + if tcp_stream not in (None, "", []): + return f"tcp:{tcp_stream}" + udp_stream = tshark_meta.get("udp_stream") + if udp_stream not in (None, "", []): + return f"udp:{udp_stream}" + + return None + + +def _attach_derived_fields(payload: Dict[str, Any]) -> None: + if payload.get("flow_id") in (None, ""): + flow_id = _derive_flow_id(payload) + if flow_id is not None: + payload["flow_id"] = flow_id + + class DatabasePool: """Asyncpg connection pool wrapper used by the packet APIs.""" @@ -509,7 +546,7 @@ class DatabasePool: """ SELECT * FROM packets - ORDER BY updated_at DESC, id DESC + ORDER BY timestamp DESC, id DESC LIMIT $1 """, limit, @@ -519,6 +556,7 @@ class DatabasePool: for row in rows: data = dict(row) _normalize_json_fields(data) + _attach_derived_fields(data) raw_val = data.get("raw") if isinstance(raw_val, (bytes, bytearray)): diff --git a/frontend/src/components/PacketViewer.tsx b/frontend/src/components/PacketViewer.tsx index 44553e3..d62b51a 100644 --- a/frontend/src/components/PacketViewer.tsx +++ b/frontend/src/components/PacketViewer.tsx @@ -26,6 +26,13 @@ const { Option } = Select; const DEFAULT_LIMIT = 200; const MAX_PACKETS = 2000; // in-memory cap +type PacketTableRow = PacketRow & { + key: string; + __kind: 'packet' | 'flow'; + flow_packet_count?: number; + children?: PacketTableRow[]; +}; + function base64ToHex(b64: string) { try { const bin = atob(b64); @@ -97,6 +104,39 @@ function packetKey(packet: PacketRow) { return String(packet.correlation_key ?? packet.packet_id ?? packet.packet_uid ?? packet.id ?? `${packet.ingress_if ?? 'if'}:${packet.timestamp ?? ''}`); } +function packetTimestampMs(packet: PacketRow) { + const time = packet.timestamp ? new Date(packet.timestamp).getTime() : 0; + return Number.isNaN(time) ? 0 : time; +} + +function sortPacketsByTimestampDesc(left: PacketRow, right: PacketRow) { + return packetTimestampMs(right) - packetTimestampMs(left); +} + +function sortPacketsByTimestampAsc(left: PacketRow, right: PacketRow) { + return packetTimestampMs(left) - packetTimestampMs(right); +} + +function getDpiDict(packet: PacketRow, key: string): Record | null { + const dpi = packet.dpi_metadata; + if (!dpi || typeof dpi !== 'object' || Array.isArray(dpi)) return null; + const value = dpi[key]; + if (!value || typeof value !== 'object' || Array.isArray(value)) return null; + return value as Record; +} + +function getFlowId(packet: PacketRow) { + if (packet.flow_id) return String(packet.flow_id); + const tcp = getDpiDict(packet, 'tcp'); + if (tcp?.stream != null) return `tcp:${String(tcp.stream)}`; + const udp = getDpiDict(packet, 'udp'); + if (udp?.stream != null) return `udp:${String(udp.stream)}`; + const tshark = getDpiDict(packet, 'tshark'); + if (tshark?.tcp_stream != null) return `tcp:${String(tshark.tcp_stream)}`; + if (tshark?.udp_stream != null) return `udp:${String(tshark.udp_stream)}`; + return null; +} + function formatEthType(packet: PacketRow) { if (packet.eth_type) return String(packet.eth_type); if (typeof packet.eth_type_raw === 'number') return `0x${packet.eth_type_raw.toString(16)}`; @@ -109,6 +149,14 @@ function formatIpProto(packet: PacketRow) { return '-'; } +function formatProtocolLabel(packet: PacketRow) { + const base = formatIpProto(packet); + const tcp = getDpiDict(packet, 'tcp'); + const tshark = getDpiDict(packet, 'tshark'); + const packetType = tcp?.packet_type ?? tshark?.tcp_packet_type; + return packetType ? `${base} ${String(packetType)}` : base; +} + /** Format timestamp to European locale with ms */ function formatTimestamp(ts?: string) { if (!ts) return '-'; @@ -163,12 +211,7 @@ export default function PacketViewer(): ReactElement { } next.sort((left, right) => { - const leftId = Number(left.id ?? 0); - const rightId = Number(right.id ?? 0); - if (!Number.isNaN(leftId) && !Number.isNaN(rightId) && leftId !== rightId) { - return rightId - leftId; - } - return String(right.timestamp ?? '').localeCompare(String(left.timestamp ?? '')); + return sortPacketsByTimestampDesc(left, right); }); if (next.length > MAX_PACKETS) return next.slice(0, MAX_PACKETS); @@ -182,7 +225,7 @@ export default function PacketViewer(): ReactElement { try { const res = await fetchPackets(limitVal); const list: PacketRow[] = res.packets ?? []; - setPackets(list); + setPackets([...list].sort(sortPacketsByTimestampDesc)); } catch (err: any) { console.error('fetchHistory error', err); message.error(err?.message ?? 'Failed to fetch packet history'); @@ -304,6 +347,57 @@ export default function PacketViewer(): ReactElement { URL.revokeObjectURL(url); }; + const tableRows = useMemo(() => { + const grouped = new Map(); + const standalone: PacketTableRow[] = []; + + for (const packet of packets) { + const flowId = getFlowId(packet); + if (!flowId) { + standalone.push({ ...packet, key: packetKey(packet), __kind: 'packet' }); + continue; + } + const items = grouped.get(flowId); + if (items) { + items.push(packet); + } else { + grouped.set(flowId, [packet]); + } + } + + const flowRows: PacketTableRow[] = []; + for (const [flowId, flowPackets] of grouped.entries()) { + const sortedPackets = [...flowPackets].sort(sortPacketsByTimestampAsc); + if (sortedPackets.length === 1) { + flowRows.push({ + ...sortedPackets[0], + flow_id: flowId, + key: packetKey(sortedPackets[0]), + __kind: 'packet', + }); + continue; + } + + const latestPacket = [...flowPackets].sort(sortPacketsByTimestampDesc)[0]; + const children = sortedPackets.map((packet) => ({ + ...packet, + flow_id: flowId, + key: packetKey(packet), + __kind: 'packet' as const, + })); + flowRows.push({ + ...latestPacket, + flow_id: flowId, + key: `flow:${flowId}`, + __kind: 'flow', + flow_packet_count: sortedPackets.length, + children, + }); + } + + return [...flowRows, ...standalone].sort(sortPacketsByTimestampDesc); + }, [packets]); + const columns = useMemo( () => [ { @@ -313,19 +407,6 @@ export default function PacketViewer(): ReactElement { width: 120, render: (val: any) => {formatTimestamp(val)}, }, - { - title: 'Corr', - key: 'correlation', - width: 140, - render: (_: any, rec: PacketRow) => ( - - {rec.packet_id ?? rec.packet_uid?.slice(0, 10) ?? '-'} - - {rec.correlation_source ?? '-'} - - - ), - }, { title: 'Path', key: 'path', @@ -397,8 +478,14 @@ export default function PacketViewer(): ReactElement { title: 'Protocol', dataIndex: 'ip_proto', key: 'ip_proto', - width: 110, - render: (_: any, rec: PacketRow) => {formatIpProto(rec)}, + width: 130, + render: (_: any, rec: PacketTableRow) => {rec.__kind === 'flow' ? `${formatIpProto(rec)} flow` : formatProtocolLabel(rec)}, + }, + { + title: 'Flow', + key: 'flow_id', + width: 130, + render: (_: any, rec: PacketTableRow) => {rec.flow_id ?? '-'}, }, { title: 'App', @@ -452,24 +539,23 @@ export default function PacketViewer(): ReactElement { ), }, { - title: 'Len', + title: 'Len/Pkts', dataIndex: 'length', key: 'length', - width: 80, - render: (n: any) => (typeof n === 'number' ? n.toLocaleString('de-DE') : '-'), - }, - { - title: 'Risk', - dataIndex: 'app_risk_score', - key: 'app_risk_score', - width: 70, - render: (n: any) => (typeof n === 'number' ? n : '-'), + width: 90, + render: (_: any, rec: PacketTableRow) => + rec.__kind === 'flow' + ? `${rec.flow_packet_count ?? 0} pkts` + : typeof rec.length === 'number' + ? rec.length.toLocaleString('de-DE') + : '-', }, { title: 'Actions', key: 'actions', width: 96, - render: (_: any, rec: PacketRow) => ( + render: (_: any, rec: PacketTableRow) => + rec.__kind === 'flow' ? null : (