#!/usr/bin/env python3 """Network intelligence — beacon detection, communication graph analysis, existing C2/malware detection, and service dependency mapping. Analyzes traffic patterns to identify beaconing connections, central/isolated nodes, known-bad indicators, and service dependencies. Feeds anomaly baselines to traffic_mimicry. """ import json import logging import math import os import sqlite3 import threading import time from collections import defaultdict from typing import Optional from modules.base import BaseModule logger = logging.getLogger("sensor.intel.net_intel") # Known malicious infrastructure patterns _KNOWN_BAD_PORTS = { 4444, 5555, 1234, 31337, 8888, 6666, 6667, 6668, 6669, # common RAT/IRC 4443, 8443, 2222, # common C2 alt ports } # DNS tunneling detection thresholds _DNS_TUNNEL_MIN_SUBDOMAIN_LEN = 30 # base64-encoded data tends to be long _DNS_TUNNEL_QUERY_RATE = 50 # queries/min to a single domain = suspicious # Beacon detection defaults _BEACON_INTERVAL_RATIO = 0.15 # stddev/mean < 0.15 = likely beacon _BEACON_MIN_CONNECTIONS = 10 # need enough samples to be meaningful class NetIntel(BaseModule): """Analyze network traffic patterns for beacons, C2, anomalies, and dependencies.""" name = "net_intel" module_type = "intel" priority = 150 requires_root = False def __init__(self, bus, state, config, engine=None): super().__init__(bus, state, config, engine) self._lock = threading.Lock() self._analysis_thread: Optional[threading.Thread] = None # Connection tracking: (src, dst, port) -> [timestamps] self._conn_times: dict[tuple, list] = defaultdict(list) # Communication graph: ip -> {peers: set, bytes_in, bytes_out} self._comm_graph: dict[str, dict] = defaultdict( lambda: {"peers": set(), "bytes_in": 0, "bytes_out": 0, "conn_count": 0} ) # DNS query tracking: domain -> [timestamps] self._dns_queries: dict[str, list] = defaultdict(list) # Detected beacons self._beacons: list[dict] = [] # Service dependency map: service_ip:port -> [client_ips] self._service_deps: dict[str, set] = defaultdict(set) # Anomalies / suspicious findings self._findings: list[dict] = [] # ------------------------------------------------------------------ # BaseModule interface # ------------------------------------------------------------------ def start(self) -> None: if self._running: return self.bus.subscribe(self._on_host_discovered, "HOST_DISCOVERED") self.bus.subscribe(self._on_beacon_detected, "BEACON_DETECTED") self._running = True self._pid = os.getpid() self._start_time = time.time() # Periodic analysis thread self._analysis_thread = threading.Thread( target=self._analysis_loop, daemon=True, name="sensor-netintel-analysis", ) self._analysis_thread.start() self.state.set_module_status(self.name, "running", pid=self._pid) logger.info("NetIntel started") def stop(self) -> None: if not self._running: return self._running = False self.bus.unsubscribe(self._on_host_discovered, "HOST_DISCOVERED") self.bus.unsubscribe(self._on_beacon_detected, "BEACON_DETECTED") # Persist findings self._save_findings() self.state.set_module_status(self.name, "stopped") logger.info( "NetIntel stopped — %d beacons, %d findings", len(self._beacons), len(self._findings), ) def status(self) -> dict: return { "running": self._running, "pid": self._pid, "uptime": time.time() - self._start_time if self._start_time else 0, "tracked_flows": len(self._conn_times), "graph_nodes": len(self._comm_graph), "beacons_detected": len(self._beacons), "findings": len(self._findings), } def configure(self, config: dict) -> None: self.config.update(config) # ------------------------------------------------------------------ # Data ingestion # ------------------------------------------------------------------ def ingest_connection(self, src_ip: str, dst_ip: str, dst_port: int, protocol: str = "tcp", bytes_sent: int = 0, bytes_recv: int = 0, timestamp: float = None) -> None: """Ingest a connection record for analysis. Called by other modules (dns_logger, packet analyzers) or directly. """ ts = timestamp or time.time() with self._lock: # Connection timing for beacon detection flow_key = (src_ip, dst_ip, dst_port) self._conn_times[flow_key].append(ts) # Communication graph self._comm_graph[src_ip]["peers"].add(dst_ip) self._comm_graph[src_ip]["bytes_out"] += bytes_sent self._comm_graph[src_ip]["conn_count"] += 1 self._comm_graph[dst_ip]["peers"].add(src_ip) self._comm_graph[dst_ip]["bytes_in"] += bytes_recv self._comm_graph[dst_ip]["conn_count"] += 1 # Service dependency service_key = f"{dst_ip}:{dst_port}" self._service_deps[service_key].add(src_ip) # Known-bad port check if dst_port in _KNOWN_BAD_PORTS: self._add_finding( "suspicious_port", f"{src_ip} -> {dst_ip}:{dst_port} ({protocol})", severity="warning", details={"src": src_ip, "dst": dst_ip, "port": dst_port}, ) def ingest_dns_query(self, client_ip: str, query_name: str, query_type: str = "A", timestamp: float = None) -> None: """Ingest a DNS query for tunneling and pattern analysis.""" ts = timestamp or time.time() with self._lock: self._dns_queries[query_name].append(ts) # Check for DNS tunneling indicators parts = query_name.split(".") if len(parts) >= 3: # Long subdomain labels suggest encoding subdomain = ".".join(parts[:-2]) if len(subdomain) >= _DNS_TUNNEL_MIN_SUBDOMAIN_LEN: base_domain = ".".join(parts[-2:]) self._add_finding( "dns_tunnel_suspect", f"Long subdomain to {base_domain} from {client_ip}: " f"{len(subdomain)} chars", severity="warning", details={ "client": client_ip, "base_domain": base_domain, "subdomain_len": len(subdomain), "query_type": query_type, }, ) def _on_host_discovered(self, event) -> None: """Track new hosts for the communication graph.""" p = event.payload ip = p.get("ip", "") if ip and ip not in self._comm_graph: with self._lock: self._comm_graph[ip]["peers"] = set() def _on_beacon_detected(self, event) -> None: """Record externally detected beacons.""" p = event.payload with self._lock: self._beacons.append({ "src": p.get("src_ip", ""), "dst": p.get("dst_ip", ""), "port": p.get("port", 0), "interval": p.get("interval", 0), "jitter": p.get("jitter", 0), "detected_at": time.time(), "source": event.source_module, }) # ------------------------------------------------------------------ # Analysis engine # ------------------------------------------------------------------ def _analysis_loop(self) -> None: """Periodic analysis of accumulated traffic data.""" interval = self.config.get("analysis_interval", 120) while self._running: time.sleep(interval) try: self._detect_beacons() self._detect_dns_tunneling() self._analyze_communication_graph() self._save_findings() except Exception: logger.exception("NetIntel analysis cycle failed") def _detect_beacons(self) -> None: """Detect beaconing connections by analyzing inter-connection intervals. A beacon has a regular interval: stddev/mean ratio < threshold. """ threshold = self.config.get( "beacon_ratio_threshold", _BEACON_INTERVAL_RATIO ) min_samples = self.config.get( "beacon_min_connections", _BEACON_MIN_CONNECTIONS ) with self._lock: for flow_key, timestamps in self._conn_times.items(): if len(timestamps) < min_samples: continue # Calculate inter-arrival intervals sorted_ts = sorted(timestamps) intervals = [ sorted_ts[i + 1] - sorted_ts[i] for i in range(len(sorted_ts) - 1) ] if not intervals: continue mean_interval = sum(intervals) / len(intervals) if mean_interval <= 0: continue variance = sum((x - mean_interval) ** 2 for x in intervals) / len(intervals) stddev = math.sqrt(variance) ratio = stddev / mean_interval if ratio < threshold: src, dst, port = flow_key beacon_info = { "src": src, "dst": dst, "port": port, "mean_interval": round(mean_interval, 2), "stddev": round(stddev, 2), "ratio": round(ratio, 4), "sample_count": len(timestamps), "detected_at": time.time(), } # Avoid duplicate beacon alerts is_dup = any( b["src"] == src and b["dst"] == dst and b["port"] == port for b in self._beacons ) if not is_dup: self._beacons.append(beacon_info) self.bus.emit("BEACON_DETECTED", { "src_ip": src, "dst_ip": dst, "port": port, "interval": mean_interval, "jitter": stddev, "ratio": ratio, }, source_module=self.name) logger.warning( "BEACON: %s -> %s:%d every %.1fs " "(ratio=%.4f, n=%d)", src, dst, port, mean_interval, ratio, len(timestamps), ) def _detect_dns_tunneling(self) -> None: """Detect potential DNS tunneling by query rate to specific domains.""" now = time.time() window = 60.0 # 1-minute window with self._lock: # Aggregate by base domain (last 2 labels) domain_rates: dict[str, int] = defaultdict(int) for query_name, timestamps in self._dns_queries.items(): parts = query_name.split(".") if len(parts) >= 2: base = ".".join(parts[-2:]) recent = sum(1 for t in timestamps if now - t < window) domain_rates[base] += recent for domain, rate in domain_rates.items(): if rate >= _DNS_TUNNEL_QUERY_RATE: self._add_finding( "dns_tunnel_rate", f"High DNS query rate to {domain}: {rate}/min", severity="critical", details={"domain": domain, "rate_per_min": rate}, ) def _analyze_communication_graph(self) -> None: """Identify central nodes, isolated hosts, and unusual patterns.""" with self._lock: if len(self._comm_graph) < 3: return # Find most-connected nodes (potential central servers / DCs) by_peers = sorted( self._comm_graph.items(), key=lambda x: len(x[1]["peers"]), reverse=True, ) total_nodes = len(self._comm_graph) central_threshold = max(total_nodes * 0.3, 5) for ip, data in by_peers[:10]: peer_count = len(data["peers"]) if peer_count >= central_threshold: self.state.set( self.name, f"central_node_{ip}", json.dumps({ "ip": ip, "peer_count": peer_count, "bytes_in": data["bytes_in"], "bytes_out": data["bytes_out"], }), ) # Find isolated hosts (single or no peers — unusual) for ip, data in self._comm_graph.items(): if len(data["peers"]) <= 1 and data["conn_count"] > 20: self._add_finding( "isolated_talker", f"{ip} has {data['conn_count']} connections " f"but only {len(data['peers'])} peer(s)", severity="info", details={"ip": ip, "peers": list(data["peers"])}, ) # ------------------------------------------------------------------ # Findings management # ------------------------------------------------------------------ def _add_finding(self, finding_type: str, description: str, severity: str = "info", details: dict = None) -> None: """Record a finding, deduplicating by type + description.""" # Simple dedup: don't re-add identical findings for f in self._findings: if f["type"] == finding_type and f["description"] == description: f["last_seen"] = time.time() f["count"] = f.get("count", 1) + 1 return self._findings.append({ "type": finding_type, "description": description, "severity": severity, "details": details or {}, "first_seen": time.time(), "last_seen": time.time(), "count": 1, }) def _save_findings(self) -> None: """Persist findings and beacons to state.""" with self._lock: beacons_ser = list(self._beacons) findings_ser = list(self._findings) # Serialize service deps deps_ser = {k: list(v) for k, v in self._service_deps.items()} self.state.set(self.name, "beacons", json.dumps(beacons_ser)) self.state.set(self.name, "findings", json.dumps(findings_ser)) self.state.set(self.name, "service_deps", json.dumps(deps_ser)) # ------------------------------------------------------------------ # Public query interface # ------------------------------------------------------------------ def get_beacons(self) -> list: """Return all detected beacons.""" with self._lock: return list(self._beacons) def get_findings(self, severity: str = None) -> list: """Return findings, optionally filtered by severity.""" with self._lock: if severity: return [f for f in self._findings if f["severity"] == severity] return list(self._findings) def get_central_nodes(self, min_peers: int = 5) -> list: """Return the most-connected nodes in the communication graph.""" with self._lock: results = [] for ip, data in self._comm_graph.items(): peer_count = len(data["peers"]) if peer_count >= min_peers: results.append({ "ip": ip, "peer_count": peer_count, "bytes_in": data["bytes_in"], "bytes_out": data["bytes_out"], "conn_count": data["conn_count"], }) results.sort(key=lambda x: x["peer_count"], reverse=True) return results def get_service_dependencies(self) -> dict: """Return service dependency map: service -> [dependent clients].""" with self._lock: return {k: list(v) for k, v in self._service_deps.items()} def get_anomaly_baseline(self) -> dict: """Return traffic pattern baseline for traffic_mimicry module. Provides: - Typical connection intervals per flow - Byte distribution per protocol - Peak/quiet hour patterns """ with self._lock: flow_intervals = {} for flow_key, timestamps in self._conn_times.items(): if len(timestamps) < 3: continue sorted_ts = sorted(timestamps) intervals = [sorted_ts[i + 1] - sorted_ts[i] for i in range(len(sorted_ts) - 1)] mean = sum(intervals) / len(intervals) flow_intervals[f"{flow_key[0]}->{flow_key[1]}:{flow_key[2]}"] = { "mean_interval": round(mean, 2), "sample_count": len(timestamps), } # Hour-of-day connection distribution hourly = defaultdict(int) for timestamps in self._conn_times.values(): for ts in timestamps: hour = time.localtime(ts).tm_hour hourly[hour] += 1 return { "flow_intervals": flow_intervals, "hourly_distribution": dict(hourly), "total_flows": len(self._conn_times), "total_nodes": len(self._comm_graph), }