RoyirinRosmontis/Private-Proxy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

feat(lite): add opt-in tun ipv6 mode

Browse Source
This commit is contained in:
RoyirinRosmontis
2026-09-19 04:20:34 +08:00
parent 2315735979
commit 9d5bc527d3
4 changed files with 80 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lite/rrpp_core/tun_backend.py
+33 -17
View File
@@ -53,6 +53,7 @@ class TunBackend:
self._ipv6_address = "fd42:7272:5050::1"
self._dns_configured = False
self._tun_dns_servers = ("8.8.8.8", "1.1.1.1")
self._ipv6_enabled = False
self._bypass_lan = True
self._cleanup_pending = False
self._state_path = self.user_dir / "tun_state.json"
@@ -251,6 +252,10 @@ class TunBackend:
continue
self._node_routes.append(address)
self._write_state()
if candidates_v6 and not self._ipv6_enabled:
if not candidates_v4:
raise TunBackendError("IPv6 模式已关闭，代理节点仅包含 IPv6 地址")
candidates_v6 = []
if candidates_v6 and not self._physical_ipv6_ifindex:
# 双栈节点仍可使用 IPv4 地址；只有 IPv6-only 节点需要物理 IPv6 出口。
if candidates_v4:
@@ -289,6 +294,8 @@ class TunBackend:
except ValueError:
return
if parsed.version == 6:
if not self._ipv6_enabled:
raise TunBackendError("IPv6 模式未启用，无法添加 IPv6 物理路由")
normalized = parsed.compressed
if normalized in self._node_ipv6_routes:
return
@@ -400,14 +407,15 @@ class TunBackend:
f"{self._device}", self._gateway, "metric=1"])
self._route_added = True
self._write_state()
self._netsh(["interface", "ipv6", "set", "address",
f"interface={self._device}", f"address={self._ipv6_address}",
"store=active"])
self._netsh(["interface", "ipv6", "add", "route", "prefix=::/0",
f"interface={self._device}", f"nexthop={self._ipv6_gateway}",
"metric=1"])
self._ipv6_route_added = True
self._write_state()
if self._ipv6_enabled:
self._netsh(["interface", "ipv6", "set", "address",
f"interface={self._device}", f"address={self._ipv6_address}",
"store=active"])
self._netsh(["interface", "ipv6", "add", "route", "prefix=::/0",
f"interface={self._device}", f"nexthop={self._ipv6_gateway}",
"metric=1"])
self._ipv6_route_added = True
self._write_state()
if not self._bypass_lan:
# Windows 物理网卡通常已有更具体的私网路由；补同前缀的 TUN
# 路由，确保“内网也代理”不会在进入 tun2socks 前被物理网卡截走。
@@ -417,12 +425,13 @@ class TunBackend:
f"{self._device}", self._gateway, "metric=1"])
self._lan_routes.append(network)
self._write_state()
for network in ("fc00::/7",):
self._netsh(["interface", "ipv6", "add", "route", f"prefix={network}",
f"interface={self._device}", f"nexthop={self._ipv6_gateway}",
"metric=1"])
self._lan_ipv6_routes.append(network)
self._write_state()
if self._ipv6_enabled:
for network in ("fc00::/7",):
self._netsh(["interface", "ipv6", "add", "route", f"prefix={network}",
f"interface={self._device}", f"nexthop={self._ipv6_gateway}",
"metric=1"])
self._lan_ipv6_routes.append(network)
self._write_state()
self._configure_dns()
def _configure_dns(self) -> None:
@@ -430,7 +439,7 @@ class TunBackend:
if os.name != "nt":
return
ipv4 = [item for item in self._tun_dns_servers if ":" not in item]
ipv6 = [item for item in self._tun_dns_servers if ":" in item]
ipv6 = [item for item in self._tun_dns_servers if ":" in item] if self._ipv6_enabled else []
if ipv4:
self._netsh([
"interface", "ipv4", "set", "dnsservers",
@@ -531,6 +540,7 @@ class TunBackend:
"node_ipv6_routes": list(self._node_ipv6_routes),
"route_added": bool(self._route_added),
"ipv6_route_added": bool(self._ipv6_route_added),
"ipv6_enabled": bool(self._ipv6_enabled),
"dns_configured": bool(self._dns_configured),
"dns_servers": list(self._tun_dns_servers),
"cleanup_pending": bool(self._cleanup_pending),
@@ -587,6 +597,7 @@ class TunBackend:
self._gateway = str(state.get("gateway") or self._gateway)
self._ipv6_address = str(state.get("ipv6_address") or self._ipv6_address)
self._ipv6_gateway = str(state.get("ipv6_gateway") or self._ipv6_gateway)
self._ipv6_enabled = bool(state.get("ipv6_enabled", bool(state.get("ipv6_route_added"))))
self._physical_ipv6_gateway = str(state.get("physical_ipv6_gateway") or "")
self._physical_ipv6_ifindex = str(state.get("physical_ipv6_ifindex") or "")
self._route_added = bool(state.get("route_added"))
@@ -625,6 +636,7 @@ class TunBackend:
def start(self, node: dict[str, Any], *, bypass_lan: bool = True,
route_rules: list[dict[str, Any]] | None = None,
dns_servers: tuple[str, ...] | list[str] | None = None,
ipv6_enabled: bool = False,
direct_target_resolver: Callable[[str, str, int], tuple[str, int] | None] | None = None,
route_rule_resolver: Callable[[str], dict[str, Any] | None] | None = None,
route_node_resolver: Callable[[dict[str, Any]], dict[str, Any] | None] | None = None,
@@ -635,6 +647,7 @@ class TunBackend:
if not self.is_admin():
raise TunBackendError("启动虚拟网卡 TUN 需要管理员权限")
self._bypass_lan = bool(bypass_lan)
self._ipv6_enabled = bool(ipv6_enabled)
if dns_servers is not None:
cleaned = tuple(str(item).strip() for item in dns_servers if str(item).strip())
if not cleaned:
@@ -651,10 +664,13 @@ class TunBackend:
interface = self._physical_interface()
source_address = self._physical_source_address(interface)
self._physical_gateway, self._physical_ifindex = self._physical_route_info()
self._physical_ipv6_gateway, self._physical_ipv6_ifindex = self._physical_ipv6_route_info()
if self._ipv6_enabled:
self._physical_ipv6_gateway, self._physical_ipv6_ifindex = self._physical_ipv6_route_info()
else:
self._physical_ipv6_gateway = self._physical_ipv6_ifindex = ""
if not self._physical_ifindex:
raise TunBackendError("找不到物理默认路由，无法保证代理节点不进入 TUN 回环")
if not self._physical_ipv6_ifindex:
if self._ipv6_enabled and not self._physical_ipv6_ifindex:
self.log("警告：未找到物理 IPv6 默认路由，IPv6 目标将仅尝试通过 IPv4 代理节点转发")
try:
proxy_addresses_by_node: dict[str, list[tuple]] = {}