diff --git a/.env.example b/.env.example index acd6414..b0738b4 100644 --- a/.env.example +++ b/.env.example @@ -54,6 +54,13 @@ NETBOT_LIVE_RING_DEFAULT_QUERY_LIMIT=250 NETBOT_LIVE_RING_MAX_QUERY_LIMIT=2000 NETBOT_LIVE_RING_TTL_SECONDS=0 +# Local, metadata-only destination attribution. No external lookup is performed. +NETBOT_SERVICE_ATTRIBUTION_ENABLED=true +NETBOT_SERVICE_ATTRIBUTION_REGISTRY=backend/app/data/service_fingerprints.json +NETBOT_SERVICE_ATTRIBUTION_DNS_WINDOW_SEC=300 +NETBOT_SERVICE_ATTRIBUTION_MAX_REASONS=8 +NETBOT_SERVICE_ATTRIBUTION_UNKNOWN_RATE_WARN=0.75 + # Bounded batch persistence. Counters and health are exposed in Ops Snapshot. NETBOT_PERSISTENCE_BATCH_ENABLED=true NETBOT_PERSISTENCE_PACKET_BATCH_SIZE=500 diff --git a/README.md b/README.md index bdb3cca..bcde07a 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,7 @@ or PCAP artifacts. | Conversation Timeline | Active | Protocol, alert, destination, and lifecycle events are correlated per flow. | Timeline summaries and metadata pass through central redaction. | | Flow Risk Scoring | Active | Explainable `0..100` scoring covers alerts, volume, DNS failures, unusual protocols/ports, and destinations. | Risk is an investigation aid, not an automated verdict. | | Offline PCAP Flow Summary | Active | Offline analysis returns flows, conversations, protocol summaries, timelines, and risk distribution. | Existing API response fields remain compatible; output stays redacted. | +| Service Attribution / Destination Intelligence | Active MVP | DNS, visible TLS SNI, HTTP Host, recent DNS-answer correlation, ASN metadata, and a local fingerprint registry produce confidence-scored destination labels with evidence. Browser/container identity and CDN infrastructure are handled conservatively; encrypted traffic without evidence remains Unknown. | Local metadata-only inference; no external lookup, TLS decryption, browser history, cookies, credentials, or payload inspection. | | Deep Packet Inspection | Active MVP | Inspect renders a searchable layer tree, safe bytes view, streams, and Expert Info. | No TLS decryption; visible metadata and ASCII previews are centrally redacted. | | Display Filters | Active MVP | Safe packet and flow filter parser covers text, equality, range, and boolean operators. | Filters run on redacted metadata and never use Python `eval`. | | Offline PCAP Deep Analysis | Active MVP | Offline results include packet details, Expert Info, and stream summaries. | Previous API fields remain compatible; raw secrets are not exposed. | @@ -256,6 +257,7 @@ and surfaced in Ops Snapshot metrics. - [Performance Pipeline](docs/PERFORMANCE_PIPELINE.md) - [Performance Validation](docs/PERFORMANCE_VALIDATION.md) - [Flow Analysis And Protocol Intelligence](docs/FLOW_ANALYSIS.md) +- [Service Attribution And Destination Intelligence](docs/SERVICE_ATTRIBUTION.md) - [Deep Packet Inspection](docs/DEEP_PACKET_INSPECTION.md) - [Agent Operational QA Checklist](docs/AGENT_QA_CHECKLIST.md) - [Deployment Overview](docs/DEPLOYMENT_OVERVIEW.md) @@ -484,9 +486,9 @@ powershell -ExecutionPolicy Bypass -File .\packaging\windows\build.ps1 - Complete the performance foundation with a flow-aware worker pool, live ring buffer, and benchmark/soak validation. -- Add conservative Service Attribution / Destination Intelligence using DNS, - TLS SNI, HTTP Host, QUIC-visible metadata, ASN, and local fingerprints. Low - confidence remains `Unknown`; no TLS decryption or credential collection. +- Validate and tune the implemented metadata-only Service Attribution registry + against authorized deployment traffic while preserving conservative Unknown + results for weak or hidden evidence. - Build a read-only Incident / Correlation Engine after attribution quality is validated, then consider a strictly read-only AI Analyst. - Versioned SQLite schema migrations and longer-lived deployment operations. diff --git a/backend/app/data/service_fingerprints.json b/backend/app/data/service_fingerprints.json new file mode 100644 index 0000000..648d3bb --- /dev/null +++ b/backend/app/data/service_fingerprints.json @@ -0,0 +1,28 @@ +{ + "version": 1, + "services": [ + {"service_name":"YouTube","category":"Video Streaming","domain_patterns":["youtube.com","*.youtube.com","googlevideo.com","*.googlevideo.com","ytimg.com","*.ytimg.com","youtu.be"],"sni_patterns":["youtube.com","*.youtube.com","googlevideo.com","*.googlevideo.com","ytimg.com","*.ytimg.com"],"http_host_patterns":["youtube.com","*.youtube.com","googlevideo.com","*.googlevideo.com","ytimg.com","*.ytimg.com"],"asn_org_patterns":["google"],"cdn":true,"risk_notes":"Known video streaming service; review only for policy or volume anomalies."}, + {"service_name":"Netflix","category":"Video Streaming","domain_patterns":["netflix.com","*.netflix.com","nflxvideo.net","*.nflxvideo.net","nflximg.net","*.nflximg.net"],"asn_org_patterns":["netflix"],"cdn":true}, + {"service_name":"Twitch","category":"Video Streaming","domain_patterns":["twitch.tv","*.twitch.tv","ttvnw.net","*.ttvnw.net"],"asn_org_patterns":["twitch"],"cdn":true}, + {"service_name":"Telegram Web","category":"Messaging","domain_patterns":["web.telegram.org","telegram.org","*.telegram.org","t.me","*.t.me"],"asn_org_patterns":["telegram"]}, + {"service_name":"WhatsApp Web","category":"Messaging","domain_patterns":["web.whatsapp.com","whatsapp.com","*.whatsapp.com","whatsapp.net","*.whatsapp.net"],"asn_org_patterns":["meta platforms","facebook"]}, + {"service_name":"Discord","category":"Messaging","domain_patterns":["discord.com","*.discord.com","discord.gg","*.discord.gg","discordapp.com","*.discordapp.com","discordapp.net","*.discordapp.net"],"asn_org_patterns":["discord"],"cdn":true}, + {"service_name":"GitHub","category":"Developer Platform","domain_patterns":["github.com","*.github.com","githubusercontent.com","*.githubusercontent.com","githubassets.com","*.githubassets.com"],"asn_org_patterns":["github"]}, + {"service_name":"GitLab","category":"Developer Platform","domain_patterns":["gitlab.com","*.gitlab.com"],"asn_org_patterns":["gitlab"]}, + {"service_name":"Docker Hub","category":"Developer Platform","domain_patterns":["docker.com","*.docker.com","docker.io","*.docker.io"],"asn_org_patterns":["docker"]}, + {"service_name":"npm Registry","category":"Developer Platform","domain_patterns":["npmjs.com","*.npmjs.com","npmjs.org","*.npmjs.org"],"asn_org_patterns":["npm"]}, + {"service_name":"PyPI","category":"Developer Platform","domain_patterns":["pypi.org","*.pypi.org","pythonhosted.org","*.pythonhosted.org"],"asn_org_patterns":["python software foundation"]}, + {"service_name":"Google","category":"Search / Cloud","domain_patterns":["google.com","*.google.com","gstatic.com","*.gstatic.com","1e100.net","*.1e100.net"],"asn_org_patterns":["google"]}, + {"service_name":"Microsoft","category":"Search / Cloud","domain_patterns":["microsoft.com","*.microsoft.com","live.com","*.live.com","office.com","*.office.com"],"asn_org_patterns":["microsoft"]}, + {"service_name":"AWS","category":"Cloud","domain_patterns":["amazonaws.com","*.amazonaws.com","aws.amazon.com"],"asn_org_patterns":["amazon","aws"],"cdn":true}, + {"service_name":"Azure","category":"Cloud","domain_patterns":["azure.com","*.azure.com","azurewebsites.net","*.azurewebsites.net"],"asn_org_patterns":["microsoft"]}, + {"service_name":"Instagram","category":"Social","domain_patterns":["instagram.com","*.instagram.com","cdninstagram.com","*.cdninstagram.com"],"asn_org_patterns":["meta platforms","facebook"],"cdn":true}, + {"service_name":"X / Twitter","category":"Social","domain_patterns":["x.com","*.x.com","twitter.com","*.twitter.com","twimg.com","*.twimg.com"],"asn_org_patterns":["twitter"]}, + {"service_name":"TikTok","category":"Social","domain_patterns":["tiktok.com","*.tiktok.com","tiktokcdn.com","*.tiktokcdn.com","byteoversea.com","*.byteoversea.com"],"asn_org_patterns":["bytedance"],"cdn":true}, + {"service_name":"Cloudflare CDN","category":"CDN","domain_patterns":["cloudflare.com","*.cloudflare.com","cloudflare.net","*.cloudflare.net"],"asn_org_patterns":["cloudflare"],"cdn":true}, + {"service_name":"Akamai","category":"CDN","domain_patterns":["akamaized.net","*.akamaized.net","akamai.net","*.akamai.net","akamaiedge.net","*.akamaiedge.net"],"asn_org_patterns":["akamai"],"cdn":true}, + {"service_name":"Fastly","category":"CDN","domain_patterns":["fastly.net","*.fastly.net","fastlylb.net","*.fastlylb.net"],"asn_org_patterns":["fastly"],"cdn":true}, + {"service_name":"Google CDN","category":"CDN","domain_patterns":["googleusercontent.com","*.googleusercontent.com","ggpht.com","*.ggpht.com"],"asn_org_patterns":["google"],"cdn":true}, + {"service_name":"Amazon CloudFront","category":"CDN","domain_patterns":["cloudfront.net","*.cloudfront.net"],"asn_org_patterns":["amazon"],"cdn":true} + ] +} diff --git a/backend/app/main.py b/backend/app/main.py index c0f4121..f0ad486 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -119,6 +119,7 @@ def _observability_snapshot() -> dict[str, Any]: "packet_queue": sniffer_service.packet_queue_stats(), "flow_worker_pool": sniffer_service.flow_worker_pool_stats(), "live_ring_buffer": sniffer_service.live_ring_buffer_stats(), + "service_attribution": sniffer_service.service_attribution_stats(), "persistence": sniffer_service.persistence_stats(), "auto_block": sniffer_service.auto_block_stats(), } diff --git a/backend/app/services/monitoring_service.py b/backend/app/services/monitoring_service.py index 0d6dabc..b8e9411 100644 --- a/backend/app/services/monitoring_service.py +++ b/backend/app/services/monitoring_service.py @@ -187,6 +187,7 @@ def build_monitoring_metrics( packet_queue = dict(observability.get("packet_queue") or {}) flow_worker_pool = dict(observability.get("flow_worker_pool") or {}) live_ring_buffer = dict(observability.get("live_ring_buffer") or {}) + service_attribution = dict(observability.get("service_attribution") or {}) event_aggregator = dict(observability.get("event_aggregator") or {}) websocket = dict(observability.get("websocket") or {}) persistence = dict(observability.get("persistence") or {}) @@ -288,6 +289,18 @@ def build_monitoring_metrics( pressure_reasons.extend( reason for reason in live_ring_reasons if reason not in pressure_reasons ) + service_attribution_reasons = [ + reason + for value in service_attribution.get("pressure_reasons") or [] + if (reason := str(value)).startswith("service_attribution_") + and len(reason) <= 80 + and reason.replace("_", "").isalnum() + ] + pressure_reasons.extend( + reason + for reason in service_attribution_reasons + if reason not in pressure_reasons + ) pressure_reasons.extend( reason for reason in persistence_reasons if reason not in pressure_reasons ) @@ -343,6 +356,7 @@ def build_monitoring_metrics( or websocket_drops >= _int(event_aggregator.get("client_queue_max") or 1000) or flow_worker_pool.get("health") == "critical" or live_ring_buffer.get("health") == "critical" + or service_attribution.get("health") == "critical" ): health = "critical" @@ -588,6 +602,41 @@ def build_monitoring_metrics( }, "pressure_reasons": live_ring_reasons, }, + "service_attribution": { + "enabled": bool(service_attribution.get("enabled", True)), + "health": str(service_attribution.get("health") or "healthy"), + "registry_size": _int(service_attribution.get("registry_size")), + "attributed_flows_total": _int( + service_attribution.get("attributed_flows_total") + ), + "unknown_flows_total": _int(service_attribution.get("unknown_flows_total")), + "high_confidence_total": _int( + service_attribution.get("high_confidence_total") + ), + "medium_confidence_total": _int( + service_attribution.get("medium_confidence_total") + ), + "low_confidence_total": _int( + service_attribution.get("low_confidence_total") + ), + "encrypted_unknown_total": _int( + service_attribution.get("encrypted_unknown_total") + ), + "cdn_only_total": _int(service_attribution.get("cdn_only_total")), + "attribution_errors_total": _int( + service_attribution.get("attribution_errors_total") + ), + "avg_attribution_latency_ms": _number( + service_attribution.get("avg_attribution_latency_ms") + ), + "p95_attribution_latency_ms": _number( + service_attribution.get("p95_attribution_latency_ms") + ), + "last_error": _safe_persistence_error( + service_attribution.get("last_error") + ), + "pressure_reasons": service_attribution_reasons, + }, "persistence": { "enabled": bool( persistence.get("persistence_enabled", persistence.get("enabled")) diff --git a/backend/app/services/service_attribution.py b/backend/app/services/service_attribution.py index 37672b2..9f1ce94 100644 --- a/backend/app/services/service_attribution.py +++ b/backend/app/services/service_attribution.py @@ -1,131 +1,726 @@ from __future__ import annotations +import ipaddress +import json +import os +import threading +import time +from collections import OrderedDict, deque +from dataclasses import asdict, dataclass +from datetime import datetime, timezone from fnmatch import fnmatch +from pathlib import Path from typing import Any from core.privacy_redaction import redact_sensitive_data, redact_sensitive_text -SERVICE_PATTERNS = ( - ( - "YouTube", - "Video Streaming", - ( - "youtube.com", - "*.youtube.com", - "googlevideo.com", - "*.googlevideo.com", - "youtu.be", - ), - ), - ( - "Google", - "Search / Cloud", - ( - "google.com", - "*.google.com", - "gstatic.com", - "*.gstatic.com", - "1e100.net", - "*.1e100.net", - ), - ), - ( - "GitHub", - "Developer", - ( - "github.com", - "*.github.com", - "githubusercontent.com", - "*.githubusercontent.com", - ), - ), - ("Telegram", "Messaging", ("telegram.org", "*.telegram.org", "t.me", "*.t.me")), - ("Cloudflare", "CDN / Security", ("cloudflare.com", "*.cloudflare.com")), +DEFAULT_REGISTRY_PATH = ( + Path(__file__).resolve().parents[1] / "data" / "service_fingerprints.json" ) +DEFAULT_DNS_CACHE_MAX = 10_000 +DEFAULT_FLOW_METRIC_CACHE_MAX = 100_000 +BROWSER_OR_CONTAINER_PROCESSES = { + "brave.exe", + "chrome.exe", + "discord.exe", + "electron.exe", + "firefox.exe", + "msedge.exe", + "opera.exe", + "slack.exe", + "telegram.exe", +} -ORG_FALLBACKS = ( - ("google", "Google Services", "Search / Cloud"), - ("cloudflare", "Cloudflare Network", "CDN / Security"), - ("github", "GitHub", "Developer"), - ("microsoft", "Microsoft Services", "Cloud"), - ("amazon", "Amazon / AWS", "Cloud / CDN"), -) + +def _env_bool(name: str, default: bool) -> bool: + value = os.environ.get(name) + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "on"} + + +def _safe_int(value: Any, default: int, minimum: int, maximum: int) -> int: + try: + parsed = int(value) + except (TypeError, ValueError): + return default + return max(minimum, min(parsed, maximum)) + + +def _timestamp(value: Any) -> float: + if isinstance(value, (int, float)): + return float(value) + try: + return datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp() + except (TypeError, ValueError): + return time.time() + + +def _domain(value: Any) -> str: + normalized = str(value or "").strip().lower().rstrip(".") + if ":" in normalized and normalized.count(":") == 1: + normalized = normalized.split(":", 1)[0] + if ( + not normalized + or len(normalized) > 253 + or "/" in normalized + or "@" in normalized + ): + return "" + return redact_sensitive_text(normalized) + + +def _safe_ip(value: Any) -> str: + candidate = str(value or "").strip() + try: + return str(ipaddress.ip_address(candidate)) + except ValueError: + return "" + + +def _percentile(values: deque[float], percentile: float) -> float: + ordered = sorted(values) + if not ordered: + return 0.0 + index = min(len(ordered) - 1, int((len(ordered) - 1) * percentile)) + return round(ordered[index], 3) + + +@dataclass(frozen=True) +class ServiceFingerprint: + service_name: str + category: str + domain_patterns: tuple[str, ...] + sni_patterns: tuple[str, ...] + http_host_patterns: tuple[str, ...] + asn_org_patterns: tuple[str, ...] + cdn: bool = False + risk_notes: str = "" + + @classmethod + def from_dict(cls, value: dict[str, Any]) -> "ServiceFingerprint": + service_name = str(value.get("service_name") or "").strip() + category = str(value.get("category") or "").strip() + domain_patterns = _patterns(value.get("domain_patterns")) + if not service_name or not category or not domain_patterns: + raise ValueError("InvalidServiceFingerprint") + return cls( + service_name=service_name[:100], + category=category[:100], + domain_patterns=domain_patterns, + sni_patterns=_patterns(value.get("sni_patterns")) or domain_patterns, + http_host_patterns=_patterns(value.get("http_host_patterns")) + or domain_patterns, + asn_org_patterns=tuple( + str(item).strip().lower()[:100] + for item in value.get("asn_org_patterns") or [] + if str(item).strip() + ), + cdn=bool(value.get("cdn")), + risk_notes=redact_sensitive_text(str(value.get("risk_notes") or ""))[:500], + ) -def _domain(packet: dict[str, Any]) -> tuple[str, str]: - candidates = ( - ("tls_sni", packet.get("tls_sni") or packet.get("sni")), - ("http_host", packet.get("http_host")), - ("dns", packet.get("dns_qname")), +def _patterns(value: Any) -> tuple[str, ...]: + return tuple( + pattern + for item in (value if isinstance(value, list) else []) + if (pattern := _domain(item)) ) - for source, value in candidates: - normalized = str(value or "").strip().lower().rstrip(".") - if normalized: - return normalized, source - return "", "" -def attribute_service(packet: dict[str, Any]) -> dict[str, Any]: - domain, source = _domain(packet) - application = str(packet.get("process_name") or "").strip() - org = str(packet.get("org") or "").strip() - encrypted = int(packet.get("sport") or packet.get("dport") or 0) == 443 or str( - packet.get("app_protocol") or "" - ).upper() in {"TLS", "HTTPS", "QUIC"} - - for service, category, patterns in SERVICE_PATTERNS: - if domain and any(fnmatch(domain, pattern) for pattern in patterns): - return redact_sensitive_data( - { - "application_name": application, - "service_name": service, - "service_category": category, - "service_domain": domain, - "service_confidence": "high", - "service_reasons": [f"{source.upper()} matched {domain}"], - "service_sources": [source], - "service_encrypted": encrypted, - "service_unknown": False, - } +@dataclass(frozen=True) +class ServiceAttributionResult: + application_name: str + service_name: str + service_category: str + domain: str + destination_ip: str + destination_port: int + attribution_confidence: str + confidence_score: int + attribution_reasons: tuple[str, ...] + attribution_sources: tuple[str, ...] + is_encrypted: bool + is_unknown: bool + is_cdn: bool + is_proxy_or_vpn_suspected: bool + risk_hint: str + + def to_dict(self) -> dict[str, Any]: + return redact_sensitive_data(asdict(self)) + + +class ServiceAttributionEngine: + """Deterministic, metadata-only destination attribution with local state.""" + + def __init__( + self, + registry_path: str | Path | None = None, + *, + enabled: bool = True, + dns_window_sec: int = 300, + max_reasons: int = 8, + unknown_rate_warn: float = 0.75, + dns_cache_max: int = DEFAULT_DNS_CACHE_MAX, + flow_metric_cache_max: int = DEFAULT_FLOW_METRIC_CACHE_MAX, + ) -> None: + self.enabled = bool(enabled) + self.registry_path = Path(registry_path or DEFAULT_REGISTRY_PATH) + self.dns_window_sec = _safe_int(dns_window_sec, 300, 1, 3600) + self.max_reasons = _safe_int(max_reasons, 8, 1, 20) + self.unknown_rate_warn = max(0.0, min(float(unknown_rate_warn), 1.0)) + self.dns_cache_max = _safe_int( + dns_cache_max, DEFAULT_DNS_CACHE_MAX, 1, 1_000_000 + ) + self.flow_metric_cache_max = _safe_int( + flow_metric_cache_max, + DEFAULT_FLOW_METRIC_CACHE_MAX, + 1, + 1_000_000, + ) + self._lock = threading.RLock() + self._dns_by_ip: OrderedDict[str, deque[tuple[float, str]]] = OrderedDict() + self._latencies: deque[float] = deque(maxlen=512) + self._flow_outcomes: OrderedDict[str, tuple[bool, str, bool, bool]] = ( + OrderedDict() + ) + self._attributed = self._unknown = 0 + self._high = self._medium = self._low = 0 + self._encrypted_unknown = self._cdn_only = self._errors = 0 + self._last_error = "" + self._registry_loaded = False + self._fingerprints = self._load_registry() + + @classmethod + def from_env(cls) -> "ServiceAttributionEngine": + try: + unknown_rate = float( + os.environ.get("NETBOT_SERVICE_ATTRIBUTION_UNKNOWN_RATE_WARN", "0.75") + ) + except ValueError: + unknown_rate = 0.75 + return cls( + os.environ.get("NETBOT_SERVICE_ATTRIBUTION_REGISTRY") + or DEFAULT_REGISTRY_PATH, + enabled=_env_bool("NETBOT_SERVICE_ATTRIBUTION_ENABLED", True), + dns_window_sec=_safe_int( + os.environ.get("NETBOT_SERVICE_ATTRIBUTION_DNS_WINDOW_SEC"), + 300, + 1, + 3600, + ), + max_reasons=_safe_int( + os.environ.get("NETBOT_SERVICE_ATTRIBUTION_MAX_REASONS"), + 8, + 1, + 20, + ), + unknown_rate_warn=unknown_rate, + ) + + def _load_registry(self) -> tuple[ServiceFingerprint, ...]: + try: + payload = json.loads(self.registry_path.read_text(encoding="utf-8")) + rows = payload.get("services") if isinstance(payload, dict) else payload + if not isinstance(rows, list): + raise ValueError("InvalidServiceRegistry") + fingerprints: list[ServiceFingerprint] = [] + for row in rows: + try: + if isinstance(row, dict): + fingerprints.append(ServiceFingerprint.from_dict(row)) + except (TypeError, ValueError): + continue + if not fingerprints: + raise ValueError("EmptyServiceRegistry") + self._registry_loaded = True + return tuple(fingerprints) + except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: + self._last_error = type(exc).__name__ + return () + + def enrich(self, packet: dict[str, Any]) -> dict[str, Any]: + result = self.attribute(packet) + public = result.to_dict() + packet["service_attribution"] = public + packet.update( + { + "service_name": public["service_name"], + "service_category": public["service_category"], + "service_domain": public["domain"], + "service_confidence": public["attribution_confidence"], + "service_confidence_score": public["confidence_score"], + "service_reasons": list(public["attribution_reasons"]), + "service_sources": list(public["attribution_sources"]), + "service_encrypted": public["is_encrypted"], + "service_unknown": public["is_unknown"], + "service_is_cdn": public["is_cdn"], + "service_risk_hint": public["risk_hint"], + } + ) + return public + + def attribute(self, packet: dict[str, Any]) -> ServiceAttributionResult: + started = time.perf_counter() + try: + self._observe_dns(packet) + result = self._attribute(packet) + except Exception as exc: # pragma: no cover - defensive hot-path guard + with self._lock: + self._errors += 1 + self._last_error = type(exc).__name__ + result = self._unknown_result( + packet, "Attribution metadata could not be evaluated safely." + ) + self._record_result(packet, result, (time.perf_counter() - started) * 1000.0) + return result + + def _attribute(self, packet: dict[str, Any]) -> ServiceAttributionResult: + if not self.enabled: + return self._unknown_result(packet, "Service attribution is disabled.") + + application = redact_sensitive_text(str(packet.get("process_name") or ""))[:120] + destination_ip = _safe_ip( + packet.get("remote_ip") or packet.get("dst") or packet.get("destination_ip") + ) + destination_port = _safe_int( + packet.get("dport") + or packet.get("destination_port") + or packet.get("sport"), + 0, + 0, + 65535, + ) + encrypted = self._is_encrypted(packet, destination_port) + proxy_or_vpn = self._is_proxy_or_vpn(packet, application) + org = redact_sensitive_text( + str( + packet.get("org") + or packet.get("asn_org") + or packet.get("organization") + or "" + ) + )[:200] + + candidates: list[tuple[str, str]] = [] + for source, value in ( + ("http_host", packet.get("http_host")), + ( + "tls_sni", + packet.get("tls_sni") + or packet.get("sni") + or packet.get("quic_sni") + or packet.get("quic_server_name"), + ), + ("dns", packet.get("dns_qname")), + ): + if normalized := _domain(value): + candidates.append((source, normalized)) + for value in packet.get("resolved_domains") or []: + if normalized := _domain(value): + candidates.append(("dns", normalized)) + if destination_ip: + for value in self._recent_domains(destination_ip, packet): + candidates.append(("dns", value)) + + evidence: list[tuple[str, str, ServiceFingerprint]] = [] + for source, candidate in candidates: + fingerprint = self._match(candidate, source) + if fingerprint: + evidence.append((source, candidate, fingerprint)) + + if evidence: + return self._result_from_evidence( + packet, + evidence, + application, + destination_ip, + destination_port, + encrypted, + proxy_or_vpn, + org, + ) + + org_match = self._match_org(org) + if org_match: + if org_match.cdn: + service_name = "CDN only" + category = "Unknown / CDN" + risk_hint = "shared_cdn_only" + else: + service_name = f"{org_match.service_name} Services" + category = org_match.category + risk_hint = "organization_only_attribution" + return ServiceAttributionResult( + application, + service_name, + category, + "", + destination_ip, + destination_port, + "low", + 30, + (f"ASN organization matched {org}", "No service domain was visible."), + ("asn_org",), + encrypted, + False, + org_match.cdn, + proxy_or_vpn, + risk_hint, + ) + return self._unknown_result(packet) + + def _result_from_evidence( + self, + packet: dict[str, Any], + evidence: list[tuple[str, str, ServiceFingerprint]], + application: str, + destination_ip: str, + destination_port: int, + encrypted: bool, + proxy_or_vpn: bool, + org: str, + ) -> ServiceAttributionResult: + priority = {"http_host": 3, "tls_sni": 2, "dns": 1} + selected_source, selected_domain, selected = max( + evidence, key=lambda item: priority[item[0]] + ) + agreeing = [ + item for item in evidence if item[2].service_name == selected.service_name + ] + conflicting = any( + item[2].service_name != selected.service_name for item in evidence + ) + base_scores = {"http_host": 90, "tls_sni": 80, "dns": 60} + score = base_scores[selected_source] + sources: list[str] = [] + reasons: list[str] = [] + source_labels = { + "http_host": "HTTP Host", + "tls_sni": "TLS SNI", + "dns": "Recent DNS query", + } + for source, domain, _fingerprint in agreeing: + if source not in sources: + sources.append(source) + reasons.append(f"{source_labels[source]} matched {domain}") + if source != selected_source: + score += 10 if source == "dns" else 15 + if org and any(pattern in org.lower() for pattern in selected.asn_org_patterns): + sources.append("asn_org") + reasons.append(f"ASN organization matched {org}") + score += 15 + reasons.append(f"Domain pattern matched {selected.service_name} fingerprint") + if "fingerprint" not in sources: + sources.append("fingerprint") + if conflicting: + reasons.append("Conflicting service metadata reduced confidence.") + score -= 20 + score = max(0, min(score, 100)) + confidence = "high" if score >= 80 else "medium" if score >= 50 else "low" + return ServiceAttributionResult( + application, + selected.service_name, + selected.category, + selected_domain, + destination_ip, + destination_port, + confidence, + score, + tuple(reasons[: self.max_reasons]), + tuple(dict.fromkeys(sources)), + encrypted, + False, + selected.cdn, + proxy_or_vpn, + f"known_{selected.category.lower().replace(' ', '_').replace('/', '_')}_service", + ) + + def _unknown_result( + self, packet: dict[str, Any], reason: str | None = None + ) -> ServiceAttributionResult: + application = redact_sensitive_text(str(packet.get("process_name") or ""))[:120] + destination_ip = _safe_ip(packet.get("remote_ip") or packet.get("dst")) + destination_port = _safe_int( + packet.get("dport") or packet.get("sport"), 0, 0, 65535 + ) + encrypted = self._is_encrypted(packet, destination_port) + reasons = [ + reason or "No visible DNS, SNI, or HTTP Host evidence was available." + ] + if application.lower() in BROWSER_OR_CONTAINER_PROCESSES: + reasons.append( + "Browser or container process identity alone is not service evidence." ) + if encrypted: + reasons.append("Destination appears encrypted.") + return ServiceAttributionResult( + application, + "Unknown encrypted destination" if encrypted else "Unknown", + "Unknown", + "", + destination_ip, + destination_port, + "low" if encrypted else "unknown", + 20 if encrypted else (5 if application else 0), + tuple(reasons[: self.max_reasons]), + (), + encrypted, + True, + False, + self._is_proxy_or_vpn(packet, application), + "unknown_encrypted_destination" if encrypted else "unknown_destination", + ) - org_lower = org.lower() - for org_pattern, service, category in ORG_FALLBACKS: - if org_pattern in org_lower: - return redact_sensitive_data( - { - "application_name": application, - "service_name": service, - "service_category": category, - "service_domain": domain, - "service_confidence": "low", - "service_reasons": [ - f"Network organization matched {redact_sensitive_text(org)}", - "Encrypted traffic did not expose a reliable service domain", - ], - "service_sources": ["asn_org"], - "service_encrypted": encrypted, - "service_unknown": False, - } + def _match(self, domain: str, source: str) -> ServiceFingerprint | None: + best: tuple[int, ServiceFingerprint] | None = None + for fingerprint in self._fingerprints: + patterns = ( + fingerprint.http_host_patterns + if source == "http_host" + else ( + fingerprint.sni_patterns + if source == "tls_sni" + else fingerprint.domain_patterns + ) ) + for pattern in patterns: + if fnmatch(domain, pattern) or domain == pattern.removeprefix("*."): + specificity = len(pattern.replace("*", "")) + if best is None or specificity > best[0]: + best = (specificity, fingerprint) + return best[1] if best else None - return { - "application_name": application, - "service_name": "Unknown Encrypted" if encrypted else "Unknown", - "service_category": "Unknown", - "service_domain": domain, - "service_confidence": "unknown", - "service_reasons": ["No reliable DNS, SNI, HTTP Host, or organization match"], - "service_sources": [], - "service_encrypted": encrypted, - "service_unknown": True, - } + def _match_org(self, org: str) -> ServiceFingerprint | None: + lowered = org.lower() + if not lowered: + return None + matches = [ + fingerprint + for fingerprint in self._fingerprints + if any(pattern in lowered for pattern in fingerprint.asn_org_patterns) + ] + matches.sort( + key=lambda fingerprint: (fingerprint.cdn, len(fingerprint.domain_patterns)) + ) + return matches[0] if matches else None + + def _observe_dns(self, packet: dict[str, Any]) -> None: + domain = _domain(packet.get("dns_qname")) + if not domain: + return + answers: list[Any] = [] + for key in ("dns_answer_ips", "resolved_ips", "dns_answers"): + value = packet.get(key) + answers.extend( + value if isinstance(value, list) else [value] if value else [] + ) + if packet.get("dns_answer_ip"): + answers.append(packet["dns_answer_ip"]) + observed_at = _timestamp(packet.get("ts") or packet.get("timestamp")) + with self._lock: + for answer in answers: + candidate = ( + answer.get("ip") or answer.get("address") + if isinstance(answer, dict) + else answer + ) + if ip := _safe_ip(candidate): + values = self._dns_by_ip.setdefault(ip, deque(maxlen=20)) + values.append((observed_at, domain)) + self._dns_by_ip.move_to_end(ip) + if len(self._dns_by_ip) > self.dns_cache_max: + self._dns_by_ip.popitem(last=False) + + def _recent_domains(self, destination_ip: str, packet: dict[str, Any]) -> list[str]: + now = _timestamp(packet.get("ts") or packet.get("timestamp")) + cutoff = now - self.dns_window_sec + with self._lock: + values = self._dns_by_ip.get(destination_ip, deque()) + if destination_ip in self._dns_by_ip: + self._dns_by_ip.move_to_end(destination_ip) + recent = [item for item in values if item[0] >= cutoff] + recent.sort(key=lambda item: abs(now - item[0])) + return [domain for _observed_at, domain in recent] + + @staticmethod + def _is_encrypted(packet: dict[str, Any], port: int) -> bool: + protocol = str( + packet.get("app_protocol") or packet.get("protocol") or "" + ).upper() + return port in {443, 8443, 9443} or protocol in {"TLS", "HTTPS", "QUIC"} + + @staticmethod + def _is_proxy_or_vpn(packet: dict[str, Any], application: str) -> bool: + text = " ".join( + [ + application, + str(packet.get("service_category") or ""), + str(packet.get("interface") or ""), + ] + ).lower() + return bool( + packet.get("proxy") or packet.get("vpn") or "vpn" in text or "proxy" in text + ) + + @staticmethod + def _flow_metric_key(packet: dict[str, Any]) -> str: + explicit = packet.get("flow_id") or packet.get("flow_key") + if explicit: + return str(explicit) + return "|".join( + str(value or "-") + for value in ( + packet.get("src"), + packet.get("dst") or packet.get("remote_ip"), + packet.get("sport"), + packet.get("dport"), + packet.get("proto") or packet.get("transport"), + packet.get("direction"), + ) + ) + + def _adjust_outcome( + self, outcome: tuple[bool, str, bool, bool], delta: int + ) -> None: + is_unknown, confidence, encrypted_unknown, cdn_only = outcome + if is_unknown: + self._unknown += delta + else: + self._attributed += delta + if confidence == "high": + self._high += delta + elif confidence == "medium": + self._medium += delta + elif confidence == "low": + self._low += delta + if encrypted_unknown: + self._encrypted_unknown += delta + if cdn_only: + self._cdn_only += delta + + def _record_result( + self, + packet: dict[str, Any], + result: ServiceAttributionResult, + latency_ms: float, + ) -> None: + with self._lock: + self._latencies.append(latency_ms) + flow_key = self._flow_metric_key(packet) + outcome = ( + result.is_unknown, + result.attribution_confidence, + result.is_unknown and result.is_encrypted, + result.service_name == "CDN only", + ) + previous = self._flow_outcomes.get(flow_key) + if previous == outcome: + return + if previous: + self._adjust_outcome(previous, -1) + self._flow_outcomes[flow_key] = outcome + self._flow_outcomes.move_to_end(flow_key) + self._adjust_outcome(outcome, 1) + if len(self._flow_outcomes) > self.flow_metric_cache_max: + self._flow_outcomes.popitem(last=False) + + def metrics(self) -> dict[str, Any]: + with self._lock: + total = self._attributed + self._unknown + unknown_rate = self._unknown / total if total else 0.0 + avg_latency = ( + round(sum(self._latencies) / len(self._latencies), 3) + if self._latencies + else 0.0 + ) + p95_latency = _percentile(self._latencies, 0.95) + reasons: list[str] = [] + if self.enabled and not self._registry_loaded: + reasons.append("service_attribution_registry_error") + if p95_latency >= 25.0: + reasons.append("service_attribution_high_latency") + if self._errors: + reasons.append("service_attribution_errors") + if total >= 20 and unknown_rate >= self.unknown_rate_warn: + reasons.append("service_attribution_high_unknown_rate") + health = "healthy" + if reasons: + health = "degraded" + if self.enabled and not self._fingerprints: + health = "critical" + if self._errors >= 25: + health = "critical" + return { + "enabled": self.enabled, + "health": health, + "registry_size": len(self._fingerprints), + "attributed_flows_total": self._attributed, + "unknown_flows_total": self._unknown, + "high_confidence_total": self._high, + "medium_confidence_total": self._medium, + "low_confidence_total": self._low, + "encrypted_unknown_total": self._encrypted_unknown, + "cdn_only_total": self._cdn_only, + "attribution_errors_total": self._errors, + "avg_attribution_latency_ms": avg_latency, + "p95_attribution_latency_ms": p95_latency, + "last_error": self._last_error, + "pressure_reasons": reasons, + } + + def reset_runtime(self) -> None: + with self._lock: + self._dns_by_ip.clear() + self._latencies.clear() + self._flow_outcomes.clear() + self._attributed = self._unknown = 0 + self._high = self._medium = self._low = 0 + self._encrypted_unknown = self._cdn_only = self._errors = 0 + if self._registry_loaded: + self._last_error = "" + + +_default_engine: ServiceAttributionEngine | None = None +_default_engine_lock = threading.Lock() + + +def default_service_attribution_engine() -> ServiceAttributionEngine: + global _default_engine + with _default_engine_lock: + if _default_engine is None: + _default_engine = ServiceAttributionEngine.from_env() + return _default_engine + + +def attribute_service(packet: dict[str, Any]) -> dict[str, Any]: + public = default_service_attribution_engine().attribute(packet).to_dict() + public.update( + { + "service_domain": public["domain"], + "service_confidence": public["attribution_confidence"], + "service_reasons": list(public["attribution_reasons"]), + "service_sources": list(public["attribution_sources"]), + "service_encrypted": public["is_encrypted"], + "service_unknown": public["is_unknown"], + } + ) + if public["is_unknown"] and public["is_encrypted"]: + public["service_name"] = "Unknown Encrypted" + return public def enrich_service_attribution(packet: dict[str, Any]) -> dict[str, Any]: - attribution = attribute_service(packet) - packet.update(attribution) - return attribution + return default_service_attribution_engine().enrich(packet) -__all__ = ["attribute_service", "enrich_service_attribution"] +__all__ = [ + "BROWSER_OR_CONTAINER_PROCESSES", + "ServiceAttributionEngine", + "ServiceAttributionResult", + "ServiceFingerprint", + "attribute_service", + "default_service_attribution_engine", + "enrich_service_attribution", +] diff --git a/backend/app/services/sniffer_persistence.py b/backend/app/services/sniffer_persistence.py index cfa1f28..bc74866 100644 --- a/backend/app/services/sniffer_persistence.py +++ b/backend/app/services/sniffer_persistence.py @@ -120,6 +120,11 @@ def persist(self, packet: dict[str, Any], alerts: list[dict[str, Any]]) -> None: "attribution_reason_unavailable" ), "attribution_source": packet.get("attribution_source"), + "service_attribution": packet.get("service_attribution"), + "service_name": packet.get("service_name"), + "service_category": packet.get("service_category"), + "service_domain": packet.get("service_domain"), + "service_confidence": packet.get("service_confidence"), } alert_rows = [ { @@ -325,8 +330,7 @@ def _run_retention_cleanup(self) -> None: return self._last_retention_cleanup = now try: - from backend.app.services.settings_service import \ - get_settings_snapshot + from backend.app.services.settings_service import get_settings_snapshot cleanup_retention( int(get_settings_snapshot().get("retention_minutes") or 0) diff --git a/backend/app/services/sniffer_service.py b/backend/app/services/sniffer_service.py index 87dabfa..7a59c30 100644 --- a/backend/app/services/sniffer_service.py +++ b/backend/app/services/sniffer_service.py @@ -14,7 +14,7 @@ from backend.app.services.flow_worker_pool import FlowWorkerPool from backend.app.services.live_ring_buffer import LiveRingBuffer from backend.app.services.packet_queue import BoundedPacketQueue -from backend.app.services.service_attribution import enrich_service_attribution +from backend.app.services.service_attribution import ServiceAttributionEngine from backend.app.services.settings_service import get_settings_snapshot from backend.app.services.sniffer_dashboard_state import SnifferDashboardState from backend.app.services.sniffer_detection_pipeline import SnifferDetectionPipeline @@ -54,6 +54,7 @@ def __init__( capture_provider: CaptureProvider | None = None, flow_service: FlowService | None = None, live_ring_buffer: LiveRingBuffer | None = None, + service_attribution: ServiceAttributionEngine | None = None, ) -> None: self._lock = threading.Lock() self._event_bus = event_bus @@ -74,6 +75,9 @@ def __init__( ) self._publisher = SnifferEventPublisher(event_bus) self._live_ring_buffer = live_ring_buffer or LiveRingBuffer.from_env() + self._service_attribution = ( + service_attribution or ServiceAttributionEngine.from_env() + ) self._packet_queue = BoundedPacketQueue( max_size=PACKET_QUEUE_MAX_SIZE, overflow_policy=PACKET_QUEUE_OVERFLOW_POLICY, @@ -186,7 +190,7 @@ def _process_packet(self, meta: dict[str, Any]) -> None: except Exception: logger.exception("Packet analysis pipeline crashed") alerts = [] - enrich_service_attribution(packet) + self._service_attribution.enrich(packet) alerts = self._assign_alert_ids(packet, alerts) for alert in alerts: self._apply_payload_policy(alert, policy.to_public_dict()) @@ -344,6 +348,7 @@ def reset_session(self) -> dict[str, Any]: self._state.reset() self._flow_service.reset() self._live_ring_buffer.clear() + self._service_attribution.reset_runtime() state = self._state.state(running=running, iface=iface) state["observability"] = self.observability() self._publisher.publish_state("sniffer:reset", state) @@ -366,6 +371,9 @@ def flow_worker_pool_stats(self) -> dict[str, Any]: def live_ring_buffer_stats(self) -> dict[str, Any]: return self._live_ring_buffer.metrics() + def service_attribution_stats(self) -> dict[str, Any]: + return self._service_attribution.metrics() + def recent_live_records( self, category: str = "all", @@ -390,6 +398,7 @@ def observability(self) -> dict[str, Any]: "packet_queue": self.packet_queue_stats(), "flow_worker_pool": self.flow_worker_pool_stats(), "live_ring_buffer": self.live_ring_buffer_stats(), + "service_attribution": self.service_attribution_stats(), "persistence": self.persistence_stats(), "auto_block": self.auto_block_stats(), } diff --git a/core/flow_engine.py b/core/flow_engine.py index f7ed600..e4e5738 100644 --- a/core/flow_engine.py +++ b/core/flow_engine.py @@ -77,6 +77,7 @@ def _safe_packet_sample( "length": int(packet.get("length") or 0), "summary": redact_sensitive_text(str(packet.get("summary") or "")), "metadata": deepcopy(protocol["metadata"]), + "service_attribution": deepcopy(packet.get("service_attribution") or {}), } @@ -132,6 +133,12 @@ def ingest( "service_confidence": packet.get("service_confidence") or "unknown", "service_reasons": list(packet.get("service_reasons") or []), "service_sources": list(packet.get("service_sources") or []), + "service_confidence_score": int( + packet.get("service_confidence_score") or 0 + ), + "service_attribution": deepcopy( + packet.get("service_attribution") or {} + ), "country": packet.get("country_name") or packet.get("country"), "asn": packet.get("asn"), "related_alert_ids": [], @@ -176,6 +183,12 @@ def ingest( "service_confidence": packet.get("service_confidence"), "service_reasons": list(packet.get("service_reasons") or []), "service_sources": list(packet.get("service_sources") or []), + "service_confidence_score": int( + packet.get("service_confidence_score") or 0 + ), + "service_attribution": deepcopy( + packet.get("service_attribution") or {} + ), } ) if len(flow["sample_packets"]) < _MAX_SAMPLE_PACKETS: diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index f74a91d..f6e5260 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -186,6 +186,34 @@ Raw PCAP artifacts are not treated like normal reports. They are exposed only through the guarded raw export path, require Full or Forensic mode, require Safe Use acceptance, require token authorization, and create audit records. +## Service Attribution / Destination Intelligence + +`ServiceAttributionEngine` runs after metadata extraction and before flow +aggregation. It correlates only locally available DNS query/answer metadata, +visible TLS SNI, HTTP Host, optional ASN organization metadata, and the bundled +`backend/app/data/service_fingerprints.json` registry. It never performs a +network lookup. + +Attribution is attached to packet summaries before `FlowEngine` ingestion, so +the same centrally redacted structure can travel through flow snapshots, +conversation views, the Live Ring Buffer, WebSocket Event Aggregator, batch +persistence, Inspect, and Flow Details. Flat service fields remain available +for backward compatibility. + +Each result includes a confidence score and label, evidence sources, human +readable reasons, encryption/CDN/proxy hints, and an explicit Unknown state. +HTTP Host and visible SNI are stronger evidence than recent DNS correlation; +agreement can raise confidence and conflicting metadata lowers it. A browser +or container process name alone is never treated as proof of a destination +service. Shared CDN/ASN evidence is labeled conservatively. + +Ops Snapshot receives only fixed counters, latency, health, and bounded +pressure-reason identifiers. It does not receive observed domains, headers, +packet text, credentials, cookies, sessions, or tokens. ECH, DoH, VPNs, +proxies, shared CDNs, and connection reuse can legitimately leave a destination +Unknown. No TLS decryption, MITM, browser history inspection, cookie/session +inspection, or credential collection is part of this layer. + ## Performance Pipeline Foundation ### Batch Persistence Foundation diff --git a/docs/PERFORMANCE_PIPELINE.md b/docs/PERFORMANCE_PIPELINE.md index 3de87ba..f410c33 100644 --- a/docs/PERFORMANCE_PIPELINE.md +++ b/docs/PERFORMANCE_PIPELINE.md @@ -584,16 +584,17 @@ behavior, or AI autonomous actions. ## Next Planned Steps -1. Service Attribution / Destination Intelligence -2. Incident / Correlation Engine -3. Read-only AI Analyst +1. Incident / Correlation Engine +2. Read-only AI Analyst ### Recorded Product Direction -After the remaining performance work, NetBotPro will add conservative Service -Attribution / Destination Intelligence. It will correlate process metadata -with DNS, TLS SNI, HTTP Host, QUIC-visible metadata, ASN, and local service -fingerprints. Missing or weak evidence must remain `Unknown / Encrypted`. +NetBotPro now includes conservative Service Attribution / Destination Intelligence +after packet metadata extraction and before flow aggregation. It +uses DNS, visible TLS SNI, HTTP Host, ASN metadata, and a local fingerprint +registry. The attribution stage adds bounded local matching work; its average +and p95 latency, unknown rate, errors, and registry health are visible in Ops +Snapshot. Missing or weak evidence remains `Unknown / Encrypted`. Incident correlation follows attribution, and a read-only AI Analyst follows incident quality validation. None of these roadmap items authorize TLS diff --git a/docs/SERVICE_ATTRIBUTION.md b/docs/SERVICE_ATTRIBUTION.md new file mode 100644 index 0000000..10fad94 --- /dev/null +++ b/docs/SERVICE_ATTRIBUTION.md @@ -0,0 +1,123 @@ +# Service Attribution And Destination Intelligence + +Process attribution alone only identifies the program that opened a connection; +it cannot tell which site or service a browser, Electron container, or messaging +client reached. Service Attribution gives an analyst a conservative answer to: +"Which known +service is this flow probably reaching, and why?" It is a local metadata +correlation layer. A label is an explainable inference, not proof of application +content or user activity. + +## Pipeline Position + +```text +Capture + -> metadata extraction and central redaction + -> Service Attribution Engine + -> Flow Engine and conversation views + -> Live Ring / Event Aggregator / Batch Persistence + -> Inspect, Flows, and Ops Snapshot +``` + +The engine is used by live capture. Its result is stored on redacted packet and +flow summaries, allowing the same attribution to be reviewed in Inspect, Flow +Details, recent ring-buffer records, and persistence without another lookup. + +## Local Evidence + +The engine may correlate: + +- visible HTTP Host metadata; +- visible TLS SNI metadata; +- DNS query names and recent DNS answer-to-IP observations; +- optional ASN organization metadata already supplied by the capture path; +- destination IP and port as context; +- process name as context only; and +- the bundled local fingerprint registry. + +The default registry is +`backend/app/data/service_fingerprints.json`. It covers common video, +messaging, developer-platform, cloud/search, social, and CDN providers. Registry +matching is deterministic and performs no outbound network request. +Recent DNS and unique-flow metric correlation caches have fixed internal caps, +so long capture sessions cannot make this layer grow without a bound. + +## Confidence And Reasons + +Every result contains a `0..100` confidence score, a `high`, `medium`, `low`, or +`unknown` label, evidence sources, and short reasons. Visible HTTP Host and TLS +SNI are stronger evidence than recent DNS correlation. Multiple agreeing +sources can increase confidence. Conflicting sources reduce confidence. + +Browser and container process names are deliberately weak context. For example, +`chrome.exe` alone does not mean Google or YouTube. ASN evidence for a shared CDN +is labeled `CDN only` unless stronger domain evidence identifies the final +service. + +## Unknown Encrypted Destinations + +Encrypted traffic with no visible DNS, SNI, or Host evidence is reported as +`Unknown encrypted destination`. NetBotPro does not invent a service name. +Common reasons include Encrypted Client Hello (ECH), DNS over HTTPS, VPN or +proxy use, encrypted QUIC, shared CDN infrastructure, browser connection reuse, +NAT with multiple hosts, missing process metadata, missing DNS history, and +missing SNI. + +Examples of intended output include `chrome.exe -> YouTube -> googlevideo.com` +with high confidence when visible SNI agrees with the registry, and +`chrome.exe -> Unknown encrypted destination` when only an encrypted endpoint +is visible. + +## Operations Metrics + +`/api/monitoring/metrics` exposes a fixed `service_attribution` object: + +- enabled and health; +- registry size; +- attributed and unknown unique-flow totals; +- high, medium, and low confidence totals; +- encrypted-unknown and CDN-only totals; +- attribution error total; +- average and p95 attribution latency; and +- safe pressure-reason identifiers. + +Metrics never include observed domains, headers, packet text, credentials, +cookies, sessions, authorization values, or tokens. A high unknown rate is a +diagnostic signal, not automatically a security problem. + +## Configuration + +| Variable | Default | Purpose | +| --- | --- | --- | +| `NETBOT_SERVICE_ATTRIBUTION_ENABLED` | `true` | Enables local attribution. | +| `NETBOT_SERVICE_ATTRIBUTION_REGISTRY` | bundled JSON path | Selects a local registry file. | +| `NETBOT_SERVICE_ATTRIBUTION_DNS_WINDOW_SEC` | `300` | Limits DNS answer correlation age. | +| `NETBOT_SERVICE_ATTRIBUTION_MAX_REASONS` | `8` | Bounds explanation length. | +| `NETBOT_SERVICE_ATTRIBUTION_UNKNOWN_RATE_WARN` | `0.75` | Sets the Ops warning threshold after a meaningful sample. | + +Invalid registry rows are skipped. A missing or wholly invalid registry does +not crash capture; it is reported as critical attribution health and flows stay +Unknown. + +## Privacy And Security Boundaries + +All displayed and persisted attribution data passes through central redaction. +This feature does not add: + +- TLS decryption or MITM; +- credential collection; +- browser history scraping; +- cookie/session inspection; +- raw payload forwarding, raw packet forwarding, or Agent PCAP forwarding; +- command/control, remote shell, or file collection; or +- IPS/auto-blocking or autonomous AI actions. + +Agent/Fleet Mode remains telemetry-only and read-only. Service Attribution runs +on the authorized capture host; it does not expand Agent collection or +forwarding behavior. + +## Safe And Authorized Use + +Use NetBotPro only on systems and networks you own or are explicitly authorized +to monitor. Attribution should guide investigation and validation, not serve as +the sole basis for disciplinary, blocking, or incident decisions. diff --git a/frontend/src/components/FlowsPanel.jsx b/frontend/src/components/FlowsPanel.jsx index bcc528c..1802aa4 100644 --- a/frontend/src/components/FlowsPanel.jsx +++ b/frontend/src/components/FlowsPanel.jsx @@ -170,21 +170,36 @@ export function FlowsPanel({ api }) { {selectedFlow ? ( <> + {(() => { + const attribution = selectedFlow.service_attribution || {}; + const serviceName = attribution.service_name || selectedFlow.service_name || "Unknown"; + const domain = attribution.domain || selectedFlow.service_domain || ""; + const confidence = attribution.attribution_confidence || selectedFlow.service_confidence || "unknown"; + const score = attribution.confidence_score ?? selectedFlow.service_confidence_score ?? 0; + const reasons = attribution.attribution_reasons || selectedFlow.service_reasons || []; + const sources = attribution.attribution_sources || selectedFlow.service_sources || []; + return ( +
+

Service attribution

+
+
Application
{attribution.application_name || selectedFlow.process_name || "Not mapped"}
+
Service
{serviceName}
+
Category
{attribution.service_category || selectedFlow.service_category || "Unknown"}
+
Domain
{domain || "Unavailable"}
+
Confidence
{confidence} {score}/100
+
Evidence
{sources.length ? sources.join(", ") : "No visible metadata evidence"}
+
+ {attribution.is_unknown && attribution.is_encrypted ?

Unknown encrypted destination. ECH, DoH, VPN, proxy, or connection reuse may hide service metadata.

: null} + {attribution.is_cdn ?

Shared CDN infrastructure may represent more than one final service.

: null} + {reasons.length ? : null} +
+ ); + })()}
Duration{selectedFlow.duration_ms} ms Process{selectedFlow.process_name || "Not mapped"} Sent / received{formatBytes(selectedFlow.bytes_sent)} / {formatBytes(selectedFlow.bytes_received)}
-
-

Destination service

-
-
Service
{selectedFlow.service_name || "Unknown"}
-
Category
{selectedFlow.service_category || "Unknown"}
-
Domain
{selectedFlow.service_domain || "Not visible in encrypted metadata"}
-
Confidence
{selectedFlow.service_confidence || "unknown"}
-
- -

Risk reasons

diff --git a/frontend/src/components/FlowsPanel.test.jsx b/frontend/src/components/FlowsPanel.test.jsx index 8f630ee..6c77365 100644 --- a/frontend/src/components/FlowsPanel.test.jsx +++ b/frontend/src/components/FlowsPanel.test.jsx @@ -41,6 +41,20 @@ describe("FlowsPanel", () => { risk_level: "high", risk_reasons: ["Unusual outbound destination"], metadata: { path: "/login?token=[REDACTED]" }, + service_attribution: { + application_name: "chrome.exe", + service_name: "YouTube", + service_category: "Video Streaming", + domain: "r1.googlevideo.com", + attribution_confidence: "high", + confidence_score: 92, + attribution_sources: ["tls_sni", "fingerprint"], + attribution_reasons: ["TLS SNI matched r1.googlevideo.com"], + is_unknown: false, + is_encrypted: true, + is_cdn: true, + secret: "raw-flow-secret", + }, related_alert_ids: [], }; const api = { @@ -52,7 +66,14 @@ describe("FlowsPanel", () => { }; render(); await waitFor(() => expect(screen.getByText("65 high")).toBeTruthy()); + expect(screen.getByText("Service attribution")).toBeTruthy(); + expect(screen.getByText("YouTube")).toBeTruthy(); + expect(screen.getByText("Video Streaming")).toBeTruthy(); + expect(screen.getByText("r1.googlevideo.com")).toBeTruthy(); + expect(screen.getByText("high 92/100")).toBeTruthy(); + expect(screen.getByText("tls_sni, fingerprint")).toBeTruthy(); + expect(screen.getByText("Shared CDN infrastructure may represent more than one final service.")).toBeTruthy(); expect(document.body.textContent).toContain("[REDACTED]"); - expect(document.body.textContent).not.toContain("Bearer real-secret"); + expect(document.body.textContent).not.toMatch(/Bearer real-secret|raw-flow-secret/); }); }); diff --git a/frontend/src/components/OpsPanel.jsx b/frontend/src/components/OpsPanel.jsx index 0617cc4..2dc011e 100644 --- a/frontend/src/components/OpsPanel.jsx +++ b/frontend/src/components/OpsPanel.jsx @@ -38,6 +38,7 @@ export function OpsPanel({ observability, operationalMetrics = null, isRefreshin const flowWorkerPool = snapshot.flowWorkerPool; const flowWorkerLastDropReason = snapshot.safeFlowWorkerDropReason || "No drops recorded"; const liveRingBuffer = snapshot.liveRingBuffer; + const serviceAttribution = snapshot.serviceAttribution; const persistence = snapshot.persistence; const eventBus = snapshot.eventBus; const eventAggregator = snapshot.eventAggregator; @@ -120,6 +121,27 @@ export function OpsPanel({ observability, operationalMetrics = null, isRefreshin + +
+ + + + + 0 ? "warning" : "healthy"} /> + + 0 ? "degraded" : "healthy"} hint={serviceAttribution.last_error || "None"} /> + + = 25 ? "warning" : "healthy"} /> + +
+ + { expect(screen.getByText("Live ring buffer query limit was capped. Reduce requested result size or inspect a narrower time range.")).toBeTruthy(); expect(document.body.textContent).not.toMatch(/raw-secret|authorization/i); }); + + it("renders service attribution health and safe recommended actions", () => { + render( + + ); + + expect(screen.getByText("Service Attribution")).toBeTruthy(); + expect(screen.getByText("22")).toBeTruthy(); + expect(screen.getAllByText("14").length).toBeGreaterThan(0); + expect(screen.getByText("Many flows could not be attributed. This may be normal with ECH, DoH, VPN, proxy, or shared CDN traffic.")).toBeTruthy(); + expect(screen.getByText("Service attribution latency is high. Review fingerprint registry size and matching rules.")).toBeTruthy(); + expect(screen.getByText("Service attribution errors were observed. Inspect backend logs and recent flow metadata.")).toBeTruthy(); + expect(document.body.textContent).not.toMatch(/raw-attribution-secret|authorization/i); + }); + + it.each(["healthy", "degraded", "critical"])( + "renders service attribution %s health", + (health) => { + render( + + ); + + expect(screen.getByText("Service Attribution")).toBeTruthy(); + expect(screen.getAllByText(`Health ${health}`).length).toBeGreaterThan(0); + } + ); }); diff --git a/frontend/src/components/ServiceAttributionUi.test.jsx b/frontend/src/components/ServiceAttributionUi.test.jsx new file mode 100644 index 0000000..d34bc7c --- /dev/null +++ b/frontend/src/components/ServiceAttributionUi.test.jsx @@ -0,0 +1,80 @@ +// @vitest-environment jsdom + +import { cleanup, render, screen } from "@testing-library/react"; +import { afterEach, describe, expect, it } from "vitest"; + +import { buildPacketInspectionModel } from "../lib/inspectionModel"; +import { DetailPanel } from "./DetailPanel"; + +afterEach(cleanup); + +describe("service attribution inspection", () => { + it("renders explainable metadata-only attribution in Inspect", () => { + const packet = { + id: "packet-service-1", + src: "10.0.0.5", + dst: "203.0.113.20", + sport: 52000, + dport: 443, + proto: "TCP", + length: 144, + service_attribution: { + application_name: "chrome.exe", + service_name: "YouTube", + service_category: "Video Streaming", + domain: "r1.googlevideo.com", + attribution_confidence: "high", + confidence_score: 95, + attribution_sources: ["tls_sni", "dns", "fingerprint"], + attribution_reasons: ["TLS SNI matched r1.googlevideo.com"], + is_encrypted: true, + is_unknown: false, + is_cdn: false, + credential: "Bearer raw-inspect-secret", + }, + }; + + render( + + ); + + expect(screen.getByText("Service Attribution")).toBeTruthy(); + expect(screen.getByText("YouTube")).toBeTruthy(); + expect(screen.getByText("Video Streaming")).toBeTruthy(); + expect(screen.getByText("r1.googlevideo.com")).toBeTruthy(); + expect(screen.getByText("tls_sni, dns, fingerprint")).toBeTruthy(); + expect(screen.getByText("TLS SNI matched r1.googlevideo.com")).toBeTruthy(); + expect(document.body.textContent).not.toContain("raw-inspect-secret"); + }); + + it("labels an encrypted destination as unknown without guessing a service", () => { + const model = buildPacketInspectionModel({ + id: "packet-service-2", + src: "10.0.0.5", + dst: "203.0.113.21", + dport: 443, + proto: "TCP", + length: 80, + service_attribution: { + service_name: "Unknown encrypted destination", + service_category: "Unknown", + attribution_confidence: "low", + confidence_score: 20, + attribution_sources: [], + attribution_reasons: ["No visible DNS, SNI, or HTTP Host evidence was available."], + is_encrypted: true, + is_unknown: true, + is_cdn: false, + }, + }); + + render(); + expect(screen.getByText("Unknown encrypted destination")).toBeTruthy(); + expect(screen.getByText("low")).toBeTruthy(); + expect(screen.getByText("No visible DNS, SNI, or HTTP Host evidence was available.")).toBeTruthy(); + }); +}); diff --git a/frontend/src/lib/inspectionModel.js b/frontend/src/lib/inspectionModel.js index 8a46582..6f09e8d 100644 --- a/frontend/src/lib/inspectionModel.js +++ b/frontend/src/lib/inspectionModel.js @@ -1308,6 +1308,7 @@ function buildRisk(packet, guess, payload, context, signals) { const direction = normalizeDirection(packet?.direction || context?.direction); const endpoints = determineEndpoints(packet); const process = processSummary(packet); + const serviceAttribution = packet?.service_attribution || {}; const behaviorLabels = Array.isArray(context?.behavior_labels) ? context.behavior_labels : Array.isArray(context?.behaviorLabels) ? context.behaviorLabels : []; const behaviorEvidence = Array.isArray(context?.behavior_evidence) ? context.behavior_evidence : Array.isArray(context?.behaviorEvidence) ? context.behaviorEvidence : []; @@ -2404,7 +2405,7 @@ function buildRelatedActivity(context) { body: cleanText(row?.body) || "Root-cause summary is not available for this sample yet", })), }, - ].filter((group) => group.items.length); + ].filter((group) => group?.items?.length); } function buildBehaviorGroups(context) { @@ -2607,6 +2608,7 @@ export function buildPacketInspectionModel(packet, options = {}) { const behaviorLabels = Array.isArray(context?.behavior_labels) ? context.behavior_labels : Array.isArray(context?.behaviorLabels) ? context.behaviorLabels : []; const riskExplanation = buildRiskExplanation(packet, risk, confidence, signals, process, context); const alertCorrelation = context?.alert_correlation || context?.alertCorrelation || {}; + const serviceAttribution = packet?.service_attribution || {}; const packetDirection = packet?.direction || context?.direction; const headline = `${formatDirection(packetDirection)} ${normalizeProto(packet?.proto) || "traffic"} ${ports.remotePort || ports.localPort || ""} traffic`.trim(); const interpretedSummary = `${sentenceCase(cleanText(packetDirection) || "Unknown")} ${normalizeProto(packet?.proto) || "traffic"} ${cleanText(packet?.dport) || cleanText(packet?.sport) || ""}`.trim(); @@ -2636,6 +2638,12 @@ export function buildPacketInspectionModel(packet, options = {}) { hint: process.reasonUnavailable || process.hint, tone: cleanText(packet?.process_name) || cleanText(packet?.pid) ? "active" : "warning", }, + { + label: "Destination Service", + value: serviceAttribution.service_name || packet?.service_name || "Unknown", + hint: `${serviceAttribution.service_category || packet?.service_category || "Unknown"} | ${serviceAttribution.attribution_confidence || packet?.service_confidence || "unknown"} confidence`, + tone: serviceAttribution.is_unknown ? "warning" : "active", + }, { label: "Why It Matters", value: actionHint(packet, risk, guess, context), @@ -2645,6 +2653,21 @@ export function buildPacketInspectionModel(packet, options = {}) { ]; const applicationGroups = [ + { + title: "Service Attribution", + rows: toRows([ + { label: "Application", value: serviceAttribution.application_name || packet?.process_name }, + { label: "Service", value: serviceAttribution.service_name || packet?.service_name }, + { label: "Domain", value: serviceAttribution.domain || packet?.service_domain }, + { label: "Category", value: serviceAttribution.service_category || packet?.service_category }, + { label: "Confidence", value: serviceAttribution.attribution_confidence || packet?.service_confidence }, + { label: "Confidence Score", value: serviceAttribution.confidence_score ?? packet?.service_confidence_score }, + { label: "Evidence Sources", value: formatList(serviceAttribution.attribution_sources || packet?.service_sources) }, + { label: "Attribution Reasons", value: formatList(serviceAttribution.attribution_reasons || packet?.service_reasons) }, + { label: "Encrypted Unknown", value: serviceAttribution.is_unknown && serviceAttribution.is_encrypted ? "Yes" : "No" }, + { label: "Shared CDN", value: serviceAttribution.is_cdn ? "Yes" : "No" }, + ]), + }, { title: "DNS", rows: toRows([ @@ -2704,6 +2727,14 @@ export function buildPacketInspectionModel(packet, options = {}) { "attribution_confidence", "attribution_reason_unavailable", "attribution_source", + "service_attribution", + "service_name", + "service_category", + "service_domain", + "service_confidence", + "service_confidence_score", + "service_reasons", + "service_sources", "src_mac", "dst_mac", "vendor_src", diff --git a/frontend/src/lib/opsHealth.js b/frontend/src/lib/opsHealth.js index f25331e..74f25f2 100644 --- a/frontend/src/lib/opsHealth.js +++ b/frontend/src/lib/opsHealth.js @@ -139,6 +139,32 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { pressure_reasons: safeLiveRingReasons(liveRingBuffer.pressure_reasons), categories: safeLiveRingCategories, }; + const serviceAttribution = operationalMetrics?.service_attribution || observability?.service_attribution || {}; + const safeServiceAttributionReasons = Array.isArray(serviceAttribution.pressure_reasons) + ? serviceAttribution.pressure_reasons.filter((reason) => [ + "service_attribution_registry_error", + "service_attribution_high_latency", + "service_attribution_errors", + "service_attribution_high_unknown_rate", + ].includes(reason)) + : []; + const safeServiceAttribution = { + enabled: serviceAttribution.enabled !== false, + health: ["healthy", "degraded", "critical"].includes(serviceAttribution.health) ? serviceAttribution.health : "healthy", + registry_size: toNumber(serviceAttribution.registry_size), + attributed_flows_total: toNumber(serviceAttribution.attributed_flows_total), + unknown_flows_total: toNumber(serviceAttribution.unknown_flows_total), + high_confidence_total: toNumber(serviceAttribution.high_confidence_total), + medium_confidence_total: toNumber(serviceAttribution.medium_confidence_total), + low_confidence_total: toNumber(serviceAttribution.low_confidence_total), + encrypted_unknown_total: toNumber(serviceAttribution.encrypted_unknown_total), + cdn_only_total: toNumber(serviceAttribution.cdn_only_total), + attribution_errors_total: toNumber(serviceAttribution.attribution_errors_total), + avg_attribution_latency_ms: toNumber(serviceAttribution.avg_attribution_latency_ms), + p95_attribution_latency_ms: toNumber(serviceAttribution.p95_attribution_latency_ms), + last_error: /^[A-Za-z0-9_]{1,80}$/.test(String(serviceAttribution.last_error || "")) ? String(serviceAttribution.last_error) : "", + pressure_reasons: safeServiceAttributionReasons, + }; const persistence = operationalMetrics?.persistence || observability?.persistence || {}; const history = observability?.history || {}; const autoBlock = observability?.auto_block || {}; @@ -254,10 +280,13 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { ? "warning" : "healthy" : "healthy"; + const serviceAttributionLevel = safeServiceAttribution.enabled + ? normalizeLevel(safeServiceAttribution.health) + : "healthy"; const freshnessLevel = ageSeconds == null || ageSeconds <= 120 ? "healthy" : ageSeconds <= 300 ? "warning" : "degraded"; const criticalFlows = toNumber(flows.risk_distribution?.critical); const highFlows = toNumber(flows.risk_distribution?.high); - const overall = worstLevel(backendLevel, freshnessLevel, packetQueueLevel, flowWorkerLevel, liveRingLevel, persistenceLevel, streamLevel, queryLevel, autoBlockLevel); + const overall = worstLevel(backendLevel, freshnessLevel, packetQueueLevel, flowWorkerLevel, liveRingLevel, serviceAttributionLevel, persistenceLevel, streamLevel, queryLevel, autoBlockLevel); const recommendedActions = []; if (backendLevel !== "healthy") { @@ -305,6 +334,18 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { if (liveRingEnabled && safeLiveRingBuffer.last_error) { recommendedActions.push("Live ring buffer reported errors. Inspect backend logs and recent live capture activity."); } + if (safeServiceAttributionReasons.includes("service_attribution_registry_error")) { + recommendedActions.push("Service attribution registry failed to load. Check service_fingerprints.json."); + } + if (safeServiceAttributionReasons.includes("service_attribution_high_unknown_rate")) { + recommendedActions.push("Many flows could not be attributed. This may be normal with ECH, DoH, VPN, proxy, or shared CDN traffic."); + } + if (safeServiceAttributionReasons.includes("service_attribution_high_latency")) { + recommendedActions.push("Service attribution latency is high. Review fingerprint registry size and matching rules."); + } + if (safeServiceAttribution.attribution_errors_total > 0) { + recommendedActions.push("Service attribution errors were observed. Inspect backend logs and recent flow metadata."); + } if (criticalFlows > 0) { recommendedActions.push("Review critical flows and related alerts first."); } else if (highFlows > 0) { @@ -399,6 +440,12 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { hint: `${liveRingUtilization.toFixed(1)}% used | Evicted ${liveRingEvicted}`, level: liveRingLevel, }, + { + label: "Attribution", + value: String(safeServiceAttribution.attributed_flows_total), + hint: `Unknown ${safeServiceAttribution.unknown_flows_total} | High ${safeServiceAttribution.high_confidence_total}`, + level: serviceAttributionLevel, + }, { label: "Write Queue", value: String(queueSize), @@ -464,6 +511,8 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { safeFlowWorkerDropReason: safeFlowWorkerPool.last_drop_reason, liveRingBuffer: safeLiveRingBuffer, liveRingLevel, + serviceAttribution: safeServiceAttribution, + serviceAttributionLevel, persistence, history, autoBlock, diff --git a/frontend/src/styles.css b/frontend/src/styles.css index a9d9d22..a7b89ac 100644 --- a/frontend/src/styles.css +++ b/frontend/src/styles.css @@ -1832,6 +1832,19 @@ a { gap: 4px; } +.flow-attribution-summary { + min-width: 0; +} + +.flow-attribution-warning { + margin: 0; + padding: 9px 11px; + border-left: 3px solid var(--warning); + background: color-mix(in srgb, var(--warning) 9%, transparent); + color: var(--text); + overflow-wrap: anywhere; +} + .flow-metadata { margin: 0; display: grid; diff --git a/packaging/pyinstaller/netbotpro_backend.spec b/packaging/pyinstaller/netbotpro_backend.spec index f7c0b5b..51e15f5 100644 --- a/packaging/pyinstaller/netbotpro_backend.spec +++ b/packaging/pyinstaller/netbotpro_backend.spec @@ -39,6 +39,7 @@ if project_root is None: desktop_entry = project_root / "backend" / "app" / "desktop_entry.py" config_dir = project_root / "config" +service_registry_dir = project_root / "backend" / "app" / "data" hiddenimports = [ "backend.app.desktop_entry", @@ -55,6 +56,7 @@ a = Analysis( binaries=[], datas=[ (str(config_dir), "config"), + (str(service_registry_dir), "backend/app/data"), ], hiddenimports=hiddenimports, hookspath=[], diff --git a/scripts/qa/packaged_backend_smoke.py b/scripts/qa/packaged_backend_smoke.py index 334404e..494688f 100644 --- a/scripts/qa/packaged_backend_smoke.py +++ b/scripts/qa/packaged_backend_smoke.py @@ -10,15 +10,27 @@ import urllib.request from pathlib import Path - PROJECT_ROOT = Path(__file__).resolve().parents[2] PACKAGED_BACKEND_DIR = PROJECT_ROOT / "packaging" / "runtime" / "backend" -DESKTOP_PACKAGED_BACKEND_DIR = PROJECT_ROOT / "desktop" / "electron" / "dist" / "win-unpacked" / "resources" / "runtime" / "backend" +DESKTOP_PACKAGED_BACKEND_DIR = ( + PROJECT_ROOT + / "desktop" + / "electron" + / "dist" + / "win-unpacked" + / "resources" + / "runtime" + / "backend" +) def expected_binary_name(platform_name: str | None = None) -> str: platform_name = platform_name or os.sys.platform - return "netbotpro-backend.exe" if platform_name.startswith("win") else "netbotpro-backend" + return ( + "netbotpro-backend.exe" + if platform_name.startswith("win") + else "netbotpro-backend" + ) def resolve_runtime_dir(runtime_dir: Path = PACKAGED_BACKEND_DIR) -> Path: @@ -43,7 +55,10 @@ def list_support_files(runtime_dir: Path = PACKAGED_BACKEND_DIR) -> list[Path]: runtime_dir = resolve_runtime_dir(runtime_dir) runtime_dir = Path(runtime_dir) binary_path = find_binary(runtime_dir) - return sorted([entry for entry in runtime_dir.iterdir() if entry.name != binary_path.name], key=lambda item: item.name) + return sorted( + [entry for entry in runtime_dir.iterdir() if entry.name != binary_path.name], + key=lambda item: item.name, + ) def _free_port() -> int: @@ -86,13 +101,27 @@ def validate_interfaces_payload(payload: dict) -> None: raise AssertionError("Interfaces recommendations must be a list when present") -def run_smoke(runtime_dir: Path = PACKAGED_BACKEND_DIR, timeout_sec: float = 20.0) -> None: +def validate_monitoring_payload(payload: dict) -> None: + attribution = payload.get("service_attribution") + if not isinstance(attribution, dict): + raise AssertionError("Monitoring payload must include service_attribution") + if int(attribution.get("registry_size") or 0) < 1: + raise AssertionError("Packaged service attribution registry is unavailable") + if attribution.get("health") == "critical": + raise AssertionError("Packaged service attribution health must not be critical") + + +def run_smoke( + runtime_dir: Path = PACKAGED_BACKEND_DIR, timeout_sec: float = 20.0 +) -> None: runtime_dir = resolve_runtime_dir(runtime_dir) runtime_dir = Path(runtime_dir) binary_path = find_binary(runtime_dir) support_files = list_support_files(runtime_dir) if not support_files: - raise AssertionError(f"Expected staged support files next to {binary_path.name}") + raise AssertionError( + f"Expected staged support files next to {binary_path.name}" + ) port = _free_port() with tempfile.TemporaryDirectory() as td: @@ -113,10 +142,20 @@ def run_smoke(runtime_dir: Path = PACKAGED_BACKEND_DIR, timeout_sec: float = 20. stderr=subprocess.DEVNULL, ) try: - status_payload = _wait_for_http(f"http://127.0.0.1:{port}/api/status", timeout_sec=timeout_sec) + status_payload = _wait_for_http( + f"http://127.0.0.1:{port}/api/status", timeout_sec=timeout_sec + ) validate_status_payload(status_payload) - interfaces_payload = _request_json(f"http://127.0.0.1:{port}/api/interfaces", timeout_sec=max(timeout_sec, 6.0)) + interfaces_payload = _request_json( + f"http://127.0.0.1:{port}/api/interfaces", + timeout_sec=max(timeout_sec, 6.0), + ) validate_interfaces_payload(interfaces_payload) + monitoring_payload = _request_json( + f"http://127.0.0.1:{port}/api/monitoring/metrics", + timeout_sec=max(timeout_sec, 6.0), + ) + validate_monitoring_payload(monitoring_payload) finally: process.terminate() try: diff --git a/tests/test_monitoring_metrics.py b/tests/test_monitoring_metrics.py index 8921a11..49a74d6 100644 --- a/tests/test_monitoring_metrics.py +++ b/tests/test_monitoring_metrics.py @@ -425,6 +425,8 @@ def test_monitoring_metrics_endpoint_returns_compact_snapshot(self): self.assertIn("websocket", payload) self.assertIn("persistence", payload) self.assertIn("flow_worker_pool", payload) + self.assertIn("service_attribution", payload) + self.assertIn("registry_size", payload["service_attribution"]) self.assertEqual(payload["health"], "healthy") def test_runtime_observability_snapshot_includes_flow_worker_pool(self): diff --git a/tests/test_packaged_backend_runtime.py b/tests/test_packaged_backend_runtime.py index 79b28a1..597a666 100644 --- a/tests/test_packaged_backend_runtime.py +++ b/tests/test_packaged_backend_runtime.py @@ -1,8 +1,8 @@ +import json import os import tempfile import unittest from pathlib import Path -import json from scripts.qa import packaged_backend_smoke from scripts.release import stage_backend_runtime @@ -16,13 +16,17 @@ def test_stage_backend_runtime_copies_full_bundle(self): runtime_dir = repo_root / "packaging" / "runtime" / "backend" bundle_dir.mkdir(parents=True) - binary_name = stage_backend_runtime.expected_binary_name("win32" if os.name == "nt" else "linux") + binary_name = stage_backend_runtime.expected_binary_name( + "win32" if os.name == "nt" else "linux" + ) (bundle_dir / binary_name).write_text("binary", encoding="utf-8") (bundle_dir / "support.dll").write_text("support", encoding="utf-8") (bundle_dir / "data").mkdir() (bundle_dir / "data" / "config.json").write_text("{}", encoding="utf-8") - staged_binary = stage_backend_runtime.stage_backend_runtime(bundle_dir, runtime_dir) + staged_binary = stage_backend_runtime.stage_backend_runtime( + bundle_dir, runtime_dir + ) self.assertEqual(staged_binary, runtime_dir / binary_name) self.assertTrue((runtime_dir / "support.dll").exists()) @@ -33,7 +37,9 @@ def test_packaged_backend_smoke_helpers_find_bundle_and_status_leak(self): runtime_dir = Path(td) / "packaging" / "runtime" / "backend" runtime_dir.mkdir(parents=True) - binary = runtime_dir / packaged_backend_smoke.expected_binary_name("win32" if os.name == "nt" else "linux") + binary = runtime_dir / packaged_backend_smoke.expected_binary_name( + "win32" if os.name == "nt" else "linux" + ) binary.write_text("binary", encoding="utf-8") support = runtime_dir / "python313.dll" support.write_text("support", encoding="utf-8") @@ -54,13 +60,30 @@ def test_packaged_backend_smoke_helpers_find_bundle_and_status_leak(self): "recommendations": ["Run Netbotpro as Administrator."], } ) + packaged_backend_smoke.validate_monitoring_payload( + { + "service_attribution": { + "health": "healthy", + "registry_size": 23, + } + } + ) with self.assertRaises(AssertionError): - packaged_backend_smoke.validate_status_payload({"ok": True, "project_root": "C:/secret"}) + packaged_backend_smoke.validate_status_payload( + {"ok": True, "project_root": "C:/secret"} + ) with self.assertRaises(AssertionError): - packaged_backend_smoke.validate_interfaces_payload({"items": [], "degraded": True, "source": "fallback"}) + packaged_backend_smoke.validate_interfaces_payload( + {"items": [], "degraded": True, "source": "fallback"} + ) def test_desktop_package_supports_cross_platform_dist_scripts(self): - package_json = Path(__file__).resolve().parents[1] / "desktop" / "electron" / "package.json" + package_json = ( + Path(__file__).resolve().parents[1] + / "desktop" + / "electron" + / "package.json" + ) data = json.loads(package_json.read_text(encoding="utf-8")) build_resources = package_json.parent / "build-resources" @@ -68,8 +91,12 @@ def test_desktop_package_supports_cross_platform_dist_scripts(self): self.assertIn("dist:win", scripts) self.assertIn("dist:linux", scripts) self.assertNotIn("electronDist", data.get("build", {})) - self.assertEqual(data.get("build", {}).get("linux", {}).get("target"), ["AppImage", "deb"]) - self.assertEqual(data.get("build", {}).get("win", {}).get("icon"), "build-resources/icon.ico") + self.assertEqual( + data.get("build", {}).get("linux", {}).get("target"), ["AppImage", "deb"] + ) + self.assertEqual( + data.get("build", {}).get("win", {}).get("icon"), "build-resources/icon.ico" + ) self.assertTrue((build_resources / "icon.ico").exists()) self.assertTrue((build_resources / "icon.png").exists()) diff --git a/tests/test_release_readiness.py b/tests/test_release_readiness.py index 9fc88b0..d7d4e0a 100644 --- a/tests/test_release_readiness.py +++ b/tests/test_release_readiness.py @@ -100,6 +100,32 @@ def test_deep_packet_inspection_docs_and_safety_are_linked(self): self.assertIn("core/tcp_analysis.py", architecture) self.assertIn("core/dns_intelligence.py", architecture) + def test_service_attribution_docs_and_boundaries_are_explicit(self): + service_docs = self._read("docs/SERVICE_ATTRIBUTION.md") + service_lower = service_docs.lower() + readme = self._read("README.md") + architecture = self._read("docs/ARCHITECTURE.md") + + self.assertIn("docs/SERVICE_ATTRIBUTION.md", readme) + self.assertIn("Service Attribution / Destination Intelligence", readme) + self.assertIn("ServiceAttributionEngine", architecture) + self.assertIn("service_fingerprints.json", architecture) + packaged_spec = self._read("packaging/pyinstaller/netbotpro_backend.spec") + self.assertIn('"backend/app/data"', packaged_spec) + self.assertIn("Unknown encrypted destination", service_docs) + self.assertIn("browser and container", service_lower) + self.assertIn("no outbound network request", service_lower) + for boundary in [ + "tls decryption", + "mitm", + "credential collection", + "browser history scraping", + "cookie/session inspection", + "command/control", + "raw payload forwarding", + ]: + self.assertIn(boundary, service_lower) + def test_performance_pipeline_docs_are_linked_and_scoped(self): readme = self._read("README.md") architecture = self._read("docs/ARCHITECTURE.md") diff --git a/tests/test_service_attribution.py b/tests/test_service_attribution.py index 89e1472..720f164 100644 --- a/tests/test_service_attribution.py +++ b/tests/test_service_attribution.py @@ -1,38 +1,273 @@ +import json +import tempfile import unittest +from pathlib import Path +from unittest.mock import patch -from backend.app.services.service_attribution import attribute_service +from backend.app.services.batch_persistence import BatchPersistenceWriter +from backend.app.services.event_aggregator import EventAggregator +from backend.app.services.live_ring_buffer import DEFAULT_CAPACITIES, LiveRingBuffer +from backend.app.services.monitoring_service import build_monitoring_metrics +from backend.app.services.service_attribution import ( + DEFAULT_REGISTRY_PATH, + ServiceAttributionEngine, + attribute_service, +) +from core.flow_engine import FlowEngine class ServiceAttributionTests(unittest.TestCase): + def setUp(self): + self.engine = ServiceAttributionEngine(DEFAULT_REGISTRY_PATH) + + def result(self, **packet): + defaults = { + "process_name": "chrome.exe", + "src": "10.0.0.5", + "dst": "198.51.100.20", + "sport": 52000, + "dport": 443, + "app_protocol": "TLS", + } + defaults.update(packet) + return self.engine.attribute(defaults).to_dict() + + def test_loads_local_registry(self): + metrics = self.engine.metrics() + self.assertGreaterEqual(metrics["registry_size"], 20) + self.assertEqual(metrics["health"], "healthy") + + def test_invalid_registry_degrades_without_crashing(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "invalid.json" + path.write_text("{not-json", encoding="utf-8") + engine = ServiceAttributionEngine(path) + result = engine.attribute({"dport": 443}).to_dict() + + self.assertTrue(result["is_unknown"]) + self.assertEqual(engine.metrics()["health"], "critical") + self.assertIn( + "service_attribution_registry_error", + engine.metrics()["pressure_reasons"], + ) + def test_googlevideo_sni_is_youtube_high_confidence(self): - result = attribute_service( + result = self.result(tls_sni="r1.googlevideo.com") + self.assertEqual(result["service_name"], "YouTube") + self.assertEqual(result["attribution_confidence"], "high") + self.assertIn("tls_sni", result["attribution_sources"]) + + def test_youtube_domain_is_high_confidence(self): + result = self.result(tls_sni="youtube.com") + self.assertEqual(result["service_name"], "YouTube") + self.assertGreaterEqual(result["confidence_score"], 80) + + def test_telegram_and_github_browser_destinations(self): + telegram = self.result(http_host="web.telegram.org") + github = self.result(http_host="github.com") + self.assertEqual(telegram["service_name"], "Telegram Web") + self.assertEqual(github["service_name"], "GitHub") + self.assertEqual(github["service_category"], "Developer Platform") + + def test_discord_cdn_domain_is_attributed_conservatively(self): + result = self.result(tls_sni="cdn.discordapp.com") + self.assertEqual(result["service_name"], "Discord") + self.assertTrue(result["is_cdn"]) + + def test_dns_only_match_is_medium_confidence(self): + result = self.result(dns_qname="pypi.org", app_protocol="DNS", dport=53) + self.assertEqual(result["service_name"], "PyPI") + self.assertEqual(result["attribution_confidence"], "medium") + + def test_recent_dns_answer_correlates_destination_ip(self): + self.engine.attribute( + { + "dns_qname": "r2.googlevideo.com", + "dns_answer_ips": ["203.0.113.44"], + "ts": "2026-07-15T10:00:00Z", + "dport": 53, + } + ) + result = self.engine.attribute( { "process_name": "chrome.exe", - "tls_sni": "r1.googlevideo.com", + "dst": "203.0.113.44", "dport": 443, + "ts": "2026-07-15T10:01:00Z", } - ) + ).to_dict() self.assertEqual(result["service_name"], "YouTube") - self.assertEqual(result["service_confidence"], "high") + self.assertIn("dns", result["attribution_sources"]) - def test_google_org_without_domain_is_conservative(self): - result = attribute_service( - {"process_name": "chrome.exe", "org": "Google LLC", "sport": 443} + def test_dns_correlation_prefers_closest_observation(self): + for domain, timestamp in ( + ("github.com", "2026-07-15T10:00:00Z"), + ("pypi.org", "2026-07-15T10:04:30Z"), + ): + self.engine.attribute( + { + "dns_qname": domain, + "dns_answer_ip": "203.0.113.45", + "ts": timestamp, + "dport": 53, + } + ) + result = self.engine.attribute( + { + "dst": "203.0.113.45", + "dport": 443, + "ts": "2026-07-15T10:05:00Z", + } + ).to_dict() + self.assertEqual(result["service_name"], "PyPI") + + def test_visible_quic_server_name_reuses_safe_sni_matching(self): + result = self.result(tls_sni="", quic_server_name="googlevideo.com") + self.assertEqual(result["service_name"], "YouTube") + self.assertIn("tls_sni", result["attribution_sources"]) + + def test_http_host_is_high_confidence(self): + result = self.result(http_host="github.com", app_protocol="HTTP", dport=80) + self.assertEqual(result["attribution_confidence"], "high") + self.assertIn("HTTP Host matched github.com", result["attribution_reasons"]) + + def test_conflicting_host_and_sni_reduce_confidence(self): + result = self.result(http_host="github.com", tls_sni="googlevideo.com") + self.assertEqual(result["service_name"], "GitHub") + self.assertEqual(result["attribution_confidence"], "medium") + self.assertTrue( + any("Conflicting" in reason for reason in result["attribution_reasons"]) + ) + + def test_cloudflare_only_evidence_is_cdn_only_low_confidence(self): + result = self.result(org="Cloudflare, Inc.", tls_sni="") + self.assertEqual(result["service_name"], "CDN only") + self.assertEqual(result["attribution_confidence"], "low") + self.assertTrue(result["is_cdn"]) + + def test_browser_process_alone_is_unknown_encrypted(self): + result = self.result() + self.assertTrue(result["is_unknown"]) + self.assertEqual(result["service_name"], "Unknown encrypted destination") + self.assertEqual(result["attribution_confidence"], "low") + + def test_unencrypted_process_only_stays_unknown(self): + result = self.result(dport=80, app_protocol="TCP") + self.assertTrue(result["is_unknown"]) + self.assertEqual(result["attribution_confidence"], "unknown") + self.assertLess(result["confidence_score"], 20) + + def test_malformed_input_does_not_crash(self): + result = self.engine.attribute( + {"dport": {}, "resolved_domains": [None, {"bad": "value"}]} + ).to_dict() + self.assertTrue(result["is_unknown"]) + + def test_metrics_count_confidence_unknown_and_latency(self): + self.result(flow_id="github-flow", http_host="github.com") + self.result( + flow_id="pypi-flow", dns_qname="pypi.org", app_protocol="DNS", dport=53 ) - self.assertEqual(result["service_name"], "Google Services") - self.assertEqual(result["service_confidence"], "low") - self.assertEqual(result["service_sources"], ["asn_org"]) + self.result(flow_id="unknown-flow") + metrics = self.engine.metrics() + self.assertEqual(metrics["attributed_flows_total"], 2) + self.assertEqual(metrics["unknown_flows_total"], 1) + self.assertEqual(metrics["high_confidence_total"], 1) + self.assertEqual(metrics["medium_confidence_total"], 1) + self.assertEqual(metrics["low_confidence_total"], 1) + self.assertGreaterEqual(metrics["avg_attribution_latency_ms"], 0) + + def test_metrics_count_unique_flows_and_replace_improved_evidence(self): + self.result(flow_id="same-flow") + self.result(flow_id="same-flow") + self.assertEqual(self.engine.metrics()["unknown_flows_total"], 1) + + self.result(flow_id="same-flow", tls_sni="github.com") + metrics = self.engine.metrics() + self.assertEqual(metrics["unknown_flows_total"], 0) + self.assertEqual(metrics["attributed_flows_total"], 1) + + def test_repeated_enrichment_failures_make_health_critical(self): + with patch.object(self.engine, "_attribute", side_effect=RuntimeError): + for index in range(25): + self.engine.attribute({"flow_id": f"failure-{index}", "dport": 443}) - def test_unknown_encrypted_destination_is_not_invented(self): - result = attribute_service({"process_name": "chrome.exe", "dport": 443}) - self.assertTrue(result["service_unknown"]) - self.assertEqual(result["service_name"], "Unknown Encrypted") + metrics = self.engine.metrics() + self.assertEqual(metrics["attribution_errors_total"], 25) + self.assertEqual(metrics["health"], "critical") - def test_sensitive_domain_text_is_redacted(self): + def test_sensitive_values_are_redacted(self): + result = self.result( + tls_sni="", + org="Authorization: Bearer raw-service-secret", + ) + self.assertNotIn("raw-service-secret", json.dumps(result)) + + def test_legacy_wrapper_remains_compatible(self): result = attribute_service( - {"http_host": "example.test", "org": "token=raw-secret", "dport": 443} + {"process_name": "chrome.exe", "tls_sni": "googlevideo.com", "dport": 443} + ) + self.assertEqual(result["service_confidence"], "high") + self.assertEqual(result["service_sources"][0], "tls_sni") + + def test_flow_ring_event_and_persistence_keep_redacted_attribution(self): + packet = { + "id": "pkt-1", + "src": "10.0.0.5", + "dst": "198.51.100.20", + "sport": 52000, + "dport": 443, + "proto": "TCP", + "length": 120, + "tls_sni": "github.com", + "process_name": "chrome.exe", + } + self.engine.enrich(packet) + flow = FlowEngine().ingest(packet) + self.assertEqual(flow["service_attribution"]["service_name"], "GitHub") + + capacities = {category: 5 for category in DEFAULT_CAPACITIES} + ring = LiveRingBuffer(capacities=capacities) + ring.append("flow", flow) + self.assertEqual( + ring.query("flow")["items"][0]["payload"]["service_attribution"][ + "service_name" + ], + "GitHub", + ) + + emitted = [] + aggregator = EventAggregator(emitted.append, flow_batch_max=1) + aggregator.publish("flow:update", flow) + self.assertEqual( + emitted[0]["updates"][0]["payload"]["service_attribution"]["service_name"], + "GitHub", + ) + aggregator.close() + + written = [] + writer = BatchPersistenceWriter( + lambda grouped: written.extend(grouped.get("flow_record", [])), + enabled=False, + ) + secret_flow = dict(flow) + secret_flow["service_attribution"] = { + **flow["service_attribution"], + "secret": "raw-persistence-secret", + } + writer.enqueue("flow_record", secret_flow) + self.assertNotIn("raw-persistence-secret", json.dumps(written)) + + def test_monitoring_snapshot_includes_safe_metrics(self): + self.result(http_host="github.com") + metrics = build_monitoring_metrics( + sniffer_state={}, + observability={"service_attribution": self.engine.metrics()}, + flow_summary={}, ) - self.assertNotIn("raw-secret", str(result)) + self.assertIn("service_attribution", metrics) + self.assertGreater(metrics["service_attribution"]["registry_size"], 0) + self.assertNotIn("domain", json.dumps(metrics["service_attribution"])) if __name__ == "__main__": diff --git a/tests/test_sniffer_service.py b/tests/test_sniffer_service.py index 8dd4ad1..ee0e8d0 100644 --- a/tests/test_sniffer_service.py +++ b/tests/test_sniffer_service.py @@ -285,7 +285,9 @@ def test_processed_packet_is_available_in_redacted_live_ring(self): { "src": "10.0.0.1", "dst": "8.8.8.8", + "dport": 443, "proto": "UDP", + "tls_sni": "github.com", "authorization": "Bearer ring-secret", } ) @@ -295,6 +297,9 @@ def test_processed_packet_is_available_in_redacted_live_ring(self): service.close() self.assertEqual(len(result["items"]), 1) + attribution = result["items"][0]["payload"]["service_attribution"] + self.assertEqual(attribution["service_name"], "GitHub") + self.assertEqual(attribution["attribution_confidence"], "high") self.assertNotIn("ring-secret", str(result)) @patch(