bigbrother/utils/networking.py

#!/usr/bin/env python3
"""Network interface helpers, BPF compilation, gratuitous ARP. No scapy dependency."""

import ctypes
import ctypes.util
import fcntl
import os
import socket
import struct
import subprocess
import time
import logging
from pathlib import Path
from typing import List, Optional, Tuple

logger = logging.getLogger("sensor.utils.networking")

SIOCGIFADDR = 0x8915
SIOCGIFHWADDR = 0x8927
SIOCSIFHWADDR = 0x8924
SIOCGIFFLAGS = 0x8913
SIOCSIFFLAGS = 0x8914
IFF_UP = 0x1
IFF_RUNNING = 0x40
ARPHRD_ETHER = 1

_libpcap = None


def _get_libpcap():
    global _libpcap
    if _libpcap is None:
        lib = ctypes.util.find_library("pcap")
        if not lib:
            raise OSError("libpcap not found")
        _libpcap = ctypes.CDLL(lib)
    return _libpcap


def _ifreq(ifname: str) -> bytes:
    return struct.pack("256s", ifname.encode("utf-8")[:15])


def get_all_interfaces() -> List[str]:
    interfaces = []
    net_path = Path("/sys/class/net")
    if net_path.exists():
        for entry in net_path.iterdir():
            name = entry.name
            if name != "lo":
                interfaces.append(name)
    return sorted(interfaces)


def get_primary_interface() -> Optional[str]:
    try:
        with open("/proc/net/route", "r") as f:
            for line in f.readlines()[1:]:
                fields = line.strip().split()
                if fields[1] == "00000000":
                    return fields[0]
    except (IOError, IndexError):
        pass
    interfaces = get_all_interfaces()
    for iface in interfaces:
        if iface.startswith(("eth", "en")):
            return iface
    return interfaces[0] if interfaces else None


def _is_excluded_interface(iface: str) -> bool:
    """Check if interface should be excluded from detection."""
    excluded_prefixes = ("lo", "tailscale", "wg", "tun", "docker", "veth", "br-")
    return iface.startswith(excluded_prefixes)


def _get_up_interface() -> Optional[str]:
    """Get first UP interface that is not excluded."""
    try:
        interfaces = get_all_interfaces()
        for iface in interfaces:
            if not _is_excluded_interface(iface):
                if is_interface_up(iface):
                    return iface
    except Exception as e:
        logger.debug("Error checking interface status: %s", e)
    return None


def detect_interface_with_retry(
    max_retries: int = 3,
    retry_delay: int = 5,
    exponential: bool = True,
    config_interface: Optional[str] = None,
) -> Optional[str]:
    """Detect primary interface with retry fallback chain.

    Retry chain:
    1. Try get_primary_interface() which checks /proc/net/route
    2. Retry with exponential backoff if interface not available (delays: 5s, 10s, 20s)
    3. Fall back to first non-excluded UP interface
    4. Final fallback: use config_interface if all else fails

    Excluded interfaces: lo, tailscale*, wg*, tun*, docker*, veth*, br-*

    Args:
        max_retries: Number of retry attempts (default 3)
        retry_delay: Initial retry delay in seconds (default 5)
        exponential: Use exponential backoff (default True, multiplies delay by 2 each retry)
        config_interface: Interface from config to use as fallback

    Returns:
        Interface name if found, otherwise None or config_interface if set.
    """
    # Try direct detection first
    iface = get_primary_interface()
    if iface and not _is_excluded_interface(iface):
        logger.debug("Detected interface from /proc/net/route: %s", iface)
        return iface

    # Retry with exponential backoff
    current_delay = retry_delay
    for attempt in range(max_retries):
        logger.debug(
            "Interface detection retry %d/%d (delay=%ds)",
            attempt + 1,
            max_retries,
            current_delay,
        )
        time.sleep(current_delay)

        # Try detection again
        iface = get_primary_interface()
        if iface and not _is_excluded_interface(iface):
            logger.info("Interface detected after retry %d: %s", attempt + 1, iface)
            return iface

        # Try first UP interface
        iface = _get_up_interface()
        if iface:
            logger.info("Fallback to UP interface after retry %d: %s", attempt + 1, iface)
            return iface

        if exponential:
            current_delay *= 2

    # Final fallback to config-supplied interface
    if config_interface:
        logger.warning(
            "Interface detection exhausted retries, using config fallback: %s",
            config_interface,
        )
        return config_interface

    logger.error("Interface detection failed after %d retries and no config fallback", max_retries)
    return None


def get_bridge_candidates() -> List[str]:
    candidates = []
    for iface in get_all_interfaces():
        if iface.startswith(("eth", "en", "usb")):
            candidates.append(iface)
    return candidates


def get_wifi_interfaces() -> List[str]:
    interfaces = []
    for iface in get_all_interfaces():
        phy_path = Path(f"/sys/class/net/{iface}/phy80211")
        wireless_path = Path(f"/sys/class/net/{iface}/wireless")
        if phy_path.exists() or wireless_path.exists() or iface.startswith("wlan"):
            interfaces.append(iface)
    return interfaces


def get_mac(interface: str) -> str:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        info = fcntl.ioctl(sock.fileno(), SIOCGIFHWADDR, _ifreq(interface))
        mac_bytes = info[18:24]
        return ":".join(f"{b:02x}" for b in mac_bytes)
    finally:
        sock.close()


def set_mac(interface: str, mac: str) -> None:
    was_up = is_interface_up(interface)
    if was_up:
        interface_down(interface)
    mac_bytes = bytes(int(b, 16) for b in mac.split(":"))
    ifreq = struct.pack("16sH6s8s", interface.encode()[:15], ARPHRD_ETHER, mac_bytes, b"\x00" * 8)
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        fcntl.ioctl(sock.fileno(), SIOCSIFHWADDR, ifreq)
    finally:
        sock.close()
    if was_up:
        interface_up(interface)


def get_ip(interface: str) -> Optional[str]:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        info = fcntl.ioctl(sock.fileno(), SIOCGIFADDR, _ifreq(interface))
        return socket.inet_ntoa(info[20:24])
    except OSError:
        return None
    finally:
        sock.close()


def is_interface_up(interface: str) -> bool:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        result = fcntl.ioctl(sock.fileno(), SIOCGIFFLAGS, _ifreq(interface))
        flags = struct.unpack("16sH", result[:18])[1]
        return bool(flags & IFF_UP)
    except OSError:
        return False
    finally:
        sock.close()


def interface_up(interface: str) -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        result = fcntl.ioctl(sock.fileno(), SIOCGIFFLAGS, _ifreq(interface))
        flags = struct.unpack("16sH", result[:18])[1]
        flags |= IFF_UP | IFF_RUNNING
        ifreq = struct.pack("16sH", interface.encode()[:15], flags)
        fcntl.ioctl(sock.fileno(), SIOCSIFFLAGS, ifreq)
    finally:
        sock.close()


def interface_down(interface: str) -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        result = fcntl.ioctl(sock.fileno(), SIOCGIFFLAGS, _ifreq(interface))
        flags = struct.unpack("16sH", result[:18])[1]
        flags &= ~IFF_UP
        ifreq = struct.pack("16sH", interface.encode()[:15], flags)
        fcntl.ioctl(sock.fileno(), SIOCSIFFLAGS, ifreq)
    finally:
        sock.close()


def create_vlan_interface(parent: str, vlan_id: int) -> str:
    vlan_iface = f"{parent}.{vlan_id}"
    subprocess.run(
        ["ip", "link", "add", "link", parent, "name", vlan_iface,
         "type", "vlan", "id", str(vlan_id)],
        check=True, capture_output=True,
    )
    interface_up(vlan_iface)
    return vlan_iface


def compile_bpf(expression: str, snaplen: int = 65535, linktype: int = 1) -> bytes:
    pcap = _get_libpcap()

    class bpf_insn(ctypes.Structure):
        _fields_ = [
            ("code", ctypes.c_ushort),
            ("jt", ctypes.c_ubyte),
            ("jf", ctypes.c_ubyte),
            ("k", ctypes.c_uint32),
        ]

    class bpf_program(ctypes.Structure):
        _fields_ = [
            ("bf_len", ctypes.c_uint),
            ("bf_insns", ctypes.POINTER(bpf_insn)),
        ]

    handle = pcap.pcap_open_dead(ctypes.c_int(linktype), ctypes.c_int(snaplen))
    if not handle:
        raise RuntimeError("pcap_open_dead failed")

    prog = bpf_program()
    expr_bytes = expression.encode("utf-8")
    ret = pcap.pcap_compile(
        handle,
        ctypes.byref(prog),
        ctypes.c_char_p(expr_bytes),
        ctypes.c_int(1),
        ctypes.c_uint32(0xFFFFFFFF),
    )
    if ret != 0:
        err = pcap.pcap_geterr(handle)
        pcap.pcap_close(handle)
        if err:
            raise ValueError(f"BPF compile failed: {ctypes.string_at(err).decode()}")
        raise ValueError("BPF compile failed")

    insn_size = ctypes.sizeof(bpf_insn)
    result = bytes(ctypes.string_at(prog.bf_insns, prog.bf_len * insn_size))
    pcap.pcap_freecode(ctypes.byref(prog))
    pcap.pcap_close(handle)
    return result


def send_gratuitous_arp(interface: str, ip: str, mac: str) -> None:
    ETH_P_ARP = 0x0806
    sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ARP))
    try:
        sock.bind((interface, 0))
        src_mac = bytes(int(b, 16) for b in mac.split(":"))
        dst_mac = b"\xff" * 6
        src_ip = socket.inet_aton(ip)

        # Ethernet header
        frame = dst_mac + src_mac + struct.pack("!H", ETH_P_ARP)
        # ARP: hardware=Ethernet, proto=IPv4, hlen=6, plen=4, op=reply(2)
        frame += struct.pack("!HHBBH", 1, 0x0800, 6, 4, 2)
        # sender MAC + IP
        frame += src_mac + src_ip
        # target MAC + IP (broadcast)
        frame += dst_mac + src_ip

        for _ in range(3):
            sock.send(frame)
    finally:
        sock.close()


def mac_to_bytes(mac: str) -> bytes:
    return bytes(int(b, 16) for b in mac.split(":"))


def bytes_to_mac(data: bytes) -> str:
    return ":".join(f"{b:02x}" for b in data)


def ip_in_subnet(ip: str, cidr: str) -> bool:
    network, prefix_len = cidr.split("/")
    prefix_len = int(prefix_len)
    ip_int = struct.unpack("!I", socket.inet_aton(ip))[0]
    net_int = struct.unpack("!I", socket.inet_aton(network))[0]
    mask = (0xFFFFFFFF << (32 - prefix_len)) & 0xFFFFFFFF
    return (ip_int & mask) == (net_int & mask)