#!/usr/bin/env python3 """Ethernet / Wi-Fi 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}$") _SSID_RE = re.compile(r"^[\x20-\x7e]{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_network_devices(): """Return [{device, type, state, connection}] for ethernet and wifi.""" 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 not in ("ethernet", "wifi"): continue devices.append({ "device": device, "type": dtype, "state": state, "connection": connection, }) return devices def list_ethernet_devices(): """Backward-compatible alias — ethernet only. """ return [d for d in list_network_devices() if d.get("type") == "ethernet"] 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,GENERAL.TYPE", "device", "show", device, ]) info = {"device": device, "addresses": [], "gateway": "", "dns": [], "connection": "", "type": ""} 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 elif key == "GENERAL.TYPE": # nmcli may report "wifi" / "802-11-wireless" / "ethernet" info["type"] = "wifi" if "wireless" in val or val == "wifi" else ( "ethernet" if "ethernet" in val else val ) return info def current_state(): devices = list_network_devices() for d in devices: details = device_details(d["device"]) for k, v in details.items(): if k == "device": continue if k == "type" and d.get("type"): continue d[k] = v 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" net_type = (data.get("type") or data.get("netType") or "ethernet").strip().lower() if net_type not in ("ethernet", "wifi"): return None, "type must be ethernet or wifi" payload = {"iface": iface, "mode": mode, "type": net_type} if net_type == "wifi": ssid = (data.get("ssid") or "").strip() if not ssid or not _SSID_RE.match(ssid): return None, "invalid ssid" payload["ssid"] = ssid password = data.get("password") if password is None: password = data.get("wifiPassword") or "" payload["password"] = str(password) 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(): state = current_state() cfg_path = network_config_path() if cfg_path.is_file(): try: with cfg_path.open(encoding="utf-8") as f: state["saved"] = json.load(f) except (OSError, json.JSONDecodeError): pass return jsonify(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", "")] if payload.get("type") == "wifi": args += ["wifi", payload["ssid"], payload.get("password", "")] rc, out, err_out = _run(args, timeout=60) if rc != 0: return jsonify({"ok": False, "error": err_out.strip() or out.strip() or f"exit {rc}"}), 500 # Persist without storing empty wifi password overwrite if omitted? store as given. save_payload = {k: v for k, v in payload.items()} save_payload.pop("disabled", None) save_network_config(save_payload) return jsonify({"ok": True, "output": out.strip()}) @network_bp.route("/api/network/disconnect", methods=["POST"]) @login_required def api_disconnect_network(): data = request.get_json(force=True, silent=True) or {} iface = (data.get("iface") or "").strip() if not _IFACE_RE.match(iface): abort(400, "invalid iface") script = str(scripts_dir() / "bt-netconfig") args = ["sudo", "-n", script, iface, "disconnect"] 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 # If saved config pointed at this iface, clear it so boot won't re-apply. cfg_path = network_config_path() if cfg_path.is_file(): try: with cfg_path.open(encoding="utf-8") as f: saved = json.load(f) if (saved.get("iface") or "").strip() == iface: saved["disabled"] = True save_network_config(saved) except (OSError, json.JSONDecodeError): pass return jsonify({"ok": True, "output": out.strip()})