gandalf/devices.py

"""Device inventory built from the monitor's cached state.

Pure functions with no I/O: app.py passes in the JSON blobs the monitor stored,
so search and the per-device pages always agree about what a device is.
"""
import re
from typing import Dict, List, Optional

# UniFi reports PDUs as switches or APs, so model codes take priority over type.
_PDU_MODELS = {'USPPDUP', 'UP6', 'USPDA2B'}
_KIND_BY_TYPE = {
    'usw': 'switch', 'udm': 'router', 'ugw': 'router', 'uxg': 'router',
    'uap': 'ap', 'usp': 'pdu',
}


_VIRTUAL_IFACE = ('lo', 'vmbr', 'bond', 'docker', 'veth', 'tap', 'fwbr', 'fwpr', 'fwln', 'br-', 'virbr', 'wg', 'tun')


def physical_interfaces(name: str, snapshot: dict) -> List[dict]:
    """A host's physical NICs in name order, with their state: [{'name', 'up', 'unused'}].

    Drawn fronts light one RJ45 per NIC in this order (first NIC on the leftmost jack).
    """
    ifaces = (((snapshot or {}).get('hosts') or {}).get(name) or {}).get('interfaces') or {}
    return [{'name': n, 'up': str(state).lower() == 'up', 'unused': str(state).lower() == 'unused'}
            for n, state in sorted(ifaces.items()) if not n.startswith(_VIRTUAL_IFACE)]


def unifi_kind(unifi_type: str, model: str) -> str:
    if model in _PDU_MODELS:
        return 'pdu'
    return _KIND_BY_TYPE.get((unifi_type or '').lower(), 'device')


def _mac_forms(mac: str) -> List[str]:
    """Return a MAC with and without separators, so either spelling is searchable."""
    return [mac, re.sub(r'[^0-9a-fA-F]', '', mac)] if mac else []


def _search_text(*parts) -> str:
    flat = []
    for p in parts:
        if isinstance(p, (list, tuple, set)):
            flat.extend(str(x) for x in p if x)
        elif p:
            flat.append(str(p))
    return ' '.join(flat).lower()


def _port_neighbors(ports: dict) -> List[str]:
    return [p['lldp']['system_name'] for p in ports.values()
            if (p.get('lldp') or {}).get('system_name')]


def build_index(snapshot: dict, link_stats: dict, modem: Optional[dict]) -> List[dict]:
    """Return every known device as {name, kind, status, ip, model, ..., search}."""
    snapshot = snapshot or {}
    link_stats = link_stats or {}
    devices: Dict[str, dict] = {}

    for name, host in (snapshot.get('hosts') or {}).items():
        link_ifaces = (link_stats.get('hosts') or {}).get(name) or {}
        ifaces = sorted(set(host.get('interfaces') or {}) | set(link_ifaces))
        devices[name] = {
            'name': name, 'kind': 'host', 'status': host.get('status', 'unknown'),
            'ip': host.get('ip', ''), 'mac': '', 'model': '', 'source': host.get('source', ''),
            'interface_count': len(ifaces),
            'search': _search_text(name, host.get('ip'), 'host server', ifaces),
        }

    # Prometheus also sees hosts that aren't in config.json, e.g. LXC app servers.
    # They have link stats but no alerting, so their state is reported as unknown.
    for name, ifaces in (link_stats.get('hosts') or {}).items():
        if name not in devices:
            ip = next((i.get('host_ip') for i in ifaces.values() if i.get('host_ip')), '')
            devices[name] = {
                'name': name, 'kind': 'host', 'status': 'unknown', 'ip': ip, 'mac': '', 'model': '',
                'source': 'prometheus', 'interface_count': len(ifaces),
                'search': _search_text(name, ip, 'host server', list(ifaces)),
            }

    switches = link_stats.get('unifi_switches') or {}
    for d in snapshot.get('unifi') or []:
        name = d.get('name', '')
        ports = (switches.get(name) or {}).get('ports') or {}
        kind = unifi_kind(d.get('type', ''), d.get('model', ''))
        devices[name] = {
            'name': name, 'kind': kind, 'status': 'up' if d.get('connected') else 'down',
            'ip': d.get('ip', ''), 'mac': d.get('mac', ''), 'model': d.get('model', ''),
            'source': 'unifi', 'interface_count': len(ports),
            'search': _search_text(
                name, d.get('ip'), _mac_forms(d.get('mac', '')), d.get('model'),
                d.get('type'), kind, list(ports), _port_neighbors(ports),
            ),
        }

    # A switch can have link stats while the device list is unavailable this cycle.
    for name, sw in switches.items():
        if name not in devices:
            ports = sw.get('ports') or {}
            devices[name] = {
                'name': name, 'kind': unifi_kind('usw', sw.get('model', '')), 'status': 'unknown',
                'ip': sw.get('ip', ''), 'mac': '', 'model': sw.get('model', ''),
                'source': 'unifi', 'interface_count': len(ports),
                'search': _search_text(name, sw.get('ip'), sw.get('model'), list(ports), _port_neighbors(ports)),
            }

    if modem and modem.get('name'):
        name = modem['name']
        if not modem.get('reachable'):
            status = 'unknown'
        else:
            status = 'down' if modem.get('wan_problems') else 'up'
        devices[name] = {
            'name': name, 'kind': 'modem', 'status': status,
            'ip': modem.get('lan_ip', ''), 'mac': '', 'model': modem.get('model', ''),
            'source': 'modem', 'interface_count': 0,
            'search': _search_text(
                name, modem.get('lan_ip'), modem.get('wan_ip'), modem.get('model'),
                'modem gateway fiber wan at&t att',
            ),
        }

    return sorted(devices.values(), key=lambda d: d['name'].lower())


_MAC_RE = re.compile(r'^[0-9a-f]{2}(:[0-9a-f]{2}){5}$')


def _norm_mac(value: Optional[str]) -> str:
    mac = (value or '').strip().lower().replace('-', ':')
    return mac if _MAC_RE.match(mac) else ''


def resolve_lldp_neighbors(
    switches: dict,
    host_macs: Dict[str, tuple],
    device_macs: Dict[str, str],
    host_ips: Dict[str, str],
) -> None:
    """Fill in LLDP neighbor names in place.

    UniFi's API reports a neighbor's MAC, port ID, and sometimes management IPs, but no
    system name. Names come from what Gandalf already knows:

    host_macs:   {mac: (host, interface)} from node_network_info
    device_macs: {mac: device name} from UniFi
    host_ips:    {ip: host}
    """
    for sw in (switches or {}).values():
        for port in (sw.get('ports') or {}).values():
            lldp = port.get('lldp')
            if not lldp or lldp.get('system_name'):
                continue
            port_mac = _norm_mac(lldp.get('port_id'))
            chassis_mac = _norm_mac(lldp.get('chassis_id'))
            if port_mac in host_macs:
                # The port ID names the exact NIC, so it also gives the interface.
                lldp['system_name'], lldp['port_id'] = host_macs[port_mac]
            elif chassis_mac in host_macs:
                lldp['system_name'] = host_macs[chassis_mac][0]
            elif chassis_mac in device_macs or port_mac in device_macs:
                lldp['system_name'] = device_macs.get(chassis_mac) or device_macs[port_mac]
            else:
                lldp['system_name'] = next(
                    (host_ips[ip] for ip in lldp.get('mgmt_ips') or [] if ip in host_ips), '',
                )


def apply_client_locations(
    switches: dict,
    clients: List[dict],
    host_macs: Dict[str, tuple],
    device_macs: Dict[str, str],
) -> None:
    """Name server neighbors from UniFi's wired client table, where LLDP gave no name.

    This finds servers that don't run lldpd. Uplink ports are skipped because they
    carry traffic from devices further downstream.
    """
    for client in clients or []:
        mac = _norm_mac(client.get('mac'))
        switch = device_macs.get(_norm_mac(client.get('sw_mac')))
        if mac not in host_macs or switch not in (switches or {}):
            continue
        for port in (switches[switch].get('ports') or {}).values():
            if port.get('port_idx') != client.get('sw_port') or port.get('is_uplink'):
                continue
            if (port.get('lldp') or {}).get('system_name'):
                continue
            host, iface = host_macs[mac]
            port['lldp'] = {
                'system_name': host, 'port_id': iface, 'chassis_id': mac,
                'port_desc': '', 'mgmt_ips': [], 'source': 'unifi-clients',
            }


def uplinks_for_host(name: str, switches: dict) -> List[dict]:
    """Find the switch ports whose LLDP neighbor is this host."""
    found = []
    for sw_name, sw in (switches or {}).items():
        for port_name, port in (sw.get('ports') or {}).items():
            lldp = port.get('lldp') or {}
            if lldp.get('system_name') == name:
                found.append({
                    'switch': sw_name, 'port': port_name, 'port_idx': port.get('port_idx'),
                    'host_interface': lldp.get('port_id', ''), 'up': port.get('up', False),
                    'speed_mbps': port.get('speed_mbps', 0),
                })
    return found


def device_detail(name: str, snapshot: dict, link_stats: dict, modem: Optional[dict]) -> Optional[dict]:
    """Return everything known about one device, or None if it doesn't exist."""
    match = next((d for d in build_index(snapshot, link_stats, modem) if d['name'] == name), None)
    if match is None:
        return None
    link_stats = link_stats or {}
    detail = {k: v for k, v in match.items() if k != 'search'}
    switches = link_stats.get('unifi_switches') or {}

    if match['kind'] == 'host':
        states = ((snapshot or {}).get('hosts') or {}).get(name, {}).get('interfaces') or {}
        link_ifaces = (link_stats.get('hosts') or {}).get(name) or {}
        interfaces = {i: {'state': s} for i, s in states.items()}
        for iface, stats in link_ifaces.items():
            interfaces.setdefault(iface, {}).update(stats)
        detail['interfaces'] = interfaces
        detail['uplinks'] = uplinks_for_host(name, switches)
    elif match['kind'] == 'modem':
        pass  # the page renders the shared WAN / Fiber panel from /api/modem
    else:
        detail['ports'] = (switches.get(name) or {}).get('ports') or {}
        # The port panel compares each port with the server on the other end
        hosts = link_stats.get('hosts') or {}
        neighbors = set(_port_neighbors(detail['ports']))
        detail['neighbor_hosts'] = {h: ifaces for h, ifaces in hosts.items() if h in neighbors}
        detail['info'] = (link_stats.get('unifi_devices') or {}).get(name) or {}
    detail['updated'] = link_stats.get('updated')
    return detail