diff --git a/proxy_chain_manager/gost_util.py b/proxy_chain_manager/gost_util.py index ec007ce..806f07b 100644 --- a/proxy_chain_manager/gost_util.py +++ b/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 - # unread would fill the OS buffer and deadlock the child process. + # stderr → rolling log file (not a pipe — pipes deadlock if unread; files never block). + # 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, - stderr=subprocess.DEVNULL, + stdout=stderr_dest, + 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 diff --git a/proxy_chain_manager/service.py b/proxy_chain_manager/service.py index 7fff1cf..0d3eb9d 100644 --- a/proxy_chain_manager/service.py +++ b/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 / 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}) diff --git a/tests/test_proxy_god.py b/tests/test_proxy_god.py index 4484ac2..0ad3945 100644 --- a/tests/test_proxy_god.py +++ b/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(""), "")