fix: audit round 3 - fail-closed leak, shared asyncio loop, GOST bundling, close-X UX, persona key lookup, +29 tests

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

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)

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)"

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()