NexusOS/modules/network/backend.py

"""Network status module.

Cross-platform connection info (via psutil), a WireGuard VPN status/toggle
that only works where NetworkManager + nmcli exist (Linux — this mirrors the
XFCE panel's network-popup.py, minus the GTK UI), and a small user-defined
list of ping targets so "is my router/VPN endpoint reachable" isn't tied to
any one hardcoded host.
"""
from __future__ import annotations

import json
import os
import platform
import re
import socket
import subprocess
import uuid
from typing import Any, Dict, List

from synapse.nexus_config import RUNTIME_DIR

_TARGETS_FILE = RUNTIME_DIR / "network_targets.json"

_PING_LATENCY_RE = re.compile(r"time[=<]\s*([\d.]+)\s*ms", re.IGNORECASE)


# --- connection info ---------------------------------------------------------------
def hostname() -> str:
    return socket.gethostname()


def primary_connection() -> Dict[str, Any]:
    import psutil

    stats = psutil.net_if_stats()
    addrs = psutil.net_if_addrs()
    for name, addr_list in addrs.items():
        st = stats.get(name)
        if not st or not st.isup:
            continue
        lname = name.lower()
        if lname.startswith(("lo", "loopback")):
            continue
        for a in addr_list:
            if a.family == socket.AF_INET and not a.address.startswith("169.254"):
                iface_type = "wifi" if any(k in lname for k in ("wlan", "wi-fi", "wireless", "wl")) else "ethernet"
                return {"interface": name, "ip": a.address, "type": iface_type}
    return {"interface": None, "ip": None, "type": "offline"}


# --- WireGuard / VPN (Linux + NetworkManager only) ----------------------------------
def _nmcli(*args: str) -> str:
    try:
        return subprocess.check_output(
            ["nmcli", "-t", "--escape", "no", *args],
            text=True, stderr=subprocess.DEVNULL, timeout=5,
        ).strip()
    except Exception:
        return ""


def _nmcli_available() -> bool:
    if os.name != "posix":
        return False
    try:
        subprocess.check_output(["nmcli", "--version"], stderr=subprocess.DEVNULL, timeout=3)
        return True
    except Exception:
        return False


def vpn_status() -> Dict[str, Any]:
    if not _nmcli_available():
        return {"available": False}
    wgs = [p for p in (line.split(":") for line
                       in _nmcli("-f", "NAME,TYPE,STATE", "connection", "show").splitlines())
           if len(p) >= 3 and p[1] == "wireguard"]
    for name, _t, state in wgs:
        if state == "activated":
            return {"available": True, "configured": True, "name": name, "connected": True}
    if wgs:
        return {"available": True, "configured": True, "name": wgs[0][0], "connected": False}
    return {"available": True, "configured": False, "name": None, "connected": False}


def vpn_toggle(enable: bool) -> Dict[str, Any]:
    status = vpn_status()
    if not status.get("configured"):
        raise RuntimeError("no WireGuard tunnel configured")
    action = "up" if enable else "down"
    subprocess.check_output(
        ["nmcli", "connection", action, status["name"]],
        stderr=subprocess.STDOUT, text=True, timeout=15,
    )
    return vpn_status()


# --- ping targets --------------------------------------------------------------------
def _load_targets() -> List[Dict[str, Any]]:
    try:
        return json.loads(_TARGETS_FILE.read_text()).get("targets", [])
    except Exception:
        return []


def _write_targets(targets: List[Dict[str, Any]]) -> None:
    _TARGETS_FILE.parent.mkdir(parents=True, exist_ok=True)
    _TARGETS_FILE.write_text(json.dumps({"targets": targets}, indent=2))


def list_targets() -> List[Dict[str, Any]]:
    return _load_targets()


def add_target(label: str, host: str) -> Dict[str, Any]:
    targets = _load_targets()
    t = {"id": uuid.uuid4().hex[:12], "label": label, "host": host}
    targets.append(t)
    _write_targets(targets)
    return t


def delete_target(target_id: str) -> None:
    _write_targets([t for t in _load_targets() if t["id"] != target_id])


def ping(host: str) -> Dict[str, Any]:
    is_windows = platform.system() == "Windows"
    cmd = ["ping", "-n", "1", "-w", "1000", host] if is_windows else ["ping", "-c", "1", "-W", "1", host]
    try:
        out = subprocess.check_output(cmd, text=True, stderr=subprocess.STDOUT, timeout=3)
        m = _PING_LATENCY_RE.search(out)
        return {"ok": True, "latency_ms": float(m.group(1)) if m else None}
    except subprocess.CalledProcessError:
        return {"ok": False, "latency_ms": None}
    except Exception as e:
        return {"ok": False, "latency_ms": None, "error": str(e)}


def ping_targets() -> List[Dict[str, Any]]:
    return [{**t, **ping(t["host"])} for t in _load_targets()]