252 lines
9.1 KiB
Python
252 lines
9.1 KiB
Python
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)
|