"""Database helper for packet persistence and retrieval.""" import asyncio import base64 import json import logging import math import statistics from datetime import datetime, timezone from typing import Any, Dict, List, Optional import asyncpg from asyncpg.pool import Pool from pydantic import ValidationError from src.Models.etherType import EtherTypeEnum, ethertype_from_int from src.Models.ip_protocol import protocol_from_number from src.Models.packets import PacketDBModel logger = logging.getLogger("af_packet_sniffer") def _db_text(value: Any) -> Any: if value is None: return None if isinstance(value, (str, int)): return value if hasattr(value, "value"): return getattr(value, "value") return str(value) def _analysis_protocol_name(app_protocol: Any, ip_proto_raw: Any, eth_type_raw: Any) -> str: if app_protocol not in (None, ""): return str(app_protocol) ip_protocol_name = _analysis_ip_protocol_name(ip_proto_raw) if ip_protocol_name is not None: return ip_protocol_name ethernet_protocol_name = _analysis_ethernet_protocol_name(eth_type_raw) if ethernet_protocol_name is not None: return ethernet_protocol_name return "UNKNOWN" def _analysis_ip_protocol_name(ip_proto_raw: Any) -> Optional[str]: if ip_proto_raw is not None: try: return str(protocol_from_number(int(ip_proto_raw))) except Exception: return f"IP_PROTO_{ip_proto_raw}" return None def _analysis_ethernet_protocol_name(eth_type_raw: Any) -> Optional[str]: if eth_type_raw is not None: try: return str(ethertype_from_int(int(eth_type_raw))) except Exception: return f"ETH_TYPE_{eth_type_raw}" return None def _analysis_host_key(ip_address: Any, mac_address: Any) -> str: return f"{ip_address or 'no-ip'}|{mac_address or 'no-mac'}" def _classify_discovery_activity( protocol_name: Optional[str], ethernet_protocol_name: Optional[str], ip_protocol_name: Optional[str], src_port: Optional[int], dst_port: Optional[int], ) -> Optional[str]: protocol_upper = str(protocol_name or "").upper() ethernet_upper = str(ethernet_protocol_name or "").upper() ip_upper = str(ip_protocol_name or "").upper() ports = {int(port) for port in (src_port, dst_port) if port is not None} if protocol_upper == "ARP" or ethernet_upper == "ARP": return "ARP" if protocol_upper in {"DHCP", "DHCPV6"} or ports & {67, 68, 546, 547}: return "DHCP" if protocol_upper == "MDNS" or 5353 in ports: return "mDNS" if protocol_upper == "SSDP" or 1900 in ports: return "SSDP" if protocol_upper == "LLMNR" or 5355 in ports: return "LLMNR" if protocol_upper == "NBNS" or ports & {137, 138}: return "NBNS" if protocol_upper == "ICMPV6" or ip_upper == "ICMPV6": return "ICMPv6 Discovery" return None def _safe_ratio(numerator: float, denominator: float) -> float: if denominator <= 0: return 0.0 return numerator / denominator def _serialize_row_for_broadcast(row: Dict[str, Any]) -> Dict[str, Any]: serialized = dict(row) _normalize_json_fields(serialized) _attach_derived_fields(serialized) serialized.pop("capture_session_id", None) raw_val = serialized.get("raw") if isinstance(raw_val, (bytes, bytearray)): serialized["raw_b64"] = base64.b64encode(raw_val).decode("ascii") serialized.pop("raw", None) for key, value in list(serialized.items()): if hasattr(value, "isoformat"): serialized[key] = value.isoformat() return serialized def _normalize_json_fields(payload: Dict[str, Any]) -> None: for key in ("dpi_metadata", "capture_metadata", "telemetry_metadata"): value = payload.get(key) if isinstance(value, str): try: parsed = json.loads(value) except json.JSONDecodeError: continue if isinstance(parsed, dict): payload[key] = parsed 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 tcp_meta = dpi_metadata.get("tcp") if isinstance(tcp_meta, dict): stream = tcp_meta.get("stream") if stream not in (None, "", []): 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"{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"{session_prefix}tcp:{tcp_stream}" udp_stream = tshark_meta.get("udp_stream") if udp_stream not in (None, "", []): return f"{session_prefix}udp:{udp_stream}" return None def _attach_derived_fields(payload: Dict[str, Any]) -> None: flow_id = _derive_flow_id(payload) if flow_id is not None: current_flow_id = payload.get("flow_id") if current_flow_id in (None, ""): payload["flow_id"] = flow_id else: session_id = payload.get("capture_session_id") if session_id not in (None, "") and current_flow_id == flow_id.split(":", 1)[1]: payload["flow_id"] = flow_id if payload.get("raw_present") is None: payload["raw_present"] = payload.get("raw") is not None if payload.get("eth_type") in (None, "") and payload.get("eth_type_raw") is not None: try: payload["eth_type"] = ethertype_from_int(int(payload["eth_type_raw"])) except Exception: payload["eth_type"] = EtherTypeEnum.UNKNOWN if payload.get("ip_proto") in (None, "") and payload.get("ip_proto_raw") is not None: try: payload["ip_proto"] = protocol_from_number(int(payload["ip_proto_raw"])) except Exception: payload["ip_proto"] = int(payload["ip_proto_raw"]) if payload.get("app_master_protocol") in (None, "") and payload.get("app_protocol") not in (None, ""): payload["app_master_protocol"] = payload.get("app_protocol") class DatabasePool: """Asyncpg connection pool wrapper used by the packet APIs.""" def __init__(self, dsn: str, min_size: int = 1, max_size: int = 5): self._dsn = dsn self._pool: Optional[Pool] = None self._min_size = min_size self._max_size = max_size self.broadcaster = None self._init_lock: Optional[asyncio.Lock] = None async def init_pool(self) -> None: """Initialize the connection pool once per process.""" if self._pool is not None: return if self._init_lock is None: self._init_lock = asyncio.Lock() async with self._init_lock: if self._pool is not None: return logger.info("Initializing DB pool (dsn=%s)", self._dsn) try: self._pool = await asyncpg.create_pool( dsn=self._dsn, min_size=self._min_size, max_size=self._max_size, ) logger.info("DB pool initialized") except Exception: logger.exception("Failed to create DB pool") raise async def close_pool(self) -> None: """Close the pool if present.""" if self._pool is None: return try: await self._pool.close() logger.info("DB pool closed") except Exception: logger.exception("Error closing DB pool") finally: self._pool = None async def insert_packet(self, pkt_info: Dict[str, Any]) -> None: """Insert one packet record and publish it to subscribers.""" await self.upsert_packet(pkt_info) async def upsert_packet(self, pkt_info: Dict[str, Any]) -> None: """Insert or update one packet record and publish it to subscribers.""" if self._pool is None: await self.init_pool() _normalize_json_fields(pkt_info) _attach_derived_fields(pkt_info) dpi_metadata = pkt_info.get("dpi_metadata") telemetry_metadata = pkt_info.get("telemetry_metadata") try: async with self._pool.acquire() as conn: row = await conn.fetchrow( """ INSERT INTO packets ( timestamp, correlation_key, packet_id, packet_uid, flow_id, capture_session_id, correlation_source, skb_mark, capture_iface, ingress_if, egress_if, verdict, verdict_reason, verdict_confidence, ingress_seen_at, egress_seen_at, verdict_seen_at, src_mac, dst_mac, eth_type_raw, vlan_id, src_ip, dst_ip, ip_proto_raw, src_port, dst_port, length, capture_sources, app_protocol, app_category, app_confidence, app_hostname, app_is_encrypted, app_risk_score, dpi_metadata, capture_metadata, telemetry_metadata, 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 ) ON CONFLICT (correlation_key) DO UPDATE SET updated_at = NOW(), timestamp = CASE WHEN packets.timestamp IS NULL THEN EXCLUDED.timestamp WHEN EXCLUDED.timestamp IS NULL THEN packets.timestamp ELSE LEAST(packets.timestamp, EXCLUDED.timestamp) END, packet_id = COALESCE(EXCLUDED.packet_id, packets.packet_id), packet_uid = COALESCE(EXCLUDED.packet_uid, packets.packet_uid), flow_id = COALESCE(EXCLUDED.flow_id, packets.flow_id), capture_session_id = COALESCE(EXCLUDED.capture_session_id, packets.capture_session_id), correlation_source = COALESCE(EXCLUDED.correlation_source, packets.correlation_source), skb_mark = COALESCE(EXCLUDED.skb_mark, packets.skb_mark), capture_iface = COALESCE(EXCLUDED.capture_iface, packets.capture_iface), ingress_if = COALESCE(EXCLUDED.ingress_if, packets.ingress_if), egress_if = COALESCE(EXCLUDED.egress_if, packets.egress_if), verdict = COALESCE(EXCLUDED.verdict, packets.verdict), verdict_reason = COALESCE(EXCLUDED.verdict_reason, packets.verdict_reason), verdict_confidence = COALESCE(EXCLUDED.verdict_confidence, packets.verdict_confidence), ingress_seen_at = COALESCE(EXCLUDED.ingress_seen_at, packets.ingress_seen_at), egress_seen_at = COALESCE(EXCLUDED.egress_seen_at, packets.egress_seen_at), verdict_seen_at = COALESCE(EXCLUDED.verdict_seen_at, packets.verdict_seen_at), src_mac = COALESCE(EXCLUDED.src_mac, packets.src_mac), dst_mac = COALESCE(EXCLUDED.dst_mac, packets.dst_mac), eth_type_raw = COALESCE(EXCLUDED.eth_type_raw, packets.eth_type_raw), vlan_id = COALESCE(EXCLUDED.vlan_id, packets.vlan_id), src_ip = COALESCE(EXCLUDED.src_ip, packets.src_ip), dst_ip = COALESCE(EXCLUDED.dst_ip, packets.dst_ip), ip_proto_raw = COALESCE(EXCLUDED.ip_proto_raw, packets.ip_proto_raw), src_port = COALESCE(EXCLUDED.src_port, packets.src_port), dst_port = COALESCE(EXCLUDED.dst_port, packets.dst_port), length = COALESCE(EXCLUDED.length, packets.length), capture_sources = ( SELECT ARRAY( SELECT DISTINCT source FROM unnest( COALESCE(packets.capture_sources, ARRAY[]::text[]) || COALESCE(EXCLUDED.capture_sources, ARRAY[]::text[]) ) AS source ) ), app_protocol = COALESCE(EXCLUDED.app_protocol, packets.app_protocol), app_category = COALESCE(EXCLUDED.app_category, packets.app_category), app_confidence = COALESCE(EXCLUDED.app_confidence, packets.app_confidence), app_hostname = COALESCE(EXCLUDED.app_hostname, packets.app_hostname), app_is_encrypted = COALESCE(EXCLUDED.app_is_encrypted, packets.app_is_encrypted), app_risk_score = COALESCE(EXCLUDED.app_risk_score, packets.app_risk_score), 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), raw = COALESCE(EXCLUDED.raw, packets.raw) RETURNING * """, pkt_info.get("timestamp"), pkt_info["correlation_key"], pkt_info.get("packet_id"), pkt_info.get("packet_uid"), pkt_info.get("flow_id"), pkt_info.get("capture_session_id"), pkt_info.get("correlation_source"), pkt_info.get("skb_mark"), pkt_info.get("capture_iface"), pkt_info.get("ingress_if"), pkt_info.get("egress_if"), pkt_info.get("verdict"), pkt_info.get("verdict_reason"), pkt_info.get("verdict_confidence"), pkt_info.get("ingress_seen_at"), pkt_info.get("egress_seen_at"), pkt_info.get("verdict_seen_at"), pkt_info.get("src_mac"), pkt_info.get("dst_mac"), pkt_info.get("eth_type_raw"), pkt_info.get("vlan_id"), pkt_info.get("src_ip"), pkt_info.get("dst_ip"), pkt_info.get("protocol_raw"), pkt_info.get("src_port"), pkt_info.get("dst_port"), pkt_info.get("length"), pkt_info.get("capture_sources"), pkt_info.get("app_protocol"), pkt_info.get("app_category"), pkt_info.get("app_confidence"), pkt_info.get("app_hostname"), pkt_info.get("app_is_encrypted"), pkt_info.get("app_risk_score"), 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, pkt_info.get("raw"), ) except Exception: logger.exception("DB upsert failed") return if row: persisted = _serialize_row_for_broadcast(dict(row)) pkt_info.update(persisted) if self.broadcaster: try: self.broadcaster.sync_publish(pkt_info) except Exception: logger.exception("Failed to publish pkt_info to broadcaster") async def backfill_packet_metadata( self, *, iface: str, eth_type_raw: Optional[int], src_ip: str, dst_ip: str, src_port: int, dst_port: int, protocol: Optional[int], length: int, observed_at_ms: int, enrichment: Dict[str, Any], window_ms: int, ) -> int: """Update recent packet rows after tshark metadata arrives asynchronously.""" if self._pool is None: await self.init_pool() lower_bound = datetime.fromtimestamp(max(observed_at_ms - window_ms, 0) / 1000.0, tz=timezone.utc) upper_bound = datetime.fromtimestamp(max(observed_at_ms + window_ms, 0) / 1000.0, tz=timezone.utc) dpi_metadata = enrichment.get("dpi_metadata") capture_sources = [str(source) for source in enrichment.get("capture_sources", []) if source] try: async with self._pool.acquire() as conn: rows = await conn.fetch( """ WITH candidate_rows AS ( SELECT id FROM packets WHERE ( ($1::int IS NOT NULL AND ip_proto_raw = $1) OR ($1::int IS NULL AND $2::int IS NOT NULL AND eth_type_raw = $2) ) AND (capture_iface = $3 OR ingress_if = $3 OR egress_if = $3) AND src_ip = $4::inet AND dst_ip = $5::inet AND COALESCE(src_port, 0) = $6 AND COALESCE(dst_port, 0) = $7 AND length = $8 AND timestamp BETWEEN $9 AND $10 AND ( app_protocol IS NULL OR app_category IS NULL OR app_confidence IS NULL OR app_hostname IS NULL OR app_is_encrypted IS NULL OR ($16::jsonb IS NOT NULL) OR (COALESCE(array_length($17::text[], 1), 0) > 0) ) ORDER BY id FOR UPDATE SKIP LOCKED ) UPDATE packets SET updated_at = NOW(), app_protocol = COALESCE(packets.app_protocol, $11), app_category = COALESCE(packets.app_category, $12), app_confidence = COALESCE(packets.app_confidence, $13), app_hostname = COALESCE(packets.app_hostname, $14), app_is_encrypted = COALESCE(packets.app_is_encrypted, $15), dpi_metadata = CASE WHEN $16::jsonb IS NULL THEN packets.dpi_metadata WHEN packets.dpi_metadata IS NULL THEN $16::jsonb ELSE packets.dpi_metadata || $16::jsonb END, flow_id = COALESCE( packets.flow_id, CASE WHEN COALESCE($16::jsonb -> 'tcp' ->> 'stream', $16::jsonb -> 'tshark' ->> 'tcp_stream') IS NOT NULL AND COALESCE($16::jsonb -> 'tcp' ->> 'stream', $16::jsonb -> 'tshark' ->> 'tcp_stream') <> '' THEN COALESCE(NULLIF(packets.capture_session_id, '') || ':', '') || 'tcp:' || COALESCE($16::jsonb -> 'tcp' ->> 'stream', $16::jsonb -> 'tshark' ->> 'tcp_stream') WHEN COALESCE($16::jsonb -> 'udp' ->> 'stream', $16::jsonb -> 'tshark' ->> 'udp_stream') IS NOT NULL AND COALESCE($16::jsonb -> 'udp' ->> 'stream', $16::jsonb -> 'tshark' ->> 'udp_stream') <> '' THEN COALESCE(NULLIF(packets.capture_session_id, '') || ':', '') || 'udp:' || COALESCE($16::jsonb -> 'udp' ->> 'stream', $16::jsonb -> 'tshark' ->> 'udp_stream') ELSE NULL END ), capture_sources = ( SELECT ARRAY( SELECT DISTINCT source FROM unnest( COALESCE(packets.capture_sources, ARRAY[]::text[]) || COALESCE($17::text[], ARRAY[]::text[]) ) AS source ) ) FROM candidate_rows WHERE packets.id = candidate_rows.id RETURNING packets.* """, protocol, eth_type_raw, iface, src_ip, dst_ip, src_port, dst_port, length, lower_bound, upper_bound, enrichment.get("app_protocol"), enrichment.get("app_category"), enrichment.get("app_confidence"), enrichment.get("app_hostname"), enrichment.get("app_is_encrypted"), json.dumps(dpi_metadata) if dpi_metadata is not None else None, capture_sources if capture_sources else None, ) except Exception: logger.exception("DB packet metadata backfill failed") return 0 if not rows: return 0 updated_count = 0 for row in rows: serialized = _serialize_row_for_broadcast(dict(row)) updated_count += 1 if self.broadcaster: try: self.broadcaster.sync_publish(serialized) except Exception: logger.exception("Failed to publish tshark-enriched packet row") return updated_count async def backfill_stream_metadata( self, *, iface: str, protocol: int, stream_kind: str, stream_id: int, observed_at_ms: int, enrichment: Dict[str, Any], window_ms: int, ) -> int: """Propagate tshark stream context across packets already tagged with the same stream id.""" if self._pool is None: await self.init_pool() lower_bound = datetime.fromtimestamp(max(observed_at_ms - window_ms, 0) / 1000.0, tz=timezone.utc) upper_bound = datetime.fromtimestamp(max(observed_at_ms + window_ms, 0) / 1000.0, tz=timezone.utc) capture_sources = [str(source) for source in enrichment.get("capture_sources", []) if source] stream_kind = str(stream_kind).lower() if stream_kind not in {"tcp", "udp"}: return 0 try: async with self._pool.acquire() as conn: rows = await conn.fetch( """ WITH candidate_rows AS ( SELECT id FROM packets WHERE ip_proto_raw = $1 AND (capture_iface = $2 OR ingress_if = $2 OR egress_if = $2) AND timestamp BETWEEN $5 AND $6 AND ( ( CASE WHEN $3 = 'tcp' THEN COALESCE(dpi_metadata -> 'tcp' ->> 'stream', '') WHEN $3 = 'udp' THEN COALESCE(dpi_metadata -> 'udp' ->> 'stream', '') ELSE '' END ) = $4 OR flow_id = ($3 || ':' || $4) OR flow_id LIKE ('%:' || $3 || ':' || $4) ) AND ( app_protocol IS NULL OR UPPER(app_protocol) IN ('TCP', 'UDP', 'IP', 'IPV4', 'IPV6', 'ETH', 'ETHERNET', 'ETHERTYPE', 'FRAME', 'DATA') OR app_category IS NULL OR app_category IN ('Transport', 'Network', 'Protocol') OR app_confidence IS NULL OR app_hostname IS NULL OR app_is_encrypted IS NULL OR (COALESCE(array_length($12::text[], 1), 0) > 0) ) ORDER BY id FOR UPDATE SKIP LOCKED ) UPDATE packets SET updated_at = NOW(), app_protocol = CASE WHEN packets.app_protocol IS NULL OR UPPER(packets.app_protocol) IN ('TCP', 'UDP', 'IP', 'IPV4', 'IPV6', 'ETH', 'ETHERNET', 'ETHERTYPE', 'FRAME', 'DATA') THEN COALESCE($7, packets.app_protocol) ELSE packets.app_protocol END, app_category = CASE WHEN packets.app_category IS NULL OR packets.app_category IN ('Transport', 'Network', 'Protocol') THEN COALESCE($8, packets.app_category) ELSE packets.app_category END, app_confidence = CASE WHEN packets.app_protocol IS NULL OR UPPER(packets.app_protocol) IN ('TCP', 'UDP', 'IP', 'IPV4', 'IPV6', 'ETH', 'ETHERNET', 'ETHERTYPE', 'FRAME', 'DATA') THEN COALESCE($9, packets.app_confidence) ELSE packets.app_confidence END, app_hostname = COALESCE(packets.app_hostname, $10), app_is_encrypted = COALESCE(packets.app_is_encrypted, $11), flow_id = COALESCE( packets.flow_id, COALESCE(NULLIF(packets.capture_session_id, '') || ':', '') || $3 || ':' || $4 ), capture_sources = ( SELECT ARRAY( SELECT DISTINCT source FROM unnest( COALESCE(packets.capture_sources, ARRAY[]::text[]) || COALESCE($12::text[], ARRAY[]::text[]) ) AS source ) ) FROM candidate_rows WHERE packets.id = candidate_rows.id RETURNING packets.* """, protocol, iface, stream_kind, str(stream_id), lower_bound, upper_bound, enrichment.get("app_protocol"), enrichment.get("app_category"), enrichment.get("app_confidence"), enrichment.get("app_hostname"), enrichment.get("app_is_encrypted"), capture_sources if capture_sources else None, ) except Exception: logger.exception("DB stream metadata backfill failed") return 0 if not rows: return 0 updated_count = 0 for row in rows: serialized = _serialize_row_for_broadcast(dict(row)) updated_count += 1 if self.broadcaster: try: self.broadcaster.sync_publish(serialized) except Exception: logger.exception("Failed to publish stream-context packet row") return updated_count async def fetch_latest(self, limit: int) -> List[PacketDBModel]: """Fetch newest packet rows as validated `PacketDBModel` instances.""" if self._pool is None: await self.init_pool() async with self._pool.acquire() as conn: rows = await conn.fetch( """ SELECT * FROM packets ORDER BY timestamp DESC, id DESC LIMIT $1 """, limit, ) result: List[PacketDBModel] = [] 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)): data["raw_b64"] = base64.b64encode(raw_val).decode("ascii") data.pop("raw", None) try: packet_model = PacketDBModel(**data) except ValidationError as exc: logger.warning( "Skipping DB row that failed PacketDBModel validation (id=%s): %s", data.get("id"), exc, ) continue result.append(packet_model) return result async def infer_interface_hosts( self, *, since: Optional[datetime] = None, limit_per_interface: int = 100, ) -> List[Dict[str, Any]]: """Infer which IP/MAC endpoints are attached to each observed interface.""" if self._pool is None: await self.init_pool() try: async with self._pool.acquire() as conn: rows = await conn.fetch( """ WITH observations AS ( SELECT ingress_if AS iface, src_ip::text AS ip_address, src_mac::text AS mac_address, timestamp, 'source_on_ingress' AS evidence FROM packets WHERE ingress_if IS NOT NULL AND (src_ip IS NOT NULL OR src_mac IS NOT NULL) AND ($1::timestamptz IS NULL OR timestamp >= $1) UNION ALL SELECT egress_if AS iface, dst_ip::text AS ip_address, dst_mac::text AS mac_address, timestamp, 'destination_on_egress' AS evidence FROM packets WHERE egress_if IS NOT NULL AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) AND ($1::timestamptz IS NULL OR timestamp >= $1) ), filtered AS ( SELECT * FROM observations WHERE iface IS NOT NULL AND COALESCE(mac_address, '') <> 'ff:ff:ff:ff:ff:ff' AND ( COALESCE(ip_address, '') <> '' OR COALESCE(mac_address, '') <> '' ) ), aggregated AS ( SELECT iface, ip_address, mac_address, COUNT(*) AS packet_count, MAX(timestamp) AS last_seen, SUM(CASE WHEN evidence = 'source_on_ingress' THEN 1 ELSE 0 END) AS source_on_ingress_count, SUM(CASE WHEN evidence = 'destination_on_egress' THEN 1 ELSE 0 END) AS destination_on_egress_count FROM filtered GROUP BY iface, ip_address, mac_address ), ranked AS ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY iface ORDER BY packet_count DESC, last_seen DESC, ip_address, mac_address ) AS row_num FROM aggregated ) SELECT iface, ip_address, mac_address, packet_count, last_seen, source_on_ingress_count, destination_on_egress_count FROM ranked WHERE row_num <= $2 ORDER BY iface, packet_count DESC, last_seen DESC, ip_address, mac_address """, since, limit_per_interface, ) except Exception: logger.exception("DB interface-host analysis failed") raise grouped: Dict[str, List[Dict[str, Any]]] = {} for row in rows: record = dict(row) iface = str(record.pop("iface")) if hasattr(record.get("last_seen"), "isoformat"): record["last_seen"] = record["last_seen"].isoformat() grouped.setdefault(iface, []).append(record) return [ { "interface": iface, "hosts": hosts, } for iface, hosts in sorted(grouped.items()) ] async def infer_interface_host_protocols( self, *, since: Optional[datetime] = None, limit_per_interface: int = 50, limit_protocols_per_host: int = 12, ) -> List[Dict[str, Any]]: """Infer interface-host attachment and aggregate observed protocols and verdicts.""" if self._pool is None: await self.init_pool() try: async with self._pool.acquire() as conn: rows = await conn.fetch( """ WITH observations AS ( SELECT ingress_if AS iface, src_ip::text AS ip_address, src_mac::text AS mac_address, NULLIF(app_protocol::text, '') AS app_protocol_name, ip_proto_raw, eth_type_raw, COALESCE(NULLIF(verdict::text, ''), 'unknown') AS verdict_name, timestamp, 'source_on_ingress' AS evidence FROM packets WHERE ingress_if IS NOT NULL AND (src_ip IS NOT NULL OR src_mac IS NOT NULL) AND ($1::timestamptz IS NULL OR timestamp >= $1) UNION ALL SELECT egress_if AS iface, dst_ip::text AS ip_address, dst_mac::text AS mac_address, NULLIF(app_protocol::text, '') AS app_protocol_name, ip_proto_raw, eth_type_raw, COALESCE(NULLIF(verdict::text, ''), 'unknown') AS verdict_name, timestamp, 'destination_on_egress' AS evidence FROM packets WHERE egress_if IS NOT NULL AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) AND ($1::timestamptz IS NULL OR timestamp >= $1) ), filtered AS ( SELECT * FROM observations WHERE iface IS NOT NULL AND COALESCE(mac_address, '') <> 'ff:ff:ff:ff:ff:ff' AND ( COALESCE(ip_address, '') <> '' OR COALESCE(mac_address, '') <> '' ) ), host_aggregated AS ( SELECT iface, ip_address, mac_address, COUNT(*) AS packet_count, MAX(timestamp) AS last_seen, SUM(CASE WHEN evidence = 'source_on_ingress' THEN 1 ELSE 0 END) AS source_on_ingress_count, SUM(CASE WHEN evidence = 'destination_on_egress' THEN 1 ELSE 0 END) AS destination_on_egress_count FROM filtered GROUP BY iface, ip_address, mac_address ), selected_hosts AS ( SELECT * FROM ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY iface ORDER BY packet_count DESC, last_seen DESC, ip_address, mac_address ) AS row_num FROM host_aggregated ) ranked_hosts WHERE row_num <= $2 ), protocol_aggregated AS ( SELECT filtered.iface, filtered.ip_address, filtered.mac_address, filtered.app_protocol_name, filtered.ip_proto_raw, filtered.eth_type_raw, COUNT(*) AS packet_count, MAX(filtered.timestamp) AS last_seen, SUM(CASE WHEN filtered.verdict_name = 'accept' THEN 1 ELSE 0 END) AS accept_count, SUM(CASE WHEN filtered.verdict_name = 'drop' THEN 1 ELSE 0 END) AS drop_count, SUM(CASE WHEN filtered.verdict_name = 'reject' THEN 1 ELSE 0 END) AS reject_count, SUM(CASE WHEN filtered.verdict_name NOT IN ('accept', 'drop', 'reject') THEN 1 ELSE 0 END) AS unknown_count FROM filtered INNER JOIN selected_hosts ON selected_hosts.iface = filtered.iface AND selected_hosts.ip_address IS NOT DISTINCT FROM filtered.ip_address AND selected_hosts.mac_address IS NOT DISTINCT FROM filtered.mac_address GROUP BY filtered.iface, filtered.ip_address, filtered.mac_address, filtered.app_protocol_name, filtered.ip_proto_raw, filtered.eth_type_raw ), ranked_protocols AS ( SELECT * FROM ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY iface, ip_address, mac_address ORDER BY packet_count DESC, last_seen DESC, app_protocol_name NULLS LAST, ip_proto_raw NULLS LAST, eth_type_raw NULLS LAST ) AS row_num FROM protocol_aggregated ) ranked WHERE row_num <= $3 ) SELECT selected_hosts.iface, selected_hosts.ip_address, selected_hosts.mac_address, selected_hosts.packet_count AS host_packet_count, selected_hosts.last_seen AS host_last_seen, selected_hosts.source_on_ingress_count, selected_hosts.destination_on_egress_count, ranked_protocols.app_protocol_name, ranked_protocols.ip_proto_raw, ranked_protocols.eth_type_raw, ranked_protocols.packet_count AS protocol_packet_count, ranked_protocols.last_seen AS protocol_last_seen, ranked_protocols.accept_count, ranked_protocols.drop_count, ranked_protocols.reject_count, ranked_protocols.unknown_count FROM selected_hosts LEFT JOIN ranked_protocols ON ranked_protocols.iface = selected_hosts.iface AND ranked_protocols.ip_address IS NOT DISTINCT FROM selected_hosts.ip_address AND ranked_protocols.mac_address IS NOT DISTINCT FROM selected_hosts.mac_address ORDER BY selected_hosts.iface, selected_hosts.packet_count DESC, selected_hosts.last_seen DESC, selected_hosts.ip_address, selected_hosts.mac_address, ranked_protocols.packet_count DESC NULLS LAST, ranked_protocols.last_seen DESC NULLS LAST, ranked_protocols.app_protocol_name NULLS LAST, ranked_protocols.ip_proto_raw NULLS LAST, ranked_protocols.eth_type_raw NULLS LAST """, since, limit_per_interface, limit_protocols_per_host, ) except Exception: logger.exception("DB interface-host-protocol analysis failed") raise grouped: Dict[str, Dict[str, Dict[str, Any]]] = {} for row in rows: record = dict(row) iface = str(record["iface"]) host_key = f"{record.get('ip_address') or 'no-ip'}|{record.get('mac_address') or 'no-mac'}" iface_hosts = grouped.setdefault(iface, {}) host_record = iface_hosts.get(host_key) if host_record is None: host_record = { "ip_address": record.get("ip_address"), "mac_address": record.get("mac_address"), "packet_count": int(record.get("host_packet_count") or 0), "last_seen": record["host_last_seen"].isoformat() if hasattr(record.get("host_last_seen"), "isoformat") else record.get("host_last_seen"), "source_on_ingress_count": int(record.get("source_on_ingress_count") or 0), "destination_on_egress_count": int(record.get("destination_on_egress_count") or 0), "protocols": [], "_protocol_index": {}, } iface_hosts[host_key] = host_record protocol_name = _analysis_protocol_name( record.get("app_protocol_name"), record.get("ip_proto_raw"), record.get("eth_type_raw"), ) if protocol_name not in (None, ""): protocol_key = str(protocol_name) ip_protocol_name = _analysis_ip_protocol_name(record.get("ip_proto_raw")) ethernet_protocol_name = _analysis_ethernet_protocol_name(record.get("eth_type_raw")) protocol_last_seen_raw = record.get("protocol_last_seen") protocol_last_seen = ( protocol_last_seen_raw.isoformat() if hasattr(protocol_last_seen_raw, "isoformat") else protocol_last_seen_raw ) protocol_index = host_record["_protocol_index"] protocol_record = protocol_index.get(protocol_key) if protocol_record is None: protocol_record = { "protocol": protocol_key, "packet_count": 0, "last_seen": protocol_last_seen, "accept_count": 0, "drop_count": 0, "reject_count": 0, "unknown_count": 0, "ethernet_protocol": ethernet_protocol_name, "ip_protocol": ip_protocol_name, "layer_paths": [], "_layer_index": {}, } protocol_index[protocol_key] = protocol_record host_record["protocols"].append(protocol_record) protocol_record["packet_count"] += int(record.get("protocol_packet_count") or 0) protocol_record["accept_count"] += int(record.get("accept_count") or 0) protocol_record["drop_count"] += int(record.get("drop_count") or 0) protocol_record["reject_count"] += int(record.get("reject_count") or 0) protocol_record["unknown_count"] += int(record.get("unknown_count") or 0) if protocol_last_seen and ( protocol_record.get("last_seen") in (None, "") or str(protocol_last_seen) > str(protocol_record.get("last_seen")) ): protocol_record["last_seen"] = protocol_last_seen if protocol_record.get("ethernet_protocol") is None and ethernet_protocol_name is not None: protocol_record["ethernet_protocol"] = ethernet_protocol_name if protocol_record.get("ip_protocol") is None and ip_protocol_name is not None: protocol_record["ip_protocol"] = ip_protocol_name layer_key = f"{ethernet_protocol_name or 'no-eth'}|{ip_protocol_name or 'no-ip'}" layer_index = protocol_record["_layer_index"] layer_record = layer_index.get(layer_key) if layer_record is None: layer_record = { "ethernet_protocol": ethernet_protocol_name, "ip_protocol": ip_protocol_name, "packet_count": 0, "last_seen": protocol_last_seen, "accept_count": 0, "drop_count": 0, "reject_count": 0, "unknown_count": 0, } layer_index[layer_key] = layer_record protocol_record["layer_paths"].append(layer_record) layer_record["packet_count"] += int(record.get("protocol_packet_count") or 0) layer_record["accept_count"] += int(record.get("accept_count") or 0) layer_record["drop_count"] += int(record.get("drop_count") or 0) layer_record["reject_count"] += int(record.get("reject_count") or 0) layer_record["unknown_count"] += int(record.get("unknown_count") or 0) if protocol_last_seen and ( layer_record.get("last_seen") in (None, "") or str(protocol_last_seen) > str(layer_record.get("last_seen")) ): layer_record["last_seen"] = protocol_last_seen result: List[Dict[str, Any]] = [] for iface, hosts in sorted(grouped.items()): for host in hosts.values(): host.pop("_protocol_index", None) for protocol in host["protocols"]: protocol.pop("_layer_index", None) protocol["layer_paths"] = sorted( protocol["layer_paths"], key=lambda item: ( -int(item.get("packet_count") or 0), str(item.get("last_seen") or ""), str(item.get("ethernet_protocol") or ""), str(item.get("ip_protocol") or ""), ), ) host["protocols"] = sorted( host["protocols"], key=lambda item: ( -int(item.get("packet_count") or 0), str(item.get("last_seen") or ""), str(item.get("protocol") or ""), ), ) sorted_hosts = sorted( hosts.values(), key=lambda item: (-int(item.get("packet_count") or 0), str(item.get("last_seen") or ""), str(item.get("ip_address") or ""), str(item.get("mac_address") or "")), ) result.append({"interface": iface, "hosts": sorted_hosts}) return result async def infer_interface_protocol_paths( self, *, since: Optional[datetime] = None, limit_paths: int = 500, ) -> List[Dict[str, Any]]: """Aggregate directional packet paths for Sankey rendering.""" if self._pool is None: await self.init_pool() try: async with self._pool.acquire() as conn: rows = await conn.fetch( """ WITH aggregated AS ( SELECT ingress_if, egress_if, src_ip::text AS src_ip_address, src_mac::text AS src_mac_address, dst_ip::text AS dst_ip_address, dst_mac::text AS dst_mac_address, NULLIF(app_protocol::text, '') AS app_protocol_name, ip_proto_raw, eth_type_raw, COUNT(*) AS packet_count, MAX(timestamp) AS last_seen, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') = 'accept' THEN 1 ELSE 0 END) AS accept_count, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') = 'drop' THEN 1 ELSE 0 END) AS drop_count, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') = 'reject' THEN 1 ELSE 0 END) AS reject_count, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') NOT IN ('accept', 'drop', 'reject') THEN 1 ELSE 0 END) AS unknown_count FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND (ingress_if IS NOT NULL OR egress_if IS NOT NULL) AND (src_ip IS NOT NULL OR src_mac IS NOT NULL) AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) AND COALESCE(src_mac::text, '') <> 'ff:ff:ff:ff:ff:ff' AND COALESCE(dst_mac::text, '') <> 'ff:ff:ff:ff:ff:ff' GROUP BY ingress_if, egress_if, src_ip::text, src_mac::text, dst_ip::text, dst_mac::text, app_protocol_name, ip_proto_raw, eth_type_raw ) SELECT * FROM aggregated ORDER BY packet_count DESC, last_seen DESC LIMIT $2 """, since, limit_paths, ) except Exception: logger.exception("DB interface-protocol-path analysis failed") raise grouped: Dict[str, Dict[str, Any]] = {} for row in rows: record = dict(row) protocol_name = _analysis_protocol_name( record.get("app_protocol_name"), record.get("ip_proto_raw"), record.get("eth_type_raw"), ) ip_protocol_name = _analysis_ip_protocol_name(record.get("ip_proto_raw")) ethernet_protocol_name = _analysis_ethernet_protocol_name(record.get("eth_type_raw")) last_seen_raw = record.get("last_seen") last_seen = last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw path_key = "|".join( [ str(record.get("ingress_if") or ""), str(record.get("src_ip_address") or ""), str(record.get("src_mac_address") or ""), str(protocol_name or ""), str(record.get("dst_ip_address") or ""), str(record.get("dst_mac_address") or ""), str(record.get("egress_if") or ""), ] ) path_record = grouped.get(path_key) if path_record is None: path_record = { "ingress_interface": record.get("ingress_if"), "egress_interface": record.get("egress_if"), "src_ip_address": record.get("src_ip_address"), "src_mac_address": record.get("src_mac_address"), "dst_ip_address": record.get("dst_ip_address"), "dst_mac_address": record.get("dst_mac_address"), "protocol": str(protocol_name), "ethernet_protocol": ethernet_protocol_name, "ip_protocol": ip_protocol_name, "packet_count": 0, "last_seen": last_seen, "accept_count": 0, "drop_count": 0, "reject_count": 0, "unknown_count": 0, } grouped[path_key] = path_record path_record["packet_count"] += int(record.get("packet_count") or 0) path_record["accept_count"] += int(record.get("accept_count") or 0) path_record["drop_count"] += int(record.get("drop_count") or 0) path_record["reject_count"] += int(record.get("reject_count") or 0) path_record["unknown_count"] += int(record.get("unknown_count") or 0) if path_record.get("ethernet_protocol") is None and ethernet_protocol_name is not None: path_record["ethernet_protocol"] = ethernet_protocol_name if path_record.get("ip_protocol") is None and ip_protocol_name is not None: path_record["ip_protocol"] = ip_protocol_name if last_seen and ( path_record.get("last_seen") in (None, "") or str(last_seen) > str(path_record.get("last_seen")) ): path_record["last_seen"] = last_seen return sorted( grouped.values(), key=lambda item: ( -int(item.get("packet_count") or 0), str(item.get("last_seen") or ""), str(item.get("ingress_interface") or ""), str(item.get("src_ip_address") or ""), str(item.get("dst_ip_address") or ""), str(item.get("egress_interface") or ""), ), ) async def analyze_conversations( self, *, since: Optional[datetime] = None, limit: int = 300, ) -> List[Dict[str, Any]]: """Aggregate directional conversations between endpoints.""" if self._pool is None: await self.init_pool() try: async with self._pool.acquire() as conn: rows = await conn.fetch( """ WITH aggregated AS ( SELECT ingress_if, egress_if, src_ip::text AS src_ip_address, src_mac::text AS src_mac_address, dst_ip::text AS dst_ip_address, dst_mac::text AS dst_mac_address, src_port, dst_port, NULLIF(app_protocol::text, '') AS app_protocol_name, ip_proto_raw, eth_type_raw, ARRAY_REMOVE(ARRAY_AGG(DISTINCT NULLIF(app_hostname::text, '')), NULL) AS hostnames, COUNT(*) AS packet_count, COALESCE(SUM(length), 0) AS byte_count, MIN(timestamp) AS first_seen, MAX(timestamp) AS last_seen, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') = 'accept' THEN 1 ELSE 0 END) AS accept_count, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') = 'drop' THEN 1 ELSE 0 END) AS drop_count, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') = 'reject' THEN 1 ELSE 0 END) AS reject_count, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') NOT IN ('accept', 'drop', 'reject') THEN 1 ELSE 0 END) AS unknown_count FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND (src_ip IS NOT NULL OR src_mac IS NOT NULL) AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) GROUP BY ingress_if, egress_if, src_ip::text, src_mac::text, dst_ip::text, dst_mac::text, src_port, dst_port, app_protocol_name, ip_proto_raw, eth_type_raw ) SELECT * FROM aggregated ORDER BY packet_count DESC, byte_count DESC, last_seen DESC LIMIT $2 """, since, limit, ) except Exception: logger.exception("DB conversation analysis failed") raise result: List[Dict[str, Any]] = [] for row in rows: record = dict(row) protocol_name = _analysis_protocol_name( record.get("app_protocol_name"), record.get("ip_proto_raw"), record.get("eth_type_raw"), ) first_seen_raw = record.get("first_seen") last_seen_raw = record.get("last_seen") result.append( { "ingress_interface": record.get("ingress_if"), "egress_interface": record.get("egress_if"), "src_ip_address": record.get("src_ip_address"), "src_mac_address": record.get("src_mac_address"), "dst_ip_address": record.get("dst_ip_address"), "dst_mac_address": record.get("dst_mac_address"), "src_port": record.get("src_port"), "dst_port": record.get("dst_port"), "protocol": str(protocol_name), "ethernet_protocol": _analysis_ethernet_protocol_name(record.get("eth_type_raw")), "ip_protocol": _analysis_ip_protocol_name(record.get("ip_proto_raw")), "hostnames": list(record.get("hostnames") or []), "packet_count": int(record.get("packet_count") or 0), "byte_count": int(record.get("byte_count") or 0), "first_seen": first_seen_raw.isoformat() if hasattr(first_seen_raw, "isoformat") else first_seen_raw, "last_seen": last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw, "accept_count": int(record.get("accept_count") or 0), "drop_count": int(record.get("drop_count") or 0), "reject_count": int(record.get("reject_count") or 0), "unknown_count": int(record.get("unknown_count") or 0), } ) return result async def analyze_host_intelligence( self, *, since: Optional[datetime] = None, limit_hosts: int = 40, ) -> List[Dict[str, Any]]: """Build host-centric intelligence with peers, services, and hostname enrichment.""" if self._pool is None: await self.init_pool() try: async with self._pool.acquire() as conn: host_rows = await conn.fetch( """ WITH observations AS ( SELECT src_ip::text AS ip_address, src_mac::text AS mac_address, ingress_if AS iface, COALESCE(length, 0) AS packet_length, timestamp, 'source' AS role FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND ingress_if IS NOT NULL AND (src_ip IS NOT NULL OR src_mac IS NOT NULL) UNION ALL SELECT dst_ip::text AS ip_address, dst_mac::text AS mac_address, egress_if AS iface, COALESCE(length, 0) AS packet_length, timestamp, 'destination' AS role FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND egress_if IS NOT NULL AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) ) SELECT ip_address, mac_address, COUNT(*) AS packet_count, COALESCE(SUM(packet_length), 0) AS byte_count, MIN(timestamp) AS first_seen, MAX(timestamp) AS last_seen, ARRAY_REMOVE(ARRAY_AGG(DISTINCT iface), NULL) AS interfaces, SUM(CASE WHEN role = 'source' THEN 1 ELSE 0 END) AS source_count, SUM(CASE WHEN role = 'destination' THEN 1 ELSE 0 END) AS destination_count FROM observations WHERE COALESCE(mac_address, '') <> 'ff:ff:ff:ff:ff:ff' GROUP BY ip_address, mac_address ORDER BY packet_count DESC, byte_count DESC, last_seen DESC LIMIT $2 """, since, limit_hosts, ) host_keys = [ _analysis_host_key(record.get("ip_address"), record.get("mac_address")) for record in (dict(row) for row in host_rows) ] if not host_keys: return [] detail_rows = await conn.fetch( """ SELECT src_ip::text AS src_ip_address, src_mac::text AS src_mac_address, dst_ip::text AS dst_ip_address, dst_mac::text AS dst_mac_address, ingress_if, egress_if, src_port, dst_port, NULLIF(app_protocol::text, '') AS app_protocol_name, ip_proto_raw, eth_type_raw, NULLIF(app_hostname::text, '') AS app_hostname, COALESCE(length, 0) AS packet_length, timestamp, COALESCE(NULLIF(verdict::text, ''), 'unknown') AS verdict_name FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND ( (COALESCE(src_ip::text, '') || '|' || COALESCE(src_mac::text, '')) = ANY($2::text[]) OR (COALESCE(dst_ip::text, '') || '|' || COALESCE(dst_mac::text, '')) = ANY($2::text[]) ) ORDER BY timestamp DESC """, since, host_keys, ) except Exception: logger.exception("DB host intelligence analysis failed") raise host_index: Dict[str, Dict[str, Any]] = {} for row in host_rows: record = dict(row) host_key = _analysis_host_key(record.get("ip_address"), record.get("mac_address")) first_seen_raw = record.get("first_seen") last_seen_raw = record.get("last_seen") host_index[host_key] = { "ip_address": record.get("ip_address"), "mac_address": record.get("mac_address"), "packet_count": int(record.get("packet_count") or 0), "byte_count": int(record.get("byte_count") or 0), "first_seen": first_seen_raw.isoformat() if hasattr(first_seen_raw, "isoformat") else first_seen_raw, "last_seen": last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw, "interfaces": list(record.get("interfaces") or []), "source_count": int(record.get("source_count") or 0), "destination_count": int(record.get("destination_count") or 0), "hostnames": set(), "_protocol_index": {}, "_peer_index": {}, "_service_index": {}, } for row in detail_rows: record = dict(row) protocol_name = str( _analysis_protocol_name( record.get("app_protocol_name"), record.get("ip_proto_raw"), record.get("eth_type_raw"), ) ) app_hostname = record.get("app_hostname") packet_length = int(record.get("packet_length") or 0) timestamp_raw = record.get("timestamp") timestamp = timestamp_raw.isoformat() if hasattr(timestamp_raw, "isoformat") else timestamp_raw src_key = _analysis_host_key(record.get("src_ip_address"), record.get("src_mac_address")) dst_key = _analysis_host_key(record.get("dst_ip_address"), record.get("dst_mac_address")) for role, host_key, peer_ip, peer_mac, service_port in ( ("source", src_key, record.get("dst_ip_address"), record.get("dst_mac_address"), record.get("src_port")), ("destination", dst_key, record.get("src_ip_address"), record.get("src_mac_address"), record.get("dst_port")), ): host_record = host_index.get(host_key) if host_record is None: continue if app_hostname: host_record["hostnames"].add(str(app_hostname)) protocol_record = host_record["_protocol_index"].setdefault( protocol_name, {"label": protocol_name, "packet_count": 0}, ) protocol_record["packet_count"] += 1 peer_key = _analysis_host_key(peer_ip, peer_mac) peer_record = host_record["_peer_index"].get(peer_key) if peer_record is None: peer_record = { "ip_address": peer_ip, "mac_address": peer_mac, "packet_count": 0, "byte_count": 0, "last_seen": timestamp, "protocols": set(), } host_record["_peer_index"][peer_key] = peer_record peer_record["packet_count"] += 1 peer_record["byte_count"] += packet_length if timestamp and ( peer_record.get("last_seen") in (None, "") or str(timestamp) > str(peer_record.get("last_seen")) ): peer_record["last_seen"] = timestamp peer_record["protocols"].add(protocol_name) if role == "destination" and service_port is not None: service_key = f"{service_port}|{protocol_name}" service_record = host_record["_service_index"].get(service_key) if service_record is None: service_record = { "port": int(service_port), "protocol": protocol_name, "packet_count": 0, "byte_count": 0, "last_seen": timestamp, "hostnames": set(), } host_record["_service_index"][service_key] = service_record service_record["packet_count"] += 1 service_record["byte_count"] += packet_length if app_hostname: service_record["hostnames"].add(str(app_hostname)) if timestamp and ( service_record.get("last_seen") in (None, "") or str(timestamp) > str(service_record.get("last_seen")) ): service_record["last_seen"] = timestamp hosts: List[Dict[str, Any]] = [] for host in host_index.values(): protocols = sorted( host["_protocol_index"].values(), key=lambda item: (-int(item.get("packet_count") or 0), str(item.get("label") or "")), )[:5] peers = sorted( host["_peer_index"].values(), key=lambda item: (-int(item.get("packet_count") or 0), str(item.get("last_seen") or "")), )[:6] services = sorted( host["_service_index"].values(), key=lambda item: (-int(item.get("packet_count") or 0), str(item.get("last_seen") or ""), int(item.get("port") or 0)), )[:6] for peer in peers: peer["protocols"] = sorted(peer["protocols"]) for service in services: service["hostnames"] = sorted(service["hostnames"]) hosts.append( { "ip_address": host.get("ip_address"), "mac_address": host.get("mac_address"), "packet_count": int(host.get("packet_count") or 0), "byte_count": int(host.get("byte_count") or 0), "first_seen": host.get("first_seen"), "last_seen": host.get("last_seen"), "interfaces": sorted(host.get("interfaces") or []), "source_count": int(host.get("source_count") or 0), "destination_count": int(host.get("destination_count") or 0), "hostnames": sorted(host["hostnames"]), "top_protocols": protocols, "peers": peers, "services": services, } ) return sorted( hosts, key=lambda item: ( -int(item.get("packet_count") or 0), -int(item.get("byte_count") or 0), str(item.get("last_seen") or ""), str(item.get("ip_address") or ""), ), ) async def analyze_discovery_activity( self, *, since: Optional[datetime] = None, limit: int = 300, ) -> List[Dict[str, Any]]: """Aggregate discovery and local service advertisement traffic.""" if self._pool is None: await self.init_pool() try: async with self._pool.acquire() as conn: rows = await conn.fetch( """ SELECT ingress_if, egress_if, src_ip::text AS src_ip_address, src_mac::text AS src_mac_address, dst_ip::text AS dst_ip_address, dst_mac::text AS dst_mac_address, src_port, dst_port, NULLIF(app_protocol::text, '') AS app_protocol_name, ip_proto_raw, eth_type_raw, NULLIF(app_hostname::text, '') AS app_hostname, COUNT(*) AS packet_count, COALESCE(SUM(length), 0) AS byte_count, MIN(timestamp) AS first_seen, MAX(timestamp) AS last_seen FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND ( eth_type_raw = 2054 OR NULLIF(app_protocol::text, '') IN ('MDNS', 'SSDP', 'LLMNR', 'NBNS', 'DHCP', 'DHCPV6') OR src_port IN (67, 68, 137, 138, 5353, 5355, 1900, 546, 547) OR dst_port IN (67, 68, 137, 138, 5353, 5355, 1900, 546, 547) OR ip_proto_raw = 58 ) GROUP BY ingress_if, egress_if, src_ip::text, src_mac::text, dst_ip::text, dst_mac::text, src_port, dst_port, app_protocol_name, ip_proto_raw, eth_type_raw, app_hostname ORDER BY packet_count DESC, last_seen DESC LIMIT $2 """, since, limit, ) except Exception: logger.exception("DB discovery activity analysis failed") raise grouped: Dict[str, Dict[str, Any]] = {} for row in rows: record = dict(row) protocol_name = _analysis_protocol_name( record.get("app_protocol_name"), record.get("ip_proto_raw"), record.get("eth_type_raw"), ) ip_protocol_name = _analysis_ip_protocol_name(record.get("ip_proto_raw")) ethernet_protocol_name = _analysis_ethernet_protocol_name(record.get("eth_type_raw")) category = _classify_discovery_activity( str(protocol_name), ethernet_protocol_name, ip_protocol_name, record.get("src_port"), record.get("dst_port"), ) if category is None: continue key = "|".join( [ category, str(record.get("ingress_if") or ""), str(record.get("egress_if") or ""), str(record.get("src_ip_address") or ""), str(record.get("src_mac_address") or ""), str(record.get("dst_ip_address") or ""), str(record.get("dst_mac_address") or ""), str(record.get("src_port") or ""), str(record.get("dst_port") or ""), ] ) first_seen_raw = record.get("first_seen") last_seen_raw = record.get("last_seen") activity = grouped.get(key) if activity is None: activity = { "category": category, "protocol": str(protocol_name), "ingress_interface": record.get("ingress_if"), "egress_interface": record.get("egress_if"), "src_ip_address": record.get("src_ip_address"), "src_mac_address": record.get("src_mac_address"), "dst_ip_address": record.get("dst_ip_address"), "dst_mac_address": record.get("dst_mac_address"), "src_port": record.get("src_port"), "dst_port": record.get("dst_port"), "hostnames": set(), "packet_count": 0, "byte_count": 0, "first_seen": first_seen_raw.isoformat() if hasattr(first_seen_raw, "isoformat") else first_seen_raw, "last_seen": last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw, } grouped[key] = activity activity["packet_count"] += int(record.get("packet_count") or 0) activity["byte_count"] += int(record.get("byte_count") or 0) if record.get("app_hostname"): activity["hostnames"].add(str(record.get("app_hostname"))) last_seen = last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw if last_seen and ( activity.get("last_seen") in (None, "") or str(last_seen) > str(activity.get("last_seen")) ): activity["last_seen"] = last_seen activities = [] for activity in grouped.values(): activity["hostnames"] = sorted(activity["hostnames"]) activities.append(activity) return sorted( activities, key=lambda item: ( -int(item.get("packet_count") or 0), str(item.get("category") or ""), str(item.get("last_seen") or ""), ), ) async def analyze_anomalies( self, *, since: Optional[datetime] = None, limit: int = 50, ) -> Dict[str, List[Dict[str, Any]]]: """Compute lightweight anomaly candidates from observed traffic.""" if self._pool is None: await self.init_pool() try: async with self._pool.acquire() as conn: scan_rows = await conn.fetch( """ SELECT src_ip::text AS src_ip_address, src_mac::text AS src_mac_address, COUNT(*) AS packet_count, COUNT(DISTINCT COALESCE(dst_ip::text, '') || '|' || COALESCE(dst_mac::text, '')) AS target_host_count, COUNT(DISTINCT COALESCE(dst_port, -1)) FILTER (WHERE dst_port IS NOT NULL) AS target_port_count, MIN(timestamp) AS first_seen, MAX(timestamp) AS last_seen FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND (src_ip IS NOT NULL OR src_mac IS NOT NULL) AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) GROUP BY src_ip::text, src_mac::text HAVING COUNT(DISTINCT COALESCE(dst_ip::text, '') || '|' || COALESCE(dst_mac::text, '')) >= 5 OR COUNT(DISTINCT COALESCE(dst_port, -1)) FILTER (WHERE dst_port IS NOT NULL) >= 8 ORDER BY target_host_count DESC, target_port_count DESC, packet_count DESC, last_seen DESC LIMIT $2 """, since, limit, ) rare_service_rows = await conn.fetch( """ WITH service_counts AS ( SELECT dst_ip::text AS dst_ip_address, dst_mac::text AS dst_mac_address, dst_port, NULLIF(app_protocol::text, '') AS app_protocol_name, ip_proto_raw, eth_type_raw, COUNT(*) AS packet_count, COUNT(DISTINCT COALESCE(src_ip::text, '') || '|' || COALESCE(src_mac::text, '')) AS client_count, ARRAY_REMOVE(ARRAY_AGG(DISTINCT NULLIF(app_hostname::text, '')), NULL) AS hostnames, MAX(timestamp) AS last_seen FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND dst_port IS NOT NULL AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) GROUP BY dst_ip::text, dst_mac::text, dst_port, app_protocol_name, ip_proto_raw, eth_type_raw ) SELECT * FROM service_counts WHERE client_count <= 2 AND packet_count <= 20 ORDER BY packet_count ASC, client_count ASC, last_seen DESC LIMIT $2 """, since, limit, ) reset_rows = await conn.fetch( """ SELECT src_ip::text AS src_ip_address, src_mac::text AS src_mac_address, dst_ip::text AS dst_ip_address, dst_mac::text AS dst_mac_address, dst_port, COUNT(*) AS total_packets, SUM( CASE WHEN COALESCE(dpi_metadata -> 'tcp' ->> 'packet_type', '') IN ('RST', 'RST-ACK') OR COALESCE(dpi_metadata -> 'tcp' -> 'flag_names', '[]'::jsonb) ? 'RST' THEN 1 ELSE 0 END ) AS reset_count, MAX(timestamp) AS last_seen FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND ip_proto_raw = 6 AND (src_ip IS NOT NULL OR src_mac IS NOT NULL) AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) GROUP BY src_ip::text, src_mac::text, dst_ip::text, dst_mac::text, dst_port HAVING SUM( CASE WHEN COALESCE(dpi_metadata -> 'tcp' ->> 'packet_type', '') IN ('RST', 'RST-ACK') OR COALESCE(dpi_metadata -> 'tcp' -> 'flag_names', '[]'::jsonb) ? 'RST' THEN 1 ELSE 0 END ) >= 2 ORDER BY reset_count DESC, total_packets DESC, last_seen DESC LIMIT $2 """, since, limit, ) drop_rows = await conn.fetch( """ SELECT src_ip::text AS src_ip_address, src_mac::text AS src_mac_address, dst_ip::text AS dst_ip_address, dst_mac::text AS dst_mac_address, NULLIF(app_protocol::text, '') AS app_protocol_name, ip_proto_raw, eth_type_raw, COUNT(*) AS total_packets, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') = 'drop' THEN 1 ELSE 0 END) AS drop_count, SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') = 'reject' THEN 1 ELSE 0 END) AS reject_count, MAX(timestamp) AS last_seen FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND (src_ip IS NOT NULL OR src_mac IS NOT NULL) AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) GROUP BY src_ip::text, src_mac::text, dst_ip::text, dst_mac::text, app_protocol_name, ip_proto_raw, eth_type_raw HAVING SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') IN ('drop', 'reject') THEN 1 ELSE 0 END) >= 2 ORDER BY (SUM(CASE WHEN COALESCE(NULLIF(verdict::text, ''), 'unknown') IN ('drop', 'reject') THEN 1 ELSE 0 END)) DESC, last_seen DESC LIMIT $2 """, since, limit, ) beacon_rows = await conn.fetch( """ SELECT src_ip::text AS src_ip_address, src_mac::text AS src_mac_address, dst_ip::text AS dst_ip_address, dst_mac::text AS dst_mac_address, dst_port, NULLIF(app_protocol::text, '') AS app_protocol_name, ip_proto_raw, eth_type_raw, ARRAY_AGG(EXTRACT(EPOCH FROM timestamp) ORDER BY timestamp) AS observed_seconds, COUNT(*) AS packet_count, MIN(timestamp) AS first_seen, MAX(timestamp) AS last_seen FROM packets WHERE ($1::timestamptz IS NULL OR timestamp >= $1) AND (src_ip IS NOT NULL OR src_mac IS NOT NULL) AND (dst_ip IS NOT NULL OR dst_mac IS NOT NULL) GROUP BY src_ip::text, src_mac::text, dst_ip::text, dst_mac::text, dst_port, app_protocol_name, ip_proto_raw, eth_type_raw HAVING COUNT(*) >= 4 ORDER BY packet_count DESC, last_seen DESC LIMIT 1000 """, since, ) except Exception: logger.exception("DB anomaly analysis failed") raise scan_candidates: List[Dict[str, Any]] = [] for row in scan_rows: record = dict(row) first_seen_raw = record.get("first_seen") last_seen_raw = record.get("last_seen") scan_candidates.append( { "src_ip_address": record.get("src_ip_address"), "src_mac_address": record.get("src_mac_address"), "packet_count": int(record.get("packet_count") or 0), "target_host_count": int(record.get("target_host_count") or 0), "target_port_count": int(record.get("target_port_count") or 0), "first_seen": first_seen_raw.isoformat() if hasattr(first_seen_raw, "isoformat") else first_seen_raw, "last_seen": last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw, } ) rare_services: List[Dict[str, Any]] = [] for row in rare_service_rows: record = dict(row) protocol_name = _analysis_protocol_name( record.get("app_protocol_name"), record.get("ip_proto_raw"), record.get("eth_type_raw"), ) last_seen_raw = record.get("last_seen") rare_services.append( { "dst_ip_address": record.get("dst_ip_address"), "dst_mac_address": record.get("dst_mac_address"), "dst_port": record.get("dst_port"), "protocol": str(protocol_name), "packet_count": int(record.get("packet_count") or 0), "client_count": int(record.get("client_count") or 0), "hostnames": list(record.get("hostnames") or []), "last_seen": last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw, } ) reset_heavy_paths: List[Dict[str, Any]] = [] for row in reset_rows: record = dict(row) last_seen_raw = record.get("last_seen") total_packets = int(record.get("total_packets") or 0) reset_count = int(record.get("reset_count") or 0) reset_heavy_paths.append( { "src_ip_address": record.get("src_ip_address"), "src_mac_address": record.get("src_mac_address"), "dst_ip_address": record.get("dst_ip_address"), "dst_mac_address": record.get("dst_mac_address"), "dst_port": record.get("dst_port"), "total_packets": total_packets, "reset_count": reset_count, "reset_ratio": round(_safe_ratio(reset_count, total_packets), 3), "last_seen": last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw, } ) drop_heavy_paths: List[Dict[str, Any]] = [] for row in drop_rows: record = dict(row) protocol_name = _analysis_protocol_name( record.get("app_protocol_name"), record.get("ip_proto_raw"), record.get("eth_type_raw"), ) last_seen_raw = record.get("last_seen") total_packets = int(record.get("total_packets") or 0) drop_count = int(record.get("drop_count") or 0) reject_count = int(record.get("reject_count") or 0) failure_count = drop_count + reject_count drop_heavy_paths.append( { "src_ip_address": record.get("src_ip_address"), "src_mac_address": record.get("src_mac_address"), "dst_ip_address": record.get("dst_ip_address"), "dst_mac_address": record.get("dst_mac_address"), "protocol": str(protocol_name), "total_packets": total_packets, "drop_count": drop_count, "reject_count": reject_count, "failure_ratio": round(_safe_ratio(failure_count, total_packets), 3), "last_seen": last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw, } ) beacon_candidates: List[Dict[str, Any]] = [] for row in beacon_rows: record = dict(row) observed_seconds = [float(value) for value in (record.get("observed_seconds") or []) if value is not None] if len(observed_seconds) < 4: continue intervals = [ observed_seconds[index] - observed_seconds[index - 1] for index in range(1, len(observed_seconds)) if observed_seconds[index] - observed_seconds[index - 1] > 0 ] if len(intervals) < 3: continue avg_interval = sum(intervals) / len(intervals) if avg_interval < 1 or avg_interval > 3600: continue if len(intervals) == 1: jitter_ratio = 0.0 else: jitter_ratio = _safe_ratio(statistics.pstdev(intervals), avg_interval) if math.isnan(jitter_ratio) or jitter_ratio > 0.25: continue protocol_name = _analysis_protocol_name( record.get("app_protocol_name"), record.get("ip_proto_raw"), record.get("eth_type_raw"), ) first_seen_raw = record.get("first_seen") last_seen_raw = record.get("last_seen") beacon_candidates.append( { "src_ip_address": record.get("src_ip_address"), "src_mac_address": record.get("src_mac_address"), "dst_ip_address": record.get("dst_ip_address"), "dst_mac_address": record.get("dst_mac_address"), "dst_port": record.get("dst_port"), "protocol": str(protocol_name), "packet_count": int(record.get("packet_count") or 0), "avg_interval_seconds": round(avg_interval, 2), "jitter_ratio": round(jitter_ratio, 3), "interval_samples": [round(interval, 2) for interval in intervals[:16]], "first_seen": first_seen_raw.isoformat() if hasattr(first_seen_raw, "isoformat") else first_seen_raw, "last_seen": last_seen_raw.isoformat() if hasattr(last_seen_raw, "isoformat") else last_seen_raw, } ) beacon_candidates = sorted( beacon_candidates, key=lambda item: ( item.get("jitter_ratio", 1.0), -int(item.get("packet_count") or 0), str(item.get("last_seen") or ""), ), )[:limit] return { "scan_candidates": scan_candidates, "beacon_candidates": beacon_candidates, "rare_services": rare_services, "reset_heavy_paths": reset_heavy_paths, "drop_heavy_paths": drop_heavy_paths, } async def clear_all_packets(self, reset_identity: bool = True) -> bool: """Truncate the packet table and optionally reset identity counters.""" if self._pool is None: await self.init_pool() restart_clause = "RESTART IDENTITY" if reset_identity else "" query = f"TRUNCATE TABLE packets {restart_clause};" try: async with self._pool.acquire() as conn: await conn.execute(query) logger.info("Successfully cleared all packets from the database (reset_id=%s)", reset_identity) return True except Exception: logger.exception("Failed to clear packets table") return False