Add service status detection and improve tray icon feedback

Signed-off-by: Hans Kokx <hans.d.kokx@gmail.com>

README.md

@@ -93,3 +93,23 @@ chmod +x ~/.local/bin/nextdns-tray
 mkdir -p ~/.config/autostart
 cp ~/.local/share/applications/nextdns-tray.desktop ~/.config/autostart/
 ```
+
+## Status Detection
+
+The tray distinguishes between service state and real DNS usage.
+
+- `Stopped`: NextDNS service is not running.
+- `Running / Not In Use`: service is running, but resolver settings do not point to NextDNS.
+- `Running / Usage Unknown`: service is running and resolver looks correct, but live verification is temporarily unavailable.
+- `Running / In Use`: service is running and usage is verified.
+
+Verification model:
+
+- Fast local checks run on each refresh cycle.
+- End-to-end validation uses `https://test.nextdns.io` and is cached for 30 seconds.
+- A start/restart action forces an immediate end-to-end re-check.
+
+Notes:
+
+- This validates host DNS routing; application-specific DoH settings may bypass host resolver behavior.
+- `curl` is used for the end-to-end check. If unavailable, status may appear as `Usage Unknown`.

home/.local/bin/nextdns-tray

@@ -6,6 +6,8 @@ import time
 import shutil
 import webbrowser
 import base64
+import json
+import ipaddress
 from PyQt6.QtWidgets import QApplication, QSystemTrayIcon, QMenu
 from PyQt6.QtGui import QIcon, QAction, QPixmap
 from PyQt6.QtCore import QTimer
@@ -35,6 +37,13 @@ class NextDNSTray:
         self.tray = QSystemTrayIcon()
         self.menu = QMenu()
 
+        # End-to-end validation cache (seconds)
+        self.e2e_interval = 30
+        self.last_e2e_check = 0.0
+        self.last_e2e_state = None
+        self.last_e2e_reason = "usage check not run yet"
+        self.force_e2e_check = True
+
         # Poll status every 5 seconds
         self.timer = QTimer()
         self.timer.timeout.connect(self.refresh_ui)
@@ -53,19 +62,180 @@ class NextDNSTray:
             # Fallback to a system icon if decoding fails
             return QIcon.fromTheme("yast-security")
 
-    def is_running(self):
-        """Check if nextdns service is active."""
+    def service_status(self):
+        """Return daemon status and a human-readable reason."""
         try:
-            result = subprocess.run([NEXTDNS_PATH, "status"], capture_output=True, text=True)
-            return "running" in result.stdout.lower()
+            result = subprocess.run(
+                [NEXTDNS_PATH, "status"],
+                capture_output=True,
+                text=True,
+                timeout=4,
+            )
+            running = "running" in (result.stdout or "").lower()
+            if running:
+                return True, "service is running"
+            if result.returncode != 0:
+                return False, "service is not running"
+            return False, "service is not running"
+        except FileNotFoundError:
+            return False, "nextdns command not found"
+        except subprocess.TimeoutExpired:
+            return False, "service status check timed out"
         except Exception:
-            return False
+            return False, "service status check failed"
+
+    def _extract_ips(self, text):
+        """Extract valid IP addresses from command output."""
+        ips = []
+        for token in text.replace("\n", " ").split():
+            value = token.strip("[](),;")
+            if "%" in value:
+                value = value.split("%", 1)[0]
+            try:
+                ipaddress.ip_address(value)
+                ips.append(value)
+            except ValueError:
+                continue
+        return ips
+
+    def resolver_uses_nextdns(self):
+        """Check whether system resolver targets local/NextDNS endpoints."""
+        candidate_ips = []
+        resolvectl = shutil.which("resolvectl")
+
+        if resolvectl:
+            try:
+                result = subprocess.run(
+                    [resolvectl, "dns"],
+                    capture_output=True,
+                    text=True,
+                    timeout=4,
+                )
+                candidate_ips = self._extract_ips(result.stdout or "")
+            except Exception:
+                candidate_ips = []
+
+        if not candidate_ips:
+            try:
+                with open("/etc/resolv.conf", "r", encoding="utf-8") as handle:
+                    for line in handle:
+                        line = line.strip()
+                        if line.lower().startswith("nameserver "):
+                            token = line.split(None, 1)[1].strip()
+                            try:
+                                ipaddress.ip_address(token)
+                                candidate_ips.append(token)
+                            except ValueError:
+                                continue
+            except Exception:
+                return False, "unable to read resolver configuration"
+
+        if not candidate_ips:
+            return False, "no DNS nameserver found in resolver configuration"
+
+        for ip_raw in candidate_ips:
+            ip = ipaddress.ip_address(ip_raw)
+            if ip.is_loopback:
+                return True, "resolver points to local DNS listener"
+            if ip.version == 4:
+                text = str(ip)
+                if text.startswith("45.90.28.") or text.startswith("45.90.30."):
+                    return True, f"resolver points to NextDNS endpoint ({text})"
+
+        first = candidate_ips[0]
+        return False, f"resolver points elsewhere ({first})"
+
+    def check_nextdns_usage(self):
+        """Validate if DNS resolution is actively going through NextDNS."""
+        curl = shutil.which("curl")
+        if not curl:
+            return None, "curl is not installed"
+
+        try:
+            result = subprocess.run(
+                [
+                    curl,
+                    "-L",
+                    "-fsS",
+                    "--max-time",
+                    "5",
+                    "https://test.nextdns.io",
+                ],
+                capture_output=True,
+                text=True,
+                timeout=6,
+            )
+            payload = json.loads(result.stdout or "{}")
+            status = str(payload.get("status", "")).lower()
+            if status == "ok":
+                return True, "verified by test.nextdns.io"
+            if status:
+                return False, f"test.nextdns.io returned status '{status}'"
+            return None, "test.nextdns.io returned an unexpected response"
+        except subprocess.CalledProcessError:
+            return None, "unable to reach test.nextdns.io"
+        except subprocess.TimeoutExpired:
+            return None, "test.nextdns.io check timed out"
+        except json.JSONDecodeError:
+            return None, "test.nextdns.io returned invalid JSON"
+        except Exception:
+            return None, "usage check failed"
+
+    def evaluate_health(self):
+        """Compute tray health state and explanatory reason."""
+        service_running, service_reason = self.service_status()
+        if not service_running:
+            return {
+                "state": "stopped",
+                "service_running": False,
+                "reason": service_reason,
+            }
+
+        resolver_ok, resolver_reason = self.resolver_uses_nextdns()
+        if not resolver_ok:
+            return {
+                "state": "running_not_in_use",
+                "service_running": True,
+                "reason": resolver_reason,
+            }
+
+        now = time.time()
+        should_check = (
+            self.force_e2e_check
+            or self.last_e2e_state is None
+            or (now - self.last_e2e_check) >= self.e2e_interval
+        )
+        if should_check:
+            self.last_e2e_state, self.last_e2e_reason = self.check_nextdns_usage()
+            self.last_e2e_check = now
+            self.force_e2e_check = False
+
+        if self.last_e2e_state is True:
+            return {
+                "state": "healthy_in_use",
+                "service_running": True,
+                "reason": self.last_e2e_reason,
+            }
+
+        if self.last_e2e_state is False:
+            return {
+                "state": "running_not_in_use",
+                "service_running": True,
+                "reason": self.last_e2e_reason,
+            }
+
+        return {
+            "state": "running_unknown",
+            "service_running": True,
+            "reason": self.last_e2e_reason,
+        }
 
     def run_cmd(self, cmd):
         """Execute start/stop/restart via pkexec."""
         try:
             # Only prompts for password when an action is taken
             subprocess.run(f"pkexec {NEXTDNS_PATH} {cmd}", shell=True, check=True)
+            self.force_e2e_check = True
             time.sleep(1.5)
             self.refresh_ui()
         except subprocess.CalledProcessError:
@@ -74,15 +244,24 @@ class NextDNSTray:
 
     def refresh_ui(self):
         """Updates icons and menu with a status header."""
-        running = self.is_running()
+        health = self.evaluate_health()
+        running = health["service_running"]
+        state = health["state"]
+        reason = health["reason"]
 
         # Main Tray Icon (The Shield)
-        if running:
+        if state == "healthy_in_use":
             self.tray.setIcon(self.custom_icon)
-            self.tray.setToolTip("NextDNS: Running")
+            self.tray.setToolTip("NextDNS: Running and in use")
+        elif state == "stopped":
+            self.tray.setIcon(self.custom_icon_disabled)
+            self.tray.setToolTip(f"NextDNS: Stopped ({reason})")
+        elif state == "running_not_in_use":
+            self.tray.setIcon(self.custom_icon_disabled)
+            self.tray.setToolTip(f"NextDNS: Running, but DNS not using NextDNS ({reason})")
         else:
             self.tray.setIcon(self.custom_icon_disabled)
-            self.tray.setToolTip("NextDNS: Stopped")
+            self.tray.setToolTip(f"NextDNS: Running (usage check unavailable: {reason})")
 
         self.menu.clear()
 
@@ -92,9 +271,15 @@ class NextDNSTray:
         # title_act.setEnabled(False)
         self.menu.addAction(title_act)
 
-        status = "      Enabled/Running" if running else "      Disabled/Stopped"
-        status_icon = "emblem-checked" if running else "emblem-unmounted"
-        toggle_status = QAction(QIcon.fromTheme(status_icon), status, self.menu)
+        status_map = {
+            "healthy_in_use": ("Running / In Use", "emblem-checked"),
+            "stopped": ("Stopped", "emblem-unmounted"),
+            "running_not_in_use": ("Running / Not In Use", "emblem-important"),
+            "running_unknown": ("Running / Usage Unknown", "dialog-question"),
+        }
+        status, status_icon = status_map.get(state, ("Unknown", "dialog-question"))
+        status_text = f"      {status}\n      {reason}"
+        toggle_status = QAction(QIcon.fromTheme(status_icon), status_text, self.menu)
         toggle_status.setEnabled(False)
         self.menu.addAction(toggle_status)