init

webconfig/network.py

@@ -0,0 +1,155 @@
+#!/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 re
+import subprocess
+
+from flask import Blueprint, abort, jsonify, request
+
+from auth import login_required
+from paths import scripts_dir
+
+network_bp = Blueprint("network", __name__)
+
+_IFACE_RE = re.compile(r"^[A-Za-z0-9_.:-]{1,32}$")
+
+
+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
+    return jsonify({"ok": True, "output": out.strip()})