core/controllers/ConnectionController.py

@@ -441,6 +441,11 @@ class ConnectionController:
             if completed_successfully or not ConnectionController.system_uses_wireguard_interface():
                 subprocess.run(('nmcli', 'connection', 'delete', 'hv-ipv6-sink'), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                 ConnectionController.terminate_tor_connection()
+                try:
+                    from core.utils.encrypted_proxy import killswitch
+                    killswitch.disarm()
+                except Exception:
+                    pass
                 SystemState.dissolve()
                 if connection_observer is not None:
                     connection_observer.notify('disconnected', {})
@@ -549,6 +554,15 @@ class ConnectionController:
             except CalledProcessError:
                 raise ConnectionError('The connection could not be established.')
 
+        # Arm killswitch for WireGuard
+        try:
+            wg_server_ip = ConnectionController.__extract_wireguard_endpoint(profile)
+            if wg_server_ip:
+                from core.utils.encrypted_proxy import killswitch
+                killswitch.arm(wg_server_ip, 'wg')
+        except Exception:
+            pass
+
         token = SystemStateController.create(profile.id)
         if connection_observer is not None:
             connection_observer.notify('connected_token', {'session_token': token})
@@ -559,6 +573,23 @@ class ConnectionController:
         except ConnectionError:
             raise ConnectionError('The connection could not be established.')
 
+    @staticmethod
+    def __extract_wireguard_endpoint(profile):
+        import re, socket
+        try:
+            config = profile.get_wireguard_configuration()
+            if config is None:
+                return None
+            match = re.search(r'Endpoint\s*=\s*([^\s:]+):\d+', config)
+            if not match:
+                return None
+            host = match.group(1)
+            if re.match(r'^(\d{1,3}\.){3}\d{1,3}$', host):
+                return host
+            return socket.gethostbyname(host)
+        except Exception:
+            return None
+
     @staticmethod
     def __with_tor_connection(*args, task: Callable[..., Any], connection_observer: Optional[ConnectionObserver] = None, **kwargs):
 

core/services/encrypted_proxy/hysteria_service.py

@@ -104,7 +104,7 @@ def enable_hysteria(username: str, password: str, server_host: str,
                 observer.notify("error", f"{Constants.SINGBOX_TUN_IF} did not appear after 15s")
             return False
 
-        if not killswitch.arm(server_ip, Constants.SINGBOX_TUN_IF):
+        if not killswitch.arm(server_ip, Constants.SINGBOX_TUN_IF, Constants.SINGBOX_INTERNAL_SUBNET):
             if observer:
                 observer.notify("error", "Failed to arm kill switch")
             return False

core/services/encrypted_proxy/vless_service.py

@@ -134,7 +134,7 @@ def enable_vless(vless_link: str, username: str, server_ip: str,
                 observer.notify("error", f"{Constants.SINGBOX_TUN_IF} did not appear after 15s")
             return False
 
-        if not killswitch.arm(server_ip, Constants.SINGBOX_TUN_IF):
+        if not killswitch.arm(server_ip, Constants.SINGBOX_TUN_IF, Constants.SINGBOX_INTERNAL_SUBNET):
             if observer:
                 observer.notify("error", "Failed to arm kill switch")
             return False

core/utils/encrypted_proxy/killswitch.py

@@ -4,12 +4,11 @@ from core.Constants import Constants
 _C = Constants()
 
 
-def arm(server_ip: str, tunnel_if: str) -> bool:
-    result = subprocess.run(
-        ["sudo", _C.KILLSWITCH_WRAPPER, "arm", server_ip, tunnel_if],
-        capture_output=True,
-        text=True,
-    )
+def arm(server_ip: str, tunnel_if: str, internal_subnet: str = None) -> bool:
+    cmd = ["sudo", _C.KILLSWITCH_WRAPPER, "arm", server_ip, tunnel_if]
+    if internal_subnet:
+        cmd.append(internal_subnet)
+    result = subprocess.run(cmd, capture_output=True, text=True)
     if result.returncode != 0:
         print(f"[killswitch] Failed to arm: {result.stderr.strip()}")
         return False

core/utils/encrypted_proxy/killswitch_monitor.py

@@ -10,12 +10,13 @@ tunnel_state: dict = {"alive": True}
 
 
 class KillswitchMonitor(SingboxProcessMonitor):
-    """CLI-specific monitor: adds killswitch + drops detection."""
+    """CLI-specific monitor for sing-box (vless/hysteria2): checks sing-box process health."""
 
     LOG_PREFIX = "hydraveil-drop"
 
     def __init__(self):
-        super().__init__(SingboxRunner())
+        super().__init__()
+        self._runner = SingboxRunner()
         self._on_disconnect  = None
         self._on_leak        = None
         self._on_displaced   = None
@@ -43,6 +44,9 @@ class KillswitchMonitor(SingboxProcessMonitor):
 
     # ── Abstract impl ─────────────────────────────────────────────────────────
 
+    def is_tunnel_alive(self) -> bool:
+        return self._runner.is_running()
+
     def on_process_alive(self) -> None:
         tunnel_state["alive"] = True
 

core/utils/encrypted_proxy/singbox.py

@@ -270,12 +270,11 @@ class SingboxProcessMonitor(ABC):
     threading model (threading.Event for CLI, Qt signals for GUI).
     """
 
-    POLL_INTERVAL = 5
+    POLL_INTERVAL = 10
 
-    def __init__(self, runner: SingboxRunner):
-        self._runner  = runner
-        self._lock    = threading.Lock()
-        self._running = False
+    def __init__(self):
+        self._lock       = threading.Lock()
+        self._running    = False
         self._stop_event = threading.Event()
         self._last_poll_ts = None
 
@@ -284,6 +283,11 @@ class SingboxProcessMonitor(ABC):
             self._running = False
         self._stop_event.set()
 
+    @abstractmethod
+    def is_tunnel_alive(self) -> bool:
+        """Return True if the tunnel is healthy."""
+        pass
+
     @abstractmethod
     def on_process_alive(self) -> None:
         pass
@@ -306,7 +310,7 @@ class SingboxProcessMonitor(ABC):
             if interrupted:
                 break
 
-            alive = self._runner.is_running()
+            alive = self.is_tunnel_alive()
             if alive:
                 self.on_process_alive()
             else:
@@ -327,5 +331,5 @@ class SingboxProcessMonitor(ABC):
             self._running = False
 
         t.join(timeout=5)
-        self._runner.stop()
+        # tunnel teardown handled by subclass
         self.on_monitor_stopped("user_requested")

core/utils/wireguard/wireguard_monitor.py

@@ -0,0 +1,104 @@
+import subprocess
+import threading
+from core.utils.encrypted_proxy.singbox import SingboxProcessMonitor
+from core.utils.encrypted_proxy.singbox import CAUSE_DISCONNECT, CAUSE_DISPLACED
+
+
+class WireGuardMonitor(SingboxProcessMonitor):
+    """Monitor for WireGuard system connections.
+    Checks if the wg interface is alive instead of a sing-box process.
+    Shared between CLI and GUI.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self._session_token  = None
+        self._on_disconnect  = None
+        self._on_displaced   = None
+        self._cause_event    = threading.Event()
+        self._cause: str | None = None
+
+    # ── Setters ───────────────────────────────────────────────────────────────
+
+    def set_on_disconnect(self, fn):
+        self._on_disconnect = fn
+
+    def set_on_displaced(self, fn):
+        self._on_displaced = fn
+
+    def set_session_token(self, token: str):
+        self._session_token = token
+
+    # ── Tunnel health check ───────────────────────────────────────────────────
+
+    def is_tunnel_alive(self) -> bool:
+        try:
+            result = subprocess.run(
+                ['ip', 'link', 'show', 'wg'],
+                capture_output=True,
+                timeout=5,
+            )
+            return result.returncode == 0
+        except (subprocess.TimeoutExpired, FileNotFoundError):
+            return False
+
+    # ── Abstract impl ─────────────────────────────────────────────────────────
+
+    def on_process_alive(self) -> None:
+        pass
+
+    def on_process_dead(self) -> None:
+        self._trigger(CAUSE_DISCONNECT)
+
+    def on_monitor_stopped(self, cause: str) -> None:
+        if cause == CAUSE_DISCONNECT and self._on_disconnect:
+            self._on_disconnect()
+        elif cause == CAUSE_DISPLACED and self._on_displaced:
+            self._on_displaced()
+
+    # ── Monitor loop override ─────────────────────────────────────────────────
+
+    def _monitor_loop(self) -> None:
+        while True:
+            with self._lock:
+                if not self._running:
+                    break
+
+            interrupted = self._stop_event.wait(timeout=self.POLL_INTERVAL)
+            if interrupted:
+                break
+
+            if not self.is_tunnel_alive():
+                self._trigger(CAUSE_DISCONNECT)
+                return
+
+            if self._session_token is not None:
+                from core.models.system.SystemState import SystemState
+                current = SystemState.get()
+                if current is None or current.session_token != self._session_token:
+                    self._trigger(CAUSE_DISPLACED)
+                    return
+
+    def _trigger(self, cause: str):
+        self._cause = cause
+        with self._lock:
+            self._running = False
+        self._cause_event.set()
+        self._stop_event.set()
+
+    # ── run_blocking override ─────────────────────────────────────────────────
+
+    def run_blocking(self) -> None:
+        with self._lock:
+            self._running = True
+
+        self._stop_event.clear()
+        self._cause_event.clear()
+        self._cause = None
+
+        t = threading.Thread(target=self._monitor_loop, daemon=True)
+        t.start()
+        self._cause_event.wait()
+        t.join(timeout=2)
+
+        self.on_monitor_stopped(self._cause or "user_requested")