From 5ace0cfae0195f88a636c70fe975aa5c74ad805a Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Mon, 20 Jul 2026 07:02:33 -0400 Subject: [PATCH] =?UTF-8?q?feat(xray-proxy):=20=D0=B4=D0=BE=D0=B1=D0=B0?= =?UTF-8?q?=D0=B2=D0=BB=D0=B5=D0=BD=20=D0=B1=D0=B5=D0=B7=D0=BE=D0=BF=D0=B0?= =?UTF-8?q?=D1=81=D0=BD=D1=8B=D0=B9=20=D0=B6=D0=B8=D0=B7=D0=BD=D0=B5=D0=BD?= =?UTF-8?q?=D0=BD=D1=8B=D0=B9=20=D1=86=D0=B8=D0=BA=D0=BB=20=D1=83=D0=BF?= =?UTF-8?q?=D1=80=D0=B0=D0=B2=D0=BB=D1=8F=D0=B5=D0=BC=D0=BE=D0=B3=D0=BE=20?= =?UTF-8?q?Xray?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - 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 --- .../issues/03-safe-process-lifecycle.md | 150 +++++-- .scratch/headless-xray-proxy/token-ledger.md | 22 ++ .../2026-07-15-headless-xray-proxy-design.md | 32 +- tests/test_xray_proxy.py | 84 ++++ tests/test_xray_proxy_managed.py | 335 +++++++++++++++- xray-proxy/.local/bin/xray-proxy | 367 +++++++++++++++--- 6 files changed, 890 insertions(+), 100 deletions(-) diff --git a/.scratch/headless-xray-proxy/issues/03-safe-process-lifecycle.md b/.scratch/headless-xray-proxy/issues/03-safe-process-lifecycle.md index f02ac5f..9433100 100644 --- a/.scratch/headless-xray-proxy/issues/03-safe-process-lifecycle.md +++ b/.scratch/headless-xray-proxy/issues/03-safe-process-lifecycle.md @@ -1,36 +1,132 @@ # 03 — Безопасный жизненный цикл управляемого процесса **What to build:** Управляемый Xray предсказуемо переживает повторные команды, -сигналы, устаревшее состояние и аварийное исчезновение супервизора, не затрагивая -чужие процессы и не оставляя секретные или временные данные. +сигналы, устаревшее состояние и аварийное исчезновение супервизора, не +затрагивая чужие процессы и не оставляя секретных данных вне закрытого +служебного каталога. **Blocked by:** 02 — Управляемый Xray: полный штатный путь; 06 — Упрощение. -**Status:** ready-for-agent +**Status:** done -**Мандат идентичности:** идентичность = PID + время старта (`/proc/PID/stat`) + -исполняемый файл ребёнка / путь скрипта супервизора; текущий argv-обход заменить +**Резолюция (2026-07-20):** реализовано и принято. Дельта: CLI 976→1235, +тесты 437→521 + 799→1112; сьюты: 49 unittest OK + 8 shell PASS (baseline +34/8). Конвейер: spec-ревью постановки (NEEDS_REWORK, 7 MAJOR + 3 MINOR — +постановка переписана до реализации) → реализация на medium → саморевью +(3 находки, починены) → слепые ревью: TASK 1 MINOR + 2 NIT, CODE 4 MAJOR + +1 MINOR (гонки идентичности/очистки) → триаж: FIXED 8, REJECTED 0 → обе +перепроверки FIXES_CONFIRMED. + +## Модель идентичности процессов + +Идентичность определяется отдельно для ребёнка-Xray и для супервизора; +«обход argv в поисках любого совпадения» (`verify_managed_state`) заменяется этим механизмом. -- [ ] Супервизор владеет пользовательской блокировкой, допускает только один - управляемый экземпляр и не заменяет запущенный профиль другим без явных - `stop` и нового `start`. -- [ ] Состояние с правами `0600` хранит PID и время старта супервизора и Xray, - путь к бинарнику и порт, но не ссылку или её секреты; служебные каталоги имеют - права `0700`. -- [ ] `status` и `stop` сверяют PID со временем старта и ожидаемым исполняемым - файлом, поэтому чужой, устаревший и повторно использованный PID никогда не - считается управляемым процессом. -- [ ] SIGINT, SIGTERM и SIGHUP передаются дочернему Xray; штатное завершение - удаляет состояние, сгенерированный JSON и временные файлы. -- [ ] Если супервизор аварийно исчез, `stop` завершает оставшийся Xray только - после полной проверки его личности и затем безопасно очищает состояние. -- [ ] Повторный запуск, смена профиля, конфликт локального порта и несогласованный - машинный адрес завершаются понятным отказом без повреждения уже работающего - экземпляра. -- [ ] Полная VLESS-ссылка и её секреты отсутствуют в stdout, stderr, служебных - файлах и аргументах дочерних процессов; временная конфигурация существует - только внутри закрытого служебного каталога. -- [ ] Автоматические тесты покрывают блокировку, повторный запуск, сигналы, - остановку, очистку, смену профиля, конфликт порта, чужой, устаревший и - повторно использованный PID, а также исчезнувший супервизор. +- **Общее для обоих:** PID + время старта. Время старта — поле 22 + (`starttime`) из `/proc/PID/stat`, целое число тиков с момента загрузки; + хранится в state как есть и сравнивается на точное равенство. Парсер + обязан сначала изолировать поле `comm` по последней `)` (comm может + содержать пробелы и скобки) и только потом индексировать поля; любая + ошибка чтения/разбора трактуется как несовпадение (fail-closed). +- **Ребёнок Xray:** канонический путь `/proc/PID/exe` (realpath) совпадает + с сохранённым `xray_path`. +- **Супервизор:** его exe — интерпретатор Python, поэтому проверяется + иначе: в state сохраняется канонический путь скрипта CLI + (`supervisor_path`, realpath на момент старта); при проверке операнд + скрипта из `/proc/PID/cmdline` резолвится (относительные пути — через + `/proc/PID/cwd`) и сравнивается с `supervisor_path`. Поддерживаемые формы + запуска (относительный путь, PATH, симлинк) не ломаются — сравнение + идёт по realpath. +- **Границы загрузки:** в state сохраняется `boot_id` + (`/proc/sys/kernel/random/boot_id`). Несовпадение с текущим означает + устаревшее состояние: записанные PID трактуются как мёртвые, сигналы не + посылаются. +- **Гонка check-then-kill:** непосредственно перед каждым `os.kill` + идентичность перечитывается заново; несовпадение — отказ без сигнала. + Остаточная гонка «процесс умер между перечиткой и kill» принята как риск + модели доверенных пользователей (pidfd не используем: Python 3.8+, + переносимость). + +## Чекбоксы + +- [x] Взаимное исключение: супервизор на всё время жизни держит + неблокирующий эксклюзивный `fcntl.flock` на lock-файле в служебном + каталоге. Второй одновременный `start` завершается понятным отказом, не + запуская процессов и не трогая чужие state/конфиг. `state.json` остаётся + данными для восстановления после сбоя, а не механизмом блокировки. + Последовательные повторный `start` и смена профиля при живом экземпляре + по-прежнему отвергаются. +- [x] Схема state (права 0600, каталоги 0700) — ровно эти ключи: + `supervisor_pid`, `xray_pid`, `xray_path`, `config_path`, `address`, + `port` (существующие) + `supervisor_path`, `supervisor_starttime`, + `xray_starttime`, `boot_id` (новые). Ссылка VLESS и её секреты в state + не попадают. +- [x] `status` и `stop` применяют модель идентичности выше; чужой, + устаревший или переиспользованный PID не признаётся управляемым + процессом (в пределах принятой остаточной гонки). +- [x] Три исхода проверки состояния, у каждого — своё поведение: + 1. **Здоровая пара** (супервизор и ребёнок живы, идентичность + подтверждена): `stop` сигналит супервизору, как сейчас. + 2. **Сирота** (супервизор исчез, ребёнок жив и подтверждён по + PID + starttime + boot_id + exe): `stop` сигналит ребёнку напрямую и + удаляет state и сгенерированный конфиг; доступность порта для этого + НЕ требуется. `status` — отдельная ненулевая диагностика «сирота»; + `env` и `run` — отказ. + 3. **Непроверяемое/устаревшее состояние:** если записанный PID жив, но + идентичность не подтверждена — никаких сигналов, state сохраняется, + диагностика с инструкцией ручного удаления; если записанные PID + мертвы (или boot_id не совпал) — безопасная очистка устаревших + файлов разрешена. +- [x] Конфликт локального порта: `start` при уже занятом порте завершается + ненулевым кодом, диагностика называет порт занятым/недоступным, чужой + слушатель не затронут, свои сгенерированные файлы убраны. Инспекция + владения слушателем (/proc/net/tcp) запрещена. +- [x] Секреты: полная VLESS-ссылка никуда не персистится; UUID, ключи, + секретоподобные значения отсутствуют в stdout, stderr, argv, + `state.json` и lock-файле. Единственное допустимое место — `client.json` + с правами 0600 внутри каталога 0700; он удаляется при штатном завершении + и при `stop` сироты. +- [x] Автоматические тесты (без root, на обычной dev-машине): + - блокировка: удержанный извне flock → `start` отказывает, ничего не + запустив и не изменив; + - переиспользованный PID моделируется правкой сохранённого `starttime` + у живого процесса (реальную гонку PID не воспроизводим); отдельно — + мёртвый PID, неверный исполняемый файл, несовпавший `boot_id`; + - SIGKILL супервизора → `stop` добивает подтверждённого сироту и чистит + state/конфиг; + - занятый порт (заранее открытый сокет) → контракт из чекбокса выше; + - юнит-тест парсера `/proc/PID/stat` на синтетических `comm` с + пробелами и `)`; + - фикстура фейкового Xray адаптирована под проверку `/proc/PID/exe` + (ребёнок должен быть настоящим исполняемым файлом — например, + `sys.executable` как «бинарник xray» со скриптами-операндами; детали + на усмотрение исполнителя); + - все сохраняемые ниже поведения остаются зелёными. + +## Сохранить, не переписывать + +Уже реализовано и покрыто тестами в 02/06 — не переделывать, только не +сломать: проброс SIGINT/SIGTERM/SIGHUP ребёнку; очистка state и конфига при +штатном завершении; права 0600/0700; последовательные отказы (повторный +start, смена профиля, несовпадение машинного адреса/порта со state); +loopback-only; секреты вне stdout/stderr/argv; foreground-супервизор; +fail-fast `pclaude`/`pcodex`; внешний путь без машинного конфига. + +Запрещено восстанавливать срезанное в 06: владение слушателем через +`/proc/net/tcp`, `profile_fingerprint`, инспекцию zombie-состояний, +SIGALRM-машинерию, поддержку второй версии Xray. + +## Не трогать + +- `bash/.bash_functions`, установщик, README — вне объёма. +- Никаких коммитов. + +## Отчёт исполнителя + +1. Изменённые файлы со строкой «зачем» на каждый. +2. Команды тестов и счётчики (`python3 -m unittest discover -s tests`, + `bash tests/test_proxy_launchers.sh`; baseline: 34 OK / 8 PASS — новых + станет больше, указать числа). +3. Чекбоксы: закрыт / не закрыт и почему. +4. Принятые по ходу решения и сомнения, если есть. diff --git a/.scratch/headless-xray-proxy/token-ledger.md b/.scratch/headless-xray-proxy/token-ledger.md index 37258bf..372e42d 100644 --- a/.scratch/headless-xray-proxy/token-ledger.md +++ b/.scratch/headless-xray-proxy/token-ledger.md @@ -27,6 +27,28 @@ output/reasoning), кросс-сравнение — ориентировочн _Заполняется по ходу; сравнивать построчно с 06._ +| Этап | Агент / модель | Effort | Токены | Прочее | +|---|---|---|---|---| +| 03: ревью постановки | Codex gpt-5.6-sol | high | in 1 126 384 (кэш 1 044 480 → свежих ~82k), out 15 830 (reasoning 9 567) | NEEDS_REWORK: 7 MAJOR + 3 MINOR | +| 03: реализация | Codex gpt-5.6-sol | **medium** | in 5 925 714 (кэш 5 678 080 → свежих ~248k), out 29 512 (reasoning 11 251)* | 88+6 команд; +657/−89; 44 OK + 8 PASS; процесс убит рестартом сессии перед отчётом, дописан через resume | + +| 03: саморевью | Codex (resume) | medium | инкремент: свежих in ~53k, out ~11.6k (reasoning ~7k) | 38 команд; found 3 fixed 3; 45 OK + 8 PASS | +| 03: слепое ревью «по задаче» | Claude Opus (субагент) | default | 73 131 суммарно | 21 tool uses, 6.3 мин; 1 MINOR + 2 NIT | +| 03: слепое ревью «по коду» | Codex gpt-5.6-sol (свежий тред) | high | in 1 103 725 (кэш 1 019 136 → свежих ~85k), out 12 797 (reasoning 8 908) | 28 команд; 4 MAJOR + 1 MINOR | +| 03: триаж/починка | Codex (resume) | medium | инкремент: свежих in ~55.5k, out ~16.9k (reasoning ~7.8k) | 46 команд; FIXED 8, REJECTED 0; 49 OK + 8 PASS | +| 03: перепроверка TASK-1..3 | Claude Opus (резюм субагента) | default | инкремент ~7.4k (суммарно 80 560) | 8 tool uses, 1.7 мин; FIXES_CONFIRMED | +| 03: перепроверка CODE-1..5 | Codex-ревьюер (resume) | medium | инкремент: свежих in ~112k, out ~3.2k | 20 команд; FIXES_CONFIRMED | + +Итог 03 (реализация на MEDIUM): Codex — свежий input ~636k + output ~77k +(ревью постановки 82k/16k, реализация ~248k/29.5k, саморевью 53k/11.6k, +CODE-ревью 85k/12.8k, триаж 55.5k/16.9k, перепроверка 112k/3.2k); +Claude-доля — 80.5k суммарно (TASK-ревью + перепроверка). Против 06 +(high-baseline: ~735k + 97k Codex, 175k Claude) — дешевле, но качество +ниже: слепые ревью нашли 4 MAJOR (у 06 — 1), все — гонки +конкурентности/восстановления; конвейер их выловил и починил. + +\* usage первого (убитого) хода не был зафиксирован (нет `turn.completed`); цифры — из resume-хода, по масштабу покрывают весь тред. + ## Наблюдения - Кэш у Codex покрывает ~93% входа ревью-этапа: длинные треды с большим diff --git a/docs/specs/2026-07-15-headless-xray-proxy-design.md b/docs/specs/2026-07-15-headless-xray-proxy-design.md index 2e3d6b9..f1fe7f8 100644 --- a/docs/specs/2026-07-15-headless-xray-proxy-design.md +++ b/docs/specs/2026-07-15-headless-xray-proxy-design.md @@ -306,14 +306,30 @@ CLI предоставляет `install`, `start`, `stop`, `status`, `env` и `r останавливает Xray. Отдельный постоянный журнал и команда `logs` не нужны: достаточно истории tmux. -Супервизор владеет блокировкой, дочерним процессом и служебным каталогом -`~/.local/state/xray-proxy/`. Файл состояния содержит PID и время старта обоих -процессов, путь к бинарнику и порт, но не VLESS-ссылку и не её секреты. `status` -и `stop` сверяют PID со временем старта и -ожидаемым исполняемым файлом, поэтому повторное использование PID не считается -управляемым процессом. `stop` посылает SIGTERM супервизору, а тот завершает Xray -и очищает состояние. Если супервизор аварийно исчез, `stop` может завершить -только дочерний Xray, идентичность которого подтверждена теми же данными. +Супервизор на всё время работы удерживает неблокирующую эксклюзивную блокировку +`fcntl.flock` в служебном каталоге `~/.local/state/xray-proxy/`. Второй `start` +отказывает до запуска процессов и изменения state или сгенерированного +конфига. `state.json` нужен для восстановления после сбоя, но не служит +блокировкой. + +Файл состояния содержит ровно `supervisor_pid`, `xray_pid`, `xray_path`, +`config_path`, `address`, `port`, `supervisor_path`, `supervisor_starttime`, +`xray_starttime` и `boot_id`. Время старта — поле 22 `/proc/PID/stat`; поле +`comm` сначала отделяется по последней `)`. Для обоих процессов проверяются PID, +точное время старта и `boot_id`. У ребёнка дополнительно сверяется канонический +`/proc/PID/exe` с `xray_path`. У Python-супервизора операнд скрипта из +`/proc/PID/cmdline` разрешается относительно `/proc/PID/cwd` и сравнивается с +каноническим `supervisor_path`. Ошибка чтения или разбора означает несовпадение. +Непосредственно перед каждым сигналом идентичность читается повторно. + +Проверка состояния различает три исхода. Для здоровой пары `stop` посылает +SIGTERM супервизору. Если супервизор исчез, но ребёнок подтверждён, `status` +сообщает о сироте с ненулевым кодом, `env` и `run` отказывают, а `stop` посылает +сигнал ребёнку напрямую без требования доступности порта и удаляет state и +конфиг. При живом, но неподтверждённом PID сигналы не посылаются, файлы +сохраняются, а диагностика требует ручной проверки и удаления. Если оба PID +мертвы или `boot_id` не совпадает, устаревшие state и конфиг можно безопасно +удалить без сигналов. У пользователя может быть только один управляемый экземпляр. Повторный `start` не заменяет его другим профилем. Для смены профиля нужны явные `stop` и `start`. diff --git a/tests/test_xray_proxy.py b/tests/test_xray_proxy.py index 88840b6..63e017a 100644 --- a/tests/test_xray_proxy.py +++ b/tests/test_xray_proxy.py @@ -1,10 +1,12 @@ #!/usr/bin/env python3 import json import os +import runpy import subprocess import sys import tempfile import unittest +from unittest import mock from pathlib import Path @@ -154,6 +156,88 @@ class XrayProxyCliTest(unittest.TestCase): self.assertIn("требует Python 3.8", result.stderr) self.assertIn("Python 3.7", result.stderr) + def test_proc_stat_parser_handles_spaces_and_closing_parentheses_in_comm(self): + module = runpy.run_path(str(CLI)) + parse_starttime = module["parse_proc_stat_starttime"] + tail = ["S"] + [str(value) for value in range(4, 22)] + ["987654"] + + for comm in ("worker", "worker name", "worker name ) with parens"): + with self.subTest(comm=comm): + self.assertEqual( + 987654, + parse_starttime("321 ({}) ".format(comm) + " ".join(tail)), + ) + for malformed in ("321 worker S 1 2", "321 (worker) S 1 2"): + with self.subTest(malformed=malformed): + with self.assertRaises(ValueError): + parse_starttime(malformed) + + def test_process_permission_error_is_not_treated_as_dead(self): + module = runpy.run_path(str(CLI)) + with mock.patch.object(module["Path"], "stat", side_effect=PermissionError()): + self.assertTrue(module["process_alive"](12345)) + + def test_same_pid_cannot_be_a_healthy_managed_pair(self): + module = runpy.run_path(str(CLI)) + function_globals = module["managed_state_outcome"].__globals__ + state = {"supervisor_pid": 12345, "xray_pid": 12345, "boot_id": "boot"} + replacements = { + "read_boot_id": mock.Mock(return_value="boot"), + "process_alive": mock.Mock(return_value=True), + "supervisor_identity_matches": mock.Mock(return_value=True), + "xray_identity_matches": mock.Mock(return_value=True), + } + with mock.patch.dict(function_globals, replacements): + self.assertEqual("unverifiable", module["managed_state_outcome"](state)) + + def test_identity_reads_starttime_after_process_paths(self): + module = runpy.run_path(str(CLI)) + function_globals = module["xray_identity_matches"].__globals__ + state = { + "supervisor_pid": 101, + "supervisor_starttime": 201, + "supervisor_path": "/cli", + "xray_pid": 102, + "xray_starttime": 202, + "xray_path": "/xray", + } + + xray_events = [] + xray_replacements = { + "process_executable": lambda _pid: xray_events.append("exe") or "/xray", + "process_starttime": lambda _pid: xray_events.append("stat") or 202, + } + with mock.patch.dict(function_globals, xray_replacements): + self.assertTrue(module["xray_identity_matches"](state)) + self.assertEqual(["exe", "stat"], xray_events) + + supervisor_events = [] + supervisor_replacements = { + "process_cmdline": lambda _pid: supervisor_events.append("cmdline") + or ["python", "/cli"], + "process_argument_resolves_to": lambda *_args: supervisor_events.append( + "script" + ) + or True, + "process_starttime": lambda _pid: supervisor_events.append("stat") or 201, + } + with mock.patch.dict(function_globals, supervisor_replacements): + self.assertTrue(module["supervisor_identity_matches"](state)) + self.assertEqual(["cmdline", "script", "stat"], supervisor_events) + + def test_initial_child_starttime_retries_while_child_is_alive(self): + module = runpy.run_path(str(CLI)) + wait_for_starttime = module["wait_for_child_starttime"] + function_globals = wait_for_starttime.__globals__ + child = mock.Mock(pid=12345) + child.poll.return_value = None + replacements = { + "process_starttime": mock.Mock(side_effect=[None, 7654321]), + "time": mock.Mock(), + } + with mock.patch.dict(function_globals, replacements): + self.assertEqual(7654321, wait_for_starttime(child)) + def test_broken_config_symlink_is_not_treated_as_missing_config(self): config_dir = self.config_home / "proxy" config_dir.mkdir() diff --git a/tests/test_xray_proxy_managed.py b/tests/test_xray_proxy_managed.py index 570253e..456c310 100644 --- a/tests/test_xray_proxy_managed.py +++ b/tests/test_xray_proxy_managed.py @@ -1,6 +1,8 @@ #!/usr/bin/env python3 +import fcntl import json import os +import shutil import signal import socket import subprocess @@ -22,8 +24,7 @@ PCS_A = "a" * 64 PCS_B = "b" * 64 -FAKE_XRAY = r'''#!/usr/bin/env python3 -import json +FAKE_XRAY_OPERAND = r'''import json import os import re import signal @@ -32,21 +33,22 @@ import sys import threading log_path = os.environ["FAKE_XRAY_LOG"] +operation = os.path.basename(sys.argv[0]) def record(kind, **values): values["kind"] = kind - values["argv"] = sys.argv[1:] - values["executable"] = os.path.realpath(sys.argv[0]) + values["argv"] = [operation] + sys.argv[1:] + values["executable"] = os.path.realpath("/proc/self/exe") with open(log_path, "a", encoding="utf-8") as stream: stream.write(json.dumps(values, sort_keys=True) + "\n") -if sys.argv[1:] == ["version"]: +if operation == "version" and not sys.argv[1:]: record("version") print(os.environ["FAKE_XRAY_VERSION"]) raise SystemExit(int(os.environ.get("FAKE_VERSION_EXIT", "0"))) -is_test = len(sys.argv) == 5 and sys.argv[1:4] == ["run", "-test", "-c"] -is_run = len(sys.argv) == 4 and sys.argv[1:3] == ["run", "-c"] +is_test = operation == "run" and len(sys.argv) == 4 and sys.argv[1:3] == ["-test", "-c"] +is_run = operation == "run" and len(sys.argv) == 3 and sys.argv[1] == "-c" if not (is_test or is_run): record("bad-argv") raise SystemExit(64) @@ -73,7 +75,7 @@ if os.environ["FAKE_XRAY_VERSION"].startswith("Xray 26.3.27"): ): raise SystemExit(65) -if sys.argv[2] == "-test": +if is_test: record("test", config=config) if os.environ.get("FAKE_TEST_FAIL") == "1": sys.stderr.write(os.environ.get("FAKE_TEST_SECRET", "test failed")) @@ -92,6 +94,8 @@ stopping = False def stop(signum, _frame): global stopping record("signal", signum=signum) + if signum == signal.SIGTERM and os.environ.get("FAKE_IGNORE_TERM") == "1": + return stopping = True signal.signal(signal.SIGTERM, stop) @@ -162,8 +166,10 @@ class ManagedXrayCliTest(unittest.TestCase): self.bin_dir.mkdir() self.log = self.root / "xray.log" self.xray = self.bin_dir / "xray" - self.xray.write_text(FAKE_XRAY, encoding="utf-8") + shutil.copy2(sys.executable, self.xray) self.xray.chmod(0o755) + for operand in ("version", "run"): + (self.root / operand).write_text(FAKE_XRAY_OPERAND, encoding="utf-8") self.environment = os.environ.copy() self.environment.update( { @@ -238,6 +244,7 @@ class ManagedXrayCliTest(unittest.TestCase): env.update(environment) return subprocess.run( [str(CLI)] + list(args), + cwd=str(self.root), text=True, capture_output=True, env=env, @@ -264,12 +271,25 @@ class ManagedXrayCliTest(unittest.TestCase): return [] return [json.loads(line) for line in self.log.read_text().splitlines()] + def managed_state_path(self): + return self.state_home / "xray-proxy/state.json" + + def read_state(self): + return json.loads(self.managed_state_path().read_text(encoding="utf-8")) + + def write_state(self, state): + self.managed_state_path().write_text( + json.dumps(state) + "\n", encoding="utf-8" + ) + self.managed_state_path().chmod(0o600) + def start(self, profile=None): args = [str(CLI), "start"] if profile is not None: args.extend(["-f", str(profile)]) process = subprocess.Popen( args, + cwd=str(self.root), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, @@ -373,10 +393,13 @@ class ManagedXrayCliTest(unittest.TestCase): process = self.start() state_path = self.state_home / "xray-proxy/state.json" config_path = self.state_home / "xray-proxy/client.json" + lock_path = self.state_home / "xray-proxy/lock" state = json.loads(state_path.read_text(encoding="utf-8")) config = json.loads(config_path.read_text(encoding="utf-8")) self.assertEqual(0o600, state_path.stat().st_mode & 0o777) self.assertEqual(0o600, config_path.stat().st_mode & 0o777) + self.assertEqual(0o600, lock_path.stat().st_mode & 0o777) + self.assertEqual("", lock_path.read_text(encoding="ascii")) self.assertEqual(0o700, state_path.parent.stat().st_mode & 0o777) self.assertEqual(str(self.xray.resolve()), state["xray_path"]) self.assertEqual("127.0.0.1", state["address"]) @@ -389,9 +412,20 @@ class ManagedXrayCliTest(unittest.TestCase): "config_path", "address", "port", + "supervisor_path", + "supervisor_starttime", + "xray_starttime", + "boot_id", }, set(state), ) + self.assertEqual(str(CLI.resolve()), state["supervisor_path"]) + self.assertIsInstance(state["supervisor_starttime"], int) + self.assertIsInstance(state["xray_starttime"], int) + self.assertEqual( + Path("/proc/sys/kernel/random/boot_id").read_text(encoding="ascii").strip(), + state["boot_id"], + ) state_text = state_path.read_text(encoding="utf-8") self.assert_no_secrets(state_text, profile_text) stream = config["outbounds"][0]["streamSettings"] @@ -415,6 +449,283 @@ class ManagedXrayCliTest(unittest.TestCase): argv_text = "\0".join(entry["argv"]) self.assert_no_secrets(argv_text, profile_text) + def test_held_lock_rejects_start_without_changing_managed_files(self): + profile_text = self.tcp_profile() + self.write_profile(profile_text) + state_dir = self.state_home / "xray-proxy" + state_dir.mkdir(parents=True, mode=0o700) + state_path = state_dir / "state.json" + config_path = state_dir / "client.json" + state_path.write_text("state sentinel\n", encoding="ascii") + config_path.write_text("config sentinel\n", encoding="ascii") + lock_path = state_dir / "lock" + with lock_path.open("w", encoding="ascii") as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + result = self.run_cli("start") + + self.assertNotEqual(0, result.returncode) + self.assertIn("блокиров", result.stderr) + self.assertEqual("state sentinel\n", state_path.read_text(encoding="ascii")) + self.assertEqual("config sentinel\n", config_path.read_text(encoding="ascii")) + self.assertEqual([], self.log_entries()) + self.assert_no_secrets( + result.stdout + result.stderr + lock_path.read_text(encoding="ascii"), + profile_text, + ) + + def test_stale_cleanup_waits_for_start_lock_and_preserves_files(self): + state_dir = self.state_home / "xray-proxy" + state_dir.mkdir(parents=True, mode=0o700) + state = { + "supervisor_pid": 999999998, + "xray_pid": 999999999, + "xray_path": str(self.xray.resolve()), + "config_path": str(state_dir / "client.json"), + "address": "127.0.0.1", + "port": self.port, + "supervisor_path": str(CLI.resolve()), + "supervisor_starttime": 1, + "xray_starttime": 1, + "boot_id": Path("/proc/sys/kernel/random/boot_id") + .read_text(encoding="ascii") + .strip(), + } + self.write_state(state) + config_path = state_dir / "client.json" + config_path.write_text("stale sentinel\n", encoding="ascii") + lock_path = state_dir / "lock" + with lock_path.open("w", encoding="ascii") as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + status = self.run_cli("status") + + self.assertNotEqual(0, status.returncode) + self.assertIn("блокиров", status.stderr) + self.assertTrue(self.managed_state_path().exists()) + self.assertEqual("stale sentinel\n", config_path.read_text(encoding="ascii")) + + def test_reused_pid_starttime_is_not_signalled_and_state_is_preserved(self): + profile_text = self.tcp_profile() + self.write_profile(profile_text) + process = self.start() + original = self.read_state() + changed = dict(original) + changed["xray_starttime"] += 1 + self.write_state(changed) + marker = self.root / "reused-pid-child-ran" + + for command in ( + ("status",), + ("env",), + ("run", "--", "touch", str(marker)), + ("stop",), + ): + with self.subTest(command=command): + result = self.run_cli(*command) + self.assertNotEqual(0, result.returncode) + self.assertIn("вручную", result.stderr) + self.assertTrue(self.managed_state_path().exists()) + self.assertFalse(marker.exists()) + self.assertTrue(process.poll() is None) + + self.write_state(original) + stdout, stderr = self.stop_and_collect(process) + self.assert_no_secrets(stdout + stderr, profile_text) + + def test_wrong_xray_executable_is_not_signalled_and_state_is_preserved(self): + profile_text = self.tcp_profile() + self.write_profile(profile_text) + process = self.start() + original = self.read_state() + changed = dict(original) + changed["xray_path"] = str(Path("/bin/false").resolve()) + self.write_state(changed) + + stopped = self.run_cli("stop") + self.assertNotEqual(0, stopped.returncode) + self.assertIn("вручную", stopped.stderr) + self.assertTrue(self.managed_state_path().exists()) + self.assertIsNone(process.poll()) + + self.write_state(original) + stdout, stderr = self.stop_and_collect(process) + self.assert_no_secrets(stdout + stderr, profile_text) + + def test_boot_id_mismatch_cleans_stale_files_without_signalling_processes(self): + profile_text = self.tcp_profile() + self.write_profile(profile_text) + process = self.start() + state = self.read_state() + state["boot_id"] = "00000000-0000-0000-0000-000000000000" + self.write_state(state) + os.kill(state["supervisor_pid"], signal.SIGKILL) + process.wait(timeout=2) + + status = self.run_cli("status") + self.assertNotEqual(0, status.returncode) + self.assertIn("устаревшее", status.stderr) + self.assertFalse(self.managed_state_path().exists()) + self.assertFalse((self.state_home / "xray-proxy/client.json").exists()) + self.assertTrue(Path("/proc/{}".format(state["xray_pid"])).exists()) + + os.kill(state["xray_pid"], signal.SIGKILL) + stdout, stderr = process.communicate(timeout=4) + self.assert_no_secrets(stdout + stderr, profile_text) + + def test_dead_pids_allow_stale_state_and_config_cleanup(self): + state_dir = self.state_home / "xray-proxy" + state_dir.mkdir(parents=True, mode=0o700) + state = { + "supervisor_pid": 999999998, + "xray_pid": 999999999, + "xray_path": str(self.xray.resolve()), + "config_path": str(state_dir / "client.json"), + "address": "127.0.0.1", + "port": self.port, + "supervisor_path": str(CLI.resolve()), + "supervisor_starttime": 1, + "xray_starttime": 1, + "boot_id": Path("/proc/sys/kernel/random/boot_id") + .read_text(encoding="ascii") + .strip(), + } + self.write_state(state) + (state_dir / "client.json").write_text("stale\n", encoding="ascii") + + status = self.run_cli("status") + self.assertNotEqual(0, status.returncode) + self.assertIn("устаревшее", status.stderr) + self.assertFalse(self.managed_state_path().exists()) + self.assertFalse((state_dir / "client.json").exists()) + + def test_sigkill_supervisor_stop_terminates_confirmed_orphan_and_cleans(self): + profile_text = self.tcp_profile() + self.write_profile(profile_text) + process = self.start() + state = self.read_state() + run_count = len( + [entry for entry in self.log_entries() if entry["kind"] == "run"] + ) + alternate_profile = self.write_profile(self.xhttp_profile(), explicit=True) + repeated = self.run_cli("start") + switched = self.run_cli("start", "-f", str(alternate_profile)) + for result in (repeated, switched): + self.assertNotEqual(0, result.returncode) + self.assertIn("блокиров", result.stderr) + self.assertEqual( + run_count, + len([entry for entry in self.log_entries() if entry["kind"] == "run"]), + ) + + lock_path = self.state_home / "xray-proxy/lock" + with lock_path.open("r+", encoding="ascii") as lock: + with self.assertRaises(OSError): + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + + os.kill(state["supervisor_pid"], signal.SIGKILL) + process.wait(timeout=2) + with lock_path.open("r+", encoding="ascii") as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + fcntl.flock(lock, fcntl.LOCK_UN) + + status = self.run_cli("status") + env = self.run_cli("env") + marker = self.root / "orphan-child-ran" + run = self.run_cli("run", "--", "touch", str(marker)) + for result in (status, env, run): + self.assertNotEqual(0, result.returncode) + self.assertIn("сирот", result.stderr) + self.assertFalse(marker.exists()) + + stopped = self.run_cli("stop") + self.assertEqual(0, stopped.returncode, stopped.stderr) + self.assertIn("сирот", stopped.stdout) + self.assertFalse(self.managed_state_path().exists()) + self.assertFalse((self.state_home / "xray-proxy/client.json").exists()) + stdout, stderr = process.communicate(timeout=2) + self.assert_no_secrets(stdout + stderr + stopped.stdout, profile_text) + + self.port = self.free_port() + self.write_machine_config("xray") + replacement = self.start() + replacement_stdout, replacement_stderr = self.stop_and_collect(replacement) + self.assert_no_secrets( + replacement_stdout + replacement_stderr, profile_text + ) + + def test_unverifiable_live_child_preserves_state_and_config(self): + profile_text = self.tcp_profile() + self.write_profile(profile_text) + self.environment["XRAY_PROXY_STOP_TIMEOUT"] = "0.2" + process = self.start() + state = self.read_state() + self.xray.unlink() + + os.kill(state["supervisor_pid"], signal.SIGTERM) + process.wait(timeout=3) + state_exists = self.managed_state_path().exists() + config_path = self.state_home / "xray-proxy/client.json" + config_exists = config_path.exists() + child_alive = Path("/proc/{}".format(state["xray_pid"])).exists() + if child_alive: + os.kill(state["xray_pid"], signal.SIGKILL) + deadline = time.monotonic() + 2 + while ( + Path("/proc/{}".format(state["xray_pid"])).exists() + and time.monotonic() < deadline + ): + time.sleep(0.02) + stdout, stderr = process.communicate(timeout=2) + + self.assertTrue(child_alive) + self.assertTrue(state_exists) + self.assertTrue(config_exists) + self.assertNotEqual(0, process.returncode) + self.assertIn("вручную", stderr) + self.assert_no_secrets(stdout + stderr, profile_text) + + def test_orphan_ignoring_sigterm_is_rechecked_killed_and_cleaned(self): + profile_text = self.tcp_profile() + self.write_profile(profile_text) + self.environment["FAKE_IGNORE_TERM"] = "1" + process = self.start() + state = self.read_state() + os.kill(state["supervisor_pid"], signal.SIGKILL) + process.wait(timeout=2) + + stopped = self.run_cli( + "stop", environment={"XRAY_PROXY_STOP_TIMEOUT": "0.2"} + ) + self.assertEqual(0, stopped.returncode, stopped.stderr) + self.assertFalse(self.managed_state_path().exists()) + self.assertFalse((self.state_home / "xray-proxy/client.json").exists()) + signals = [ + entry["signum"] + for entry in self.log_entries() + if entry["kind"] == "signal" + ] + self.assertEqual([signal.SIGTERM], signals) + stdout, stderr = process.communicate(timeout=2) + self.assert_no_secrets(stdout + stderr + stopped.stdout, profile_text) + + def test_occupied_port_fails_without_touching_listener_or_leaving_generated_files(self): + profile_text = self.tcp_profile() + self.write_profile(profile_text) + with socket.socket() as listener: + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind(("127.0.0.1", self.port)) + listener.listen(1) + result = self.run_cli("start") + with socket.create_connection(("127.0.0.1", self.port), timeout=1): + pass + + self.assertNotEqual(0, result.returncode) + self.assertIn("порт {}".format(self.port), result.stderr) + self.assertIn("занят", result.stderr) + self.assertEqual([], self.log_entries()) + self.assertFalse(self.managed_state_path().exists()) + self.assertFalse((self.state_home / "xray-proxy/client.json").exists()) + self.assert_no_secrets(result.stdout + result.stderr, profile_text) + def test_versions_and_profile_failures_stop_before_test_or_run_without_leaks(self): invalid_versions = ( "Xray 25.12.8", @@ -556,6 +867,7 @@ class ManagedXrayCliTest(unittest.TestCase): config.write_text('{"value":' + constant + '}\n', encoding="utf-8") result = subprocess.run( [str(self.xray), "run", "-test", "-c", str(config)], + cwd=str(self.root), text=True, capture_output=True, env=self.environment, @@ -576,6 +888,7 @@ class ManagedXrayCliTest(unittest.TestCase): self.write_machine_config("xray") process = subprocess.Popen( [str(CLI), "start"], + cwd=str(self.root), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, @@ -744,8 +1057,8 @@ class ManagedXrayCliTest(unittest.TestCase): proxy_bin.mkdir() (proxy_bin / "xray-proxy").symlink_to(CLI) invocations = ( - ([str(CLI)], REPO_ROOT, self.environment), - ([os.path.relpath(str(CLI), str(REPO_ROOT))], REPO_ROOT, self.environment), + ([str(CLI)], self.root, self.environment), + ([os.path.relpath(str(CLI), str(self.root))], self.root, self.environment), ( ["xray-proxy"], self.root, diff --git a/xray-proxy/.local/bin/xray-proxy b/xray-proxy/.local/bin/xray-proxy index 4eee61f..35d9c44 100755 --- a/xray-proxy/.local/bin/xray-proxy +++ b/xray-proxy/.local/bin/xray-proxy @@ -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",