fix(proxy-client): порт свободен сразу после остановки по Ctrl+C

Проверка порта перед запуском открывала сокет без SO_REUSEADDR и
отказывала, пока на порту доживали закрытые соединения Xray (TIME_WAIT),
хотя сам Xray запустился бы. Живой слушатель флаг на Linux не обходит,
так что занятый порт по-прежнему обнаруживается.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
ddadminandClaude Opus 5.5 committed 2026-09-28 08:51:34 -04:00
1 parent b492f8f4eb
commit 205e995f26
2 files changed
+43

No files matched your search

+4
View File
@@ -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))
+39
View File
@@ -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()