From 260e53aa4794d3941da818791c9223e2e52e8a92 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Tue, 28 Jul 2026 13:02:11 +0545 Subject: [PATCH] feat(wifi): add DNS service with nmcli backend --- src/services/dns.py | 176 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 176 insertions(+) create mode 100644 src/services/dns.py diff --git a/src/services/dns.py b/src/services/dns.py new file mode 100644 index 0000000..e7cd459 --- /dev/null +++ b/src/services/dns.py @@ -0,0 +1,176 @@ +import subprocess +import threading + +from fabric.utils import GLib, logger + +DEFAULT_PROVIDERS = [ + {"label": "Cloudflare", "primary": "1.1.1.1", "secondary": "1.0.0.1"}, + {"label": "Google", "primary": "8.8.8.8", "secondary": "8.8.4.4"}, + {"label": "OpenDNS", "primary": "208.67.222.222", "secondary": "208.67.220.220"}, + {"label": "AdGuard", "primary": "94.140.14.14", "secondary": "94.140.15.15"}, + {"label": "Quad9", "primary": "9.9.9.9", "secondary": "149.112.112.112"}, +] + + +def _nmcli(*args, timeout=15): + """Run nmcli and return (returncode, stdout, stderr).""" + try: + r = subprocess.run( + ["nmcli", *args], + capture_output=True, + text=True, + timeout=timeout, + ) + return r.returncode, r.stdout.strip(), r.stderr.strip() + except FileNotFoundError: + logger.warning("[Dns] nmcli not found") + return 127, "", "nmcli not found" + except subprocess.TimeoutExpired: + logger.warning("[Dns] nmcli timed out") + return -1, "", "timeout" + except Exception as e: + logger.error(f"[Dns] nmcli failed: {e}") + return 1, "", str(e) + + +def _active_connection_name(): + """Return the NM connection name (id) of the first active connection.""" + rc, out, _ = _nmcli("-t", "-f", "NAME,DEVICE", "connection", "show", "--active") + if rc != 0 or not out: + return None + for line in out.splitlines(): + parts = line.split(":") + if len(parts) >= 2: + return parts[0] + return None + + +def _active_device(): + """Return the interface name of the first active wifi/ethernet device.""" + rc, out, _ = _nmcli("-t", "-f", "DEVICE,TYPE,STATE", "device", "status") + if rc != 0 or not out: + return None + for line in out.splitlines(): + parts = line.split(":") + if ( + len(parts) >= 3 + and parts[1] in ("wifi", "ethernet") + and parts[2] == "connected" + ): + return parts[0] + return None + + +def get_dns(callback=None): + """Get current IPv4 DNS servers for the active connection. + + Returns an empty list when DHCP-provided DNS is in use (auto mode). + Calls ``callback(dns_servers: list[str])`` on the main thread. + """ + + def _work(): + try: + conn_name = _active_connection_name() + if not conn_name: + GLib.idle_add(callback, []) if callback else None + return + + # Check if custom DNS is enabled + rc, out, _ = _nmcli( + "-t", + "-f", + "ipv4.ignore-auto-dns", + "connection", + "show", + conn_name, + ) + if rc == 0 and out.strip() == "no": + GLib.idle_add(callback, []) if callback else None + return + + rc, out, _ = _nmcli("-t", "-f", "ipv4.dns", "connection", "show", conn_name) + if rc != 0 or not out: + GLib.idle_add(callback, []) if callback else None + return + servers = [] + for line in out.splitlines(): + line = line.strip() + if line and line != "--": + servers.append(line) + GLib.idle_add(callback, servers) if callback else None + except Exception as e: + logger.error(f"[Dns] Failed to get DNS: {e}") + if callback: + GLib.idle_add(callback, []) + + threading.Thread(target=_work, daemon=True).start() + + +def set_dns(dns_servers, callback=None): + """Set IPv4 DNS servers on the active connection and apply. + + Pass an empty list to clear custom DNS (revert to DHCP). + Manual DNS uses ``device reapply`` for instant effect. + Auto mode uses ``connection up`` because DHCP renew is required. + Calls ``callback(success: bool)`` on the main thread. + """ + + def _work(): + try: + conn_name = _active_connection_name() + if not conn_name: + logger.warning("[Dns] No active connection") + if callback: + GLib.idle_add(callback, False) + return + + if dns_servers: + for server in dns_servers: + rc, _, err = _nmcli( + "connection", "modify", conn_name, "ipv4.dns", server + ) + if rc != 0: + logger.warning(f"[Dns] Failed to set dns {server}: {err}") + _nmcli( + "connection", + "modify", + conn_name, + "ipv4.ignore-auto-dns", + "yes", + ) + # Fast apply for manual DNS + device = _active_device() + if device: + rc, _, err = _nmcli("device", "reapply", device, timeout=10) + if rc != 0: + logger.warning( + f"[Dns] reapply failed ({device}): {err}, " + "falling back to connection up" + ) + rc, _, err = _nmcli("connection", "up", conn_name, timeout=20) + else: + rc, _, err = _nmcli("connection", "up", conn_name, timeout=20) + else: + # Revert to DHCP — must reconnect to renew DHCP DNS + _nmcli( + "connection", + "modify", + conn_name, + "ipv4.ignore-auto-dns", + "no", + ) + rc, _, err = _nmcli("connection", "up", conn_name, timeout=20) + + ok = rc == 0 + if ok: + logger.info(f"[Dns] DNS set to {dns_servers or 'auto'}") + else: + logger.warning(f"[Dns] apply failed: {err}") + if callback: + GLib.idle_add(callback, ok) + except Exception as e: + logger.error(f"[Dns] Failed to set DNS: {e}") + if callback: + GLib.idle_add(callback, False) + + threading.Thread(target=_work, daemon=True).start()