try new correlation af packet bridge mode
All checks were successful
Build and Deploy MITM Webserver / traffic_target (push) Successful in 0s
Build and Deploy MITM Webserver / build (push) Successful in 11s

This commit is contained in:
2026-04-12 18:27:41 +02:00
parent 4521b9d99e
commit 9c6eaaa7b2
8 changed files with 214 additions and 11 deletions

View File

@@ -1,11 +1,26 @@
"""Pydantic model for packet rows returned by the backend.""" """Pydantic model for packet rows returned by the backend."""
from datetime import datetime from datetime import datetime
from typing import Optional, Union from typing import Literal, Optional, Union
from pydantic import BaseModel, ConfigDict, Field, IPvAnyAddress 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): class PacketDBModel(BaseModel):
"""Normalized packet representation used across DB and API layers.""" """Normalized packet representation used across DB and API layers."""
@@ -48,6 +63,18 @@ class PacketDBModel(BaseModel):
"dpi_metadata": {"method": "GET"}, "dpi_metadata": {"method": "GET"},
"capture_metadata": {"capture_mode": "tc_ingress", "packet_id": "123456"}, "capture_metadata": {"capture_mode": "tc_ingress", "packet_id": "123456"},
"telemetry_metadata": {"event_type": "egress", "iface": "eth1", "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": "accept",
"verdict_reason": "egress-observed", "verdict_reason": "egress-observed",
"verdict_confidence": "high", "verdict_confidence": "high",
@@ -92,6 +119,10 @@ class PacketDBModel(BaseModel):
dpi_metadata: Optional[dict] = Field(None, description="Raw metadata from DPI/flow enrichment.") 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.") 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.") 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: Optional[str] = None
verdict_reason: Optional[str] = None verdict_reason: Optional[str] = None
verdict_confidence: Optional[str] = None verdict_confidence: Optional[str] = None

View File

@@ -98,6 +98,7 @@ class PacketInfo(TypedDict, total=False):
app_risk_score: Optional[int] app_risk_score: Optional[int]
dpi_metadata: Optional[Dict[str, Any]] dpi_metadata: Optional[Dict[str, Any]]
capture_metadata: Optional[Dict[str, Any]] capture_metadata: Optional[Dict[str, Any]]
capture_observation: Optional[Dict[str, Any]]
ip_id: Optional[int] ip_id: Optional[int]
icmp_type: Optional[int] icmp_type: Optional[int]
icmp_code: 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 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( def parse_packet(
pkt, pkt,
bridge_label: str, bridge_label: str,
@@ -219,14 +248,17 @@ def parse_packet(
logger.debug("Packet captured on %s (bridge_label %s)", pkt_iface, bridge_label) 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 = { pkt_info: PacketInfo = {
"timestamp": _packet_timestamp(pkt), "timestamp": pkt_timestamp,
"capture_session_id": capture_session_id, "capture_session_id": capture_session_id,
"iface": pkt_iface, "iface": pkt_iface,
"capture_iface": None, "capture_iface": None,
"length": len(pkt), "length": len(pkt),
"raw": bytes(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, "packet_id": capture_metadata.get("packet_id") if capture_metadata else None,
"skb_mark": capture_metadata.get("skb_mark") if capture_metadata else None, "skb_mark": capture_metadata.get("skb_mark") if capture_metadata else None,
"src_mac": None, "src_mac": None,
@@ -250,6 +282,13 @@ def parse_packet(
"app_risk_score": None, "app_risk_score": None,
"dpi_metadata": None, "dpi_metadata": None,
"capture_metadata": capture_metadata, "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, "ip_id": None,
"icmp_type": None, "icmp_type": None,
"icmp_code": None, "icmp_code": None,

View File

@@ -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: 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) value = payload.get(key)
if isinstance(value, str): if isinstance(value, str):
try: try:
parsed = json.loads(value) parsed = json.loads(value)
except json.JSONDecodeError: except json.JSONDecodeError:
continue continue
if isinstance(parsed, dict): if isinstance(parsed, dict) or (key == "capture_observations" and isinstance(parsed, list)):
payload[key] = parsed payload[key] = parsed
@@ -227,6 +227,13 @@ class DatabasePool:
min_size=self._min_size, min_size=self._min_size,
max_size=self._max_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") logger.info("DB pool initialized")
except Exception: except Exception:
logger.exception("Failed to create DB pool") logger.exception("Failed to create DB pool")
@@ -259,6 +266,7 @@ class DatabasePool:
dpi_metadata = pkt_info.get("dpi_metadata") dpi_metadata = pkt_info.get("dpi_metadata")
telemetry_metadata = pkt_info.get("telemetry_metadata") telemetry_metadata = pkt_info.get("telemetry_metadata")
capture_observations = pkt_info.get("capture_observations")
try: try:
async with self._pool.acquire() as conn: async with self._pool.acquire() as conn:
@@ -302,10 +310,11 @@ class DatabasePool:
dpi_metadata, dpi_metadata,
capture_metadata, capture_metadata,
telemetry_metadata, telemetry_metadata,
capture_observations,
raw raw
) VALUES( ) VALUES(
$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20, $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 ON CONFLICT (correlation_key) DO UPDATE SET
updated_at = NOW(), updated_at = NOW(),
@@ -357,6 +366,7 @@ class DatabasePool:
dpi_metadata = COALESCE(EXCLUDED.dpi_metadata, packets.dpi_metadata), dpi_metadata = COALESCE(EXCLUDED.dpi_metadata, packets.dpi_metadata),
capture_metadata = COALESCE(EXCLUDED.capture_metadata, packets.capture_metadata), capture_metadata = COALESCE(EXCLUDED.capture_metadata, packets.capture_metadata),
telemetry_metadata = COALESCE(EXCLUDED.telemetry_metadata, packets.telemetry_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) raw = COALESCE(EXCLUDED.raw, packets.raw)
RETURNING * RETURNING *
""", """,
@@ -397,6 +407,7 @@ class DatabasePool:
json.dumps(dpi_metadata) if dpi_metadata is not None else None, 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(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(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"), pkt_info.get("raw"),
) )
except Exception: except Exception:

View File

@@ -23,6 +23,21 @@ def _utcnow() -> datetime:
return datetime.now(timezone.utc) 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: class PacketTracker:
"""Deduplicate packet observations and persist one upserted row per packet.""" """Deduplicate packet observations and persist one upserted row per packet."""
@@ -73,6 +88,11 @@ class PacketTracker:
telemetry_metadata = payload.get("telemetry_metadata") or {} telemetry_metadata = payload.get("telemetry_metadata") or {}
if isinstance(telemetry_metadata, dict): if isinstance(telemetry_metadata, dict):
related_ifaces.add(telemetry_metadata.get("iface")) 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: if related_ifaces & targets:
self._entries.pop(correlation_key, None) self._entries.pop(correlation_key, None)
@@ -124,6 +144,18 @@ class PacketTracker:
payload["telemetry_metadata"] = event payload["telemetry_metadata"] = event
payload["last_observed_at"] = now_ts payload["last_observed_at"] = now_ts
self._add_capture_source(payload, "telemetry") 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(): for key, value in event.items():
if value is None or key in {"event_type", "reason", "reason_code", "iface", "packet_uid", "correlation_key"}: if value is None or key in {"event_type", "reason", "reason_code", "iface", "packet_uid", "correlation_key"}:
continue continue
@@ -189,6 +221,7 @@ class PacketTracker:
"capture_sources": [], "capture_sources": [],
"capture_metadata": None, "capture_metadata": None,
"telemetry_metadata": None, "telemetry_metadata": None,
"capture_observations": [],
}, },
"persisted": False, "persisted": False,
"dirty": True, "dirty": True,
@@ -236,12 +269,33 @@ class PacketTracker:
if source not in capture_sources: if source not in capture_sources:
capture_sources.append(source) 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: def _merge_packet_info(self, entry: Dict[str, Any], pkt_info: Dict[str, Any], now_ts: float) -> None:
payload = entry["payload"] payload = entry["payload"]
changed = False changed = False
self._add_capture_source(payload, pkt_info.get("capture_source") or "af_packet") 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(): for key, value in pkt_info.items():
if key in {"iface", "capture_source"}: if key in {"iface", "capture_source", "capture_observation"}:
continue continue
if value is None: if value is None:
continue continue
@@ -266,7 +320,14 @@ class PacketTracker:
payload["capture_iface"] = iface payload["capture_iface"] = iface
changed = True 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_if"] = iface
payload["ingress_seen_at"] = _utcnow() payload["ingress_seen_at"] = _utcnow()
changed = True changed = True

View File

@@ -149,6 +149,7 @@ function buildReducedMetadata(packet: PacketRow | null) {
dpi_metadata: dpi, dpi_metadata: dpi,
capture_metadata: packet.capture_metadata ?? null, capture_metadata: packet.capture_metadata ?? null,
telemetry_metadata: packet.telemetry_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 reducedMetadata = useMemo(() => buildReducedMetadata(packet), [packet]);
const fullPacketJson = useMemo(() => formatJson(packet), [packet]); const fullPacketJson = useMemo(() => formatJson(packet), [packet]);
const decodedPayload = useMemo(() => getDecodedPayload(packet), [packet]); const decodedPayload = useMemo(() => getDecodedPayload(packet), [packet]);
const captureObservationCount = Array.isArray(packet?.capture_observations) ? packet.capture_observations.length : 0;
const downloadRaw = () => { const downloadRaw = () => {
if (!packet?.raw_b64) return; if (!packet?.raw_b64) return;
@@ -548,11 +550,20 @@ export default function PacketInspectModal({
<Descriptions.Item label="Application">{packet?.app_protocol ?? packet?.app_master_protocol ?? '-'}</Descriptions.Item> <Descriptions.Item label="Application">{packet?.app_protocol ?? packet?.app_master_protocol ?? '-'}</Descriptions.Item>
<Descriptions.Item label="Host">{packet?.app_hostname ?? '-'}</Descriptions.Item> <Descriptions.Item label="Host">{packet?.app_hostname ?? '-'}</Descriptions.Item>
<Descriptions.Item label="Verdict">{packet?.verdict ?? '-'}</Descriptions.Item> <Descriptions.Item label="Verdict">{packet?.verdict ?? '-'}</Descriptions.Item>
<Descriptions.Item label="Observations">{captureObservationCount || '-'}</Descriptions.Item>
<Descriptions.Item label="Source">{packet?.src_ip ? `${packet.src_ip}${packet?.src_port ? `:${packet.src_port}` : ''}` : '-'}</Descriptions.Item> <Descriptions.Item label="Source">{packet?.src_ip ? `${packet.src_ip}${packet?.src_port ? `:${packet.src_port}` : ''}` : '-'}</Descriptions.Item>
<Descriptions.Item label="Destination">{packet?.dst_ip ? `${packet.dst_ip}${packet?.dst_port ? `:${packet.dst_port}` : ''}` : '-'}</Descriptions.Item> <Descriptions.Item label="Destination">{packet?.dst_ip ? `${packet.dst_ip}${packet?.dst_port ? `:${packet.dst_port}` : ''}` : '-'}</Descriptions.Item>
<Descriptions.Item label="Path">{[packet?.ingress_if, packet?.egress_if].filter(Boolean).join(' -> ') || '-'}</Descriptions.Item> <Descriptions.Item label="Path">{[packet?.ingress_if, packet?.egress_if].filter(Boolean).join(' -> ') || '-'}</Descriptions.Item>
<Descriptions.Item label="Capture Sources">{packet?.capture_sources?.join(', ') || '-'}</Descriptions.Item> <Descriptions.Item label="Capture Sources">{packet?.capture_sources?.join(', ') || '-'}</Descriptions.Item>
</Descriptions> </Descriptions>
{captureObservationCount > 0 ? (
<div>
<Text strong>Capture observations</Text>
<pre style={{ whiteSpace: 'pre-wrap', wordBreak: 'break-word', fontSize: 12, marginTop: 8 }}>
{formatJson(packet?.capture_observations)}
</pre>
</div>
) : null}
<div> <div>
<Text strong>Reduced metadata</Text> <Text strong>Reduced metadata</Text>
<pre style={{ whiteSpace: 'pre-wrap', wordBreak: 'break-word', fontSize: 12, marginTop: 8 }}>{formatJson(reducedMetadata)}</pre> <pre style={{ whiteSpace: 'pre-wrap', wordBreak: 'break-word', fontSize: 12, marginTop: 8 }}>{formatJson(reducedMetadata)}</pre>

View File

@@ -296,6 +296,23 @@ function getStringList(value: unknown): string[] {
.filter(Boolean); .filter(Boolean);
} }
function getObservationIfaces(packet: PacketRow): string[] {
const observations = Array.isArray(packet.capture_observations) ? packet.capture_observations : [];
const seen = new Set<string>();
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) { function isLikelyText(bytes: Uint8Array) {
if (bytes.length === 0) return false; if (bytes.length === 0) return false;
let printable = 0; let printable = 0;
@@ -1181,9 +1198,9 @@ export default function PacketViewer(): ReactElement {
render: (val: any) => <Text>{formatTimestamp(val)}</Text>, render: (val: any) => <Text>{formatTimestamp(val)}</Text>,
}, },
{ {
title: 'Path', title: 'Path / Seen On',
key: 'path', key: 'path',
width: 150, width: 190,
render: (_: any, rec: PacketRow) => ( render: (_: any, rec: PacketRow) => (
<Space wrap size={[6, 6]}> <Space wrap size={[6, 6]}>
{[rec.ingress_if, rec.egress_if].filter(Boolean).length > 0 ? ( {[rec.ingress_if, rec.egress_if].filter(Boolean).length > 0 ? (
@@ -1204,7 +1221,24 @@ export default function PacketViewer(): ReactElement {
)} )}
</> </>
) : ( ) : (
<Text type="secondary">-</Text> (() => {
const observationIfaces = getObservationIfaces(rec);
const summary = getObservationSummary(rec);
return observationIfaces.length > 0 ? (
<>
{observationIfaces.map((iface) => (
<Tooltip key={iface} title={`observed on ${iface}`}>
<Tag style={{ ...colorForName(iface), borderRadius: 6 }}>{iface}</Tag>
</Tooltip>
))}
<Tooltip title={`${summary.captureCount} capture observation(s), ${summary.telemetryCount} telemetry observation(s)`}>
<Tag color="default">{summary.total} obs</Tag>
</Tooltip>
</>
) : (
<Text type="secondary">-</Text>
);
})()
)} )}
</Space> </Space>
), ),

View File

@@ -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 { export interface PacketRow {
id?: number | string; id?: number | string;
timestamp?: string; timestamp?: string;
@@ -36,6 +49,7 @@ export interface PacketRow {
dpi_metadata?: Record<string, unknown> | null; dpi_metadata?: Record<string, unknown> | null;
capture_metadata?: Record<string, unknown> | null; capture_metadata?: Record<string, unknown> | null;
telemetry_metadata?: Record<string, unknown> | null; telemetry_metadata?: Record<string, unknown> | null;
capture_observations?: PacketObservationRow[] | null;
verdict?: string | null; verdict?: string | null;
verdict_reason?: string | null; verdict_reason?: string | null;
verdict_confidence?: string | null; verdict_confidence?: string | null;

View File

@@ -95,10 +95,12 @@ CREATE TABLE IF NOT EXISTS packets (
dpi_metadata JSONB, dpi_metadata JSONB,
capture_metadata JSONB, capture_metadata JSONB,
telemetry_metadata JSONB, telemetry_metadata JSONB,
capture_observations JSONB,
-- Full packet dump -- Full packet dump
raw BYTEA 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_correlation_key ON packets(correlation_key);
CREATE UNIQUE INDEX IF NOT EXISTS idx_packets_packet_uid ON packets(packet_uid); 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); CREATE INDEX IF NOT EXISTS idx_packets_flow_id ON packets(flow_id);