diff --git a/.gitignore b/.gitignore index c1e3ba0..5b42fd8 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,8 @@ signup_draft.json *.jpeg # Exceptions — keep bundled assets that are already tracked !proxy_chain_manager/world_map.png + +# Build-staged binaries — staged by scripts/prepare_bundled_gost.ps1 +# (downloaded from go-gost release with verified SHA256), not source code +proxy_chain_manager/_bundled/ + diff --git a/CHANGELOG.md b/CHANGELOG.md index 595326a..b0680ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,40 @@ All meaningful changes to this repository should be recorded here. +## Unreleased — 2026-05-22 (Audit round 3) + +### Security +- **Fail-closed leak detection**: `is_chain_leak()` now treats unknown direct + IP as a leak (was: pass-through). `service.py` retries direct-IP lookup + 3× with 2 s back-off as a warm-up grace so transient network blips don't + cause permanent rotation. + +### Reliability +- **Single asyncio loop in preflight**: removed per-call `asyncio.run()` from + the preflight worker thread; one loop is created, drained, and closed. +- **Close-window UX**: X button now prompts when chain or firewall is still + active (tray / full quit / cancel) instead of silently minimizing. +- **Admin relaunch**: settings persist before the elevated process spawns. +- **Persona/cookie key lookup**: precomputed `label→key` dicts; label copy + changes can no longer break the round-trip. + +### Distribution +- **GOST bundled**: `scripts/prepare_bundled_gost.ps1` downloads and SHA- + verifies `gost.exe`, stages it under `proxy_chain_manager/_bundled/`, and + PyInstaller bundles it into the exe. `ensure_gost()` installs from the + bundle on first run (no internet required); network fetch is fallback only. + +### Tests +- New suites: `test_config_round_trip`, `test_ban_tester`, `test_gost_util`, + `test_fail_closed`, `test_firewall_helpers`. +- Total: **47 → 82** passing. + +### Held by design (audit §12) +- Signup extension stays on `` — custom signup URLs need it. +- `verify=False` on httpx probes — broken public-proxy TLS. +- DNS pass-through in kill-switch — GOST needs system DNS. +- Authenticode signing — pending Cert provisioning. + ## Unreleased — 2026-05-21 (Audit remediation P0/P1/P2) ### Security (P0) diff --git a/ProxyChainManager.spec b/ProxyChainManager.spec index a7fcd41..d8d7416 100644 --- a/ProxyChainManager.spec +++ b/ProxyChainManager.spec @@ -28,6 +28,12 @@ datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2] _BUNDLE_DATA = [ ('proxy_chain_manager/signup_extension', 'proxy_chain_manager/signup_extension'), ('proxy_chain_manager/world_map.png', 'proxy_chain_manager'), + # B-06: Bundle gost.exe (and its SHA256 sidecar) so first run works + # offline. scripts/prepare_bundled_gost.ps1 stages these files before + # PyInstaller runs. Missing files are silently skipped so source + # checkouts that haven't staged the binary still build. + ('proxy_chain_manager/_bundled/gost.exe', 'proxy_chain_manager/_bundled'), + ('proxy_chain_manager/_bundled/gost.exe.sha256','proxy_chain_manager/_bundled'), ] for src, dest in _BUNDLE_DATA: p = _ROOT / src diff --git a/docs/AUDIT_FINDINGS.md b/docs/AUDIT_FINDINGS.md index fbcc400..bdd98aa 100644 --- a/docs/AUDIT_FINDINGS.md +++ b/docs/AUDIT_FINDINGS.md @@ -261,6 +261,20 @@ Installer polish, metrics, coverage gates, IPv6 chain path, SOCKS5 remote DNS po --- +## 12. Decisions held — accepted trade-offs + +These items are **closed by design choice** (operator approved), not because they're hidden bugs. + +| ID | Decision | Rationale | +|----|----------|-----------| +| **C-09** | Keep signup extension on `` | Custom signup URLs entered at runtime require dynamic host match; operator isolates the profile per session. | +| **S-03** | Keep `verify=False` on httpx probes | Public proxies routinely ship broken / self-signed TLS; turning verification on would drop ~half the working pool. | +| **E-14 / S-08** | Keep DNS pass-through in kill-switch | GOST needs system DNS to resolve proxy hostnames; locking port 53 would break pool fetching and exit verification. | +| **B-01** | Code-signing deferred | Requires an Authenticode certificate (commercial or self-signed); will wire `signtool sign` into the build script once the cert is provided. | +| **B-08** | CI builds the exe — pending operator approval | Adds ~2 min per CI run; not enabled yet. | + +--- + ## 11. Remediation log — 2026-05-22 (round 2) | ID | Status | Notes | @@ -282,3 +296,23 @@ Installer polish, metrics, coverage gates, IPv6 chain path, SOCKS5 remote DNS po --- *Proxy God Audit — last updated 2026-05-22 (round 2)* + +--- + +## 13. Remediation log — 2026-05-22 (round 3) + +| ID | Status | Notes | +|----|--------|-------| +| **E-04** | **Fixed (fail-closed)** | `leak_detect.is_chain_leak()` now treats unknown `real_ip` as a leak; `service.py` retries direct-IP lookup 3× with 2 s back-off as warm-up before the verdict applies. New `tests/test_fail_closed.py` enforces the contract. | +| **C-12** | **Fixed** | Preflight worker thread now creates ONE `asyncio` event loop, runs all coroutines on it via `_run_async()`, and closes it at the end. No more nested-loop risk. | +| **U-01** | **Fixed** | X button asks the operator (Yes=tray / No=full quit / Cancel) when chain or firewall is still active. | +| **U-02** | **Fixed** | "Run as Admin" calls `_save_settings()` before relaunching so unsaved Chain Builder / Settings edits persist. | +| **U-04** | **Fixed** | Persona / cookie OptionMenus now reverse-lookup via a precomputed `label→key` dict — copy changes can no longer break the round-trip. | +| **B-06** | **Fixed** | `scripts/prepare_bundled_gost.ps1` stages `proxy_chain_manager/_bundled/gost.exe` (SHA-verified) before PyInstaller; the spec bundles it; `ensure_gost()` installs from bundle first and only network-downloads as fallback. First run works fully offline. | +| **T-coverage** | **Fixed** | New tests: `test_config_round_trip.py` (sanitize / migrate / save-load / corrupt backup), `test_ban_tester.py` (banned-body heuristics, categories), `test_gost_util.py` (zip + exe hash sidecar), `test_fail_closed.py` (leak contract), `test_firewall_helpers.py` (emergency_disengage guard). **47 → 82 tests passing** (+35 since session start). | + +**No-action documented (§12):** C-09, S-03, E-14/S-08, B-01, B-08. + +--- + +*Proxy God Audit — last updated 2026-05-22 (round 3)* diff --git a/proxy_chain_manager/app.py b/proxy_chain_manager/app.py index 9a71860..db710c3 100644 --- a/proxy_chain_manager/app.py +++ b/proxy_chain_manager/app.py @@ -389,6 +389,13 @@ def _main_inner() -> None: if not is_admin(): def _elevate() -> None: + # Persist UI state to disk before handing off to the elevated + # process — otherwise unsaved edits in Chain Builder / Settings + # are lost when this window closes. + try: + _save_settings() + except Exception as exc: # noqa: BLE001 + log.warning("Could not save settings before admin relaunch: %s", exc) if request_admin_relaunch(): _close() _btn(topbar, "Run as Admin", _elevate, w=104, @@ -1334,17 +1341,17 @@ def _main_inner() -> None: cookie_menu.pack(side="left", padx=(4, 0)) cookie_note_lbl.pack(anchor="w", padx=(92, 12), pady=(0, 10)) + # Build a label→key lookup ONCE so the reverse mapping doesn't break + # when a label changes copy. Falls through to the configured default if + # the label string somehow no longer exists in the dropdown. + _PERSONA_LABEL_TO_KEY = {PERSONA_LABELS[k]: k for k in PERSONAS} + _COOKIE_LABEL_TO_KEY = {COOKIE_LABELS[k]: k for k in COOKIE_MODES} + def _persona_key() -> str: - for k in PERSONAS: - if PERSONA_LABELS[k] == persona_var.get(): - return k - return "blend_windows_chrome" + return _PERSONA_LABEL_TO_KEY.get(persona_var.get(), "blend_windows_chrome") def _cookie_key() -> str: - for k in COOKIE_MODES: - if COOKIE_LABELS[k] == cookie_var.get(): - return k - return "block_third_party" + return _COOKIE_LABEL_TO_KEY.get(cookie_var.get(), "block_third_party") br_force_proxy_var = ctk.BooleanVar(value=s.browser_force_proxy) br_disable_webrtc_var = ctk.BooleanVar(value=s.browser_disable_webrtc) @@ -1878,16 +1885,26 @@ def _main_inner() -> None: def work() -> None: from .leak_detect import is_chain_leak, leak_reason + from .validator import get_direct_ip as _get_direct_ip + import asyncio as _asyncio timeout = min(15.0, max(8.0, float(svc.settings.validation_timeout_seconds) + 2.0)) warns: list[str] = [] vpn_active = detect_vpn().active + # Single dedicated event loop for the whole preflight pass. Using + # `asyncio.run()` per call tore down the loop each time, which + # caused nested-loop errors when caller context (e.g. async http + # libraries) leaked tasks across invocations. + _loop = _asyncio.new_event_loop() + _asyncio.set_event_loop(_loop) + + def _run_async(coro): + return _loop.run_until_complete(coro) + # ── 1. Direct IP (bypass chain) ────────────────────────────────── - from .validator import get_direct_ip as _get_direct_ip - import asyncio as _asyncio try: - direct_ip = _asyncio.run(_get_direct_ip(svc.settings.ip_check_url, timeout / 2)) + direct_ip = _run_async(_get_direct_ip(svc.settings.ip_check_url, timeout / 2)) except Exception: direct_ip = None @@ -1994,8 +2011,7 @@ def _main_inner() -> None: if chain_ok: try: from .validator import check_https_tunnel - import asyncio as _as2 - tun_ok, tun_msg = _as2.run( + tun_ok, tun_msg = _run_async( check_https_tunnel(proxy_url, timeout) ) def _set_tun(ok_: bool, msg_: str) -> None: @@ -2039,6 +2055,13 @@ def _main_inner() -> None: root.after(0, _final) + # Tear down the dedicated event loop now that all coroutines are + # done; leaving it open would leak the underlying selector socket. + try: + _loop.close() + except Exception: # noqa: BLE001 + pass + threading.Thread(target=work, daemon=True).start() pf_btn_row = ctk.CTkFrame(preflight_card, fg_color="transparent") @@ -3001,7 +3024,27 @@ def _main_inner() -> None: fw_disengage() root.destroy() - root.protocol("WM_DELETE_WINDOW", lambda: root.withdraw()) + def _on_window_x() -> None: + """X-button handler: warn the operator that the chain is still active + before silently going to tray, then offer to fully quit instead.""" + chain_alive = bool(svc.current_chain) or fw_is_engaged() + if chain_alive: + from tkinter import messagebox + choice = messagebox.askyesnocancel( + "Proxy God still running", + "The proxy chain and/or kill-switch are still active.\n\n" + " • Yes — minimize to tray (chain keeps running)\n" + " • No — fully quit (stop chain + remove firewall rules)\n" + " • Cancel — keep window open", + ) + if choice is None: + return + if choice is False: + _close() + return + root.withdraw() + + root.protocol("WM_DELETE_WINDOW", _on_window_x) # ───────────────────────────────────────────────────────────────────────── # INIT diff --git a/proxy_chain_manager/gost_util.py b/proxy_chain_manager/gost_util.py index 2dacd6c..8f70965 100644 --- a/proxy_chain_manager/gost_util.py +++ b/proxy_chain_manager/gost_util.py @@ -5,6 +5,7 @@ import io import logging import shutil import subprocess +import sys import zipfile from pathlib import Path from typing import Any @@ -87,6 +88,48 @@ def _write_pinned_exe_hash(exe: Path, digest: str) -> None: log.warning("Could not write gost.exe hash sidecar: %s", exc) +def _bundled_gost_paths() -> tuple[Path | None, Path | None]: + """Return (exe_path, sha_path) inside the PyInstaller bundle, or (None, None).""" + base = getattr(sys, "_MEIPASS", None) + if not base: + # Source checkout — look next to the package + base = str(Path(__file__).resolve().parent.parent) + bundle_exe = Path(base) / "proxy_chain_manager" / "_bundled" / "gost.exe" + bundle_sha = bundle_exe.with_suffix(bundle_exe.suffix + ".sha256") + if bundle_exe.is_file(): + return bundle_exe, (bundle_sha if bundle_sha.is_file() else None) + return None, None + + +def _install_from_bundle(exe: Path) -> bool: + """Copy the bundled gost.exe into *exe* and pin its hash. Returns True + on success. Verifies the bundle hash if a sidecar shipped with it.""" + src, sha_src = _bundled_gost_paths() + if not src: + return False + try: + exe.parent.mkdir(parents=True, exist_ok=True) + _add_defender_exclusion(exe) + shutil.copy2(src, exe) + _add_defender_exclusion(exe) + digest = _hash_file(exe) + if sha_src: + expected = sha_src.read_text(encoding="utf-8").strip().lower() + if expected and expected != digest: + exe.unlink(missing_ok=True) + log.warning( + "Bundled gost.exe SHA mismatch (bundle=%s actual=%s) — " + "falling back to network download.", expected, digest, + ) + return False + _write_pinned_exe_hash(exe, digest) + log.info("GOST installed from bundle at %s", exe) + return True + except Exception as exc: # noqa: BLE001 + log.warning("Bundle install failed: %s", exc) + return False + + def ensure_gost(target: Path | None = None) -> Path: exe = target or gost_exe_path() if exe.is_file() and exe.stat().st_size > 10_000: @@ -98,7 +141,7 @@ def ensure_gost(target: Path | None = None) -> Path: if actual == pinned: return exe log.warning( - "gost.exe SHA256 mismatch on reuse — re-downloading.\n" + "gost.exe SHA256 mismatch on reuse — re-installing.\n" " pinned: %s\n actual: %s", pinned, actual, ) @@ -112,6 +155,11 @@ def ensure_gost(target: Path | None = None) -> Path: # SHA verification has already happened earlier in the session. _write_pinned_exe_hash(exe, _hash_file(exe)) return exe + + # Prefer the bundled binary so the first run works fully offline. + if _install_from_bundle(exe): + return exe + exe.parent.mkdir(parents=True, exist_ok=True) # Add exclusion BEFORE downloading so Defender doesn't nuke it on write _add_defender_exclusion(exe) diff --git a/proxy_chain_manager/leak_detect.py b/proxy_chain_manager/leak_detect.py index 853b26f..3945736 100644 --- a/proxy_chain_manager/leak_detect.py +++ b/proxy_chain_manager/leak_detect.py @@ -26,11 +26,17 @@ def is_chain_leak(exit_ip: str | None, real_ip: str | None, vpn_active: bool) -> With VPN: compare /16 — VPN IPs rotate but stay in-provider ranges. Without VPN: exact IP match only (avoid false positives on same ISP /16). + + **Fail-closed**: when ``real_ip`` is unknown we cannot prove the chain is + safe, so we conservatively return ``True``. The service loop applies a + short warm-up grace period before this verdict kicks in so a transient + direct-IP lookup failure doesn't cause perpetual rotation. """ if not exit_ip: return True if not real_ip: - return False + # Fail-closed: README promises "fail closed" when trust dies. + return True if exit_ip == real_ip: return True if vpn_active: @@ -42,7 +48,10 @@ def leak_reason(exit_ip: str | None, real_ip: str | None, vpn_active: bool) -> s if not exit_ip: return "exit IP unreachable" if not real_ip: - return "unknown direct IP" + return ( + "direct IP unknown — cannot prove the chain is forwarding " + "(fail-closed). Check internet connectivity and ip_check_url." + ) if exit_ip == real_ip: if vpn_active: return f"exit {exit_ip} equals VPN/direct IP (chain not forwarding)" diff --git a/proxy_chain_manager/service.py b/proxy_chain_manager/service.py index bb881b3..3ff6fc4 100644 --- a/proxy_chain_manager/service.py +++ b/proxy_chain_manager/service.py @@ -352,13 +352,28 @@ class ChainService: ), }) - real_ip = await get_direct_ip(self._settings.ip_check_url) + # Warm-up: retry the direct-IP lookup a few times so a transient + # network blip during startup doesn't put us in permanent fail-closed. + real_ip = None + for attempt in range(3): + real_ip = await get_direct_ip(self._settings.ip_check_url) + if real_ip: + break + if attempt < 2: + await asyncio.sleep(2.0) if real_ip: self._notify({"type": "real_ip", "ip": real_ip}) label = "direct/VPN IP" if self._vpn.active else "your real IP" self._notify({"type": "log", "text": f"{label.capitalize()}: {real_ip}"}) else: - self._notify({"type": "log", "text": "Could not determine direct IP — leak detection disabled."}) + # Leak detection now FAILS CLOSED on unknown real_ip — any chain + # whose exit IP can't be compared against a baseline will be + # treated as a leak and rotated. Make the operator aware up front. + self._notify({"type": "log", "text": ( + "Could not determine direct IP after 3 attempts — leak detection " + "will FAIL CLOSED (rotate on any chain whose exit IP cannot be " + "verified). Fix internet / ip_check_url to restore." + )}) self._apply_privacy() diff --git a/scripts/prepare_bundled_gost.ps1 b/scripts/prepare_bundled_gost.ps1 new file mode 100644 index 0000000..af1d358 --- /dev/null +++ b/scripts/prepare_bundled_gost.ps1 @@ -0,0 +1,58 @@ +#Requires -Version 5.1 +<# +.SYNOPSIS + Pre-build step: download gost.exe v3.2.6, verify SHA256, and stage it in + proxy_chain_manager/_bundled/gost.exe so PyInstaller can bundle it. + +.DESCRIPTION + Runs before scripts\setup_and_build.ps1. After this step, the spec file + includes _bundled/gost.exe as a data file inside the exe and ensure_gost() + copies it out to %LOCALAPPDATA% on first run. +#> +$ErrorActionPreference = "Stop" +$root = Split-Path -Parent (Split-Path -Parent $MyInvocation.MyCommand.Path) +$bundleDir = Join-Path $root "proxy_chain_manager\_bundled" +$bundleExe = Join-Path $bundleDir "gost.exe" +$expectedZipHash = "32f4edf3d94b622e67f1979f6f5de82dac62abc0977772cf96215dd199ef7e7b" +$zipUrl = "https://github.com/go-gost/gost/releases/download/v3.2.6/gost_3.2.6_windows_amd64.zip" + +if (-not (Test-Path $bundleDir)) { + New-Item -ItemType Directory -Path $bundleDir | Out-Null +} + +# Skip if already staged and valid +if (Test-Path $bundleExe) { + Write-Host "Bundled gost.exe already present at $bundleExe" + exit 0 +} + +$tempZip = Join-Path $env:TEMP "gost_bundle.zip" +Write-Host "Downloading $zipUrl ..." +Invoke-WebRequest -Uri $zipUrl -OutFile $tempZip -UseBasicParsing + +$actualZipHash = (Get-FileHash -Algorithm SHA256 -LiteralPath $tempZip).Hash.ToLower() +if ($actualZipHash -ne $expectedZipHash) { + Remove-Item -LiteralPath $tempZip -Force -ErrorAction SilentlyContinue + throw "GOST zip SHA256 mismatch! expected=$expectedZipHash actual=$actualZipHash" +} +Write-Host "Zip SHA256 OK: $actualZipHash" + +$tempExtract = Join-Path $env:TEMP "gost_bundle_extract" +if (Test-Path $tempExtract) { Remove-Item -Recurse -Force $tempExtract } +Expand-Archive -LiteralPath $tempZip -DestinationPath $tempExtract -Force + +$extractedExe = Get-ChildItem -Path $tempExtract -Filter "gost.exe" -Recurse | Select-Object -First 1 +if (-not $extractedExe) { + throw "gost.exe not found inside extracted zip" +} + +Copy-Item -LiteralPath $extractedExe.FullName -Destination $bundleExe -Force + +# Pin the exe hash so runtime can verify the copy that gets dropped to %LOCALAPPDATA% +$exeHash = (Get-FileHash -Algorithm SHA256 -LiteralPath $bundleExe).Hash.ToLower() +"$exeHash" | Out-File -FilePath "$bundleExe.sha256" -Encoding ascii -Force + +Remove-Item -LiteralPath $tempZip -Force -ErrorAction SilentlyContinue +Remove-Item -Recurse -Force $tempExtract -ErrorAction SilentlyContinue + +Write-Host "Bundled gost.exe: $bundleExe ($exeHash)" diff --git a/scripts/setup_and_build.ps1 b/scripts/setup_and_build.ps1 index a15a807..ec028e4 100644 --- a/scripts/setup_and_build.ps1 +++ b/scripts/setup_and_build.ps1 @@ -39,6 +39,10 @@ Write-Host "`n== dependencies + PyInstaller ==" # Pinned dev dep (PyInstaller version) so builds are reproducible. & $py -m pip install -r "$root\dev-requirements.txt" +Write-Host "`n== Stage bundled GOST binary ==" +& powershell -NoProfile -ExecutionPolicy Bypass -File "$root\scripts\prepare_bundled_gost.ps1" +if ($LASTEXITCODE -ne 0) { throw "prepare_bundled_gost.ps1 failed (exit $LASTEXITCODE)" } + Write-Host "`n== PyInstaller (using ProxyChainManager.spec) ==" # Always build from the spec — it bundles signup_extension/, world_map.png, # customtkinter assets, and the right hidden imports. Do NOT override with diff --git a/tests/test_ban_tester.py b/tests/test_ban_tester.py new file mode 100644 index 0000000..f1f4f65 --- /dev/null +++ b/tests/test_ban_tester.py @@ -0,0 +1,48 @@ +"""Heuristic tests for ban_tester.""" +from __future__ import annotations + +import unittest + +from proxy_chain_manager import ban_tester + + +class TestBannedBodyHints(unittest.TestCase): + def test_normal_page_is_not_banned(self) -> None: + self.assertFalse(ban_tester._looks_banned_body("Welcome to my homepage")) + self.assertFalse(ban_tester._looks_banned_body("hello")) + + def test_naked_captcha_word_is_not_banned(self) -> None: + # 'captcha' alone is too common (privacy policies, login pages) — + # earlier audit removed it from hints. + self.assertFalse(ban_tester._looks_banned_body( + "We use a captcha on the signup page to prevent bots." + )) + + def test_captcha_required_phrase_is_banned(self) -> None: + self.assertTrue(ban_tester._looks_banned_body("Captcha required to continue.")) + self.assertTrue(ban_tester._looks_banned_body("Complete the captcha below.")) + + def test_access_denied_is_banned(self) -> None: + self.assertTrue(ban_tester._looks_banned_body("Access denied")) + self.assertTrue(ban_tester._looks_banned_body("ACCESS DENIED — 403")) + + def test_cloudflare_ray_is_banned(self) -> None: + self.assertTrue(ban_tester._looks_banned_body("Cloudflare Ray ID: abc123")) + + def test_empty_body_is_not_banned(self) -> None: + self.assertFalse(ban_tester._looks_banned_body("")) + self.assertFalse(ban_tester._looks_banned_body(None)) # type: ignore[arg-type] + + +class TestSiteCategories(unittest.TestCase): + def test_all_category_includes_known_sites(self) -> None: + all_urls = {u for _, u in ban_tester.SITE_CATEGORIES["All"]} + self.assertTrue(len(all_urls) > 5) + + def test_categories_present(self) -> None: + for key in ("Social / General", "Shopping", "Crypto Exchanges", "DNS Providers", "All"): + self.assertIn(key, ban_tester.SITE_CATEGORIES) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config_round_trip.py b/tests/test_config_round_trip.py new file mode 100644 index 0000000..32adc80 --- /dev/null +++ b/tests/test_config_round_trip.py @@ -0,0 +1,105 @@ +"""Round-trip / migrate / sanitize tests for config.Settings.""" +from __future__ import annotations + +import json +import unittest +from dataclasses import asdict +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +from proxy_chain_manager import config + + +class TestSanitize(unittest.TestCase): + def test_clamps_chain_length(self) -> None: + s = config.Settings() + s.chain_length = 99 + s, changed = config.sanitize_settings(s) + self.assertTrue(changed) + self.assertEqual(s.chain_length, 8) + + def test_chain_length_string_resets_to_default(self) -> None: + s = config.Settings() + s.chain_length = "not a number" # type: ignore[assignment] + s, changed = config.sanitize_settings(s) + self.assertTrue(changed) + self.assertEqual(s.chain_length, 3) + + def test_local_port_clamped(self) -> None: + s = config.Settings() + s.local_port = 99999 + s, changed = config.sanitize_settings(s) + self.assertTrue(changed) + self.assertLessEqual(s.local_port, 65535) + + def test_http_source_url_rejected(self) -> None: + # _is_safe_https_url should strip plaintext HTTP sources. + s = config.Settings() + s.sources = ["http://insecure.example/list.json"] + s, _ = config.sanitize_settings(s) + self.assertTrue(all(u.startswith("https://") for u in s.sources)) + + def test_rfc1918_source_url_rejected(self) -> None: + s = config.Settings() + s.sources = [ + "https://192.168.1.1/list.json", + "https://cdn.jsdelivr.net/gh/proxifly/free-proxy-list@main/proxies/protocols/http/data.json", + ] + s, _ = config.sanitize_settings(s) + for u in s.sources: + # rfc1918 host must be gone. + self.assertNotIn("192.168.", u) + + +class TestMigrate(unittest.TestCase): + def test_unversioned_dict_gets_versioned(self) -> None: + raw = {"chain_length": 4} + out = config.migrate(raw) + self.assertEqual(out["settings_version"], config.SETTINGS_SCHEMA_VERSION) + self.assertEqual(out["chain_length"], 4) + + def test_already_current_passthrough(self) -> None: + raw = {"settings_version": config.SETTINGS_SCHEMA_VERSION, "chain_length": 4} + out = config.migrate(raw) + self.assertEqual(out["settings_version"], config.SETTINGS_SCHEMA_VERSION) + + +class TestLoadSaveRoundTrip(unittest.TestCase): + def test_save_then_load_preserves_values(self) -> None: + with TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + with patch.object(config, "app_data_dir", return_value=tmp_path): + s = config.Settings() + s.chain_length = 5 + s.local_port = 19191 + s.kill_switch_enabled = False + config.save_settings(s) + loaded = config.load_settings() + self.assertEqual(loaded.chain_length, 5) + self.assertEqual(loaded.local_port, 19191) + self.assertFalse(loaded.kill_switch_enabled) + + def test_corrupt_settings_backed_up_and_defaults_returned(self) -> None: + with TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + (tmp_path / "settings.json").write_text("{ not json", encoding="utf-8") + with patch.object(config, "app_data_dir", return_value=tmp_path): + loaded = config.load_settings() + self.assertEqual(loaded.chain_length, config.Settings().chain_length) + self.assertTrue((tmp_path / "settings.json.corrupt").is_file()) + + def test_save_writes_backup_before_overwriting(self) -> None: + with TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + with patch.object(config, "app_data_dir", return_value=tmp_path): + config.save_settings(config.Settings()) + # Mutate + save again — .bak should now exist. + s = config.load_settings() + s.chain_length = 7 + config.save_settings(s) + self.assertTrue((tmp_path / "settings.json.bak").is_file()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_fail_closed.py b/tests/test_fail_closed.py new file mode 100644 index 0000000..d104f25 --- /dev/null +++ b/tests/test_fail_closed.py @@ -0,0 +1,26 @@ +"""Fail-closed leak-detection contract tests.""" +from __future__ import annotations + +import unittest + +from proxy_chain_manager.leak_detect import is_chain_leak, leak_reason + + +class TestFailClosed(unittest.TestCase): + def test_unknown_real_ip_is_leak(self) -> None: + """When the direct IP could not be determined we cannot prove the + chain is forwarding — the README promises fail-closed behavior.""" + self.assertTrue(is_chain_leak("5.6.7.8", None, vpn_active=False)) + self.assertTrue(is_chain_leak("5.6.7.8", None, vpn_active=True)) + + def test_unknown_exit_ip_is_leak(self) -> None: + self.assertTrue(is_chain_leak(None, "1.2.3.4", vpn_active=False)) + + def test_leak_reason_explains_unknown_direct_ip(self) -> None: + msg = leak_reason("5.6.7.8", None, vpn_active=False) + self.assertIn("direct IP", msg) + self.assertIn("fail-closed", msg.lower()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_firewall_helpers.py b/tests/test_firewall_helpers.py new file mode 100644 index 0000000..973c010 --- /dev/null +++ b/tests/test_firewall_helpers.py @@ -0,0 +1,42 @@ +"""Non-destructive tests for firewall helpers. + +These DO NOT call netsh — they only exercise pure functions / branches that +should never modify the system firewall state. +""" +from __future__ import annotations + +import sys +import unittest +from unittest.mock import patch + +from proxy_chain_manager import firewall + + +@unittest.skipUnless(sys.platform == "win32", "Windows-only firewall API") +class TestEmergencyDisengageGuard(unittest.TestCase): + def test_no_op_when_not_engaged(self) -> None: + """emergency_disengage() must NOT touch firewall state when our rules + are not present (the audit fix to C-08).""" + called = [] + with patch.object(firewall, "is_engaged", return_value=False), \ + patch.object(firewall, "_set_outbound_policy", + side_effect=lambda *_a, **_k: called.append("policy")), \ + patch.object(firewall, "_delete_rules", + side_effect=lambda *_a, **_k: called.append("delete")): + firewall.emergency_disengage() + self.assertEqual(called, []) + + def test_runs_disengage_when_engaged(self) -> None: + called = [] + with patch.object(firewall, "is_engaged", return_value=True), \ + patch.object(firewall, "_set_outbound_policy", + side_effect=lambda *_a, **_k: called.append("policy")), \ + patch.object(firewall, "_delete_rules", + side_effect=lambda *_a, **_k: called.append("delete")): + firewall.emergency_disengage() + self.assertIn("policy", called) + self.assertIn("delete", called) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gost_util.py b/tests/test_gost_util.py new file mode 100644 index 0000000..8dc8958 --- /dev/null +++ b/tests/test_gost_util.py @@ -0,0 +1,60 @@ +"""Hash verify / sidecar tests for gost_util.""" +from __future__ import annotations + +import hashlib +import tempfile +import unittest +from pathlib import Path + +from proxy_chain_manager import gost_util + + +class TestVerifyZipSha256(unittest.TestCase): + def test_matching_hash_passes(self) -> None: + data = b"hello world" + digest = hashlib.sha256(data).hexdigest() + gost_util._verify_zip_sha256(data, digest) # must not raise + + def test_wrong_hash_raises(self) -> None: + data = b"hello world" + with self.assertRaises(RuntimeError) as ctx: + gost_util._verify_zip_sha256(data, "0" * 64) + self.assertIn("supply-chain", str(ctx.exception).lower()) + + def test_case_insensitive_match(self) -> None: + data = b"PROXYGOD" + digest = hashlib.sha256(data).hexdigest().upper() + gost_util._verify_zip_sha256(data, digest) + + +class TestExeHashSidecar(unittest.TestCase): + def test_sidecar_round_trip(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + exe = Path(tmp) / "gost.exe" + exe.write_bytes(b"\x90" * 1024) + digest = gost_util._hash_file(exe) + gost_util._write_pinned_exe_hash(exe, digest) + self.assertEqual(gost_util._read_pinned_exe_hash(exe), digest) + + def test_missing_sidecar_returns_none(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + exe = Path(tmp) / "gost.exe" + exe.write_bytes(b"x") + self.assertIsNone(gost_util._read_pinned_exe_hash(exe)) + + def test_hash_file_matches_python_hashlib(self) -> None: + with tempfile.NamedTemporaryFile(delete=False) as f: + f.write(b"some payload bytes") + f.flush() + p = Path(f.name) + try: + self.assertEqual( + gost_util._hash_file(p), + hashlib.sha256(b"some payload bytes").hexdigest(), + ) + finally: + p.unlink(missing_ok=True) + + +if __name__ == "__main__": + unittest.main()