Button/winDeployScripts/software/webconfig/network.py

#!/usr/bin/env python3
"""Ethernet network configuration via NetworkManager (nmcli).

Reading current state is done directly (read-only nmcli). Applying a new
configuration is delegated to the privileged helper script `bt-netconfig`
(invoked through sudo) so the web service can keep running unprivileged.
"""
import ipaddress
import json
import re
import subprocess

from flask import Blueprint, abort, jsonify, request

from auth import login_required
from config_store import config_path
from paths import scripts_dir

network_bp = Blueprint("network", __name__)

_IFACE_RE = re.compile(r"^[A-Za-z0-9_.:-]{1,32}$")


def network_config_path():
    return config_path().parent / "network.json"


def save_network_config(payload):
    path = network_config_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".json.tmp")
    with tmp.open("w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
        f.write("\n")
    tmp.replace(path)


def _run(args, timeout=10):
    try:
        out = subprocess.run(args, capture_output=True, text=True, timeout=timeout)
        return out.returncode, out.stdout, out.stderr
    except FileNotFoundError:
        return 127, "", f"{args[0]} not found"
    except subprocess.TimeoutExpired:
        return 124, "", "timeout"


def list_ethernet_devices():
    """Return [{device, state, connection}] for ethernet interfaces."""
    rc, out, _ = _run([
        "nmcli", "-t", "-f", "DEVICE,TYPE,STATE,CONNECTION", "device", "status",
    ])
    devices = []
    if rc != 0:
        return devices
    for line in out.splitlines():
        parts = line.split(":")
        if len(parts) < 4:
            continue
        device, dtype, state, connection = parts[0], parts[1], parts[2], parts[3]
        if dtype != "ethernet":
            continue
        devices.append({"device": device, "state": state, "connection": connection})
    return devices


def device_details(device):
    """Return current IP4 config for a device."""
    rc, out, _ = _run([
        "nmcli", "-t", "-f",
        "IP4.ADDRESS,IP4.GATEWAY,IP4.DNS,GENERAL.CONNECTION",
        "device", "show", device,
    ])
    info = {"device": device, "addresses": [], "gateway": "", "dns": [],
            "connection": ""}
    if rc != 0:
        return info
    for line in out.splitlines():
        if ":" not in line:
            continue
        key, _, val = line.partition(":")
        val = val.strip()
        if not val:
            continue
        if key.startswith("IP4.ADDRESS"):
            info["addresses"].append(val)
        elif key == "IP4.GATEWAY":
            info["gateway"] = val
        elif key.startswith("IP4.DNS"):
            info["dns"].append(val)
        elif key == "GENERAL.CONNECTION":
            info["connection"] = val
    return info


def current_state():
    devices = list_ethernet_devices()
    for d in devices:
        d.update({k: v for k, v in device_details(d["device"]).items()
                  if k != "device"})
    return {"devices": devices}


def _valid_ip(value):
    try:
        ipaddress.ip_address(value)
        return True
    except ValueError:
        return False


def _validate_payload(data):
    iface = (data.get("iface") or "").strip()
    if not _IFACE_RE.match(iface):
        return None, "invalid iface"
    mode = (data.get("mode") or "dhcp").strip().lower()
    if mode not in ("dhcp", "static"):
        return None, "mode must be dhcp or static"
    payload = {"iface": iface, "mode": mode}
    if mode == "static":
        address = (data.get("address") or "").strip()
        if not _valid_ip(address):
            return None, "invalid address"
        try:
            prefix = int(data.get("prefix", 24))
        except (TypeError, ValueError):
            return None, "invalid prefix"
        if not (1 <= prefix <= 32):
            return None, "prefix must be 1..32"
        gateway = (data.get("gateway") or "").strip()
        if gateway and not _valid_ip(gateway):
            return None, "invalid gateway"
        dns = [d.strip() for d in (data.get("dns") or "").replace(";", ",").split(",")
               if d.strip()]
        for d in dns:
            if not _valid_ip(d):
                return None, f"invalid dns: {d}"
        payload.update({
            "address": address,
            "prefix": prefix,
            "gateway": gateway,
            "dns": ",".join(dns),
        })
    return payload, None


@network_bp.route("/api/network", methods=["GET"])
@login_required
def api_get_network():
    return jsonify(current_state())


@network_bp.route("/api/network", methods=["PUT"])
@login_required
def api_set_network():
    data = request.get_json(force=True, silent=True) or {}
    payload, err = _validate_payload(data)
    if err:
        abort(400, err)

    script = str(scripts_dir() / "bt-netconfig")
    args = ["sudo", "-n", script, payload["iface"], payload["mode"]]
    if payload["mode"] == "static":
        args += [payload["address"], str(payload["prefix"]),
                 payload.get("gateway", ""), payload.get("dns", "")]

    rc, out, err_out = _run(args, timeout=30)
    if rc != 0:
        return jsonify({"ok": False, "error": err_out.strip() or out.strip()
                        or f"exit {rc}"}), 500
    save_network_config(payload)
    return jsonify({"ok": True, "output": out.strip()})