Tor System Router v9 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, tun2socks 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
Technical specs
OS Windows (needs Administrator) · Components Tor Expert Bundle + tun2socks + Wintun · Protocol SOCKS5 over Tor · Adapter tor0 (10.192.0.2/24) · Ports auto-assigned (SOCKS · DNS · Control)
📜 Full source code — Tor System Router v9 (copy, save as a .py, run as Administrator)
import ctypes
import json
import os
import random
import re
import shutil
import socket
import subprocess
import sys
import tarfile
import threading
import time
import tkinter as tk
import urllib.request
import zipfile
from pathlib import Path
from tkinter import messagebox, scrolledtext, ttk
BASE_DIR = Path(r"C:\\TorSetup")
CONFIG_PATH = Path(__file__).with_suffix(".config.json")
EXIT_RATING_PATH = BASE_DIR / "exit_rating.json"
TUN_NAME = "tor0"
TUN_IP = "10.192.0.2"
TUN_MASK = "255.255.255.0"
TUN_GATEWAY = "10.192.0.1"
FW_RULE_PREFIX = "TorRouterGUI_"
FALLBACK_TOR_VERSION = "14.5.1"
FALLBACK_WINTUN_URL = "https://www.wintun.net/builds/wintun-0.14.1.zip"
EXIT_SCORE_BAD = -100
EXIT_SCORE_CAPTCHA = -50
EXIT_CAPTCHA_THRESHOLD = 5
def is_admin() -> bool:
try:
return ctypes.windll.shell32.IsUserAnAdmin() != 0
except Exception:
return False
def relaunch_as_admin():
params = " ".join(f'"{a}"' for a in sys.argv)
ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, params, None, 1)
sys.exit(0)
def run(cmd, check=False):
proc = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, "CREATE_NO_WINDOW") else 0,
)
out = (proc.stdout or "") + (proc.stderr or "")
if check and proc.returncode != 0:
raise RuntimeError(
f"Command failed (rc={proc.returncode}): {cmd}\n{out}"
)
return proc.returncode, out
def find_free_port(*candidates):
for port in candidates:
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("127.0.0.1", port))
s.close()
return port
except OSError:
continue
while True:
port = random.randint(15000, 65000)
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("127.0.0.1", port))
s.close()
return port
except OSError:
continue
def get_processes_using_port(port):
pids = []
try:
rc, out = run(f'netstat -ano | findstr ":{port}"', check=False)
for line in out.split("\n"):
if "LISTENING" in line or "ESTABLISHED" in line:
parts = line.split()
if len(parts) >= 5:
pid = parts[-1]
if pid.isdigit():
pids.append(int(pid))
except Exception:
pass
return list(set(pids))
def kill_processes_by_pid(pids):
for pid in pids:
try:
run(f'taskkill /PID {pid} /F', check=False)
except Exception:
pass
def kill_processes_by_name(names):
for name in names:
try:
run(f'taskkill /F /IM {name}', check=False)
except Exception:
pass
def is_port_free(port):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(1)
s.bind(("127.0.0.1", port))
s.close()
return True
except OSError:
return False
class DNSForwarder:
def __init__(self, listen_port: int, target_port: int):
self.listen_port = listen_port
self.target_port = target_port
self.sock = None
self.thread = None
self.running = False
def start(self):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.bind(("127.0.0.1", self.listen_port))
self.running = True
self.thread = threading.Thread(target=self._loop, daemon=True)
self.thread.start()
def _handle_one(self, data: bytes, addr):
try:
fwd = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
fwd.settimeout(5)
fwd.sendto(data, ("127.0.0.1", self.target_port))
resp, _ = fwd.recvfrom(512)
self.sock.sendto(resp, addr)
fwd.close()
except Exception:
pass
def _loop(self):
while self.running:
try:
data, addr = self.sock.recvfrom(512)
threading.Thread(
target=self._handle_one, args=(data, addr), daemon=True
).start()
except Exception:
pass
def stop(self):
self.running = False
if self.sock:
try:
self.sock.close()
except Exception:
pass
class ExitRating:
def __init__(self, path: Path):
self.path = path
self.data = self._load()
def _load(self):
if self.path.exists():
try:
with open(self.path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {}
def _save(self):
try:
with open(self.path, "w", encoding="utf-8") as f:
json.dump(self.data, f, indent=2)
except Exception:
pass
def record(self, exit_ip: str, success: bool, captcha: bool = False, error: bool = False):
if exit_ip not in self.data:
self.data[exit_ip] = {"ok": 0, "fail": 0, "captcha": 0, "score": 0}
entry = self.data[exit_ip]
if success:
entry["ok"] += 1
entry["score"] += 10
if captcha:
entry["captcha"] += 1
entry["score"] -= 30
if error:
entry["fail"] += 1
entry["score"] -= 20
self._save()
def get_score(self, exit_ip: str) -> int:
return self.data.get(exit_ip, {}).get("score", 0)
def get_captcha_count(self, exit_ip: str) -> int:
return self.data.get(exit_ip, {}).get("captcha", 0)
def is_bad_exit(self, exit_ip: str) -> bool:
if not exit_ip or exit_ip == "unknown":
return False
score = self.get_score(exit_ip)
captcha_count = self.get_captcha_count(exit_ip)
return score < EXIT_SCORE_BAD or captcha_count >= EXIT_CAPTCHA_THRESHOLD
def is_suspicious_exit(self, exit_ip: str) -> bool:
if not exit_ip or exit_ip == "unknown":
return False
score = self.get_score(exit_ip)
captcha_count = self.get_captcha_count(exit_ip)
return score < EXIT_SCORE_CAPTCHA or captcha_count >= 3
def get_blacklist(self) -> list:
return [ip for ip, data in self.data.items()
if data.get("score", 0) < EXIT_SCORE_BAD
or data.get("captcha", 0) >= EXIT_CAPTCHA_THRESHOLD]
class TorControlClient:
def __init__(self, port: int, log_fn=None):
self.port = port
self.log = log_fn or (lambda x: None)
self.sock = None
self.connected = False
self._lock = threading.Lock()
self.current_exit = None
self.current_exit_ip = None
def connect(self, timeout=10):
try:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(timeout)
self.sock.connect(("127.0.0.1", self.port))
banner = self._recv_line()
if not banner.startswith("250"):
self.log(f"[control] Unexpected banner: {banner}")
return False
cookie_path = BASE_DIR / "tordata" / "control_auth_cookie"
if not cookie_path.exists():
self.log("[control] Cookie file not found, trying without auth")
self._send("AUTHENTICATE")
else:
with open(cookie_path, "rb") as f:
cookie_hex = f.read().hex()
self._send(f"AUTHENTICATE {cookie_hex}")
resp = self._recv_line()
if not resp.startswith("250"):
self.log(f"[control] Auth error: {resp}")
return False
self._send("SETEVENTS CIRC STREAM")
resp = self._recv_line()
if not resp.startswith("250"):
self.log(f"[control] Event subscription error: {resp}")
self.connected = True
self.log(f"[control] Connected to ControlPort {self.port}")
return True
except Exception as e:
self.log(f"[control] Connection error: {e}")
return False
def _send(self, cmd: str):
self.sock.sendall(f"{cmd}\r\n".encode())
def _recv_line(self) -> str:
buf = b""
while True:
chunk = self.sock.recv(1)
if not chunk:
break
buf += chunk
if buf.endswith(b"\r\n"):
break
return buf.decode().strip()
def _recv_multiline(self) -> list:
lines = []
while True:
line = self._recv_line()
lines.append(line)
if line.startswith("250 ") and line != "250-":
break
return lines
def get_current_exit(self) -> tuple[str, str]:
try:
with self._lock:
self._send("GETINFO circuit-status")
lines = self._recv_multiline()
for line in lines:
if line.startswith("250+circuit-status=") or line.startswith("250-circuit-status="):
continue
if line.startswith("250 "):
break
parts = line.split()
if len(parts) >= 3 and parts[1] == "BUILT":
hops = parts[2].split(",")
if hops:
last_hop = hops[-1]
if "~" in last_hop:
fp = last_hop.split("~")[0].lstrip("$")
name = last_hop.split("~")[1]
self.current_exit = name
ip = self._get_exit_ip(fp)
self.current_exit_ip = ip
return name, ip
return None, None
except Exception as e:
self.log(f"[control] Error getting exit: {e}")
return None, None
def _get_exit_ip(self, fingerprint: str) -> str:
try:
with self._lock:
self._send(f"GETINFO ns/id/{fingerprint}")
lines = self._recv_multiline()
for line in lines:
if "a " in line and "." in line:
match = re.search(r"a (\d+\.\d+\.\d+\.\d+)", line)
if match:
return match.group(1)
return None
except Exception:
return None
def newnym(self) -> bool:
try:
with self._lock:
self._send("SIGNAL NEWNYM")
resp = self._recv_line()
if resp.startswith("250"):
self.log("[control] NEWNYM: new circuit requested")
self.current_exit = None
self.current_exit_ip = None
return True
else:
self.log(f"[control] NEWNYM error: {resp}")
return False
except Exception as e:
self.log(f"[control] NEWNYM exception: {e}")
return False
def drop_guards(self) -> bool:
try:
with self._lock:
self._send("SIGNAL DROPGUARDS")
resp = self._recv_line()
if resp.startswith("250"):
self.log("[control] DROPGUARDS: guards dropped")
return True
return False
except Exception as e:
self.log(f"[control] DROPGUARDS exception: {e}")
return False
def close(self):
if self.sock:
try:
self.sock.close()
except Exception:
pass
self.connected = False
def monitor_events(self, callback):
def _loop():
while self.connected:
try:
line = self._recv_line()
if line.startswith("650 "):
event = line[4:]
callback(event)
except Exception:
break
threading.Thread(target=_loop, daemon=True).start()
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
@dataclass
class ComponentStatus:
name: str
installed: bool = False
version: Optional[str] = None
latest_version: Optional[str] = None
path: Optional[Path] = None
needs_update: bool = False
needs_install: bool = False
error: Optional[str] = None
info: str = ""
class Component(ABC):
def __init__(self, log_fn=None, progress_fn=None):
self.log = log_fn or (lambda x: None)
self.progress = progress_fn or (lambda c, t: None)
self.status = ComponentStatus(name=self.name)
@property
@abstractmethod
def name(self) -> str:
pass
@abstractmethod
def check(self) -> ComponentStatus:
pass
@abstractmethod
def download(self, target_dir: Path) -> Path:
pass
@abstractmethod
def install(self, archive_path: Path, install_dir: Path) -> Path:
pass
def full_update(self, install_dir: Path) -> ComponentStatus:
self.status = self.check()
if not self.status.installed:
self.log(f"[x] {self.name} not found")
elif self.status.needs_update:
self.log(f"[^] {self.name} {self.status.version} -> {self.status.latest_version}")
else:
self.log(f"[ok] {self.name} is up to date ({self.status.version})")
return self.status
if self.status.needs_install or self.status.needs_update:
install_dir.mkdir(parents=True, exist_ok=True)
temp_dir = install_dir / "_temp"
temp_dir.mkdir(parents=True, exist_ok=True)
try:
self.log(f" Downloading {self.name}...")
archive = self.download(temp_dir)
self.log(f" Installing {self.name}...")
result_path = self.install(archive, install_dir)
self.status.path = result_path
self.status.installed = True
self.status.needs_install = False
self.status.needs_update = False
self.status.version = self.status.latest_version
self.log(f"[ok] {self.name} installed: {result_path}")
except Exception as e:
self.status.error = str(e)
self.log(f"[x] {self.name} install error: {e}")
finally:
if temp_dir.exists():
shutil.rmtree(temp_dir, ignore_errors=True)
return self.status
class TorComponent(Component):
@property
def name(self) -> str:
return "Tor Expert Bundle"
def _get_latest_version(self) -> str:
try:
req = urllib.request.Request(
"https://dist.torproject.org/torbrowser/?C=M;O=D",
headers={"User-Agent": "Mozilla/5.0"},
)
with urllib.request.urlopen(req, timeout=15) as resp:
html = resp.read().decode("utf-8", errors="ignore")
versions = re.findall(r'href="(\d+\.\d+\.\d+)/"', html)
if versions:
def key(v):
return tuple(int(x) for x in v.split("."))
return sorted(set(versions), key=key, reverse=True)[0]
except Exception as e:
self.log(f" Failed to get Tor version ({e})")
return FALLBACK_TOR_VERSION
def _find_installed(self) -> tuple[Optional[Path], Optional[str]]:
config = {}
if CONFIG_PATH.exists():
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
config = json.load(f)
except Exception:
pass
cached = config.get("tor_exe")
if cached and Path(cached).exists():
return Path(cached), self._get_exe_version(Path(cached))
p = shutil.which("tor")
if p:
return Path(p), self._get_exe_version(Path(p))
home = Path.home()
pf = Path(os.environ.get("PROGRAMFILES", r"C:\\Program Files"))
pf86 = Path(os.environ.get("PROGRAMFILES(X86)", r"C:\\Program Files (x86)"))
local = Path(os.environ.get("LOCALAPPDATA", r""))
candidates = [
BASE_DIR / "tor" / "tor.exe",
BASE_DIR / "Tor" / "tor.exe",
pf / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
pf86 / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
local / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
home / "Desktop" / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
home / "Downloads" / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
]
for c in candidates:
if c.exists():
return c, self._get_exe_version(c)
rc, out = run(
'powershell -NoProfile -Command "'
'Get-ChildItem -Path C:\\, $env:LOCALAPPDATA, $env:PROGRAMFILES, $env:PROGRAMFILES(X86) '
'-Filter tor.exe -Recurse -ErrorAction SilentlyContinue -Depth 4 | '
'Select-Object -First 1 -ExpandProperty FullName"'
)
pth = out.strip().splitlines()[-1].strip() if out.strip() else ""
if pth and Path(pth).exists():
return Path(pth), self._get_exe_version(Path(pth))
return None, None
def _get_exe_version(self, exe_path: Path) -> Optional[str]:
match = re.search(r'(\d+\.\d+\.\d+)', str(exe_path))
if match:
return match.group(1)
return None
def check(self) -> ComponentStatus:
self.status = ComponentStatus(name=self.name)
self.status.latest_version = self._get_latest_version()
exe_path, version = self._find_installed()
if exe_path:
self.status.installed = True
self.status.path = exe_path
self.status.version = version or "unknown"
if version and version != self.status.latest_version:
self.status.needs_update = True
self.status.info = f"Version {self.status.latest_version} available"
else:
self.status.info = "Up to date"
else:
self.status.needs_install = True
self.status.info = "Not installed"
return self.status
def download(self, target_dir: Path) -> Path:
version = self.status.latest_version or FALLBACK_TOR_VERSION
url = (
f"https://dist.torproject.org/torbrowser/{version}/"
f"tor-expert-bundle-windows-x86_64-{version}.tar.gz"
)
archive = target_dir / "tor-download.tar.gz"
try:
self._download_file(url, archive)
except Exception as e:
self.log(f" Primary URL failed ({e}). Trying fallback...")
fallback_url = (
"https://archive.torproject.org/tor-package-archive/torbrowser/"
f"{version}/tor-expert-bundle-windows-x86_64-{version}.tar.gz"
)
self._download_file(fallback_url, archive)
return archive
def _download_file(self, url: str, dest: Path):
self.log(f" -> {url}")
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=120) as resp:
total = int(resp.headers.get("Content-Length", 0))
block = 8192
downloaded = 0
with open(dest, "wb") as f:
while True:
chunk = resp.read(block)
if not chunk:
break
f.write(chunk)
downloaded += len(chunk)
if total > 0:
self.progress(downloaded, total)
self.progress(1, 1)
def install(self, archive_path: Path, install_dir: Path) -> Path:
with tarfile.open(archive_path, "r:gz") as tar:
tar.extractall(path=install_dir)
exe = install_dir / "tor" / "tor.exe"
if not exe.exists():
for p in install_dir.rglob("tor.exe"):
exe = p
break
if not exe.exists():
raise RuntimeError("tor.exe not found after extraction")
config = {}
if CONFIG_PATH.exists():
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
config = json.load(f)
except Exception:
pass
config["tor_exe"] = str(exe)
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
return exe
class Tun2SocksComponent(Component):
@property
def name(self) -> str:
return "tun2socks"
def _get_latest_version(self) -> str:
try:
req = urllib.request.Request(
"https://github.com/xjasonlyu/tun2socks/releases/latest",
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"},
)
with urllib.request.urlopen(req, timeout=20) as resp:
final_url = resp.geturl()
html = resp.read().decode("utf-8", errors="ignore")
m = re.search(r'/tag/([^/"\s]+)', final_url)
if m:
return m.group(1)
except Exception as e:
self.log(f" Failed to get tun2socks version ({e})")
return "v2.5.2"
def _find_installed(self) -> tuple[Optional[Path], Optional[str]]:
config = {}
if CONFIG_PATH.exists():
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
config = json.load(f)
except Exception:
pass
cached = config.get("tun2socks_exe")
if cached and Path(cached).exists():
return Path(cached), self._get_exe_version(Path(cached))
p = shutil.which("tun2socks-windows-amd64") or shutil.which("tun2socks")
if p:
return Path(p), self._get_exe_version(Path(p))
candidates = [
BASE_DIR / "tun2socks" / "tun2socks-windows-amd64.exe",
BASE_DIR / "tun2socks" / "tun2socks.exe",
]
for c in candidates:
if c.exists():
return c, self._get_exe_version(c)
rc, out = run(
'powershell -NoProfile -Command "'
'Get-ChildItem -Path C:\\, $env:LOCALAPPDATA, $env:PROGRAMFILES, $env:PROGRAMFILES(X86) '
'-Filter tun2socks-windows-amd64.exe -Recurse -ErrorAction SilentlyContinue -Depth 4 | '
'Select-Object -First 1 -ExpandProperty FullName"'
)
pth = out.strip().splitlines()[-1].strip() if out.strip() else ""
if pth and Path(pth).exists():
return Path(pth), self._get_exe_version(Path(pth))
return None, None
def _get_exe_version(self, exe_path: Path) -> Optional[str]:
match = re.search(r'v(\d+\.\d+\.\d+)', str(exe_path))
if match:
return match.group(1)
return None
def check(self) -> ComponentStatus:
self.status = ComponentStatus(name=self.name)
self.status.latest_version = self._get_latest_version()
exe_path, version = self._find_installed()
if exe_path:
self.status.installed = True
self.status.path = exe_path
self.status.version = version or "unknown"
self.status.info = "Installed"
else:
self.status.needs_install = True
self.status.info = "Not installed"
return self.status
def download(self, target_dir: Path) -> Path:
version = self.status.latest_version or "v2.5.2"
download_url = (
"https://github.com/xjasonlyu/tun2socks/releases/download/"
f"{version}/tun2socks-windows-amd64.zip"
)
archive = target_dir / "tun2socks.zip"
self._download_file(download_url, archive)
return archive
def _download_file(self, url: str, dest: Path):
self.log(f" -> {url}")
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=120) as resp:
total = int(resp.headers.get("Content-Length", 0))
block = 8192
downloaded = 0
with open(dest, "wb") as f:
while True:
chunk = resp.read(block)
if not chunk:
break
f.write(chunk)
downloaded += len(chunk)
if total > 0:
self.progress(downloaded, total)
self.progress(1, 1)
def install(self, archive_path: Path, install_dir: Path) -> Path:
with zipfile.ZipFile(archive_path, "r") as z:
z.extractall(install_dir)
exe = install_dir / "tun2socks-windows-amd64.exe"
if not exe.exists():
for p in install_dir.rglob("tun2socks-windows-amd64.exe"):
exe = p
break
if not exe.exists():
raise RuntimeError("tun2socks not found after extraction")
config = {}
if CONFIG_PATH.exists():
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
config = json.load(f)
except Exception:
pass
config["tun2socks_exe"] = str(exe)
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
return exe
class WintunComponent(Component):
@property
def name(self) -> str:
return "Wintun"
def _get_latest_version(self) -> str:
try:
req = urllib.request.Request(
"https://www.wintun.net/",
headers={"User-Agent": "Mozilla/5.0"},
)
with urllib.request.urlopen(req, timeout=15) as resp:
html = resp.read().decode("utf-8", errors="ignore")
m = re.search(r'href="(https://www\.wintun\.net/builds/wintun-([\d\.]+)\.zip)"', html)
if m:
return m.group(2)
except Exception as e:
self.log(f" Failed to get Wintun version ({e})")
return "0.14.1"
def _find_installed(self) -> tuple[Optional[Path], Optional[str]]:
config = {}
if CONFIG_PATH.exists():
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
config = json.load(f)
except Exception:
pass
cached = config.get("wintun_dll")
if cached and Path(cached).exists():
return Path(cached), self._get_dll_version(Path(cached))
tun2socks = config.get("tun2socks_exe")
if tun2socks:
dll = Path(tun2socks).parent / "wintun.dll"
if dll.exists():
return dll, self._get_dll_version(dll)
rc, out = run(
'powershell -NoProfile -Command "'
'Get-ChildItem -Path C:\\Windows\\System32, C:\\Windows\\SysWOW64 '
'-Filter wintun.dll -Recurse -ErrorAction SilentlyContinue | '
'Select-Object -First 1 -ExpandProperty FullName"'
)
pth = out.strip().splitlines()[-1].strip() if out.strip() else ""
if pth and Path(pth).exists():
return Path(pth), self._get_dll_version(Path(pth))
return None, None
def _get_dll_version(self, dll_path: Path) -> Optional[str]:
match = re.search(r'wintun-([\d\.]+)', str(dll_path))
if match:
return match.group(1)
return None
def check(self) -> ComponentStatus:
self.status = ComponentStatus(name=self.name)
self.status.latest_version = self._get_latest_version()
dll_path, version = self._find_installed()
if dll_path:
self.status.installed = True
self.status.path = dll_path
self.status.version = version or "unknown"
if version and version != self.status.latest_version:
self.status.needs_update = True
self.status.info = f"Version {self.status.latest_version} available"
else:
self.status.info = "Up to date"
else:
self.status.needs_install = True
self.status.info = "Not installed"
return self.status
def download(self, target_dir: Path) -> Path:
version = self.status.latest_version or "0.14.1"
url = f"https://www.wintun.net/builds/wintun-{version}.zip"
archive = target_dir / "wintun.zip"
try:
self._download_file(url, archive)
except Exception:
url = FALLBACK_WINTUN_URL
self._download_file(url, archive)
return archive
def _download_file(self, url: str, dest: Path):
self.log(f" -> {url}")
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=120) as resp:
total = int(resp.headers.get("Content-Length", 0))
block = 8192
downloaded = 0
with open(dest, "wb") as f:
while True:
chunk = resp.read(block)
if not chunk:
break
f.write(chunk)
downloaded += len(chunk)
if total > 0:
self.progress(downloaded, total)
self.progress(1, 1)
def install(self, archive_path: Path, install_dir: Path) -> Path:
with zipfile.ZipFile(archive_path, "r") as z:
z.extractall(install_dir)
dll = install_dir / "wintun" / "bin" / "amd64" / "wintun.dll"
if not dll.exists():
for p in install_dir.rglob("wintun.dll"):
dll = p
break
if not dll.exists():
raise RuntimeError("wintun.dll not found after extraction")
config = {}
if CONFIG_PATH.exists():
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
config = json.load(f)
except Exception:
pass
tun2socks = config.get("tun2socks_exe")
if tun2socks:
dest = Path(tun2socks).parent / "wintun.dll"
shutil.copy2(dll, dest)
dll = dest
extracted = install_dir / "wintun"
if extracted.exists():
shutil.rmtree(extracted)
config["wintun_dll"] = str(dll)
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
return dll
class ComponentManager:
def __init__(self, log_fn=None, progress_fn=None):
self.log = log_fn or (lambda x: None)
self.progress = progress_fn or (lambda c, t: None)
self.components: list = []
self._register_default_components()
def _register_default_components(self):
self.components = [
TorComponent(self.log, self.progress),
Tun2SocksComponent(self.log, self.progress),
WintunComponent(self.log, self.progress),
]
def register(self, component: Component):
self.components.append(component)
self.log(f"[components] Registered: {component.name}")
def check_all(self) -> list:
self.log("=== Checking for updates ===")
results = []
for comp in self.components:
status = comp.check()
results.append(status)
if status.installed and not status.needs_update:
self.log(f"[ok] {status.name} up to date ({status.version})")
elif status.needs_update:
self.log(f"[^] {status.name} update available ({status.version} -> {status.latest_version})")
else:
self.log(f"[x] {status.name} not installed")
return results
def update_all(self) -> list:
self.log("=== Updating components ===")
results = []
for comp in self.components:
self.log(f"\n--- {comp.name} ---")
status = comp.full_update(BASE_DIR)
results.append(status)
self.log("\n=== Update complete ===")
return results
def get_component(self, name: str) -> Optional[Component]:
for comp in self.components:
if comp.name.lower() == name.lower():
return comp
return None
class DependencyManager:
def __init__(self, log_callback=None, progress_callback=None):
self.log = log_callback or (lambda x: None)
self.progress = progress_callback or (lambda cur, total: None)
self.config = self._load_config()
def _load_config(self):
if CONFIG_PATH.exists():
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {}
def _save_config(self):
try:
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
json.dump(self.config, f, indent=2)
except Exception as e:
self.log(f"Failed to save config: {e}")
def find_tor(self) -> Path | None:
cached = self.config.get("tor_exe")
if cached and Path(cached).exists():
return Path(cached)
p = shutil.which("tor")
if p:
return Path(p)
home = Path.home()
pf = Path(os.environ.get("PROGRAMFILES", r"C:\\Program Files"))
pf86 = Path(os.environ.get("PROGRAMFILES(X86)", r"C:\\Program Files (x86)"))
local = Path(os.environ.get("LOCALAPPDATA", r""))
candidates = [
BASE_DIR / "tor" / "tor.exe",
BASE_DIR / "Tor" / "tor.exe",
pf / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
pf86 / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
local / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
home / "Desktop" / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
home / "Downloads" / "Tor Browser" / "Browser" / "TorBrowser" / "Tor" / "tor.exe",
]
for c in candidates:
if c.exists():
self.config["tor_exe"] = str(c)
self._save_config()
return c
self.log("Searching for tor.exe via PowerShell...")
rc, out = run(
'powershell -NoProfile -Command "'
'Get-ChildItem -Path C:\\, $env:LOCALAPPDATA, $env:PROGRAMFILES, $env:PROGRAMFILES(X86) '
'-Filter tor.exe -Recurse -ErrorAction SilentlyContinue -Depth 4 | '
'Select-Object -First 1 -ExpandProperty FullName"'
)
pth = out.strip().splitlines()[-1].strip() if out.strip() else ""
if pth and Path(pth).exists():
self.config["tor_exe"] = pth
self._save_config()
return Path(pth)
return None
def find_tun2socks(self) -> Path | None:
cached = self.config.get("tun2socks_exe")
if cached and Path(cached).exists():
return Path(cached)
p = shutil.which("tun2socks-windows-amd64") or shutil.which("tun2socks")
if p:
return Path(p)
candidates = [
BASE_DIR / "tun2socks" / "tun2socks-windows-amd64.exe",
BASE_DIR / "tun2socks" / "tun2socks.exe",
]
for c in candidates:
if c.exists():
self.config["tun2socks_exe"] = str(c)
self._save_config()
return c
self.log("Searching for tun2socks via PowerShell...")
rc, out = run(
'powershell -NoProfile -Command "'
'Get-ChildItem -Path C:\\, $env:LOCALAPPDATA, $env:PROGRAMFILES, $env:PROGRAMFILES(X86) '
'-Filter tun2socks-windows-amd64.exe -Recurse -ErrorAction SilentlyContinue -Depth 4 | '
'Select-Object -First 1 -ExpandProperty FullName"'
)
pth = out.strip().splitlines()[-1].strip() if out.strip() else ""
if pth and Path(pth).exists():
self.config["tun2socks_exe"] = pth
self._save_config()
return Path(pth)
return None
def find_wintun(self, search_dir: Path) -> Path | None:
cached = self.config.get("wintun_dll")
if cached and Path(cached).exists():
return Path(cached)
dll = search_dir / "wintun.dll"
if dll.exists():
self.config["wintun_dll"] = str(dll)
self._save_config()
return dll
self.log("Searching for wintun.dll via PowerShell...")
rc, out = run(
'powershell -NoProfile -Command "'
'Get-ChildItem -Path C:\\Windows\\System32, C:\\Windows\\SysWOW64 '
'-Filter wintun.dll -Recurse -ErrorAction SilentlyContinue | '
'Select-Object -First 1 -ExpandProperty FullName"'
)
pth = out.strip().splitlines()[-1].strip() if out.strip() else ""
if pth and Path(pth).exists():
self.config["wintun_dll"] = pth
self._save_config()
return Path(pth)
return None
def _download_file(self, url: str, dest: Path):
self.log(f" -> {url}")
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=120) as resp:
total = int(resp.headers.get("Content-Length", 0))
block = 8192
downloaded = 0
with open(dest, "wb") as f:
while True:
chunk = resp.read(block)
if not chunk:
break
f.write(chunk)
downloaded += len(chunk)
if total > 0:
self.progress(downloaded, total)
self.progress(1, 1)
def _get_latest_tor_version(self) -> str:
try:
req = urllib.request.Request(
"https://dist.torproject.org/torbrowser/?C=M;O=D",
headers={"User-Agent": "Mozilla/5.0"},
)
with urllib.request.urlopen(req, timeout=15) as resp:
html = resp.read().decode("utf-8", errors="ignore")
versions = re.findall(r'href="(\d+\.\d+\.\d+)/"', html)
if versions:
def key(v):
return tuple(int(x) for x in v.split("."))
return sorted(set(versions), key=key, reverse=True)[0]
except Exception as e:
self.log(f"Failed to get latest Tor version ({e}), using fallback.")
return FALLBACK_TOR_VERSION
def download_tor(self, target_dir: Path) -> Path:
version = self._get_latest_tor_version()
self.log(f"Latest Tor version: {version}")
url = (
f"https://dist.torproject.org/torbrowser/{version}/"
f"tor-expert-bundle-windows-x86_64-{version}.tar.gz"
)
archive = target_dir / "tor-download.tar.gz"
target_dir.mkdir(parents=True, exist_ok=True)
self.log("Downloading Tor Expert Bundle...")
try:
self._download_file(url, archive)
except Exception as e:
self.log(f"Primary URL failed ({e}). Trying fallback...")
fallback_url = (
"https://archive.torproject.org/tor-package-archive/torbrowser/"
f"{version}/tor-expert-bundle-windows-x86_64-{version}.tar.gz"
)
self._download_file(fallback_url, archive)
self.log("Extracting Tor...")
with tarfile.open(archive, "r:gz") as tar:
tar.extractall(path=target_dir)
exe = target_dir / "tor" / "tor.exe"
if not exe.exists():
for p in target_dir.rglob("tor.exe"):
exe = p
break
if not exe.exists():
raise RuntimeError("tor.exe not found after extraction")
archive.unlink(missing_ok=True)
self.config["tor_exe"] = str(exe)
self._save_config()
self.log(f"Tor installed: {exe}")
return exe
def download_tun2socks(self, target_dir: Path) -> Path:
self.log("Getting latest tun2socks release info...")
download_url = None
version = None
try:
req = urllib.request.Request(
"https://github.com/xjasonlyu/tun2socks/releases/latest",
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"},
)
with urllib.request.urlopen(req, timeout=20) as resp:
final_url = resp.geturl()
html = resp.read().decode("utf-8", errors="ignore")
m = re.search(r'/tag/([^/"\s]+)', final_url)
if m:
version = m.group(1)
pattern = r'href="(/xjasonlyu/tun2socks/releases/download/[^"]*windows-amd64[^"]*\.zip)"'
match = re.search(pattern, html)
if match:
download_url = "https://github.com" + match.group(1)
self.log(f"Found release {version}, asset: {download_url}")
except Exception as e:
self.log(f"Release page parse failed ({e}).")
if not download_url:
version = "v2.5.2"
download_url = (
"https://github.com/xjasonlyu/tun2socks/releases/download/"
f"{version}/tun2socks-windows-amd64.zip"
)
self.log(f"Using fallback version {version}")
archive = target_dir / "tun2socks.zip"
target_dir.mkdir(parents=True, exist_ok=True)
self.log("Downloading tun2socks...")
self._download_file(download_url, archive)
with zipfile.ZipFile(archive, "r") as z:
z.extractall(target_dir)
exe = target_dir / "tun2socks-windows-amd64.exe"
if not exe.exists():
for p in target_dir.rglob("tun2socks-windows-amd64.exe"):
exe = p
break
if not exe.exists():
raise RuntimeError("tun2socks not found after extraction")
archive.unlink(missing_ok=True)
self.config["tun2socks_exe"] = str(exe)
self._save_config()
self.log(f"tun2socks installed: {exe}")
return exe
def download_wintun(self, target_dir: Path) -> Path:
self.log("Determining Wintun URL...")
url = FALLBACK_WINTUN_URL
try:
req = urllib.request.Request(
"https://www.wintun.net/",
headers={"User-Agent": "Mozilla/5.0"},
)
with urllib.request.urlopen(req, timeout=15) as resp:
html = resp.read().decode("utf-8", errors="ignore")
m = re.search(r'href="(https://www\.wintun\.net/builds/wintun-[\d\.]+\.zip)"', html)
if m:
url = m.group(1)
except Exception as e:
self.log(f"Failed to get latest Wintun ({e}), using fallback.")
archive = target_dir / "wintun.zip"
self.log("Downloading Wintun...")
self._download_file(url, archive)
with zipfile.ZipFile(archive, "r") as z:
z.extractall(target_dir)
dll = target_dir / "wintun" / "bin" / "amd64" / "wintun.dll"
if not dll.exists():
for p in target_dir.rglob("wintun.dll"):
dll = p
break
if not dll.exists():
raise RuntimeError("wintun.dll not found after extraction")
tun2socks_exe = Path(self.config.get("tun2socks_exe", target_dir / "tun2socks-windows-amd64.exe"))
dest = tun2socks_exe.parent / "wintun.dll"
shutil.copy2(dll, dest)
archive.unlink(missing_ok=True)
extracted = target_dir / "wintun"
if extracted.exists():
shutil.rmtree(extracted)
self.config["wintun_dll"] = str(dest)
self._save_config()
self.log(f"wintun.dll installed: {dest}")
return dest
def ensure_all(self) -> tuple[Path, Path, Path]:
self.log("=== Checking dependencies ===")
tor = self.find_tor()
if tor:
self.log(f"[OK] Tor found: {tor}")
else:
self.log("[!!] Tor not found. Installing automatically...")
tor = self.download_tor(BASE_DIR)
tun2socks = self.find_tun2socks()
if tun2socks:
self.log(f"[OK] tun2socks found: {tun2socks}")
else:
self.log("[!!] tun2socks not found. Installing automatically...")
tun2socks = self.download_tun2socks(BASE_DIR / "tun2socks")
wintun = self.find_wintun(tun2socks.parent)
if wintun:
self.log(f"[OK] wintun.dll found: {wintun}")
else:
self.log("[!!] wintun.dll not found. Installing automatically...")
wintun = self.download_wintun(BASE_DIR / "tun2socks")
return tor, tun2socks, wintun
class TorRouter:
def __init__(self, log_fn, dep_manager: DependencyManager):
self.log = log_fn
self.dep = dep_manager
self.tor_exe: Path | None = None
self.tun2socks_exe: Path | None = None
self.wintun_dll: Path | None = None
self.tor_proc = None
self.tun2socks_proc = None
self.dns_forwarder: DNSForwarder | None = None
self.control_client: TorControlClient | None = None
self.phys_alias = None
self.phys_ip = None
self.phys_metric_saved = None
self.tun_ifindex = None
self.socks_port = None
self.dns_port = None
self.control_port = None
self.running = False
self.exit_rating = ExitRating(EXIT_RATING_PATH)
self._last_exit_ip = None
self._circuit_change_count = 0
self._lock = threading.Lock()
def detect_physical_adapter(self):
rc, out = run(
'powershell -NoProfile -Command '
'"Get-NetIPConfiguration | Where-Object {$_.IPv4DefaultGateway -ne $null} '
'| Select-Object -First 1 -ExpandProperty InterfaceAlias"'
)
alias = out.strip().splitlines()[-1].strip() if out.strip() else ""
if not alias:
raise RuntimeError("Failed to find active network adapter with internet.")
self.phys_alias = alias
rc, out = run(
f'powershell -NoProfile -Command '
f'"(Get-NetIPAddress -InterfaceAlias \'{alias}\' -AddressFamily IPv4).IPAddress"'
)
ip = out.strip().splitlines()[-1].strip() if out.strip() else ""
if not re.match(r"^\d+\.\d+\.\d+\.\d+$", ip):
raise RuntimeError("Failed to determine physical adapter IP.")
self.phys_ip = ip
self.log(f"Physical adapter: {alias} ({ip})")
def pick_ports(self):
self.socks_port = find_free_port(9050, 9150, 9151, 9152)
self.dns_port = find_free_port(53, 9053, 9153)
self.control_port = find_free_port(9051, 9154)
self.log(f"Selected ports: SOCKS={self.socks_port}, DNS={self.dns_port}, Control={self.control_port}")
def write_torrc(self):
data_dir = BASE_DIR / "tordata"
data_dir.mkdir(parents=True, exist_ok=True)
torrc_path = BASE_DIR / "torrc"
bad_exits = self.exit_rating.get_blacklist()
exclude_nodes = ""
if bad_exits:
exclude_ips = ",".join(bad_exits[:20])
exclude_nodes = f"ExcludeExitNodes {{{exclude_ips}}}\n"
self.log(f"[torrc] Excluded bad exits: {len(bad_exits)} (in torrc: {len(bad_exits[:20])})")
content = f"""\
SocksPort 127.0.0.1:{self.socks_port}
DNSPort 127.0.0.1:{self.dns_port}
ControlPort 127.0.0.1:{self.control_port}
CookieAuthentication 1
OutboundBindAddress {self.phys_ip}
DataDirectory {data_dir}
AvoidDiskWrites 1
SafeLogging 0
Log info stdout
KeepalivePeriod 60
CircuitBuildTimeout 35
LearnCircuitBuildTimeout 0
MaxClientCircuitsPending 16
NumEntryGuards 6
NewCircuitPeriod 300
MaxCircuitDirtiness 1800
{exclude_nodes}"""
with open(torrc_path, "w", encoding="utf-8") as f:
f.write(content)
self.log("[connection] Direct connection mode.")
self.log(f"torrc written: {torrc_path}")
def start_tor(self):
if not self.tor_exe or not self.tor_exe.exists():
raise RuntimeError(f"tor.exe not found: {self.tor_exe}")
torrc = BASE_DIR / "torrc"
self.tor_proc = subprocess.Popen(
[str(self.tor_exe), "-f", str(torrc)],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="utf-8",
errors="replace",
creationflags=subprocess.CREATE_NO_WINDOW,
)
self.log("Tor started, waiting for bootstrap 100%...")
start = time.time()
while time.time() - start < 90:
line = self.tor_proc.stdout.readline()
if not line:
if self.tor_proc.poll() is not None:
raise RuntimeError("tor.exe exited before full startup.")
continue
self.log("[tor] " + line.strip())
if "Bootstrapped 100%" in line:
self.log("Tor fully bootstrapped.")
break
else:
raise RuntimeError("Tor did not bootstrap within 90 seconds.")
threading.Thread(target=self._read_tor_output, daemon=True).start()
self.log(f"Checking SOCKS on 127.0.0.1:{self.socks_port}...")
for attempt in range(10):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(2)
s.connect(("127.0.0.1", self.socks_port))
s.close()
self.log("SOCKS port responding.")
return
except Exception:
time.sleep(0.5)
raise RuntimeError("SOCKS port not responding after bootstrap.")
def _read_tor_output(self):
while self.tor_proc and self.tor_proc.poll() is None:
line = self.tor_proc.stdout.readline()
if not line:
break
self.log("[tor] " + line.rstrip())
def connect_control_port(self):
self.control_client = TorControlClient(self.control_port, self.log)
if self.control_client.connect():
self.log("[control] ControlPort connected and authenticated")
name, ip = self.control_client.get_current_exit()
if ip:
self._last_exit_ip = ip
self.log(f"[control] Current exit: {name} ({ip})")
score = self.exit_rating.get_score(ip)
self.log(f"[control] Exit rating {ip}: {score}")
if self.exit_rating.is_bad_exit(ip):
self.log("[control] Exit in blacklist! Requesting new circuit...")
self.control_client.newnym()
else:
self.log("[control] Could not determine current exit")
self.control_client.monitor_events(self._on_control_event)
else:
self.log("[control] Could not connect to ControlPort")
def _on_control_event(self, event: str):
if event.startswith("CIRC"):
if "BUILT" in event:
threading.Thread(target=self._update_exit_info, daemon=True).start()
def _update_exit_info(self):
if not self.control_client:
return
time.sleep(1)
name, ip = self.control_client.get_current_exit()
if ip and ip != self._last_exit_ip:
self._last_exit_ip = ip
self.log(f"[control] New exit: {name} ({ip})")
score = self.exit_rating.get_score(ip)
self.log(f"[control] Exit rating {ip}: {score}")
if self.exit_rating.is_bad_exit(ip):
self.log("[control] New exit in blacklist! Requesting change...")
self.control_client.newnym()
elif self.exit_rating.is_suspicious_exit(ip):
self.log("[control] New exit is suspicious (CAPTCHA).")
def check_exit_quality(self):
self.log("=== Checking exit quality ===")
sites = [
("http://check.torproject.org", "Tor Check"),
("http://neverssl.com", "NeverSSL"),
]
current_ip = None
for url, name in sites:
t0 = time.time()
try:
proxy_handler = urllib.request.ProxyHandler({
'http': f'socks5://127.0.0.1:{self.socks_port}',
'https': f'socks5://127.0.0.1:{self.socks_port}'
})
opener = urllib.request.build_opener(proxy_handler)
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with opener.open(req, timeout=15) as r:
elapsed = time.time() - t0
self.log(f" {name}: OK ({r.status}) in {elapsed:.2f}s")
if "check.torproject.org" in url:
html = r.read().decode('utf-8', errors='ignore')
ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', html)
if ip_match:
current_ip = ip_match.group(1)
except Exception as e:
self.log(f" {name}: FAIL ({e})")
if current_ip:
self.log(f"[check] Detected exit IP: {current_ip}")
score = self.exit_rating.get_score(current_ip)
self.log(f"[check] Rating: {score}")
if self.exit_rating.is_bad_exit(current_ip):
self.log("[check] Exit in blacklist! Requesting new circuit...")
if self.control_client:
self.control_client.newnym()
return False
elif self.exit_rating.is_suspicious_exit(current_ip):
self.log("[check] Exit is suspicious.")
return True
def request_new_identity(self, reason: str = "manual"):
self.log(f"[control] Requesting new circuit (reason: {reason})")
if self.control_client and self.control_client.connected:
self.control_client.newnym()
self._circuit_change_count += 1
else:
self.log("[control] ControlPort unavailable, cannot change circuit")
def wait_for_tun_adapter(self, timeout=30):
self.log("Waiting for tor0 adapter...")
start = time.time()
while time.time() - start < timeout:
rc, out = run(
'powershell -NoProfile -Command '
'"(Get-NetAdapter -Name tor0 -ErrorAction SilentlyContinue).Status"'
)
status = out.strip().splitlines()[-1].strip() if out.strip() else ""
if status == "Up":
self.log("Adapter tor0 is Up.")
break
time.sleep(0.5)
else:
raise RuntimeError(f"Adapter tor0 did not become Up within {timeout} seconds.")
rc, out = run(
'powershell -NoProfile -Command '
'"(Get-NetAdapter -Name tor0 -ErrorAction SilentlyContinue).ifIndex"'
)
try:
self.tun_ifindex = int(out.strip().splitlines()[-1].strip())
self.log(f"ifIndex tor0: {self.tun_ifindex}")
except Exception as e:
raise RuntimeError(f"Failed to get ifIndex for tor0: {e}")
def start_tun2socks(self):
if not self.tun2socks_exe or not self.tun2socks_exe.exists():
raise RuntimeError(f"tun2socks not found: {self.tun2socks_exe}")
if not self.wintun_dll or not self.wintun_dll.exists():
raise RuntimeError(f"wintun.dll not found: {self.wintun_dll}")
cmd = [
str(self.tun2socks_exe),
"-device", f"tun://{TUN_NAME}",
"-proxy", f"socks5://127.0.0.1:{self.socks_port}",
"-restapi", "127.0.0.1:60080",
"-tcp-auto-tuning",
"-tcp-rcvbuf", "2097152",
"-tcp-sndbuf", "2097152",
"-mtu", "1500",
"-udp-timeout", "60s",
"-loglevel", "warning",
]
self.tun2socks_proc = subprocess.Popen(
cmd,
cwd=str(self.tun2socks_exe.parent),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="utf-8",
errors="replace",
creationflags=subprocess.CREATE_NO_WINDOW,
)
self.log("tun2socks started.")
threading.Thread(target=self._read_tun2socks_log, daemon=True).start()
self.wait_for_tun_adapter()
def _read_tun2socks_log(self):
while self.tun2socks_proc:
line = self.tun2socks_proc.stdout.readline()
if not line:
break
self.log("[tun2socks] " + line.rstrip())
def configure_tun_adapter(self):
rc, out = run(
f'netsh interface ip set address "{TUN_NAME}" static {TUN_IP} {TUN_MASK}',
check=False,
)
self.log(f"[netsh set address] rc={rc} | {out.strip()}")
if rc != 0:
raise RuntimeError(f"Failed to set IP on {TUN_NAME}: {out}")
rc, out = run(f'netsh interface ipv4 set interface "{TUN_NAME}" metric=1', check=False)
self.log(f"[netsh set metric] rc={rc} | {out.strip()}")
if rc != 0:
raise RuntimeError(f"Failed to set metric on {TUN_NAME}: {out}")
self.log(f"Adapter {TUN_NAME} configured: {TUN_IP}/{TUN_MASK}")
def apply_routing(self):
rc, out = run(
f'powershell -NoProfile -Command '
f'"(Get-NetIPInterface -InterfaceAlias \'{self.phys_alias}\' -AddressFamily IPv4).InterfaceMetric"'
)
try:
self.phys_metric_saved = int(out.strip().splitlines()[-1])
except Exception:
self.phys_metric_saved = 25
self.log("Physical route preserved with high metric.")
rc, out = run(
f'powershell -NoProfile -Command '
f'"Set-NetIPInterface -InterfaceAlias \'{self.phys_alias}\' -InterfaceMetric 9999"',
check=False,
)
self.log(f"[Set-NetIPInterface metric] rc={rc} | {out.strip()}")
self.configure_tun_adapter()
rc, out = run(
'powershell -NoProfile -Command '
f'"New-NetRoute '
f'-DestinationPrefix 0.0.0.0/0 '
f'-InterfaceIndex {self.tun_ifindex} '
f'-NextHop 0.0.0.0 '
f'-RouteMetric 1 '
f'-ErrorAction Stop"'
)
self.log(f"[New-NetRoute] rc={rc} | {out.strip()}")
if rc != 0:
self.log(out)
raise RuntimeError(f"Failed to add default route: {out}")
time.sleep(1)
rc, out = run(
'powershell -NoProfile -Command '
f'"Remove-NetRoute -DestinationPrefix ::/0 -InterfaceAlias \'{self.phys_alias}\' '
f'-Confirm:$false -ErrorAction SilentlyContinue"',
check=False,
)
self.log(f"[Remove-NetRoute IPv6] rc={rc} | {out.strip()}")
rc, out = run(
f'powershell -NoProfile -Command '
f'"Set-DnsClientServerAddress -InterfaceAlias \'{self.phys_alias}\' '
f'-ResetServerAddresses -ErrorAction SilentlyContinue"',
check=False,
)
self.log(f"[Reset DNS phys] rc={rc} | {out.strip()}")
rc, out = run(f'netsh interface ip set dns "{TUN_NAME}" static 127.0.0.1 primary', check=False)
self.log(f"[netsh DNS tor0] rc={rc} | {out.strip()}")
if self.dns_port != 53:
self.log(f"DNS port {self.dns_port} != 53, starting UDP forwarder 127.0.0.1:53 -> {self.dns_port}")
self.dns_forwarder = DNSForwarder(53, self.dns_port)
self.dns_forwarder.start()
time.sleep(0.5)
else:
self.log("Tor listening on DNS port 53, forwarder not needed.")
self._verify_routing()
self.log("Routing and DNS switched to Tor.")
def _verify_routing(self):
self.log("Verifying routing table...")
rc, out = run(
'powershell -NoProfile -Command '
f'"Get-NetRoute -DestinationPrefix 0.0.0.0/0 | '
f'Where-Object {{$_.InterfaceIndex -eq {self.tun_ifindex}}}"'
)
self.log(out)
if rc != 0 or not out.strip():
raise RuntimeError("Default route via tor0 not found.")
self.log("Routing confirmed.")
def restore_routing(self):
if self.dns_forwarder:
self.dns_forwarder.stop()
self.dns_forwarder = None
self.log("DNS forwarder stopped.")
if self.tun_ifindex:
rc, out = run(
'powershell -NoProfile -Command '
f'"Remove-NetRoute '
f'-DestinationPrefix 0.0.0.0/0 '
f'-InterfaceIndex {self.tun_ifindex} '
f'-Confirm:$false -ErrorAction SilentlyContinue"',
check=False,
)
self.log(f"[restore Remove-NetRoute tor0] rc={rc} | {out.strip()}")
if self.phys_alias:
metric = self.phys_metric_saved or 25
rc, out = run(
f'powershell -NoProfile -Command '
f'"Set-NetIPInterface -InterfaceAlias \'{self.phys_alias}\' '
f'-InterfaceMetric {metric}"',
check=False,
)
self.log(f"[restore metric] rc={rc} | {out.strip()}")
rc, out = run(
f'powershell -NoProfile -Command '
f'"Set-DnsClientServerAddress -InterfaceAlias \'{self.phys_alias}\' '
f'-ResetServerAddresses -ErrorAction SilentlyContinue"',
check=False,
)
self.log(f"[restore DNS] rc={rc} | {out.strip()}")
self.log("Routing restored.")
def enable_killswitch(self):
run(f'netsh advfirewall firewall delete rule name="{FW_RULE_PREFIX}*"', check=False)
rules = [
f'netsh advfirewall firewall add rule name="{FW_RULE_PREFIX}AllowLoopback" dir=out action=allow remoteip=127.0.0.1',
f'netsh advfirewall firewall add rule name="{FW_RULE_PREFIX}AllowTor" dir=out action=allow program="{self.tor_exe}"',
f'netsh advfirewall firewall add rule name="{FW_RULE_PREFIX}AllowTun2socks" dir=out action=allow program="{self.tun2socks_exe}"',
f'netsh advfirewall firewall add rule name="{FW_RULE_PREFIX}AllowTunAdapter" dir=out action=allow interfacetype=any localip={TUN_IP}',
f'netsh advfirewall firewall add rule name="{FW_RULE_PREFIX}BlockIPv6" dir=out action=block protocol=41',
]
for r in rules:
run(r, check=False)
self.log("Killswitch enabled.")
def disable_killswitch(self):
run(f'netsh advfirewall firewall delete rule name="{FW_RULE_PREFIX}AllowLoopback"', check=False)
run(f'netsh advfirewall firewall delete rule name="{FW_RULE_PREFIX}AllowTor"', check=False)
run(f'netsh advfirewall firewall delete rule name="{FW_RULE_PREFIX}AllowTun2socks"', check=False)
run(f'netsh advfirewall firewall delete rule name="{FW_RULE_PREFIX}AllowTunAdapter"', check=False)
run(f'netsh advfirewall firewall delete rule name="{FW_RULE_PREFIX}BlockIPv6"', check=False)
self.log("Killswitch disabled.")
def _monitor_tor(self):
while self.running:
if self.tor_proc:
if self.tor_proc.poll() is None:
self.log("[monitor] TOR PROCESS OK")
else:
self.log(f"[monitor] !!! TOR EXIT CODE {self.tor_proc.returncode} !!!")
else:
self.log("[monitor] TOR PROC IS NONE")
try:
s = socket.create_connection(
("127.0.0.1", self.socks_port),
timeout=2
)
s.close()
self.log("[monitor] SOCKS LISTENING")
except Exception as e:
self.log(f"[monitor] SOCKS DEAD: {e}")
time.sleep(15)
def _monitor_http(self):
while self.running:
try:
urllib.request.urlopen("http://neverssl.com", timeout=10)
self.log("[monitor] HTTP OK")
except Exception as e:
self.log(f"[monitor] HTTP FAIL: {e}")
time.sleep(15)
def _monitor_loop(self):
self.log("[monitor] Monitoring started (every 15 sec)...")
while self.running:
time.sleep(15)
if not self.running:
break
if self.tun2socks_proc and self.tun2socks_proc.poll() is not None:
self.log("[monitor] tun2socks EXITED!")
continue
rc, out = run(
'powershell -NoProfile -Command '
'"(Get-NetAdapter -Name tor0 -ErrorAction SilentlyContinue).Status"'
)
status = out.strip().splitlines()[-1].strip() if out.strip() else "???"
if status != "Up":
self.log(f"[monitor] tor0 status: {status}")
rc, out = run(
'powershell -NoProfile -Command '
'"Get-NetRoute -DestinationPrefix 0.0.0.0/0 | '
'Select-Object InterfaceIndex, InterfaceAlias, NextHop, RouteMetric"'
)
self.log("[monitor] Routes 0.0.0.0/0:")
self.log(out.strip())
if self.control_client and self.control_client.connected:
name, ip = self.control_client.get_current_exit()
if ip and ip != self._last_exit_ip:
self._last_exit_ip = ip
self.log(f"[monitor] Exit changed: {name} ({ip})")
score = self.exit_rating.get_score(ip)
if self.exit_rating.is_bad_exit(ip):
self.log(f"[monitor] Exit {ip} in blacklist (score={score})")
def start(self):
self.tor_exe, self.tun2socks_exe, self.wintun_dll = self.dep.ensure_all()
self.detect_physical_adapter()
self.pick_ports()
self.write_torrc()
self.start_tor()
self.connect_control_port()
threading.Thread(target=self._monitor_tor, daemon=True).start()
threading.Thread(target=self._monitor_http, daemon=True).start()
self.start_tun2socks()
self.apply_routing()
self.check_exit_quality()
self.enable_killswitch()
self.running = True
threading.Thread(target=self._monitor_loop, daemon=True).start()
self.log(">>> All system traffic now goes through Tor. <<<")
def stop(self):
self.log("Stopping and rolling back...")
if self.control_client:
self.control_client.close()
self.control_client = None
self.disable_killswitch()
self.restore_routing()
if self.tun2socks_proc and self.tun2socks_proc.poll() is None:
self.tun2socks_proc.terminate()
if self.tor_proc and self.tor_proc.poll() is None:
self.tor_proc.terminate()
self.running = False
self.log("Stopped, system returned to normal network.")
def kill_all_tor_processes(self):
self.log("[cleanup] Terminating all Tor/PT processes...")
if self.tor_proc and self.tor_proc.pid:
try:
run(f'taskkill /PID {self.tor_proc.pid} /F', check=False)
except Exception:
pass
processes = ["tor.exe", "tun2socks-windows-amd64.exe", "tun2socks.exe"]
for proc in processes:
run(f'taskkill /F /IM {proc}', check=False)
time.sleep(1)
self.log("[cleanup] Processes terminated")
def kill_processes_on_ports(self, ports: list):
for port in ports:
if port and port > 0:
pids = get_processes_using_port(port)
kill_processes_by_pid(pids)
time.sleep(0.5)
def cleanup_tor_data(self):
self.log("[cleanup] Cleaning temporary files...")
pt_state_dir = BASE_DIR / "pt_state"
if pt_state_dir.exists():
try:
shutil.rmtree(pt_state_dir)
except Exception:
pass
cookie_path = BASE_DIR / "tordata" / "control_auth_cookie"
if cookie_path.exists():
try:
cookie_path.unlink()
except Exception:
pass
lock_path = BASE_DIR / "tordata" / "lock"
if lock_path.exists():
try:
lock_path.unlink()
except Exception:
pass
def full_reset(self):
self.log("\n" + "=" * 50)
self.log("FULL RESET...")
self.log("=" * 50)
self.stop()
self.kill_all_tor_processes()
ports = [self.socks_port, self.dns_port, self.control_port]
self.kill_processes_on_ports(ports)
self.cleanup_tor_data()
self.socks_port = None
self.dns_port = None
self.control_port = None
self.running = False
self.log("=" * 50)
self.log("[cleanup] Full reset complete")
self.log("=" * 50)
class App:
def __init__(self, root):
self.root = root
root.title("Tor System Router v9")
root.geometry("800x620")
self.dep_manager = DependencyManager(self.log, self.set_progress)
self.comp_manager = ComponentManager(self.log, self.set_progress)
self.router = TorRouter(self.log, self.dep_manager)
self.status_var = tk.StringVar(value="Stopped")
tk.Label(root, textvariable=self.status_var, font=("Segoe UI", 13, "bold")).pack(pady=6)
btn_frame = tk.Frame(root)
btn_frame.pack(pady=4)
self.start_btn = tk.Button(btn_frame, text="Start", width=14, command=self.on_start)
self.start_btn.grid(row=0, column=0, padx=5)
self.stop_btn = tk.Button(btn_frame, text="Stop", width=14, command=self.on_stop, state="disabled")
self.stop_btn.grid(row=0, column=1, padx=5)
self.check_btn = tk.Button(btn_frame, text="Check IP", width=14, command=self.on_check_ip)
self.check_btn.grid(row=0, column=2, padx=5)
self.newnym_btn = tk.Button(btn_frame, text="New Circuit", width=14, command=self.on_newnym)
self.newnym_btn.grid(row=0, column=3, padx=5)
self.reset_btn = tk.Button(btn_frame, text="Full Reset", width=14, command=self.on_full_reset)
self.reset_btn.grid(row=0, column=4, padx=5)
self.exit_frame = tk.LabelFrame(root, text="Current Exit Node", font=("Segoe UI", 9))
self.exit_frame.pack(fill="x", padx=10, pady=(0, 4))
update_frame = tk.Frame(root)
update_frame.pack(pady=(0, 4))
self.update_btn = tk.Button(
update_frame,
text="Check Updates",
width=30,
command=self.on_check_updates
)
self.update_btn.pack()
self.comp_frame = tk.LabelFrame(root, text="Components", font=("Segoe UI", 9))
self.comp_frame.pack(fill="x", padx=10, pady=(0, 4))
self.comp_vars = {}
self._init_comp_panel()
self.exit_info_var = tk.StringVar(value="Exit: — | IP: — | Rating: —")
tk.Label(self.exit_frame, textvariable=self.exit_info_var, font=("Consolas", 9)).pack(anchor="w", padx=5, pady=2)
self.progress = ttk.Progressbar(root, mode="determinate", maximum=100)
self.progress.pack(fill="x", padx=10, pady=(0, 4))
self.log_box = scrolledtext.ScrolledText(root, height=16, state="disabled", font=("Consolas", 9))
self.log_box.pack(fill="both", expand=True, padx=8, pady=8)
root.protocol("WM_DELETE_WINDOW", self.on_close)
self._update_exit_timer()
self._update_comp_display()
def _init_comp_panel(self):
for comp_name in ["Tor Expert Bundle", "tun2socks", "Wintun"]:
var = tk.StringVar(value=f"{comp_name}: checking...")
self.comp_vars[comp_name] = var
tk.Label(self.comp_frame, textvariable=var, font=("Consolas", 9)).pack(anchor="w", padx=5, pady=1)
def _update_comp_display(self):
for comp in self.comp_manager.components:
status = comp.status
if status.name in self.comp_vars:
if status.error:
text = f"{status.name}: ERROR — {status.error[:30]}"
elif status.needs_install:
text = f"{status.name}: NOT INSTALLED"
elif status.needs_update:
text = f"{status.name}: UPDATE AVAILABLE ({status.version} -> {status.latest_version})"
elif status.installed:
text = f"{status.name}: OK ({status.version})"
else:
text = f"{status.name}: unknown"
self.comp_vars[status.name].set(text)
def log(self, msg):
def _append():
self.log_box.configure(state="normal")
self.log_box.insert("end", msg + "\n")
lines = int(self.log_box.index("end-1c").split(".")[0])
if lines > 1000:
self.log_box.delete("1.0", "200.0")
self.log_box.see("end")
self.log_box.configure(state="disabled")
self.root.after(0, _append)
def set_progress(self, current, total):
if total <= 0:
return
pct = int(current / total * 100)
self.root.after(0, lambda: self.progress.configure(value=pct))
def _update_exit_timer(self):
if self.router.control_client and self.router.control_client.connected:
try:
name, ip = self.router.control_client.get_current_exit()
if ip:
score = self.router.exit_rating.get_score(ip)
captcha = self.router.exit_rating.get_captcha_count(ip)
status = "OK" if not self.router.exit_rating.is_bad_exit(ip) else "BLOCKED"
self.exit_info_var.set(
f"Exit: {name or '?'} | IP: {ip} | Rating: {score} | CAPTCHA: {captcha} | {status}"
)
else:
self.exit_info_var.set("Exit: detecting... | IP: — | Rating: —")
except Exception:
pass
self.root.after(5000, self._update_exit_timer)
def on_start(self):
self.start_btn.config(state="disabled")
self.status_var.set("Checking dependencies...")
threading.Thread(target=self._start_thread, daemon=True).start()
def _start_thread(self):
try:
self.router.start()
self.status_var.set("ACTIVE — traffic through Tor")
self.root.after(0, lambda: self.stop_btn.config(state="normal"))
except Exception as e:
self.log(f"ERROR: {e}")
self.status_var.set("Start error")
self.root.after(0, lambda: self.start_btn.config(state="normal"))
try:
self.router.stop()
except Exception:
pass
def on_stop(self):
self.stop_btn.config(state="disabled")
self.status_var.set("Stopping...")
threading.Thread(target=self._stop_thread, daemon=True).start()
def _stop_thread(self):
try:
self.router.stop()
except Exception as e:
self.log(f"Stop error: {e}")
self.status_var.set("Stopped")
self.root.after(0, lambda: self.start_btn.config(state="normal"))
self.exit_info_var.set("Exit: — | IP: — | Rating: —")
def on_check_ip(self):
threading.Thread(target=self._check_ip_thread, daemon=True).start()
def _check_ip_thread(self):
try:
self.log("Checking IP via HTTPS...")
proxy_handler = urllib.request.ProxyHandler({
'http': f'socks5://127.0.0.1:{self.router.socks_port}',
'https': f'socks5://127.0.0.1:{self.router.socks_port}'
})
opener = urllib.request.build_opener(proxy_handler)
req = urllib.request.Request(
"https://check.torproject.org/api/ip",
headers={"User-Agent": "Mozilla/5.0"},
)
with opener.open(req, timeout=30) as resp:
data = json.loads(resp.read().decode())
ip = data.get('IP')
is_tor = data.get('IsTor')
self.log(f"IP: {ip} | Through Tor: {is_tor}")
if ip:
if is_tor:
self.router.exit_rating.record(ip, success=True)
self.log(f"[rating] Exit {ip}: +10 (successful check)")
else:
self.router.exit_rating.record(ip, success=False, error=True)
self.log(f"[rating] Exit {ip}: -20 (not through Tor!)")
if self.router.exit_rating.is_bad_exit(ip):
self.log(f"[rating] Exit {ip} in blacklist!")
self.router.request_new_identity("bad_exit_detected")
except Exception as e:
self.log(f"Failed to check IP: {e}")
self.router.exit_rating.record("unknown", success=False, error=True)
def on_newnym(self):
if self.router.control_client and self.router.control_client.connected:
self.log("[gui] Manual new circuit request...")
self.router.request_new_identity("manual_gui")
else:
self.log("[gui] ControlPort not connected")
def on_full_reset(self):
if self.router.running:
if not messagebox.askyesno("Confirm", "Full reset will stop Tor and clean everything. Continue?"):
return
self.start_btn.config(state="disabled")
self.stop_btn.config(state="disabled")
self.status_var.set("Full reset...")
threading.Thread(target=self._reset_thread, daemon=True).start()
def _reset_thread(self):
try:
self.router.full_reset()
self.status_var.set("Ready to start")
self.root.after(0, lambda: self.start_btn.config(state="normal"))
except Exception as e:
self.log(f"Reset error: {e}")
self.status_var.set("Reset error")
self.root.after(0, lambda: self.start_btn.config(state="normal"))
finally:
self.root.after(0, lambda: self.stop_btn.config(state="disabled"))
def on_close(self):
if self.router.running:
if not messagebox.askyesno("Exit", "Tor routing is active. Stop and exit?"):
return
self.router.stop()
self.root.destroy()
def on_check_updates(self):
self.update_btn.config(state="disabled")
self.log("\n" + "=" * 50)
self.log("CHECKING UPDATES...")
self.log("=" * 50)
threading.Thread(target=self._update_thread, daemon=True).start()
def _update_thread(self):
try:
results = self.comp_manager.update_all()
self.root.after(0, self._update_comp_display)
all_ok = all(r.installed and not r.needs_update and not r.error for r in results)
if all_ok:
self.log("\n[ok] All components are up to date.")
else:
self.log("\n[Done] Update check complete.")
for r in results:
if r.error:
self.log(f" [x] {r.name}: {r.error}")
elif r.needs_install:
self.log(f" [x] {r.name}: installation failed")
elif r.needs_update:
self.log(f" [^] {r.name}: update available but not installed")
else:
self.log(f" [ok] {r.name}: {r.version}")
except Exception as e:
self.log(f"[x] Update check error: {e}")
finally:
self.root.after(0, lambda: self.update_btn.config(state="normal"))
def main():
if not is_admin():
relaunch_as_admin()
return
root = tk.Tk()
App(root)
root.mainloop()
if __name__ == "__main__":
main()
━━━━━━ 
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
!