flow showing

2026-03-09 12:54:57 +01:00
parent a111d5ea00
commit a49ee5b9bd
4 changed files with 166 additions and 36 deletions
--- 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.")
--- 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)):
--- 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<string, unknown> | 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<string, unknown>;
+}
+
+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<PacketTableRow[]>(() => {
+    const grouped = new Map<string, PacketRow[]>();
+    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) => <Text>{formatTimestamp(val)}</Text>,
      },
-      {
-        title: 'Corr',
-        key: 'correlation',
-        width: 140,
-        render: (_: any, rec: PacketRow) => (
-          <Space direction="vertical" size={0}>
-            <Text>{rec.packet_id ?? rec.packet_uid?.slice(0, 10) ?? '-'}</Text>
-            <Text type="secondary" style={{ fontSize: 12 }}>
-              {rec.correlation_source ?? '-'}
-            </Text>
-          </Space>
-        ),
-      },
      {
        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) => <Text>{formatIpProto(rec)}</Text>,
+        width: 130,
+        render: (_: any, rec: PacketTableRow) => <Text>{rec.__kind === 'flow' ? `${formatIpProto(rec)} flow` : formatProtocolLabel(rec)}</Text>,
+      },
+      {
+        title: 'Flow',
+        key: 'flow_id',
+        width: 130,
+        render: (_: any, rec: PacketTableRow) => <Text>{rec.flow_id ?? '-'}</Text>,
      },
      {
        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 : (
          <Space>
            <Tooltip title="View raw + tshark">
              <Button
@@ -551,12 +637,16 @@ export default function PacketViewer(): ReactElement {
          </div>
        ) : (
          <Table
-            rowKey={(rec: PacketRow) => packetKey(rec)}
-            dataSource={packets}
+            rowKey={(rec: PacketTableRow) => rec.key}
+            dataSource={tableRows}
            columns={columns as any}
            pagination={{ pageSize: 25 }}
            size="small"
            bordered
+            expandable={{
+              defaultExpandAllRows: true,
+              rowExpandable: (record: PacketTableRow) => Array.isArray(record.children) && record.children.length > 0,
+            }}
          />
        )}
      </Card>
--- a/frontend/src/types/packets.ts
+++ b/frontend/src/types/packets.ts
@@ -24,6 +24,7 @@ export interface PacketRow {
  length?: number | null;
  raw_present?: boolean | null;
  capture_sources?: string[] | null;
+  flow_id?: string | null;
  raw_b64?: string | null;
  app_protocol?: string | null;
  app_master_protocol?: string | null;