diff --git a/README.md b/README.md index 595c2bb..59561b4 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,14 @@ python run.py **Administrator** = full kill-switch. No admin = still runs; firewall enforcement steps aside. +### Self-test (core logic, no GUI) + +```cmd +python -m unittest discover -s tests -v +``` + +Hits config sanitization, fetcher smoke, validator timeouts, and a live `get_direct_ip` check (skips if you’re offline). + --- ## Build the binary diff --git a/proxy_chain_manager/app.py b/proxy_chain_manager/app.py index ea0e913..a6a24b7 100644 --- a/proxy_chain_manager/app.py +++ b/proxy_chain_manager/app.py @@ -697,11 +697,11 @@ def main() -> None: use_pinned_chain=bool(use_manual_var.get()), pinned_chain=list(manual_chain), manual_exit_proxy=normalize_proxy_url(exit_proxy_entry.get()), - health_check_seconds=max(10, int(entries["health"].get().strip())), - full_refresh_seconds=max(60, int(entries["refresh"].get().strip())), + health_check_seconds=min(3600, max(10, int(entries["health"].get().strip()))), + full_refresh_seconds=min(86400, max(60, int(entries["refresh"].get().strip()))), validation_concurrency=max(1, int(entries["conc"].get().strip())), max_candidates=max(10, int(entries["maxc"].get().strip())), - validation_timeout_seconds=max(2.0, float(entries["timeout"].get().strip())), + validation_timeout_seconds=min(120.0, max(2.0, float(entries["timeout"].get().strip()))), prefer_elite=bool(elite_var.get()), kill_switch_enabled=bool(ks_var.get()), proxy_bypass=entries["bypass"].get().strip() or Settings().proxy_bypass, diff --git a/proxy_chain_manager/config.py b/proxy_chain_manager/config.py index cc30cc2..5ad45eb 100644 --- a/proxy_chain_manager/config.py +++ b/proxy_chain_manager/config.py @@ -95,16 +95,24 @@ def sanitize_settings(s: Settings) -> tuple[Settings, bool]: s.local_port = 18888 changed = True try: - if int(s.health_check_seconds) < 10: + hc = int(s.health_check_seconds) + if hc < 10: s.health_check_seconds = 10 changed = True + elif hc > 3600: + s.health_check_seconds = 3600 + changed = True except (TypeError, ValueError): s.health_check_seconds = 180 changed = True try: - if int(s.full_refresh_seconds) < 60: + fr = int(s.full_refresh_seconds) + if fr < 60: s.full_refresh_seconds = 60 changed = True + elif fr > 86400: + s.full_refresh_seconds = 86400 + changed = True except (TypeError, ValueError): s.full_refresh_seconds = 1800 changed = True @@ -123,9 +131,13 @@ def sanitize_settings(s: Settings) -> tuple[Settings, bool]: s.max_candidates = 400 changed = True try: - if float(s.validation_timeout_seconds) < 2.0: + vt = float(s.validation_timeout_seconds) + if vt < 2.0: s.validation_timeout_seconds = 2.0 changed = True + elif vt > 120.0: + s.validation_timeout_seconds = 120.0 + changed = True except (TypeError, ValueError): s.validation_timeout_seconds = 12.0 changed = True diff --git a/proxy_chain_manager/service.py b/proxy_chain_manager/service.py index 70c7cb0..43e8d0b 100644 --- a/proxy_chain_manager/service.py +++ b/proxy_chain_manager/service.py @@ -24,16 +24,6 @@ Notify = Callable[[dict[str, Any]], None] _FETCH_POOL = ThreadPoolExecutor(max_workers=4, thread_name_prefix="fetcher") -def _filter_by_mode(pool: list[str], mode: str) -> list[str]: - """Filter pool by obfuscation/protocol mode.""" - if mode == "http_only": - return [u for u in pool if u.startswith("http://")] - if mode == "socks5_only": - return [u for u in pool if u.startswith("socks5://")] - # "random_mix" and "auto" — keep all, caller will shuffle - return list(pool) - - class ChainService: """Background rotating proxy chain using GOST.""" @@ -323,24 +313,44 @@ class ChainService: # POOL MANAGEMENT # ───────────────────────────────────────────────────────────────────────── + @staticmethod + def _apply_mode_filter(candidates: list[str], mode: str) -> list[str]: + if mode == "http_only": + return [u for u in candidates if u.startswith("http://")] + if mode == "socks5_only": + return [u for u in candidates if u.startswith("socks5://")] + if mode == "random_mix": + out = list(candidates) + random.shuffle(out) + return out + return list(candidates) + def _pick_chain(self) -> list[str]: """Pick chain_length hops: optional fixed last hop + random prefix from pool.""" s = self._settings + chain = self._pick_chain_for_mode(s.obfuscation_mode) + if chain: + return chain + if s.obfuscation_mode != "auto" and not s.use_pinned_chain: + self._notify({ + "type": "log", + "text": "Obfuscation filter left nothing usable — retrying this pick with ALL protocols (auto).", + }) + return self._pick_chain_for_mode("auto") + return [] + + def _pick_chain_for_mode(self, mode: str) -> list[str]: + """Build one chain using given mode (http_only / socks5_only / random_mix / auto).""" + s = self._settings k = max(1, s.chain_length) manual = self._manual_exit_url() if manual and not s.use_pinned_chain: - # Last hop is always manual; earlier hops come from pool (never duplicate manual) mid_need = k - 1 - candidates = [ + base = [ u for u in self._available if u not in self._blacklist and u != manual ] - if s.obfuscation_mode == "http_only": - candidates = [u for u in candidates if u.startswith("http://")] - elif s.obfuscation_mode == "socks5_only": - candidates = [u for u in candidates if u.startswith("socks5://")] - elif s.obfuscation_mode == "random_mix": - random.shuffle(candidates) + candidates = self._apply_mode_filter(base, mode) if mid_need == 0: return [manual] @@ -354,14 +364,8 @@ class ChainService: self._used.add(manual) return prefix + [manual] - # Default: all hops from rotating pool - candidates = [u for u in self._available if u not in self._blacklist] - if s.obfuscation_mode == "http_only": - candidates = [u for u in candidates if u.startswith("http://")] - elif s.obfuscation_mode == "socks5_only": - candidates = [u for u in candidates if u.startswith("socks5://")] - elif s.obfuscation_mode == "random_mix": - random.shuffle(candidates) + base = [u for u in self._available if u not in self._blacklist] + candidates = self._apply_mode_filter(base, mode) if len(candidates) < k: return [] @@ -380,12 +384,19 @@ class ChainService: # Fetch all sources concurrently in thread pool (they are blocking) async def _fetch_one(url: str) -> list[str]: try: - rows = await loop.run_in_executor( - _FETCH_POOL, lambda: fetch_proxy_json(url) + rows = await asyncio.wait_for( + loop.run_in_executor( + _FETCH_POOL, + lambda u=url: fetch_proxy_json(u, timeout=50.0), + ), + timeout=55.0, ) entries = normalize_entries(rows, s.prefer_elite) self._notify({"type": "log", "text": f"Fetched {len(entries)} from source."}) return entries + except asyncio.TimeoutError: + self._notify({"type": "log", "text": f"Fetch timed out (55s): {url[:60]}..."}) + return [] except Exception as e: self._notify({"type": "log", "text": f"Fetch error: {e!s}"}) return [] diff --git a/proxy_chain_manager/validator.py b/proxy_chain_manager/validator.py index bbc8d32..9db74a2 100644 --- a/proxy_chain_manager/validator.py +++ b/proxy_chain_manager/validator.py @@ -44,6 +44,8 @@ async def validate_proxies( timeout_seconds: float, on_progress: Callable[[int, int], None] | None = None, ) -> list[str]: + if not proxy_urls: + return [] sem = asyncio.Semaphore(max(1, concurrency)) ok: list[str] = [] lock = asyncio.Lock() @@ -86,50 +88,69 @@ async def check_chain_exit_ip( timeout_seconds: float, ) -> str | None: """Query the IP-check URL through the local chain proxy. - Returns the exit IP string on success, or None on failure. - Tries the configured URL first, then falls back to alternatives.""" - urls = [check_url] + [u for u in _IP_FALLBACKS if u != check_url] - t = httpx.Timeout(timeout_seconds, connect=min(10.0, timeout_seconds)) + Bounded total time — never stacks one slow request per fallback URL forever.""" + per = max(5.0, min(20.0, float(timeout_seconds))) + budget = max(15.0, min(60.0, float(timeout_seconds) * 2 + 5.0)) + urls = [check_url] + [u for u in _IP_FALLBACKS if u != check_url][:2] + + async def _run() -> str | None: + t = httpx.Timeout(per, connect=min(8.0, per)) + try: + async with httpx.AsyncClient( + proxy=listen_proxy, + timeout=t, + verify=False, + follow_redirects=True, + ) as c: + for url in urls: + try: + r = await c.get(url) + if r.status_code == 200: + ip = _parse_ip(r.text) + if ip: + return ip + except Exception: + continue + except Exception: + pass + return None + try: - async with httpx.AsyncClient( - proxy=listen_proxy, - timeout=t, - verify=False, - follow_redirects=True, - ) as c: - for url in urls: - try: - r = await c.get(url) - if r.status_code == 200: - ip = _parse_ip(r.text) - if ip: - return ip - except Exception: - continue - except Exception: - pass - return None + return await asyncio.wait_for(_run(), timeout=budget) + except asyncio.TimeoutError: + log.debug("check_chain_exit_ip timed out after %.1fs", budget) + return None async def get_direct_ip(check_url: str, timeout_seconds: float = 15.0) -> str | None: """Get our own exit IP without the proxy chain (so we can compare).""" - urls = [check_url] + [u for u in _IP_FALLBACKS if u != check_url] - t = httpx.Timeout(timeout_seconds) + per = max(5.0, min(15.0, float(timeout_seconds))) + budget = max(12.0, min(45.0, float(timeout_seconds) * 2)) + urls = [check_url] + [u for u in _IP_FALLBACKS if u != check_url][:2] + + async def _run() -> str | None: + t = httpx.Timeout(per) + try: + async with httpx.AsyncClient( + timeout=t, + verify=False, + follow_redirects=True, + ) as c: + for url in urls: + try: + r = await c.get(url) + if r.status_code == 200: + ip = _parse_ip(r.text) + if ip: + return ip + except Exception: + continue + except Exception: + pass + return None + try: - async with httpx.AsyncClient( - timeout=t, - verify=False, - follow_redirects=True, - ) as c: - for url in urls: - try: - r = await c.get(url) - if r.status_code == 200: - ip = _parse_ip(r.text) - if ip: - return ip - except Exception: - continue - except Exception: - pass - return None + return await asyncio.wait_for(_run(), timeout=budget) + except asyncio.TimeoutError: + log.debug("get_direct_ip timed out after %.1fs", budget) + return None diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..d4839a6 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# Tests package diff --git a/tests/test_proxy_god.py b/tests/test_proxy_god.py new file mode 100644 index 0000000..25e0605 --- /dev/null +++ b/tests/test_proxy_god.py @@ -0,0 +1,99 @@ +"""Automated checks for Proxy God core logic (no GUI). Run: python -m unittest discover -s tests -v""" +from __future__ import annotations + +import asyncio +import unittest + +from proxy_chain_manager.config import Settings, normalize_proxy_url, sanitize_settings +from proxy_chain_manager.fetcher import fetch_proxy_json, normalize_entries +from proxy_chain_manager.validator import ( + check_chain_exit_ip, + get_direct_ip, + validate_proxies, +) + + +class TestConfig(unittest.TestCase): + def test_normalize_proxy_url(self) -> None: + self.assertEqual(normalize_proxy_url(""), "") + self.assertEqual(normalize_proxy_url("1.2.3.4:8080"), "http://1.2.3.4:8080") + self.assertEqual(normalize_proxy_url("socks5://x:1"), "socks5://x:1") + + def test_sanitize_clamps_extremes(self) -> None: + s = Settings( + chain_length=0, + health_check_seconds=999999, + full_refresh_seconds=10, + validation_timeout_seconds=900.0, + ) + s, changed = sanitize_settings(s) + self.assertTrue(changed) + self.assertEqual(s.chain_length, 1) + self.assertEqual(s.health_check_seconds, 3600) + self.assertEqual(s.full_refresh_seconds, 60) + self.assertEqual(s.validation_timeout_seconds, 120.0) + + +class TestFetcher(unittest.TestCase): + def test_normalize_entries_empty(self) -> None: + self.assertEqual(normalize_entries([], False), []) + + def test_fetch_smoke(self) -> None: + """One real HTTP GET — may fail offline; then skip assertion.""" + try: + rows = fetch_proxy_json( + "https://cdn.jsdelivr.net/gh/proxifly/free-proxy-list@main/proxies/protocols/http/data.json", + timeout=25.0, + ) + except Exception: + self.skipTest("network unavailable") + self.assertIsInstance(rows, list) + if rows: + self.assertTrue(isinstance(rows[0], dict)) + + +class TestValidator(unittest.TestCase): + def test_validate_empty_list_returns_fast(self) -> None: + async def run() -> list[str]: + return await validate_proxies( + [], + "https://api.ipify.org?format=json", + 8, + 5.0, + ) + + self.assertEqual(asyncio.run(run()), []) + + def test_validate_dead_proxy_fast(self) -> None: + async def run() -> list[str]: + return await validate_proxies( + ["http://127.0.0.1:1"], + "https://api.ipify.org?format=json", + 1, + 3.0, + ) + + self.assertEqual(asyncio.run(run()), []) + + def test_get_direct_ip_network(self) -> None: + async def run() -> str | None: + return await get_direct_ip("https://api.ipify.org?format=json", 12.0) + + ip = asyncio.run(run()) + if ip is None: + self.skipTest("could not reach ipify (offline or blocked)") + self.assertGreater(len(ip), 4) + + def test_check_chain_exit_ip_invalid_proxy_quick(self) -> None: + async def run() -> str | None: + return await check_chain_exit_ip( + "http://127.0.0.1:1", + "https://api.ipify.org?format=json", + 8.0, + ) + + self.assertIsNone(asyncio.run(run())) + + +if __name__ == "__main__": + unittest.main()