first commit

This commit is contained in:
Александр Ебаклаков
2026-04-06 14:07:11 +03:00
commit 9a2f1387f9
13 changed files with 1758 additions and 0 deletions

5
.gitignore vendored Normal file
View File

@@ -0,0 +1,5 @@
bin/tun2socks*
bin/wintun.dll
__pycache__/
*.pyc
.DS_Store

129
README.md Normal file
View File

@@ -0,0 +1,129 @@
# 🌐 Proxy Switcher
**Кроссплатформенный прозрачный прокси** — направляет **100% трафика ОС** через заданный прокси-сервер с помощью TUN-интерфейса и [tun2socks](https://github.com/xjasonlyu/tun2socks).
> В отличие от системных настроек прокси (которые игнорируются играми, CLI-утилитами, фоновыми сервисами), proxy-switcher работает на уровне сетевого стека ОС — перехватывает **весь** трафик, включая TCP и UDP.
## ⚡ Быстрый старт
### 1. Скачайте tun2socks
```bash
# macOS (Apple Silicon)
curl -L -o tun2socks.zip https://github.com/xjasonlyu/tun2socks/releases/download/v2.6.0/tun2socks-darwin-arm64.zip
unzip tun2socks.zip && mv tun2socks-darwin-arm64 bin/ && rm tun2socks.zip
# macOS (Intel)
curl -L -o tun2socks.zip https://github.com/xjasonlyu/tun2socks/releases/download/v2.6.0/tun2socks-darwin-amd64.zip
unzip tun2socks.zip && mv tun2socks-darwin-amd64 bin/ && rm tun2socks.zip
# Windows — скачайте архив и wintun.dll:
# https://github.com/xjasonlyu/tun2socks/releases/download/v2.6.0/tun2socks-windows-amd64.zip
# Извлеките tun2socks-windows-amd64.exe в папку bin/
# https://www.wintun.net/ → извлеките wintun.dll в папку bin/
```
### 2. Запустите
```bash
# macOS / Linux (требует sudo)
sudo python -m src.main --proxy socks5://your-proxy-server:1080
# С авторизацией
sudo python -m src.main --proxy socks5://user:password@proxy:1080
# HTTP прокси
sudo python -m src.main --proxy http://proxy:8080
# Тестовый режим (показывает команды, не выполняет)
sudo python -m src.main --proxy socks5://proxy:1080 --dry-run
```
### 3. Отключение
Нажмите `Ctrl+C` — все маршруты автоматически восстановятся.
## 🏗 Архитектура
```
┌──────────────────────────────────────────────────────┐
│ Python Manager (main.py) │
│ ┌─────────┐ ┌──────────────┐ ┌────────────────┐ │
│ │ Privile-│ │ Network │ │ Platform │ │
│ │ ges │ │ Utils │ │ Router (ABC) │ │
│ │ check │ │ (gateway │ │ ┌────────────┐ │ │
│ │ │ │ detection) │ │ │macOS Router│ │ │
│ └─────────┘ └──────────────┘ │ │Win Router │ │ │
│ │ └────────────┘ │ │
│ ┌──────────────────────────┐ └────────────────┘ │
│ │ Tun2socks Manager │ │
│ │ (subprocess lifecycle) │ │
│ └────────────┬─────────────┘ │
└───────────────┼──────────────────────────────────────┘
│ subprocess
┌───────────▼───────────┐
│ tun2socks binary │
│ (Go + gVisor stack) │
│ ┌─────────────────┐ │
│ │ TUN Interface │◄─── All OS traffic
│ │ (utun / wintun) │ │
│ └────────┬────────┘ │
│ │ IP→SOCKS │
│ ┌────────▼────────┐ │
│ │ Proxy Server │──── Internet
│ └─────────────────┘ │
└───────────────────────┘
```
## 📋 CLI-параметры
| Параметр | Описание | По умолчанию |
|---|---|---|
| `--proxy`, `-p` | URL прокси (обязательный) | — |
| `--tun-name` | Имя TUN-интерфейса | `utun123` (macOS) / `wintun` (Windows) |
| `--tun-ip` | IP-адрес TUN | `198.18.0.1` (macOS) / `192.168.123.1` (Win) |
| `--dns` | DNS-серверы | `8.8.8.8 8.8.4.4` |
| `--interface`, `-i` | Основной интерфейс ОС | auto-detect |
| `--log-level`, `-l` | Уровень логов | `info` |
| `--dry-run` | Только показать команды | `false` |
## 🔒 Требования
- **Python** 3.8+
- **Права root/Administrator** (создание TUN + маршруты)
- **tun2socks** binary в `bin/` или PATH
- **Windows**: дополнительно `wintun.dll`
## 📁 Структура проекта
```
proxy-switcher/
├── bin/ # Бинарники tun2socks (скачиваются)
├── src/
│ ├── __init__.py
│ ├── main.py # CLI + оркестратор
│ ├── config.py # Конфигурация (dataclass)
│ ├── privileges.py # Проверка прав admin/root
│ ├── network_utils.py # Детекция шлюза + DNS resolve
│ ├── platform_router.py # Абстракция маршрутизации
│ ├── macos_router.py # macOS: ifconfig + route
│ ├── windows_router.py # Windows: netsh
│ └── tun2socks_manager.py # Управление процессом tun2socks
├── requirements.txt
└── README.md
```
## ⚠ Важные ограничения
1. **Прокси не на localhost**: Прокси-сервер должен быть на удалённом хосте. Если прокси на `127.0.0.1`, возникнет routing loop.
2. **Один экземпляр**: Не запускайте несколько экземпляров одновременно.
3. **Graceful shutdown**: Всегда используйте `Ctrl+C` для корректного отключения. При аварийном завершении маршруты могут остаться в ОС.
## 📜 Лицензия
MIT
## 📝 TODO
- [ ] Добавить отдельные CLI флаги `--user` и `--password` для прокси с авторизацией (сейчас поддерживается через передачу параметров внутри `--proxy URL`).
- [ ] Добавить скрипт или команду для автоматического скачивания нужного бинарника `tun2socks` под текущую ОС и архитектуру.

8
bin/.gitkeep Normal file
View File

@@ -0,0 +1,8 @@
# Place tun2socks binaries here.
# Download from: https://github.com/xjasonlyu/tun2socks/releases
#
# Expected files:
# tun2socks-darwin-arm64 (macOS Apple Silicon)
# tun2socks-darwin-amd64 (macOS Intel)
# tun2socks-windows-amd64.exe (Windows)
# wintun.dll (Windows only — from https://www.wintun.net/)

20
requirements.txt Normal file
View File

@@ -0,0 +1,20 @@
# proxy-switcher — dependencies
#
# Ядро приложения НЕ ТРЕБУЕТ внешних pip-зависимостей.
# Все модули используют только стандартную библиотеку Python 3.8+:
# - subprocess, os, sys, signal, atexit, platform, shutil
# - dataclasses, abc, logging, argparse, urllib.parse, socket, time
# - ctypes (для Windows admin check)
# - pathlib, re, typing
#
# Единственная внешняя зависимость — бинарник tun2socks (Go),
# который скачивается отдельно в папку bin/.
#
# Ниже — опциональные зависимости для расширений:
#
# --- Опционально: GUI (системный трей) ---
# pystray>=0.19 # Cross-platform system tray icon
# Pillow>=10.0 # Required by pystray for icon rendering
#
# --- Опционально: Авто-скачивание бинарников ---
# httpx>=0.27 # HTTP client for downloading tun2socks releases

1
src/__init__.py Normal file
View File

@@ -0,0 +1 @@
# proxy-switcher — Cross-platform TUN-based proxy routing manager

164
src/config.py Normal file
View File

@@ -0,0 +1,164 @@
"""
config.py — Модель конфигурации для proxy-switcher.
Содержит dataclass ProxyConfig с валидацией параметров:
- proxy URL (socks5://host:port, http://host:port)
- TUN interface parameters (name, IP, mask)
- DNS servers для TUN-интерфейса
- Уровень логирования tun2socks
"""
from __future__ import annotations
import platform
import re
import socket
from dataclasses import dataclass, field
from typing import List, Optional
from urllib.parse import urlparse
# ==============================================================================
# Константы по умолчанию
# ==============================================================================
# Подсеть 198.18.0.0/15 (RFC 5737 — зарезервирована для бенчмарков, безопасна
# как приватный адрес TUN-интерфейса, не пересекается с реальными сетями)
DEFAULT_TUN_IP = "198.18.0.1"
DEFAULT_TUN_MASK = "255.254.0.0" # /15
# Для Windows tun2socks использует wintun driver с несколько другой схемой адресации
DEFAULT_TUN_IP_WIN = "192.168.123.1"
DEFAULT_TUN_MASK_WIN = "255.255.255.0" # /24
# DNS-серверы, которые будут использоваться через TUN
DEFAULT_DNS_SERVERS = ["8.8.8.8", "8.8.4.4"]
# Поддерживаемые схемы прокси
SUPPORTED_PROXY_SCHEMES = {"socks5", "socks4", "http", "https", "ss", "ssh", "relay"}
@dataclass
class ProxyConfig:
"""Полная конфигурация для proxy-switcher.
Attributes:
proxy_url: URL прокси-сервера (e.g., socks5://user:pass@1.2.3.4:1080)
tun_name: Имя TUN-интерфейса (автоматически по платформе)
tun_ip: IP-адрес TUN-интерфейса
tun_mask: Маска подсети TUN
dns_servers: Список DNS-серверов для TUN-интерфейса
log_level: Уровень логирования tun2socks (debug/info/warn/error)
interface: Имя основного сетевого интерфейса (auto-detect если None)
dry_run: Если True — только печатает команды, не выполняет
"""
proxy_url: str
tun_name: Optional[str] = None
tun_ip: Optional[str] = None
tun_mask: Optional[str] = None
dns_servers: List[str] = field(default_factory=lambda: list(DEFAULT_DNS_SERVERS))
log_level: str = "info"
interface: Optional[str] = None
dry_run: bool = False
def __post_init__(self) -> None:
"""Валидация и автозаполнение полей после инициализации."""
self._validate_proxy_url()
self._set_platform_defaults()
# --------------------------------------------------------------------------
# Валидация
# --------------------------------------------------------------------------
def _validate_proxy_url(self) -> None:
"""Проверяет корректность proxy_url."""
parsed = urlparse(self.proxy_url)
# Проверяем схему
scheme = parsed.scheme.lower()
if scheme not in SUPPORTED_PROXY_SCHEMES:
raise ValueError(
f"Unsupported proxy scheme: '{scheme}'. "
f"Supported: {', '.join(sorted(SUPPORTED_PROXY_SCHEMES))}"
)
# Проверяем наличие хоста
if not parsed.hostname:
raise ValueError(f"Proxy URL must contain a hostname: {self.proxy_url}")
# Проверяем наличие порта
if not parsed.port:
raise ValueError(f"Proxy URL must contain a port: {self.proxy_url}")
# Предупреждение о localhost — routing loop!
if parsed.hostname in ("127.0.0.1", "localhost", "::1", "0.0.0.0"):
raise ValueError(
f"⚠ Proxy on localhost ({parsed.hostname}) will cause a "
f"routing loop! The proxy server must be on a remote host."
)
def _set_platform_defaults(self) -> None:
"""Устанавливает значения по умолчанию в зависимости от платформы."""
system = platform.system()
if system == "Darwin":
if self.tun_name is None:
self.tun_name = "utun123"
if self.tun_ip is None:
self.tun_ip = DEFAULT_TUN_IP
if self.tun_mask is None:
self.tun_mask = DEFAULT_TUN_MASK
elif system == "Windows":
if self.tun_name is None:
self.tun_name = "wintun"
if self.tun_ip is None:
self.tun_ip = DEFAULT_TUN_IP_WIN
if self.tun_mask is None:
self.tun_mask = DEFAULT_TUN_MASK_WIN
else:
# Linux / другие — используем macOS-подобные значения
if self.tun_name is None:
self.tun_name = "tun0"
if self.tun_ip is None:
self.tun_ip = DEFAULT_TUN_IP
if self.tun_mask is None:
self.tun_mask = DEFAULT_TUN_MASK
# --------------------------------------------------------------------------
# Утилиты
# --------------------------------------------------------------------------
@property
def proxy_host(self) -> str:
"""Извлекает hostname из proxy URL."""
return urlparse(self.proxy_url).hostname or ""
@property
def proxy_port(self) -> int:
"""Извлекает port из proxy URL."""
return urlparse(self.proxy_url).port or 0
@property
def proxy_ip(self) -> str:
"""Резолвит hostname прокси в IP-адрес.
Необходимо для добавления маршрута-bypass (чтобы трафик к самому
прокси шёл напрямую, а не через TUN → бесконечная петля).
"""
host = self.proxy_host
try:
# Если уже IP — вернётся как есть
socket.inet_aton(host)
return host
except socket.error:
# DNS resolve
return socket.gethostbyname(host)
@property
def is_macos(self) -> bool:
return platform.system() == "Darwin"
@property
def is_windows(self) -> bool:
return platform.system() == "Windows"

156
src/macos_router.py Normal file
View File

@@ -0,0 +1,156 @@
"""
macos_router.py — Реализация маршрутизации для macOS (Darwin).
Особенности macOS:
1. tun2socks сам создаёт utun-интерфейс при запуске (не нужно создавать вручную)
2. IP назначается через ifconfig: `ifconfig utun123 198.18.0.1 198.18.0.1 up`
3. Маршруты добавляются через route: `route add -net X.0.0.0/Y 198.18.0.1`
4. macOS не поддерживает `route add default` для TUN — используется split-routing
(набор подсетей, полностью покрывающих 0.0.0.0/0)
5. Для предотвращения routing loop — отдельный маршрут для IP прокси-сервера
через оригинальный gateway
Split-routing подсети (покрывают всё адресное пространство IPv4):
1.0.0.0/8 — покрывает 1.x.x.x
2.0.0.0/7 — покрывает 2.x.x.x 3.x.x.x
4.0.0.0/6 — покрывает 4.x.x.x 7.x.x.x
8.0.0.0/5 — покрывает 8.x.x.x 15.x.x.x
16.0.0.0/4 — покрывает 16.x.x.x 31.x.x.x
32.0.0.0/3 — покрывает 32.x.x.x 63.x.x.x
64.0.0.0/2 — покрывает 64.x.x.x 127.x.x.x
128.0.0.0/1 — покрывает 128.x.x.x 255.x.x.x
198.18.0.0/15 — подсеть самого TUN-интерфейса
"""
from __future__ import annotations
import logging
from .platform_router import PlatformRouter
logger = logging.getLogger(__name__)
# Подсети для split-routing (покрывают 0.0.0.0/0 без замены default route)
SPLIT_ROUTES = [
"1.0.0.0/8",
"2.0.0.0/7",
"4.0.0.0/6",
"8.0.0.0/5",
"16.0.0.0/4",
"32.0.0.0/3",
"64.0.0.0/2",
"128.0.0.0/1",
"198.18.0.0/15",
]
class MacOSRouter(PlatformRouter):
"""Реализация маршрутизации для macOS.
Полный цикл:
1. setup_interface() → ifconfig utun123 <ip> <ip> up
2. add_proxy_bypass_route() → route add -host <proxy_ip> <orig_gateway>
3. add_routes() → route add -net <subnet> <tun_ip> (×9)
4. cleanup() → route delete ... (в обратном порядке)
"""
def setup_interface(self) -> None:
"""Назначает IP-адрес utun-интерфейсу и поднимает его.
На macOS TUN — point-to-point интерфейс, поэтому задаётся
и локальный, и удалённый адрес (оба одинаковые для TUN).
Команда: sudo ifconfig utun123 198.18.0.1 198.18.0.1 up
"""
tun = self.config.tun_name
ip = self.config.tun_ip
logger.info(f"⚙ Configuring interface {tun} with IP {ip}")
self._run_cmd([
"sudo", "ifconfig", tun,
ip, # local address
ip, # destination address (point-to-point)
"up",
])
self._interface_configured = True
logger.info(f"✓ Interface {tun} is UP with IP {ip}")
def add_proxy_bypass_route(self) -> None:
"""Добавляет маршрут-исключение: трафик к прокси → оригинальный шлюз.
Без этого маршрута пакеты к прокси-серверу пойдут через TUN →
tun2socks → прокси → TUN → ... (бесконечная петля).
Команда: sudo route add -host <proxy_ip> <original_gateway>
"""
logger.info(
f"⚙ Adding bypass route: {self.proxy_ip}{self.original_gateway}"
)
self._run_cmd([
"sudo", "route", "add",
"-host", self.proxy_ip,
self.original_gateway,
])
logger.info(f"✓ Proxy bypass route added for {self.proxy_ip}")
def add_routes(self) -> None:
"""Добавляет split-routing маршруты через TUN-интерфейс.
Вместо изменения default route (что на macOS проблематично для TUN),
добавляем набор более специфичных подсетей, которые суммарно
покрывают всё адресное пространство IPv4.
Более специфичные маршруты имеют приоритет над менее специфичными
(longest prefix match), поэтому:
- Трафик к прокси → идёт через -host маршрут (самый специфичный)
- Весь остальной → идёт через наши подсети → TUN → прокси
"""
tun_ip = self.config.tun_ip
logger.info(f"⚙ Adding {len(SPLIT_ROUTES)} split routes via {tun_ip}")
for subnet in SPLIT_ROUTES:
self._run_cmd([
"sudo", "route", "add",
"-net", subnet,
tun_ip,
])
logger.debug(f" + route {subnet}{tun_ip}")
self._routes_added = True
logger.info(f"✓ All {len(SPLIT_ROUTES)} routes added through TUN")
def cleanup(self) -> None:
"""Удаляет все добавленные маршруты (graceful shutdown).
Удаляет маршруты в обратном порядке, игнорируя ошибки
(маршрут может уже не существовать, если tun2socks упал раньше).
Идемпотентный — безопасно вызывать повторно.
"""
logger.info("🧹 Starting route cleanup...")
errors = 0
# 1. Удаляем split-routing маршруты
if self._routes_added:
for subnet in reversed(SPLIT_ROUTES):
if not self._run_cmd_silent([
"sudo", "route", "delete", "-net", subnet,
]):
errors += 1
self._routes_added = False
# 2. Удаляем bypass-маршрут для прокси
self._run_cmd_silent([
"sudo", "route", "delete", "-host", self.proxy_ip,
])
if errors:
logger.warning(f"{errors} route(s) failed to delete (may already be gone)")
else:
logger.info("✓ All routes cleaned up successfully")

343
src/main.py Normal file
View File

@@ -0,0 +1,343 @@
#!/usr/bin/env python3
"""
main.py — Точка входа и оркестратор proxy-switcher.
Управляет полным жизненным циклом:
1. Парсинг CLI-аргументов
2. Проверка / получение прав администратора
3. Определение оригинального шлюза и сетевого интерфейса
4. Resolve IP-адреса прокси-сервера
5. Запуск tun2socks (создание TUN-интерфейса)
6. Настройка TUN-интерфейса (IP, DNS)
7. Добавление маршрута-bypass для прокси
8. Добавление маршрутов через TUN (перехват всего трафика)
9. Ожидание Ctrl+C / SIGTERM
10. Graceful shutdown (cleanup маршрутов, остановка tun2socks)
Использование:
sudo python -m src.main --proxy socks5://proxy-host:1080
sudo python -m src.main --proxy http://user:pass@proxy:8080 --log-level debug
sudo python -m src.main --proxy socks5://proxy:1080 --dry-run
"""
from __future__ import annotations
import argparse
import atexit
import logging
import platform
import signal
import sys
import time
from typing import Optional
from .config import ProxyConfig, SUPPORTED_PROXY_SCHEMES
from .network_utils import get_default_gateway
from .platform_router import PlatformRouter
from .privileges import ensure_admin
from .tun2socks_manager import Tun2socksManager
# ==============================================================================
# Логирование
# ==============================================================================
LOG_FORMAT = "%(asctime)s [%(levelname)-5s] %(name)s: %(message)s"
LOG_DATE_FORMAT = "%H:%M:%S"
def setup_logging(level: str) -> None:
"""Настраивает логирование для всего приложения."""
numeric_level = getattr(logging, level.upper(), logging.INFO)
logging.basicConfig(
level=numeric_level,
format=LOG_FORMAT,
datefmt=LOG_DATE_FORMAT,
)
logger = logging.getLogger(__name__)
# ==============================================================================
# CLI-аргументы
# ==============================================================================
def parse_args() -> argparse.Namespace:
"""Парсит аргументы командной строки."""
parser = argparse.ArgumentParser(
prog="proxy-switcher",
description=(
"🌐 Cross-platform transparent proxy — routes ALL OS traffic "
"through a proxy server via TUN interface + tun2socks."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"Examples:\n"
" sudo python -m src.main --proxy socks5://1.2.3.4:1080\n"
" sudo python -m src.main --proxy http://user:pass@proxy:8080\n"
" sudo python -m src.main --proxy socks5://proxy:1080 --dry-run\n"
"\n"
f"Supported proxy schemes: {', '.join(sorted(SUPPORTED_PROXY_SCHEMES))}\n"
"\n"
"⚠ Requires root/administrator privileges.\n"
"⚠ Proxy must NOT be on localhost (routing loop)."
),
)
parser.add_argument(
"--proxy", "-p",
required=True,
help=(
"Proxy server URL. "
"Format: scheme://[user:pass@]host:port "
"(e.g., socks5://1.2.3.4:1080)"
),
)
parser.add_argument(
"--tun-name",
default=None,
help="TUN interface name (default: utun123 on macOS, wintun on Windows)",
)
parser.add_argument(
"--tun-ip",
default=None,
help="IP address for TUN interface (default: 198.18.0.1)",
)
parser.add_argument(
"--dns",
nargs="+",
default=None,
help="DNS servers for TUN (default: 8.8.8.8 8.8.4.4)",
)
parser.add_argument(
"--interface", "-i",
default=None,
help=(
"Primary network interface name (auto-detected if not set). "
"e.g., en0 on macOS, Wi-Fi on Windows"
),
)
parser.add_argument(
"--log-level", "-l",
default="info",
choices=["debug", "info", "warn", "error"],
help="Log level (default: info)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Print commands without executing them (for testing)",
)
return parser.parse_args()
# ==============================================================================
# Создание платформенного роутера
# ==============================================================================
def create_router(
config: ProxyConfig,
gateway: str,
iface: str,
proxy_ip: str,
) -> PlatformRouter:
"""Создаёт подходящую реализацию PlatformRouter для текущей ОС.
Returns:
MacOSRouter или WindowsRouter.
Raises:
RuntimeError: Если платформа не поддерживается.
"""
system = platform.system()
if system == "Darwin":
from .macos_router import MacOSRouter
return MacOSRouter(config, gateway, iface, proxy_ip)
elif system == "Windows":
from .windows_router import WindowsRouter
return WindowsRouter(config, gateway, iface, proxy_ip)
else:
raise RuntimeError(
f"Platform '{system}' is not yet supported. "
f"Currently supported: macOS (Darwin), Windows."
)
# ==============================================================================
# Главный оркестратор
# ==============================================================================
def main() -> int:
"""Главная функция — полный жизненный цикл proxy-switcher.
Returns:
Exit code (0 = success, 1 = error).
"""
args = parse_args()
setup_logging(args.log_level)
# Баннер
print(
"\n"
" ╔═══════════════════════════════════════════╗\n"
" ║ 🌐 Proxy Switcher — TUN-based routing ║\n"
" ║ Ctrl+C to disconnect gracefully ║\n"
" ╚═══════════════════════════════════════════╝\n"
)
# ── Шаг 1: Проверка прав ──────────────────────────────────────────────
logger.info("Step 1/7: Checking privileges...")
if args.dry_run:
logger.info("⏭ Skipping privilege check in dry-run mode")
else:
ensure_admin()
# ── Шаг 2: Создание конфигурации ──────────────────────────────────────
logger.info("Step 2/7: Building configuration...")
try:
config = ProxyConfig(
proxy_url=args.proxy,
tun_name=args.tun_name,
tun_ip=args.tun_ip,
dns_servers=args.dns,
log_level=args.log_level,
interface=args.interface,
dry_run=args.dry_run,
)
except ValueError as e:
logger.error(f"Configuration error: {e}")
return 1
if config.dry_run:
logger.info("🔍 DRY-RUN mode: commands will be printed but NOT executed")
# ── Шаг 3: Определение оригинального шлюза ────────────────────────────
logger.info("Step 3/7: Detecting default gateway...")
try:
orig_gateway, orig_iface = get_default_gateway()
except RuntimeError as e:
logger.error(f"Failed to detect gateway: {e}")
return 1
# Если пользователь указал интерфейс — используем его
if config.interface:
orig_iface = config.interface
logger.info(f"Using user-specified interface: {orig_iface}")
logger.info(f"Original gateway: {orig_gateway} via {orig_iface}")
# ── Шаг 4: Resolve IP прокси-сервера ──────────────────────────────────
logger.info("Step 4/7: Resolving proxy IP...")
try:
proxy_ip = config.proxy_ip
except RuntimeError as e:
logger.error(f"Failed to resolve proxy: {e}")
return 1
logger.info(f"Proxy IP: {proxy_ip} ({config.proxy_host}:{config.proxy_port})")
# ── Шаг 5-7: Запуск с гарантированным cleanup ────────────────────────
tun2socks = Tun2socksManager(config)
router = create_router(config, orig_gateway, orig_iface, proxy_ip)
# Регистрируем cleanup на atexit (страховка)
def cleanup() -> None:
"""Гарантированная очистка при любом завершении."""
logger.info("🔄 Running cleanup...")
try:
router.cleanup()
except Exception as e:
logger.error(f"Router cleanup error: {e}")
try:
tun2socks.stop()
except Exception as e:
logger.error(f"tun2socks stop error: {e}")
atexit.register(cleanup)
# Обработка SIGTERM (kill / systemd stop)
def signal_handler(signum: int, frame) -> None:
sig_name = signal.Signals(signum).name
logger.info(f"Received {sig_name}, shutting down...")
cleanup()
sys.exit(0)
signal.signal(signal.SIGTERM, signal_handler)
try:
# ── Шаг 5: Запуск tun2socks ──────────────────────────────────────
logger.info("Step 5/7: Starting tun2socks...")
tun2socks.start(network_interface=orig_iface)
# ── Шаг 6: Настройка TUN-интерфейса ──────────────────────────────
logger.info("Step 6/7: Configuring TUN interface...")
router.setup_interface()
# ── Шаг 7: Настройка маршрутизации ────────────────────────────────
logger.info("Step 7/7: Setting up routing...")
router.add_proxy_bypass_route()
router.add_routes()
# ── Готово! ───────────────────────────────────────────────────────
print(
"\n"
" ╔═══════════════════════════════════════════╗\n"
" ║ ✅ ALL TRAFFIC NOW ROUTED VIA PROXY ║\n"
f" ║ Proxy: {config.proxy_url:<33}\n"
f" ║ TUN: {config.tun_name:<33}\n"
f" ║ TUN IP: {config.tun_ip:<32}\n"
" ║ ║\n"
" ║ Press Ctrl+C to disconnect ║\n"
" ╚═══════════════════════════════════════════╝\n"
)
# Блокируемся до Ctrl+C или завершения tun2socks
while tun2socks.is_running():
time.sleep(1)
# Если tun2socks сам завершился — это ошибка
logger.warning("⚠ tun2socks process exited unexpectedly!")
return 1
except KeyboardInterrupt:
print("\n")
logger.info("🛑 Ctrl+C received, disconnecting...")
return 0
except Exception as e:
logger.error(f"❌ Fatal error: {e}")
return 1
finally:
# Cleanup ВСЕГДА выполняется (даже при exception)
cleanup()
# Снимаем atexit чтобы не вызывать cleanup дважды
atexit.unregister(cleanup)
# ==============================================================================
# Entry point
# ==============================================================================
if __name__ == "__main__":
sys.exit(main())

208
src/network_utils.py Normal file
View File

@@ -0,0 +1,208 @@
"""
network_utils.py — Кроссплатформенные утилиты для работы с сетью.
Функции:
- get_default_gateway() → (gateway_ip, interface_name)
- resolve_host(hostname) → ip_address
"""
from __future__ import annotations
import logging
import platform
import re
import socket
import subprocess
from typing import Tuple
logger = logging.getLogger(__name__)
def get_default_gateway() -> Tuple[str, str]:
"""Определяет текущий шлюз по умолчанию и основной сетевой интерфейс.
Returns:
Кортеж (gateway_ip, interface_name).
Например: ("192.168.1.1", "en0") на macOS
("192.168.1.1", "Wi-Fi") на Windows
Raises:
RuntimeError: Если не удалось определить шлюз.
"""
system = platform.system()
if system == "Darwin":
return _get_gateway_macos()
elif system == "Windows":
return _get_gateway_windows()
elif system == "Linux":
return _get_gateway_linux()
else:
raise RuntimeError(f"Unsupported platform: {system}")
def _get_gateway_macos() -> Tuple[str, str]:
"""Определяет шлюз на macOS через `route -n get default`.
Пример вывода:
route to: default
destination: default
gateway: 192.168.1.1
interface: en0
flags: <UP,GATEWAY,DONE,STATIC,PRCLONING,AUTOCONF>
"""
try:
result = subprocess.run(
["route", "-n", "get", "default"],
capture_output=True,
text=True,
timeout=10,
)
output = result.stdout
gateway = ""
interface = ""
for line in output.splitlines():
line = line.strip()
if line.startswith("gateway:"):
gateway = line.split(":", 1)[1].strip()
elif line.startswith("interface:"):
interface = line.split(":", 1)[1].strip()
if not gateway or not interface:
raise RuntimeError(
f"Could not parse gateway/interface from route output:\n{output}"
)
logger.info(f"Detected default gateway: {gateway} via {interface}")
return gateway, interface
except subprocess.TimeoutExpired:
raise RuntimeError("Timeout while detecting default gateway")
except FileNotFoundError:
raise RuntimeError("'route' command not found")
def _get_gateway_windows() -> Tuple[str, str]:
"""Определяет шлюз на Windows через `netsh interface ipv4 show route`.
Ищет маршрут 0.0.0.0/0 с минимальным metric.
Альтернативно парсит `ipconfig` для получения имени интерфейса.
"""
try:
# Получаем маршрут по умолчанию
result = subprocess.run(
[
"powershell", "-Command",
"(Get-NetRoute -DestinationPrefix '0.0.0.0/0' | "
"Sort-Object RouteMetric | Select-Object -First 1 | "
"Format-List NextHop, InterfaceAlias)"
],
capture_output=True,
text=True,
timeout=15,
)
output = result.stdout
gateway = ""
interface = ""
for line in output.splitlines():
line = line.strip()
if line.startswith("NextHop"):
gateway = line.split(":", 1)[1].strip()
elif line.startswith("InterfaceAlias"):
interface = line.split(":", 1)[1].strip()
if not gateway or not interface:
# Fallback: попробуем через netsh
return _get_gateway_windows_netsh()
logger.info(f"Detected default gateway: {gateway} via {interface}")
return gateway, interface
except (subprocess.TimeoutExpired, FileNotFoundError):
return _get_gateway_windows_netsh()
def _get_gateway_windows_netsh() -> Tuple[str, str]:
"""Fallback-метод определения шлюза через netsh на Windows."""
try:
result = subprocess.run(
["netsh", "interface", "ipv4", "show", "route"],
capture_output=True,
text=True,
timeout=10,
)
# Ищем строку с 0.0.0.0/0
for line in result.stdout.splitlines():
if "0.0.0.0/0" in line:
parts = line.split()
# Типичный формат: Prefix NextHop Metric IfIndex InterfaceAlias
# Парсим по позиции
for i, part in enumerate(parts):
if "0.0.0.0/0" in part:
if i + 1 < len(parts):
gateway = parts[i + 1]
interface = parts[-1] if len(parts) > i + 3 else "unknown"
return gateway, interface
raise RuntimeError("Could not find default route in netsh output")
except Exception as e:
raise RuntimeError(f"Failed to detect default gateway on Windows: {e}")
def _get_gateway_linux() -> Tuple[str, str]:
"""Определяет шлюз на Linux через `ip route show default`."""
try:
result = subprocess.run(
["ip", "route", "show", "default"],
capture_output=True,
text=True,
timeout=10,
)
# Формат: default via 192.168.1.1 dev eth0 ...
match = re.search(
r"default\s+via\s+(\S+)\s+dev\s+(\S+)", result.stdout
)
if match:
gateway = match.group(1)
interface = match.group(2)
logger.info(f"Detected default gateway: {gateway} via {interface}")
return gateway, interface
raise RuntimeError(
f"Could not parse default route from: {result.stdout}"
)
except FileNotFoundError:
raise RuntimeError("'ip' command not found")
def resolve_host(hostname: str) -> str:
"""Резолвит hostname в IP-адрес.
Если передан уже IP — возвращает его без изменений.
Args:
hostname: DNS-имя или IP-адрес.
Returns:
IPv4-адрес в виде строки.
"""
try:
socket.inet_aton(hostname)
return hostname # Уже IP
except socket.error:
pass
try:
ip = socket.gethostbyname(hostname)
logger.info(f"Resolved {hostname}{ip}")
return ip
except socket.gaierror as e:
raise RuntimeError(f"Failed to resolve proxy host '{hostname}': {e}")

151
src/platform_router.py Normal file
View File

@@ -0,0 +1,151 @@
"""
platform_router.py — Абстрактный базовый класс для управления маршрутизацией.
Определяет интерфейс, который должны реализовать платформенные модули
(macos_router.py, windows_router.py). Обеспечивает единообразный контракт:
setup_interface() — назначить IP адрес TUN-интерфейсу
add_proxy_bypass_route() — исключить IP прокси из маршрутизации через TUN
add_routes() — перенаправить весь трафик через TUN
cleanup() — откатить все изменения (восстановить оригинал)
"""
from __future__ import annotations
import logging
import subprocess
from abc import ABC, abstractmethod
from typing import List
from .config import ProxyConfig
logger = logging.getLogger(__name__)
class PlatformRouter(ABC):
"""Абстрактный базовый класс для платформенной маршрутизации.
Отвечает за конфигурацию TUN-интерфейса (IP-адрес) и манипуляцию
таблицей маршрутизации ОС. Конкретные реализации — в наследниках.
Attributes:
config: ProxyConfig с параметрами подключения.
original_gateway: Оригинальный шлюз по умолчанию (для restore).
original_iface: Оригинальный сетевой интерфейс (для restore).
proxy_ip: Resolve-нутый IP прокси-сервера.
_routes_added: Флаг, указывающий что маршруты были добавлены.
"""
def __init__(
self,
config: ProxyConfig,
original_gateway: str,
original_iface: str,
proxy_ip: str,
) -> None:
self.config = config
self.original_gateway = original_gateway
self.original_iface = original_iface
self.proxy_ip = proxy_ip
self._routes_added = False
self._interface_configured = False
# ==========================================================================
# Абстрактные методы — реализуются наследниками
# ==========================================================================
@abstractmethod
def setup_interface(self) -> None:
"""Назначает IP-адрес TUN-интерфейсу и поднимает его.
Вызывается ПОСЛЕ запуска tun2socks (который создаёт интерфейс).
"""
...
@abstractmethod
def add_proxy_bypass_route(self) -> None:
"""Добавляет маршрут-исключение для IP прокси-сервера.
Трафик к прокси-серверу должен идти через оригинальный шлюз,
а НЕ через TUN-интерфейс. Иначе — бесконечная петля маршрутизации.
"""
...
@abstractmethod
def add_routes(self) -> None:
"""Перенаправляет весь трафик через TUN-интерфейс.
Использует split-routing (набор подсетей, покрывающих 0.0.0.0/0)
вместо замены default route, чтобы не ломать существующую маршрутизацию.
"""
...
@abstractmethod
def cleanup(self) -> None:
"""Откатывает ВСЕ изменения маршрутизации и интерфейса.
Должен быть идемпотентным (безопасно вызывать повторно).
Вызывается в finally-блоке и при обработке SIGTERM.
"""
...
# ==========================================================================
# Утилиты для наследников
# ==========================================================================
def _run_cmd(self, cmd: List[str], check: bool = True) -> subprocess.CompletedProcess:
"""Выполняет системную команду с логированием.
Args:
cmd: Список аргументов команды (как для subprocess.run).
check: Если True — бросает исключение при ненулевом exit code.
Returns:
CompletedProcess с результатом выполнения.
"""
cmd_str = " ".join(cmd)
if self.config.dry_run:
logger.info(f"[DRY-RUN] Would execute: {cmd_str}")
return subprocess.CompletedProcess(cmd, 0, "", "")
logger.debug(f"Executing: {cmd_str}")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30,
)
if result.returncode != 0 and check:
logger.error(
f"Command failed (exit {result.returncode}): {cmd_str}\n"
f" stdout: {result.stdout.strip()}\n"
f" stderr: {result.stderr.strip()}"
)
raise subprocess.CalledProcessError(
result.returncode, cmd, result.stdout, result.stderr
)
return result
except subprocess.TimeoutExpired:
logger.error(f"Command timed out: {cmd_str}")
raise
def _run_cmd_silent(self, cmd: List[str]) -> bool:
"""Выполняет команду без exceptions при ошибке.
Используется в cleanup() где ошибки неизбежны (маршрут может
уже не существовать).
Returns:
True если команда выполнилась успешно (exit 0).
"""
try:
result = self._run_cmd(cmd, check=False)
return result.returncode == 0
except Exception:
return False

123
src/privileges.py Normal file
View File

@@ -0,0 +1,123 @@
"""
privileges.py — Проверка и получение прав администратора/root.
Создание TUN-интерфейсов и изменение таблицы маршрутизации требуют
привилегий суперпользователя на обеих платформах:
- macOS: root (через sudo)
- Windows: Administrator (через UAC / runas)
"""
from __future__ import annotations
import logging
import os
import platform
import sys
logger = logging.getLogger(__name__)
def is_admin() -> bool:
"""Проверяет, запущен ли скрипт с правами администратора.
Returns:
True если текущий процесс имеет права root/Administrator.
"""
system = platform.system()
if system in ("Darwin", "Linux"):
# Unix: проверяем effective user ID
return os.geteuid() == 0
elif system == "Windows":
# Windows: проверяем через Windows API
try:
import ctypes
return bool(ctypes.windll.shell32.IsUserAnAdmin())
except (AttributeError, OSError):
return False
else:
logger.warning(f"Unknown platform: {system}. Cannot check privileges.")
return False
def ensure_admin() -> None:
"""Проверяет права администратора. Если их нет — выводит инструкцию и завершает.
На Windows пытается перезапустить скрипт с UAC-диалогом.
На macOS/Linux — выводит инструкцию запуска через sudo.
"""
if is_admin():
logger.debug("✓ Running with admin/root privileges.")
return
system = platform.system()
if system == "Windows":
_elevate_windows()
else:
_elevate_unix()
def _elevate_unix() -> None:
"""Выводит инструкцию для запуска через sudo и завершает процесс.
Не пытаемся запустить sudo программатически, т.к. это создаёт проблемы
с TTY и перехватом stdout/stderr.
"""
script = " ".join(sys.argv)
logger.error(
"❌ This program requires root privileges to create TUN interfaces "
"and modify routing tables."
)
print(
"\n"
"╔══════════════════════════════════════════════════════════════╗\n"
"║ 🔒 ROOT PRIVILEGES REQUIRED ║\n"
"╠══════════════════════════════════════════════════════════════╣\n"
"║ Please re-run with sudo: ║\n"
f"║ $ sudo {sys.executable} {script:<40}\n"
"╚══════════════════════════════════════════════════════════════╝"
)
sys.exit(1)
def _elevate_windows() -> None:
"""Пытается перезапустить скрипт с правами администратора через UAC.
Использует ShellExecuteW с verb='runas' для вызова UAC-диалога.
Текущий (непривилегированный) процесс завершается после запуска нового.
"""
try:
import ctypes
logger.info("🔒 Requesting UAC elevation...")
# Собираем команду: python.exe <script> <args>
params = " ".join(sys.argv)
# ShellExecuteW(hwnd, verb, file, parameters, directory, showCmd)
# verb='runas' вызывает UAC prompt
result = ctypes.windll.shell32.ShellExecuteW(
None, # hwnd — нет родительского окна
"runas", # verb — запустить с повышенными правами
sys.executable, # file — путь к python.exe
params, # parameters — аргументы скрипта
None, # directory — текущая директория
1, # nShowCmd — SW_SHOWNORMAL
)
# ShellExecuteW возвращает HINSTANCE > 32 при успехе
if result <= 32:
logger.error(f"❌ UAC elevation failed (error code: {result}).")
print("\n Please run this script as Administrator manually.")
sys.exit(1)
# Текущий процесс завершается — новый запущен с правами
sys.exit(0)
except Exception as e:
logger.error(f"❌ Failed to elevate privileges: {e}")
print("\n Please run this script as Administrator manually.")
sys.exit(1)

283
src/tun2socks_manager.py Normal file
View File

@@ -0,0 +1,283 @@
"""
tun2socks_manager.py — Управление процессом tun2socks.
Отвечает за:
1. Определение пути к бинарнику tun2socks (по платформе + архитектуре)
2. Формирование CLI-аргументов
3. Запуск через subprocess.Popen
4. Мониторинг здоровья процесса
5. Graceful shutdown (SIGTERM → wait → SIGKILL)
"""
from __future__ import annotations
import logging
import os
import platform
import shutil
import subprocess
import sys
import time
from pathlib import Path
from typing import Optional
from .config import ProxyConfig
logger = logging.getLogger(__name__)
# Таймаут ожидания корректного завершения tun2socks (секунды)
SHUTDOWN_TIMEOUT = 5
# Задержка после запуска tun2socks для создания TUN-интерфейса (секунды)
STARTUP_DELAY = 2
class Tun2socksManager:
"""Менеджер жизненного цикла процесса tun2socks.
Attributes:
config: ProxyConfig с параметрами подключения.
process: subprocess.Popen — запущенный процесс tun2socks.
"""
def __init__(self, config: ProxyConfig) -> None:
self.config = config
self.process: Optional[subprocess.Popen] = None
self._binary_path: Optional[str] = None
# ==========================================================================
# Определение бинарника
# ==========================================================================
def _find_binary(self) -> str:
"""Находит бинарник tun2socks для текущей платформы.
Порядок поиска:
1. ./bin/tun2socks-<platform>-<arch>[.exe]
2. ./bin/tun2socks[.exe]
3. tun2socks в PATH (shutil.which)
Returns:
Абсолютный путь к бинарнику.
Raises:
FileNotFoundError: Если бинарник не найден.
"""
system = platform.system().lower() # 'darwin', 'windows', 'linux'
machine = platform.machine().lower()
# Нормализация архитектуры
arch_map = {
"x86_64": "amd64",
"amd64": "amd64",
"arm64": "arm64",
"aarch64": "arm64",
}
arch = arch_map.get(machine, machine)
# Расширение файла
ext = ".exe" if system == "windows" else ""
# Директория bin/ относительно корня проекта
project_root = Path(__file__).resolve().parent.parent
bin_dir = project_root / "bin"
# Кандидаты для поиска
candidates = [
bin_dir / f"tun2socks-{system}-{arch}{ext}",
bin_dir / f"tun2socks{ext}",
]
for candidate in candidates:
if candidate.is_file():
logger.info(f"Found tun2socks binary: {candidate}")
return str(candidate)
# Fallback: ищем в PATH
found = shutil.which("tun2socks")
if found:
logger.info(f"Found tun2socks in PATH: {found}")
return found
# Не найден — подробная ошибка с инструкцией
raise FileNotFoundError(
f"\n"
f"╔══════════════════════════════════════════════════════════════╗\n"
f"║ ❌ tun2socks binary not found ║\n"
f"╠══════════════════════════════════════════════════════════════╣\n"
f"║ Expected locations: ║\n"
f"{candidates[0]} \n"
f"{candidates[1]} \n"
f"║ or 'tun2socks' in system PATH ║\n"
f"║ ║\n"
f"║ Download from: ║\n"
f"║ https://github.com/xjasonlyu/tun2socks/releases ║\n"
f"║ ║\n"
f"║ Platform: {system}/{arch} \n"
f"╚══════════════════════════════════════════════════════════════╝"
)
# ==========================================================================
# Запуск
# ==========================================================================
def start(self, network_interface: str) -> None:
"""Запускает процесс tun2socks.
Args:
network_interface: Имя основного сетевого интерфейса ОС (en0, Wi-Fi, etc.)
tun2socks привязывается к нему для исходящих соединений.
Raises:
FileNotFoundError: Если бинарник не найден.
RuntimeError: Если процесс не запустился.
"""
# На macOS: используем формат utun://utunN (с префиксом utun://)
# На Windows: используем просто имя "wintun"
device = self.config.tun_name
# В dry-run режиме не ищем бинарник — просто показываем команду
if self.config.dry_run:
cmd_str = (
f"tun2socks -device {device} "
f"-proxy {self.config.proxy_url} "
f"-loglevel {self.config.log_level}"
)
if network_interface:
cmd_str += f" -interface {network_interface}"
logger.info(f"[DRY-RUN] Would start: {cmd_str}")
return
binary = self._find_binary()
self._binary_path = binary
# Формируем аргументы CLI
cmd = [
binary,
"-device", device,
"-proxy", self.config.proxy_url,
"-loglevel", self.config.log_level,
]
# Привязка к основному интерфейсу (предотвращает routing loop
# на уровне tun2socks для исходящих соединений)
if network_interface:
cmd.extend(["-interface", network_interface])
cmd_str = " ".join(cmd)
logger.info(f"🚀 Starting tun2socks: {cmd_str}")
# Убеждаемся, что бинарник исполняемый (macOS/Linux)
if not self.config.is_windows:
os.chmod(binary, 0o755)
try:
self.process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
# На Windows: не создаём новое окно консоли
creationflags=(
subprocess.CREATE_NO_WINDOW
if self.config.is_windows
else 0
),
)
except PermissionError:
raise RuntimeError(
f"Permission denied for '{binary}'. "
f"Try: chmod +x {binary}"
)
# Ждём startup — tun2socks должен создать TUN-интерфейс
logger.info(
f"⏳ Waiting {STARTUP_DELAY}s for TUN interface creation..."
)
time.sleep(STARTUP_DELAY)
# Проверяем, что процесс не умер
if self.process.poll() is not None:
_, stderr = self.process.communicate(timeout=5)
raise RuntimeError(
f"tun2socks exited immediately (exit code: {self.process.returncode}).\n"
f"stderr: {stderr.strip()}"
)
logger.info(f"✓ tun2socks started (PID: {self.process.pid})")
# ==========================================================================
# Мониторинг
# ==========================================================================
def is_running(self) -> bool:
"""Проверяет, работает ли процесс tun2socks."""
if self.process is None:
return False
return self.process.poll() is None
def wait(self) -> int:
"""Блокирующее ожидание завершения процесса.
Returns:
Exit code процесса.
"""
if self.process is None:
return -1
return self.process.wait()
# ==========================================================================
# Остановка
# ==========================================================================
def stop(self) -> None:
"""Корректно останавливает tun2socks (graceful shutdown).
Стратегия:
1. SIGTERM (мягкая остановка)
2. Ожидание SHUTDOWN_TIMEOUT секунд
3. SIGKILL (принудительная остановка), если не завершился
Идемпотентный — безопасно вызывать повторно.
"""
if self.process is None:
return
if self.process.poll() is not None:
logger.debug(
f"tun2socks already exited (code: {self.process.returncode})"
)
self.process = None
return
pid = self.process.pid
logger.info(f"🛑 Stopping tun2socks (PID: {pid})...")
if self.config.dry_run:
logger.info("[DRY-RUN] Would terminate tun2socks process")
return
# Шаг 1: мягкая остановка
try:
self.process.terminate()
logger.debug(
f"Sent SIGTERM to PID {pid}, waiting {SHUTDOWN_TIMEOUT}s..."
)
self.process.wait(timeout=SHUTDOWN_TIMEOUT)
logger.info(f"✓ tun2socks stopped gracefully (PID: {pid})")
except subprocess.TimeoutExpired:
# Шаг 2: принудительная остановка
logger.warning(
f"⚠ tun2socks did not stop in {SHUTDOWN_TIMEOUT}s, "
f"sending SIGKILL..."
)
self.process.kill()
self.process.wait(timeout=5)
logger.info(f"✓ tun2socks killed (PID: {pid})")
except Exception as e:
logger.error(f"Error stopping tun2socks: {e}")
finally:
self.process = None

167
src/windows_router.py Normal file
View File

@@ -0,0 +1,167 @@
"""
windows_router.py — Реализация маршрутизации для Windows.
Особенности Windows:
1. Требуется wintun.dll (драйвер TUN для Windows) рядом с tun2socks.exe или в PATH
2. tun2socks создаёт интерфейс "wintun" при запуске
3. IP назначается через: netsh interface ipv4 set address name="wintun" ...
4. DNS через: netsh interface ipv4 set dnsservers name="wintun" ...
5. Маршрут: netsh interface ipv4 add route 0.0.0.0/0 "wintun" <gateway> metric=1
6. На Windows можно добавить единственный default route 0.0.0.0/0 с низким metric
(в отличие от macOS, где нужен split-routing)
"""
from __future__ import annotations
import logging
from .platform_router import PlatformRouter
logger = logging.getLogger(__name__)
class WindowsRouter(PlatformRouter):
"""Реализация маршрутизации для Windows.
Полный цикл:
1. setup_interface() → netsh set address + set dnsservers
2. add_proxy_bypass_route() → netsh add route <proxy_ip>/32
3. add_routes() → netsh add route 0.0.0.0/0 metric=1
4. cleanup() → netsh delete route ...
"""
def setup_interface(self) -> None:
"""Назначает IP-адрес и DNS wintun-интерфейсу.
На Windows tun2socks + wintun создаёт адаптер автоматически,
но IP и DNS нужно настроить вручную через netsh.
Команды:
netsh interface ipv4 set address name="wintun"
source=static addr=192.168.123.1 mask=255.255.255.0
netsh interface ipv4 set dnsservers name="wintun"
static address=8.8.8.8 register=none validate=no
"""
tun = self.config.tun_name
ip = self.config.tun_ip
mask = self.config.tun_mask
logger.info(f"⚙ Configuring interface {tun} with IP {ip}/{mask}")
# Назначаем IP-адрес
self._run_cmd([
"netsh", "interface", "ipv4", "set", "address",
f"name={tun}",
"source=static",
f"addr={ip}",
f"mask={mask}",
])
# Назначаем DNS-серверы
if self.config.dns_servers:
primary_dns = self.config.dns_servers[0]
self._run_cmd([
"netsh", "interface", "ipv4", "set", "dnsservers",
f"name={tun}",
"static",
f"address={primary_dns}",
"register=none",
"validate=no",
])
# Добавляем дополнительные DNS-серверы (если есть)
for dns in self.config.dns_servers[1:]:
self._run_cmd([
"netsh", "interface", "ipv4", "add", "dnsservers",
f"name={tun}",
f"address={dns}",
"validate=no",
])
logger.debug(f" DNS servers: {', '.join(self.config.dns_servers)}")
self._interface_configured = True
logger.info(f"✓ Interface {tun} configured with IP {ip}")
def add_proxy_bypass_route(self) -> None:
"""Добавляет маршрут-исключение: трафик к прокси → оригинальный шлюз.
Команда:
netsh interface ipv4 add route <proxy_ip>/32
<original_interface> <original_gateway> metric=1
"""
logger.info(
f"⚙ Adding bypass route: {self.proxy_ip}"
f"{self.original_gateway} via {self.original_iface}"
)
self._run_cmd([
"netsh", "interface", "ipv4", "add", "route",
f"{self.proxy_ip}/32",
self.original_iface,
self.original_gateway,
"metric=1",
])
logger.info(f"✓ Proxy bypass route added for {self.proxy_ip}")
def add_routes(self) -> None:
"""Добавляет default route через TUN-интерфейс с низким metric.
На Windows можно добавить маршрут 0.0.0.0/0 напрямую, задав
metric=1 (ниже, чем у оригинального default route), что даёт
приоритет TUN-интерфейсу.
Команда:
netsh interface ipv4 add route 0.0.0.0/0
"wintun" 192.168.123.1 metric=1
"""
tun = self.config.tun_name
tun_ip = self.config.tun_ip
logger.info(f"⚙ Adding default route via {tun} (metric=1)")
self._run_cmd([
"netsh", "interface", "ipv4", "add", "route",
"0.0.0.0/0",
tun,
tun_ip,
"metric=1",
])
self._routes_added = True
logger.info(f"✓ Default route added through {tun}")
def cleanup(self) -> None:
"""Удаляет все добавленные маршруты (graceful shutdown).
Идемпотентный — безопасно вызывать повторно.
"""
logger.info("🧹 Starting route cleanup...")
tun = self.config.tun_name
tun_ip = self.config.tun_ip
errors = 0
# 1. Удаляем default route через TUN
if self._routes_added:
if not self._run_cmd_silent([
"netsh", "interface", "ipv4", "delete", "route",
"0.0.0.0/0", tun, tun_ip,
]):
errors += 1
self._routes_added = False
# 2. Удаляем bypass-маршрут для прокси
self._run_cmd_silent([
"netsh", "interface", "ipv4", "delete", "route",
f"{self.proxy_ip}/32",
self.original_iface,
self.original_gateway,
])
if errors:
logger.warning(f"{errors} route(s) failed to delete (may already be gone)")
else:
logger.info("✓ All routes cleaned up successfully")