feat: add declarative sing-box egress control plane

2026-08-27 00:58:52 +05:00
commit b8d19b2c3e
54 changed files with 3337 additions and 0 deletions
@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from vpn_egressctl.renderer_1_13_19 import render_bytes, render_config
+from vpn_egressctl.uri import parse_hysteria2_uri
+
+from tests.helpers import make_policy
+
+
+class RendererTests(unittest.TestCase):
+    def test_complete_production_shape(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            policy = make_policy(Path(directory))
+            endpoint = parse_hysteria2_uri(
+                "hysteria2://AUTH@fi.api.withen.pro:443/?insecure=0&obfs=salamander&obfs-password=OBFS"
+            )
+            config = render_config(policy, endpoint)
+        self.assertEqual(list(config), ["log", "dns", "inbounds", "outbounds", "route"])
+        tun = config["inbounds"][0]
+        self.assertEqual(tun["route_exclude_address"], ["10.20.0.0/24", "10.30.0.0/24", "127.0.0.0/8"])
+        self.assertEqual(tun["iproute2_table_index"], 2022)
+        self.assertEqual(tun["auto_redirect_input_mark"], "0x2023")
+        self.assertEqual(tun["auto_redirect_output_mark"], "0x2024")
+        self.assertEqual(tun["auto_redirect_reset_mark"], "0x2025")
+        outbound = config["outbounds"][0]
+        self.assertEqual(outbound["server"], "fi.api.withen.pro")
+        self.assertEqual(outbound["password"], "AUTH")
+        self.assertEqual(outbound["obfs"]["password"], "OBFS")
+        self.assertEqual(outbound["bind_interface"], "eth0")
+        self.assertEqual(config["dns"]["servers"][1]["detour"], "hy2-out")
+        self.assertEqual(config["route"]["final"], "hy2-out")
+        self.assertNotIn("185.156.108.141", render_bytes(policy, endpoint).decode())
+
+    def test_multi_port_mapping(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            config = render_config(
+                make_policy(Path(directory)),
+                parse_hysteria2_uri("hy2://x@example.com:443,5000-6000"),
+            )
+        outbound = config["outbounds"][0]
+        self.assertNotIn("server_port", outbound)
+        self.assertEqual(outbound["server_ports"], ["443", "5000:6000"])
+
+    def test_deterministic_utf8_json(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            policy = make_policy(Path(directory))
+            endpoint = parse_hysteria2_uri("hy2://x@example.com#Тест")
+            first = render_bytes(policy, endpoint)
+            second = render_bytes(policy, endpoint)
+        self.assertEqual(first, second)
+        self.assertTrue(first.endswith(b"\n"))
+        json.loads(first)
+
+
+if __name__ == "__main__":
+    unittest.main()