maxdmitrov
972757ed75
Панель 3.9.0 меняет подключение через API ядра: старое удаляет, а новое с маскировкой Hysteria2 ядро 26.6.27 не принимает, и 443/udp закрывается (kit net masq on, kit net fp, kit net on). Теперь после правки kit сверяет порты и при нужде перезапускает ядро (server/restartXrayService, запасной путь – systemctl restart x-ui). kit check: процесс Xray (дочерний у x-ui) и порты всех включённых подключений, TCP и UDP отдельно (nginx на 443/tcp больше не считается за Hysteria2). kit fix перезапускает ядро. Тест: tools/test/kit_ports_test.py.
87 lines
3.9 KiB
Python
87 lines
3.9 KiB
Python
"""Проверки портов подключений в kit.sh (inbound_listens, inbounds_down, xray_heal): TCP и UDP отдельно,
|
|
перезапуск ядра только если правка закрыла порт. python3 tools/test/kit_ports_test.py"""
|
|
import json, os, re, subprocess, tempfile, unittest
|
|
|
|
root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
src = open(os.path.join(root, "scripts", "kit.sh"), encoding="utf-8").read()
|
|
funcs = "".join(re.search(r"^%s\(\) \{.*?^\}\n" % n, src, re.S | re.M).group(0)
|
|
for n in ("port_net", "inbound_listens", "inbounds_down", "xray_heal"))
|
|
|
|
INBOUNDS = [
|
|
{"remark": "REALITY", "protocol": "vless", "port": 10443, "enable": True},
|
|
{"remark": "Hysteria2", "protocol": "hysteria", "port": 443, "enable": True},
|
|
{"remark": "Shadowsocks", "protocol": "shadowsocks", "port": 8388, "enable": True},
|
|
{"remark": "VMess-WS", "protocol": "vmess", "port": 10452, "enable": False},
|
|
]
|
|
|
|
# ss и api – заглушки: открытые порты лежат в файле «t:443», «u:443»; перезапуск ядра открывает порты из $AFTER.
|
|
STUBS = r'''
|
|
set -Eeuo pipefail
|
|
say() { echo "SAY: $*"; }; warn() { echo "WARN: $*"; }; sleep() { :; }
|
|
ss() { local p=${1:3:1} port=${2##*:}; grep -qx "$p:$port" "$LISTEN" && echo LISTEN || true; }
|
|
api() { cat "$LIST"; }
|
|
xray_restart_core() { echo RESTART; cp "$AFTER" "$LISTEN"; }
|
|
'''
|
|
|
|
|
|
def run(listen, body, after=None):
|
|
d = tempfile.mkdtemp()
|
|
paths = {k: os.path.join(d, k) for k in ("listen", "list", "after")}
|
|
for k, text in (("listen", "\n".join(listen) + "\n"), ("after", "\n".join(after if after is not None else listen) + "\n"),
|
|
("list", json.dumps(INBOUNDS))):
|
|
with open(paths[k], "w") as f:
|
|
f.write(text)
|
|
env = dict(os.environ, LISTEN=paths["listen"], LIST=paths["list"], AFTER=paths["after"])
|
|
r = subprocess.run(["bash", "-c", STUBS + funcs + body], capture_output=True, text=True, env=env)
|
|
return r.returncode, r.stdout + r.stderr
|
|
|
|
|
|
ALL = ["t:10443", "u:443", "t:8388", "u:8388"]
|
|
|
|
|
|
class Ports(unittest.TestCase):
|
|
def test_all_listen(self):
|
|
code, out = run(ALL + ["t:443"], 'inbounds_down; echo END')
|
|
self.assertEqual(code, 0, out)
|
|
self.assertEqual(out.strip(), "END")
|
|
|
|
def test_nginx_tcp_is_not_hysteria_udp(self):
|
|
code, out = run(["t:10443", "t:443", "t:8388", "u:8388"], 'inbounds_down')
|
|
self.assertEqual(out.strip(), "Hysteria2\t443\tudp", out)
|
|
|
|
def test_shadowsocks_needs_both(self):
|
|
code, out = run(["t:10443", "u:443", "t:8388"], 'inbounds_down')
|
|
self.assertEqual(out.strip(), "Shadowsocks\t8388\ttcp+udp", out)
|
|
|
|
def test_disabled_ignored(self):
|
|
code, out = run(ALL, 'inbounds_down | grep -c VMess || true')
|
|
self.assertEqual(out.strip(), "0")
|
|
|
|
def test_bad_port_value(self):
|
|
code, out = run(ALL, 'inbound_listens "443;id" udp && echo YES || echo NO')
|
|
self.assertEqual(out.strip(), "NO")
|
|
|
|
def test_heal_restarts_when_patch_closed_port(self):
|
|
code, out = run(["t:10443", "t:8388", "u:8388"], 'xray_heal ""', after=ALL)
|
|
self.assertEqual(code, 0, out)
|
|
self.assertIn("RESTART", out)
|
|
self.assertIn("Hysteria2", out)
|
|
self.assertNotIn("WARN", out)
|
|
|
|
def test_heal_quiet_when_nothing_closed(self):
|
|
code, out = run(ALL, 'xray_heal ""')
|
|
self.assertEqual(out.strip(), "")
|
|
|
|
def test_heal_ignores_port_down_before_patch(self):
|
|
code, out = run(["t:10443", "t:8388", "u:8388"], 'xray_heal "$(printf "Hysteria2\\t443\\tudp")"')
|
|
self.assertNotIn("RESTART", out)
|
|
|
|
def test_heal_warns_if_restart_did_not_help(self):
|
|
code, out = run(["t:10443", "t:8388", "u:8388"], 'xray_heal ""')
|
|
self.assertIn("RESTART", out)
|
|
self.assertIn("WARN: После перезапуска ядра не слушают: Hysteria2:443", out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|