feat(xray-proxy): добавлен безопасный жизненный цикл управляемого Xray

- Зачем:
  - stop/status опирались на обход argv и могли счесть чужой или
    переиспользованный PID управляемым процессом, а параллельный start
    и аварийная гибель супервизора не имели определённого поведения.
- Что:
  - введена модель идентичности: PID + starttime (поле 22 /proc/PID/stat),
    /proc/PID/exe для ребёнка, канонический путь скрипта для супервизора,
    boot_id как граница загрузки; перечитка перед каждым сигналом.
  - добавлены lifetime-flock супервизора, три исхода проверки состояния
    (здоровая пара / подтверждённый сирота / непроверяемое), отказ при
    занятом порте пробным bind и очистка stale только под lock.
  - тесты расширены до 49 (blank flock, подмена starttime, чужой exe,
    boot_id, SIGKILL супервизора, занятый порт, парсер stat); спека
    приведена к модели идентичности; тикет 03 закрыт, леджер дополнен.
- Проверка:
  - python3 -m unittest discover -s tests — 49 OK.
  - bash tests/test_proxy_launchers.sh — 8 PASS.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
ddadminandClaude Fable 5 committed 2026-07-20 07:02:33 -04:00
1 parent f2d303c233
commit 5ace0cfae0
6 files changed
+890 -100

No files matched your search

+313 -54
View File
@@ -11,6 +11,7 @@ if sys.version_info < (3, 8):
import argparse
import configparser
import fcntl
import http.client
import ipaddress
import json
@@ -63,6 +64,10 @@ STATE_KEYS = {
"config_path",
"address",
"port",
"supervisor_path",
"supervisor_starttime",
"xray_starttime",
"boot_id",
}
@@ -126,6 +131,10 @@ def generated_config_path():
return state_home() / "client.json"
def lock_path():
return state_home() / "lock"
def validate_address(value):
try:
parsed = ipaddress.ip_address(value)
@@ -588,10 +597,54 @@ def run_xray_test(path, config_path):
def process_alive(pid):
try:
os.kill(pid, 0)
Path("/proc/{}".format(pid)).stat()
return True
except OSError:
except FileNotFoundError:
return False
except OSError:
return True
def parse_proc_stat_starttime(text):
comm_end = text.rfind(")")
if comm_end < 0:
raise ValueError("missing comm terminator")
fields = text[comm_end + 1 :].split()
if len(fields) <= 19:
raise ValueError("missing starttime")
return int(fields[19])
def process_starttime(pid):
try:
text = Path("/proc/{}/stat".format(pid)).read_text(encoding="ascii")
return parse_proc_stat_starttime(text)
except (OSError, UnicodeError, ValueError):
return None
def process_executable(pid):
try:
return str(Path("/proc/{}/exe".format(pid)).resolve(strict=True))
except (OSError, RuntimeError):
return None
def read_boot_id():
try:
value = Path("/proc/sys/kernel/random/boot_id").read_text(
encoding="ascii"
).strip()
except (OSError, UnicodeError):
return None
return value or None
def require_boot_id():
value = read_boot_id()
if value is None:
raise ProxyError("не удалось определить идентификатор текущей загрузки")
return value
def process_cmdline(pid):
@@ -602,16 +655,6 @@ def process_cmdline(pid):
return [item.decode("utf-8", "surrogateescape") for item in raw.split(b"\0") if item]
def process_parent(pid):
try:
for line in Path("/proc/{}/status".format(pid)).read_text(encoding="ascii").splitlines():
if line.startswith("PPid:"):
return int(line.split()[1])
except (OSError, ValueError):
pass
return None
def process_argument_resolves_to(pid, arguments, expected_path):
try:
process_cwd = Path("/proc/{}/cwd".format(pid)).resolve(strict=True)
@@ -648,34 +691,134 @@ def read_managed_state():
raise ProxyError("состояние управляемого Xray не подтверждено")
if not isinstance(value, dict) or set(value) != STATE_KEYS:
raise ProxyError("состояние управляемого Xray не подтверждено")
if not all(isinstance(value[key], int) and not isinstance(value[key], bool) for key in ("supervisor_pid", "xray_pid", "port")):
integer_keys = (
"supervisor_pid",
"xray_pid",
"port",
"supervisor_starttime",
"xray_starttime",
)
if not all(
isinstance(value[key], int) and not isinstance(value[key], bool)
for key in integer_keys
):
raise ProxyError("состояние управляемого Xray не подтверждено")
if not all(isinstance(value[key], str) for key in ("xray_path", "config_path", "address")):
string_keys = (
"xray_path",
"config_path",
"address",
"supervisor_path",
"boot_id",
)
if not all(isinstance(value[key], str) for key in string_keys):
raise ProxyError("состояние управляемого Xray не подтверждено")
return value
def supervisor_identity_matches(state):
pid = state["supervisor_pid"]
if pid <= 1:
return False
arguments = process_cmdline(pid)
if len(arguments) < 2:
return False
if not process_argument_resolves_to(
pid, [arguments[1]], Path(state["supervisor_path"])
):
return False
return process_starttime(pid) == state["supervisor_starttime"]
def xray_identity_matches(state):
pid = state["xray_pid"]
if pid <= 1 or process_executable(pid) != state["xray_path"]:
return False
return process_starttime(pid) == state["xray_starttime"]
def wait_for_child_starttime(child):
while True:
value = process_starttime(child.pid)
if value is not None:
return value
if child.poll() is not None:
raise ProxyError("Xray завершился до сохранения идентичности процесса")
time.sleep(0.01)
def signal_supervisor(state, signal_number):
if read_boot_id() != state["boot_id"] or not supervisor_identity_matches(state):
return False
try:
os.kill(state["supervisor_pid"], signal_number)
except OSError:
return False
return True
def signal_xray(state, signal_number):
if read_boot_id() != state["boot_id"] or not xray_identity_matches(state):
return False
try:
os.kill(state["xray_pid"], signal_number)
except OSError:
return False
return True
def managed_state_outcome(state):
current_boot_id = read_boot_id()
if current_boot_id is None:
return "unverifiable"
if current_boot_id != state["boot_id"]:
return "stale"
if state["supervisor_pid"] == state["xray_pid"]:
return "unverifiable"
supervisor_alive = process_alive(state["supervisor_pid"])
xray_alive = process_alive(state["xray_pid"])
if not supervisor_alive and not xray_alive:
return "stale"
supervisor_matches = supervisor_identity_matches(state)
xray_matches = xray_identity_matches(state)
if supervisor_matches and xray_matches:
return "healthy"
if not supervisor_alive and xray_matches:
return "orphan"
return "unverifiable"
def cleanup_stale_files():
for path in (state_path(), generated_config_path()):
try:
path.unlink()
except FileNotFoundError:
pass
def unverifiable_state_error():
return ProxyError(
"личность процесса не подтверждена; сигналы не отправлены, "
"удалите state.json и client.json вручную после проверки процессов"
)
def verify_managed_state(config=None, require_port=True):
state = read_managed_state()
supervisor = state["supervisor_pid"]
child = state["xray_pid"]
if supervisor <= 1 or child <= 1 or supervisor == child:
raise ProxyError("личность управляемого Xray не подтверждена")
supervisor_args = process_cmdline(supervisor)
child_args = process_cmdline(child)
cli = Path(__file__).resolve()
if (
not process_alive(supervisor)
or not process_argument_resolves_to(supervisor, supervisor_args, cli)
or "start" not in supervisor_args
or not process_alive(child)
or state["xray_path"] not in child_args
or "run" not in child_args
or "-c" not in child_args
or state["config_path"] not in child_args
or process_parent(child) != supervisor
):
raise ProxyError("личность процесса управляемого Xray не подтверждена")
outcome = managed_state_outcome(state)
if outcome == "stale":
lock = acquire_start_lock()
try:
state = read_managed_state()
outcome = managed_state_outcome(state)
if outcome == "stale":
cleanup_stale_files()
raise ProxyError("устаревшее состояние управляемого Xray очищено")
finally:
lock.close()
if outcome == "orphan":
raise ProxyError("обнаружен Xray-сирота; выполните xray-proxy stop")
if outcome != "healthy":
raise unverifiable_state_error()
if config is not None and (
config.proxy_type != "xray"
or config.address != state["address"]
@@ -687,6 +830,26 @@ def verify_managed_state(config=None, require_port=True):
return state
def acquire_start_lock():
private_directory(state_home())
try:
descriptor = os.open(str(lock_path()), os.O_RDWR | os.O_CREAT, 0o600)
lock = os.fdopen(descriptor, "r+")
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
os.chmod(str(lock_path()), 0o600)
except BlockingIOError:
if "lock" in locals():
lock.close()
raise ProxyError("другой управляемый запуск уже удерживает блокировку")
except OSError as error:
if "lock" in locals():
lock.close()
elif "descriptor" in locals():
os.close(descriptor)
raise ProxyError("не удалось открыть файл блокировки: {}".format(error))
return lock
def safe_summary(profile, local_port):
query = profile.query
print(
@@ -823,8 +986,39 @@ def command_start(config, profile_file):
raise ProxyError("для type=xray поле address должно быть loopback IP-адресом")
if not address.is_loopback:
raise ProxyError("для type=xray требуется loopback IP-адрес")
lock = acquire_start_lock()
try:
return command_start_locked(config, profile_file, address)
finally:
lock.close()
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:
listener.bind((str(address), port))
except OSError:
raise ProxyError("локальный порт {} занят или недоступен".format(port))
def command_start_locked(config, profile_file, address):
if state_path().exists() or state_path().is_symlink():
raise ProxyError("файл состояния уже существует; управляемый запуск отменён")
try:
existing_state = read_managed_state()
except ProxyError:
raise unverifiable_state_error()
outcome = managed_state_outcome(existing_state)
if outcome == "stale":
cleanup_stale_files()
elif outcome == "orphan":
raise ProxyError("обнаружен Xray-сирота; сначала выполните xray-proxy stop")
elif outcome == "healthy":
raise ProxyError("управляемый Xray уже запущен")
else:
raise unverifiable_state_error()
check_port_available(address, config.port)
profile_path = Path(profile_file) if profile_file else xray_config_home() / "default.url"
raw_profile = read_profile(profile_path)
@@ -847,9 +1041,15 @@ def command_start(config, profile_file):
previous = {}
for handled in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
previous[handled] = signal.signal(handled, request_stop)
boot_id = require_boot_id()
supervisor_path = str(Path(__file__).resolve(strict=True))
supervisor_starttime = process_starttime(os.getpid())
if supervisor_starttime is None:
raise ProxyError("не удалось сохранить идентичность супервизора")
child = subprocess.Popen(
[executable, "run", "-c", str(generated_config_path())]
)
xray_starttime = wait_for_child_starttime(child)
state = {
"supervisor_pid": os.getpid(),
"xray_pid": child.pid,
@@ -857,6 +1057,10 @@ def command_start(config, profile_file):
"config_path": str(generated_config_path()),
"address": config.address,
"port": config.port,
"supervisor_path": supervisor_path,
"supervisor_starttime": supervisor_starttime,
"xray_starttime": xray_starttime,
"boot_id": boot_id,
}
write_private_json(state_path(), state)
deadline = time.monotonic() + max(1.0, preflight_timeout())
@@ -872,17 +1076,25 @@ def command_start(config, profile_file):
last_error = error
time.sleep(0.03)
if last_error is not None:
raise ProxyError("локальный порт дочернего Xray не стал доступен")
raise ProxyError(
"локальный порт {} занят или недоступен".format(config.port)
)
diagnose_ip(config)
while child.poll() is None and stopping["signal"] is None:
time.sleep(0.05)
if stopping["signal"] is not None and child.poll() is None:
os.kill(child.pid, stopping["signal"])
try:
child.wait(timeout=stop_timeout())
except subprocess.TimeoutExpired:
child.kill()
child.wait()
if not signal_xray(state, stopping["signal"]):
if child.poll() is None:
raise unverifiable_state_error()
else:
try:
child.wait(timeout=stop_timeout())
except subprocess.TimeoutExpired:
if not signal_xray(state, signal.SIGKILL):
if child.poll() is None:
raise unverifiable_state_error()
else:
child.wait()
return_code = child.wait()
if stopping["signal"] is not None:
return 0
@@ -893,13 +1105,22 @@ def command_start(config, profile_file):
if "previous" in locals():
for handled, handler in previous.items():
signal.signal(handled, handler)
if "child" in locals() and child.poll() is None:
child.terminate()
try:
child.wait(timeout=stop_timeout())
except subprocess.TimeoutExpired:
child.kill()
child.wait()
preserve_managed_files = False
if "child" in locals() and child.poll() is None and "state" in locals():
if signal_xray(state, signal.SIGTERM):
try:
child.wait(timeout=stop_timeout())
except subprocess.TimeoutExpired:
if signal_xray(state, signal.SIGKILL):
child.wait()
elif child.poll() is None:
preserve_managed_files = True
elif child.poll() is None:
preserve_managed_files = True
elif "child" in locals() and child.poll() is None:
preserve_managed_files = True
if preserve_managed_files:
raise unverifiable_state_error()
cleanup_managed_files(os.getpid())
@@ -914,12 +1135,35 @@ def stop_timeout():
return value
def command_stop():
timeout = stop_timeout()
state = verify_managed_state(require_port=True)
try:
os.kill(state["supervisor_pid"], signal.SIGTERM)
except OSError:
def stop_managed_state(timeout, state, outcome):
if outcome == "stale":
cleanup_stale_files()
raise ProxyError("устаревшее состояние управляемого Xray очищено")
if outcome == "unverifiable":
raise unverifiable_state_error()
if outcome == "orphan":
if not signal_xray(state, signal.SIGTERM):
raise ProxyError("личность Xray-сироты изменилась; сигнал не отправлен")
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if not process_alive(state["xray_pid"]):
cleanup_stale_files()
print("Xray-сирота остановлен, служебные файлы удалены")
return 0
time.sleep(0.05)
if not signal_xray(state, signal.SIGKILL):
raise ProxyError("личность Xray-сироты изменилась; SIGKILL не отправлен")
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if not process_alive(state["xray_pid"]):
cleanup_stale_files()
print("Xray-сирота остановлен, служебные файлы удалены")
return 0
time.sleep(0.05)
raise ProxyError("остановка Xray-сироты после SIGKILL превысила таймаут")
check_endpoint(state["address"], state["port"])
if not signal_supervisor(state, signal.SIGTERM):
raise ProxyError("не удалось послать сигнал подтверждённому супервизору")
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
@@ -930,6 +1174,21 @@ def command_stop():
raise ProxyError("штатная остановка управляемого Xray превысила таймаут")
def command_stop():
timeout = stop_timeout()
state = read_managed_state()
outcome = managed_state_outcome(state)
if outcome in ("stale", "orphan"):
lock = acquire_start_lock()
try:
state = read_managed_state()
outcome = managed_state_outcome(state)
return stop_managed_state(timeout, state, outcome)
finally:
lock.close()
return stop_managed_state(timeout, state, outcome)
def build_parser():
parser = argparse.ArgumentParser(
prog="xray-proxy",