Tor System Router v12 is a Windows application that routes all your system traffic through the Tor network by creating a virtual TUN adapter (tor0) and configuring system-wide routing — a complete, one-app anonymity layer with a friendly GUI. 
A separate thread has been created because the program uses two different implementation approaches. The only things that will change in this thread as development progresses are the code and the version number.
What it does
Full system routing — every app’s traffic goes through Tor via a virtual network adapter (not just the browser)
Automatic component management — downloads + updates Tor Expert Bundle, sing-box and Wintun for you
Exit-node rating system — scores exit nodes on speed, CAPTCHA rate and errors, and remembers the good ones
Smart circuit management — auto-switches circuits when an exit looks blacklisted or suspicious
ControlPort integration — live monitoring + control of your Tor circuits
DNS forwarding — all DNS queries resolve through Tor (no sneaky ISP lookups)
Killswitch — firewall rules block every leak the moment Tor is active
GUI dashboard — start/stop, switch circuit, check IP, update components in one click
Full reset — cleans up processes, ports and temp files
Version 12 Removed Snowflake and Meek modes, leaving only the standard Tor mode and bridges. Added optional security measures that you can enable based on your needs. Since each of these protections may affect website functionality, the choice is left entirely to you. Integrated support for Winget has also been added, allowing you to search for and install the required components directly from within the application.
Technical specs
OS Windows (needs Administrator) · Components Tor Expert Bundle + sing-box + Wintun · Protocol SOCKS5 over Tor · Adapter · Ports auto-assigned (SOCKS · DNS · Control)
📜 Full source code — Tor System Router v12 (copy, save as a .py, run as Administrator)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TorLeakGuard - full single-file build (Windows 11).
Routes the whole PC through Tor (sing-box TUN); bridges obfs4 / webtunnel;
anti-DPI padding; IPv6/DNS/WebRTC leak protection; manifest-driven longevity;
winget (auto + manual GUI).
NOTE: snowflake & meek support removed by design - only obfs4/webtunnel bridges
remain. PT discovery reads Tor Browser's OWN torrc (source of truth), so obfs4/
webtunnel work even if the binary is renamed/relocated in your build.
"""
import binascii
import ctypes
import hashlib
import json
import os
import re
import shutil
import socket
import string
import subprocess
import sys
import tarfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from pathlib import Path
import tkinter as tk
from tkinter import ttk
from tkinter import filedialog
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
PROGRAM_VERSION = "2026.07.24.7"
APP_NAME = "TorLeakGuard"
APP_DIR = Path(os.environ.get("LOCALAPPDATA", Path.home())) / "TorLeakGuard"
DOWNLOAD_DIR = APP_DIR / "downloads"
TOR_DIR = APP_DIR / "tor"
SINGBOX_DIR = APP_DIR / "sing-box"
TOR_DATA = APP_DIR / "tor-data"
PT_DIR = APP_DIR / "pt"
TORRC = APP_DIR / "torrc"
SINGBOX_CONFIG = APP_DIR / "sing-box.json"
STATE_FILE = APP_DIR / "state.json"
VERSIONS_FILE = APP_DIR / "versions.json"
SETTINGS_FILE = APP_DIR / "settings.json"
SOURCES_FILE = APP_DIR / "sources.json"
MANIFEST_CACHE_FILE = APP_DIR / "manifest_cache.json"
TRACKLIST_FILE = APP_DIR / "tracklist.txt"
TOR_LOG = APP_DIR / "tor.log"
SINGBOX_LOG = APP_DIR / "sing-box.log"
SINGBOX_CONSOLE_LOG = APP_DIR / "sing-box-console.log"
TOR_BROWSER_TEMP_DIR = Path(
os.environ.get("TEMP") or os.environ.get("TMP") or str(Path.home())
) / "TorLeakGuard_TB_Temp"
SCRIPT_PATH = Path(sys.argv[0]).resolve()
ROAMING_DIR = Path(os.environ.get("APPDATA", str(Path.home()))) / "TorLeakGuard"
WINTUN_VERSION = "0.14.1"
CREATE_NO_WINDOW = 0x08000000 if os.name == "nt" else 0
HTTP_HEADERS = {"User-Agent": "TorLeakGuard/1.0 (Windows11; local privacy tool)"}
# Only obfs-family + webtunnel remain (snowflake & meek removed by design).
PT_GROUPS = [
["lyrebird.exe", "obfs4proxy.exe"],
]
ALL_PT_NAMES = [n for g in PT_GROUPS for n in g]
TOR_PT_NAMES = ["lyrebird.exe", "obfs4proxy.exe"]
KNOWN_PT = {"obfs2", "obfs3", "obfs4", "webtunnel"}
# Map our internal "kind" -> transport names as they appear in Tor Browser torrc.
KIND2TRANSPORT = {
"obfs": ["obfs4", "obfs3", "obfs2"],
"obfs4": ["obfs4", "obfs3", "obfs2"],
"lyrebird": ["obfs4", "webtunnel"],
}
# Source-of-truth cache: Tor Browser torrc -> {transport_name: (exe_abs, full_exec_tail)}
_TB_PT_MAP = None
TB_EXEC_TAIL = {}
TB_EXEC_EXE = {}
KILL_PS = (
r"Get-CimInstance Win32_Process | Where-Object { "
r"$_.ExecutablePath -like '*\TorLeakGuard\*' -and "
r"($_.Name -eq 'tor.exe' -or $_.Name -eq 'sing-box.exe' -or "
r"$_.Name -eq 'obfs4proxy.exe' -or $_.Name -eq 'lyrebird.exe') } | ForEach-Object { "
r"Stop-Process -Id $_.ProcessId -Force -ErrorAction SilentlyContinue }"
)
CIRCUM_HINT = (
"Circumvention: Tor hides SNI/IP inside the tunnel (DPI site-blocks don't apply). "
"If your ISP blocks Tor itself use Bridges (obfs4=noise, webtunnel=HTTPS). "
"'Check bridge status' proves a bridge carries traffic ([ACTIVE]=real ESTABLISHED conn). "
"RESET=emergency eject (never waits). Limits: Tor speed capped by Tor net; raw ISP bandwidth cap not bypassable."
)
PRIVACY_HINT = (
"Always-on (no breakage): DNS via Tor, IPv6 blocked, WebRTC blocked (UDP off), padding. "
"Real browser fingerprinting (canvas/WebGL/audio/window) can ONLY be fixed by the browser -> Tor Browser. "
"Cookie theft over network already prevented by Tor+TLS; JS/XSS theft is a browser problem. "
"STRICT options below are OFF by default (may break some sites)."
)
WINGET_HINT = (
"winget is a first-class path: auto-used for 7-Zip and for any component whose winget_id is set in the "
"manifest. NOTE: winget installs Tor/sing-box into Program Files (NOT our folder), so it cannot feed our "
"TUN scheme directly - use the manual search below to install any module into the system when auto-detect misses it."
)
SYSTEM_HINT = (
"PT discovery reads Tor Browser's OWN torrc as the source of truth (works even if the obfs4/webtunnel "
"binary has a different name/location). Only obfs4 / webtunnel bridges are supported."
)
_ABORT = {"flag": False}
_RESTORED = {"done": False}
class Aborted(Exception):
pass
def abort_now():
_ABORT["flag"] = True
def abort_clear():
_ABORT["flag"] = False
def check_abort():
if _ABORT["flag"]:
raise Aborted()
def _dec(b):
if b is None:
return ""
if isinstance(b, str):
return b
for enc in ("utf-8", "cp1251", "cp866", "latin-1"):
try:
return b.decode(enc)
except Exception:
continue
return b.decode("latin-1", errors="replace")
def run_cmd(cmd, timeout=None):
try:
p = subprocess.run(
cmd,
capture_output=True,
creationflags=CREATE_NO_WINDOW,
timeout=timeout,
)
return p.returncode, _dec(p.stdout), _dec(p.stderr)
except subprocess.TimeoutExpired:
return -1, "", "timeout"
except Exception as e:
return -1, "", str(e)
def run_silent(cmd, timeout=None):
try:
subprocess.run(
cmd,
creationflags=CREATE_NO_WINDOW,
timeout=timeout,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return True
except Exception:
return False
def ensure_dirs():
for d in (APP_DIR, DOWNLOAD_DIR, TOR_DIR, SINGBOX_DIR, TOR_DATA, PT_DIR):
d.mkdir(parents=True, exist_ok=True)
def is_admin():
try:
return bool(ctypes.windll.shell32.IsUserAnAdmin())
except Exception:
return False
def restart_as_admin(action=None):
script = Path(sys.argv[0]).resolve()
params = f'"{script}"'
if action == "start":
params += " --autostart"
elif action == "install":
params += " --autoinstall"
elif action == "update":
params += " --autoupdate"
elif action == "stop":
params += " --autostop"
ret = ctypes.windll.shell32.ShellExecuteW(
None, "runas", sys.executable, params, None, 1
)
if ret <= 32:
raise RuntimeError("Failed to request administrator rights.")
sys.exit(0)
def safe_unlink(path):
try:
path = Path(path)
if path.exists():
path.unlink()
except Exception:
pass
def ps_quote(s):
return "'" + str(s).replace("'", "''") + "'"
def load_versions():
if VERSIONS_FILE.exists():
try:
return json.loads(VERSIONS_FILE.read_text(encoding="utf-8"))
except Exception:
return {}
return {}
def save_versions(data):
ensure_dirs()
VERSIONS_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8")
def load_settings():
defaults = {
"mode": "normal",
"bridges": "",
"pt_path": "",
"prevent_circuit_change": True,
"padding": True,
"block_insecure_http": False,
"block_webrtc_hosts": False,
}
if SETTINGS_FILE.exists():
try:
data = json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
if "pt_path" not in data and data.get("obfs4proxy"):
data["pt_path"] = data.get("obfs4proxy")
defaults.update(data)
except Exception:
pass
return defaults
def save_settings(settings):
ensure_dirs()
SETTINGS_FILE.write_text(json.dumps(settings, indent=2), encoding="utf-8")
def load_sources():
ensure_dirs()
if not SOURCES_FILE.exists():
save_sources(DEFAULT_SOURCES)
return json.loads(json.dumps(DEFAULT_SOURCES))
try:
data = json.loads(SOURCES_FILE.read_text(encoding="utf-8"))
merged = json.loads(json.dumps(DEFAULT_SOURCES))
merged.update(data)
return merged
except Exception:
return json.loads(json.dumps(DEFAULT_SOURCES))
def save_sources(data):
ensure_dirs()
SOURCES_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8")
DEFAULT_MANIFEST = {
"manifest_version": 1,
"components": {
"tor": {
"enabled": True,
"target_dir": "tor",
"binary_name": "tor.exe",
"version_source": {"kind": "torproject_listing"},
"url_templates": [
"https://dist.torproject.org/torbrowser/{version}/tor-expert-bundle-windows-x86_64-{version}.tar.gz",
"https://dist.torproject.org/torbrowser/{version}/tor-expert-bundle-windows-x86_64-{version}.zip",
"https://archive.torproject.org/tor-package-archive/torbrowser/{version}/tor-expert-bundle-windows-x86_64-{version}.tar.gz",
],
"sha256_url_template": "https://dist.torproject.org/torbrowser/{version}/sha256sums-signed-build.txt",
"winget_id": "",
},
"sing-box": {
"enabled": True,
"target_dir": "sing-box",
"binary_name": "sing-box.exe",
"version_source": {"kind": "github_api", "repo": "SagerNet/sing-box"},
"url_templates": [
"https://github.com/SagerNet/sing-box/releases/download/v{version}/sing-box-{version}-windows-amd64.zip",
],
"sha256_url_template": "https://github.com/SagerNet/sing-box/releases/download/v{version}/sing-box-{version}-windows-amd64.zip.sha256sum",
"winget_id": "",
},
"wintun": {
"enabled": True,
"target_dir": "sing-box",
"binary_name": "wintun.dll",
"post": "wintun",
"version_source": {"kind": "fixed", "version": WINTUN_VERSION},
"url_templates": ["https://www.wintun.net/builds/wintun-{version}.zip"],
"winget_id": "",
},
},
"program_update": {
"version_url": "",
"file_url_template": "",
"version_regex": r"([0-9]{4}\.[0-9]{2}\.[0-9]{2}(?:\.[0-9]+)?)",
},
}
DEFAULT_SOURCES = {
"manifest_mirrors": [],
"program_update_mirrors": [],
"program_file_url_template": "",
"winget_enabled": True,
"auto_discover": True,
"privacy_defaults": {"block_insecure_http": False, "block_webrtc_hosts": False},
}
DEFAULT_WINGET_MAP = {"7zip": "7zip.7zip"}
DEFAULT_DISCOVERY_SEEDS = []
DEFAULT_DISCOVERY_QUERIES = ["TorLeakGuard manifest", "tor-leak-guard", "TorLeakGuard in:name"]
def http_get_bytes(url, timeout=60):
req = urllib.request.Request(url, headers=HTTP_HEADERS)
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.read()
def http_get_text(url, timeout=60):
return http_get_bytes(url, timeout=timeout).decode("utf-8", errors="ignore")
def fetch_text_safe(url, timeout=20):
try:
return http_get_text(url, timeout=timeout)
except Exception:
return None
def url_exists(url):
try:
req = urllib.request.Request(url, method="HEAD", headers=HTTP_HEADERS)
with urllib.request.urlopen(req, timeout=30):
return True
except Exception:
return False
def download_file(url, dest, log):
dest = Path(dest)
dest.parent.mkdir(parents=True, exist_ok=True)
log(f"Downloading: {url}")
req = urllib.request.Request(url, headers=HTTP_HEADERS)
with urllib.request.urlopen(req, timeout=180) as resp, open(dest, "wb") as f:
while True:
chunk = resp.read(1024 * 1024)
if not chunk:
break
f.write(chunk)
log(f"Saved: {dest}")
def _load_json_file(p):
try:
p = Path(p)
if not p.exists():
return None
d = json.loads(p.read_text(encoding="utf-8"))
return d if isinstance(d, dict) else None
except Exception:
return None
def validate_manifest(d):
return isinstance(d, dict) and ("components" in d or "manifest_version" in d)
def manifest_vault_paths():
ps = [MANIFEST_CACHE_FILE, APP_DIR / ".manifest.vault"]
try:
ROAMING_DIR.mkdir(parents=True, exist_ok=True)
ps.append(ROAMING_DIR / ".manifest.vault")
except Exception:
pass
try:
ps.append(SCRIPT_PATH.with_name(SCRIPT_PATH.name + ".manifest.vault"))
except Exception:
pass
return ps
def sources_vault_paths():
ps = [SOURCES_FILE, APP_DIR / ".sources.vault"]
try:
ROAMING_DIR.mkdir(parents=True, exist_ok=True)
ps.append(ROAMING_DIR / ".sources.vault")
except Exception:
pass
try:
ps.append(SCRIPT_PATH.with_name(SCRIPT_PATH.name + ".sources.vault"))
except Exception:
pass
return ps
def _count_vaults(paths):
n = 0
for p in paths:
try:
if Path(p).exists():
n += 1
except Exception:
pass
return n
def save_manifest_everywhere(data, log):
text = json.dumps(data, indent=2, ensure_ascii=False)
for p in manifest_vault_paths():
try:
Path(p).parent.mkdir(parents=True, exist_ok=True)
Path(p).write_text(text, encoding="utf-8")
except Exception:
pass
def save_sources_everywhere(data, log):
text = json.dumps(data, indent=2, ensure_ascii=False)
for p in sources_vault_paths():
try:
Path(p).parent.mkdir(parents=True, exist_ok=True)
Path(p).write_text(text, encoding="utf-8")
except Exception:
pass
def github_discover(queries, log):
out = []
for q in queries:
api = (
"https://api.github.com/search/repositories?q="
+ urllib.parse.quote(q)
+ "&per_page=5&sort=updated"
)
text = fetch_text_safe(api, 20)
if not text:
continue
try:
data = json.loads(text)
except Exception:
continue
for it in (data.get("items") or [])[:5]:
owner = (it.get("owner") or {}).get("login")
repo = it.get("name")
br = it.get("default_branch") or "main"
if not owner or not repo:
continue
for path in (
"manifest.json",
"TorLeakGuard/manifest.json",
"torleakguard/manifest.json",
"data/manifest.json",
"build/manifest.json",
):
out.append(f"https://raw.githubusercontent.com/{owner}/{repo}/{br}/{path}")
return out
def ensure_restored(log):
if _RESTORED["done"]:
return
_RESTORED["done"] = True
if not validate_manifest(_load_json_file(MANIFEST_CACHE_FILE)):
healed = False
for p in manifest_vault_paths():
if Path(p) == Path(MANIFEST_CACHE_FILE):
continue
d = _load_json_file(p)
if validate_manifest(d):
try:
MANIFEST_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
MANIFEST_CACHE_FILE.write_text(json.dumps(d, indent=2), encoding="utf-8")
healed = True
log(f"Restored manifest cache from vault copy: {p}")
break
except Exception:
pass
if not healed:
try:
MANIFEST_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
MANIFEST_CACHE_FILE.write_text(
json.dumps(DEFAULT_MANIFEST, indent=2), encoding="utf-8"
)
log("Manifest cache was missing - recreated from the copy embedded inside the program.")
except Exception:
pass
if not SOURCES_FILE.exists():
healed = False
for p in sources_vault_paths():
if Path(p) == Path(SOURCES_FILE):
continue
d = _load_json_file(p)
if isinstance(d, dict):
try:
SOURCES_FILE.write_text(json.dumps(d, indent=2), encoding="utf-8")
healed = True
log(f"Restored sources from vault copy: {p}")
break
except Exception:
pass
if not healed:
save_sources(DEFAULT_SOURCES)
log("sources.json was missing - recreated from the defaults embedded inside the program.")
log(
f"Vault copies on disk: manifest={_count_vaults(manifest_vault_paths())}, "
f"sources={_count_vaults(sources_vault_paths())} (each saved in up to 4 places + embedded)."
)
def capability_inventory():
return [
"Tor + sing-box TUN full-PC routing (Windows 11)",
"Auto-install Tor / sing-box / Wintun (manifest-driven engine)",
"Manifest sources: explicit mirrors + embedded seeds + GitHub AUTO-discovery",
"Self-healing vaults: manifest & sources saved in 4 places + embedded-in-code copy",
"Auto-CREATE & FILL manifest on disk (resolved_version) even fully offline",
"winget MANDATORY: auto for 7-Zip + per-component winget_id + MANUAL search/install GUI",
"PT SOURCE OF TRUTH: reads Tor Browser's own torrc exec paths (obfs4/webtunnel work even if renamed/relocated)",
"PT auto-find fallback: winget Tor Browser folder + all-drives scan + manual 'Tor Browser folder' picker",
"Locale-safe subprocess (bytes+decode) -> winget search never crashes on cp1251/utf-8",
"Bridges: obfs4 / webtunnel only (snowflake & meek removed by design; exact Tor Browser exec line reused)",
"Bridge status proof: [ACTIVE] = real ESTABLISHED connection to that bridge IP",
"Anti-DPI: connection & circuit padding + one-click anti-DPI preset",
"IPv6 leak block via Windows Firewall (AddressFamily IPv6)",
"DNS via Tor (Tor DNSPort + sing-box hijack-dns)",
"WebRTC leak block (UDP disabled; optional TCP STUN/TURN host block)",
"Optional STRICT: block plain-HTTP (protects cookies on http sites)",
"External tracker block-list (tracklist.txt) - survives code updates",
"Internet probe after TUN with automatic rollback (never a fake 'Started')",
"Per-variant sing-box auto-selection (strict_route x stack x sniff)",
"Tor ControlPort: bootstrap progress + manual NEWNYM circuit change",
"RESET emergency eject: aborts tasks, force-kills by PID, no UAC, keeps window",
"Stop (restore internet); tun2socks kill; occupied-port freeing",
"Program self-update with .bak backup (URLs from manifest/sources)",
"Runtime capability inventory (this very list) to verify completeness",
"Portable per-user paths (%LOCALAPPDATA%) - works for any user / any PC",
"7-Zip auto-detect (PATH + registry + common dirs) and auto-install",
"Copy/paste + right-click context menu in bridge & log fields",
"Auto-discovery fills manifest; vaults heal on missing files",
"Manual winget search/install for any module when auto-detect misses it",
]
def fetch_remote_manifest(log):
sources = load_sources()
candidates = []
candidates += [u for u in (sources.get("manifest_mirrors") or []) if u]
candidates += [u for u in DEFAULT_DISCOVERY_SEEDS if u]
if sources.get("auto_discover", True):
candidates += github_discover(DEFAULT_DISCOVERY_QUERIES, log)
seen = set()
uniq = []
for u in candidates:
if u not in seen:
seen.add(u)
uniq.append(u)
log(f"Manifest discovery: {len(uniq)} candidate URL(s) (mirrors + seeds + GitHub auto-search).")
for u in uniq:
text = fetch_text_safe(u, timeout=20)
if not text:
continue
try:
data = json.loads(text)
except Exception:
continue
if validate_manifest(data):
save_manifest_everywhere(data, log)
log(f"Manifest loaded from remote: {u}")
return data, "remote"
for p in manifest_vault_paths():
data = _load_json_file(p)
if validate_manifest(data):
save_manifest_everywhere(data, log)
log(f"Manifest loaded from local vault copy: {p}")
return data, "vault"
log("Manifest: no remote/vault available - using the copy embedded inside the program.")
return json.loads(json.dumps(DEFAULT_MANIFEST)), "embedded"
def merge_manifest(remote):
base = json.loads(json.dumps(DEFAULT_MANIFEST))
if not isinstance(remote, dict):
return base
base_comps = base.setdefault("components", {})
for name, spec in (remote.get("components") or {}).items():
if isinstance(spec, dict):
cur = dict(base_comps.get(name, {}))
cur.update(spec)
base_comps[name] = cur
if isinstance(remote.get("program_update"), dict):
pu = dict(base.get("program_update", {}))
pu.update(remote["program_update"])
base["program_update"] = pu
return base
def _ver_torproject_listing(log):
try:
html = http_get_text("https://dist.torproject.org/torbrowser/", timeout=30)
vs = re.findall(r'href="(\d+\.\d+(?:\.\d+)?)/"', html)
vs = sorted(set(vs), key=lambda v: tuple(int(x) for x in v.split(".")))
return vs[-1] if vs else None
except Exception as e:
log(f"torproject listing version parse failed: {e}")
return None
def _ver_github_api(repo, log):
try:
data = json.loads(
http_get_text(f"https://api.github.com/repos/{repo}/releases/latest", timeout=30)
)
return data.get("tag_name", "").lstrip("v") or None
except Exception as e:
log(f"github api version parse failed ({repo}): {e}")
return None
def _ver_regex_url(spec, log):
u = spec.get("url")
rgx = spec.get("regex")
if not u or not rgx:
return None
text = fetch_text_safe(u, timeout=30)
if not text:
return None
m = re.search(rgx, text)
return m.group(1) if m else None
def resolve_version(spec, log):
vs = spec.get("version_source") or {}
kind = vs.get("kind")
if kind == "fixed":
return vs.get("version")
if kind == "torproject_listing":
return _ver_torproject_listing(log)
if kind == "github_api":
return _ver_github_api(vs.get("repo", ""), log)
if kind == "regex_url":
return _ver_regex_url(vs, log)
return None
def winget_available():
return bool(shutil.which("winget"))
def winget_id_for(name, spec):
wid = (spec or {}).get("winget_id")
if wid:
return wid
return DEFAULT_WINGET_MAP.get(name, "")
def winget_install(pkg_id, log, upgrade=False):
if not winget_available():
log("winget not available on this system.")
return False
verb = "upgrade" if upgrade else "install"
log(f"winget {verb}: {pkg_id}")
rc, out, err = run_cmd(
[
"winget", verb, "--id", pkg_id, "-e", "--silent",
"--accept-package-agreements", "--accept-source-agreements",
],
timeout=600,
)
tail = (out + err).strip()
if tail:
log("winget output: " + tail[-400:])
return rc == 0
def _parse_winget_table(text):
results = []
id_re = re.compile(r'^[A-Za-z0-9_-]+(\.[A-Za-z0-9_-]+)+$')
ver_re = re.compile(r'^\d')
for line in text.splitlines():
s = line.rstrip()
if not s.strip():
continue
if set(s.strip()) <= set("- "):
continue
if "Name" in s and "Id" in s and "Version" in s:
continue
toks = s.split()
if len(toks) < 2:
continue
id_idx = -1
for i, t in enumerate(toks):
if id_re.match(t) and re.search(r'[A-Za-z]', t):
id_idx = i
break
if id_idx <= 0:
continue
pkg_id = toks[id_idx]
ver = ""
for t in toks[id_idx + 1:]:
if ver_re.match(t):
ver = t
break
name = " ".join(toks[:id_idx]).strip()
if name and pkg_id:
results.append((name, pkg_id, ver))
return results
def winget_search(query, log):
if not winget_available():
log("winget not available on this system.")
return []
log(f"winget search: {query}")
rc, out, err = run_cmd(
["winget", "search", "--query", query, "--accept-source-agreements"],
timeout=120,
)
results = _parse_winget_table(out + err)
log(f"winget search returned {len(results)} result(s).")
return results
def winget_list(query="", log=None):
if not winget_available():
return []
cmd = ["winget", "list", "--accept-source-agreements"]
if query:
cmd += ["--query", query]
rc, out, err = run_cmd(cmd, timeout=120)
return _parse_winget_table(out + err)
def _winget_package_dirs():
dirs = []
la = os.environ.get("LOCALAPPDATA", "")
if la:
dirs.append(Path(la) / "Microsoft" / "WinGet" / "Links")
dirs.append(Path(os.environ.get("ProgramFiles", "C:/Program Files")))
dirs.append(Path(os.environ.get("ProgramFiles(x86)", "C:/Program Files (x86)")))
return dirs
def _winget_packages_root():
la = os.environ.get("LOCALAPPDATA", "")
if not la:
return None
p = Path(la) / "Microsoft" / "WinGet" / "Packages"
return p if p.exists() else None
def _logical_drives():
drives = []
try:
buf = ctypes.create_unicode_buffer(256)
n = ctypes.windll.kernel32.GetLogicalDriveStringsW(255, buf)
if n:
raw = buf[:n]
drives = [d for d in raw.split("\x00") if d and len(d) >= 3 and d[1] == ":"]
except Exception:
pass
if not drives:
for letter in string.ascii_uppercase:
drives.append(f"{letter}:\\")
return drives
def _walk_for_tb_folders(root, max_depth=5, _depth=0):
found = []
if _depth > max_depth:
return found
try:
entries = list(Path(root).iterdir())
except Exception:
return found
for e in entries:
try:
if not e.is_dir():
continue
except Exception:
continue
low = e.name.lower()
if low in ("tor browser", "torbrowser") or low.startswith("tor browser") or low.startswith("torbrowser"):
found.append(e)
continue
found += _walk_for_tb_folders(e, max_depth, _depth + 1)
return found
_DRIVE_SCAN_CACHE = None
def scan_drives_for_torbrowser(log=None, force=False):
global _DRIVE_SCAN_CACHE
if _DRIVE_SCAN_CACHE is not None and not force:
return _DRIVE_SCAN_CACHE
found = []
pkg = _winget_packages_root()
if pkg is not None:
found += _walk_for_tb_folders(pkg, max_depth=6)
for d in _logical_drives():
try:
if not Path(d).exists():
continue
except Exception:
continue
found += _walk_for_tb_folders(d, max_depth=5)
seen = set()
uniq = []
for p in found:
try:
rp = p.resolve()
except Exception:
rp = p
key = str(rp).lower()
if key not in seen:
seen.add(key)
uniq.append(p)
_DRIVE_SCAN_CACHE = uniq
if log is not None:
log(f"Drive scan: {len(uniq)} Tor Browser folder(s) located.")
for p in uniq[:12]:
log(f" - {p}")
return uniq
def tor_browser_pt_dirs():
dirs = []
for env in ("ProgramFiles", "ProgramFiles(x86)", "LOCALAPPDATA"):
base = os.environ.get(env)
if base:
dirs.append(Path(base) / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "PluggableTransports")
dirs.append(Path(base) / "Tor Browser" / "Browser" / "TorBrowser" / "Tor")
h = Path.home()
dirs.append(h / "Desktop" / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "PluggableTransports")
dirs.append(h / "Downloads" / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "PluggableTransports")
for tb in scan_drives_for_torbrowser():
dirs.append(tb / "Browser" / "TorBrowser" / "Tor" / "PluggableTransports")
dirs.append(tb / "TorBrowser" / "Tor" / "PluggableTransports")
dirs.append(tb)
return dirs
# ---------------------------------------------------------------------------
# SOURCE OF TRUTH: read Tor Browser's own torrc for ClientTransportPlugin lines.
# Finds obfs4/webtunnel regardless of binary name/location.
# ---------------------------------------------------------------------------
def _parse_exec_path(tail):
tail = tail.strip()
if not tail:
return ""
if tail[0] == '"':
end = tail.find('"', 1)
if end > 0:
return tail[1:end]
return tail[1:]
return tail.split(None, 1)[0]
def find_torbrowser_torrc():
cands = []
static_bases = []
for env in ("ProgramFiles", "ProgramFiles(x86)", "LOCALAPPDATA"):
b = os.environ.get(env)
if b:
static_bases.append(Path(b) / "Tor Browser")
static_bases.append(Path.home() / "Desktop" / "Tor Browser")
static_bases.append(Path.home() / "Downloads" / "Tor Browser")
for tb in static_bases:
cands += [
tb / "Browser" / "TorBrowser" / "Data" / "Tor" / "torrc",
tb / "TorBrowser" / "Data" / "Tor" / "torrc",
tb / "Data" / "Tor" / "torrc",
]
for tb in scan_drives_for_torbrowser():
cands += [
tb / "Browser" / "TorBrowser" / "Data" / "Tor" / "torrc",
tb / "TorBrowser" / "Data" / "Tor" / "torrc",
tb / "Data" / "Tor" / "torrc",
tb / "torrc",
]
for c in cands:
try:
if c.exists():
return c
except Exception:
pass
return None
def get_tb_pt_map(force=False, log=None):
global _TB_PT_MAP, TB_EXEC_TAIL, TB_EXEC_EXE
if _TB_PT_MAP is not None and not force:
return _TB_PT_MAP
TB_EXEC_TAIL = {}
TB_EXEC_EXE = {}
_TB_PT_MAP = {}
trc = find_torbrowser_torrc()
if trc is None:
if log:
log("Tor Browser torrc not found (Tor Browser may never have run, or portable layout differs).")
return _TB_PT_MAP
try:
text = trc.read_text(encoding="utf-8", errors="ignore")
except Exception as e:
if log:
log(f"Could not read Tor Browser torrc: {e}")
return _TB_PT_MAP
if log:
log(f"Reading Tor Browser torrc (source of truth): {trc}")
torrc_dir = trc.parent
rx = re.compile(r'(?im)^\s*ClientTransportPlugin\s+([^\n]+?)\s+exec\s+(.+?)\s*$')
for m in rx.finditer(text):
names_part = m.group(1).strip()
tail = m.group(2).strip()
exe = _parse_exec_path(tail)
if exe and not os.path.isabs(exe):
exe = str((torrc_dir / exe).resolve())
for nm in names_part.split():
_TB_PT_MAP[nm] = (exe, tail)
TB_EXEC_TAIL[nm] = tail
TB_EXEC_EXE[nm] = exe
if log:
if _TB_PT_MAP:
log(f"Tor Browser torrc declares PT plugins: {sorted(_TB_PT_MAP.keys())}")
for nm, (exe, tail) in _TB_PT_MAP.items():
exists = bool(exe and Path(exe).exists())
log(f" [TB-torrc] {nm} -> {exe} (exists={exists})")
else:
log("Tor Browser torrc has no ClientTransportPlugin lines (this build may ship no PT binaries).")
return _TB_PT_MAP
def copy_pt_from_torbrowser_config(log):
"""Copy PT binaries referenced by Tor Browser's torrc into our PT_DIR."""
tb = get_tb_pt_map(force=True, log=log)
ensure_dirs()
copied = 0
for nm, (exe, tail) in tb.items():
if not exe:
continue
ep = Path(exe)
if not ep.exists():
log(f" skip {nm}: file not present at {exe}")
continue
try:
shutil.copy2(ep, PT_DIR / ep.name)
copied += 1
log(f"Copied PT for {nm}: {ep} -> {PT_DIR / ep.name}")
try:
for sib in ep.parent.iterdir():
if sib.is_file() and sib.suffix.lower() == ".dll" and sib.name.lower() != ep.name.lower():
try:
shutil.copy2(sib, PT_DIR / sib.name)
except Exception:
pass
except Exception:
pass
except Exception as e:
log(f"Could not copy {nm}: {e}")
if copied:
log(f"Copied {copied} PT binary(ies) from Tor Browser config into {PT_DIR}.")
else:
log("No copyable PT binaries found via Tor Browser torrc.")
return copied
def copy_pt_from_folder(folder, log):
folder = Path(folder)
if not folder.exists():
log(f"Folder does not exist: {folder}")
return 0
ensure_dirs()
copied = 0
for name in TOR_PT_NAMES:
try:
hit = next((p for p in folder.rglob(name) if p.is_file()), None)
except Exception:
hit = None
if hit:
try:
shutil.copy2(hit, PT_DIR / name)
copied += 1
log(f"Copied {name} from {hit} -> {PT_DIR / name}")
except Exception as e:
log(f"Could not copy {name}: {e}")
if copied == 0:
log(f"No Tor PT binaries by known name found under {folder} (trying Tor Browser torrc next).")
copied += copy_pt_from_torbrowser_config(log)
else:
log(f"Copied {copied} PT file(s) into {PT_DIR}. They will now be used by our tor.exe.")
return copied
def find_in_system(names):
found = {}
for n in names:
w = shutil.which(n)
if w:
found[n] = Path(w)
quick = _winget_package_dirs() + tor_browser_pt_dirs() + [TOR_DIR, PT_DIR, APP_DIR, SINGBOX_DIR]
for n in names:
if n in found:
continue
for d in quick:
try:
c = d / n
if c.exists():
found[n] = c
break
except Exception:
pass
pkg = _winget_packages_root()
if pkg is not None:
for n in names:
if n in found:
continue
try:
for p in pkg.rglob(n):
if p.is_file():
found[n] = p
break
except Exception:
pass
tb = get_tb_pt_map()
for nm, (exe, tail) in tb.items():
if exe and Path(exe).exists():
found.setdefault(Path(exe).name, Path(exe))
return found
def _find_binary(target_dir, binary_name):
root = Path(target_dir)
if not root.exists():
return None
try:
for p in root.rglob(binary_name):
if p.is_file():
return p
except Exception:
pass
return None
def _extract_to(archive, dest_dir, log):
dest_dir = Path(dest_dir)
dest_dir.mkdir(parents=True, exist_ok=True)
if archive.name.lower().endswith(".exe"):
if not ensure_7zip(log):
return False
return extract_with_7z(archive, dest_dir, log)
try:
extract_archive(archive, dest_dir, log)
return True
except Exception as e:
log(f"Extraction error: {e}")
return False
def _winget_last_resort(name, spec, target_dir, binary_name, log):
if not load_sources().get("winget_enabled", True) or not winget_available():
return False
wid = winget_id_for(name, spec)
if wid:
log(f"[manifest] {name}: winget fallback install ({wid})")
return winget_install(wid, log)
log(f"[manifest] {name}: no winget_id known; use manual winget search in the GUI to install it.")
return False
def install_component_generic(name, spec, manifest, log, force=False):
if not spec.get("enabled", True):
return False
target_dir = APP_DIR / spec.get("target_dir", name)
binary_name = spec.get("binary_name")
versions = load_versions()
if not force and binary_name:
existing = _find_binary(target_dir, binary_name)
if existing and versions.get(f"manifest_{name}"):
log(f"[manifest] {name}: already present ({existing})")
return True
version = resolve_version(spec, log)
if not version:
log(f"[manifest] {name}: could not resolve version")
return _winget_last_resort(name, spec, target_dir, binary_name, log)
log(f"[manifest] {name}: resolved version {version}")
archive_path = None
for tpl in (spec.get("url_templates") or []):
check_abort()
url = tpl.replace("{version}", version)
fname = url.rsplit("/", 1)[-1].split("?")[0] or f"{name}-{version}.bin"
dest = DOWNLOAD_DIR / fname
try:
download_file(url, dest, log)
archive_path = dest
break
except Exception as e:
log(f"[manifest] {name}: download failed ({url}): {e}")
safe_unlink(dest)
continue
if not archive_path:
log(f"[manifest] {name}: all URL templates failed -> trying winget")
return _winget_last_resort(name, spec, target_dir, binary_name, log)
sha_tpl = spec.get("sha256_url_template")
if sha_tpl:
sums_url = sha_tpl.replace("{version}", version).replace("{filename}", archive_path.name)
sums = fetch_text_safe(sums_url, timeout=30)
if sums:
try:
verify_sha256_from_text(sums, archive_path.name, archive_path, log)
except ValueError:
safe_unlink(archive_path)
log(f"[manifest] {name}: SHA256 mismatch, aborting component")
return False
except Exception:
pass
if target_dir.exists() and spec.get("post") != "wintun":
shutil.rmtree(target_dir, ignore_errors=True)
ok = _extract_to(archive_path, target_dir, log)
safe_unlink(archive_path)
if not ok:
log(f"[manifest] {name}: extraction failed -> trying winget")
return _winget_last_resort(name, spec, target_dir, binary_name, log)
if spec.get("post") == "wintun":
src = find_wintun_dll_in_tree(target_dir)
if not src:
log(f"[manifest] {name}: wintun.dll not found after extraction")
return False
shutil.copy2(src, SINGBOX_DIR / "wintun.dll")
shutil.copy2(src, APP_DIR / "wintun.dll")
se = get_singbox_exe()
if se:
shutil.copy2(src, Path(se).parent / "wintun.dll")
found = _find_binary(target_dir, binary_name) if binary_name else True
if not found:
log(f"[manifest] {name}: binary {binary_name} not found after extraction -> trying winget")
return _winget_last_resort(name, spec, target_dir, binary_name, log)
versions[f"manifest_{name}"] = version
save_versions(versions)
log(f"[manifest] {name}: installed v{version} -> {found if isinstance(found, Path) else target_dir}")
return True
def apply_manifest_install(log, force=False):
remote, origin = fetch_remote_manifest(log)
manifest = merge_manifest(remote)
log(
f"Manifest origin: {origin} (components: "
f"{', '.join(sorted((manifest.get('components') or {}).keys()))})"
)
results = {}
for name, spec in (manifest.get("components") or {}).items():
check_abort()
ok = False
try:
ok = install_component_generic(name, spec, manifest, log, force=force)
except Aborted:
raise
except Exception as e:
log(f"[manifest] {name}: generic install error: {e}")
if not ok and name == "tor":
log("[manifest] tor: built-in installer fallback")
try:
install_tor(log, force=force)
ok = True
except Aborted:
raise
except Exception as e:
log(f"built-in tor install error: {e}")
elif not ok and name == "sing-box":
log("[manifest] sing-box: built-in installer fallback")
try:
install_singbox(log, force=force)
ok = True
except Aborted:
raise
except Exception as e:
log(f"built-in sing-box install error: {e}")
elif not ok and name == "wintun":
log("[manifest] wintun: built-in installer fallback")
try:
install_wintun(log, force=force)
ok = True
except Aborted:
raise
except Exception as e:
log(f"built-in wintun install error: {e}")
results[name] = ok
vers = load_versions()
for nm in (manifest.get("components") or {}):
rv = vers.get(f"manifest_{nm}") or vers.get(nm)
if rv:
manifest["components"][nm]["resolved_version"] = rv
save_manifest_everywhere(manifest, log)
return manifest, results
def _program_update_info(manifest):
sources = load_sources()
pu = (manifest or {}).get("program_update") or {}
version_url = pu.get("version_url") or ""
file_tpl = pu.get("file_url_template") or sources.get("program_file_url_template") or ""
mirrors = sources.get("program_update_mirrors", []) or []
regex = pu.get("version_regex") or r"([0-9]+\.[0-9]+\.[0-9]+)"
return version_url, file_tpl, mirrors, regex
def check_program_update(log, manifest):
version_url, file_tpl, mirrors, regex = _program_update_info(manifest)
remote_ver = None
cand = ([version_url] if version_url else []) + [m for m in mirrors if m]
for u in cand:
text = fetch_text_safe(u, timeout=20)
if not text:
continue
m = re.search(regex, text)
if m:
remote_ver = m.group(1)
log(f"Program remote version: {remote_ver} (from {u})")
break
if not remote_ver:
log("Program update: no reachable version source (set program_update in manifest / sources.json).")
return False, None, None
def key(v):
return [int(x) if x.isdigit() else 0 for x in re.split(r"[.\-]", v)]
if key(remote_ver) <= key(PROGRAM_VERSION):
log(f"Program is up to date (local {PROGRAM_VERSION} >= remote {remote_ver}).")
return False, remote_ver, None
file_url = file_tpl.replace("{version}", remote_ver) if file_tpl else ""
log(f"Program update available: {PROGRAM_VERSION} -> {remote_ver}")
return True, remote_ver, file_url or None
def do_program_update(log, manifest):
has, ver, file_url = check_program_update(log, manifest)
if not has or not file_url:
log("Program update: nothing to download (no file URL configured).")
return False
try:
cur = Path(sys.argv[0]).resolve()
new_path = cur.with_name(cur.name + ".new")
bak_path = cur.with_name(cur.name + ".bak")
download_file(file_url, new_path, log)
try:
new_path.read_text(encoding="utf-8")
except Exception:
safe_unlink(new_path)
log("Program update: downloaded file is not valid text, aborted.")
return False
try:
if cur.exists():
shutil.copy2(cur, bak_path)
except Exception as e:
log(f"Program update: backup warning: {e}")
log(f"Program update downloaded to: {new_path}")
log(f"Backup of current file: {bak_path}")
log("To apply: close this program and run the .new file (rename over the old one if you want).")
return True
except Exception as e:
log(f"Program update download error: {e}")
return False
def sha256_file(path):
h = hashlib.sha256()
with open(path, "rb") as f:
while True:
chunk = f.read(1024 * 1024)
if not chunk:
break
h.update(chunk)
return h.hexdigest().lower()
def verify_sha256_from_text(sum_text, filename, filepath, log):
actual = sha256_file(filepath)
for line in sum_text.splitlines():
if filename in line:
parts = line.split()
if not parts:
continue
expected = parts[0].strip().lower().lstrip("*")
if expected == actual:
log(f"SHA256 OK: {filename}")
return True
raise ValueError(
f"SHA256 mismatch for {filename}\nexpected: {expected}\nactual: {actual}"
)
log(f"SHA256 for {filename} not found, check skipped.")
return False
def extract_zip(zip_path, dest_dir, log):
zip_path = Path(zip_path)
dest_dir = Path(dest_dir)
dest_dir.mkdir(parents=True, exist_ok=True)
log(f"Extracting ZIP {zip_path.name} -> {dest_dir}")
with zipfile.ZipFile(zip_path) as zf:
zf.extractall(dest_dir)
def extract_archive(path, dest_dir, log):
path = Path(path)
dest_dir = Path(dest_dir)
dest_dir.mkdir(parents=True, exist_ok=True)
log(f"Extracting {path.name} -> {dest_dir}")
low = path.name.lower()
if low.endswith(".zip"):
with zipfile.ZipFile(path) as zf:
zf.extractall(dest_dir)
elif low.endswith((".tar.gz", ".tgz", ".tar.xz", ".txz", ".tar")):
with tarfile.open(path, "r:*") as tf:
try:
tf.extractall(dest_dir, filter="data")
except TypeError:
tf.extractall(dest_dir)
else:
raise RuntimeError(f"Unsupported archive format: {path.name}")
_SEVENZIP_CACHE = None
def read_7z_from_registry():
found = []
try:
import winreg
except Exception:
return found
for hive in (winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER):
for kp in (r"SOFTWARE\7-Zip", r"SOFTWARE\WOW6432Node\7-Zip"):
try:
key = winreg.OpenKey(hive, kp)
except OSError:
continue
try:
i = 0
while True:
try:
name, value, _ = winreg.EnumValue(key, i)
except OSError:
break
if isinstance(value, str) and value.strip():
p = Path(value.strip())
if p.exists():
found.append(p)
i += 1
finally:
try:
winreg.CloseKey(key)
except Exception:
pass
return found
def find_7z():
global _SEVENZIP_CACHE
if _SEVENZIP_CACHE and _SEVENZIP_CACHE.exists():
return _SEVENZIP_CACHE
for n in ("7z.exe", "7zz.exe", "7za.exe"):
w = shutil.which(n)
if w:
_SEVENZIP_CACHE = Path(w)
return _SEVENZIP_CACHE
dirs = [
Path("C:/Program Files/7-Zip"),
Path("C:/Program Files (x86)/7-Zip"),
Path(os.environ.get("LOCALAPPDATA", "")) / "7-Zip",
Path(os.environ.get("ProgramFiles", "")) / "7-Zip",
Path(os.environ.get("ProgramFiles(x86)", "")) / "7-Zip",
Path("C:/7-Zip"),
Path("D:/7-Zip"),
Path("E:/7-Zip"),
]
for rp in read_7z_from_registry():
dirs.append(rp)
for d in dirs:
try:
if not d or not d.exists():
continue
except Exception:
continue
for n in ("7z.exe", "7zz.exe", "7za.exe"):
c = d / n
if c.exists():
_SEVENZIP_CACHE = c
return c
return None
def add_7z_to_path():
exe = find_7z()
extra = ([exe.parent] if exe else []) + [
Path("C:/Program Files/7-Zip"),
Path("C:/Program Files (x86)/7-Zip"),
]
cur = os.environ.get("PATH", "")
add = [str(d) for d in extra if d.exists() and str(d) not in cur]
if add:
os.environ["PATH"] = os.pathsep.join(add) + os.pathsep + cur
def extract_with_7z(archive, dest_dir, log):
exe = find_7z()
if not exe:
log("7-Zip not found - cannot extract exe archive.")
return False
dest_dir = Path(dest_dir)
dest_dir.mkdir(parents=True, exist_ok=True)
log(f"Extracting via 7-Zip ({exe})...")
rc, out, err = run_cmd([str(exe), "x", str(archive), f"-o{dest_dir}", "-y"], timeout=600)
log(f"7-Zip returncode={rc} (0 and 1 = success)")
return rc in (0, 1)
def find_file(root, name):
root = Path(root)
if not root.exists():
return None
try:
for p in root.rglob(name):
if p.is_file():
return p
except Exception:
pass
return None
def find_wintun_dll_in_tree(root):
root = Path(root)
if not root.exists():
return None
cands = []
try:
for p in root.rglob("wintun.dll"):
if p.is_file():
cands.append(p)
except Exception:
pass
if not cands:
return None
def score(p):
low = str(p).lower().replace("\\", "/")
if "/amd64/" in low or "/x64/" in low:
return 0
if "/arm64/" in low:
return 2
if "/x86/" in low or "/386/" in low:
return 3
return 1
cands.sort(key=score)
return cands[0]
def get_tor_exe():
return find_file(TOR_DIR, "tor.exe")
def get_singbox_exe():
return find_file(SINGBOX_DIR, "sing-box.exe")
def get_wintun_dll():
for c in (SINGBOX_DIR / "wintun.dll", APP_DIR / "wintun.dll"):
if c.exists():
return c
found = find_file(SINGBOX_DIR, "wintun.dll")
if found:
return found
sys32 = Path(os.environ.get("SystemRoot", "C:\\Windows")) / "System32" / "wintun.dll"
return sys32 if sys32.exists() else None
def components_status(log):
tor = get_tor_exe()
sing = get_singbox_exe()
wintun = get_wintun_dll()
log(f"tor.exe: {tor if tor else 'not found'}")
log(f"sing-box.exe: {sing if sing else 'not found'}")
log(f"wintun.dll: {wintun if wintun else 'not found'}")
return bool(tor and sing and wintun)
def get_latest_tor_version(log):
log("Fetching Tor version list...")
html = http_get_text("https://dist.torproject.org/torbrowser/")
vs = re.findall(r'href="(\d+\.\d+(?:\.\d+)?)/"', html)
vs = sorted(set(vs), key=lambda v: tuple(int(x) for x in v.split(".")))
if not vs:
raise RuntimeError("Could not get Tor version list from dist.torproject.org")
latest = vs[-1]
log(f"Latest Tor version: {latest}")
return latest
def get_tor_expert_bundle_url(version, log):
base = f"https://dist.torproject.org/torbrowser/{version}/"
try:
html = http_get_text(base)
links = re.findall(r'href="([^"]+)"', html)
cands = []
for link in links:
name = link.split("/")[-1]
if not name:
continue
low = name.lower()
if (
"expert-bundle" in low
and "windows" in low
and "x86_64" in low
and (low.endswith(".zip") or low.endswith(".tar.gz") or low.endswith(".tar.xz"))
):
url = link if link.startswith("http") else base + name
pr = 0 if low.endswith(".zip") else (1 if low.endswith(".tar.gz") else 2)
cands.append((pr, url, name))
if cands:
cands.sort(key=lambda x: x[0])
return cands[0][1], cands[0][2]
except Exception as e:
log(f"Could not get Tor Expert Bundle file list: {e}")
for name in (
f"tor-expert-bundle-windows-x86_64-{version}.zip",
f"tor-expert-bundle-windows-x86_64-{version}.tar.gz",
f"tor-expert-bundle-windows-x86_64-{version}.tar.xz",
):
url = base + name
if url_exists(url):
return url, name
raise RuntimeError("Tor Expert Bundle file for Windows x86_64 not found.")
def install_tor(log, force=False):
ensure_dirs()
check_abort()
exe = get_tor_exe()
versions = load_versions()
latest = get_latest_tor_version(log)
if exe and versions.get("tor") == latest and not force:
log(f"Tor already installed and up to date: {latest}")
return latest
if exe and force and versions.get("tor") == latest:
log(f"Tor is up to date: {latest}. Reinstall not needed.")
return latest
url, fn = get_tor_expert_bundle_url(latest, log)
ap = DOWNLOAD_DIR / fn
download_file(url, ap, log)
check_abort()
try:
st = http_get_text(f"https://dist.torproject.org/torbrowser/{latest}/sha256sums-signed-build.txt")
verify_sha256_from_text(st, fn, ap, log)
except urllib.error.HTTPError:
log("Tor SHA256 file not found, check skipped.")
except ValueError:
safe_unlink(ap)
raise
except Exception as e:
log(f"Tor SHA256 check skipped: {e}")
if TOR_DIR.exists():
log("Removing old Tor folder...")
shutil.rmtree(TOR_DIR, ignore_errors=True)
extract_archive(ap, TOR_DIR, log)
safe_unlink(ap)
versions = load_versions()
versions["tor"] = latest
save_versions(versions)
te = get_tor_exe()
if not te:
raise RuntimeError("Tor extracted but tor.exe not found.")
log(f"Tor installed: {te}")
return latest
def get_latest_singbox_release(log):
log("Fetching latest sing-box version...")
data = json.loads(http_get_text("https://api.github.com/repos/SagerNet/sing-box/releases/latest"))
version = data.get("tag_name", "").lstrip("v")
if not version:
raise RuntimeError("Could not get sing-box version from GitHub API")
an = f"sing-box-{version}-windows-amd64.zip"
au = None
for a in data.get("assets", []):
if a.get("name") == an:
au = a.get("browser_download_url")
break
if not au:
au = f"https://github.com/SagerNet/sing-box/releases/download/v{version}/{an}"
log(f"Latest sing-box version: {version}")
return version, au, an
def install_singbox(log, force=False):
ensure_dirs()
check_abort()
exe = get_singbox_exe()
versions = load_versions()
latest, url, an = get_latest_singbox_release(log)
if exe and versions.get("sing-box") == latest and not force:
log(f"sing-box up to date: {latest}")
return latest
if exe and force and versions.get("sing-box") == latest:
log(f"sing-box up to date: {latest}")
return latest
zp = DOWNLOAD_DIR / an
download_file(url, zp, log)
check_abort()
try:
st = http_get_text(url + ".sha256sum")
verify_sha256_from_text(st, an, zp, log)
except urllib.error.HTTPError:
log("sing-box SHA256 not found, skipped.")
except ValueError:
safe_unlink(zp)
raise
except Exception as e:
log(f"sing-box SHA256 skipped: {e}")
if SINGBOX_DIR.exists():
log("Removing old sing-box folder...")
shutil.rmtree(SINGBOX_DIR, ignore_errors=True)
extract_zip(zp, SINGBOX_DIR, log)
safe_unlink(zp)
versions = load_versions()
versions["sing-box"] = latest
save_versions(versions)
se = get_singbox_exe()
if not se:
raise RuntimeError("sing-box extracted but sing-box.exe not found.")
log(f"sing-box installed: {se}")
return latest
def install_wintun(log, force=False):
ensure_dirs()
check_abort()
dll = get_wintun_dll()
versions = load_versions()
if dll and versions.get("wintun") == WINTUN_VERSION and not force:
log(f"Wintun installed: {dll}")
return WINTUN_VERSION
if dll and force and versions.get("wintun") == WINTUN_VERSION:
log("Wintun up to date.")
return WINTUN_VERSION
zn = f"wintun-{WINTUN_VERSION}.zip"
zp = DOWNLOAD_DIR / zn
td = DOWNLOAD_DIR / "wintun-extract"
download_file(f"https://www.wintun.net/builds/{zn}", zp, log)
check_abort()
if td.exists():
shutil.rmtree(td, ignore_errors=True)
extract_zip(zp, td, log)
src = find_wintun_dll_in_tree(td)
if not src:
raise RuntimeError("wintun.dll not found in Wintun archive")
log(f"Found wintun.dll in archive: {src}")
SINGBOX_DIR.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, SINGBOX_DIR / "wintun.dll")
shutil.copy2(src, APP_DIR / "wintun.dll")
se = get_singbox_exe()
if se:
shutil.copy2(src, Path(se).parent / "wintun.dll")
shutil.rmtree(td, ignore_errors=True)
safe_unlink(zp)
versions = load_versions()
versions["wintun"] = WINTUN_VERSION
save_versions(versions)
log("Wintun installed.")
return WINTUN_VERSION
def remove_ipv6_block(log):
run_cmd(
[
"powershell", "-NoProfile", "-Command",
"Get-NetFirewallRule -DisplayName 'TorLeakGuard*' -ErrorAction SilentlyContinue | "
"Remove-NetFirewallRule -ErrorAction SilentlyContinue",
]
)
log("TorLeakGuard IPv6 rules removed, if any existed.")
def add_ipv6_block(log, tor_exe=None, pt_exes=None):
pt_exes = pt_exes or []
cmds = [
"Get-NetFirewallRule -DisplayName 'TorLeakGuard*' -ErrorAction SilentlyContinue | "
"Remove-NetFirewallRule -ErrorAction SilentlyContinue"
]
if tor_exe and Path(tor_exe).exists():
path = str(Path(tor_exe).resolve())
cmds.append(
f"New-NetFirewallRule -Name {ps_quote('TorLeakGuard Allow IPv6 Tor')} "
f"-DisplayName {ps_quote('TorLeakGuard Allow IPv6 Tor')} "
f"-Direction Outbound -Action Allow -AddressFamily IPv6 -Program {ps_quote(path)}"
)
for i, pt in enumerate(pt_exes, start=1):
if pt and Path(pt).exists():
path = str(Path(pt).resolve())
cmds.append(
f"New-NetFirewallRule -Name {ps_quote(f'TorLeakGuard Allow IPv6 PT {i}')} "
f"-DisplayName {ps_quote(f'TorLeakGuard Allow IPv6 PT {i}')} "
f"-Direction Outbound -Action Allow -AddressFamily IPv6 -Program {ps_quote(path)}"
)
cmds.append(
f"New-NetFirewallRule -Name {ps_quote('TorLeakGuard Block IPv6 Out')} "
f"-DisplayName {ps_quote('TorLeakGuard Block IPv6 Out')} "
f"-Direction Outbound -Action Block -AddressFamily IPv6"
)
rc, out, err = run_cmd(["powershell", "-NoProfile", "-Command", "; ".join(cmds)])
if rc != 0:
log(f"Could not configure IPv6 rules: {err.strip()}")
else:
log("IPv6 firewall configured: Tor/PT allowed, other IPv6 outbound blocked.")
def reset_processes(log):
log("Stopping TorLeakGuard processes...")
run_silent(["powershell", "-NoProfile", "-Command", KILL_PS])
run_silent(["netsh", "interface", "set", "interface", "name=TorLeakGuard", "admin=disabled"])
remove_ipv6_block(log)
time.sleep(1)
def reset_files(log, full=False):
log("Cleaning working files...")
for f in [TORRC, SINGBOX_CONFIG, STATE_FILE, TOR_LOG, SINGBOX_LOG, SINGBOX_CONSOLE_LOG]:
safe_unlink(f)
if full:
log("Full clean of tor-data...")
if TOR_DATA.exists():
shutil.rmtree(TOR_DATA, ignore_errors=True)
log("Files cleaned.")
def warn_if_tun2socks(log):
rc, out, err = run_cmd(["tasklist", "/FO", "CSV", "/NH"])
for line in out.splitlines():
if "tun2socks" in line.lower():
log("WARNING: tun2socks process detected. It may conflict with sing-box TUN.")
log("If you don't need tun2socks - close it before starting.")
break
def _txt_select_all(w):
w.tag_add("sel", "1.0", "end")
return "break"
def _txt_copy(w):
try:
text = w.get("sel.first", "sel.last")
except tk.TclError:
return "break"
try:
w.clipboard_clear()
w.clipboard_append(text)
except tk.TclError:
pass
return "break"
def _txt_cut(w):
_txt_copy(w)
try:
w.delete("sel.first", "sel.last")
except tk.TclError:
pass
return "break"
def _txt_paste(w):
try:
text = w.clipboard_get()
except tk.TclError:
return "break"
try:
w.delete("sel.first", "sel.last")
except tk.TclError:
pass
try:
w.insert("insert", text)
except tk.TclError:
pass
return "break"
def bind_text_clipboard(w):
w.bind("<Control-a>", lambda e: _txt_select_all(e.widget))
w.bind("<Control-A>", lambda e: _txt_select_all(e.widget))
w.bind("<Control-c>", lambda e: _txt_copy(e.widget))
w.bind("<Control-C>", lambda e: _txt_copy(e.widget))
w.bind("<Control-Insert>", lambda e: _txt_copy(e.widget))
w.bind("<Control-x>", lambda e: _txt_cut(e.widget))
w.bind("<Control-X>", lambda e: _txt_cut(e.widget))
w.bind("<Shift-Delete>", lambda e: _txt_cut(e.widget))
w.bind("<Control-v>", lambda e: _txt_paste(e.widget))
w.bind("<Control-V>", lambda e: _txt_paste(e.widget))
w.bind("<Shift-Insert>", lambda e: _txt_paste(e.widget))
menu = tk.Menu(w, tearoff=0)
menu.add_command(label="Cut", command=lambda: _txt_cut(w))
menu.add_command(label="Copy", command=lambda: _txt_copy(w))
menu.add_command(label="Paste", command=lambda: _txt_paste(w))
menu.add_separator()
menu.add_command(label="Select all", command=lambda: _txt_select_all(w))
def _popup(event):
try:
menu.tk_popup(event.x_root, event.y_root)
finally:
menu.grab_release()
return "break"
w.bind("<Button-3>", _popup)
def parse_bridge_rows(text):
rows = []
for raw in text.splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue
toks = line.split()
if not toks:
continue
if toks[0].lower() == "bridge":
toks = toks[1:]
if len(toks) < 2:
continue
rows.append((toks[0], toks[1]))
return rows
def get_pt_runtime():
pids = set()
rc, out, err = run_cmd(
[
"powershell", "-NoProfile", "-Command",
"Get-Process -Name lyrebird,obfs4proxy "
"-ErrorAction SilentlyContinue | Select-Object -ExpandProperty Id",
]
)
for line in out.splitlines():
line = line.strip()
if line.isdigit():
pids.add(int(line))
remote = set()
if pids:
rc, out, err = run_cmd(["netstat", "-ano"])
for line in out.splitlines():
parts = line.split()
if len(parts) < 5:
continue
if parts[3] != "ESTABLISHED":
continue
pid = parts[4]
if pid.isdigit() and int(pid) in pids:
remote.add(parts[2].strip())
return pids, remote
def read_tor_log_tail(max_bytes=300000):
try:
if not TOR_LOG.exists():
return ""
size = TOR_LOG.stat().st_size
with open(TOR_LOG, "rb") as f:
if size > max_bytes:
f.seek(size - max_bytes)
return f.read().decode(errors="ignore")
except Exception:
return ""
def tor_quote_path(p):
s = Path(p).as_posix()
return f'"{s}"' if " " in s else s
def normalize_bridge_lines(text):
out = []
for raw in text.splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue
if re.match(r'(?i)^(UseBridges|ClientTransportPlugin)\b', line):
continue
m = re.match(r'(?i)^bridge\s+(.*)$', line)
if m:
line = m.group(1).strip()
if line:
out.append(f"Bridge {line}")
return out
def bridge_transports(bridge_lines):
tr = set()
for line in bridge_lines:
parts = line.split()
if len(parts) >= 2 and parts[0].lower() == "bridge":
t = parts[1].lower()
if t in KNOWN_PT:
tr.add(t)
return tr
def pt_group_satisfied_in(d):
d = Path(d)
return all(any((d / n).exists() for n in g) for g in PT_GROUPS)
def pt_ready():
return pt_group_satisfied_in(PT_DIR)
def pt_search_roots():
roots = [PT_DIR, TOR_DIR]
roots += tor_browser_pt_dirs()
roots += [TOR_BROWSER_TEMP_DIR, DOWNLOAD_DIR / "tor-browser-extract", APP_DIR / "tor-browser"]
return roots
def find_file_in_roots(names, roots):
for root in roots:
root = Path(root)
if not root.exists():
continue
for name in names:
try:
for p in root.rglob(name):
if p.is_file():
return p
except Exception:
pass
return None
def list_pt_like_files(root):
root = Path(root)
found = []
if not root.exists():
return found
try:
for p in root.rglob("*"):
if p.is_file() and any(k in p.name.lower() for k in ("lyrebird", "obfs")):
found.append(str(p))
except Exception:
pass
return found
def list_top_entries(root, log, limit=40):
root = Path(root)
if not root.exists():
log(f" folder does not exist: {root}")
return
try:
entries = sorted(os.listdir(root))
except Exception as e:
log(f" could not read {root}: {e}")
return
if not entries:
log(f" folder is EMPTY: {root}")
return
log(f" top level of {root}:")
for e in entries[:limit]:
log(" " + e)
if len(entries) > limit:
log(f" ...and {len(entries) - limit} more")
def copy_groups_from_roots(roots, log, overwrite=False):
copied = []
exclude = {Path(r).resolve() for r in roots if Path(r).resolve() == PT_DIR.resolve()}
for group in PT_GROUPS:
if not overwrite and any((PT_DIR / n).exists() for n in group):
continue
src = find_file_in_roots(group, [r for r in roots if Path(r).resolve() not in exclude])
if src:
try:
shutil.copy2(src, PT_DIR / src.name)
copied.append(src.name)
log(f"Copied PT file: {src} -> {PT_DIR / src.name}")
except Exception as e:
log(f"Could not copy {src}: {e}")
return copied
def missing_group_names():
return [g for g in PT_GROUPS if not any((PT_DIR / n).exists() for n in g)]
def get_7zip_installer_url(log):
try:
html = http_get_text("https://www.7-zip.org/download.html")
links = re.findall(r'href="a/(7z\d+-x64\.exe)"', html)
if links:
def ver(s):
m = re.match(r'7z(\d+)-x64\.exe', s)
return int(m.group(1)) if m else 0
links = sorted(set(links), key=ver)
chosen = links[-1]
log(f"Found 7-Zip installer on site: {chosen}")
return "https://www.7-zip.org/a/" + chosen
except Exception as e:
log(f"Could not get 7-Zip download page: {e}")
for v in ("2409", "2408", "2407", "2406", "2301", "2201"):
url = f"https://www.7-zip.org/a/7z{v}-x64.exe"
if url_exists(url):
log(f"Using fallback 7-Zip installer: {url}")
return url
return "https://www.7-zip.org/a/7z2409-x64.exe"
def ensure_7zip(log):
add_7z_to_path()
ex = find_7z()
if ex:
log(f"7-Zip found automatically: {ex}")
return True
if not is_admin():
log("WARNING: no admin rights - 7-Zip install may fail.")
if winget_available():
log("winget: installing 7-Zip...")
rc, out, err = run_cmd(
[
"winget", "install", "--id", "7zip.7zip", "-e", "--silent",
"--accept-package-agreements", "--accept-source-agreements",
],
timeout=600,
)
if (out + err).strip():
log("winget output: " + (out + err).strip()[-600:])
add_7z_to_path()
if find_7z():
log("7-Zip installed via winget.")
return True
url = get_7zip_installer_url(log)
log(f"Downloading 7-Zip installer: {url}")
try:
dest = DOWNLOAD_DIR / Path(url).name
download_file(url, dest, log)
run_silent([str(dest), "/S"])
deadline = time.time() + 120
while time.time() < deadline:
add_7z_to_path()
if find_7z():
safe_unlink(dest)
log("7-Zip installed from installer.")
return True
time.sleep(2)
safe_unlink(dest)
add_7z_to_path()
log("7-Zip installer finished but 7z.exe not found in usual paths.")
except Exception as e:
log(f"7-Zip install error: {e}")
final = find_7z()
log(f"7-Zip after all attempts: {final if final else 'NOT found'}")
return bool(final)
def extract_tor_browser_archive(archive, extract_dir, log):
extract_dir = Path(extract_dir)
if extract_dir.exists():
shutil.rmtree(extract_dir, ignore_errors=True)
if archive.name.lower().endswith(".exe"):
if not ensure_7zip(log):
log("7-Zip unavailable - cannot extract Tor Browser exe.")
return False
return extract_with_7z(archive, extract_dir, log)
try:
extract_archive(archive, extract_dir, log)
return True
except Exception as e:
log(f"Tor Browser archive extraction error: {e}")
return False
def quiet_install_tor_browser(installer, install_dir, log):
install_dir = Path(install_dir)
if install_dir.exists():
shutil.rmtree(install_dir, ignore_errors=True)
log(f"Silent install of Tor Browser into {install_dir}")
run_silent([str(installer), "/S", f"/D={install_dir}"])
deadline = time.time() + 300
while time.time() < deadline:
if find_file_in_roots(ALL_PT_NAMES, [install_dir]):
time.sleep(3)
return True
time.sleep(2)
log("Silent Tor Browser install produced no PT files in time.")
return False
def install_pt_components(log, force=False):
ensure_dirs()
check_abort()
versions = load_versions()
latest = None
try:
latest = get_latest_tor_version(log)
except Exception as e:
log(f"Could not get latest Tor Browser version: {e}")
roots = pt_search_roots()
log("Collecting PT from local sources (tor bundle / winget Tor Browser / all drives / extracts)...")
copy_groups_from_roots(roots, log, overwrite=True)
if pt_ready():
log("PT components ready from local sources.")
versions["pt_attempted"] = True
save_versions(versions)
return True
log("Reading Tor Browser torrc (source of truth) for PT plugins...")
copy_pt_from_torbrowser_config(log)
if pt_ready():
log("PT components ready via Tor Browser torrc.")
versions["pt_attempted"] = True
save_versions(versions)
return True
log("Local PT incomplete - fetching fresh Tor Browser to extract PT...")
if latest:
if not ensure_7zip(log):
log("WARNING: 7-Zip unavailable - PT extraction will likely fail.")
check_abort()
url = get_tor_browser_url(latest, log)
if url:
dest = DOWNLOAD_DIR / Path(url).name
try:
download_file(url, dest, log)
check_abort()
ed = DOWNLOAD_DIR / "tor-browser-extract"
if extract_tor_browser_archive(dest, ed, log):
log("Contents of extracted Tor Browser:")
list_top_entries(ed, log)
like = list_pt_like_files(ed)
if like:
log("PT-like files in extracted Tor Browser:")
for f in like:
log(" " + f)
else:
log("No PT files found in extracted Tor Browser.")
copy_groups_from_roots([ed], log, overwrite=True)
else:
log("Tor Browser extraction returned False.")
shutil.rmtree(ed, ignore_errors=True)
safe_unlink(dest)
except Exception as e:
log(f"Tor Browser download/extraction error: {e}")
check_abort()
if not pt_ready():
try:
iurl = get_tor_browser_installer_url(latest, log)
idest = DOWNLOAD_DIR / Path(iurl).name
download_file(iurl, idest, log)
check_abort()
if quiet_install_tor_browser(idest, TOR_BROWSER_TEMP_DIR, log):
like = list_pt_like_files(TOR_BROWSER_TEMP_DIR)
if like:
log("PT-like files after silent Tor Browser install:")
for f in like:
log(" " + f)
copy_groups_from_roots([TOR_BROWSER_TEMP_DIR], log, overwrite=True)
shutil.rmtree(TOR_BROWSER_TEMP_DIR, ignore_errors=True)
safe_unlink(idest)
except Exception as e:
log(f"Silent Tor Browser install error: {e}")
copy_groups_from_roots(pt_search_roots(), log, overwrite=True)
versions = load_versions()
versions["pt_attempted"] = True
save_versions(versions)
if pt_ready():
if latest:
versions["pt_tor_browser"] = latest
save_versions(versions)
log("PT components ready.")
return True
log("Could not fetch PT files for groups:")
for g in missing_group_names():
log(" - " + " / ".join(g))
log("7-Zip present: " + ("yes" if find_7z() else "NO"))
log("Tip: press 'Import PT from Tor Browser' or 'Tor Browser folder...' to copy PT binaries manually.")
return False
def pt_status(log, mode="normal"):
log("Checking PT components:")
roots = pt_search_roots()
for group in PT_GROUPS:
found = next((PT_DIR / n for n in group if (PT_DIR / n).exists()), None) or find_file_in_roots(group, roots)
label = " / ".join(group)
if found:
log(f"{label}: {found}")
elif mode == "normal":
log(f"{label}: not required in plain mode (needed only for bridges)")
else:
log(f"{label}: not found")
tb = get_tb_pt_map()
if tb:
log("Tor Browser torrc (source of truth) PT plugins:")
for nm, (exe, tail) in tb.items():
exists = bool(exe and Path(exe).exists())
log(f" [TB-torrc] {nm} -> {exe} (exists={exists})")
return pt_ready()
def find_pt_executable(kind, settings_path=None):
names = {
"obfs": ["lyrebird.exe", "obfs4proxy.exe"],
"obfs4": ["lyrebird.exe", "obfs4proxy.exe"],
"lyrebird": ["lyrebird.exe"],
}.get(kind, [])
cands = [PT_DIR / n for n in names]
nl = [n.lower() for n in names]
if settings_path:
p = Path(settings_path)
if p.exists():
if p.is_file() and p.name.lower() in nl:
cands.append(p)
base = p if p.is_dir() else p.parent
cands += [base / n for n in names]
for n in names:
w = shutil.which(n)
if w:
cands.append(Path(w))
for root in (APP_DIR, SINGBOX_DIR, TOR_DIR, PT_DIR):
f = find_file(root, n)
if f:
cands.append(f)
for d in tor_browser_pt_dirs():
cands += [d / n for n in names]
for c in cands:
try:
if c and Path(c).exists():
return Path(c)
except Exception:
pass
tb = get_tb_pt_map()
for tn in KIND2TRANSPORT.get(kind, []):
if tn in tb:
exe, tail = tb[tn]
if exe and Path(exe).exists():
return Path(exe)
return None
def resolve_pt_map(transports, settings_pt_path, log):
get_tb_pt_map(log=log)
pt_map = {}
pt_exes = []
for t in sorted(transports):
exe = None
if t in ("obfs2", "obfs3", "obfs4"):
exe = find_pt_executable("obfs", settings_pt_path)
if not exe:
raise RuntimeError("lyrebird.exe or obfs4proxy.exe not found.\nThe program will try to install PT automatically.")
elif t == "webtunnel":
exe = find_pt_executable("lyrebird", settings_pt_path)
if not exe:
raise RuntimeError("webtunnel requires lyrebird.exe.")
else:
raise RuntimeError(f"Unsupported bridge transport: {t} (only obfs4/webtunnel are supported).")
pt_map[t] = exe
if exe not in pt_exes:
pt_exes.append(exe)
log(f"PT for {t}: {exe}")
return pt_map, pt_exes
def client_transport_plugin_lines(pt_map):
out = []
for t, exe in sorted(pt_map.items()):
tail = TB_EXEC_TAIL.get(t)
texe = TB_EXEC_EXE.get(t)
if tail and texe and Path(texe).exists():
out.append(f"ClientTransportPlugin {t} exec {tail}")
else:
out.append(f"ClientTransportPlugin {t} exec {tor_quote_path(exe)}")
return out
def install_components(log, force=False):
reset_processes(log)
manifest, results = apply_manifest_install(log, force=force)
log(f"Manifest install results: {results}")
check_abort()
install_pt_components(log, force=force)
if components_status(log):
log("Core components ready.")
else:
raise RuntimeError("After install not all core components were found.")
pt_status(log, mode="bridge")
return manifest
def kill_tun2socks(log):
log("Stopping tun2socks...")
run_silent(
[
"powershell", "-NoProfile", "-Command",
"Get-Process | Where-Object { $_.ProcessName -like 'tun2socks*' } | "
"Stop-Process -Force -ErrorAction SilentlyContinue",
]
)
run_silent(
[
"powershell", "-NoProfile", "-Command",
"Get-Service | Where-Object { $_.Name -like 'tun2socks*' -or "
"$_.DisplayName -like 'tun2socks*' } | Stop-Service -Force -ErrorAction SilentlyContinue",
]
)
for name in ("tun2socks.exe", "tun2socks64.exe", "tun2socks-windows-amd64.exe", "tun2socks-windows-386.exe"):
run_silent(["taskkill", "/F", "/IM", name])
def kill_all_conflicting(log):
log("Stopping all conflicting processes...")
for name in (
"tor.exe", "sing-box.exe", "obfs4proxy.exe", "lyrebird.exe",
"tun2socks.exe", "tun2socks64.exe",
"tun2socks-windows-amd64.exe", "tun2socks-windows-386.exe",
):
run_silent(["taskkill", "/F", "/IM", name])
kill_tun2socks(log)
def get_pids_for_port(port):
pids = set()
rc, out, err = run_cmd(["netstat", "-ano"])
for line in out.splitlines():
parts = line.split()
if len(parts) < 5:
continue
if parts[1].endswith(f":{port}") and parts[-1].isdigit():
pids.add(int(parts[-1]))
return pids
def kill_processes_on_ports(log):
log("Freeing occupied ports...")
ports = {9050, 9150, 9250, 9041, 5354, 1080, 10808}
if STATE_FILE.exists():
try:
st = json.loads(STATE_FILE.read_text(encoding="utf-8"))
for k in ("socks_port", "dns_port", "control_port"):
if st.get(k):
ports.add(int(st[k]))
except Exception:
pass
cur = os.getpid()
for port in sorted(ports):
for pid in get_pids_for_port(port):
if pid <= 4 or pid == cur:
continue
log(f"Killing PID {pid}, port {port}")
run_silent(["taskkill", "/F", "/PID", str(pid)])
def clean_temp_files(log):
log("Cleaning temporary files...")
if DOWNLOAD_DIR.exists():
for p in DOWNLOAD_DIR.iterdir():
try:
if p.is_file():
safe_unlink(p)
elif p.is_dir():
shutil.rmtree(p, ignore_errors=True)
except Exception:
pass
for d in (DOWNLOAD_DIR / "tor-browser-extract", DOWNLOAD_DIR / "wintun-extract", TOR_BROWSER_TEMP_DIR):
try:
if d.exists():
shutil.rmtree(d, ignore_errors=True)
except Exception:
pass
log("Temporary files cleaned.")
def get_free_port(proto="tcp"):
t = socket.SOCK_STREAM if proto == "tcp" else socket.SOCK_DGRAM
with socket.socket(socket.AF_INET, t) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def wait_tcp_port(port, timeout=60):
end = time.time() + timeout
while time.time() < end:
check_abort()
try:
with socket.create_connection(("127.0.0.1", port), timeout=1):
return True
except OSError:
time.sleep(0.5)
return False
def singbox_env():
env = os.environ.copy()
env["ENABLE_DEPRECATED_LEGACY_DNS_SERVERS"] = "true"
return env
def load_tracker_domains():
d = []
if TRACKLIST_FILE.exists():
try:
for line in TRACKLIST_FILE.read_text(encoding="utf-8", errors="ignore").splitlines():
x = line.strip().lower()
if not x or x.startswith("#"):
continue
x = x.split("/")[-1]
if x:
d.append(x)
except Exception:
pass
return d
class ControlUnavailable(Exception):
pass
def _ctrl_recv_response(sock, buf, deadline):
lines = []
while True:
while b"\r\n" in buf:
line, buf = buf.split(b"\r\n", 1)
t = line.decode(errors="ignore")
lines.append(t)
if len(t) >= 4 and t[3] == " ":
return lines, buf
left = deadline - time.time()
if left <= 0:
raise socket.timeout("timed out")
sock.settimeout(left)
chunk = sock.recv(4096)
if not chunk:
raise ConnectionError("control connection closed")
buf += chunk
def _ctrl_open(control_port, deadline):
left = max(0.5, min(4.0, deadline - time.time()))
try:
sock = socket.create_connection(("127.0.0.1", control_port), timeout=left)
except (socket.timeout, OSError) as e:
raise ControlUnavailable(f"connect: {e}")
try:
sock.settimeout(max(0.5, deadline - time.time()))
buf = b""
banner, buf = _ctrl_recv_response(sock, buf, deadline)
if not banner or not banner[-1].startswith("250"):
raise ControlUnavailable(f"banner: {banner}")
cp = TOR_DATA / "control_auth_cookie"
if not cp.exists():
raise ControlUnavailable("control_auth_cookie not found")
sock.sendall(f"AUTHENTICATE {binascii.hexlify(cp.read_bytes()).decode()}\r\n".encode())
resp, buf = _ctrl_recv_response(sock, buf, deadline)
if not resp or not resp[-1].startswith("250"):
raise ControlUnavailable(f"auth: {resp}")
return sock, buf
except ControlUnavailable:
try:
sock.close()
except Exception:
pass
raise
except Exception as e:
try:
sock.close()
except Exception:
pass
raise ControlUnavailable(f"open: {e}")
def _ctrl_command(control_port, command, timeout=6):
deadline = time.time() + timeout
sock, buf = _ctrl_open(control_port, deadline)
try:
sock.sendall((command + "\r\n").encode())
resp, buf = _ctrl_recv_response(sock, buf, deadline)
return resp
except ControlUnavailable:
raise
except Exception as e:
raise ControlUnavailable(f"cmd: {e}")
finally:
try:
sock.close()
except Exception:
pass
def _get_bootstrap(cp):
resp = _ctrl_command(cp, "GETINFO status/bootstrap-phase", timeout=4)
text = "\n".join(resp)
done = ("TAG=done" in text) or ("PROGRESS=100" in text)
m = re.search(r"PROGRESS=(\d+)", text)
return (int(m.group(1)) if m else -1), done
def tor_wait_bootstrap(cp, timeout, log):
deadline = time.time() + timeout
cf = 0
seen = False
lp = -1
while time.time() < deadline:
check_abort()
try:
prog, done = _get_bootstrap(cp)
cf = 0
seen = True
if prog != lp:
log(f"Tor bootstrap: {prog}%")
lp = prog
if done or prog >= 100:
log("Tor bootstrap: ready (100%).")
return True
except ControlUnavailable:
cf += 1
if not seen and cf >= 4:
log("ControlPort not responding - skipping bootstrap wait.")
return False
except Exception as e:
cf += 1
log(f"bootstrap poll: {e}")
if cf >= 4:
log("ControlPort unstable - skipping bootstrap wait.")
return False
time.sleep(2)
log("Tor did not reach 100% in time (continuing)." if seen else "ControlPort silent (continuing).")
return False
def socks5_probe(socks_port, timeout_each=12):
for host, port in (("1.1.1.1", 443), ("8.8.8.8", 443), ("9.9.9.9", 53)):
try:
with socket.create_connection(("127.0.0.1", socks_port), timeout=timeout_each) as s:
s.sendall(b"\x05\x01\x00")
h = s.recv(2)
if len(h) < 2 or h[0] != 5 or h[1] != 0:
continue
s.sendall(b"\x05\x01\x00\x01" + socket.inet_aton(host) + port.to_bytes(2, "big"))
r = s.recv(10)
if len(r) >= 2 and r[0] == 5 and r[1] == 0:
return True, f"{host}:{port}"
except Exception:
continue
return False, ""
def write_torrc(socks_port, dns_port, control_port, settings, pt_map, log):
ensure_dirs()
lines = [
"# Managed by TorLeakGuard",
f"DataDirectory {TOR_DATA.as_posix()}",
f"SocksPort 127.0.0.1:{socks_port}",
f"DNSPort 127.0.0.1:{dns_port}",
"ClientUseIPv4 1",
"ClientUseIPv6 0",
"ClientPreferIPv6ORPort 0",
"ClientPreferIPv6DirPort 0",
"IPv6Exit 0",
"AutomapHostsOnResolve 1",
"VirtualAddrNetworkIPv4 10.192.0.0/10",
f"ControlPort 127.0.0.1:{control_port}",
"CookieAuthentication 1",
]
lines.append(
"MaxCircuitDirtiness 31536000" if settings.get("prevent_circuit_change", True)
else "MaxCircuitDirtiness 600"
)
if settings.get("padding", True):
lines += [
"# Anti traffic-analysis padding",
"ConnectionPadding 1",
"ReducedConnectionPadding 0",
"CircuitPadding 1",
"ReducedCircuitPadding 0",
]
if settings.get("mode", "normal") == "bridge":
bl = normalize_bridge_lines(settings.get("bridges", ""))
if not bl:
raise RuntimeError("Bridges mode on but bridge lines empty.")
lines.append("UseBridges 1")
tr = bridge_transports(bl)
if tr:
if not pt_map:
pt_map, _ = resolve_pt_map(tr, settings.get("pt_path", ""), log)
lines += client_transport_plugin_lines(pt_map)
lines += bl
else:
lines.append("UseBridges 0")
TORRC.write_text("\n".join(lines) + "\n", encoding="utf-8")
log(f"torrc written: {TORRC}")
def build_singbox_config(socks_port, dns_port, tor_exe, sing_exe, strict_route, stack,
pt_exes=None, sniff_action=True, settings=None):
pt_exes = pt_exes or []
settings = settings or {}
pn = ["tor.exe", "tor", "sing-box.exe", "sing-box"]
pp = [str(tor_exe), str(sing_exe)]
for pt in pt_exes:
pt = Path(pt)
pn += [pt.name, pt.stem]
pp.append(str(pt))
inbound = {
"type": "tun",
"tag": "tun-in",
"interface_name": "TorLeakGuard",
"address": ["172.19.0.1/30"],
"mtu": 1500,
"auto_route": True,
"strict_route": strict_route,
"stack": stack,
}
rr = []
if sniff_action:
rr.append({"action": "sniff"})
rr.append({"port": [53], "action": "hijack-dns"})
rr += [
{"process_name": pn, "outbound": "direct"},
{"process_path": pp, "outbound": "direct"},
{"ip_cidr": ["127.0.0.0/8"], "outbound": "direct"},
{"network": "udp", "port": [67, 68], "outbound": "direct"},
{"ip_version": 6, "outbound": "block"},
{"network": "udp", "outbound": "block"},
{
"ip_cidr": ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16",
"169.254.0.0/16", "224.0.0.0/4", "255.255.255.255/32"],
"outbound": "block",
},
]
if settings.get("block_webrtc_hosts"):
rr.append({
"domain_suffix": [
"stun.l.google.com", "stun1.l.google.com", "stun2.l.google.com",
"stun3.l.google.com", "stun4.l.google.com", "stun.services.mozilla.com",
"stun.voip.blackberry.com", "global.stun.twilio.com",
],
"outbound": "block",
})
if settings.get("block_insecure_http"):
rr.append({"port": [80], "outbound": "block"})
if settings.get("block_webrtc_hosts") or settings.get("block_insecure_http"):
td = load_tracker_domains()
if td:
rr.append({"domain_suffix": td, "outbound": "block"})
return {
"log": {"level": "info", "output": SINGBOX_LOG.as_posix()},
"dns": {
"servers": [{"type": "udp", "tag": "tor-dns", "server": "127.0.0.1", "server_port": dns_port}],
"rules": [
{"query_type": ["A", "AAAA", "CNAME", "TXT", "MX", "NS", "SOA", "SRV", "PTR"],
"server": "tor-dns"}
],
"final": "tor-dns",
},
"inbounds": [inbound],
"outbounds": [
{"type": "socks", "tag": "tor", "server": "127.0.0.1", "server_port": socks_port, "version": "5"},
{"type": "direct", "tag": "direct"},
{"type": "block", "tag": "block"},
],
"route": {"rules": rr, "final": "tor"},
}
class App:
def __init__(self, root):
self.root = root
self.root.title(f"TorLeakGuard v{PROGRAM_VERSION}: Tor + sing-box + obfs4/webtunnel + longevity + winget")
self.root.geometry("1080x1280")
self.busy = False
self.handles = []
self.spawned = []
self.last_newnym = 0.0
self.manifest = None
self.manifest_origin = "builtin"
self._winget_ids = []
self.settings = load_settings()
self.mode_var = tk.StringVar(value=self.settings.get("mode", "normal"))
self.prevent_var = tk.BooleanVar(value=self.settings.get("prevent_circuit_change", True))
self.padding_var = tk.BooleanVar(value=self.settings.get("padding", True))
self.block_http_var = tk.BooleanVar(value=self.settings.get("block_insecure_http", False))
self.block_webrtc_var = tk.BooleanVar(value=self.settings.get("block_webrtc_hosts", False))
self.pt_path_var = tk.StringVar(value=self.settings.get("pt_path", ""))
self.pt_label_var = tk.StringVar(value="obfs4/lyrebird: ? (press 'Check bridge status')")
self.status_var = tk.StringVar(value=f"program v{PROGRAM_VERSION} | manifest: (not loaded)")
self.winget_query_var = tk.StringVar(value="")
self.reset_tun2socks_var = tk.BooleanVar(value=True)
self.reset_ports_var = tk.BooleanVar(value=False)
self.reset_temp_var = tk.BooleanVar(value=True)
self.reset_tordata_var = tk.BooleanVar(value=False)
self.reset_all_proc_var = tk.BooleanVar(value=False)
self._build_ui()
self._build_extra_ui()
self._build_winget_ui()
self._build_system_ui()
ensure_restored(self.log)
self.root.protocol("WM_DELETE_WINDOW", self.on_close)
def _build_ui(self):
pad = {"padx": 8, "pady": 4}
ttk.Label(self.root, textvariable=self.status_var).pack(fill=tk.X, padx=8, pady=2)
top = ttk.Frame(self.root)
top.pack(fill=tk.X, **pad)
ttk.Button(top, text="Check components", command=self.on_check).pack(side=tk.LEFT, padx=2)
ttk.Button(top, text="Install / repair components", command=self.on_install).pack(side=tk.LEFT, padx=2)
ttk.Button(top, text="Check for updates", command=self.on_update).pack(side=tk.LEFT, padx=2)
ttk.Button(top, text="Refresh manifest", command=self.on_refresh_manifest).pack(side=tk.LEFT, padx=2)
ttk.Button(top, text="Check program update", command=self.on_check_program_update).pack(side=tk.LEFT, padx=2)
mid = ttk.Frame(self.root)
mid.pack(fill=tk.X, **pad)
ttk.Button(mid, text="Start Tor + protection", command=self.on_start).pack(side=tk.LEFT, padx=2)
ttk.Button(mid, text="Stop (restore internet)", command=self.on_stop).pack(side=tk.LEFT, padx=2)
ttk.Button(mid, text="RESET (emergency eject)", command=self.on_reset).pack(side=tk.LEFT, padx=2)
ttk.Button(mid, text="New circuit (NEWNYM)", command=self.on_newnym).pack(side=tk.LEFT, padx=2)
rf = ttk.LabelFrame(self.root, text="RESET / emergency eject options")
rf.pack(fill=tk.X, **pad)
ttk.Label(
rf,
text="RESET never waits: aborts tasks, force-kills by PID, drops TUN+IPv6, cleans files, keeps window open.",
justify=tk.LEFT, wraplength=1020,
).pack(anchor=tk.W, padx=6, pady=2)
ttk.Checkbutton(rf, text="Stop tun2socks", variable=self.reset_tun2socks_var).pack(anchor=tk.W, padx=6, pady=1)
ttk.Checkbutton(rf, text="Force-free occupied ports", variable=self.reset_ports_var).pack(anchor=tk.W, padx=6, pady=1)
ttk.Checkbutton(rf, text="Clean temporary files", variable=self.reset_temp_var).pack(anchor=tk.W, padx=6, pady=1)
ttk.Checkbutton(rf, text="Full clean of tor-data", variable=self.reset_tordata_var).pack(anchor=tk.W, padx=6, pady=1)
ttk.Checkbutton(rf, text="Stop ALL conflicting processes by name (dangerous)", variable=self.reset_all_proc_var).pack(anchor=tk.W, padx=6, pady=1)
mf = ttk.LabelFrame(self.root, text="Connection mode")
mf.pack(fill=tk.X, **pad)
ttk.Radiobutton(mf, text="Plain Tor (protected)", variable=self.mode_var, value="normal").pack(anchor=tk.W, padx=6, pady=2)
ttk.Radiobutton(mf, text="Connect via bridges (obfs4 / webtunnel)", variable=self.mode_var, value="bridge").pack(anchor=tk.W, padx=6, pady=2)
ttk.Checkbutton(mf, text="Don't rotate circuit automatically (MaxCircuitDirtiness = 1 year)", variable=self.prevent_var).pack(anchor=tk.W, padx=6, pady=2)
cf = ttk.LabelFrame(self.root, text="Circumvention & anti-analysis (DPI / throttling)")
cf.pack(fill=tk.X, **pad)
ttk.Checkbutton(cf, text="Enable connection & circuit padding (defeats traffic-pattern throttling / analysis)", variable=self.padding_var).pack(anchor=tk.W, padx=6, pady=2)
cb = ttk.Frame(cf)
cb.pack(fill=tk.X, padx=6, pady=2)
ttk.Button(cb, text="Apply recommended anti-DPI preset", command=self.on_apply_preset).pack(side=tk.LEFT)
ttk.Label(cb, text="(switches to Bridges + padding; then paste obfs4/webtunnel bridges below)").pack(side=tk.LEFT, padx=8)
ttk.Label(cf, text=CIRCUM_HINT, justify=tk.LEFT, wraplength=1020).pack(anchor=tk.W, padx=8, pady=4)
pf = ttk.LabelFrame(self.root, text="Privacy hardening: fingerprint & cookie/session leak (optional)")
pf.pack(fill=tk.X, **pad)
ttk.Label(pf, text="Always ON (no breakage): DNS via Tor, IPv6 blocked, WebRTC blocked (UDP off), padding.", justify=tk.LEFT, wraplength=1020).pack(anchor=tk.W, padx=8, pady=2)
ttk.Checkbutton(pf, text="STRICT: block WebRTC STUN/TURN hosts over TCP too (may break some video calls)", variable=self.block_webrtc_var).pack(anchor=tk.W, padx=8, pady=1)
ttk.Checkbutton(pf, text="STRICT: block insecure plain-HTTP (http-only sites will fail)", variable=self.block_http_var).pack(anchor=tk.W, padx=8, pady=1)
pb = ttk.Frame(pf)
pb.pack(fill=tk.X, padx=8, pady=2)
ttk.Button(pb, text="Open Tor Browser download page", command=self.on_open_torbrowser).pack(side=tk.LEFT)
ttk.Button(pb, text="Edit tracker block-list", command=self.on_edit_tracklist).pack(side=tk.LEFT, padx=6)
ttk.Label(pb, text="(tracklist.txt = one domain per line; external -> survives updates)").pack(side=tk.LEFT, padx=6)
ttk.Label(pf, text=PRIVACY_HINT, justify=tk.LEFT, wraplength=1020).pack(anchor=tk.W, padx=8, pady=4)
bf = ttk.LabelFrame(self.root, text="Bridges from Tor Browser: obfs4 / webtunnel")
bf.pack(fill=tk.BOTH, expand=True, **pad)
ptf = ttk.Frame(bf)
ptf.pack(fill=tk.X, padx=4, pady=4)
ttk.Label(ptf, text="PT file:").pack(side=tk.LEFT)
ttk.Entry(ptf, textvariable=self.pt_path_var).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=6)
ttk.Button(ptf, text="Choose PT file", command=self.on_choose_pt).pack(side=tk.LEFT)
ttk.Button(ptf, text="Tor Browser folder...", command=self.on_choose_tb_dir).pack(side=tk.LEFT, padx=4)
ttk.Button(ptf, text="Import PT from Tor Browser", command=self.on_import_tb_pt).pack(side=tk.LEFT, padx=4)
bw = ttk.Frame(bf)
bw.pack(fill=tk.BOTH, expand=True, padx=4, pady=4)
self.bridges_text = tk.Text(bw, height=5, wrap=tk.NONE)
bv = ttk.Scrollbar(bw, orient=tk.VERTICAL, command=self.bridges_text.yview)
bh = ttk.Scrollbar(bw, orient=tk.HORIZONTAL, command=self.bridges_text.xview)
self.bridges_text.config(yscrollcommand=bv.set, xscrollcommand=bh.set)
self.bridges_text.grid(row=0, column=0, sticky="nsew")
bv.grid(row=0, column=1, sticky="ns")
bh.grid(row=1, column=0, sticky="ew")
bw.rowconfigure(0, weight=1)
bw.columnconfigure(0, weight=1)
self.bridges_text.insert(tk.END, self.settings.get("bridges", ""))
bind_text_clipboard(self.bridges_text)
sb = ttk.Frame(bf)
sb.pack(fill=tk.X, padx=4, pady=2)
ttk.Button(sb, text="Check bridge status", command=self.on_check_bridges).pack(side=tk.LEFT)
ttk.Label(sb, text=" [ACTIVE]=real conn | [no connection]=plugin up elsewhere | [NO PT PROCESS]=missing", wraplength=700, justify=tk.LEFT).pack(side=tk.LEFT, padx=6)
ttk.Label(bf, textvariable=self.pt_label_var).pack(anchor=tk.W, padx=6, pady=2)
self.bridge_status_lb = tk.Listbox(bf, height=4, activestyle="none")
self.bridge_status_lb.pack(fill=tk.X, padx=4, pady=4)
self.log_frame = ttk.LabelFrame(self.root, text="Log")
self.log_frame.pack(fill=tk.BOTH, expand=True, **pad)
self.log_text = tk.Text(self.log_frame, wrap=tk.WORD, state=tk.DISABLED)
self.log_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
sc = ttk.Scrollbar(self.log_frame, command=self.log_text.yview)
sc.pack(side=tk.RIGHT, fill=tk.Y)
self.log_text.config(yscrollcommand=sc.set)
bind_text_clipboard(self.log_text)
self.log("Ready.")
self.log(f"Working folder: {APP_DIR}")
self.log(f"Program version: {PROGRAM_VERSION}")
self.log("Install/update/start/stop request UAC. RESET never does (always works).")
self.log("PT (obfs4/webtunnel): read from Tor Browser's OWN torrc (source of truth) - works even if the binary is renamed/relocated.")
def _build_extra_ui(self):
ex = ttk.Frame(self.root)
ex.pack(fill=tk.X, padx=8, pady=2, before=self.log_frame)
ttk.Button(ex, text="Rebuild vault copies", command=self.on_rebuild_vaults).pack(side=tk.LEFT, padx=2)
ttk.Button(ex, text="Inventory / self-test", command=self.on_inventory).pack(side=tk.LEFT, padx=2)
ttk.Label(ex, text="(manifest/sources saved in 4 places + a copy embedded inside the program)", wraplength=620, justify=tk.LEFT).pack(side=tk.LEFT, padx=6)
def _build_winget_ui(self):
wf = ttk.LabelFrame(self.root, text="winget: manual search & install (fallback when auto-detect misses a module)")
wf.pack(fill=tk.X, padx=8, pady=2, before=self.log_frame)
row = ttk.Frame(wf)
row.pack(fill=tk.X, padx=4, pady=3)
ttk.Label(row, text="Query:").pack(side=tk.LEFT)
ttk.Entry(row, textvariable=self.winget_query_var, width=28).pack(side=tk.LEFT, padx=4)
ttk.Button(row, text="Search winget", command=self.on_winget_search).pack(side=tk.LEFT, padx=2)
ttk.Button(row, text="Install selected (winget)", command=self.on_winget_install).pack(side=tk.LEFT, padx=2)
ttk.Label(row, text="(select a row, then Install)").pack(side=tk.LEFT, padx=6)
self.winget_lb = tk.Listbox(wf, height=5, activestyle="none")
self.winget_lb.pack(fill=tk.X, padx=4, pady=3)
ttk.Label(wf, text=WINGET_HINT, justify=tk.LEFT, wraplength=1020).pack(anchor=tk.W, padx=6, pady=3)
def _build_system_ui(self):
sf = ttk.LabelFrame(self.root, text="Installed via winget / found in system (scanner)")
sf.pack(fill=tk.X, padx=8, pady=2, before=self.log_frame)
srow = ttk.Frame(sf)
srow.pack(fill=tk.X, padx=4, pady=3)
ttk.Button(srow, text="Scan system (winget + PATH + drives)", command=self.on_scan_system).pack(side=tk.LEFT, padx=2)
ttk.Label(srow, text="(auto-runs after winget install / Install / Update)").pack(side=tk.LEFT, padx=6)
self.system_lb = tk.Listbox(sf, height=9, activestyle="none")
self.system_lb.pack(fill=tk.X, padx=4, pady=3)
ttk.Label(sf, text=SYSTEM_HINT, justify=tk.LEFT, wraplength=1020).pack(anchor=tk.W, padx=6, pady=3)
def log(self, msg):
def _a():
self.log_text.config(state=tk.NORMAL)
self.log_text.insert(tk.END, msg + "\n")
self.log_text.see(tk.END)
self.log_text.config(state=tk.DISABLED)
self.root.after(0, _a)
def set_status(self, origin):
self.manifest_origin = origin
ho = "ON" if self.block_http_var.get() else "off"
wo = "ON" if self.block_webrtc_var.get() else "off"
self.root.after(
0,
lambda: self.status_var.set(
f"program v{PROGRAM_VERSION} | manifest: {origin} | strict http={ho} webrtc-tcp={wo}"
),
)
def run_thread(self, func):
if self.busy:
self.log("Another operation running. Wait or press RESET.")
return
self.busy = True
def w():
try:
func()
except Aborted:
self.log("Background task aborted by RESET.")
except Exception as e:
self.log(f"Error: {e}")
finally:
self.root.after(0, lambda: setattr(self, "busy", False))
threading.Thread(target=w, daemon=True).start()
def collect_settings(self):
return {
"mode": self.mode_var.get(),
"bridges": self.bridges_text.get("1.0", tk.END).strip(),
"pt_path": self.pt_path_var.get().strip(),
"prevent_circuit_change": bool(self.prevent_var.get()),
"padding": bool(self.padding_var.get()),
"block_insecure_http": bool(self.block_http_var.get()),
"block_webrtc_hosts": bool(self.block_webrtc_var.get()),
}
def collect_reset_options(self):
return {
"tun2socks": bool(self.reset_tun2socks_var.get()),
"ports": bool(self.reset_ports_var.get()),
"temp": bool(self.reset_temp_var.get()),
"tor_data": bool(self.reset_tordata_var.get()),
"all_processes": bool(self.reset_all_proc_var.get()),
}
def save_current_settings(self):
save_settings(self.collect_settings())
def on_close(self):
try:
self.save_current_settings()
except Exception:
pass
self.root.destroy()
def on_check(self):
self.run_thread(self.task_check)
def on_install(self):
if not is_admin():
self.log("Install needs UAC.")
try:
restart_as_admin("install")
except Exception as e:
self.log(str(e))
return
self.run_thread(self.task_install)
def on_update(self):
if not is_admin():
self.log("Update needs UAC.")
try:
restart_as_admin("update")
except Exception as e:
self.log(str(e))
return
self.run_thread(self.task_update)
def on_refresh_manifest(self):
self.run_thread(self.task_refresh_manifest)
def on_check_program_update(self):
self.run_thread(self.task_check_program_update)
def on_start(self):
if not is_admin():
self.log("Start needs UAC.")
try:
restart_as_admin("start")
except Exception as e:
self.log(str(e))
return
s = self.collect_settings()
self.run_thread(lambda: self.task_start(s))
def on_stop(self):
if not is_admin():
self.log("Stop needs UAC.")
try:
restart_as_admin("stop")
except Exception as e:
self.log(str(e))
return
self.run_thread(self.task_stop)
def on_reset(self):
self.log("=== EMERGENCY RESET === forcing immediate stop...")
abort_now()
self.busy = False
o = self.collect_reset_options()
threading.Thread(target=lambda: self._emergency_reset_work(o), daemon=True).start()
def on_newnym(self):
self.run_thread(self.task_newnym)
def on_apply_preset(self):
self.mode_var.set("bridge")
self.padding_var.set(True)
self.log("Anti-DPI preset applied: Bridges + padding ON.")
def on_open_torbrowser(self):
try:
os.startfile("https://www.torproject.org/download/")
except Exception as e:
self.log(f"Could not open browser: {e}")
def on_edit_tracklist(self):
ensure_dirs()
if not TRACKLIST_FILE.exists():
TRACKLIST_FILE.write_text(
"# One tracker/cookie domain per line (suffix match). External -> survives updates.\n"
"# doubleclick.net\n# google-analytics.com\n",
encoding="utf-8",
)
try:
os.startfile(str(TRACKLIST_FILE))
except Exception as e:
self.log(f"Could not open tracklist: {e}")
def on_check_bridges(self):
t = self.bridges_text.get("1.0", tk.END)
self.run_thread(lambda: self.refresh_bridge_status(t, True))
def on_choose_pt(self):
f = filedialog.askopenfilename(
title="Choose PT file",
filetypes=[("Executable files", "*.exe"), ("All files", "*.*")],
)
if f:
self.pt_path_var.set(f)
self.log(f"Chosen PT file: {f}")
def on_choose_tb_dir(self):
d = filedialog.askdirectory(title="Select your Tor Browser folder (any location / any name)")
if not d:
return
self.run_thread(lambda: self._copy_tb_dir(d))
def on_import_tb_pt(self):
self.run_thread(self._import_tb_pt)
def on_rebuild_vaults(self):
self.run_thread(self.task_rebuild_vaults)
def on_inventory(self):
self.run_thread(self.task_inventory)
def on_winget_search(self):
self.run_thread(self.task_winget_search)
def on_winget_install(self):
self.run_thread(self.task_winget_install)
def on_scan_system(self):
self.run_thread(self._do_system_scan)
def _copy_tb_dir(self, d):
self.log(f"Copying Tor PT binaries from chosen folder: {d}")
n = copy_pt_from_folder(d, self.log)
if n:
self.log("Re-running bridge status so the new PT binaries show up...")
t = self.bridges_text.get("1.0", tk.END)
self.refresh_bridge_status(t, True)
self._do_system_scan()
def _import_tb_pt(self):
self.log("Importing PT binaries referenced by Tor Browser's torrc (source of truth)...")
n = copy_pt_from_torbrowser_config(self.log)
t = self.bridges_text.get("1.0", tk.END)
self.refresh_bridge_status(t, True)
self._do_system_scan()
if n:
self.log("Import done. Our tor.exe will now use these PT binaries for bridges.")
else:
self.log("Import found nothing copyable. Use obfs4/webtunnel via lyrebird.")
def _bridge_status_compute(self, bt):
obfs = find_pt_executable("obfs", None)
pt_text = f"obfs4/lyrebird: {'FOUND' if obfs else 'MISSING'}"
rows = parse_bridge_rows(bt)
pids, remote = get_pt_runtime()
lines = []
act = False
if not rows:
lines.append("(no bridge lines parsed)")
else:
for tr, ipport in rows:
ip = ipport.split(":")[0] if ":" in ipport else ipport
a = (ipport in remote) or any(r == ipport or r.startswith(ip + ":") for r in remote)
st = "[ACTIVE]" if a else ("[NO PT PROCESS]" if not pids else "[no connection]")
if a:
act = True
lines.append(f"{tr:10} {ipport:26} {st}")
tail = read_tor_log_tail()
low = tail.lower()
hbl = (
"conn_bridge" in low or "connected to a bridge" in low
or "managed proxy" in low or ("bootstrapped" in low and "bridge" in low)
)
ro = ("connecting to a relay" in low) and not hbl
if not rows:
v = "No bridges in the field."
elif not pids:
v = "PT plugin NOT running -> bridges CANNOT work."
elif act:
v = "BRIDGES IN USE: yes (open connection to a bridge IP)."
else:
v = "PT running but no bridge connection yet."
if ro:
v += " WARNING: tor.log shows 'connecting to a relay'."
elif hbl and rows:
v += " tor.log confirms bridge bootstrap."
return pt_text, lines, v, tail
def _apply_bridge_status_ui(self, pt_text, lines):
self.pt_label_var.set(pt_text)
self.bridge_status_lb.delete(0, tk.END)
for ln in lines:
self.bridge_status_lb.insert(tk.END, ln)
def refresh_bridge_status(self, bt, log_too=False):
pt_text, lines, v, tail = self._bridge_status_compute(bt)
self.root.after(0, lambda: self._apply_bridge_status_ui(pt_text, lines))
if log_too:
self.log("---- Bridge status ----")
self.log(pt_text)
for ln in lines:
self.log(" " + ln)
self.log("VERDICT: " + v)
def _do_system_scan(self):
self.log("Scanning system for Tor PT binaries + winget packages (incl. all drives)...")
scan_drives_for_torbrowser(self.log, force=True)
found = find_in_system(TOR_PT_NAMES)
lines = []
lines.append("== Tor pluggable transports found in system ==")
for n in TOR_PT_NAMES:
if n in found:
lines.append(f"[TOR-PT OK] {n} -> {found[n]}")
else:
lines.append(f"[missing] {n}")
tb = get_tb_pt_map(force=True, log=self.log)
lines.append("== Tor Browser torrc PT plugins (source of truth) ==")
if not tb:
lines.append("(no ClientTransportPlugin lines / torrc not found)")
for nm, (exe, tail) in tb.items():
ex = bool(exe and Path(exe).exists())
lines.append(f"[TB-torrc] {nm} exec {exe} (file exists={ex})")
related = []
seen = set()
for q in ("tor", "obfs", "lyrebird"):
for row in winget_list(q):
if row[1] in seen:
continue
seen.add(row[1])
related.append(row)
lines.append("== winget packages related to tor/bridges (INFO only) ==")
if not related:
lines.append("(none found via winget list)")
for name, pid, ver in related:
note = ""
if pid.lower().startswith("torproject"):
note = " <-- Tor Browser: its PTs are read from its torrc"
lines.append(f"[winget] {name} | {pid} | {ver}{note}")
def _fill():
self.system_lb.delete(0, tk.END)
for ln in lines:
self.system_lb.insert(tk.END, ln)
self.root.after(0, _fill)
for ln in lines:
self.log(" " + ln)
def task_check(self):
ensure_dirs()
ensure_restored(self.log)
self.log("Checking components...")
if components_status(self.log):
self.log("Core components found.")
else:
self.log("Not all core components found. Press 'Install / repair components'.")
self.log("=== Capability inventory (proof nothing is missing) ===")
for s in capability_inventory():
self.log(" - " + s)
self.log(f"Total capabilities listed: {len(capability_inventory())}")
pt_status(self.log, mode=self.mode_var.get())
def task_install(self):
ensure_dirs()
abort_clear()
self.log("Installing / repairing (manifest-driven)...")
self.manifest = install_components(self.log, force=False)
self.set_status("installed")
self._do_system_scan()
def task_update(self):
ensure_dirs()
abort_clear()
self.log("Checking updates (manifest-driven)...")
self.manifest = install_components(self.log, force=True)
self.set_status("updated")
self._do_system_scan()
def task_refresh_manifest(self):
ensure_dirs()
self.log("Refreshing manifest from mirrors...")
r, o = fetch_remote_manifest(self.log)
self.manifest = merge_manifest(r)
self.set_status(o)
self.log(f"Manifest components now: {sorted((self.manifest.get('components') or {}).keys())}")
def task_check_program_update(self):
ensure_dirs()
if self.manifest is None:
r, o = fetch_remote_manifest(self.log)
self.manifest = merge_manifest(r)
self.set_status(o)
has, ver, fu = check_program_update(self.log, self.manifest)
if has and fu:
self.log("Downloading program update (current backed up to .bak)...")
do_program_update(self.log, self.manifest)
def task_rebuild_vaults(self):
ensure_dirs()
ensure_restored(self.log)
m = _load_json_file(MANIFEST_CACHE_FILE)
if not validate_manifest(m):
m = json.loads(json.dumps(DEFAULT_MANIFEST))
save_manifest_everywhere(m, self.log)
save_sources_everywhere(load_sources(), self.log)
self.log(
f"Vault copies rebuilt: manifest={_count_vaults(manifest_vault_paths())}, "
f"sources={_count_vaults(sources_vault_paths())}."
)
def task_inventory(self):
ensure_dirs()
ensure_restored(self.log)
self.log("=== Capability inventory ===")
inv = capability_inventory()
for s in inv:
self.log(" - " + s)
self.log(f"Total: {len(inv)} | version {PROGRAM_VERSION}")
def task_winget_search(self):
q = self.winget_query_var.get().strip()
if not q:
self.log("Enter a winget query first.")
return
res = winget_search(q, self.log)
self._winget_ids = []
def _fill():
self.winget_lb.delete(0, tk.END)
if not res:
self.winget_lb.insert(tk.END, "(no results or winget unavailable)")
return
for name, pid, ver in res:
self._winget_ids.append(pid)
self.winget_lb.insert(tk.END, f"{name} | {pid} | {ver}")
self.root.after(0, _fill)
def task_winget_install(self):
if not winget_available():
self.log("winget not available on this system.")
return
sel = self.winget_lb.curselection()
if not sel or not self._winget_ids:
self.log("Search first, then select a row to install.")
return
pid = self._winget_ids[sel[0]]
if not is_admin():
self.log("winget install may need UAC; trying anyway...")
if winget_install(pid, self.log):
self.log(f"winget install finished for {pid}.")
self.log("Re-scanning system so the new package shows up below...")
self._do_system_scan()
else:
self.log(f"winget install failed/needs elevation for {pid}. Run program as admin and retry.")
def task_stop(self):
ensure_dirs()
self.log("Stop: restoring normal internet...")
reset_processes(self.log)
for h in self.handles:
try:
h.close()
except Exception:
pass
self.handles = []
self.spawned = []
safe_unlink(STATE_FILE)
self.log("Stop done.")
def _kill_saved_pids(self):
if not STATE_FILE.exists():
return
try:
st = json.loads(STATE_FILE.read_text(encoding="utf-8"))
except Exception:
return
for k in ("tor_pid", "sing_box_pid"):
pid = st.get(k)
if pid:
run_silent(["taskkill", "/F", "/PID", str(pid)])
self.log(f"Emergency kill PID {pid}.")
def _kill_spawned(self):
for p in list(self.spawned):
try:
if p.poll() is None:
p.kill()
self.log(f"Emergency kill spawned PID {p.pid}.")
except Exception:
pass
self.spawned = []
def _emergency_reset_work(self, o):
try:
self._kill_spawned()
self._kill_saved_pids()
if o.get("all_processes"):
kill_all_conflicting(self.log)
elif o.get("tun2socks"):
kill_tun2socks(self.log)
reset_processes(self.log)
if o.get("ports"):
kill_processes_on_ports(self.log)
for h in list(self.handles):
try:
h.close()
except Exception:
pass
self.handles = []
reset_files(self.log, full=o.get("tor_data", False))
if o.get("temp"):
clean_temp_files(self.log)
self.log("=== EMERGENCY RESET done === window stays open.")
except Exception as e:
self.log(f"Emergency reset error: {e}")
finally:
abort_clear()
def task_newnym(self):
if not STATE_FILE.exists():
self.log("Start Tor + protection first.")
return
now = time.time()
if now - self.last_newnym < 10:
self.log("Too frequent. Wait 10s.")
return
try:
st = json.loads(STATE_FILE.read_text(encoding="utf-8"))
except Exception as e:
self.log(f"state.json error: {e}")
return
cp = st.get("control_port")
if not cp:
self.log("No control_port. Restart protection.")
return
try:
r = _ctrl_command(int(cp), "SIGNAL NEWNYM", timeout=6)
if not r or not r[-1].startswith("250"):
raise RuntimeError(str(r))
self.last_newnym = now
self.log("NEWNYM sent. Restart browser for a fully new IP.")
except ControlUnavailable:
self.log("ControlPort unavailable. Use Stop then Start to rebuild circuits.")
except Exception as e:
self.log(f"NEWNYM error: {e}")
def rollback(self, tp, sp, log):
log("ROLLBACK: restoring normal internet...")
for p in (sp, tp):
if p is None:
continue
try:
if p.poll() is None:
p.kill()
except Exception:
pass
for h in self.handles:
try:
h.close()
except Exception:
pass
self.handles = []
self.spawned = []
run_silent(["netsh", "interface", "set", "interface", "name=TorLeakGuard", "admin=disabled"])
remove_ipv6_block(log)
safe_unlink(STATE_FILE)
log("Rollback done.")
def task_start(self, settings):
ensure_dirs()
abort_clear()
save_settings(settings)
self.set_status(self.manifest_origin)
self.log("Preparing to start...")
self.spawned = []
warn_if_tun2socks(self.log)
reset_processes(self.log)
for h in self.handles:
try:
h.close()
except Exception:
pass
self.handles = []
reset_files(self.log, full=False)
if not components_status(self.log):
self.log("Components missing. Installing (manifest-driven)...")
self.manifest = install_components(self.log, force=False)
check_abort()
te = get_tor_exe()
se = get_singbox_exe()
if not te or not se:
self.log("tor.exe or sing-box.exe not found. Start impossible.")
return
w = get_wintun_dll()
if w and se:
t = Path(se).parent / "wintun.dll"
try:
if not t.exists():
shutil.copy2(w, t)
except Exception as e:
self.log(f"wintun copy warning: {e}")
vs = load_versions()
if not pt_ready() and not vs.get("pt_attempted"):
self.log("Fetching PT (one-time)...")
try:
install_pt_components(self.log, force=False)
except Aborted:
raise
except Exception as e:
self.log(f"PT auto-fetch failed: {e}")
pt_map = {}
pt_exes = []
if settings.get("mode") == "bridge":
bl = normalize_bridge_lines(settings.get("bridges", ""))
tr = bridge_transports(bl)
if tr:
try:
pt_map, pt_exes = resolve_pt_map(tr, settings.get("pt_path", ""), self.log)
except Exception as e:
self.log(str(e))
self.log("Installing PT automatically...")
install_pt_components(self.log, force=False)
pt_map, pt_exes = resolve_pt_map(tr, settings.get("pt_path", ""), self.log)
socks_port = get_free_port("tcp")
dns_port = get_free_port("udp")
control_port = get_free_port("tcp")
self.log(f"Ports: SOCKS={socks_port}, DNS={dns_port}, Control={control_port}")
write_torrc(socks_port, dns_port, control_port, settings, pt_map, log=self.log)
if settings.get("padding", True):
self.log("Padding enabled.")
if settings.get("block_insecure_http"):
self.log("STRICT: plain-HTTP blocked.")
if settings.get("block_webrtc_hosts"):
self.log("STRICT: WebRTC TCP hosts blocked.")
self.log("Starting Tor...")
tlf = open(TOR_LOG, "wb")
self.handles.append(tlf)
tp = subprocess.Popen(
[str(te), "-f", str(TORRC)],
stdout=tlf, stderr=tlf, creationflags=CREATE_NO_WINDOW, cwd=str(TOR_DIR),
)
self.spawned.append(tp)
if not wait_tcp_port(socks_port, timeout=60):
self.log("Tor SOCKS not up in time.")
self.rollback(tp, None, self.log)
return
self.log("Tor SOCKS listening.")
cp = TOR_DATA / "control_auth_cookie"
dl = time.time() + 30
while time.time() < dl and not cp.exists():
check_abort()
time.sleep(0.3)
if not cp.exists():
self.log("control_auth_cookie not found.")
tor_wait_bootstrap(control_port, timeout=20, log=self.log)
started = False
sp = None
chosen = None
variants = [
{"strict_route": sr, "stack": st, "sniff_action": sn}
for sr in (True, False) for st in ("system", "gvisor") for sn in (True, False)
]
env = singbox_env()
for v in variants:
check_abort()
self.log(f"Trying sing-box: sr={v['strict_route']} stack={v['stack']} sniff={v['sniff_action']}")
cfg = build_singbox_config(
socks_port, dns_port, te, se, v["strict_route"], v["stack"],
pt_exes=pt_exes, sniff_action=v["sniff_action"], settings=settings,
)
SINGBOX_CONFIG.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
rc, out, err = run_cmd([str(se), "check", "-c", str(SINGBOX_CONFIG)], timeout=120)
if rc != 0:
self.log("sing-box check failed:")
self.log((err or out).strip())
continue
scf = open(SINGBOX_CONSOLE_LOG, "wb")
self.handles.append(scf)
sp = subprocess.Popen(
[str(se), "run", "-c", str(SINGBOX_CONFIG)],
stdout=scf, stderr=scf, creationflags=CREATE_NO_WINDOW, cwd=str(APP_DIR), env=env,
)
self.spawned.append(sp)
time.sleep(4)
if sp.poll() is not None:
self.log("sing-box exited immediately.")
sp = None
continue
ok, tgt = socks5_probe(socks_port, timeout_each=12)
if ok:
started = True
chosen = v
self.log(f"Internet probe via Tor PASSED ({tgt}).")
break
self.log("Probe failed - trying next variant...")
try:
if sp.poll() is None:
sp.kill()
except Exception:
pass
sp = None
run_silent(["netsh", "interface", "set", "interface", "name=TorLeakGuard", "admin=disabled"])
time.sleep(1)
if not started:
self.log("No variant gave working internet via Tor.")
self.rollback(tp, sp, self.log)
return
add_ipv6_block(self.log, tor_exe=te, pt_exes=pt_exes)
STATE_FILE.write_text(
json.dumps(
{
"tor_pid": tp.pid,
"sing_box_pid": sp.pid,
"socks_port": socks_port,
"dns_port": dns_port,
"control_port": control_port,
"mode": settings.get("mode", "normal"),
"pt_exes": [str(x) for x in pt_exes],
"started_at": time.strftime("%Y-%m-%d %H:%M:%S"),
},
indent=2,
),
encoding="utf-8",
)
self.log("Started.")
self.log(f"SOCKS5 Tor: 127.0.0.1:{socks_port}")
self.log(f"DNS Tor: 127.0.0.1:{dns_port}")
self.log(f"Control: 127.0.0.1:{control_port}")
self.log(f"sing-box: sr={chosen['strict_route']} stack={chosen['stack']} sniff={chosen['sniff_action']}")
self.log("Mode: " + ("bridges." if settings.get("mode") == "bridge" else "plain Tor."))
self.log("Automatic rotation " + ("disabled." if settings.get("prevent_circuit_change", True) else "default."))
if settings.get("mode") == "bridge":
self.refresh_bridge_status(settings.get("bridges", ""), True)
def main():
ensure_dirs()
root = tk.Tk()
app = App(root)
if "--autostart" in sys.argv:
root.after(500, app.on_start)
if "--autoinstall" in sys.argv:
root.after(500, app.on_install)
if "--autoupdate" in sys.argv:
root.after(500, app.on_update)
if "--autostop" in sys.argv:
root.after(500, app.on_stop)
root.mainloop()
if __name__ == "__main__":
main()
# === END OF FILE ===
━━━━━━ 
First thing after you connect — prove it’s actually anonymous ━━━━━━
A system-wide router is only as good as its leaks. Open these once Tor is up — takes 30 seconds:
Am I on Tor? official green-light check → https://check.torproject.org/ (scriptable JSON for a killswitch → https://check.torproject.org/api/ip)
All-in-one leak dash (IP + DNS + WebRTC + IPv6) → https://ipleak.net/ ↳ Tor is IPv4-only, so any IPv6 row = leak- DNS leak — extended → https://dnsleaktest.com/ (no ISP resolver should appear)
WebRTC leak (system Tor does NOT stop this) → https://browserleaks.com/webrtc
Fingerprint check → https://coveryourtracks.eff.org/
━━━━━━ Get the most anonymity out of it — do & don’t ━━━━━━
Never torrent over Tor — BitTorrent leaks your real IP straight to peers, killswitch or not → https://blog.torproject.org/bittorrent-over-tor-isnt-good-idea/
The anonymity DoNot checklist (mistakes that unmask you) → https://www.whonix.org/wiki/DoNot
Stream isolation so different apps don’t share one circuit → https://www.whonix.org/wiki/Stream_Isolation
Authoritative transparent-proxy torrc/config reference → https://gitlab.torproject.org/legacy/trac/-/wikis/doc/TransparentProxy
━━━━━━ 
If Tor is blocked on your network (school / country / ISP) ━━━━━━
- Grab fresh obfs4 / WebTunnel bridge lines → https://bridges.torproject.org/
- How to drop them into a plain tor daemon’s torrc (exactly this setup) → https://support.torproject.org/little-t-tor/circumvention/using-bridges/
━━━━━━ Exit-node data (pairs perfectly with the built-in rating system) ━━━━━━
Cross-check what the app scores against the live Tor consensus:
Relay Search — look up any exit’s flags/bandwidth/uptime → https://metrics.torproject.org/rs.html
Live bulk exit list (one IP per line, diff-able) → https://check.torproject.org/torbulkexitlist
Full node feed with flags → https://www.dan.me.uk/tornodes
Verify or rebuild the components yourself (don’t just trust a binary)
- Tor Expert Bundle (official, +
.ascsignature) → https://www.torproject.org/download/tor/ 
tun2socks (official releases + source) → https://github.com/xjasonlyu/tun2socks/releases
Wintun (signed DLL, official) → https://www.wintun.net/
Related anonymity tools worth knowing
Tallow / TorWall — another transparent Tor firewall for Windows → https://github.com/basil00/Tallow- OnionFruit — Windows Tor proxy gateway with a GUI → https://github.com/dragonfruitnetwork/onionfruit
Whonix (VM, strongest isolation) → https://www.whonix.org/ · Tails (amnesic live USB) → https://tails.net/- Tor Browser — the hardened, anti-fingerprint baseline → https://www.torproject.org/download/
Continuation of this topic → 🕵️ Make Every App on Your Windows PC Anonymous Through Tor — One Click, No VPN, No Setup, Completely Free
!