Method __init__

traffic_analyzer.py:279–357

(self, shared_data=None, interface: str = None)

    # Passive host discovery: periodically flush observed LAN hosts into the
    # hosts DB so traffic-only hosts (firewalled, never scanned) get tracked.
    PASSIVE_SYNC_INTERVAL = 30.0         # seconds between DB flushes
    PASSIVE_MIN_PACKETS = 5              # min packets to upsert if no MAC yet
    PASSIVE_MAX_LISTEN_PORTS = 100       # cap per host
    PASSIVE_LAN_PREFIX = 24              # /24 around each known local/gateway IP

    # Bounded state. Alerts and DNS queries are ring buffers already, but
    # host_stats, connections and the beacon flow history gain an entry per new
    # peer/flow seen and have no natural ceiling — on a box left capturing for a
    # week they are the only thing that grows without limit. Cap them and evict
    # the least-recently-seen entries.
    #
    # The caps are what makes this feature honest on a 512MB Pi Zero 2 W: at the
    # low-memory numbers below the tracked state tops out around 3MB, on top of
    # ~18MB for the analyzer itself and ~8MB for tcpdump.
    STATE_PRUNE_INTERVAL = 60.0          # seconds between eviction sweeps
    MAX_TRACKED_HOSTS = 4000
    MAX_TRACKED_CONNECTIONS = 8000
    MAX_TRACKED_FLOWS = 2000             # each holds a 64-sample ring buffer
    LOW_MEMORY_RAM_GB = 1.0              # below this, use the divided caps
    LOW_MEMORY_DIVISOR = 8

    # Rate limiting
    MAX_ALERTS_PER_MINUTE = 10
    STATS_RETENTION_HOURS = 24

    def __init__(self, shared_data=None, interface: str = None):
        self.shared_data = shared_data
        self.interface = interface or self._detect_interface()
        
        self._running = False
        self._capture_process: Optional[subprocess.Popen] = None
        self._capture_thread: Optional[threading.Thread] = None
        self._analysis_thread: Optional[threading.Thread] = None
        
        self._packet_queue: queue.Queue = queue.Queue(maxsize=10000)
        self._lock = threading.Lock()
        
        # Local IP addresses to exclude from alerts (Ragnar's own IPs)
        self._local_ips: set = self._detect_local_ips()

        # Default gateway IPs: exempt from port-scan heuristic.
        self._gateway_ips: set = self._detect_gateway_ips()

        # LAN networks for passive host discovery (derived from local+gateway IPs).
        self._lan_networks: List[ipaddress.IPv4Network] = self._derive_lan_networks()

        # Passive discovery state:
        # _mac_by_ip: IP -> MAC (from ARP replies)
        # _listening_ports: IP -> set of service ports it appears to serve
        # _hostname_by_ip: IP -> hostname (from observed DNS replies)
        self._mac_by_ip: Dict[str, str] = {}
        self._listening_ports: Dict[str, set] = {}
        self._hostname_by_ip: Dict[str, str] = {}
        self._last_passive_sync: float = time.time()
        
        # Statistics storage
        self.host_stats: Dict[str, HostTrafficStats] = {}