diff --git a/.env.example b/.env.example index 970aea3..acd6414 100644 --- a/.env.example +++ b/.env.example @@ -41,6 +41,19 @@ NETBOT_FLOW_WORKER_SHUTDOWN_TIMEOUT_SEC=5 NETBOT_FLOW_WORKER_ERROR_THRESHOLD=25 NETBOT_FLOW_WORKER_SLOW_JOB_MS=100 +# Bounded, redacted recent live summaries. TTL 0 uses capacity-only eviction. +NETBOT_LIVE_RING_ENABLED=true +NETBOT_LIVE_RING_PACKET_MAX=5000 +NETBOT_LIVE_RING_FLOW_MAX=2000 +NETBOT_LIVE_RING_ALERT_MAX=1000 +NETBOT_LIVE_RING_EXPERT_MAX=1000 +NETBOT_LIVE_RING_PROTOCOL_MAX=2000 +NETBOT_LIVE_RING_AGENT_MAX=1000 +NETBOT_LIVE_RING_OPS_MAX=1000 +NETBOT_LIVE_RING_DEFAULT_QUERY_LIMIT=250 +NETBOT_LIVE_RING_MAX_QUERY_LIMIT=2000 +NETBOT_LIVE_RING_TTL_SECONDS=0 + # 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 4acbe41..4eda697 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,7 @@ or PCAP artifacts. | 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. | | Bounded Packet Intake Queue | Foundation step | Queue pressure metrics, accepted/drop counters, overflow policies, worker liveness, high-water mark, and Ops Snapshot packet queue visibility are tested. | First engine-level intake hardening step; it is separate from processing workers. | | Flow-aware Worker Pool | Foundation step | Stable per-flow lanes preserve ordering while different flows process concurrently; bounded queue pressure, latency, failures, and worker health are visible in Ops Snapshot. | This is not a benchmark claim or the complete performance engine. | +| Live Ring Buffer | Foundation step | Recent packet, flow, alert, and Expert summaries are held in per-category bounded memory with eviction/query metrics and Ops Snapshot visibility. | Stored and returned records are redacted summaries; raw packet payloads and credentials are excluded. | | Batch Persistence / Storage Backpressure | Foundation step | Redacted packet, alert, and flow records use bounded batches with queue health, write latency, retry, backlog, failure, and Ops Snapshot visibility. | Audit and report exports remain synchronous; this is not a distributed storage engine or the complete performance pipeline. | | WebSocket Event Aggregator | Foundation step | Realtime packet/alert batching, slow-client protection, WebSocket pressure metrics, and Ops Snapshot visibility are tested. | Realtime delivery batching only. | | Demo and operational QA | Ready | Token-safe demo and Agent script behavior is tested. | Demo launchers and status commands do not print raw tokens. | @@ -187,6 +188,12 @@ bidirectional flow keys select FIFO worker lanes, preserving order within a flow while allowing different flows to process concurrently. It exposes bounded backlog, drops/rejections, failures, worker liveness, and processing latency. +The Live Ring Buffer keeps a queryable window of recent redacted packet, flow, +alert, and Expert summaries. Every category has a hard capacity, oldest records +are evicted at capacity, and query limits are capped. Ops Snapshot shows memory +utilization, evictions, query pressure, and safe error types. This is a bounded +live context layer, not raw packet storage or the complete performance engine. + Packet intake queue tuning is controlled by: - `NETBOT_PACKET_QUEUE_MAX_SIZE`: default `2000`; use `1000` for small/local @@ -205,6 +212,15 @@ Packet intake queue tuning is controlled by: - `NETBOT_FLOW_WORKER_ERROR_THRESHOLD`: default `25` failures or drops before critical health. - `NETBOT_FLOW_WORKER_SLOW_JOB_MS`: default `100` milliseconds. +- `NETBOT_LIVE_RING_ENABLED`: default `true`; disables only recent in-memory + storage when set to `false`. +- `NETBOT_LIVE_RING_PACKET_MAX`, `FLOW_MAX`, `ALERT_MAX`, `EXPERT_MAX`, + `PROTOCOL_MAX`, `AGENT_MAX`, and `OPS_MAX`: per-category hard capacities; + defaults are `5000`, `2000`, `1000`, `1000`, `2000`, `1000`, and `1000`. +- `NETBOT_LIVE_RING_DEFAULT_QUERY_LIMIT` / `MAX_QUERY_LIMIT`: defaults `250` / + `2000`; oversized requests are capped and counted. +- `NETBOT_LIVE_RING_TTL_SECONDS`: default `0` for capacity-only eviction; + positive values also prune expired summaries. - `NETBOT_PERSISTENCE_BATCH_ENABLED`: batching toggle; default `true`. `false` uses compatible synchronous writes. - `NETBOT_PERSISTENCE_PACKET_BATCH_SIZE` / `PACKET_FLUSH_MS`: `500` / `1000`. diff --git a/backend/app/main.py b/backend/app/main.py index 665ade4..c0f4121 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -118,6 +118,7 @@ def _observability_snapshot() -> dict[str, Any]: "history": history_service.metrics(), "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(), "persistence": sniffer_service.persistence_stats(), "auto_block": sniffer_service.auto_block_stats(), } @@ -220,6 +221,34 @@ def api_monitoring_metrics( ) +@app.get("/api/live/recent") +def api_live_recent( + type: str = "all", + limit: int | None = None, + flow_key: str = "", + since: str | None = None, + _: None = Depends(require_trusted_client), + __: None = Depends(require_local_token), +) -> dict[str, Any]: + try: + return sniffer_service.recent_live_records( + type, + limit=limit, + flow_key=flow_key, + since=since, + ) + except (TypeError, ValueError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.get("/api/live/ring/metrics") +def api_live_ring_metrics( + _: None = Depends(require_trusted_client), + __: None = Depends(require_local_token), +) -> dict[str, Any]: + return sniffer_service.live_ring_buffer_stats() + + def _agent_headers( agent_id: str = Header("", alias="X-NetBot-Agent-Id"), agent_token: str = Header("", alias="X-NetBot-Agent-Token"), diff --git a/backend/app/services/live_ring_buffer.py b/backend/app/services/live_ring_buffer.py new file mode 100644 index 0000000..89efb46 --- /dev/null +++ b/backend/app/services/live_ring_buffer.py @@ -0,0 +1,391 @@ +from __future__ import annotations + +import os +import threading +import uuid +from collections import deque +from copy import deepcopy +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from typing import Any + +from backend.app.services.redaction import redact_sensitive_data, redact_sensitive_text + +SUPPORTED_CATEGORIES = ( + "packet", + "flow", + "alert", + "expert_info", + "protocol_metadata", + "agent_status", + "ops_event", +) + +DEFAULT_CAPACITIES = { + "packet": 5000, + "flow": 2000, + "alert": 1000, + "expert_info": 1000, + "protocol_metadata": 2000, + "agent_status": 1000, + "ops_event": 1000, +} + +ENV_CAPACITY_NAMES = { + "packet": "NETBOT_LIVE_RING_PACKET_MAX", + "flow": "NETBOT_LIVE_RING_FLOW_MAX", + "alert": "NETBOT_LIVE_RING_ALERT_MAX", + "expert_info": "NETBOT_LIVE_RING_EXPERT_MAX", + "protocol_metadata": "NETBOT_LIVE_RING_PROTOCOL_MAX", + "agent_status": "NETBOT_LIVE_RING_AGENT_MAX", + "ops_event": "NETBOT_LIVE_RING_OPS_MAX", +} + +RAW_PAYLOAD_KEYS = { + "payload_ascii", + "payload_hex", + "raw_payload", + "packet_bytes", + "pcap", + "pcap_bytes", +} + + +def _env_bool(name: str, default: bool) -> bool: + value = str(os.environ.get(name, str(default))).strip().lower() + if value in {"1", "true", "yes", "on"}: + return True + if value in {"0", "false", "no", "off"}: + return False + return default + + +def _env_int(name: str, default: int, *, minimum: int, maximum: int) -> int: + try: + value = int(os.environ.get(name, str(default))) + except (TypeError, ValueError): + return default + return value if minimum <= value <= maximum else default + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _iso_timestamp(value: str | None = None) -> str: + if not value: + return _utc_now().isoformat() + parsed = _parse_timestamp(value) + return parsed.isoformat() if parsed else _utc_now().isoformat() + + +def _parse_timestamp(value: str | None) -> datetime | None: + if not value: + return None + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except (TypeError, ValueError): + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _safe_payload(value: Any) -> Any: + redacted = redact_sensitive_data(deepcopy(value)) + if isinstance(redacted, dict): + return { + key: "" if str(key).lower() in RAW_PAYLOAD_KEYS else _safe_payload(item) + for key, item in redacted.items() + } + if isinstance(redacted, list): + return [_safe_payload(item) for item in redacted] + return redacted + + +@dataclass(frozen=True) +class LiveRingBufferRecord: + id: str + type: str + timestamp: str + flow_key: str + payload: Any + source: str + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +class LiveRingBuffer: + """Thread-safe, capacity-bounded recent live analysis storage.""" + + def __init__( + self, + *, + enabled: bool = True, + capacities: dict[str, int] | None = None, + default_query_limit: int = 250, + max_query_limit: int = 2000, + ttl_seconds: int = 0, + ) -> None: + configured = capacities or DEFAULT_CAPACITIES + self.enabled = bool(enabled) + self.capacities = { + category: max( + 1, int(configured.get(category, DEFAULT_CAPACITIES[category])) + ) + for category in SUPPORTED_CATEGORIES + } + self.max_query_limit = max(1, min(int(max_query_limit), 10_000)) + self.default_query_limit = max( + 1, min(int(default_query_limit), self.max_query_limit) + ) + self.ttl_seconds = max(0, int(ttl_seconds)) + self._buffers = {category: deque() for category in SUPPORTED_CATEGORIES} + self._evicted_by_category = {category: 0 for category in SUPPORTED_CATEGORIES} + self._lock = threading.RLock() + self._records_added_total = 0 + self._records_evicted_total = 0 + self._records_dropped_total = 0 + self._query_count_total = 0 + self._query_limit_rejected_total = 0 + self._last_added_at = "" + self._last_evicted_at = "" + self._last_error = "" + + @classmethod + def from_env(cls) -> "LiveRingBuffer": + capacities = { + category: _env_int( + ENV_CAPACITY_NAMES[category], + default, + minimum=1, + maximum=1_000_000, + ) + for category, default in DEFAULT_CAPACITIES.items() + } + max_query_limit = _env_int( + "NETBOT_LIVE_RING_MAX_QUERY_LIMIT", 2000, minimum=1, maximum=10_000 + ) + return cls( + enabled=_env_bool("NETBOT_LIVE_RING_ENABLED", True), + capacities=capacities, + default_query_limit=_env_int( + "NETBOT_LIVE_RING_DEFAULT_QUERY_LIMIT", + 250, + minimum=1, + maximum=max_query_limit, + ), + max_query_limit=max_query_limit, + ttl_seconds=_env_int( + "NETBOT_LIVE_RING_TTL_SECONDS", + 0, + minimum=0, + maximum=31_536_000, + ), + ) + + def append( + self, + category: str, + payload: Any, + *, + flow_key: str = "", + timestamp: str | None = None, + source: str = "live_capture", + ) -> dict[str, Any] | None: + if not self.enabled: + return None + if category not in self._buffers: + with self._lock: + self._records_dropped_total += 1 + self._last_error = "UnsupportedCategory" + return None + try: + record = LiveRingBufferRecord( + id=f"ring-{uuid.uuid4().hex}", + type=category, + timestamp=_iso_timestamp(timestamp), + flow_key=redact_sensitive_text(str(flow_key or ""))[:256], + payload=_safe_payload(payload), + source=redact_sensitive_text(str(source or "live_capture"))[:80], + ) + with self._lock: + self._prune_expired_locked() + buffer = self._buffers[category] + if len(buffer) >= self.capacities[category]: + buffer.popleft() + self._evicted_by_category[category] += 1 + self._records_evicted_total += 1 + self._last_evicted_at = _utc_now().isoformat() + buffer.append(record) + self._records_added_total += 1 + self._last_added_at = record.timestamp + return record.to_dict() + except Exception as exc: # pragma: no cover - defensive guard + with self._lock: + self._records_dropped_total += 1 + self._last_error = type(exc).__name__ + return None + + def query( + self, + category: str = "all", + *, + limit: int | None = None, + flow_key: str = "", + since: str | None = None, + ) -> dict[str, Any]: + requested_limit = ( + self.default_query_limit if limit is None else max(1, int(limit)) + ) + effective_limit = min(requested_limit, self.max_query_limit) + since_dt = _parse_timestamp(since) + with self._lock: + self._query_count_total += 1 + if requested_limit > self.max_query_limit: + self._query_limit_rejected_total += 1 + self._prune_expired_locked() + categories = SUPPORTED_CATEGORIES if category == "all" else (category,) + if any(item not in self._buffers for item in categories): + raise ValueError("Unsupported live ring buffer category") + records = [ + record + for item in categories + for record in reversed(self._buffers[item]) + ] + if flow_key: + records = [record for record in records if record.flow_key == flow_key] + if since_dt: + records = [ + record + for record in records + if ( + _parse_timestamp(record.timestamp) + or datetime.min.replace(tzinfo=timezone.utc) + ) + >= since_dt + ] + if category == "all": + records.sort(key=lambda record: record.timestamp, reverse=True) + truncated = ( + len(records) > effective_limit or requested_limit > effective_limit + ) + items = [record.to_dict() for record in records[:effective_limit]] + return { + "items": items, + "limit": effective_limit, + "type": category, + "truncated": truncated, + "generated_at": _utc_now().isoformat(), + } + + def snapshot(self, limit_per_category: int | None = None) -> dict[str, Any]: + limit = limit_per_category or self.default_query_limit + return { + category: self.query(category, limit=limit)["items"] + for category in SUPPORTED_CATEGORIES + } + + def clear(self) -> None: + with self._lock: + for buffer in self._buffers.values(): + buffer.clear() + + def reset(self) -> None: + with self._lock: + self.clear() + self._evicted_by_category = { + category: 0 for category in SUPPORTED_CATEGORIES + } + self._records_added_total = 0 + self._records_evicted_total = 0 + self._records_dropped_total = 0 + self._query_count_total = 0 + self._query_limit_rejected_total = 0 + self._last_added_at = "" + self._last_evicted_at = "" + self._last_error = "" + + def metrics(self) -> dict[str, Any]: + with self._lock: + self._prune_expired_locked() + categories = { + category: { + "records": len(buffer), + "capacity": self.capacities[category], + "utilization_percent": round( + len(buffer) / self.capacities[category] * 100.0, 2 + ), + "evicted_total": self._evicted_by_category[category], + } + for category, buffer in self._buffers.items() + } + total_records = sum(item["records"] for item in categories.values()) + total_capacity = sum(item["capacity"] for item in categories.values()) + utilization = round(total_records / total_capacity * 100.0, 2) + pressure_reasons = self._pressure_reasons(categories) + health = ( + "critical" + if self._last_error + else ("degraded" if pressure_reasons else "healthy") + ) + return { + "enabled": self.enabled, + "health": health, + "total_records": total_records, + "total_capacity": total_capacity, + "utilization_percent": utilization, + "records_added_total": self._records_added_total, + "records_evicted_total": self._records_evicted_total, + "records_dropped_total": self._records_dropped_total, + "query_count_total": self._query_count_total, + "query_limit_rejected_total": self._query_limit_rejected_total, + "last_added_at": self._last_added_at, + "last_evicted_at": self._last_evicted_at, + "last_error": self._last_error, + "ttl_seconds": self.ttl_seconds, + "default_query_limit": self.default_query_limit, + "max_query_limit": self.max_query_limit, + "categories": categories, + "pressure_reasons": pressure_reasons, + } + + def _pressure_reasons(self, categories: dict[str, dict[str, Any]]) -> list[str]: + reasons: list[str] = [] + if any(item["utilization_percent"] >= 90.0 for item in categories.values()): + reasons.append("live_ring_high_utilization") + frequent_evictions = any( + self._evicted_by_category[category] + >= max(10, self.capacities[category] // 10) + for category in SUPPORTED_CATEGORIES + ) + if frequent_evictions: + reasons.append("live_ring_frequent_evictions") + if self._query_limit_rejected_total: + reasons.append("live_ring_query_limit_rejections") + if self._last_error: + reasons.append("live_ring_errors") + return reasons + + def _prune_expired_locked(self) -> None: + if self.ttl_seconds <= 0: + return + cutoff = _utc_now().timestamp() - self.ttl_seconds + for category, buffer in self._buffers.items(): + while buffer: + timestamp = _parse_timestamp(buffer[0].timestamp) + if timestamp and timestamp.timestamp() >= cutoff: + break + buffer.popleft() + self._evicted_by_category[category] += 1 + self._records_evicted_total += 1 + self._last_evicted_at = _utc_now().isoformat() + + +__all__ = [ + "DEFAULT_CAPACITIES", + "LiveRingBuffer", + "LiveRingBufferRecord", + "SUPPORTED_CATEGORIES", +] diff --git a/backend/app/services/monitoring_service.py b/backend/app/services/monitoring_service.py index 199e91e..0d6dabc 100644 --- a/backend/app/services/monitoring_service.py +++ b/backend/app/services/monitoring_service.py @@ -152,6 +152,29 @@ def _safe_flow_worker_reasons(value: Any) -> list[str]: ) +def _safe_live_ring_reasons(value: Any) -> list[str]: + allowed = { + "live_ring_high_utilization", + "live_ring_frequent_evictions", + "live_ring_query_limit_rejections", + "live_ring_errors", + } + return ( + [str(reason) for reason in value if str(reason) in allowed] + if isinstance(value, list) + else [] + ) + + +def _safe_live_ring_error(value: Any) -> str: + error_type = str(value or "") + return ( + error_type + if len(error_type) <= 80 and error_type.replace("_", "").isalnum() + else "" + ) + + def build_monitoring_metrics( *, sniffer_state: dict[str, Any], @@ -163,6 +186,7 @@ def build_monitoring_metrics( event_bus = dict(observability.get("event_bus") or {}) 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 {}) event_aggregator = dict(observability.get("event_aggregator") or {}) websocket = dict(observability.get("websocket") or {}) persistence = dict(observability.get("persistence") or {}) @@ -258,6 +282,12 @@ def build_monitoring_metrics( pressure_reasons.extend( reason for reason in flow_worker_reasons if reason not in pressure_reasons ) + live_ring_reasons = _safe_live_ring_reasons( + live_ring_buffer.get("pressure_reasons") or [] + ) + pressure_reasons.extend( + reason for reason in live_ring_reasons if reason not in pressure_reasons + ) pressure_reasons.extend( reason for reason in persistence_reasons if reason not in pressure_reasons ) @@ -312,6 +342,7 @@ def build_monitoring_metrics( or websocket_send_errors >= 3 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" ): health = "critical" @@ -510,6 +541,53 @@ def build_monitoring_metrics( "last_slow_job_at": str(flow_worker_pool.get("last_slow_job_at") or ""), "pressure_reasons": flow_worker_reasons, }, + "live_ring_buffer": { + "enabled": bool(live_ring_buffer.get("enabled", False)), + "health": str(live_ring_buffer.get("health") or "healthy"), + "total_records": _int(live_ring_buffer.get("total_records")), + "total_capacity": _int(live_ring_buffer.get("total_capacity")), + "utilization_percent": _number(live_ring_buffer.get("utilization_percent")), + "records_added_total": _int(live_ring_buffer.get("records_added_total")), + "records_evicted_total": _int( + live_ring_buffer.get("records_evicted_total") + ), + "records_dropped_total": _int( + live_ring_buffer.get("records_dropped_total") + ), + "query_count_total": _int(live_ring_buffer.get("query_count_total")), + "query_limit_rejected_total": _int( + live_ring_buffer.get("query_limit_rejected_total") + ), + "last_added_at": str(live_ring_buffer.get("last_added_at") or ""), + "last_evicted_at": str(live_ring_buffer.get("last_evicted_at") or ""), + "last_error": _safe_live_ring_error(live_ring_buffer.get("last_error")), + "ttl_seconds": _int(live_ring_buffer.get("ttl_seconds")), + "default_query_limit": _int(live_ring_buffer.get("default_query_limit")), + "max_query_limit": _int(live_ring_buffer.get("max_query_limit")), + "categories": { + category: { + "records": _int(values.get("records")), + "capacity": _int(values.get("capacity")), + "utilization_percent": _number(values.get("utilization_percent")), + "evicted_total": _int(values.get("evicted_total")), + } + for category, values in dict( + live_ring_buffer.get("categories") or {} + ).items() + if category + in { + "packet", + "flow", + "alert", + "expert_info", + "protocol_metadata", + "agent_status", + "ops_event", + } + and isinstance(values, dict) + }, + "pressure_reasons": live_ring_reasons, + }, "persistence": { "enabled": bool( persistence.get("persistence_enabled", persistence.get("enabled")) diff --git a/backend/app/services/sniffer_service.py b/backend/app/services/sniffer_service.py index 180ac4d..87dabfa 100644 --- a/backend/app/services/sniffer_service.py +++ b/backend/app/services/sniffer_service.py @@ -12,6 +12,7 @@ from backend.app.services.event_bus import EventBus from backend.app.services.flow_service import FlowService 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.settings_service import get_settings_snapshot @@ -20,6 +21,7 @@ from backend.app.services.sniffer_event_publisher import SnifferEventPublisher from backend.app.services.sniffer_persistence import SnifferPersistence from core.capture import CaptureProvider, CaptureSession, SystemCaptureProvider +from core.expert_info import packet_expert_items from core.flow_engine import flow_id_for ensure_project_root_on_path() @@ -51,6 +53,7 @@ def __init__( event_bus: EventBus, capture_provider: CaptureProvider | None = None, flow_service: FlowService | None = None, + live_ring_buffer: LiveRingBuffer | None = None, ) -> None: self._lock = threading.Lock() self._event_bus = event_bus @@ -70,6 +73,7 @@ def __init__( flow_writer=flow_writer if callable(flow_writer) else None ) self._publisher = SnifferEventPublisher(event_bus) + self._live_ring_buffer = live_ring_buffer or LiveRingBuffer.from_env() self._packet_queue = BoundedPacketQueue( max_size=PACKET_QUEUE_MAX_SIZE, overflow_policy=PACKET_QUEUE_OVERFLOW_POLICY, @@ -193,6 +197,46 @@ def _process_packet(self, meta: dict[str, Any]) -> None: flow = self._flow_service.ingest(packet, alerts) self._state.add_packet(packet) self._state.add_alerts(alerts) + flow_key = str((flow or {}).get("flow_id") or flow_id_for(packet)) + self._live_ring_buffer.append( + "packet", + packet, + flow_key=flow_key, + timestamp=str(packet.get("ts") or packet.get("timestamp") or ""), + source="live_capture", + ) + if isinstance(flow, dict): + self._live_ring_buffer.append( + "flow", + flow, + flow_key=flow_key, + timestamp=str(flow.get("last_seen") or ""), + source="flow_worker_pool", + ) + for alert in alerts: + self._live_ring_buffer.append( + "alert", + alert, + flow_key=flow_key, + timestamp=str(alert.get("ts") or alert.get("timestamp") or ""), + source="detection_engine", + ) + try: + expert_items = packet_expert_items(packet, flow_key) + except Exception as exc: # pragma: no cover - defensive hot-path guard + logger.warning( + "live ring expert generation failed error_type=%s", + type(exc).__name__, + ) + expert_items = [] + for expert_item in expert_items: + self._live_ring_buffer.append( + "expert_info", + expert_item, + flow_key=flow_key, + timestamp=str(expert_item.get("last_seen") or ""), + source="expert_info", + ) self._persistence.persist(packet, alerts) if self._central_flow_persistence and isinstance(flow, dict): self._persistence.persist_flow(flow) @@ -299,6 +343,7 @@ def reset_session(self) -> dict[str, Any]: iface = self._iface self._state.reset() self._flow_service.reset() + self._live_ring_buffer.clear() state = self._state.state(running=running, iface=iface) state["observability"] = self.observability() self._publisher.publish_state("sniffer:reset", state) @@ -318,6 +363,24 @@ def packet_queue_stats(self) -> dict[str, Any]: def flow_worker_pool_stats(self) -> dict[str, Any]: return self._flow_worker_pool.stats() + def live_ring_buffer_stats(self) -> dict[str, Any]: + return self._live_ring_buffer.metrics() + + def recent_live_records( + self, + category: str = "all", + *, + limit: int | None = None, + flow_key: str = "", + since: str | None = None, + ) -> dict[str, Any]: + return self._live_ring_buffer.query( + category, + limit=limit, + flow_key=flow_key, + since=since, + ) + def auto_block_stats(self) -> dict[str, int | float]: return self._detection_pipeline.stats() @@ -326,6 +389,7 @@ def observability(self) -> dict[str, Any]: "event_bus": self._event_bus.stats(), "packet_queue": self.packet_queue_stats(), "flow_worker_pool": self.flow_worker_pool_stats(), + "live_ring_buffer": self.live_ring_buffer_stats(), "persistence": self.persistence_stats(), "auto_block": self.auto_block_stats(), } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 7babe85..b802bdc 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -53,6 +53,7 @@ flowchart TB IntakeQueue["Bounded Packet Intake Queue"] QueueWorker["Packet Queue Dispatcher"] FlowWorkers["Flow-aware Worker Pool
Ordered Flow Lanes"] + LiveRing["Live Ring Buffer
Bounded Redacted History"] EventAggregator["Event Aggregator"] Parser["Packet Parser"] Layer7["Layer 7 / TLS Metadata"] @@ -110,7 +111,7 @@ flowchart TB Events --> LiveClient CapturePolicy --> Provider Provider --> IntakeQueue --> QueueWorker --> FlowWorkers --> Parser --> Layer7 --> ProtocolIntel --> FlowEngine - FlowWorkers --> EventAggregator --> Events + Redaction --> LiveRing --> EventAggregator --> Events ProtocolIntel --> TCPIntel ProtocolIntel --> DNSIntel ProtocolIntel --> HTTPIntel @@ -226,9 +227,17 @@ parallel. Incomplete metadata uses a safe fallback lane. Worker queues remain bounded and expose backlog, utilization, failures, drops/rejections, processing latency, and liveness to `/api/monitoring/metrics` and Ops Snapshot. -The Event Aggregator remains the realtime delivery-pressure boundary after -processing, while Batch Persistence remains the storage-pressure boundary. -Live ring buffer and benchmark/soak validation remain future steps. +The Live Ring Buffer sits after processing and central redaction. It holds +separate bounded deques for recent packet, flow, alert, Expert, protocol, +Agent-status, and ops summaries. Current capture integration writes packet, +flow, alert, and Expert records. Oldest records are evicted at each category's +hard capacity, optional TTL pruning is bounded, and read APIs cap result size. +It supports Inspect/Dashboard recovery and future read-only context without +retaining raw payloads or allowing unbounded memory growth. + +The Event Aggregator remains the realtime delivery-pressure boundary after the +ring buffer, while Batch Persistence remains the storage-pressure boundary. +Benchmark/soak validation remains a future step. ## WebSocket Event Aggregator diff --git a/docs/PERFORMANCE_PIPELINE.md b/docs/PERFORMANCE_PIPELINE.md index 6764df9..32408bc 100644 --- a/docs/PERFORMANCE_PIPELINE.md +++ b/docs/PERFORMANCE_PIPELINE.md @@ -26,6 +26,7 @@ Capture -> Bounded Packet Intake Queue -> Flow-aware Worker Pool -> Existing Packet Processing +-> Live Ring Buffer -> Bounded Batch Persistence (packets, alerts, flow snapshots) -> Event Aggregator -> Batched WebSocket Updates @@ -35,9 +36,10 @@ Capture The capture callback copies packet metadata into `BoundedPacketQueue` and returns quickly. Its dispatcher sends accepted metadata to bounded flow-aware worker lanes. Each lane runs the existing processing path: payload policy, -protocol metadata, flow ingestion, detection, dashboard state, persistence -enqueue, and websocket publishing. The Event Aggregator then batches -high-frequency realtime updates before websocket fan-out. +protocol metadata, flow ingestion, detection, and dashboard state. Redacted +summaries then enter the Live Ring Buffer before persistence enqueue and +websocket publishing. The Event Aggregator batches high-frequency realtime +updates before websocket fan-out. ## Environment Variables @@ -452,13 +454,74 @@ Live frontend buffers remain bounded: Table virtualization remains a later step. +## Live Ring Buffer + +`LiveRingBuffer` provides a bounded, thread-safe window of recent live analysis +data after packet/flow/DPI processing and central redaction. It prevents the +Inspect and operational read paths from depending on an unbounded in-memory +list while preserving recent context across WebSocket bursts. + +Current capture integration stores four useful categories: + +- packet summaries, maximum `5000`; +- flow updates, maximum `2000`; +- alerts, maximum `1000`; +- Expert Info records, maximum `1000`. + +Protocol metadata, Agent status, and ops event buffers are configured and +bounded for compatible internal use, but are not force-fed by this step. Their +defaults are `2000`, `1000`, and `1000` records. All capacities have safe +fallbacks and can never become unlimited. + +When a category reaches capacity, the oldest record is evicted before the new +record is appended. `NETBOT_LIVE_RING_TTL_SECONDS=0` means capacity-only +eviction; a positive value also prunes expired records. Append and query +operations use a short reentrant lock and copy/redact payloads before storage. +Raw payload fields are removed from ring records even when capture policy +permits a guarded forensic workflow. + +Read-only endpoints: + +- `GET /api/live/recent` supports bounded `type`, `limit`, `flow_key`, and + `since` queries; +- `GET /api/live/ring/metrics` exposes counters and safe health fields; +- `GET /api/monitoring/metrics` includes the same `live_ring_buffer` section. + +Both endpoints use the existing trusted-client and local-token dependencies. +Oversized limits are capped at `NETBOT_LIVE_RING_MAX_QUERY_LIMIT` and counted. +The Ring Buffer complements rather than replaces the Event Aggregator: +WebSocket batching controls delivery pressure, while the ring controls recent +in-memory history. Batch Persistence still owns durable storage pressure, and +the Flow-aware Worker Pool still owns parallel processing pressure. + +Environment variables: + +| Variable | Default | Purpose | +| --- | --- | --- | +| `NETBOT_LIVE_RING_ENABLED` | `true` | Enable bounded recent live storage. | +| `NETBOT_LIVE_RING_PACKET_MAX` | `5000` | Packet summary capacity. | +| `NETBOT_LIVE_RING_FLOW_MAX` | `2000` | Flow update capacity. | +| `NETBOT_LIVE_RING_ALERT_MAX` | `1000` | Alert capacity. | +| `NETBOT_LIVE_RING_EXPERT_MAX` | `1000` | Expert Info capacity. | +| `NETBOT_LIVE_RING_PROTOCOL_MAX` | `2000` | Protocol summary capacity. | +| `NETBOT_LIVE_RING_AGENT_MAX` | `1000` | Agent status capacity. | +| `NETBOT_LIVE_RING_OPS_MAX` | `1000` | Operational event capacity. | +| `NETBOT_LIVE_RING_DEFAULT_QUERY_LIMIT` | `250` | Default recent-result limit. | +| `NETBOT_LIVE_RING_MAX_QUERY_LIMIT` | `2000` | Hard API result cap. | +| `NETBOT_LIVE_RING_TTL_SECONDS` | `0` | Optional TTL; zero disables TTL pruning. | + +Metrics include enabled/health state, total records/capacity/utilization, +added/evicted/dropped totals, query and query-limit counters, safe timestamps, +safe error type, per-category utilization, and fixed pressure reason enums. +High utilization, frequent evictions, or capped queries degrade health; an +internal read/write error is critical and contributes to overall Ops health. + ## Current Limitations This is not the complete performance engine yet. Not implemented yet: -- Live Ring Buffer; - Benchmark / Soak Tests; - Optional ClickHouse or external metrics backend. @@ -469,12 +532,11 @@ behavior, or AI autonomous actions. ## Next Planned Steps -1. Live Ring Buffer -2. Benchmark and Soak Tests -3. Performance Validation Report -4. Service Attribution / Destination Intelligence -5. Incident / Correlation Engine -6. Read-only AI Analyst +1. Benchmark and Soak Tests +2. Performance Validation Report +3. Service Attribution / Destination Intelligence +4. Incident / Correlation Engine +5. Read-only AI Analyst ### Recorded Product Direction diff --git a/frontend/src/components/OpsPanel.jsx b/frontend/src/components/OpsPanel.jsx index 282f46b..0617cc4 100644 --- a/frontend/src/components/OpsPanel.jsx +++ b/frontend/src/components/OpsPanel.jsx @@ -37,6 +37,7 @@ export function OpsPanel({ observability, operationalMetrics = null, isRefreshin const packetQueueLastDropReason = snapshot.safeLastDropReason || "No drops recorded"; const flowWorkerPool = snapshot.flowWorkerPool; const flowWorkerLastDropReason = snapshot.safeFlowWorkerDropReason || "No drops recorded"; + const liveRingBuffer = snapshot.liveRingBuffer; const persistence = snapshot.persistence; const eventBus = snapshot.eventBus; const eventAggregator = snapshot.eventAggregator; @@ -119,6 +120,41 @@ export function OpsPanel({ observability, operationalMetrics = null, isRefreshin + +
+ + + + + 0 ? "degraded" : "healthy"} /> + 0 ? "warning" : "healthy"} /> + + + + +
+ {Object.keys(liveRingBuffer.categories || {}).length ? ( +
+ {Object.entries(liveRingBuffer.categories).map(([category, values]) => ( +
+
+ {category.replaceAll("_", " ")} + {Number(values.utilization_percent || 0).toFixed(1)}% used +
+ {values.records || 0}/{values.capacity || 0} records + {values.evicted_total || 0} evicted +
+ ))} +
+ ) : null} + + { { worker_id: 0, worker_alive: true, queue_depth: 0, queue_max: 500, processed_total: 12 }, ], }, + live_ring_buffer: { + enabled: true, + health: "healthy", + total_records: 12, + total_capacity: 100, + utilization_percent: 12, + records_added_total: 20, + categories: { + packet: { records: 10, capacity: 80, utilization_percent: 12.5, evicted_total: 0 }, + alert: { records: 2, capacity: 20, utilization_percent: 10, evicted_total: 0 }, + }, + }, event_aggregator: { packet_batch_ms: 500, packet_batch_max: 250, @@ -88,6 +100,8 @@ describe("OpsPanel", () => { expect(screen.getByText("Capture and Flow Pressure")).toBeTruthy(); expect(screen.getByText("Packet Intake Queue")).toBeTruthy(); expect(screen.getByText("Flow Worker Pool")).toBeTruthy(); + expect(screen.getByText("Live Ring Buffer")).toBeTruthy(); + expect(screen.getByText("packet")).toBeTruthy(); expect(screen.getByText("Worker 0")).toBeTruthy(); expect(screen.getByText("WebSocket Event Aggregator")).toBeTruthy(); expect(screen.getByText("Batches Sent")).toBeTruthy(); @@ -599,4 +613,87 @@ describe("OpsPanel", () => { expect(screen.getByText("A flow worker appears unhealthy. Restart capture or inspect backend runtime logs.")).toBeTruthy(); expect(document.body.textContent).not.toMatch(/raw-token|authorization/i); }); + + it.each(["healthy", "degraded", "critical"])( + "renders live ring buffer %s health", + (health) => { + render( + + ); + + expect(screen.getByText("Live Ring Buffer")).toBeTruthy(); + expect(screen.getAllByText(`Health ${health}`).length).toBeGreaterThan(0); + } + ); + + it("renders disabled live ring buffer state", () => { + render( + + ); + + expect(screen.getByText("Live Ring Buffer")).toBeTruthy(); + expect(screen.getAllByText("Disabled").length).toBeGreaterThan(0); + }); + + it("renders live ring pressure actions and sanitizes diagnostics", () => { + render( + + ); + + expect(screen.getByText("Live ring buffer is near capacity. Increase category caps or reduce live capture pressure.")).toBeTruthy(); + expect(screen.getByText("Live ring buffer is evicting old records frequently. This is safe but recent history may be shorter than expected.")).toBeTruthy(); + 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); + }); }); diff --git a/frontend/src/lib/opsHealth.js b/frontend/src/lib/opsHealth.js index 9ff14c7..f25331e 100644 --- a/frontend/src/lib/opsHealth.js +++ b/frontend/src/lib/opsHealth.js @@ -84,6 +84,21 @@ function safeFlowWorkerReasons(value) { return Array.isArray(value) ? value.filter((reason) => allowed.has(reason)) : []; } +function safeLiveRingError(value) { + const errorType = String(value || ""); + return errorType.length <= 80 && /^[A-Za-z0-9_]+$/.test(errorType) ? errorType : ""; +} + +function safeLiveRingReasons(value) { + const allowed = new Set([ + "live_ring_high_utilization", + "live_ring_frequent_evictions", + "live_ring_query_limit_rejections", + "live_ring_errors", + ]); + return Array.isArray(value) ? value.filter((reason) => allowed.has(reason)) : []; +} + export function buildOpsSnapshot(observability, operationalMetrics = null) { const eventBus = observability?.event_bus || {}; const eventAggregator = operationalMetrics?.event_aggregator || observability?.event_aggregator || {}; @@ -96,6 +111,34 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { last_drop_reason: safeFlowWorkerDropReason(flowWorkerPool.last_drop_reason), pressure_reasons: safeFlowWorkerReasons(flowWorkerPool.pressure_reasons), }; + const liveRingBuffer = operationalMetrics?.live_ring_buffer || observability?.live_ring_buffer || {}; + const safeLiveRingCategories = Object.fromEntries( + Object.entries(liveRingBuffer.categories || {}) + .filter(([category]) => ["packet", "flow", "alert", "expert_info", "protocol_metadata", "agent_status", "ops_event"].includes(category)) + .map(([category, values]) => [category, { + records: toNumber(values?.records), + capacity: toNumber(values?.capacity), + utilization_percent: toNumber(values?.utilization_percent), + evicted_total: toNumber(values?.evicted_total), + }]), + ); + const safeLiveRingBuffer = { + enabled: liveRingBuffer.enabled === true, + health: ["healthy", "degraded", "critical"].includes(liveRingBuffer.health) ? liveRingBuffer.health : "healthy", + total_records: toNumber(liveRingBuffer.total_records), + total_capacity: toNumber(liveRingBuffer.total_capacity), + utilization_percent: toNumber(liveRingBuffer.utilization_percent), + records_added_total: toNumber(liveRingBuffer.records_added_total), + records_evicted_total: toNumber(liveRingBuffer.records_evicted_total), + records_dropped_total: toNumber(liveRingBuffer.records_dropped_total), + query_count_total: toNumber(liveRingBuffer.query_count_total), + query_limit_rejected_total: toNumber(liveRingBuffer.query_limit_rejected_total), + last_added_at: String(liveRingBuffer.last_added_at || ""), + last_evicted_at: String(liveRingBuffer.last_evicted_at || ""), + last_error: safeLiveRingError(liveRingBuffer.last_error), + pressure_reasons: safeLiveRingReasons(liveRingBuffer.pressure_reasons), + categories: safeLiveRingCategories, + }; const persistence = operationalMetrics?.persistence || observability?.persistence || {}; const history = observability?.history || {}; const autoBlock = observability?.auto_block || {}; @@ -128,6 +171,10 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { const flowWorkerRejected = toNumber(flowWorkerPool.jobs_rejected_total); const flowWorkerSlowJobs = toNumber(flowWorkerPool.slow_jobs_total); const flowWorkerP95 = toNumber(flowWorkerPool.p95_processing_latency_ms); + const liveRingEnabled = safeLiveRingBuffer.enabled; + const liveRingUtilization = safeLiveRingBuffer.utilization_percent; + const liveRingEvicted = safeLiveRingBuffer.records_evicted_total; + const liveRingQueryRejected = safeLiveRingBuffer.query_limit_rejected_total; const droppedWrites = toNumber(persistence.events_dropped_total ?? persistence.dropped_writes); const failedWrites = toNumber(persistence.events_failed_total ?? persistence.failed_writes); const persistenceUtilization = toNumber(persistence.utilization_percent ?? persistence.queue_utilization_percent); @@ -198,10 +245,19 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { ? "warning" : "healthy" : "healthy"; + const liveRingLevel = liveRingEnabled + ? normalizeLevel(safeLiveRingBuffer.health) !== "healthy" + ? normalizeLevel(safeLiveRingBuffer.health) + : safeLiveRingBuffer.last_error || safeLiveRingBuffer.records_dropped_total > 0 + ? "degraded" + : liveRingUtilization >= 90 || liveRingQueryRejected > 0 + ? "warning" + : "healthy" + : "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, persistenceLevel, streamLevel, queryLevel, autoBlockLevel); + const overall = worstLevel(backendLevel, freshnessLevel, packetQueueLevel, flowWorkerLevel, liveRingLevel, persistenceLevel, streamLevel, queryLevel, autoBlockLevel); const recommendedActions = []; if (backendLevel !== "healthy") { @@ -237,6 +293,18 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { if (flowWorkerDropped > 0 || flowWorkerRejected > 0) { recommendedActions.push("Flow worker jobs were dropped due to processing pressure. Review worker queue size and overflow policy."); } + if (liveRingEnabled && safeLiveRingBuffer.pressure_reasons.includes("live_ring_high_utilization")) { + recommendedActions.push("Live ring buffer is near capacity. Increase category caps or reduce live capture pressure."); + } + if (liveRingEnabled && safeLiveRingBuffer.pressure_reasons.includes("live_ring_frequent_evictions")) { + recommendedActions.push("Live ring buffer is evicting old records frequently. This is safe but recent history may be shorter than expected."); + } + if (liveRingEnabled && liveRingQueryRejected > 0) { + recommendedActions.push("Live ring buffer query limit was capped. Reduce requested result size or inspect a narrower time range."); + } + if (liveRingEnabled && safeLiveRingBuffer.last_error) { + recommendedActions.push("Live ring buffer reported errors. Inspect backend logs and recent live capture activity."); + } if (criticalFlows > 0) { recommendedActions.push("Review critical flows and related alerts first."); } else if (highFlows > 0) { @@ -325,6 +393,12 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { hint: `${flowWorkerDepth}/${flowWorkerMax} queued | P95 ${formatMs(flowWorkerP95)}`, level: flowWorkerLevel, }, + { + label: "Live Ring", + value: liveRingEnabled ? `${safeLiveRingBuffer.total_records}/${safeLiveRingBuffer.total_capacity}` : "Disabled", + hint: `${liveRingUtilization.toFixed(1)}% used | Evicted ${liveRingEvicted}`, + level: liveRingLevel, + }, { label: "Write Queue", value: String(queueSize), @@ -388,6 +462,8 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { flowWorkerPool: safeFlowWorkerPool, flowWorkerLevel, safeFlowWorkerDropReason: safeFlowWorkerPool.last_drop_reason, + liveRingBuffer: safeLiveRingBuffer, + liveRingLevel, persistence, history, autoBlock, diff --git a/tests/test_live_ring_buffer.py b/tests/test_live_ring_buffer.py new file mode 100644 index 0000000..1df421d --- /dev/null +++ b/tests/test_live_ring_buffer.py @@ -0,0 +1,292 @@ +from __future__ import annotations + +import json +import os +import threading +import unittest +from unittest.mock import patch + +from fastapi.testclient import TestClient + +from backend.app.main import app, sniffer_service +from backend.app.security import require_local_token, require_trusted_client +from backend.app.services.live_ring_buffer import ( + DEFAULT_CAPACITIES, + LiveRingBuffer, +) +from backend.app.services.monitoring_service import build_monitoring_metrics + + +class LiveRingBufferTests(unittest.TestCase): + def make_buffer(self, **overrides): + capacities = {category: 3 for category in DEFAULT_CAPACITIES} + capacities.update(overrides.pop("capacities", {})) + return LiveRingBuffer(capacities=capacities, **overrides) + + def test_creates_all_category_buffers_with_configured_capacity(self): + ring = self.make_buffer(capacities={"packet": 2, "flow": 4}) + + metrics = ring.metrics() + + self.assertEqual(metrics["categories"]["packet"]["capacity"], 2) + self.assertEqual(metrics["categories"]["flow"]["capacity"], 4) + self.assertEqual(len(metrics["categories"]), 7) + + def test_appends_supported_live_records(self): + ring = self.make_buffer() + + for category in ("packet", "flow", "alert", "expert_info"): + ring.append(category, {"id": category}, flow_key="flow-1") + + self.assertEqual(ring.metrics()["total_records"], 4) + self.assertEqual(ring.query("flow")["items"][0]["flow_key"], "flow-1") + + def test_evicts_oldest_and_tracks_counter(self): + ring = self.make_buffer(capacities={"packet": 2}) + ring.append("packet", {"id": 1}) + ring.append("packet", {"id": 2}) + ring.append("packet", {"id": 3}) + + items = ring.query("packet", limit=10)["items"] + metrics = ring.metrics() + + self.assertEqual([item["payload"]["id"] for item in items], [3, 2]) + self.assertEqual(metrics["records_evicted_total"], 1) + self.assertEqual(metrics["categories"]["packet"]["evicted_total"], 1) + + def test_query_by_type_flow_key_since_and_latest_order(self): + ring = self.make_buffer() + ring.append( + "packet", + {"id": "old"}, + flow_key="flow-a", + timestamp="2026-01-01T00:00:00Z", + ) + ring.append( + "packet", + {"id": "new"}, + flow_key="flow-a", + timestamp="2026-01-02T00:00:00Z", + ) + ring.append("packet", {"id": "other"}, flow_key="flow-b") + + result = ring.query( + "packet", + flow_key="flow-a", + since="2026-01-01T12:00:00Z", + ) + + self.assertEqual([item["payload"]["id"] for item in result["items"]], ["new"]) + + def test_query_limit_is_capped_and_visible(self): + ring = self.make_buffer(default_query_limit=2, max_query_limit=2) + for index in range(3): + ring.append("packet", {"id": index}) + + result = ring.query("packet", limit=999) + + self.assertEqual(result["limit"], 2) + self.assertTrue(result["truncated"]) + self.assertEqual(ring.metrics()["query_limit_rejected_total"], 1) + self.assertIn( + "live_ring_query_limit_rejections", + ring.metrics()["pressure_reasons"], + ) + + def test_disabled_ring_does_not_store_records(self): + ring = self.make_buffer(enabled=False) + + self.assertIsNone(ring.append("packet", {"id": 1})) + self.assertEqual(ring.metrics()["total_records"], 0) + + def test_ttl_prunes_expired_records(self): + ring = self.make_buffer(ttl_seconds=60) + ring.append("packet", {"id": "expired"}, timestamp="2020-01-01T00:00:00Z") + + self.assertEqual(ring.query("packet")["items"], []) + self.assertEqual(ring.metrics()["records_evicted_total"], 1) + + def test_invalid_env_values_fall_back_to_safe_defaults(self): + with patch.dict( + os.environ, + { + "NETBOT_LIVE_RING_PACKET_MAX": "unlimited", + "NETBOT_LIVE_RING_DEFAULT_QUERY_LIMIT": "bad", + "NETBOT_LIVE_RING_MAX_QUERY_LIMIT": "0", + "NETBOT_LIVE_RING_TTL_SECONDS": "-1", + }, + clear=False, + ): + ring = LiveRingBuffer.from_env() + + self.assertEqual(ring.capacities["packet"], 5000) + self.assertEqual(ring.default_query_limit, 250) + self.assertEqual(ring.max_query_limit, 2000) + self.assertEqual(ring.ttl_seconds, 0) + + def test_records_are_redacted_and_raw_payload_fields_removed(self): + ring = self.make_buffer() + secret = "ring-secret-value" + + ring.append( + "packet", + { + "authorization": f"Bearer {secret}", + "cookie": f"session={secret}", + "path": f"/login?token={secret}", + "payload_ascii": secret, + "nested": {"password": secret}, + }, + ) + + serialized = json.dumps(ring.query("packet")) + self.assertNotIn(secret, serialized) + self.assertIn("[REDACTED]", serialized) + self.assertEqual( + ring.query("packet")["items"][0]["payload"]["payload_ascii"], "" + ) + + def test_metrics_never_expose_payload_values(self): + ring = self.make_buffer() + ring.append("alert", {"secret": "do-not-leak"}) + + serialized = json.dumps(ring.metrics()) + + self.assertNotIn("do-not-leak", serialized) + self.assertNotIn("payload", serialized) + + def test_health_transitions_for_utilization_evictions_and_errors(self): + ring = self.make_buffer(capacities={"packet": 2}) + self.assertEqual(ring.metrics()["health"], "healthy") + + ring.append("packet", {"id": 1}) + ring.append("packet", {"id": 2}) + self.assertEqual(ring.metrics()["health"], "degraded") + self.assertIn("live_ring_high_utilization", ring.metrics()["pressure_reasons"]) + + self.assertIsNone(ring.append("unsupported", {})) + self.assertEqual(ring.metrics()["health"], "critical") + self.assertEqual(ring.metrics()["last_error"], "UnsupportedCategory") + + def test_concurrent_append_and_query_is_safe(self): + ring = self.make_buffer(capacities={"packet": 50}) + errors = [] + + def writer(offset): + try: + for index in range(200): + ring.append("packet", {"id": offset + index}) + except Exception as exc: # pragma: no cover - assertion capture + errors.append(exc) + + def reader(): + try: + for _ in range(200): + ring.query("packet", limit=20) + except Exception as exc: # pragma: no cover - assertion capture + errors.append(exc) + + threads = [threading.Thread(target=writer, args=(i * 1000,)) for i in range(3)] + threads.append(threading.Thread(target=reader)) + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual(errors, []) + self.assertLessEqual(ring.metrics()["categories"]["packet"]["records"], 50) + + def test_clear_and_reset(self): + ring = self.make_buffer(capacities={"packet": 1}) + ring.append("packet", {"id": 1}) + ring.append("packet", {"id": 2}) + ring.clear() + self.assertEqual(ring.metrics()["total_records"], 0) + self.assertEqual(ring.metrics()["records_evicted_total"], 1) + + ring.reset() + self.assertEqual(ring.metrics()["records_evicted_total"], 0) + self.assertEqual(ring.metrics()["records_added_total"], 0) + + +class LiveRingMonitoringTests(unittest.TestCase): + def test_monitoring_metrics_include_ring_and_pressure(self): + metrics = build_monitoring_metrics( + sniffer_state={}, + observability={ + "live_ring_buffer": { + "enabled": True, + "health": "degraded", + "total_records": 9, + "total_capacity": 10, + "utilization_percent": 90, + "records_added_total": 12, + "records_evicted_total": 2, + "records_dropped_total": 0, + "query_count_total": 4, + "query_limit_rejected_total": 1, + "last_added_at": "2026-01-01T00:00:00+00:00", + "last_evicted_at": "2026-01-01T00:00:01+00:00", + "last_error": "", + "categories": { + "packet": { + "records": 9, + "capacity": 10, + "utilization_percent": 90, + "evicted_total": 2, + } + }, + "pressure_reasons": ["live_ring_high_utilization"], + } + }, + flow_summary={}, + ) + + ring = metrics["live_ring_buffer"] + self.assertEqual(ring["total_records"], 9) + self.assertEqual(ring["categories"]["packet"]["capacity"], 10) + self.assertIn("live_ring_high_utilization", metrics["pressure_reasons"]) + self.assertEqual(metrics["health"], "degraded") + + +class LiveRingApiTests(unittest.TestCase): + def setUp(self): + app.dependency_overrides[require_trusted_client] = lambda: None + app.dependency_overrides[require_local_token] = lambda: None + self.client = TestClient(app) + self.original_ring = sniffer_service._live_ring_buffer + sniffer_service._live_ring_buffer = LiveRingBuffer( + capacities={category: 3 for category in DEFAULT_CAPACITIES}, + max_query_limit=2, + ) + + def tearDown(self): + sniffer_service._live_ring_buffer = self.original_ring + app.dependency_overrides.clear() + + def test_recent_endpoint_returns_only_redacted_bounded_records(self): + sniffer_service._live_ring_buffer.append( + "packet", + {"id": "packet-1", "authorization": "Bearer api-secret"}, + flow_key="flow-1", + ) + + response = self.client.get("/api/live/recent?type=packet&limit=999") + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["limit"], 2) + self.assertNotIn("api-secret", response.text) + + def test_recent_endpoint_rejects_unknown_category(self): + response = self.client.get("/api/live/recent?type=unknown") + self.assertEqual(response.status_code, 400) + + def test_ring_metrics_endpoint_and_monitoring_snapshot(self): + self.assertEqual(self.client.get("/api/live/ring/metrics").status_code, 200) + response = self.client.get("/api/monitoring/metrics") + self.assertEqual(response.status_code, 200) + self.assertIn("live_ring_buffer", response.json()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_release_readiness.py b/tests/test_release_readiness.py index 3e98651..d4660f9 100644 --- a/tests/test_release_readiness.py +++ b/tests/test_release_readiness.py @@ -112,6 +112,11 @@ def test_performance_pipeline_docs_are_linked_and_scoped(self): self.assertIn("Flow-aware Worker Pool", performance) self.assertIn("NETBOT_FLOW_WORKER_COUNT", performance) self.assertIn("Flow-aware Worker Pool", architecture) + self.assertIn("Live Ring Buffer", readme) + self.assertIn("Live Ring Buffer", performance) + self.assertIn("Live Ring Buffer", architecture) + self.assertIn("NETBOT_LIVE_RING_PACKET_MAX", performance) + self.assertIn("GET /api/live/recent", performance) self.assertIn("Queue pressure metrics", readme) self.assertIn("Ops Snapshot packet queue visibility", readme) self.assertIn("NETBOT_PACKET_QUEUE_MAX_SIZE", readme) diff --git a/tests/test_sniffer_service.py b/tests/test_sniffer_service.py index 232dca8..8dd4ad1 100644 --- a/tests/test_sniffer_service.py +++ b/tests/test_sniffer_service.py @@ -3,6 +3,7 @@ from unittest.mock import patch from backend.app.services.event_bus import EventBus +from backend.app.services.live_ring_buffer import DEFAULT_CAPACITIES, LiveRingBuffer from backend.app.services.sniffer_service import ( CaptureStartUnavailableError, SnifferService, @@ -269,6 +270,56 @@ def test_disabled_flow_workers_use_existing_packet_processing_path(self): finally: service.close() + def test_processed_packet_is_available_in_redacted_live_ring(self): + ring = LiveRingBuffer( + capacities={category: 5 for category in DEFAULT_CAPACITIES} + ) + service = SnifferService( + EventBus(), + capture_provider=_FakeCaptureProvider(), + flow_service=_FakeFlowService(), + live_ring_buffer=ring, + ) + try: + service._on_packet( + { + "src": "10.0.0.1", + "dst": "8.8.8.8", + "proto": "UDP", + "authorization": "Bearer ring-secret", + } + ) + self.assertTrue(service.drain_packet_queue(timeout_sec=1.0)) + result = service.recent_live_records("packet", limit=5) + finally: + service.close() + + self.assertEqual(len(result["items"]), 1) + self.assertNotIn("ring-secret", str(result)) + + @patch( + "backend.app.services.sniffer_service.packet_expert_items", + side_effect=ValueError("Authorization: Bearer should-not-log"), + ) + def test_expert_generation_failure_does_not_stop_packet_processing( + self, _mock_expert + ): + service = SnifferService( + EventBus(), + capture_provider=_FakeCaptureProvider(), + flow_service=_FakeFlowService(), + ) + try: + service._on_packet({"src": "10.0.0.1", "dst": "8.8.8.8", "proto": "UDP"}) + self.assertTrue(service.drain_packet_queue(timeout_sec=1.0)) + result = service.recent_live_records("packet", limit=5) + stats = service.packet_queue_stats() + finally: + service.close() + + self.assertEqual(len(result["items"]), 1) + self.assertTrue(stats["worker_alive"]) + if __name__ == "__main__": unittest.main()