"""Проверки раздельной маршрутизации kit-sub: python3 tools/test/rules_test.py""" import json, os, sys, tempfile, unittest, importlib.util root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) tmp = tempfile.mkdtemp() cfgp = os.path.join(tmp, "config.json") json.dump({"listen": "127.0.0.1", "port": 1, "path": "/sub/", "upstream": "http://127.0.0.1:1/"}, open(cfgp, "w")) os.environ["KIT_SUB_CONFIG"] = cfgp os.environ["KIT_SUB_RULES"] = os.path.join(tmp, "rules.yaml") spec = importlib.util.spec_from_file_location("kit_sub", os.path.join(root, "scripts", "kit-sub.py")) ks = importlib.util.module_from_spec(spec); spec.loader.exec_module(ks) import yaml BASE = yaml.safe_dump({"proxies": [{"name": "a", "type": "vless"}, {"name": "b", "type": "vless"}], "proxy-groups": [{"name": "Авто", "type": "url-test", "proxies": ["a", "b"]}], "rules": ["MATCH,Авто"], "dns": {"enable": False}}).encode() def write(text): with open(ks.RULES_FILE, "w") as f: f.write(text) class Rules(unittest.TestCase): def tearDown(self): if os.path.exists(ks.RULES_FILE): os.remove(ks.RULES_FILE) def apply(self, text): write(text) loaded = ks.load_rules(report=lambda m: None) self.assertIsNotNone(loaded) return yaml.safe_load(ks.apply_rules(BASE, loaded)) def test_no_file(self): self.assertIsNone(ks.load_rules()) def test_basic(self): c = self.apply("via_vpn:\n - GEOSITE,youtube\n - DOMAIN-SUFFIX, example.org \n - IP-CIDR,1.2.3.0/24\n") r = c["rules"] self.assertEqual(r[-1], "MATCH,DIRECT") self.assertIn("GEOSITE,youtube,Авто", r) self.assertIn("DOMAIN-SUFFIX,example.org,Авто", r) self.assertIn("IP-CIDR,1.2.3.0/24,Авто,no-resolve", r) # адресные правила не разрешают имена self.assertLess(r.index("IP-CIDR,192.168.0.0/16,DIRECT,no-resolve"), r.index("GEOSITE,youtube,Авто")) self.assertEqual(c["mode"], "rule") def test_dns_no_leak(self): c = self.apply("via_vpn:\n - GEOSITE,youtube\n - DOMAIN-SUFFIX,example.org\n - DOMAIN,api.example.com\n") d = c["dns"] self.assertEqual(d["enhanced-mode"], "fake-ip") self.assertEqual(d["nameserver"], ["system"]) self.assertNotIn("fallback", d) self.assertEqual(d["nameserver-policy"]["geosite:youtube"], ks.DOH_DEFAULT) self.assertIn("+.example.org", d["nameserver-policy"]) self.assertIn("api.example.com", d["nameserver-policy"]) # DoH-серверы идут через VPN, а не напрямую self.assertIn("IP-CIDR,1.1.1.1/32,Авто,no-resolve", c["rules"]) self.assertFalse(c["sniffer"]["override-destination"]) def test_direct_dns(self): c = self.apply("direct_dns:\n - https://dns.example.net/dns-query\n - http://evil/x\n - 'https://a b/'\nvia_vpn:\n - GEOSITE,openai\n") self.assertEqual(c["dns"]["nameserver"], ["https://dns.example.net/dns-query"]) def test_bad_rules_dropped(self): write("via_vpn:\n - MATCH,DIRECT\n - PROCESS-NAME,x\n - SCRIPT,abc\n - 'DOMAIN,a.com,DIRECT'\n - 'DOMAIN,a b'\n - GEOSITE,youtube\n - 'DOMAIN-SUFFIX,x.org\\nMATCH,Авто'\n") loaded = ks.load_rules(report=lambda m: None) self.assertEqual(loaded[0], [("GEOSITE", "youtube")]) def test_broken_files_leave_subscription_alone(self): for text in ("", "via_vpn: 5", "via_vpn: []", "- a\n- b", ":\n - [", "via_vpn:\n - MATCH,DIRECT"): write(text) self.assertIsNone(ks.load_rules(report=lambda m: None), text) with open(ks.RULES_FILE, "wb") as f: f.write(b"via_vpn:\n" + b" - DOMAIN,a.com\n" * 6000) self.assertIsNone(ks.load_rules(report=lambda m: None)) # слишком большой def test_limits_and_duplicates(self): write("via_vpn:\n" + "".join(f" - DOMAIN,h{i}.example.org\n" for i in range(400)) + " - DOMAIN,h1.example.org\n") rules, _ = ks.load_rules(report=lambda m: None) self.assertEqual(len(rules), ks.RULES_MAX_COUNT) self.assertEqual(len(set(rules)), len(rules)) def test_no_proxies_untouched(self): write("via_vpn:\n - GEOSITE,youtube\n") loaded = ks.load_rules(report=lambda m: None) self.assertEqual(ks.apply_rules(b"proxies: []\n", loaded), b"proxies: []\n") def test_forgiving_input(self): # Что если: строчные типы, BOM и CRLF из Windows-редактора, лишние пробелы with open(ks.RULES_FILE, "wb") as f: f.write("\ufeffvia_vpn:\r\n - geosite , youtube\r\n - domain-suffix,Example.ORG\r\n".encode("utf-8")) rules, _ = ks.load_rules(report=lambda m: None) self.assertEqual(rules, [("GEOSITE", "youtube"), ("DOMAIN-SUFFIX", "Example.ORG")]) def test_bad_cidr_and_short_keyword(self): write("via_vpn:\n - IP-CIDR,999.1.1.1/8\n - IP-CIDR,1.2.3.4/99\n - IP-CIDR6,1.2.3.0/24\n - IP-CIDR,2001:db8::/32\n" " - IP-CIDR,203.0.113.0/24\n - IP-CIDR6,2001:db8::/32\n - DOMAIN-KEYWORD,a\n - DOMAIN-KEYWORD,video\n") rules, _ = ks.load_rules(report=lambda m: None) self.assertEqual(rules, [("IP-CIDR", "203.0.113.0/24"), ("IP-CIDR6", "2001:db8::/32"), ("DOMAIN-KEYWORD", "video")]) def test_template_from_kit_sh_is_valid(self): # Что если: шаблон kit net split on сам окажется с ошибкой import re t = open(os.path.join(root, "scripts", "kit.sh"), encoding="utf-8").read() body = t[t.index("split_template() {"):] body = body[body.index("<<'YAML'\n") + 9: body.index("\nYAML\n")] write(body) rules, _ = ks.load_rules(report=lambda m: self.fail(m)) self.assertGreater(len(rules), 40) self.assertIn(("GEOIP", "telegram"), rules) self.assertNotIn(("DOMAIN-SUFFIX", "example.org"), rules) # пример закомментирован def test_rules_only_for_mihomo_apps(self): # Что если: Karing/Hiddify просят формат Clash – правила им не нужны (у них своя маршрутизация) def gets_rules(ua): clash = bool(ks.CLASH_UA.search(ua)) return clash and not ks.NO_AWG_UA.search(ua) for ua in ("FlClash/0.8.80", "Clash.Meta/1.18", "mihomo/1.19", "clash-verge/v2.0", "ClashVerge/2.0", "Mihomo Party/1.5"): self.assertTrue(gets_rules(ua), ua) for ua in ("Karing/1.2.5", "Karing/1.2.5 clash.meta", "Hiddify/2.0 clash", "sing-box/1.9", "Happ/1.0", "Stash/2.7", "v2rayN/6.0", "Mozilla/5.0"): self.assertFalse(gets_rules(ua), ua) def test_routing_payloads_for_panel(self): import base64 write("via_vpn:\n - GEOSITE,youtube\n - DOMAIN-SUFFIX,example.org\n - DOMAIN,api.example.com\n - GEOIP,telegram\n - IP-CIDR,203.0.113.0/24\n") loaded = ks.load_rules(report=lambda m: None) link = ks.routing_payload("happ", loaded) self.assertTrue(link.startswith("happ://routing/onadd/")) h = json.loads(base64.b64decode(link.split("/onadd/")[1])) self.assertEqual(h["Name"], "3X-UI_autoVPS") self.assertEqual(h["GlobalProxy"], "false") self.assertEqual(h["ProxySites"], ["geosite:youtube", "domain:example.org", "full:api.example.com"]) self.assertEqual(h["ProxyIp"], ["geoip:telegram", "203.0.113.0/24"]) j = json.loads(ks.routing_payload("json", loaded)) self.assertEqual(j["RouteOrder"], "block-proxy-direct") self.assertEqual(j["DirectSites"], ["regexp:.*"]) # остальное напрямую: панель сама добавляет «всё через прокси» в конец self.assertIn("0.0.0.0/0", j["DirectIp"]) self.assertEqual(j["ProxyIp"][:2], ["1.1.1.1", "8.8.8.8"]) self.assertEqual(j["DomesticDNSDomain"], "localhost") self.assertLess(len(link), 8192) # предел панели на значение заголовка Routing def test_full_default_template_fits_panel_limit(self): t = open(os.path.join(root, "scripts", "kit.sh"), encoding="utf-8").read() body = t[t.index("split_template() {"):] body = body[body.index("<<'YAML'\n") + 9: body.index("\nYAML\n")] write(body) loaded = ks.load_rules(report=lambda m: self.fail(m)) self.assertLess(len(ks.routing_payload("happ", loaded)), 8192) if __name__ == "__main__": unittest.main(verbosity=1)