diff --git a/backend/.env.example b/backend/.env.example new file mode 100644 index 0000000..629b798 --- /dev/null +++ b/backend/.env.example @@ -0,0 +1,22 @@ +BACKEND_DB_DSN=postgresql://mitm_user:mitm_password@localhost:5432/mitm_db +BACKEND_LOG_LEVEL=DEBUG +BACKEND_DB_POOL_MIN_SIZE=1 +BACKEND_DB_POOL_MAX_SIZE=5 +BACKEND_BROADCAST_QUEUE_MAXSIZE=1024 +BACKEND_PACKET_TRACKER_FINALIZE_DELAY_SECONDS=0.25 +BACKEND_PACKET_TRACKER_RETENTION_SECONDS=10.0 +BACKEND_PACKET_TRACKER_MIN_FLUSH_INTERVAL_SECONDS=0.05 +BACKEND_PACKET_TRACKER_PERSIST_TIMEOUT_SECONDS=2.0 +BACKEND_PACKET_TRACKER_STOP_JOIN_TIMEOUT_SECONDS=2.0 +BACKEND_PACKET_TRACKER_REJECT_CORRELATION_WINDOW_SECONDS=1.0 +BACKEND_SNIFFER_BUFFER_CAPACITY=20000 +BACKEND_SNIFFER_SOCKET_RCVBUF_BYTES=4194304 +BACKEND_SNIFFER_SELECTOR_TIMEOUT_SECONDS=1.0 +BACKEND_SNIFFER_RECV_BYTES=65536 +BACKEND_SNIFFER_BUFFER_DRAIN_INTERVAL_SECONDS=5.0 +BACKEND_SNIFFER_THREAD_JOIN_TIMEOUT_SECONDS=2.0 +BACKEND_TELEMETRY_PROCESS_STOP_TIMEOUT_SECONDS=3.0 +BACKEND_TELEMETRY_READER_JOIN_TIMEOUT_SECONDS=2.0 +BACKEND_NDPI_MAX_FLOWS=200000 +BACKEND_NDPI_FLOW_TTL_SECONDS=120.0 +BACKEND_NDPI_CLEANUP_INTERVAL_PACKETS=10000 diff --git a/backend/src/config.py b/backend/src/config.py new file mode 100644 index 0000000..36db3c7 --- /dev/null +++ b/backend/src/config.py @@ -0,0 +1,84 @@ +"""Runtime configuration for the backend service.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + + +def _env_str(name: str, default: str) -> str: + value = os.getenv(name) + return value if value not in (None, "") else default + + +def _env_int(name: str, default: int) -> int: + value = os.getenv(name) + if value in (None, ""): + return default + return int(value) + + +def _env_float(name: str, default: float) -> float: + value = os.getenv(name) + if value in (None, ""): + return default + return float(value) + + +@dataclass(frozen=True) +class BackendSettings: + db_dsn: str + log_level: str + db_pool_min_size: int + db_pool_max_size: int + broadcaster_queue_maxsize: int + packet_tracker_finalize_delay_seconds: float + packet_tracker_retention_seconds: float + packet_tracker_min_flush_interval_seconds: float + packet_tracker_persist_timeout_seconds: float + packet_tracker_stop_join_timeout_seconds: float + packet_tracker_reject_correlation_window_seconds: float + sniffer_buffer_capacity: int + sniffer_socket_rcvbuf_bytes: int + sniffer_selector_timeout_seconds: float + sniffer_recv_bytes: int + sniffer_buffer_drain_interval_seconds: float + sniffer_thread_join_timeout_seconds: float + telemetry_process_stop_timeout_seconds: float + telemetry_reader_join_timeout_seconds: float + ndpi_max_flows: int + ndpi_flow_ttl_seconds: float + ndpi_cleanup_interval_packets: int + + +def load_settings() -> BackendSettings: + return BackendSettings( + db_dsn=_env_str("BACKEND_DB_DSN", "postgresql://mitm_user:mitm_password@localhost:5432/mitm_db"), + log_level=_env_str("BACKEND_LOG_LEVEL", "DEBUG"), + db_pool_min_size=_env_int("BACKEND_DB_POOL_MIN_SIZE", 1), + db_pool_max_size=_env_int("BACKEND_DB_POOL_MAX_SIZE", 5), + broadcaster_queue_maxsize=_env_int("BACKEND_BROADCAST_QUEUE_MAXSIZE", 1024), + packet_tracker_finalize_delay_seconds=_env_float("BACKEND_PACKET_TRACKER_FINALIZE_DELAY_SECONDS", 0.25), + packet_tracker_retention_seconds=_env_float("BACKEND_PACKET_TRACKER_RETENTION_SECONDS", 10.0), + packet_tracker_min_flush_interval_seconds=_env_float("BACKEND_PACKET_TRACKER_MIN_FLUSH_INTERVAL_SECONDS", 0.05), + packet_tracker_persist_timeout_seconds=_env_float("BACKEND_PACKET_TRACKER_PERSIST_TIMEOUT_SECONDS", 2.0), + packet_tracker_stop_join_timeout_seconds=_env_float("BACKEND_PACKET_TRACKER_STOP_JOIN_TIMEOUT_SECONDS", 2.0), + packet_tracker_reject_correlation_window_seconds=_env_float( + "BACKEND_PACKET_TRACKER_REJECT_CORRELATION_WINDOW_SECONDS", + 1.0, + ), + sniffer_buffer_capacity=_env_int("BACKEND_SNIFFER_BUFFER_CAPACITY", 20_000), + sniffer_socket_rcvbuf_bytes=_env_int("BACKEND_SNIFFER_SOCKET_RCVBUF_BYTES", 4 * 1024 * 1024), + sniffer_selector_timeout_seconds=_env_float("BACKEND_SNIFFER_SELECTOR_TIMEOUT_SECONDS", 1.0), + sniffer_recv_bytes=_env_int("BACKEND_SNIFFER_RECV_BYTES", 65_536), + sniffer_buffer_drain_interval_seconds=_env_float("BACKEND_SNIFFER_BUFFER_DRAIN_INTERVAL_SECONDS", 5.0), + sniffer_thread_join_timeout_seconds=_env_float("BACKEND_SNIFFER_THREAD_JOIN_TIMEOUT_SECONDS", 2.0), + telemetry_process_stop_timeout_seconds=_env_float("BACKEND_TELEMETRY_PROCESS_STOP_TIMEOUT_SECONDS", 3.0), + telemetry_reader_join_timeout_seconds=_env_float("BACKEND_TELEMETRY_READER_JOIN_TIMEOUT_SECONDS", 2.0), + ndpi_max_flows=_env_int("BACKEND_NDPI_MAX_FLOWS", 200_000), + ndpi_flow_ttl_seconds=_env_float("BACKEND_NDPI_FLOW_TTL_SECONDS", 120.0), + ndpi_cleanup_interval_packets=_env_int("BACKEND_NDPI_CLEANUP_INTERVAL_PACKETS", 10_000), + ) + + +settings = load_settings() diff --git a/backend/src/main.py b/backend/src/main.py index 00d4491..6ccc57d 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -13,13 +13,16 @@ import src.shared_objects as shared_objects from src.api import nft_manager from src.api import packet_api from src.api import packet_scripting_api +from src.config import settings from src.utilities.database import DatabasePool from src.utilities.packet_broadcaster import PacketBroadcaster -DB_DSN = "postgresql://mitm_user:mitm_password@localhost:5432/mitm_db" - -logging.basicConfig(level=logging.DEBUG) -shared_objects.db = DatabasePool(DB_DSN) +logging.basicConfig(level=getattr(logging, settings.log_level.upper(), logging.DEBUG)) +shared_objects.db = DatabasePool( + settings.db_dsn, + min_size=settings.db_pool_min_size, + max_size=settings.db_pool_max_size, +) app = FastAPI( root_path="/api", @@ -53,7 +56,7 @@ async def on_startup() -> None: raise try: - shared_objects.broadcaster = PacketBroadcaster(loop) + shared_objects.broadcaster = PacketBroadcaster(loop, queue_maxsize=settings.broadcaster_queue_maxsize) shared_objects.db.broadcaster = shared_objects.broadcaster except Exception: logging.exception("Failed to create/attach broadcaster") diff --git a/backend/src/network_sniffer.py b/backend/src/network_sniffer.py index 9f47178..76be59b 100644 --- a/backend/src/network_sniffer.py +++ b/backend/src/network_sniffer.py @@ -31,6 +31,7 @@ from src.utilities.interface_bridge_helpers import ( check_interface_up, get_bridge_ports_once, ) +from src.config import settings from src.utilities.bridge_telemetry import bridge_telemetry_manager from src.utilities.ndpi_classifier import ndpi_classifier from src.utilities.packet_identity import build_packet_uid @@ -100,7 +101,7 @@ class PacketInfo(TypedDict, total=False): # small bounded buffer for packets produced before shared_objects is ready _PACKET_BUFFER: List[PacketInfo] = [] -_BUFFER_CAPACITY = 20000 +_BUFFER_CAPACITY = settings.sniffer_buffer_capacity # Background asyncio loop used for internal tasks in this module (kept but not used for DB pool) async_loop = asyncio.new_event_loop() @@ -396,7 +397,7 @@ def parse_packet(pkt, bridge_label: str) -> None: # ------------------------- # AF_PACKET socket utilities # ------------------------- -def _create_af_packet_socket(ifname: str, rx_buf: int = 4 * 1024 * 1024) -> Optional[socket.socket]: +def _create_af_packet_socket(ifname: str, rx_buf: Optional[int] = None) -> Optional[socket.socket]: """ Create and bind an AF_PACKET raw socket to interface. Non-blocking socket returned or None on failure. @@ -411,7 +412,7 @@ def _create_af_packet_socket(ifname: str, rx_buf: int = 4 * 1024 * 1024) -> Opti return None try: - s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, rx_buf) + s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, rx_buf or settings.sniffer_socket_rcvbuf_bytes) except Exception: logger.debug("SO_RCVBUF set failed for %s (non-fatal)", ifname) @@ -507,7 +508,7 @@ def _session_reader_loop(session_id: str) -> None: pass try: - events = sel.select(timeout=1.0) + events = sel.select(timeout=settings.sniffer_selector_timeout_seconds) except Exception: logger.exception("Selector error in session %s", session_id) time.sleep(0.1) @@ -515,7 +516,7 @@ def _session_reader_loop(session_id: str) -> None: if not events: # occasionally attempt to drain buffer if DB becomes available - if time.time() - last_drain_ts > 5: + if time.time() - last_drain_ts > settings.sniffer_buffer_drain_interval_seconds: try: drain_buffer_to_shared_db() except Exception: @@ -527,7 +528,7 @@ def _session_reader_loop(session_id: str) -> None: sock: socket.socket = key.fileobj iface: str = key.data try: - raw = sock.recv(65536) + raw = sock.recv(settings.sniffer_recv_bytes) if not raw: continue except BlockingIOError: @@ -639,7 +640,7 @@ def stop_afpacket_sniffer(session_id: Optional[str] = None, target: Optional[str s["stop_event"].set() t = s.get("thread") if t and isinstance(t, threading.Thread): - t.join(timeout=2) + t.join(timeout=settings.sniffer_thread_join_timeout_seconds) _sync_bridge_telemetry() logger.info("Stopped session %s", session_id) return diff --git a/backend/src/utilities/bridge_telemetry.py b/backend/src/utilities/bridge_telemetry.py index 0959289..f1e0173 100644 --- a/backend/src/utilities/bridge_telemetry.py +++ b/backend/src/utilities/bridge_telemetry.py @@ -12,6 +12,7 @@ import threading from pathlib import Path from typing import Iterable, Optional +from src.config import settings from src.utilities.packet_tracker import packet_tracker logger = logging.getLogger("bridge_telemetry") @@ -87,14 +88,14 @@ class BridgeTelemetryManager: if process is not None and process.poll() is None: try: process.send_signal(signal.SIGTERM) - process.wait(timeout=3) + process.wait(timeout=settings.telemetry_process_stop_timeout_seconds) except subprocess.TimeoutExpired: process.kill() except Exception: logger.exception("Failed to stop bridge telemetry collector cleanly") if reader is not None and reader.is_alive(): - reader.join(timeout=2) + reader.join(timeout=settings.telemetry_reader_join_timeout_seconds) def _read_loop(self, process: subprocess.Popen[str]) -> None: stdout = process.stdout diff --git a/backend/src/utilities/ndpi_classifier.py b/backend/src/utilities/ndpi_classifier.py index 9b32670..d7ca36c 100644 --- a/backend/src/utilities/ndpi_classifier.py +++ b/backend/src/utilities/ndpi_classifier.py @@ -15,6 +15,8 @@ from typing import Any, Dict, Optional, Tuple from scapy.all import IP, IPv6, TCP, UDP # type: ignore +from src.config import settings + logger = logging.getLogger("ndpi_classifier") ndpi = None # type: ignore @@ -52,9 +54,9 @@ class NDPIClassifier: def __init__(self) -> None: self._engine = None self._flows: Dict[Tuple[Any, ...], _FlowState] = {} - self._max_flows = 200_000 - self._flow_ttl_seconds = 120.0 - self._cleanup_interval_packets = 10_000 + self._max_flows = settings.ndpi_max_flows + self._flow_ttl_seconds = settings.ndpi_flow_ttl_seconds + self._cleanup_interval_packets = settings.ndpi_cleanup_interval_packets self._packet_counter = 0 self._initialized = False diff --git a/backend/src/utilities/packet_tracker.py b/backend/src/utilities/packet_tracker.py index 8ae82bb..e6d13fc 100644 --- a/backend/src/utilities/packet_tracker.py +++ b/backend/src/utilities/packet_tracker.py @@ -12,6 +12,7 @@ from typing import Any, Dict, List, Optional import src.shared_objects as shared_objects from src.Models.etherType import EtherTypeEnum, ethertype_from_int from src.Models.ip_protocol import protocol_from_number +from src.config import settings from src.utilities.packet_identity import build_packet_uid logger = logging.getLogger("packet_tracker") @@ -41,7 +42,7 @@ class PacketTracker: def stop(self) -> None: self._stop_event.set() - self._thread.join(timeout=2) + self._thread.join(timeout=settings.packet_tracker_stop_join_timeout_seconds) def observe_packet(self, pkt_info: Dict[str, Any]) -> str: """Merge parsed packet information into a pending packet entry.""" @@ -221,7 +222,7 @@ class PacketTracker: if not src_ip or not dst_ip or protocol_raw is None: return None - cutoff = time.time() - 1.0 + cutoff = time.time() - settings.packet_tracker_reject_correlation_window_seconds for entry in self._entries.values(): payload = entry["payload"] if entry["last_observed_at"] < cutoff: @@ -292,7 +293,7 @@ class PacketTracker: try: fut = asyncio.run_coroutine_threadsafe(web_db.upsert_packet(payload), web_loop) - fut.result(timeout=2.0) + fut.result(timeout=settings.packet_tracker_persist_timeout_seconds) with self._lock: current = self._entries.get(entry["packet_uid"]) if current is not None: @@ -303,4 +304,8 @@ class PacketTracker: logger.exception("Failed to persist packet %s", entry["packet_uid"]) -packet_tracker = PacketTracker() +packet_tracker = PacketTracker( + finalize_delay_seconds=settings.packet_tracker_finalize_delay_seconds, + retention_seconds=settings.packet_tracker_retention_seconds, + min_flush_interval_seconds=settings.packet_tracker_min_flush_interval_seconds, +) diff --git a/setup_build_server.sh b/setup_build_server.sh index 58bdc64..7ff51d9 100755 --- a/setup_build_server.sh +++ b/setup_build_server.sh @@ -81,6 +81,7 @@ After=network.target [Service] User=$USER_ROOT WorkingDirectory=$BACKEND_DIR +EnvironmentFile=-$BACKEND_DIR/.env ExecStart=$BACKEND_DIR/venv/bin/uvicorn src.main:app --host 127.0.0.1 --port 8000 Restart=always