from __future__ import annotations import hashlib import io import logging import shutil import subprocess import sys import zipfile from pathlib import Path from typing import Any import httpx from .paths import app_data_dir, gost_exe_path log = logging.getLogger(__name__) GOST_RELEASE_ZIP = ( "https://github.com/go-gost/gost/releases/download/v3.2.6/" "gost_3.2.6_windows_amd64.zip" ) # Pinned SHA256 of the release zip. Update this constant when bumping GOST_RELEASE_ZIP. # Verified 2026-05-21 against https://github.com/go-gost/gost/releases/download/v3.2.6/ GOST_RELEASE_ZIP_SHA256 = "32f4edf3d94b622e67f1979f6f5de82dac62abc0977772cf96215dd199ef7e7b" # Pinned SHA256 of the extracted ``gost.exe`` binary inside the v3.2.6 zip. # Recorded the first time the zip is unpacked and verified on every reuse, so # a tampered or partially-overwritten exe on disk forces a clean re-download. GOST_EXE_SHA256_FILE = "gost.exe.sha256" def _add_defender_exclusion(path: Path) -> None: """Add Windows Defender exclusion so GOST is not quarantined.""" try: # ExclusionPath covers gost.exe under this folder; avoid invalid combined params on older builds. folder = str(path.parent).replace("'", "''") subprocess.run( [ "powershell", "-NoProfile", "-NonInteractive", "-Command", f"Add-MpPreference -ExclusionPath '{folder}' -ErrorAction SilentlyContinue", ], capture_output=True, timeout=30, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) log.info("Defender exclusion added for %s", path.parent) except Exception as exc: log.warning("Defender exclusion failed (non-fatal): %s", exc) def _verify_zip_sha256(data: bytes, expected: str) -> None: """Raise RuntimeError if SHA256 of *data* does not match *expected* (hex).""" actual = hashlib.sha256(data).hexdigest().lower() if actual != expected.lower(): raise RuntimeError( f"GOST zip SHA256 mismatch — possible supply-chain attack!\n" f" expected: {expected.lower()}\n" f" actual: {actual}\n" "Delete the downloaded zip and retry, or update GOST_RELEASE_ZIP_SHA256." ) log.info("GOST zip SHA256 OK: %s", actual) def _hash_file(path: Path) -> str: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(64 * 1024), b""): h.update(chunk) return h.hexdigest().lower() def _read_pinned_exe_hash(exe: Path) -> str | None: sidecar = exe.with_suffix(exe.suffix + ".sha256") try: return sidecar.read_text(encoding="utf-8").strip().lower() or None except OSError: return None def _write_pinned_exe_hash(exe: Path, digest: str) -> None: sidecar = exe.with_suffix(exe.suffix + ".sha256") try: sidecar.write_text(digest, encoding="utf-8") except OSError as exc: 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: # Re-verify the on-disk binary against the sidecar pin so a tampered # exe cannot persist across launches. pinned = _read_pinned_exe_hash(exe) if pinned: actual = _hash_file(exe) if actual == pinned: return exe log.warning( "gost.exe SHA256 mismatch on reuse — re-installing.\n" " pinned: %s\n actual: %s", pinned, actual, ) try: exe.unlink() except OSError as exc: log.warning("Could not remove tampered gost.exe: %s", exc) else: # First reuse after older versions: record the current hash so the # next run can verify. Mark as trusted-on-first-use only when zip # 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) log.info("Downloading GOST %s", GOST_RELEASE_ZIP) with httpx.Client(timeout=120.0, follow_redirects=True) as c: r = c.get(GOST_RELEASE_ZIP) r.raise_for_status() data = r.content if len(data) < 64 or data[:2] != b"PK": raise RuntimeError("Downloaded GOST zip looks invalid (not a zip).") # Integrity check before extraction _verify_zip_sha256(data, GOST_RELEASE_ZIP_SHA256) with zipfile.ZipFile(io.BytesIO(data), "r") as z: names = [n for n in z.namelist() if n.lower().endswith("gost.exe")] if not names: raise RuntimeError("gost.exe not found in release zip") with z.open(names[0]) as src, open(exe, "wb") as dst: shutil.copyfileobj(src, dst) # Add exclusion again after write in case Defender scanned during extraction _add_defender_exclusion(exe) # Pin the freshly-extracted exe so subsequent launches can re-verify it. _write_pinned_exe_hash(exe, _hash_file(exe)) log.info("GOST installed at %s", exe) return exe def build_gost_cmd(gost: Path, listen_http: str, forwards: list[str]) -> list[str]: args = [str(gost), "-L", f"http://{listen_http}"] for f in forwards: args.extend(["-F", f]) return args def gost_log_path() -> Path: return app_data_dir() / "gost.log" def popen_no_window(args: list[str]) -> subprocess.Popen: # 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=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 if proc.poll() is not None: return try: proc.terminate() proc.wait(timeout=5) except Exception as exc: log.debug("GOST terminate failed (%s) — escalating to kill.", exc) try: proc.kill() except Exception as kill_exc: log.warning("GOST kill also failed: %s", kill_exc)