diff --git a/proxy-client/.local/bin/proxy-client b/proxy-client/.local/bin/proxy-client index 4fe981f..c2a0c7a 100755 --- a/proxy-client/.local/bin/proxy-client +++ b/proxy-client/.local/bin/proxy-client @@ -1317,6 +1317,10 @@ def check_port_available(address, port): family = socket.AF_INET6 if address.version == 6 else socket.AF_INET try: with socket.socket(family, socket.SOCK_STREAM) as listener: + # Без флага bind отказывает, пока после остановки Xray на порту + # доживают закрытые соединения (TIME_WAIT); живой слушатель + # на Linux флаг не обходит, и занятый порт по-прежнему виден. + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) listener.bind((str(address), port)) except OSError: raise ProxyError("локальный порт {} занят или недоступен".format(port)) diff --git a/tests/test_proxy_client.py b/tests/test_proxy_client.py index 7d7ce38..d9e1503 100644 --- a/tests/test_proxy_client.py +++ b/tests/test_proxy_client.py @@ -5,6 +5,7 @@ import json import hashlib import os import runpy +import socket import subprocess import sys import tempfile @@ -1302,5 +1303,43 @@ class VlessProfileTest(unittest.TestCase): self.module["parse_vless"](url) +class PortCheckTest(unittest.TestCase): + def setUp(self): + self.module = runpy.run_path(str(CLI)) + self.check = self.module["check_port_available"] + self.address = self.module["ipaddress"].ip_address("127.0.0.1") + + @staticmethod + def listener(): + # Как у Xray: Go ставит SO_REUSEADDR на слушающий сокет. + sock = socket.socket() + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind(("127.0.0.1", 0)) + sock.listen() + return sock + + def test_port_is_free_right_after_xray_closed_its_connections(self): + listener = self.listener() + port = listener.getsockname()[1] + client = socket.create_connection(("127.0.0.1", port)) + accepted, _peer = listener.accept() + # После Ctrl+C Xray первым закрывает соединения, и ядро держит их + # «хвосты» на локальном порту ещё около минуты. + accepted.close() + listener.close() + try: + self.check(self.address, port) + finally: + client.close() + + def test_live_listener_is_still_reported_as_busy(self): + listener = self.listener() + try: + with self.assertRaisesRegex(self.module["ProxyError"], "занят"): + self.check(self.address, listener.getsockname()[1]) + finally: + listener.close() + + if __name__ == "__main__": unittest.main()