|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +import json |
| 4 | +import os |
| 5 | +import sys |
| 6 | +import time |
| 7 | +import unittest |
| 8 | +from unittest import mock |
| 9 | + |
| 10 | +from singbox2proxy import SingBoxProxy |
| 11 | + |
| 12 | + |
| 13 | +TEST_LINK = os.environ.get("TEST_SINGBOX_LINK") |
| 14 | + |
| 15 | + |
| 16 | +def _read_config(proxy: SingBoxProxy) -> dict: |
| 17 | + path = proxy.create_config_file() |
| 18 | + try: |
| 19 | + with open(path, "r", encoding="utf-8") as f: |
| 20 | + return json.load(f) |
| 21 | + finally: |
| 22 | + if os.path.exists(path): |
| 23 | + os.remove(path) |
| 24 | + |
| 25 | + |
| 26 | +def _find_tun_inbound(config: dict) -> dict | None: |
| 27 | + for inbound in config.get("inbounds", []): |
| 28 | + if inbound.get("type") == "tun": |
| 29 | + return inbound |
| 30 | + return None |
| 31 | + |
| 32 | + |
| 33 | +class TestTunConfig(unittest.TestCase): |
| 34 | + def test_tun_disabled_by_default(self): |
| 35 | + proxy = SingBoxProxy("socks://127.0.0.1:1080", config_only=True) |
| 36 | + config = _read_config(proxy) |
| 37 | + self.assertIsNone(_find_tun_inbound(config)) |
| 38 | + |
| 39 | + def test_tun_basic_fields(self): |
| 40 | + proxy = SingBoxProxy("socks://127.0.0.1:1080", config_only=True, tun_enabled=True) |
| 41 | + config = _read_config(proxy) |
| 42 | + tun = _find_tun_inbound(config) |
| 43 | + self.assertIsNotNone(tun) |
| 44 | + self.assertEqual(tun["type"], "tun") |
| 45 | + self.assertTrue(tun["auto_route"]) |
| 46 | + self.assertTrue(tun["strict_route"]) |
| 47 | + self.assertEqual(tun["stack"], "system") |
| 48 | + self.assertEqual(tun["address"], ["172.19.0.1/30"]) |
| 49 | + |
| 50 | + def test_tun_auto_redirect_on_linux(self): |
| 51 | + with mock.patch.object(sys, "platform", "linux"): |
| 52 | + proxy = SingBoxProxy("socks://127.0.0.1:1080", config_only=True, tun_enabled=True) |
| 53 | + with mock.patch.object(SingBoxProxy, "_parse_core_version", return_value=(1, 13, 0)): |
| 54 | + config = _read_config(proxy) |
| 55 | + self.assertTrue(_find_tun_inbound(config).get("auto_redirect")) |
| 56 | + |
| 57 | + def test_tun_no_auto_redirect_on_windows(self): |
| 58 | + with mock.patch.object(sys, "platform", "win32"): |
| 59 | + proxy = SingBoxProxy("socks://127.0.0.1:1080", config_only=True, tun_enabled=True) |
| 60 | + with mock.patch.object(SingBoxProxy, "_parse_core_version", return_value=(1, 13, 0)): |
| 61 | + config = _read_config(proxy) |
| 62 | + self.assertNotIn("auto_redirect", _find_tun_inbound(config)) |
| 63 | + |
| 64 | + def test_tun_auto_redirect_skipped_for_old_core(self): |
| 65 | + with mock.patch.object(sys, "platform", "linux"): |
| 66 | + proxy = SingBoxProxy("socks://127.0.0.1:1080", config_only=True, tun_enabled=True) |
| 67 | + with mock.patch.object(SingBoxProxy, "_parse_core_version", return_value=(1, 9, 0)): |
| 68 | + config = _read_config(proxy) |
| 69 | + self.assertNotIn("auto_redirect", _find_tun_inbound(config)) |
| 70 | + |
| 71 | + def test_tun_auto_redirect_explicit_off(self): |
| 72 | + with mock.patch.object(sys, "platform", "linux"): |
| 73 | + proxy = SingBoxProxy( |
| 74 | + "socks://127.0.0.1:1080", |
| 75 | + config_only=True, |
| 76 | + tun_enabled=True, |
| 77 | + tun_auto_redirect=False, |
| 78 | + ) |
| 79 | + with mock.patch.object(SingBoxProxy, "_parse_core_version", return_value=(1, 13, 0)): |
| 80 | + config = _read_config(proxy) |
| 81 | + self.assertNotIn("auto_redirect", _find_tun_inbound(config)) |
| 82 | + |
| 83 | + |
| 84 | +def _has_privileges() -> bool: |
| 85 | + if sys.platform.startswith("win"): |
| 86 | + try: |
| 87 | + import ctypes |
| 88 | + |
| 89 | + return bool(ctypes.windll.shell32.IsUserAnAdmin()) |
| 90 | + except Exception: |
| 91 | + return False |
| 92 | + return hasattr(os, "geteuid") and os.geteuid() == 0 |
| 93 | + |
| 94 | + |
| 95 | +@unittest.skipUnless( |
| 96 | + TEST_LINK |
| 97 | + and (sys.platform.startswith("linux") or sys.platform.startswith("win")) |
| 98 | + and _has_privileges(), |
| 99 | + "Live TUN test requires Linux/Windows + root/admin + TEST_SINGBOX_LINK", |
| 100 | +) |
| 101 | +class TestTunLiveSystemVpn(unittest.TestCase): |
| 102 | + IP_URL = "https://api.ipify.org?format=json" |
| 103 | + |
| 104 | + def _fetch_proxy_ip(self) -> str: |
| 105 | + proxy = SingBoxProxy(TEST_LINK) |
| 106 | + try: |
| 107 | + return proxy.get(self.IP_URL, timeout=15).json()["ip"] |
| 108 | + finally: |
| 109 | + proxy.stop() |
| 110 | + |
| 111 | + def _fetch_system_ip(self) -> str: |
| 112 | + import requests |
| 113 | + |
| 114 | + return requests.get(self.IP_URL, timeout=15).json()["ip"] |
| 115 | + |
| 116 | + def test_tun_captures_system_traffic(self): |
| 117 | + proxy_ip = self._fetch_proxy_ip() |
| 118 | + baseline_ip = self._fetch_system_ip() |
| 119 | + self.assertNotEqual(proxy_ip, baseline_ip) |
| 120 | + |
| 121 | + tun_proxy = SingBoxProxy(TEST_LINK, tun_enabled=True) |
| 122 | + try: |
| 123 | + time.sleep(5) |
| 124 | + observed = self._fetch_system_ip() |
| 125 | + finally: |
| 126 | + tun_proxy.stop() |
| 127 | + |
| 128 | + self.assertEqual(observed, proxy_ip) |
| 129 | + |
| 130 | + |
| 131 | +if __name__ == "__main__": |
| 132 | + unittest.main() |
0 commit comments