156 lines
4.8 KiB
Python
156 lines
4.8 KiB
Python
#!/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()})
|