bigbrother/net_alerter/test_net_alerter.py

#!/usr/bin/env python3
"""
Test suite for net_alerter.py — alert formatting fixes and false positive suppression.
"""

import sys
import threading
import time
from pathlib import Path
from unittest.mock import patch, MagicMock

# Add parent dir to path so we can import net_alerter
sys.path.insert(0, str(Path(__file__).parent))

import net_alerter


def test_hostname_dedup_when_hostname_equals_ip():
    """Test that hostname is not repeated when DNS fails (hostname == IP)."""
    # Simulate a device where hostname lookup returned the IP address itself
    mac = "88:a2:9e:8e:f2:90"
    ip = "10.0.0.0"
    hostname = ip  # This is what happens when reverse DNS fails

    # Format the alert using internal logic
    msg = net_alerter.format_arrival(hostname, ip, "Apple", mac)

    # Should show IP only once
    assert msg == "[NET] ARRIVED: 10.0.0.0 [Apple] MAC:88:a2:9e:8e:f2:90"
    assert msg.count("10.0.0.0") == 1, "IP should appear exactly once"


def test_hostname_shown_when_different_from_ip():
    """Test that hostname is shown separately when it differs from IP."""
    mac = "88:a2:9e:8e:f2:90"
    ip = "10.0.0.0"
    hostname = "mydevice.local"

    # Format the alert
    msg = net_alerter.format_arrival(hostname, ip, "Apple", mac)

    # Should show both hostname and IP
    assert msg == "[NET] ARRIVED: mydevice.local (10.0.0.0) [Apple] MAC:88:a2:9e:8e:f2:90"
    assert "mydevice.local" in msg
    assert "10.0.0.0" in msg


def test_oui_lookup_with_inline_dict():
    """Test that OUI lookup returns vendor for known MACs."""
    # These are real OUI prefixes
    test_cases = [
        ("88:a2:9e", "Apple"),  # Apple
        ("00:1a:7d", "Apple"),  # Apple
        ("18:5f:3f", "Apple"),  # Apple
        ("a4:c3:f0", "Apple"),  # Apple
        ("28:87:ba", "Google"),  # Google
        ("54:27:58", "Google"),  # Google
        ("34:15:13", "Amazon"),  # Amazon
        ("b8:27:eb", "Raspberry Pi"),  # Raspberry Pi
    ]

    for mac_prefix, expected_vendor in test_cases:
        vendor = net_alerter.lookup_oui(mac_prefix + ":00:00:00")
        assert vendor == expected_vendor, f"Expected {expected_vendor} for {mac_prefix}, got {vendor}"


def test_oui_lookup_unknown_vendor():
    """Test that unknown OUI returns 'unknown'."""
    vendor = net_alerter.lookup_oui("aa:bb:cc:dd:ee:ff")
    assert vendor == "unknown", f"Expected 'unknown' for unknown OUI, got {vendor}"


def test_departure_format_dedup():
    """Test that departure message also deduplicates hostname and IP."""
    mac = "88:a2:9e:8e:f2:90"
    ip = "10.0.0.0"
    hostname = ip
    duration = "5m 30s"

    msg = net_alerter.format_departure(hostname, ip, "Apple", mac, duration)
    assert msg == "[NET] DEPARTED: 10.0.0.0 [Apple] MAC:88:a2:9e:8e:f2:90 — present 5m 30s"
    assert msg.count("10.0.0.0") == 1


def test_infrastructure_ips_never_alert():
    """Test that infrastructure IPs (gateway, broadcast, self) don't trigger alerts."""
    # Reset known_devices
    net_alerter.known_devices.clear()

    # Mock send_alert to track calls
    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        # Seed infrastructure IPs with infrastructure flag
        net_alerter.known_devices = {
            "ff:ff:ff:ff:ff:ff": {"ip": "10.0.0.0", "hostname": "255", "vendor": "Broadcast", "first_seen": time.time(), "last_seen": time.time(), "infrastructure": True},
            "00:00:00:00:00:00": {"ip": "10.0.0.0", "hostname": "1", "vendor": "Gateway", "first_seen": time.time(), "last_seen": time.time(), "infrastructure": True},
            "12:34:56:78:90:ab": {"ip": "10.0.0.0", "hostname": "10", "vendor": "Self", "first_seen": time.time(), "last_seen": time.time(), "infrastructure": True},
        }

        # Try to trigger departure for infrastructure IPs - should be suppressed
        net_alerter.on_departure("ff:ff:ff:ff:ff:ff")
        net_alerter.on_departure("00:00:00:00:00:00")
        net_alerter.on_departure("12:34:56:78:90:ab")

        # No alerts should be sent
        assert len(alert_calls) == 0, f"Infrastructure IPs should not trigger alerts, but got: {alert_calls}"
    finally:
        net_alerter.send_alert = original_send_alert


def test_departure_debounce_15min():
    """Test that departure alerts are debounced for 15 minutes."""
    net_alerter.known_devices.clear()
    net_alerter.departure_timers.clear()

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "aa:bb:cc:dd:ee:ff"

        # Add device to known_devices
        net_alerter.known_devices[mac] = {
            "ip": "10.0.0.0",
            "hostname": "testhost",
            "vendor": "Test",
            "first_seen": time.time(),
            "last_seen": time.time()
        }

        # Trigger departure
        net_alerter.on_departure(mac)

        # Should NOT have sent alert yet (still in debounce period)
        assert len(alert_calls) == 0, f"Departure should be debounced, but got alert: {alert_calls}"

        # Timer should be created
        assert mac in net_alerter.departure_timers, "Departure timer should be created"

        # Cancel the timer (cleanup)
        net_alerter.departure_timers[mac].cancel()
        del net_alerter.departure_timers[mac]
    finally:
        net_alerter.send_alert = original_send_alert


def test_re_arrival_within_5min_suppresses_alert():
    """Test that device re-arrival within 5 min of departure suppresses ARRIVED alert."""
    net_alerter.known_devices.clear()
    net_alerter.departure_timers.clear()
    net_alerter.last_departed_time.clear()

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "aa:bb:cc:dd:ee:ff"
        ip = "10.0.0.0"

        # Device exists
        net_alerter.known_devices[mac] = {
            "ip": ip,
            "hostname": "testhost",
            "vendor": "Test",
            "first_seen": time.time(),
            "last_seen": time.time()
        }

        # Trigger departure (but don't let the timer fire)
        net_alerter.on_departure(mac)

        # Record departure time (simulating alert being sent after debounce)
        net_alerter.last_departed_time[mac] = time.time()

        # Now device re-arrives within 5 minutes
        net_alerter.on_arrival(mac, ip, "testhost")

        # Should NOT have sent ARRIVED alert (suppressed)
        assert len(alert_calls) == 0, f"Re-arrival within 5 min should be suppressed, but got: {alert_calls}"

        # Cleanup
        if mac in net_alerter.departure_timers:
            net_alerter.departure_timers[mac].cancel()
    finally:
        net_alerter.send_alert = original_send_alert


def test_dhcp_renewal_dedup_30min():
    """Test that DHCP renewal from known device (< 30 min old) doesn't send ARRIVED alert."""
    net_alerter.known_devices.clear()

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "aa:bb:cc:dd:ee:ff"
        ip = "10.0.0.0"

        # Device already known and seen < 30 min ago
        now = time.time()
        net_alerter.known_devices[mac] = {
            "ip": ip,
            "hostname": "testhost",
            "vendor": "Test",
            "first_seen": now - 600,  # 10 min ago
            "last_seen": now - 100,   # 100 sec ago
        }

        # DHCP renewal (arrival)
        net_alerter.on_arrival(mac, ip, "testhost")

        # Should NOT have sent alert (last_seen within 30 min)
        assert len(alert_calls) == 0, f"DHCP renewal < 30 min should be deduped, but got: {alert_calls}"
    finally:
        net_alerter.send_alert = original_send_alert


if __name__ == "__main__":
    test_hostname_dedup_when_hostname_equals_ip()
    print("✓ test_hostname_dedup_when_hostname_equals_ip")

    test_hostname_shown_when_different_from_ip()
    print("✓ test_hostname_shown_when_different_from_ip")

    test_oui_lookup_with_inline_dict()
    print("✓ test_oui_lookup_with_inline_dict")

    test_oui_lookup_unknown_vendor()
    print("✓ test_oui_lookup_unknown_vendor")

    test_departure_format_dedup()
    print("✓ test_departure_format_dedup")

    test_infrastructure_ips_never_alert()
    print("✓ test_infrastructure_ips_never_alert")

    test_departure_debounce_15min()
    print("✓ test_departure_debounce_15min")

    test_re_arrival_within_5min_suppresses_alert()
    print("✓ test_re_arrival_within_5min_suppresses_alert")

    test_dhcp_renewal_dedup_30min()
    print("✓ test_dhcp_renewal_dedup_30min")

    print("\nAll tests passed!")