flow id sessions prefix

2026-03-10 21:16:32 +01:00
parent 57b0ac5b25
commit c370374a8a
4 changed files with 98 additions and 29 deletions
--- a/backend/src/network_sniffer.py
+++ b/backend/src/network_sniffer.py
@@ -63,6 +63,7 @@ class PacketInfo(TypedDict, total=False):
    """TypedDict for parsed packet data used by persistence and telemetry."""

    timestamp: datetime
+    capture_session_id: Optional[str]
    correlation_key: str
    correlation_source: str
    packet_id: Optional[str]
@@ -199,7 +200,12 @@ def _merge_enrichment(pkt_info: PacketInfo, enrichment: Dict[str, Any]) -> None:
            pkt_info[key] = value


-def parse_packet(pkt, bridge_label: str, capture_metadata: Optional[Dict[str, Any]] = None) -> None:
+def parse_packet(
+    pkt,
+    bridge_label: str,
+    capture_metadata: Optional[Dict[str, Any]] = None,
+    capture_session_id: Optional[str] = None,
+) -> None:
    """
    Parse a scapy Packet object into a normalized PacketInfo and schedule DB insert.
    bridge_label indicates whether the packet was captured as part of a bridge-snapshot or single-interface.
@@ -212,6 +218,7 @@ def parse_packet(pkt, bridge_label: str, capture_metadata: Optional[Dict[str, An

    pkt_info: PacketInfo = {
        "timestamp": _packet_timestamp(pkt),
+        "capture_session_id": capture_session_id,
        "iface": pkt_iface,
        "capture_iface": None,
        "length": len(pkt),
@@ -454,11 +461,12 @@ def parse_packet_bytes(
    packet_bytes: bytes,
    iface: str,
    capture_metadata: Optional[Dict[str, Any]] = None,
+    capture_session_id: Optional[str] = None,
 ) -> None:
    """Parse one raw Ethernet frame using the shared Scapy packet path."""
    pkt = Ether(packet_bytes)
    pkt.sniffed_on = iface
-    parse_packet(pkt, iface, capture_metadata=capture_metadata)
+    parse_packet(pkt, iface, capture_metadata=capture_metadata, capture_session_id=capture_session_id)


 # -------------------------
@@ -640,7 +648,7 @@ def _session_reader_loop(session_id: str) -> None:
                try:
                    pkt = Ether(raw)
                    pkt.sniffed_on = iface
-                    parse_packet(pkt, label)
+                    parse_packet(pkt, label, capture_session_id=session_id)
                    logger.debug("Captured packet on %s in session %s (len=%d)", iface, session_id, len(raw))
                except Exception:
                    logger.exception("Failed to parse/process packet from %s in session %s", iface, session_id)
--- a/backend/src/utilities/database.py
+++ b/backend/src/utilities/database.py
@@ -56,6 +56,9 @@ def _normalize_json_fields(payload: Dict[str, Any]) -> None:


 def _derive_flow_id(payload: Dict[str, Any]) -> Optional[str]:
+    session_id = payload.get("capture_session_id")
+    session_prefix = f"{session_id}:" if session_id not in (None, "") else ""
+
    dpi_metadata = payload.get("dpi_metadata")
    if not isinstance(dpi_metadata, dict):
        return None
@@ -64,22 +67,22 @@ def _derive_flow_id(payload: Dict[str, Any]) -> Optional[str]:
    if isinstance(tcp_meta, dict):
        stream = tcp_meta.get("stream")
        if stream not in (None, "", []):
-            return f"tcp:{stream}"
+            return f"{session_prefix}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}"
+            return f"{session_prefix}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}"
+            return f"{session_prefix}tcp:{tcp_stream}"
        udp_stream = tshark_meta.get("udp_stream")
        if udp_stream not in (None, "", []):
-            return f"udp:{udp_stream}"
+            return f"{session_prefix}udp:{udp_stream}"

    return None

--- a/frontend/src/api/apiClient.ts
+++ b/frontend/src/api/apiClient.ts
@@ -157,6 +157,16 @@ export const fetchPackets = async (limit = 100): Promise<FetchPacketsResponse> =
  return res.data;
 };

+export const getPacketsWebSocketUrl = (subscribeRecent = 0): string => {
+  const url = new URL(BASE);
+  url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
+  url.pathname = `${url.pathname.replace(/\/$/, '')}/packets/ws/packets`;
+  if (subscribeRecent > 0) {
+    url.searchParams.set('subscribe_recent', String(subscribeRecent));
+  }
+  return url.toString();
+};
+
 export const clearPackets = async (): Promise<any> => {
  const res = await api.delete('/packets/packets');
  return res.data;
--- a/frontend/src/components/PacketViewer.tsx
+++ b/frontend/src/components/PacketViewer.tsx
@@ -7,25 +7,27 @@ import {
  Descriptions,
  Modal,
  Row,
+  Segmented,
  Select,
  Space,
  Spin,
-  Switch,
  Table,
  Tag,
  Tabs,
  Tooltip,
  Typography,
  message,
+  theme,
 } from 'antd';
 import { ReactElement, ReactNode, useCallback, useEffect, useMemo, useRef, useState } from 'react';
-import { clearPackets, fetchPackets } from '../api/apiClient';
+import { clearPackets, fetchPackets, getPacketsWebSocketUrl } from '../api/apiClient';
 import type { PacketRow } from '../types/packets';

 const { Text, Title } = Typography;
 const { Option } = Select;

 const DEFAULT_LIMIT = 200;
+const DEFAULT_PAGE_SIZE = 25;
 const MAX_PACKETS = 2000; // in-memory cap

 const ARP_OPCODE_LABELS: Record<number, string> = {
@@ -915,10 +917,13 @@ function formatTimestamp(ts?: string) {
 * PacketViewer component
 */
 export default function PacketViewer(): ReactElement {
+  const { token } = theme.useToken();
  const [packets, setPackets] = useState<PacketRow[]>([]);
  const [loading, setLoading] = useState<boolean>(true);
  const [statusLoading, setStatusLoading] = useState<boolean>(false);
  const [limit, setLimit] = useState<number>(DEFAULT_LIMIT);
+  const [mode, setMode] = useState<'live' | 'history'>('live');
+  const [pageSize, setPageSize] = useState<number>(DEFAULT_PAGE_SIZE);
  const [paused, setPaused] = useState<boolean>(false);
  const wsRef = useRef<WebSocket | null>(null);
  const [hexModalOpen, setHexModalOpen] = useState(false);
@@ -967,18 +972,20 @@ export default function PacketViewer(): ReactElement {
    }
  }, []);

-  // open websocket
-  const openWs = useCallback(() => {
+  const closeWs = useCallback(() => {
    if (wsRef.current) {
      try {
        wsRef.current.close();
      } catch {}
      wsRef.current = null;
    }
+  }, []);

-    const loc = window.location;
-    const protocol = loc.protocol === 'https:' ? 'wss' : 'ws';
-    const wsUrl = `ws://mitm.lan/api/packets/ws/packets?subscribe_recent=20`;
+  // open websocket
+  const openWs = useCallback((subscribeRecent: number) => {
+    closeWs();
+
+    const wsUrl = getPacketsWebSocketUrl(subscribeRecent);
    const ws = new WebSocket(wsUrl);
    wsRef.current = ws;

@@ -1014,7 +1021,7 @@ export default function PacketViewer(): ReactElement {
    ws.onclose = () => {
      wsRef.current = null;
    };
-  }, [paused, pushNew]);
+  }, [closeWs, paused, pushNew]);

  // pause handling: when unpausing, flush queuedDuringPause into list
  useEffect(() => {
@@ -1027,26 +1034,28 @@ export default function PacketViewer(): ReactElement {
    }
  }, [paused, pushNew]);

-  // start up: fetch history and open ws
+  // start up / mode change: fetch history, optionally attach to live websocket
  useEffect(() => {
    setLoading(true);
    fetchHistory(limit).then(() => {
-      openWs();
+      if (mode === 'live') {
+        openWs(limit);
+      } else {
+        closeWs();
+      }
    });
    return () => {
-      if (wsRef.current) {
-        try {
-          wsRef.current.close();
-        } catch {}
-        wsRef.current = null;
-      }
+      closeWs();
    };
-  }, [fetchHistory, limit, openWs]);
+  }, [closeWs, fetchHistory, limit, mode, openWs]);

  const handleRefresh = async () => {
    setLoading(true);
    try {
      await fetchHistory(limit);
+      if (mode === 'live') {
+        openWs(limit);
+      }
    } finally {
      setLoading(false);
    }
@@ -1365,19 +1374,44 @@ export default function PacketViewer(): ReactElement {
            gap: 6px;
            font-weight: 600;
          }
+
+          .packet-viewer .packet-mode-toggle.ant-segmented {
+            background: ${token.colorFillTertiary};
+          }
+
+          .packet-viewer .packet-mode-toggle .ant-segmented-item-selected {
+            background: ${token.colorPrimary};
+            color: ${token.colorTextLightSolid};
+          }
+
+          .packet-viewer .packet-mode-toggle .ant-segmented-item-selected:hover {
+            color: ${token.colorTextLightSolid};
+          }
        `}
      </style>
      <Row justify="space-between" align="middle" style={{ marginBottom: 12 }}>
        <Col>
          <Title level={3}>Packets</Title>
          <div>
-            <Text type="secondary">Live packet viewer — history + live stream</Text>
+            <Text type="secondary">
+              {mode === 'live' ? 'Live packet viewer via websocket stream' : 'Historical packet viewer from database'}
+            </Text>
          </div>
        </Col>

        <Col>
          <Space>
-            <Text>History</Text>
+            <Segmented
+              className="packet-mode-toggle"
+              value={mode}
+              onChange={(value) => setMode(value as 'live' | 'history')}
+              options={[
+                { label: 'Live packets', value: 'live' },
+                { label: 'History', value: 'history' },
+              ]}
+            />
+
+            <Text>{mode === 'live' ? 'Recent buffer' : 'History size'}</Text>
            <Select value={limit} onChange={(v) => setLimit(Number(v))} style={{ width: 120 }}>
              <Option value={50}>50</Option>
              <Option value={100}>100</Option>
@@ -1385,8 +1419,19 @@ export default function PacketViewer(): ReactElement {
              <Option value={500}>500</Option>
            </Select>

-            <Text>Live</Text>
-            <Switch checked={!paused} onChange={(checked) => handlePauseToggle(!checked ? true : false)} />
+            <Text>Page size</Text>
+            <Select value={pageSize} onChange={(v) => setPageSize(Number(v))} style={{ width: 120 }}>
+              <Option value={25}>25</Option>
+              <Option value={50}>50</Option>
+              <Option value={100}>100</Option>
+              <Option value={200}>200</Option>
+            </Select>
+
+            {mode === 'live' ? (
+              <Button onClick={() => handlePauseToggle(!paused)}>
+                {paused ? 'Resume live' : 'Pause live'}
+              </Button>
+            ) : null}

            <Button icon={<SyncOutlined />} onClick={handleRefresh} loading={statusLoading} />
            <Button
@@ -1415,7 +1460,10 @@ export default function PacketViewer(): ReactElement {
            rowKey={(rec: PacketTableRow) => rec.key}
            dataSource={tableRows}
            columns={columns as any}
-            pagination={{ pageSize: 25 }}
+            pagination={{
+              pageSize,
+              showSizeChanger: false,
+            }}
            size="small"
            bordered
            expandable={{