", _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/](https://check.torproject.org/) (scriptable JSON for a killswitch β [https://check.torproject.org/api/ip](https://check.torproject.org/api/ip))
- π **All-in-one leak dash** (IP + DNS + WebRTC + IPv6) β [https://ipleak.net/](https://ipleak.net/) β³ Tor is IPv4-only, so _any_ IPv6 row = leak
- π§ **DNS leak β extended** β [https://dnsleaktest.com/](https://dnsleaktest.com/) (no ISP resolver should appear)
- π‘ **WebRTC leak** (system Tor does NOT stop this) β [https://browserleaks.com/webrtc](https://browserleaks.com/webrtc)
- π΅ **Fingerprint check** β [https://coveryourtracks.eff.org/](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/](https://blog.torproject.org/bittorrent-over-tor-isnt-good-idea/)
- π The anonymity **DoNot checklist** (mistakes that unmask you) β [https://www.whonix.org/wiki/DoNot](https://www.whonix.org/wiki/DoNot)
- 𧩠**Stream isolation** so different apps donβt share one circuit β [https://www.whonix.org/wiki/Stream\_Isolation](https://www.whonix.org/wiki/Stream_Isolation)
- π Authoritative transparent-proxy torrc/config reference β [https://gitlab.torproject.org/legacy/trac/-/wikis/doc/TransparentProxy](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/](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/](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](https://metrics.torproject.org/rs.html)
- π Live **bulk exit list** (one IP per line, diff-able) β [https://check.torproject.org/torbulkexitlist](https://check.torproject.org/torbulkexitlist)
- π Full node feed with flags β [https://www.dan.me.uk/tornodes](https://www.dan.me.uk/tornodes)
## π¦ Verify or rebuild the components yourself (donβt just trust a binary)
- π§
Tor Expert Bundle (official, +`.asc` signature) β [https://www.torproject.org/download/tor/](https://www.torproject.org/download/tor/)
- π tun2socks (official releases + source) β [https://github.com/xjasonlyu/tun2socks/releases](https://github.com/xjasonlyu/tun2socks/releases)
- 𧡠Wintun (signed DLL, official) β [https://www.wintun.net/](https://www.wintun.net/)
## 𧩠Related anonymity tools worth knowing
- πͺ **Tallow / TorWall** β another transparent Tor firewall for Windows β [https://github.com/basil00/Tallow](https://github.com/basil00/Tallow)
- π§
**OnionFruit** β Windows Tor proxy gateway with a GUI β [https://github.com/dragonfruitnetwork/onionfruit](https://github.com/dragonfruitnetwork/onionfruit)
- π° **Whonix** (VM, strongest isolation) β [https://www.whonix.org/](https://www.whonix.org/) Β· **Tails** (amnesic live USB) β [https://tails.net/](https://tails.net/)
- π **Tor Browser** β the hardened, anti-fingerprint baseline β [https://www.torproject.org/download/](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](https://onehack.st/t/make-every-app-on-your-windows-pc-anonymous-through-tor-one-click-no-vpn-no-setup-completely-free/324036)
---
### Author:  [@nikitin.sergey1998](https://onehack.st/u/nikitin.sergey1998)
#### Post date: [July 23, 2026, 9:36am UTC](https://onehack.st/t/tor-system-router-v12-route-your-entire-windows-pc-through-tor-gui-no-vpn-update-24-07-2026/324081/2 "2026-07-23T09:36:55Z")
A new update in less than 24 hoursβawesome!
---
### Author:  [@Edgar](https://onehack.st/u/Edgar)
#### Post date: [July 23, 2026, 9:52am UTC](https://onehack.st/t/tor-system-router-v12-route-your-entire-windows-pc-through-tor-gui-no-vpn-update-24-07-2026/324081/3 "2026-07-23T09:52:19Z")
Every issue I notice gets fixed within 24 hoursβoften even faster.
---
### Author:  [@nikitin.sergey1998](https://onehack.st/u/nikitin.sergey1998)
#### Post date: [July 23, 2026, 6:12pm UTC](https://onehack.st/t/tor-system-router-v12-route-your-entire-windows-pc-through-tor-gui-no-vpn-update-24-07-2026/324081/4 "2026-07-23T18:12:03Z")
Thank you for the quick response!
---
### Author:  [@Edgar](https://onehack.st/u/Edgar)
#### Post date: [July 24, 2026, 9:46am UTC](https://onehack.st/t/tor-system-router-v12-route-your-entire-windows-pc-through-tor-gui-no-vpn-update-24-07-2026/324081/5 "2026-07-24T09:46:11Z")
At this point, Version 12 is considered the final release. Future updates will only be made if changes to Tor require them. The application already includes the maximum level of protection and all the features I intended to implement. Adding more functionality at this stage would likely compromise website compatibility, so there is little practical benefit in doing so. The software has been designed with long-term stability in mind and should remain reliable for many years to come.