VPN-aware leak detection, GOST stderr logging

- _is_same_network: /16 subnet comparison catches NordVPN IP rotation
  (exit_ip != real_ip exact-match missed same-VPN exits on rotated IPs)
- Leak now blacklists the entire dead chain, not just rotate
- GOST stderr/stdout -> gost.log (file, never deadlocks vs pipe)
  with 512KB rotation; last 8 lines shown in UI when GOST dies
- read_gost_log_tail helper for live debugging
- Health check uses same subnet check for consistency
- 5 new subnet tests

Made-with: Cursor

proxy_chain_manager/gost_util.py

@@ -6,10 +6,11 @@ import shutil
 import subprocess
 import zipfile
 from pathlib import Path
+from typing import Any
 
 import httpx
 
-from .paths import gost_exe_path
+from .paths import app_data_dir, gost_exe_path
 
 log = logging.getLogger(__name__)
 
@@ -71,19 +72,48 @@ def build_gost_cmd(gost: Path, listen_http: str, forwards: list[str]) -> list[st
     return args
 
 
+def gost_log_path() -> Path:
+    return app_data_dir() / "gost.log"
+
+
 def popen_no_window(args: list[str]) -> subprocess.Popen:
-    # Use DEVNULL for both pipes — GOST writes logs to stderr, leaving pipes
+    # stderr → rolling log file (not a pipe — pipes deadlock if unread; files never block).
-    # unread would fill the OS buffer and deadlock the child process.
+    # stdout is also captured to the same file since GOST v3 mixes output channels.
     cr = getattr(subprocess, "CREATE_NO_WINDOW", 0)
+    log_path = gost_log_path()
+    try:
+        _rotate_gost_log(log_path)
+        stderr_dest: Any = open(log_path, "ab")  # noqa: WPS515 — intentional long-lived file handle
+    except OSError:
+        stderr_dest = subprocess.DEVNULL
     return subprocess.Popen(
         args,
         stdin=subprocess.DEVNULL,
-        stdout=subprocess.DEVNULL,
+        stdout=stderr_dest,
-        stderr=subprocess.DEVNULL,
+        stderr=stderr_dest,
         creationflags=cr,
     )
 
 
+def _rotate_gost_log(path: Path, max_bytes: int = 512 * 1024) -> None:
+    """Keep gost.log under max_bytes by truncating the oldest half when it grows too large."""
+    try:
+        if path.is_file() and path.stat().st_size > max_bytes:
+            data = path.read_bytes()
+            path.write_bytes(data[len(data) // 2:])
+    except OSError:
+        pass
+
+
+def read_gost_log_tail(lines: int = 40) -> str:
+    """Return the last N lines of gost.log for display in the UI."""
+    try:
+        text = gost_log_path().read_text(encoding="utf-8", errors="replace")
+        return "\n".join(text.splitlines()[-lines:])
+    except OSError:
+        return "(gost.log not found)"
+
+
 def terminate_process(proc: subprocess.Popen | None) -> None:
     if proc is None:
         return

proxy_chain_manager/service.py

@@ -18,7 +18,7 @@ from .config import (
 )
 from .fetcher import fetch_proxy_json, normalize_entries
 from .firewall import disengage as fw_disengage, engage as fw_engage, is_admin
-from .gost_util import build_gost_cmd, ensure_gost, popen_no_window, terminate_process
+from .gost_util import build_gost_cmd, ensure_gost, popen_no_window, read_gost_log_tail, terminate_process
 from .sysproxy import clear_system_proxy, set_system_proxy
 from .validator import check_chain_exit_ip, get_direct_ip, validate_proxies
 
@@ -30,6 +30,32 @@ Notify = Callable[[dict[str, Any]], None]
 _FETCH_POOL = ThreadPoolExecutor(max_workers=8, thread_name_prefix="fetcher")
 
 
+def _is_same_network(ip_a: str | None, ip_b: str | None, prefix_len: int = 16) -> bool:
+    """True if two IPv4 addresses share the same /<prefix_len> subnet.
+
+    With NordVPN (or any VPN), the VPN provider rotates IPs so an exact-match
+    comparison misses leaks where the chain exits through the VPN tunnel directly.
+    A /16 check catches same-ISP/same-VPN exit while still allowing genuine
+    unrelated proxies that happen to share a /24 with the VPN exit.
+    Returns False if either address is None or non-IPv4.
+    """
+    if not ip_a or not ip_b:
+        return False
+    try:
+        a_parts = [int(x) for x in ip_a.split(".")]
+        b_parts = [int(x) for x in ip_b.split(".")]
+        if len(a_parts) != 4 or len(b_parts) != 4:
+            return False
+
+        def to_int(parts: list[int]) -> int:
+            return (parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3]
+
+        mask = (0xFFFFFFFF << (32 - prefix_len)) & 0xFFFFFFFF
+        return (to_int(a_parts) & mask) == (to_int(b_parts) & mask)
+    except Exception:
+        return False
+
+
 class ChainService:
     """Background rotating proxy chain using GOST."""
 
@@ -307,6 +333,11 @@ class ChainService:
                 self._available = [x for x in self._available if x != h]
             self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": None})
             self._notify({"type": "log", "text": "GOST exited immediately — proxies blacklisted."})
+            tail = read_gost_log_tail(20)
+            if tail.strip():
+                for ln in tail.splitlines()[-8:]:
+                    if ln.strip():
+                        self._notify({"type": "log", "text": f"  GOST> {ln.rstrip()}"})
             return False
 
         local_proxy = f"http://{listen}"
@@ -327,9 +358,26 @@ class ChainService:
             self._proc = None
             return False
 
-        if real_ip and exit_ip == real_ip:
+        if real_ip and _is_same_network(exit_ip, real_ip):
             self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": exit_ip})
-            self._notify({"type": "log", "text": f"Leak! Exit={exit_ip} == real IP. Rotating."})
+            self._notify({
+                "type": "log",
+                "text": (
+                    f"Leak detected! Exit={exit_ip} shares network with real IP {real_ip} "
+                    f"(same /16 subnet — chain not forwarding, likely exiting via VPN directly). Rotating."
+                ),
+            })
+            log.warning(
+                "Subnet leak: exit=%s real=%s — chain proxy not forwarding. Blacklisting chain.",
+                exit_ip, real_ip,
+            )
+            # Blacklist the whole chain so we don't reuse broken proxies
+            fixed = self._manual_exit_url()
+            for h in chain:
+                if fixed and h == fixed:
+                    continue
+                self._blacklist.add(h)
+                self._available = [x for x in self._available if x != h]
             terminate_process(self._proc)
             self._proc = None
             return False
@@ -362,9 +410,13 @@ class ChainService:
             exit_ip = await check_chain_exit_ip(local_proxy, self._settings.ip_check_url, timeout)
             log.debug("Periodic exit IP check %.2fs → %s", time.monotonic() - t1, exit_ip or "none")
 
-            if not exit_ip or (real_ip and exit_ip == real_ip):
+            if not exit_ip or (real_ip and _is_same_network(exit_ip, real_ip)):
+                reason = (
+                    "exit IP gone" if not exit_ip
+                    else f"exit {exit_ip} matches real network {real_ip} (VPN leak)"
+                )
                 self._notify({"type": "hops", "hops": chain, "status": "dead", "exit_ip": exit_ip})
-                self._notify({"type": "log", "text": "Health check failed — rotating."})
+                self._notify({"type": "log", "text": f"Health check failed ({reason}) — rotating."})
                 break
 
             self._notify({"type": "hops", "hops": chain, "status": "healthy", "exit_ip": exit_ip})

tests/test_proxy_god.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 import asyncio
 import unittest
 
+from proxy_chain_manager.service import _is_same_network
+
 from proxy_chain_manager.config import (
     Settings,
     merge_proxy_credentials,
@@ -20,6 +22,24 @@ from proxy_chain_manager.validator import (
 )
 
 
+class TestSameNetwork(unittest.TestCase):
+    def test_exact_match(self) -> None:
+        self.assertTrue(_is_same_network("1.2.3.4", "1.2.3.4"))
+
+    def test_same_slash16(self) -> None:
+        self.assertTrue(_is_same_network("185.220.101.5", "185.220.200.9"))
+
+    def test_different_slash16(self) -> None:
+        self.assertFalse(_is_same_network("185.220.101.5", "104.28.3.99"))
+
+    def test_none_safe(self) -> None:
+        self.assertFalse(_is_same_network(None, "1.2.3.4"))
+        self.assertFalse(_is_same_network("1.2.3.4", None))
+
+    def test_non_ipv4_safe(self) -> None:
+        self.assertFalse(_is_same_network("not-an-ip", "1.2.3.4"))
+
+
 class TestConfig(unittest.TestCase):
     def test_normalize_proxy_url(self) -> None:
         self.assertEqual(normalize_proxy_url(""), "")