add config
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 10s
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 10s
This commit is contained in:
22
backend/.env.example
Normal file
22
backend/.env.example
Normal file
@@ -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
|
||||
84
backend/src/config.py
Normal file
84
backend/src/config.py
Normal file
@@ -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()
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user