Compare commits

..
4 Commits
Author SHA1 Message Date
ddadmin 203e515479 feat(proxy-client): добавлен мастер установки системной службы
- Зачем: установка с нуля и переход из ручного Xray одной командой.
- Что: план с подтверждением или --yes, скрытый ввод ссылки, проверка Xray,
  перенос настроек, запуск службы и переключение домашнего профиля;
  безопасное продолжение после сбоя и сохранение существующих настроек.
- Проверка: cross-review Claude и исправление находок; по 108 тестов на
  Python 3.8 и 3.14 под PTY, 18 Bash-проверок. После аудита уточнена
  фикстура повторного запуска и отдельно проверена на обеих версиях Python.
  Настоящие службы и настройки машины не изменялись.
2026-11-04 21:15:23 +01:00
ddadmin 45938cde98 refactor(proxy-client): введён явный контекст запуска
Выбор путей, прав профиля, вывода Xray и подсказки остановки перенесён из глобальных переменных в неизменяемый RuntimeContext. Команды и управление процессом получают контекст явно.

Проверка: прежние 72 теста без изменений прошли на Python 3.8 и 3.14; оба shell-набора прошли.
2026-11-03 07:17:23 +01:00
ddadmin 384b4f215f fix(proxy-client): исправлено восстановление установки службы
Заданы права системных каталогов независимо от umask, добавлен журнал незавершённой установки и исправлена подсказка прав профиля. Проверены явные флаги systemd через подставные API ОС.

Проверка: 72 теста на Python 3.8 и 3.14, оба shell-набора. Настоящий boot/systemd остаётся непроверенным.
2026-11-02 22:30:43 +01:00
ddadmin b34eb6ac4c feat(proxy-client): добавлены профили и системная служба
Добавлены независимые пользовательские профили, их выбор в Bash-командах и явная установка службы systemd. Инструкции размещены в README пакета.

Проверка: 67 тестов на Python 3.8 и 3.14, shell-наборы и интеграция Bash с CLI. Загрузка ОС и восстановление настоящей службы требуют отдельной проверки.
2026-11-02 22:16:12 +01:00
10 changed files with 2245 additions and 91 deletions

No files matched your search

+10
View File
@@ -0,0 +1,10 @@
# Термины dotfiles
## Прокси
**Профиль прокси** — именованный набор настроек одного прокси: его тип,
адрес и порт для приложений, исключения и, для управляемого Xray, VLESS-ссылка.
Не следует называть профилем только VLESS-ссылку.
**VLESS-ссылка** — параметры подключения к удалённому серверу VLESS.
Она не задаёт локальный порт прокси и исключения для приложений.
@@ -4,6 +4,9 @@
Статус: реализовано; 2026-09-28 упрощено по принципу ниже Статус: реализовано; 2026-09-28 упрощено по принципу ниже
Расширение от 2026-11-02:
[пользовательские профили и системная служба](2026-11-02-proxy-client-profiles-service-design.md).
Точное поведение, форматы, сообщения и команды описаны в коде, Точное поведение, форматы, сообщения и команды описаны в коде,
`proxy-client --help` и [`proxy-client/README.md`](../../proxy-client/README.md). `proxy-client --help` и [`proxy-client/README.md`](../../proxy-client/README.md).
Здесь — то, чего по коду не видно: задача, границы, выбор и его причины, риски. Здесь — то, чего по коду не видно: задача, границы, выбор и его причины, риски.
@@ -0,0 +1,223 @@
# Профили proxy-client и системная служба
Дата: 2026-11-02
Статус: реализовано; запуск после загрузки ОС и восстановление настоящей
службы systemd требуют отдельной проверки на Linux-ВМ.
## Задача и границы
На Linux-ВМ, компьютерах и ноутбуках нужен прокси, доступный после загрузки
без входа пользователя и ручного запуска. Приложения уже знают его адрес.
Одновременно пользователь должен иметь возможность запустить другой прокси
на другом порту в открытом терминале.
Это расширяет [первоначальный дизайн](2026-07-15-proxy-client-design.md):
добавляет системную службу и несколько пользовательских профилей.
Остальные ограничения и принцип личного инструмента сохраняются.
Термины определены в [CONTEXT.md](../../CONTEXT.md).
Не добавляем несколько системных экземпляров, фоновый пользовательский режим,
TUN, системную маршрутизацию, автоматический выбор прокси или поддержку Windows.
Для системного режима требуется systemd; ручной режим сохраняет независимость
от него. Пакет по-прежнему можно использовать отдельно от остальных dotfiles.
## Пользовательские профили
Профиль по умолчанию сохраняет нынешние файлы
`~/.config/proxy-client/config` и `default.url`. Отсутствие выбора профиля
не меняет поведения существующих команд, включая прежний внешний прокси
при отсутствии конфигурации. Учитываются существующие настройки XDG.
Дополнительные профили размещаются в
`~/.config/proxy-client/profiles/<имя>/`, каждый со своими `config` и
`default.url`. Имя `default` обозначает существующий профиль по умолчанию.
Неизвестное имя даёт ошибку, а не переход к другому профилю.
Один общий параметр выбора профиля действует на запуск, остановку, проверку,
вывод окружения и запуск дочерних команд. Существующий выбор файла VLESS-ссылки
через `start -f` остаётся разовым переопределением ссылки выбранного профиля.
Bash-команды без явного выбора продолжают использовать профиль по умолчанию.
Каждый профиль получает собственные сведения о процессе, блокировку и
сгенерированный JSON. Запуск и остановка одного не затрагивают остальные.
Два экземпляра могут работать одновременно, если их локальные порты различны.
Занятый порт даёт ошибку; чужой процесс не останавливается.
Ручной запуск остаётся на переднем плане, с выводом в терминал и остановкой
через Ctrl+C либо команду остановки выбранного профиля. Само обращение к прокси
через `run` или `env` ничего не запускает.
## Bash-команды
`proxy_on` без параметров выбирает профиль по умолчанию, с `--profile` —
именованный. Он меняет только окружение текущего терминала после успешной
проверки. `proxy_off` очищает переменные, но не останавливает процессы.
`myip` использует текущее окружение.
`pclaude` и `pcodex` принимают ведущий `--proxy-profile` для выбора профиля.
Без него остаётся профиль по умолчанию; остальные аргументы и код возврата
дочерней программы сохраняются. Обычный `--profile` относится к дочерней
программе. Bash-модуль передаёт выбор CLI и не дублирует правила профилей.
Выбор через `proxy_on` не меняет профиль по умолчанию: обычная команда
наследует окружение, а обёртка с префиксом `p` явно выбирает свой профиль.
Это различие должно быть показано в README пакета.
## Системная служба
Одна служба `proxy-client.service` читает один комплект настроек из
`/etc/proxy-client/`. Ей не нужны именованные профили или домашний каталог
пользователя. Исполняемые файлы устанавливаются в системное расположение,
без ссылок на пользовательский каталог исходников. Xray работает под отдельной
непривилегированной учётной записью и слушает только loopback.
Временные файлы процесса, блокировка и сгенерированный JSON находятся в
закрытом `/run/proxy-client/`. Секреты доступны только администратору и
служебной учётной записи. Пользовательские команды остановки не управляют
системным экземпляром: за него отвечает systemd.
Установка оформляется одной подкомандой `service install`. В терминале она
собирает недостающие настройки, показывает план и после подтверждения
выполняет полный переход на службу. `--yes` (`--auto-yes`) принимает план
без вопроса. Без терминала для нового подключения нужен закрытый `--file`;
секрет не передаётся в argv или окружении. В терминале применяется скрытый
ввод. Новый профиль получает адрес `127.0.0.1` и порт `12334` по умолчанию.
Для переноса используются сохранённые `config` и `default.url` выбранного
профиля; `--file` переопределяет ссылку. Xray ищется в `~/.local/bin`, затем
в `PATH` и системной установке. Если его нет, скачивается закреплённая версия
с проверкой SHA-256. Параметр `--xray` позволяет указать бинарник явно.
Полный переход состоит из подготовки снимка, проверки Xray, установки
системных файлов, остановки выбранного ручного процесса и запуска службы.
Домашний `config` меняется на `external` только после проверки активного
состояния службы и доступности её порта; прежний файл сохраняется рядом.
Остановка пользовательского процесса и изменение домашнего профиля
выполняются без повышения прав. Sudo применяется только к системной части.
Закрытый снимок `service-setup.json` в каталоге состояния профиля позволяет
продолжить незавершённый переход. Он содержит исходный домашний config,
подготовленный config, VLESS-ссылку и путь для ручного восстановления;
права файла `600`, каталога `700`. При повторе домашние и системные настройки
сверяются со снимком до остановки процессов. Изменённые настройки не
перезаписываются, исходный файл VLESS-ссылки тоже проверяется. Ошибка конфликта
указывает команду удаления снимка, чтобы начать заново с текущими настройками;
этот сброс не меняет сами настройки и процессы. После успеха снимок удаляется.
Ошибка запуска сохраняет
прежний домашний профиль; восстановление ручного процесса требует явной
команды, прежние соединения не восстанавливаются.
Наличие системных файлов само по себе не означает завершённый переход:
если домашний профиль ещё `xray`, мастер сверяет его с системным и завершает
перенос. Другой профиль не заменяет существующие системные настройки.
После завершённого перехода настройки службы самостоятельны. Обычное
обновление сохраняет их и включает перезапуск в подтверждаемый план;
домашний профиль не меняется и может отсутствовать.
`--install-only`, `--config-dir` и `--root`
сохраняют режим установки файлов без мастера. Он требует готовых настроек
и Xray. Флаги `--enable`, `--start`, `--restart` задают действия явно.
`--root` не допускает действий с systemd. В этом режиме установка сама
домашний профиль не меняет и ручной процесс не останавливает.
Без терминала и без явного выбора режима команда завершается с подсказкой.
Systemd отвечает за запуск, остановку и перезапуск после аварии. Явная
остановка должна оставлять службу остановленной до следующего запуска.
Если автозапуск включён, после перезагрузки она запускается снова.
Отсутствие внешней сети при загрузке не должно приводить к окончательному
отказу службы. Готовность локального порта и доступность удалённого маршрута
различаются: автозапуск не гарантирует успешность первого запроса любого
приложения. Для приложений, не умеющих повторять запрос, порядок запуска и
ожидание готовности потребуют отдельной настройки.
## Журналы и совместная работа
Вывод службы направляется в journald. Отдельный текстовый журнал и собственная
ротация не нужны. Хранение между загрузками зависит от настройки journald;
установщик проверяет её и сообщает результат, не меняя глобальную политику
журналирования машины. Секреты не должны попадать в журнал.
Если системная служба занимает порт прежнего пользовательского Xray, второй
запуск на этом порту завершается ошибкой. Чтобы использовать системный прокси
через пользовательские команды, подходит профиль типа `external`. Для второго
Xray нужен другой порт. Мастер меняет домашний профиль по подтверждённому
плану; режим установки только файлов его не меняет.
## Устройство модулей
Применяем `codebase-design`: выбор профиля и все его пути должны определяться
в одном месте, чтобы команды не повторяли правила размещения файлов.
Пути, правила доступа к профилю, вывод Xray и подсказка остановки собраны
в неизменяемый контекст запуска. Он создаётся при выборе пользовательского
или системного режима и явно передаётся функциям. Управление процессом
использует готовые значения, без глобального переключателя режима.
Системная установка скрывает размещение, права и взаимодействие с systemd
за небольшим интерфейсом; не копирует разбор VLESS и управление Xray.
Внутри модуль отделяет подготовку плана от его применения: план хранит
снимок настроек и исходного процесса, применение управляет порядком
действий и возобновлением. Проверки проходят через публичный CLI.
Внешние команды подменяются через `PATH`, терминальный ввод, сокеты и загрузка
архива — подменой функций стандартной библиотеки в тестовом `sitecustomize`.
Разбор настроек и работа с файлами выполняются настоящим кодом.
Не вводим обобщённую поддержку других менеджеров служб и дополнительные
слои ради возможного будущего использования. Реально различаются
пользовательское и системное размещение, а также способ управления процессом.
## Проверки
Согласованы три интерфейса для новых тестов TDD:
1. CLI пользовательских профилей: прежнее поведение без имени, выбор
именованного профиля, два одновременных процесса, независимая остановка,
отказ при занятом порте и правильное окружение дочерней команды.
2. Установка службы через CLI и управление через systemd: размещение файлов,
сохранение настроек при повторной установке, явное включение автозапуска,
запуск и остановка, восстановление после аварии и доступность журналов.
3. Bash-команды: выбор профиля, передача аргументов и кода возврата,
сохранение окружения при ошибке и отсутствие остановки процессов.
Тесты проверяют наблюдаемое поведение через эти интерфейсы, а не внутренние
функции и последовательности их вызовов. Идём по одному сценарию:
падающий тест, минимальная реализация, следующий сценарий.
Локальные проверки используют временные каталоги, синтетические профили и
подставные внешние программы. Это не доказывает работу настоящего systemd.
Окончательная проверка системного режима нужна в отдельной Linux-среде
с systemd: запуск до входа пользователя после загрузки, восстановление после
аварии, явная остановка и одновременная работа ручного экземпляра.
Изменение текущей машины или её перезагрузка не входят в проверку по умолчанию.
Проверка после cross-review 2026-11-04: полный набор под псевдотерминалом
прошёл на Python 3.8 и 3.14 — по 108 тестов, ещё 12 проверок Bash-команд
и 6 проверок пользовательского установщика.
Мастер проверен через CLI: настройка с нуля, перенос, обновление, отмена,
отказы и продолжение перехода. Скрытый ввод проверен в псевдотерминале.
Системные действия и скачивание в этих тестах подменены; настоящие службы
не запускались. Shellcheck в среде отсутствует.
При первоначальной реализации отдельно проверены Bash-команды вместе
с настоящим CLI и двумя локальными тестовыми портами. Unit прошёл статическую
проверку в изолированном корне с тестовыми базовыми target-файлами.
Эти результаты не заменяют проверку загрузки ОС.
## Документация и альтернативы
Подробные инструкции размещены в
[`proxy-client/README.md`](../../proxy-client/README.md): профили, системная
установка и обновление, управление, автозапуск, журналы и диагностика.
В корневом README достаточно существующей ссылки на пакет. Этот документ
хранит причины решений; рабочие команды и примеры живут в README пакета.
Пользовательская служба с linger позволила бы запускать прокси до входа,
но выбрано явное разделение системных и домашних настроек. Несколько
именованных системных служб пока не нужны. Запуск в терминале сохраняется
для дополнительных прокси, но сам по себе не решает задачу загрузки машины.
Правила systemd сверены с документацией
[каталогов службы](https://www.freedesktop.org/software/systemd/man/latest/systemd.exec.html#RuntimeDirectory=),
[перезапуска](https://www.freedesktop.org/software/systemd/man/latest/systemd.service.html#Restart=)
и [хранения журналов](https://www.freedesktop.org/software/systemd/man/latest/journald.conf.html#Storage=).
+15 -2
View File
@@ -32,8 +32,12 @@ _proxy_apply_env() {
} }
proxy_on() { proxy_on() {
if (( $# != 0 )) && { (( $# != 2 )) || [[ "$1" != --profile || -z "${2:-}" ]]; }; then
echo "Использование: proxy_on [--profile NAME]" >&2
return 2
fi
local output local output
if ! output="$(_proxy_cli env)" || ! _proxy_apply_env "$output"; then if ! output="$(_proxy_cli "$@" env)" || ! _proxy_apply_env "$output"; then
echo "Ошибка: прокси не включён" >&2 echo "Ошибка: прокси не включён" >&2
return 1 return 1
fi fi
@@ -53,7 +57,16 @@ _proxy_run() {
return 2 return 2
fi fi
shift shift
_proxy_cli run -- "$command_name" "$@" local -a profile_args=()
if [[ "${1:-}" == --proxy-profile ]]; then
if (( $# < 2 )) || [[ -z "$2" ]]; then
echo "Ошибка: --proxy-profile требует имя профиля" >&2
return 2
fi
profile_args=(--profile "$2")
shift 2
fi
_proxy_cli "${profile_args[@]}" run -- "$command_name" "$@"
} }
pclaude() { _proxy_run claude "$@"; } pclaude() { _proxy_run claude "$@"; }
+693 -76
View File
@@ -13,7 +13,9 @@ import argparse
import configparser import configparser
import fcntl import fcntl
import hashlib import hashlib
import getpass
import http.client import http.client
import io
import ipaddress import ipaddress
import json import json
import os import os
@@ -28,8 +30,11 @@ import tempfile
import time import time
import urllib.request import urllib.request
import uuid import uuid
import warnings
import zipfile import zipfile
from pathlib import Path from pathlib import Path
from dataclasses import dataclass
from typing import Optional
from urllib.parse import parse_qsl, unquote, urlsplit from urllib.parse import parse_qsl, unquote, urlsplit
@@ -120,23 +125,57 @@ class VlessProfile:
return self.query["security"] return self.query["security"]
def config_home(): @dataclass(frozen=True)
root = os.environ.get("XDG_CONFIG_HOME") class RuntimeContext:
return (Path(root).expanduser() if root else Path.home() / ".config") / "proxy-client" """Размещение и правила выбранного запуска; создаётся до выполнения команды."""
config_dir: Path
state_dir: Path
xray_path: Optional[Path]
profile_forbidden_mode: int
profile_repair: str
xray_output: Optional[int]
stop_hint: str
@property
def state_path(self):
return self.state_dir / "state.json"
@property
def generated_config_path(self):
return self.state_dir / "client.json"
def state_home(): def user_context(profile="default", home=None):
root = os.environ.get("XDG_STATE_HOME") if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}", profile):
base = Path(root).expanduser() if root else Path.home() / ".local/state" raise ProxyError("недопустимое имя профиля")
return base / "proxy-client" config_root = os.environ.get("XDG_CONFIG_HOME")
state_root = os.environ.get("XDG_STATE_HOME")
home = home if home is not None else Path.home()
config_dir = (Path(config_root).expanduser() if config_root else home / ".config") / "proxy-client"
state_dir = (Path(state_root).expanduser() if state_root else home / ".local/state") / "proxy-client"
stop_hint = "Ctrl+C или proxy-client stop"
if profile != "default":
config_dir = config_dir / "profiles" / profile
state_dir = state_dir / "profiles" / profile
if not (config_dir / "config").is_file():
raise ProxyError("неизвестный профиль: {}".format(profile))
stop_hint = "Ctrl+C или proxy-client --profile {} stop".format(profile)
return RuntimeContext(config_dir, state_dir, None, 0o077,
"chmod 600 {0}", None, stop_hint)
def state_path(): def service_context(arguments):
return state_home() / "state.json" config_dir = Path(arguments.config_dir)
state_dir = Path(arguments.runtime_dir)
xray_path = Path(arguments.xray)
def generated_config_path(): if not all(path.is_absolute() for path in (config_dir, state_dir, xray_path)):
return state_home() / "client.json" raise ProxyError("системные пути должны быть абсолютными")
if not (config_dir / "config").is_file():
raise ProxyError("нет системного config")
return RuntimeContext(config_dir, state_dir, xray_path, 0o037,
"sudo chown root:proxy-client {0}; sudo chmod 640 {0}",
subprocess.DEVNULL, "sudo systemctl stop proxy-client")
def validate_address(value): def validate_address(value):
@@ -157,8 +196,8 @@ def validate_address(value):
raise ProxyError("поле address должно быть IP-адресом или простым ASCII-именем") raise ProxyError("поле address должно быть IP-адресом или простым ASCII-именем")
def load_config(): def load_config(context):
path = config_home() / "config" path = context.config_dir / "config"
if not path.exists(): if not path.exists():
if path.is_symlink(): if path.is_symlink():
raise ProxyError("машинный конфиг {} ссылается на отсутствующий файл".format(path)) raise ProxyError("машинный конфиг {} ссылается на отсутствующий файл".format(path))
@@ -173,7 +212,10 @@ def load_config():
def parse_config(text): def parse_config(text):
parser = configparser.ConfigParser(interpolation=None, empty_lines_in_values=False) parser = configparser.ConfigParser(interpolation=None, empty_lines_in_values=False)
parser.read_string(text, source="config") try:
parser.read_string(text, source="config")
except configparser.Error:
raise ProxyError("неверный формат config; ожидается секция [proxy] с параметрами") from None
if parser.sections() != ["proxy"]: if parser.sections() != ["proxy"]:
raise ProxyError("нужна ровно одна секция [proxy]") raise ProxyError("нужна ровно одна секция [proxy]")
section = parser["proxy"] section = parser["proxy"]
@@ -225,9 +267,9 @@ def check_reachable(config):
) )
def check_proxy(config): def check_proxy(config, context):
if config.proxy_type == "xray": if config.proxy_type == "xray":
require_supervised_xray(config) require_supervised_xray(config, context)
check_reachable(config) check_reachable(config)
@@ -274,13 +316,15 @@ def write_private_json(path, value):
path.chmod(0o600) path.chmod(0o600)
def load_profile(path): def load_profile(path, context):
# Права проверяем у конечного файла: у симлинка они ничего не значат. # Права проверяем у конечного файла: у симлинка они ничего не значат.
resolved = path.expanduser().resolve(strict=True) resolved = path.expanduser().resolve(strict=True)
if resolved.stat().st_mode & 0o077: if resolved.stat().st_mode & context.profile_forbidden_mode:
target = shlex.quote(str(resolved))
repair = context.profile_repair.format(target)
raise ProxyError( raise ProxyError(
"файл профиля {} доступен другим пользователям; выполните: chmod 600 {}".format( "файл профиля {} доступен другим пользователям; выполните: {}".format(
resolved, shlex.quote(str(resolved)) resolved, repair
) )
) )
lines = [line.strip() for line in resolved.read_text(encoding="utf-8").splitlines()] lines = [line.strip() for line in resolved.read_text(encoding="utf-8").splitlines()]
@@ -531,8 +575,8 @@ def build_xray_config(profile, local_address, local_port):
} }
def resolve_xray(): def resolve_xray(context):
found = shutil.which("xray") found = str(context.xray_path) if context.xray_path is not None else shutil.which("xray")
if found is None: if found is None:
raise ProxyError("xray не найден в PATH; установите его: proxy-client install") raise ProxyError("xray не найден в PATH; установите его: proxy-client install")
return str(Path(found).resolve(strict=True)) return str(Path(found).resolve(strict=True))
@@ -607,7 +651,8 @@ def install_xray(destination, downloader):
shutil.rmtree(str(staging), ignore_errors=True) shutil.rmtree(str(staging), ignore_errors=True)
def command_install(replace=False, downloader=download_file): def command_install(replace=False, downloader=download_file, context=None):
context = context if context is not None else user_context()
destination = Path.home() / ".local/bin/xray" destination = Path.home() / ".local/bin/xray"
if destination.is_dir(): if destination.is_dir():
raise ProxyError("назначение {} является каталогом".format(destination)) raise ProxyError("назначение {} является каталогом".format(destination))
@@ -624,7 +669,7 @@ def command_install(replace=False, downloader=download_file):
else: else:
install_xray(destination, downloader) install_xray(destination, downloader)
print("Xray {} установлен: {}".format(XRAY_VERSION, destination)) print("Xray {} установлен: {}".format(XRAY_VERSION, destination))
if managed_xray_running(): if managed_xray_running(context):
print("Работающий Xray остался прежним; перезапустите proxy-client start") print("Работающий Xray остался прежним; перезапустите proxy-client start")
found = shutil.which("xray") found = shutil.which("xray")
if found is None or Path(found).resolve() != destination.resolve(): if found is None or Path(found).resolve() != destination.resolve():
@@ -651,8 +696,8 @@ def run_xray_test(path, config_path):
raise ProxyError("созданная конфигурация не прошла xray run -test") raise ProxyError("созданная конфигурация не прошла xray run -test")
def read_state(): def read_state(context):
path = state_path() path = context.state_path
try: try:
state = json.loads(path.read_text(encoding="utf-8")) state = json.loads(path.read_text(encoding="utf-8"))
keys = ("supervisor_pid", "xray_pid", "xray_path", "address", "port") keys = ("supervisor_pid", "xray_pid", "xray_path", "address", "port")
@@ -685,32 +730,32 @@ def supervisor_running(state):
return b"proxy-client" in cmdline return b"proxy-client" in cmdline
def remove_state_files(): def remove_state_files(context):
for path in (state_path(), generated_config_path()): for path in (context.state_path, context.generated_config_path):
try: try:
path.unlink() path.unlink()
except FileNotFoundError: except FileNotFoundError:
pass pass
def managed_state(): def managed_state(context):
"""Состояние работающего Xray; устаревшие служебные файлы удаляет.""" """Состояние работающего Xray; устаревшие служебные файлы удаляет."""
state = read_state() state = read_state(context)
if not xray_running(state): if not xray_running(state):
remove_state_files() remove_state_files(context)
raise ProxyError("управляемый Xray не запущен; устаревшие служебные файлы удалены") raise ProxyError("управляемый Xray не запущен; устаревшие служебные файлы удалены")
return state return state
def managed_xray_running(): def managed_xray_running(context):
try: try:
return xray_running(read_state()) return xray_running(read_state(context))
except ProxyError: except ProxyError:
return False return False
def require_supervised_xray(config): def require_supervised_xray(config, context):
state = managed_state() state = managed_state(context)
if not supervisor_running(state): if not supervisor_running(state):
raise ProxyError( raise ProxyError(
"Xray работает без proxy-client start (его процесс завершён); " "Xray работает без proxy-client start (его процесс завершён); "
@@ -720,7 +765,7 @@ def require_supervised_xray(config):
raise ProxyError( raise ProxyError(
"Xray запущен на {}:{}, а {} указывает {}:{}; чтобы применить конфиг, " "Xray запущен на {}:{}, а {} указывает {}:{}; чтобы применить конфиг, "
"выполните proxy-client stop, затем proxy-client start".format( "выполните proxy-client stop, затем proxy-client start".format(
state["address"], state["port"], config_home() / "config", state["address"], state["port"], context.config_dir / "config",
config.address, config.port, config.address, config.port,
) )
) )
@@ -756,9 +801,9 @@ def check_port_available(address, port):
raise ProxyError("локальный порт {} занят или недоступен".format(port)) raise ProxyError("локальный порт {} занят или недоступен".format(port))
def acquire_start_lock(): def acquire_start_lock(context):
private_directory(state_home()) private_directory(context.state_dir)
lock = (state_home() / "lock").open("a") lock = (context.state_dir / "lock").open("a")
try: try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError: except BlockingIOError:
@@ -832,8 +877,8 @@ def interrupt(_signum, _frame):
raise KeyboardInterrupt raise KeyboardInterrupt
def command_status(config): def command_status(config, context):
check_proxy(config) check_proxy(config, context)
if config.proxy_type == "external": if config.proxy_type == "external":
print("{} (external) доступен по адресу {}:{}".format(config.name, config.address, config.port)) print("{} (external) доступен по адресу {}:{}".format(config.name, config.address, config.port))
else: else:
@@ -841,21 +886,21 @@ def command_status(config):
return 0 return 0
def command_env(config): def command_env(config, context):
check_proxy(config) check_proxy(config, context)
lines = ["{}={}".format(key, value) for key, value in environment_values(config)] lines = ["{}={}".format(key, value) for key, value in environment_values(config)]
sys.stdout.write("\n".join(lines) + "\n") sys.stdout.write("\n".join(lines) + "\n")
return 0 return 0
def command_run(config, remainder): def command_run(config, remainder, context):
if not remainder or remainder[0] != "--" or len(remainder) == 1: if not remainder or remainder[0] != "--" or len(remainder) == 1:
raise ProxyError("использование: proxy-client run -- COMMAND ARG...") raise ProxyError("использование: proxy-client run -- COMMAND ARG...")
command = remainder[1:] command = remainder[1:]
if shutil.which(command[0]) is None: if shutil.which(command[0]) is None:
sys.stderr.write("Ошибка: команда не найдена в PATH: {}\n".format(command[0])) sys.stderr.write("Ошибка: команда не найдена в PATH: {}\n".format(command[0]))
return 127 return 127
check_proxy(config) check_proxy(config, context)
child_environment = os.environ.copy() child_environment = os.environ.copy()
child_environment.update(dict(environment_values(config))) child_environment.update(dict(environment_values(config)))
try: try:
@@ -866,10 +911,10 @@ def command_run(config, remainder):
return exit_code(completed.returncode) return exit_code(completed.returncode)
def command_start(config, profile_file): def command_start(config, profile_file, context):
if config.proxy_type != "xray": if config.proxy_type != "xray":
raise ProxyError( raise ProxyError(
"start доступен только для type=xray; измените {}".format(config_home() / "config") "start доступен только для type=xray; измените {}".format(context.config_dir / "config")
) )
try: try:
loopback = ipaddress.ip_address(config.address).is_loopback loopback = ipaddress.ip_address(config.address).is_loopback
@@ -877,39 +922,43 @@ def command_start(config, profile_file):
loopback = False loopback = False
if not loopback: if not loopback:
raise ProxyError("для type=xray поле address должно быть loopback-адресом, например 127.0.0.1") raise ProxyError("для type=xray поле address должно быть loopback-адресом, например 127.0.0.1")
lock = acquire_start_lock() lock = acquire_start_lock(context)
try: try:
return run_managed_xray(config, profile_file) return run_managed_xray(config, profile_file, context)
finally: finally:
lock.close() lock.close()
def run_managed_xray(config, profile_file): def run_managed_xray(config, profile_file, context):
if state_path().exists(): if context.state_path.exists():
if xray_running(read_state()): if xray_running(read_state(context)):
raise ProxyError( raise ProxyError(
"Xray остался от прошлого proxy-client start; сначала выполните proxy-client stop" "Xray остался от прошлого proxy-client start; сначала выполните proxy-client stop"
) )
remove_state_files() remove_state_files(context)
check_port_available(ipaddress.ip_address(config.address), config.port) check_port_available(ipaddress.ip_address(config.address), config.port)
profile_path = Path(profile_file) if profile_file else config_home() / "default.url" profile_path = Path(profile_file) if profile_file else context.config_dir / "default.url"
profile = load_profile(profile_path) profile = load_profile(profile_path, context)
executable = resolve_xray() executable = resolve_xray(context)
check_xray_version(executable) check_xray_version(executable)
write_private_json( write_private_json(
generated_config_path(), build_xray_config(profile, config.address, config.port) context.generated_config_path, build_xray_config(profile, config.address, config.port)
) )
child = None child = None
try: try:
run_xray_test(executable, generated_config_path()) run_xray_test(executable, context.generated_config_path)
print("Xray: {} ({})".format(executable, XRAY_VERSION)) print("Xray: {} ({})".format(executable, XRAY_VERSION))
print_summary(profile, config.port) print_summary(profile, config.port)
# SIGTERM (от proxy-client stop) и SIGHUP обрабатываем как Ctrl+C. # SIGTERM (от proxy-client stop) и SIGHUP обрабатываем как Ctrl+C.
signal.signal(signal.SIGTERM, interrupt) signal.signal(signal.SIGTERM, interrupt)
signal.signal(signal.SIGHUP, interrupt) signal.signal(signal.SIGHUP, interrupt)
child = subprocess.Popen([executable, "run", "-c", str(generated_config_path())]) child = subprocess.Popen(
[executable, "run", "-c", str(context.generated_config_path)],
stdout=context.xray_output,
stderr=context.xray_output,
)
write_private_json( write_private_json(
state_path(), context.state_path,
{ {
"supervisor_pid": os.getpid(), "supervisor_pid": os.getpid(),
"xray_pid": child.pid, "xray_pid": child.pid,
@@ -924,8 +973,8 @@ def run_managed_xray(config, profile_file):
raise ProxyError("Xray не открыл локальный порт {}".format(config.port)) raise ProxyError("Xray не открыл локальный порт {}".format(config.port))
diagnose_ip(config) diagnose_ip(config)
print( print(
"Управляемый Xray работает на {}:{}; остановка: Ctrl+C или proxy-client stop".format( "Управляемый Xray работает на {}:{}; остановка: {}".format(
config.address, config.port config.address, config.port, context.stop_hint
), ),
flush=True, flush=True,
) )
@@ -940,11 +989,11 @@ def run_managed_xray(config, profile_file):
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
child.kill() child.kill()
child.wait() child.wait()
remove_state_files() remove_state_files(context)
def command_stop(): def command_stop(context):
state = managed_state() state = managed_state(context)
# Живой супервизор сам остановит Xray; иначе Xray гасим напрямую. # Живой супервизор сам остановит Xray; иначе Xray гасим напрямую.
running_supervisor = supervisor_running(state) running_supervisor = supervisor_running(state)
target = state["supervisor_pid"] if running_supervisor else state["xray_pid"] target = state["supervisor_pid"] if running_supervisor else state["xray_pid"]
@@ -958,17 +1007,574 @@ def command_stop():
os.kill(state["xray_pid"], signal.SIGKILL) os.kill(state["xray_pid"], signal.SIGKILL)
if not wait_until(stopped, STOP_TIMEOUT): if not wait_until(stopped, STOP_TIMEOUT):
raise ProxyError("управляемый Xray не остановился за {} с".format(2 * STOP_TIMEOUT + 1)) raise ProxyError("управляемый Xray не остановился за {} с".format(2 * STOP_TIMEOUT + 1))
remove_state_files() remove_state_files(context)
print("Управляемый Xray остановлен") print("Управляемый Xray остановлен")
return 0 return 0
SERVICE_UNIT = """# Managed by proxy-client
[Unit]
Description=Proxy client (Xray)
After=network.target
StartLimitIntervalSec=0
[Service]
Type=simple
User=proxy-client
Group=proxy-client
RuntimeDirectory=proxy-client
RuntimeDirectoryMode=0700
UMask=0077
ExecStart=/usr/local/lib/proxy-client/proxy-client service run
Restart=on-failure
RestartSec=5
TimeoutStopSec=15
StandardOutput=journal
StandardError=journal
NoNewPrivileges=yes
ProtectSystem=strict
ProtectHome=yes
PrivateTmp=yes
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
[Install]
WantedBy=multi-user.target
"""
def report_journal_storage(root):
# Приоритет каталогов соответствует journald.conf; одноимённый drop-in
# из /etc перекрывает поставляемый системой файл.
directories = ["usr/lib/systemd", "usr/local/lib/systemd", "run/systemd", "etc/systemd"]
main = next((root / directory / "journald.conf"
for directory in reversed(directories)
if (root / directory / "journald.conf").exists()), None)
dropins = {}
for directory in directories:
for path in (root / directory / "journald.conf.d").glob("*.conf"):
dropins[path.name] = path
storage = None
for path in ([main] if main else []) + [dropins[name] for name in sorted(dropins)]:
section = ""
for line in path.read_text().splitlines():
line = line.strip()
if line.startswith("["):
section = line
elif section == "[Journal]" and re.match(r"Storage\s*=", line):
storage = line.split("=", 1)[1].strip() or None
if storage is None:
print("journald: Storage не задан; сохранение журнала зависит от сборки systemd. "
"Проверьте journalctl --list-boots после перезагрузки.")
return
persistent = storage == "persistent" or (
storage == "auto" and (root / "var/log/journal").is_dir())
print("journald: Storage={}; журнал {} между загрузками (по конфигурации).".format(
storage, "сохраняется" if persistent else "не сохраняется"))
def make_install_directory(path):
if path.is_dir():
return
make_install_directory(path.parent)
path.mkdir()
path.chmod(0o755)
def install_file(target, data, mode):
make_install_directory(target.parent)
descriptor, temporary = tempfile.mkstemp(prefix=".install-", dir=str(target.parent))
try:
with os.fdopen(descriptor, "wb") as stream:
stream.write(data)
os.fchmod(stream.fileno(), mode)
os.replace(temporary, str(target))
finally:
if os.path.exists(temporary):
os.unlink(temporary)
def service_config(text):
config = parse_config(text)
try:
loopback = ipaddress.ip_address(config.address).is_loopback
except ValueError:
loopback = False
if config.proxy_type != "xray" or not loopback:
raise ProxyError("службе нужен type=xray и loopback-адрес")
return config
def validate_service_source(source):
missing = [name for name in ("config", "default.url") if not (source / name).is_file()]
if missing:
raise ProxyError("не хватает настроек {} в {}; запустите proxy-client service install "
"в терминале для настройки или укажите готовый --config-dir".format(
", ".join(missing), source))
config = service_config((source / "config").read_text(encoding="utf-8"))
load_profile(source / "default.url", user_context())
return config
@dataclass(frozen=True)
class ServiceSetupPlan:
"""Снимок перехода; config_text=None означает обновление без изменения HOME."""
context: RuntimeContext
original_config: Optional[bytes]
config_text: Optional[str]
profile_text: Optional[str]
manual_state: Optional[dict] = None
resume: bool = False
recovery_file: Optional[str] = None
source_profile: Optional[str] = None
def service_is_installed(root):
manifest = root / "usr/local/lib/proxy-client/installed.json"
return manifest.is_file() or manifest.with_name("installed.pending.json").is_file()
def render_machine_config(config, proxy_type):
parser = configparser.ConfigParser(interpolation=None)
parser["proxy"] = {"type": proxy_type, "name": config.name,
"address": config.address, "port": str(config.port),
"no_proxy_extra": ",".join(config.no_proxy_extra)}
output = io.StringIO()
parser.write(output)
return output.getvalue()
def setup_reset_hint(context):
snapshot = context.state_dir / "service-setup.json"
return ("Чтобы начать заново, удалите только снимок: rm -- {}\n"
"Системные и домашние настройки эта команда не меняет.").format(shlex.quote(str(snapshot)))
def check_setup_source(plan):
path = Path(plan.source_profile) if plan.source_profile else plan.context.config_dir / "default.url"
if plan.source_profile:
changed = not path.is_file() or path.read_text(encoding="utf-8").strip() != plan.profile_text
if not changed:
load_profile(path, plan.context)
else:
changed = path.exists() or path.is_symlink()
if changed:
raise ProxyError("исходный файл VLESS-ссылки изменился; переход остановлен. " +
setup_reset_hint(plan.context))
def prepare_service_setup(args):
context = user_context(args.profile)
pending = context.state_dir / "service-setup.json"
if pending.exists():
if pending.is_symlink() or pending.stat().st_mode & 0o077:
raise ProxyError("снимок перехода должен быть обычным закрытым файлом: {}".format(pending))
if args.profile_file or args.port is not None:
raise ProxyError("есть незавершённый переход; повторите без --file и --port. " +
setup_reset_hint(context))
try:
data = json.loads(pending.read_text(encoding="utf-8"))
original = data["original_config"].encode() if data["original_config"] is not None else None
service_config(data["config_text"])
parse_vless(data["profile_text"])
source_profile = data["source_profile"]
if source_profile is not None and not isinstance(source_profile, str):
raise ValueError("неверный путь источника")
except (KeyError, TypeError, ValueError, AttributeError, ProxyError, configparser.Error):
raise ProxyError("снимок незавершённого перехода повреждён или устарел: {}. ".format(pending) +
setup_reset_hint(context))
manual_state = read_state(context) if managed_xray_running(context) else None
plan = ServiceSetupPlan(context, original, data["config_text"], data["profile_text"],
manual_state, True, data.get("recovery_file"), source_profile)
check_setup_source(plan)
return plan
config_path = context.config_dir / "config"
original = config_path.read_bytes() if config_path.exists() else None
config = parse_config(original.decode("utf-8")) if original is not None else None
# Установленные файлы ещё не означают завершённый переход из ручного режима.
if (Path("/") / "usr/local/lib/proxy-client/installed.json").is_file() and (
config is None or config.proxy_type != "xray"):
if args.profile_file or args.port is not None:
raise ProxyError("настройки службы уже существуют; измените /etc/proxy-client, "
"обновление сохраняет их")
return ServiceSetupPlan(context, None, None, None)
if config_path.is_symlink():
raise ProxyError("домашний config является симлинком; используйте --install-only")
profile_path = Path(args.profile_file).expanduser() if args.profile_file else context.config_dir / "default.url"
if profile_path.is_symlink() and not profile_path.exists():
raise ProxyError("{} — битый симлинк; исправьте ссылку или выберите другой --file".format(profile_path))
source_profile = None
if profile_path.is_file():
load_profile(profile_path, context)
profile_text = profile_path.read_text(encoding="utf-8").strip()
source_profile = str(profile_path.absolute())
elif args.profile_file or not sys.stdin.isatty():
raise ProxyError("нет VLESS-ссылки; укажите закрытый файл через --file или запустите в терминале")
else:
# getpass не должен переходить к видимому вводу, если терминал недоступен.
with warnings.catch_warnings():
warnings.simplefilter("error", getpass.GetPassWarning)
try:
profile_text = getpass.getpass("VLESS-ссылка (ввод скрыт): ").strip()
except getpass.GetPassWarning:
raise ProxyError("скрытый ввод недоступен; передайте ссылку через закрытый --file")
parse_vless(profile_text)
if config is None or config.proxy_type != "xray":
port = DEFAULT_PORT if args.port is None else args.port
if args.port is None and not args.yes:
raw = input("Локальный порт [{}]: ".format(DEFAULT_PORT)).strip()
if raw:
try:
port = int(raw)
except ValueError:
raise ProxyError("порт должен быть числом от 1 до 65535")
config = ProxyConfig("xray", "Xray", "127.0.0.1", port,
config.no_proxy_extra if config else DEFAULT_NO_PROXY_EXTRA)
elif args.port is not None:
config.port = args.port
text = render_machine_config(config, "xray")
service_config(text)
manual_state = read_state(context) if managed_xray_running(context) else None
return ServiceSetupPlan(context, original, text, profile_text, manual_state,
recovery_file=str(profile_path.resolve()) if args.profile_file else None,
source_profile=source_profile)
def find_service_xray(explicit, home, root):
if explicit:
path = Path(explicit)
if not path.is_absolute():
raise ProxyError("xray должен быть абсолютным путём")
return path
local = home / ".local/bin/xray"
if local.is_file():
return local
found = shutil.which("xray")
if found:
return Path(found).resolve()
installed = root / "usr/local/lib/proxy-client/xray"
return installed if installed.is_file() else None
def service_systemctl(*arguments, check=True):
return subprocess.run(["sudo", "--", "systemctl", *arguments], check=check)
def service_ready(config):
return subprocess.run(["systemctl", "is-active", "--quiet", "proxy-client.service"],
check=False).returncode == 0 and port_open(config.address, config.port)
def switch_home_to_service(plan):
path = plan.context.config_dir / "config"
config = service_config(plan.config_text)
replacement = render_machine_config(config, "external").encode()
current = path.read_bytes() if path.exists() else None
if not path.is_symlink() and current == replacement:
return
if path.is_symlink() or current != plan.original_config:
raise ProxyError("домашний config изменился во время установки; он оставлен без изменений. " +
setup_reset_hint(plan.context))
private_directory(plan.context.config_dir)
if plan.original_config is not None:
descriptor, backup = tempfile.mkstemp(prefix="config.before-service.", dir=str(path.parent))
with os.fdopen(descriptor, "wb") as stream:
stream.write(plan.original_config)
print("Прежний config сохранён: {}".format(backup))
install_file(path, replacement, 0o600)
def apply_service_setup(args, plan, xray):
if plan.config_text is None:
with tempfile.TemporaryDirectory(prefix="proxy-service-update-") as temporary:
if xray is None:
xray = Path(temporary) / "xray"
install_xray(xray, download_file)
options = vars(args).copy()
options.update(config_dir=str(plan.context.config_dir), xray=str(xray),
profile="default", enable=True, start=False, restart=True)
result = service_install(argparse.Namespace(**options))
if result == 0:
print("Служба обновлена и готова; её настройки и домашний профиль сохранены.")
return result
config = service_config(plan.config_text)
check_setup_source(plan)
config_path = plan.context.config_dir / "config"
current = config_path.read_bytes() if config_path.exists() else None
if config_path.is_symlink() or current not in (
plan.original_config, render_machine_config(config, "external").encode()):
raise ProxyError("домашний config изменился; переход остановлен без его замены. " +
setup_reset_hint(plan.context))
replacing_installed = service_is_installed(Path("/"))
if not replacing_installed and (plan.manual_state is None or (config.address, config.port) != (
plan.manual_state["address"], plan.manual_state["port"])):
check_port_available(ipaddress.ip_address(config.address), config.port)
with tempfile.TemporaryDirectory(prefix="proxy-service-") as temporary:
source = Path(temporary)
install_file(source / "config", plan.config_text.encode(), 0o600)
install_file(source / "default.url", (plan.profile_text + "\n").encode(), 0o600)
if xray is None:
print("Скачивание проверенного Xray {}…".format(XRAY_VERSION))
xray = source / "xray"
install_xray(xray, download_file)
check_xray_version(str(xray))
generated = source / "client.json"
write_private_json(generated, build_xray_config(parse_vless(plan.profile_text),
config.address, config.port))
run_xray_test(str(xray), generated)
private_directory(plan.context.state_dir)
pending = plan.context.state_dir / "service-setup.json"
install_file(pending, json.dumps({
"original_config": plan.original_config.decode() if plan.original_config is not None else None,
"config_text": plan.config_text, "profile_text": plan.profile_text,
"recovery_file": plan.recovery_file,
"source_profile": plan.source_profile,
}).encode(), 0o600)
options = vars(args).copy()
options.update(config_dir=str(source), xray=str(xray), profile="default",
enable=False, start=False, restart=False, expect_profile=True)
result = service_install(argparse.Namespace(**options))
if result:
if pending.exists():
sys.stderr.write("Снимок сохранён для повторной попытки. " +
setup_reset_hint(plan.context) + "\n")
return result
check_setup_source(plan)
if managed_xray_running(plan.context):
if read_state(plan.context) != plan.manual_state:
raise ProxyError("ручной Xray изменился во время установки; переход остановлен")
command_stop(plan.context)
try:
if replacing_installed:
service_systemctl("stop", "proxy-client.service")
check_port_available(ipaddress.ip_address(config.address), config.port)
service_systemctl("enable", "--now", "proxy-client.service")
if not wait_until(lambda: service_ready(config), 10):
raise ProxyError("служба не стала готова")
except (ProxyError, OSError, subprocess.CalledProcessError, KeyboardInterrupt) as error:
cleanup = service_systemctl("disable", "--now", "proxy-client.service", check=False)
recovery = ["proxy-client"]
if args.profile != "default":
recovery.extend(["--profile", args.profile])
recovery.append("start")
if plan.recovery_file:
recovery.extend(["-f", plan.recovery_file])
hint = (" Ручной Xray остановлен; вернуть его: {}.".format(
" ".join(shlex.quote(part) for part in recovery)) if plan.manual_state else "")
if cleanup.returncode:
hint += " Не удалось отключить службу; выполните sudo systemctl disable --now proxy-client."
raise ProxyError("переход не завершён; домашний config не изменён. "
"Проверьте sudo journalctl -u proxy-client -b." + hint) from error
try:
switch_home_to_service(plan)
except (ProxyError, OSError) as error:
raise ProxyError("служба уже запущена, но домашний профиль не удалось переключить: {}".format(error))
pending.unlink()
print("Служба готова на {}:{}. Домашний профиль переключён на external.".format(
config.address, config.port))
return 0
def setup_service(args):
if os.geteuid() == 0:
raise ProxyError("для настройки и перехода запустите proxy-client service install без sudo; "
"для установки только файлов используйте --install-only")
plan = prepare_service_setup(args)
xray = find_service_xray(args.xray, Path.home(), Path("/"))
if xray is None:
print("Xray {} будет скачан и проверен перед установкой.".format(XRAY_VERSION))
if plan.config_text is None:
print("План: обновить и перезапустить службу, включить автозапуск; настройки сохранить.")
else:
config = service_config(plan.config_text)
print("План: установить системную службу на {}:{}, включить автозапуск "
"и переключить текущий профиль.".format(config.address, config.port))
if plan.original_config and parse_config(plan.original_config.decode()).proxy_type == "external":
print("Текущий внешний прокси будет заменён локальной службой; прежний config сохранится.")
if plan.resume:
print("Продолжение незавершённого перехода; изменённые настройки не будут заменены.")
if plan.manual_state is not None:
print("После установки файлов выбранный ручной Xray будет остановлен; "
"текущие соединения прервутся.")
if not args.yes and input("Продолжить? [y/N]: ").strip().lower() not in ("y", "yes", "д", "да"):
print("Установка отменена.")
return 0
return apply_service_setup(args, plan, xray)
def command_service_install(args):
if args.port is not None and not 1 <= args.port <= 65535:
raise ProxyError("порт должен быть числом от 1 до 65535")
if args.yes and (args.root or args.install_only or args.config_dir):
raise ProxyError("--yes нельзя совмещать с --root, --install-only или --config-dir")
if (args.profile_file or args.port is not None) and (args.root or args.install_only or args.config_dir):
raise ProxyError("--file и --port относятся к настройке; их нельзя совмещать "
"с --root, --install-only или --config-dir")
if args.yes or (sys.stdin.isatty() and not args.root and not args.install_only
and not args.config_dir):
return setup_service(args)
if not (args.root or args.install_only or args.config_dir):
raise ProxyError("без терминала выберите --yes для полного перехода "
"или --install-only для установки только файлов")
return service_install(args)
def service_install(args):
if args.config_dir and args.profile != "default":
raise ProxyError("выберите --profile или --config-dir, не оба сразу")
root = Path(args.root) if args.root else Path("/")
if not root.is_absolute():
raise ProxyError("root должен быть абсолютным путём")
if args.root and (args.enable or args.start or args.restart):
raise ProxyError("--root несовместим с --enable, --start и --restart")
home = Path.home()
if os.geteuid() == 0 and "SUDO_UID" in os.environ and (not args.config_dir or not args.xray):
import pwd
try:
home = Path(pwd.getpwuid(int(os.environ["SUDO_UID"])).pw_dir)
except (ValueError, KeyError):
raise ProxyError("не удалось определить пользователя sudo; укажите --config-dir и --xray")
source = Path(args.config_dir) if args.config_dir else user_context(args.profile, home=home).config_dir
xray = find_service_xray(args.xray, home, root)
if xray is None:
raise ProxyError("Xray не найден; запустите proxy-client service install в терминале "
"для настройки или выполните proxy-client install")
if not all(path.is_absolute() for path in (source, xray, root)):
raise ProxyError("config-dir, xray и root должны быть абсолютными путями")
if not service_is_installed(root):
validate_service_source(source)
if not args.root and os.geteuid() != 0:
# Пути выбираем до sudo: оно сбрасывает HOME, XDG и PATH.
command = ["sudo", "--", sys.executable, str(Path(__file__).resolve()),
"service", "install", "--config-dir", str(source), "--xray", str(xray)]
if args.enable:
command.append("--enable")
if args.start:
command.append("--start")
if args.restart:
command.append("--restart")
if args.expect_profile:
command.append("--expect-profile")
print("Установка системной службы через sudo.", flush=True)
return exit_code(subprocess.run(command, check=False).returncode)
settings = root / "etc/proxy-client"
# При обновлении служба уже имеет собственные настройки вне HOME.
config_path = settings / "config" if (settings / "config").exists() else source / "config"
config = service_config(config_path.read_text(encoding="utf-8"))
profile_path = settings / "default.url" if (settings / "default.url").exists() else source / "default.url"
profile_context = service_context(argparse.Namespace(config_dir=config_path.parent,
runtime_dir=root / "run/proxy-client", xray=xray)) if profile_path.parent == settings else user_context()
profile = load_profile(profile_path, profile_context)
if args.expect_profile:
expected_config = validate_service_source(source)
expected_profile = load_profile(source / "default.url", user_context())
if vars(config) != vars(expected_config) or build_xray_config(profile, config.address, config.port) != \
build_xray_config(expected_profile, expected_config.address, expected_config.port):
raise ProxyError("системные настройки отличаются от снимка перехода; они оставлены без изменений. "
"Для их смены отредактируйте /etc/proxy-client отдельно")
check_xray_version(str(xray))
with tempfile.TemporaryDirectory(prefix="proxy-service-test-") as temporary:
generated = Path(temporary) / "client.json"
write_private_json(generated, build_xray_config(profile, config.address, config.port))
run_xray_test(str(xray), generated)
files = {
"usr/local/lib/proxy-client/proxy-client": (Path(__file__).read_bytes(), 0o755),
"usr/local/lib/proxy-client/xray": (xray.read_bytes(), 0o755),
"etc/systemd/system/proxy-client.service": (SERVICE_UNIT.encode(), 0o644),
"usr/lib/sysusers.d/proxy-client.conf": (
b'u proxy-client - "Proxy client" /nonexistent /usr/sbin/nologin\n', 0o644),
}
manifest = root / "usr/local/lib/proxy-client/installed.json"
pending = manifest.with_name("installed.pending.json")
# До любой записи проверяем все назначения и компоненты пути.
targets = [root / name for name in files] + [
manifest, pending, settings, settings / "config", settings / "default.url"]
for target in targets:
if any(part.is_symlink() for part in [target, *target.parents]):
raise ProxyError("установка через симлинк запрещена: {}".format(target))
owned = json.loads(manifest.read_text()) if manifest.exists() else {}
interrupted = json.loads(pending.read_text()) if pending.exists() else {}
for name in files:
target = root / name
if target.exists() and (
not target.is_file() or
hashlib.sha256(target.read_bytes()).hexdigest() not in
[owned.get(name)] + interrupted.get(name, [])
):
raise ProxyError("чужой или изменённый файл: {}".format(target))
if not manifest.exists() and not pending.exists() and settings.exists():
raise ProxyError("чужой каталог настроек: {}".format(settings))
make_install_directory(manifest.parent)
manifest.parent.chmod(0o755)
# До замены сохраняем допустимые старые и новые хеши. После сбоя повтор
# распознает уже записанную часть, не принимая произвольные чужие файлы.
planned = {}
for name, (data, _) in files.items():
target = root / name
planned[name] = [hashlib.sha256(data).hexdigest()]
if target.exists():
planned[name].append(hashlib.sha256(target.read_bytes()).hexdigest())
install_file(pending, json.dumps(planned).encode(), 0o644)
for name, (data, mode) in files.items():
install_file(root / name, data, mode)
settings.mkdir(parents=True, exist_ok=True)
settings.chmod(0o750)
for name in ("config", "default.url"):
target = settings / name
if not target.exists():
install_file(target, (source / name).read_bytes(), 0o640)
install_file(manifest, json.dumps({
name: hashlib.sha256(data).hexdigest() for name, (data, _) in files.items()
}).encode(), 0o644)
pending.unlink()
if not args.root:
import grp
subprocess.run(["systemd-sysusers", "/usr/lib/sysusers.d/proxy-client.conf"], check=True)
gid = grp.getgrnam("proxy-client").gr_gid
for path in (settings, settings / "config", settings / "default.url"):
os.chown(str(path), 0, gid)
subprocess.run(["systemctl", "daemon-reload"], check=True)
if args.enable:
subprocess.run(["systemctl", "enable", "proxy-client.service"], check=True)
if args.start:
subprocess.run(["systemctl", "start", "proxy-client.service"], check=True)
if args.restart:
subprocess.run(["systemctl", "stop", "proxy-client.service"], check=True)
check_port_available(ipaddress.ip_address(config.address), config.port)
subprocess.run(["systemctl", "start", "proxy-client.service"], check=True)
if not wait_until(lambda: service_ready(config), 10):
raise ProxyError("служба не стала готова; проверьте sudo journalctl -u proxy-client -b")
print("Файлы службы установлены.")
report_journal_storage(root)
return 0
def build_parser(): def build_parser():
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="proxy-client", prog="proxy-client",
description="Безопасный запуск команд через явно настроенный прокси.", description="Безопасный запуск команд через явно настроенный прокси.",
) )
parser.add_argument("--profile", default="default", help="имя пользовательского профиля")
subparsers = parser.add_subparsers(dest="subcommand") subparsers = parser.add_subparsers(dest="subcommand")
service = subparsers.add_parser("service", help="системная служба")
actions = service.add_subparsers(dest="service_action", required=True)
serve = actions.add_parser("run", help="запуск из systemd")
serve.add_argument("--config-dir", default="/etc/proxy-client")
serve.add_argument("--runtime-dir", default="/run/proxy-client")
serve.add_argument("--xray", default="/usr/local/lib/proxy-client/xray")
setup = actions.add_parser("install", help="установить системную службу")
setup.add_argument("--config-dir", help="каталог настроек (по умолчанию текущий профиль)")
setup.add_argument("--xray", help="готовый Xray (по умолчанию ~/.local/bin/xray или PATH)")
setup.add_argument("--root", help="изолированный корень без systemctl")
setup.add_argument("--enable", action="store_true")
setup.add_argument("--start", action="store_true")
setup.add_argument("--restart", action="store_true", help="перезапустить и проверить готовность службы")
setup.add_argument("--yes", "--auto-yes", action="store_true",
help="принять план полного перехода без вопроса")
setup.add_argument("--install-only", action="store_true",
help="установить файлы без мастера и переключения домашнего профиля")
setup.add_argument("-f", "--file", dest="profile_file", help="закрытый файл VLESS-ссылки для настройки")
setup.add_argument("--port", type=int, help="локальный порт при настройке (по умолчанию 12334)")
setup.add_argument("--expect-profile", action="store_true", help=argparse.SUPPRESS)
install_parser = subparsers.add_parser("install", help="установить проверенный Xray") install_parser = subparsers.add_parser("install", help="установить проверенный Xray")
install_parser.add_argument( install_parser.add_argument(
"--replace", action="store_true", help="заменить существующий ~/.local/bin/xray" "--replace", action="store_true", help="заменить существующий ~/.local/bin/xray"
@@ -992,20 +1598,31 @@ def main():
return 2 return 2
try: try:
if arguments.subcommand == "service":
if arguments.service_action == "install":
return command_service_install(arguments)
if arguments.profile != "default":
raise ProxyError("service run не использует пользовательские профили")
context = service_context(arguments)
return command_start(load_config(context), None, context)
context = user_context(arguments.profile)
if arguments.subcommand == "install": if arguments.subcommand == "install":
return command_install(arguments.replace) return command_install(arguments.replace, context=context)
if arguments.subcommand == "stop": if arguments.subcommand == "stop":
return command_stop() return command_stop(context)
config = load_config() config = load_config(context)
if arguments.subcommand == "status": if arguments.subcommand == "status":
return command_status(config) return command_status(config, context)
if arguments.subcommand == "env": if arguments.subcommand == "env":
return command_env(config) return command_env(config, context)
if arguments.subcommand == "run": if arguments.subcommand == "run":
return command_run(config, arguments.arguments) return command_run(config, arguments.arguments, context)
if arguments.subcommand == "start": if arguments.subcommand == "start":
return command_start(config, arguments.profile_file) return command_start(config, arguments.profile_file, context)
except (ProxyError, OSError) as error: except (EOFError, KeyboardInterrupt):
sys.stderr.write("Установка или команда прервана пользователем.\n")
return 130
except (ProxyError, OSError, ValueError, subprocess.CalledProcessError) as error:
sys.stderr.write("Ошибка: {}\n".format(error)) sys.stderr.write("Ошибка: {}\n".format(error))
return 1 return 1
return 2 return 2
+259 -7
View File
@@ -1,7 +1,7 @@
# proxy-client # proxy-client
`proxy-client` — переносимый Linux/Bash-интерфейс для запуска команд через `proxy-client` — переносимый Linux/Bash-интерфейс для запуска команд через
внешний прокси или через управляемый пользовательский Xray. Он добавляет команды внешний прокси или через управляемый Xray. Он добавляет команды
`proxy_on`, `proxy_off`, `pclaude` и `pcodex`, но не подменяет обычные `claude` `proxy_on`, `proxy_off`, `pclaude` и `pcodex`, но не подменяет обычные `claude`
и `codex`. и `codex`.
@@ -78,7 +78,7 @@ source ~/.bash_custom
## Как выбирается прокси ## Как выбирается прокси
Режим определяется только машинным файлом Без выбора профиля режим определяется файлом
`~/.config/proxy-client/config`: `~/.config/proxy-client/config`:
| Машинный конфиг | Используется | Что запускать | | Машинный конфиг | Используется | Что запускать |
@@ -212,14 +212,55 @@ proxy-client install
proxy-client start proxy-client start
``` ```
Другой закрытый профиль можно выбрать без замены `default.url`: Другой файл VLESS-ссылки можно выбрать без замены `default.url`:
```bash ```bash
proxy-client start -f ~/.config/proxy-client/work.url proxy-client start -f ~/.config/proxy-client/work.url
``` ```
Для переключения профиля сначала выполните `proxy-client stop`. Все файлы с Для смены ссылки работающего экземпляра сначала выполните `proxy-client stop`.
настоящими VLESS-ссылками должны иметь права `600` и не должны попадать в Git. Все файлы с настоящими VLESS-ссылками должны иметь права `600` и не должны
попадать в Git.
## Именованные профили
Профиль объединяет настройки локального входа, исключения и, для Xray,
VLESS-ссылку. Без указания имени используется `default`: прежние `config` и
`default.url` остаются на своих местах. Дополнительные профили живут в
`~/.config/proxy-client/profiles/<имя>/`. При заданном `XDG_CONFIG_HOME`
вместо `~/.config` используется этот каталог.
Например, создайте профиль `work`:
```bash
mkdir -p ~/.config/proxy-client/profiles/work
cp ~/.config/proxy-client/examples/config.xray ~/.config/proxy-client/profiles/work/config
$EDITOR ~/.config/proxy-client/profiles/work/config
```
В `config` задайте свободный порт, например `12335`, если основной прокси
занимает `12334`. Поместите VLESS-ссылку в `profiles/work/default.url` и
ограничьте доступ:
```bash
chmod 600 ~/.config/proxy-client/profiles/work/default.url
proxy-client --profile work start
```
Команда остаётся в терминале. В другом терминале можно проверить или
остановить именно этот профиль:
```bash
proxy-client --profile work status
proxy-client --profile work stop
```
Параметр `--profile` ставится перед подкомандой. Неизвестный профиль даёт
ошибку; перехода на `default` нет. Разные профили могут работать одновременно
на разных портах. Остановка одного не затрагивает остальные.
`start -f FILE` переопределяет только файл ссылки выбранного профиля на этот
запуск. Его порт и исключения по-прежнему берутся из `config`.
## Ежедневное использование ## Ежедневное использование
@@ -243,6 +284,24 @@ proxy-client start -f ~/.config/proxy-client/work.url
обёрнуты и не выполняют предварительную проверку, но после `proxy_on` наследуют обёрнуты и не выполняют предварительную проверку, но после `proxy_on` наследуют
прокси-переменные, как и любой другой дочерний процесс. прокси-переменные, как и любой другой дочерний процесс.
Для именованного профиля:
```bash
proxy_on --profile work
pclaude --proxy-profile work
pcodex --proxy-profile work --search
proxy-client --profile work run -- COMMAND ARG...
```
У `pclaude` и `pcodex` параметр `--proxy-profile` должен быть первым.
Обычный `--profile` передаётся запускаемой программе.
`proxy_on --profile work` меняет окружение терминала, но не профиль по
умолчанию: обычный `codex` наследует это окружение, а `pcodex` без параметра
выберет `default`. `proxy_off` только очищает переменные — работающие прокси,
включая системную службу, остаются запущенными. `myip` использует текущее
окружение.
### Жизненный цикл Xray ### Жизненный цикл Xray
```bash ```bash
@@ -253,7 +312,7 @@ proxy-client stop
``` ```
`start` сначала проверяет версию и конфигурацию Xray, затем остаётся на переднем `start` сначала проверяет версию и конфигурацию Xray, затем остаётся на переднем
плане. Systemd, фоновый автозапуск и скрытый запуск из `run` не используются. плане. Скрытого запуска из `run` нет; системная служба устанавливается отдельно.
Один loopback-порт принимает HTTP CONNECT и SOCKS5. Один loopback-порт принимает HTTP CONNECT и SOCKS5.
Остановить Xray можно двумя способами: `Ctrl+C` в окне `start` или Остановить Xray можно двумя способами: `Ctrl+C` в окне `start` или
@@ -262,7 +321,200 @@ proxy-client stop
остаться работать: `status` и `env` тогда сообщат об ошибке, а `stop` добьёт его остаться работать: `status` и `env` тогда сообщат об ошибке, а `stop` добьёт его
и уберёт файлы. `stop` работает и после изменения машинного типа на `external`. и уберёт файлы. `stop` работает и после изменения машинного типа на `external`.
Второй `start`, пока первый работает, завершится ошибкой. Второй `start` того же профиля, пока первый работает, завершится ошибкой.
## Системная служба
Этот режим нужен, когда приложения должны получать прокси после загрузки
Linux, до входа пользователя. Требуются systemd, `systemd-sysusers` и права
администратора для установки. Служба одна; именованные профили остаются
пользовательскими.
### Установка и переход из tmux
Запустите от своего пользователя, без `sudo`:
```bash
proxy-client service install
```
В терминале команда показывает план и просит одно подтверждение.
Если настроек нет, она запросит VLESS-ссылку скрытым вводом и предложит
локальный порт `12334`. Если Xray отсутствует, скачает закреплённую версию
и проверит контрольную сумму. Ссылка не попадает в аргументы процессов и журнал.
Если профиль уже настроен, используются его `config` и `default.url`.
После подтверждения команда проверяет конфигурацию Xray, устанавливает
системные файлы через `sudo`, останавливает свой ручной Xray и запускает службу
с автозапуском. Текущие соединения при переходе прервутся. Чужой процесс,
занимающий порт, команда не останавливает.
Когда служба активна и локальный порт доступен, домашний `config` переключается
на `type=external`. Поэтому `proxy_on`, `status` и `run` сразу работают со
службой. Прежний файл сохраняется рядом как `config.before-service.*`.
Готовность локального порта не гарантирует доступность удалённого сервера.
Если файлы службы уже установлены через `--install-only`, но домашний профиль
ещё имеет `type=xray`, мастер завершит перенос. Сначала он проверит совпадение
настроек с системными; другой профиль не заменит уже настроенную службу.
Для именованного профиля:
```bash
proxy-client --profile work service install
```
Для принятия плана без вопроса есть `--yes` и его синоним `--auto-yes`:
```bash
proxy-client service install --yes
```
При настройке с нуля без терминала передайте закрытый файл со ссылкой:
```bash
chmod 600 ~/connection.url
proxy-client service install --auto-yes --file ~/connection.url --port 12334
```
`--file` также подходит, если ручной Xray запускался с разовой ссылкой
через `start -f`. Без него берётся сохранённый `default.url`.
`--yes` принимает план, но не заменяет ввод недостающей ссылки и не отменяет
проверки или запрос пароля `sudo`.
### Ошибка или прерывание
До успешной установки системных файлов ручной Xray продолжает работать.
Если запуск службы не удался после его остановки, домашний профиль остаётся
прежним. Команда пытается отключить неудачно запущенную службу и показывает,
как вернуть ручной запуск. Восстановить прежние соединения и процесс в tmux
автоматически нельзя.
Повторите `proxy-client service install`: незавершённый переход продолжится
по закрытому снимку `service-setup.json` в каталоге состояния выбранного
профиля. Снимок содержит VLESS-ссылку, имеет права `600` в каталоге `700`
и удаляется после успеха. Если домашние или системные настройки успели
измениться, команда остановится, чтобы их не затереть. Проверяется также
исходный `default.url` или файл, переданный через `--file`, до остановки
ручного Xray. После запуска служба использует свою копию ссылки; позднее
изменение исходной ссылки не отменяет завершение перехода.
Чтобы отказаться от старого снимка и прочитать текущие настройки заново,
удалите только `service-setup.json`: точную команду `rm -- …` CLI печатает
при конфликте. Для профиля по умолчанию без переопределения XDG это:
```bash
rm -- ~/.local/state/proxy-client/service-setup.json
proxy-client service install
```
Сброс снимка не меняет домашние файлы, настройки службы и работающие процессы.
Если системные настройки уже отличаются от желаемых, их нужно изменить
отдельно в `/etc/proxy-client` — мастер их не перезаписывает.
### Только установка файлов
Для прежнего режима без мастера и переключения домашнего профиля:
```bash
proxy-client service install --install-only
```
Без терминала явно выберите `--yes` для полного перехода или `--install-only`
для установки файлов; команда без выбора режима завершится с подсказкой.
Режим установки файлов требует готовые
настройки и Xray, а при их отсутствии выдаёт подсказку до вызова `sudo`.
`--enable`, `--start` и `--restart` явно включают автозапуск, запуск сейчас
или перезапуск с проверкой готовности. Без этих флагов служба не запускается.
Xray ищется в `~/.local/bin`, затем в `PATH` и системной установке.
Параметр `--xray FILE` переопределяет выбор. Для установки из произвольного
каталога есть `--config-dir DIR`: он тоже выбирает режим установки файлов.
Пути должны быть абсолютными; `--config-dir` несовместим с именованным
`--profile`. Прямой запуск через `sudo` подходит для `--install-only`;
нестандартный каталог XDG в этом случае задавайте через `--config-dir`.
### Доступ к прокси
Xray слушает только числовой loopback-адрес: `127.0.0.1` или `::1`.
Проверка выполняется при установке и каждом запуске. Адреса всех интерфейсов
(`0.0.0.0`, `::`) и локальной сети не принимаются. Другие процессы на этой же
машине могут пользоваться прокси.
Установщик копирует настройки в `/etc/proxy-client/`, а CLI и Xray — в
`/usr/local/lib/proxy-client/`. Служба больше не зависит от домашнего каталога.
Изменения исходного пользовательского профиля на неё не влияют. Системные
настройки имеют владельца `root`, группу `proxy-client` и права `640`;
каталог — `750`. Процесс работает под отдельным пользователем `proxy-client`.
Временные файлы находятся в закрытом `/run/proxy-client/`.
Управление и журнал:
```bash
sudo systemctl status proxy-client
sudo systemctl stop proxy-client
sudo systemctl start proxy-client
sudo systemctl restart proxy-client
sudo journalctl -u proxy-client -b
sudo journalctl -u proxy-client -f
```
Для изменения настроек отредактируйте `/etc/proxy-client/config` или
`/etc/proxy-client/default.url` с правами администратора, затем перезапустите
службу. Сохраняйте владельца, группу и права файлов.
После аварии systemd повторяет запуск с паузой в пять секунд. Явная команда
`stop` оставляет службу остановленной, но не отменяет запуск при следующей
загрузке. Чтобы отключить автозапуск и остановить её сразу:
```bash
sudo systemctl disable --now proxy-client
```
Чтобы включить обратно:
```bash
sudo systemctl enable --now proxy-client
```
При обновлении повторите команду из актуальной копии пакета:
```bash
proxy-client service install
```
Мастер предложит обновить и перезапустить службу. Настройки в
`/etc/proxy-client` сохраняются, домашний профиль не меняется и может
отсутствовать. Без терминала используйте `--yes` для этого же действия.
Режим `--install-only` по-прежнему не перезапускает службу без явного флага.
Чужие или вручную изменённые исполняемые файлы и unit-файл установщик
не заменяет; изменения параметров systemd делайте через
`sudo systemctl edit proxy-client`. Файлы заменяются
по одному атомарно, общего отката всей установки при сбое нет. После
устранения причины сбоя повторите установку: журнал незавершённой установки
позволяет продолжить её, сохраняя проверку принадлежности файлов.
Вывод CLI поступает в journald. Сырой вывод Xray в системном режиме подавлен,
чтобы сообщения с фрагментами конфигурации не раскрывали секреты; детальные
сетевые журналы Xray поэтому недоступны. Установщик сообщает настройку хранения
journald, но не меняет её. Сохранение журнала после перезагрузки зависит от
настроек машины; проверить предыдущие загрузки можно через
`sudo journalctl --list-boots`.
Открытый локальный порт не гарантирует доступность удалённого сервера.
На ноутбуке сеть может появиться позже службы. Приложениям без повторения
запросов понадобится отдельная настройка ожидания готовности.
Для `proxy_on`, `pclaude`, `pcodex` и `run` системная служба — внешний прокси:
укажите в нужном пользовательском профиле `type = external`, адрес
`127.0.0.1` и порт службы. Дополнительный профиль типа `xray` запускайте вручную
на другом порту. Пользовательский `stop` системной службой не управляет.
Для проверки размещения файлов без установки на машину есть
`service install --config-dir DIR --xray FILE --root /absolute/staging`.
Этот режим не вызывает systemd, не создаёт системную учётную запись и не
назначает ей владельцев; он несовместим с `--enable` и `--start`.
## Прокси в настройках Codex и Claude Code ## Прокси в настройках Codex и Claude Code
+54 -4
View File
@@ -15,7 +15,7 @@ from urllib.parse import quote
REPO_ROOT = Path(__file__).resolve().parents[1] REPO_ROOT = Path(__file__).resolve().parents[1]
CLI = REPO_ROOT / "proxy-client/.local/bin/proxy-client" CLI = Path(os.environ.get("PROXY_CLIENT_TEST_CLI", str(REPO_ROOT / "proxy-client/.local/bin/proxy-client")))
UUID = "11111111-2222-4333-8444-555555555555" UUID = "11111111-2222-4333-8444-555555555555"
PATH_SECRET = "/tenant/PATH_SENTINEL_a1b2" PATH_SECRET = "/tenant/PATH_SENTINEL_a1b2"
PROFILE = ( PROFILE = (
@@ -28,7 +28,7 @@ REALITY_PROFILE = (
"vless://{}@edge.example:443?type=tcp&security=reality&encryption=none" "vless://{}@edge.example:443?type=tcp&security=reality&encryption=none"
"&flow=xtls-rprx-vision&sni=front.example&fp=chrome&pbk={}&sid={}#Synthetic" "&flow=xtls-rprx-vision&sni=front.example&fp=chrome&pbk={}&sid={}#Synthetic"
).format(UUID, PBK, SID) ).format(UUID, PBK, SID)
READY = "остановка: Ctrl+C или proxy-client stop" READY = "Управляемый Xray работает на"
# Поддельный Xray — симлинк на интерпретатор Python: тогда /proc/PID/exe # Поддельный Xray — симлинк на интерпретатор Python: тогда /proc/PID/exe
# совпадает с разрешённым путём к xray, как у настоящего бинарника. # совпадает с разрешённым путём к xray, как у настоящего бинарника.
@@ -102,6 +102,56 @@ class ManagedXrayTest(unittest.TestCase):
self.write_machine_config() self.write_machine_config()
self.profile = self.write_profile(self.config_dir / "default.url") self.profile = self.write_profile(self.config_dir / "default.url")
def test_profiles_run_independently_and_export_selected_environment(self):
first = self.start()
named = self.config_dir / "profiles/second"
named.mkdir(parents=True)
second_port = free_port()
(named / "config").write_text(
(self.config_dir / "config").read_text().replace(str(self.port), str(second_port)))
# Переопределение -f должно работать без default.url у второго профиля.
second = self.start("-f", str(self.profile), cli_prefix=["--profile", "second", "start"])
self.assertIn("proxy-client --profile second stop", self.start_output)
status = self.run_cli("--profile", "second", "status")
self.assertEqual(status.returncode, 0, status.stderr)
env = self.run_cli("--profile", "second", "env")
self.assertIn("127.0.0.1:" + str(second_port), env.stdout)
run = self.run_cli("--profile", "second", "run", "--", sys.executable,
"-c", "import os,sys; print(os.environ['HTTP_PROXY']); sys.exit(23)")
self.assertEqual(run.returncode, 23, run.stderr)
self.assertIn("127.0.0.1:" + str(second_port), run.stdout)
collision = self.run_cli("--profile", "second", "start", "-f", str(self.profile))
self.assertEqual(collision.returncode, 1)
stopped = self.run_cli("--profile", "second", "stop")
self.assertEqual(stopped.returncode, 0, stopped.stderr)
second.communicate(timeout=10)
self.assertIsNone(first.poll())
self.assertEqual(self.run_cli("--profile", "default", "status").returncode, 0)
self.assertEqual(self.run_cli("stop").returncode, 0)
first.communicate(timeout=10)
def test_service_run_uses_explicit_paths_and_pinned_binary(self):
runtime = self.root / "service-runtime"
process = self.start(cli_prefix=[
"service", "run", "--config-dir", str(self.config_dir),
"--runtime-dir", str(runtime), "--xray", str(self.xray),
], PATH="/usr/bin:/bin")
self.assertTrue((runtime / "state.json").exists())
self.assertFalse(self.state_dir.exists())
process.terminate()
process.communicate(timeout=10)
self.assertEqual(process.returncode, 0)
self.assertFalse((runtime / "client.json").exists())
def test_system_profile_permission_hint_preserves_group_read(self):
self.profile.chmod(0o644)
result = self.run_cli("service", "run", "--config-dir", str(self.config_dir),
"--runtime-dir", str(self.root / "system-runtime"),
"--xray", str(self.xray))
self.assertEqual(1, result.returncode)
self.assertIn("chmod 640", result.stderr)
self.assertIn("chown root:proxy-client", result.stderr)
def write_machine_config(self, address="127.0.0.1"): def write_machine_config(self, address="127.0.0.1"):
self.config_dir.mkdir(parents=True, exist_ok=True) self.config_dir.mkdir(parents=True, exist_ok=True)
(self.config_dir / "config").write_text( (self.config_dir / "config").write_text(
@@ -126,10 +176,10 @@ class ManagedXrayTest(unittest.TestCase):
check=False, check=False,
) )
def start(self, *arguments, **environment): def start(self, *arguments, cli_prefix=None, **environment):
"""Запускает start и ждёт строку готовности; возвращает процесс.""" """Запускает start и ждёт строку готовности; возвращает процесс."""
process = subprocess.Popen( process = subprocess.Popen(
[str(CLI), "start"] + list(arguments), [str(CLI)] + (cli_prefix or ["start"]) + list(arguments),
cwd=str(self.root), cwd=str(self.root),
env=dict(self.environment, **environment), env=dict(self.environment, **environment),
text=True, text=True,
+115 -2
View File
@@ -33,8 +33,13 @@ assert_eq() {
PROXY_CLIENT_BIN="$TEST_ROOT/proxy-client" PROXY_CLIENT_BIN="$TEST_ROOT/proxy-client"
cat > "$PROXY_CLIENT_BIN" <<'EOF' cat > "$PROXY_CLIENT_BIN" <<'EOF'
#!/usr/bin/env bash #!/usr/bin/env bash
if [ -n "${FAKE_CALL_LOG:-}" ]; then
printf '<%s>\n' "$@" >> "$FAKE_CALL_LOG"
fi
if [ "$1" = env ]; then if [ "$1" = env ]; then
printf '%s\n' "$FAKE_ENV_OUTPUT" printf '%s\n' "$FAKE_ENV_OUTPUT"
elif [ "$#" -eq 3 ] && [ "$1" = --profile ] && [ "$2" = work ] && [ "$3" = env ]; then
printf '%s\n' "$FAKE_PROFILE_ENV_OUTPUT"
else else
printf '<%s>\n' "$@" printf '<%s>\n' "$@"
fi fi
@@ -69,6 +74,42 @@ source "$SHELL_MODULE"
) )
ok "proxy_on применяет восемь переменных" ok "proxy_on применяет восемь переменных"
# Выбор профиля меняет окружение только текущей оболочки.
(
unset "${PROXY_KEYS[@]}"
FAKE_ENV_OUTPUT="$VALID_ENV" \
FAKE_PROFILE_ENV_OUTPUT='HTTP_PROXY=http://work.test:4321
HTTPS_PROXY=http://work.test:4321
ALL_PROXY=socks5://work.test:4321
NO_PROXY=work.test,localhost
http_proxy=http://work.test:4321
https_proxy=http://work.test:4321
all_proxy=socks5://work.test:4321
no_proxy=work.test,localhost' \
proxy_on --profile work >/dev/null
assert_eq http://work.test:4321 "$HTTP_PROXY" "proxy_on не выбрал профиль work"
assert_eq socks5://work.test:4321 "$all_proxy" "SOCKS выбранного профиля"
assert_eq work.test,localhost "$NO_PROXY" "исключения выбранного профиля"
)
ok "proxy_on выбирает именованный профиль"
# Неверный синтаксис proxy_on отклоняется до изменения окружения.
assert_invalid_proxy_on() (
for key in "${PROXY_KEYS[@]}"; do export "$key=old"; done
local rc=0
FAKE_ENV_OUTPUT="$VALID_ENV" proxy_on "$@" >/dev/null 2>&1 || rc=$?
assert_eq 2 "$rc" "proxy_on принял неверные аргументы: $*"
for key in "${PROXY_KEYS[@]}"; do
assert_eq old "${!key}" "$key изменился после неверных аргументов"
done
)
assert_invalid_proxy_on env
assert_invalid_proxy_on --profile
assert_invalid_proxy_on --profile ""
assert_invalid_proxy_on --profile work extra
assert_invalid_proxy_on --unknown work
ok "proxy_on отклоняет неверные аргументы без изменения окружения"
# Ошибка CLI или непонятный вывод не меняют окружение ни на одну переменную. # Ошибка CLI или непонятный вывод не меняют окружение ни на одну переменную.
for case_spec in "1|$VALID_ENV" "0|$VALID_ENV for case_spec in "1|$VALID_ENV" "0|$VALID_ENV
EXTRA_PROXY=bad" "0|BROKEN-LINE"; do EXTRA_PROXY=bad" "0|BROKEN-LINE"; do
@@ -85,15 +126,33 @@ EXTRA_PROXY=bad" "0|BROKEN-LINE"; do
done done
ok "неуспешный env не меняет окружение" ok "неуспешный env не меняет окружение"
# Даже пригодный вывод не применяется, если проверка выбранного прокси не прошла.
(
for key in "${PROXY_KEYS[@]}"; do export "$key=old"; done
unset no_proxy
if FAKE_RC=1 FAKE_PROFILE_ENV_OUTPUT="$VALID_ENV" proxy_on --profile work >/dev/null 2>&1; then
fail "proxy_on принял отказ проверки выбранного профиля"
fi
for key in "${PROXY_KEYS[@]}"; do
if [[ "$key" == no_proxy ]]; then
[ -z "${!key+set}" ] || fail "$key появился после ошибки"
else
assert_eq old "${!key}" "$key изменился после ошибки выбранного профиля"
fi
done
)
ok "отказ проверки выбранного профиля сохраняет окружение"
# proxy_off очищает оба регистра. # proxy_off очищает оба регистра.
( (
for key in "${PROXY_KEYS[@]}"; do export "$key=x"; done for key in "${PROXY_KEYS[@]}"; do export "$key=x"; done
proxy_off >/dev/null FAKE_CALL_LOG="$TEST_ROOT/off-calls" proxy_off >/dev/null
[ ! -e "$TEST_ROOT/off-calls" ] || fail "proxy_off вызвал CLI"
for key in "${PROXY_KEYS[@]}"; do for key in "${PROXY_KEYS[@]}"; do
[ -z "${!key+set}" ] || fail "$key не очищен" [ -z "${!key+set}" ] || fail "$key не очищен"
done done
) )
ok "proxy_off очищает восемь переменных" ok "proxy_off очищает восемь переменных без вызова CLI"
# Обёртки передают CLI команду, аргументы и код возврата без искажений. # Обёртки передают CLI команду, аргументы и код возврата без искажений.
( (
@@ -116,6 +175,60 @@ ok "proxy_off очищает восемь переменных"
) )
ok "pclaude и pcodex передают аргументы и код выхода" ok "pclaude и pcodex передают аргументы и код выхода"
# Ведущий --proxy-profile относится к прокси, обычный --profile — к программе.
(
for wrapper in pclaude pcodex; do
rc=0
output="$(FAKE_RC=42 "$wrapper" --proxy-profile work --profile personal "a b" "")" || rc=$?
assert_eq 42 "$rc" "$wrapper не сохранил код возврата с профилем"
assert_eq $'<--profile>\n<work>\n<run>\n<-->\n<'"${wrapper#p}"$'>\n<--profile>\n<personal>\n<a b>\n<>' "$output" \
"$wrapper исказил выбор профиля или аргументы ребенка"
done
)
ok "обёртки выбирают профиль прокси и сохраняют аргументы ребенка"
# Отсутствующее имя профиля не должно запускать дочернюю команду.
(
for wrapper in pclaude pcodex; do
for empty_name in missing empty; do
args=(--proxy-profile)
if [[ "$empty_name" == empty ]]; then args+=(""); fi
rc=0
output="$("$wrapper" "${args[@]}" 2>/dev/null)" || rc=$?
assert_eq 2 "$rc" "$wrapper принял --proxy-profile без имени"
assert_eq "" "$output" "$wrapper вызвал CLI без имени профиля"
done
done
)
ok "обёртки отклоняют --proxy-profile без имени"
# Только ведущий --proxy-profile выбирает профиль прокси.
(
for wrapper in pclaude pcodex; do
output="$("$wrapper" --profile personal --proxy-profile child "a b" "")"
assert_eq $'<run>\n<-->\n<'"${wrapper#p}"$'>\n<--profile>\n<personal>\n<--proxy-profile>\n<child>\n<a b>\n<>' "$output" \
"$wrapper забрал аргументы ребенка"
done
)
ok "обычный --profile и неведущий --proxy-profile остаются ребенку"
# Запуск с профилем не меняет окружение вызывающей оболочки.
(
for key in "${PROXY_KEYS[@]}"; do export "$key=old"; done
unset no_proxy
for wrapper in pclaude pcodex; do
"$wrapper" --proxy-profile work >/dev/null
for key in "${PROXY_KEYS[@]}"; do
if [[ "$key" == no_proxy ]]; then
[ -z "${!key+set}" ] || fail "$wrapper экспортировал $key"
else
assert_eq old "${!key}" "$wrapper изменил $key"
fi
done
done
)
ok "обёртки сохраняют окружение вызывающей оболочки"
# .bash_custom подключает и общие функции, и proxy-модуль. # .bash_custom подключает и общие функции, и proxy-модуль.
( (
home="$TEST_ROOT/home" home="$TEST_ROOT/home"
+353
View File
@@ -0,0 +1,353 @@
"""Публичный CLI профилей и установка в изолированный корень."""
import os
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
CLI = Path(__file__).resolve().parents[1] / 'proxy-client/.local/bin/proxy-client'
class ProfilesServiceTest(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.root = Path(tmp.name)
self.config = self.root / 'config/proxy-client'
self.config.mkdir(parents=True)
self.env = dict(os.environ, HOME=str(self.root), XDG_CONFIG_HOME=str(self.root / 'config'),
XDG_STATE_HOME=str(self.root / 'state'))
def cli(self, *args):
return subprocess.run([str(CLI), *map(str, args)], env=self.env,
stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=15)
def test_unknown_profile_never_falls_back_even_for_stop(self):
for command in ('status', 'env', 'run', 'start', 'stop'):
result = self.cli('--profile', 'missing', command)
self.assertEqual(result.returncode, 1, result.stderr)
self.assertIn('неизвестный профиль', result.stderr)
result = self.cli('--profile', '../escape', 'stop')
self.assertEqual(result.returncode, 1)
self.assertIn('имя профиля', result.stderr)
def install_args(self):
source = self.root / 'source'
source.mkdir(exist_ok=True)
(source / 'config').write_text('[proxy]\ntype=xray\nname=Service\nno_proxy_extra=\naddress=127.0.0.1\nport=12345\n')
(source / 'default.url').write_text(
'vless://11111111-2222-4333-8444-555555555555@cdn.example:443'
'?type=xhttp&security=tls&encryption=none&sni=cdn.example&fp=chrome&path=%2Ftest\n')
(source / 'default.url').chmod(0o600)
xray = self.root / 'xray'
xray.write_text('#!/bin/sh\n[ "$1" != version ] || echo "Xray 26.10.10 (fake)"\nexit 0\n')
xray.chmod(0o755)
return ['service', 'install', '--config-dir', source, '--xray', xray,
'--root', self.root / 'stage']
def test_staged_install_is_standalone_and_preserves_config(self):
args = self.install_args()
result = self.cli(*args)
self.assertEqual(result.returncode, 0, result.stderr)
stage = self.root / 'stage'
unit = (stage / 'etc/systemd/system/proxy-client.service').read_text()
self.assertIn('User=proxy-client', unit)
self.assertIn('Restart=on-failure', unit)
self.assertIn('RuntimeDirectoryMode=0700', unit)
self.assertIn('StandardOutput=journal', unit)
self.assertNotIn(str(self.root), unit)
self.assertTrue((stage / 'usr/local/lib/proxy-client/xray').is_file())
self.assertTrue((stage / 'usr/lib/sysusers.d/proxy-client.conf').is_file())
config = stage / 'etc/proxy-client/config'
config.write_text(config.read_text() + '# preserved\n')
result = self.cli(*args)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('# preserved', config.read_text())
self.assertEqual((stage / 'etc/proxy-client/default.url').stat().st_mode & 0o777, 0o640)
self.assertIn('journald', result.stdout)
def test_install_defaults_copy_current_profile_and_local_xray(self):
args = self.install_args()
source, xray = args[3], args[5]
for name in ('config', 'default.url'):
(source / name).replace(self.config / name)
local_xray = self.root / '.local/bin/xray'
local_xray.parent.mkdir(parents=True)
xray.replace(local_xray)
result = self.cli('service', 'install', '--root', self.root / 'stage')
self.assertEqual(0, result.returncode, result.stderr)
settings = self.root / 'stage/etc/proxy-client'
for name in ('config', 'default.url'):
self.assertEqual((self.config / name).read_bytes(), (settings / name).read_bytes())
self.assertEqual(local_xray.read_bytes(),
(self.root / 'stage/usr/local/lib/proxy-client/xray').read_bytes())
def test_missing_settings_fail_before_sudo(self):
args = self.install_args()
fake = self.root / 'fake-os'
fake.mkdir()
(fake / 'sitecustomize.py').write_text('import os\nos.geteuid = lambda: 1000\n')
log = self.root / 'sudo-called'
sudo = fake / 'sudo'
sudo.write_text('#!/bin/sh\ntouch "$TEST_SUDO_LOG"\nexit 23\n')
sudo.chmod(0o755)
self.env.update(PYTHONPATH=str(fake), TEST_SUDO_LOG=str(log),
PATH=str(fake) + os.pathsep + os.environ['PATH'])
result = self.cli('service', 'install', '--install-only', '--xray', args[5])
self.assertEqual(1, result.returncode, result.stderr)
self.assertIn('настрой', result.stderr)
self.assertIn('терминал', result.stderr)
self.assertNotIn('No such file', result.stderr)
self.assertFalse(log.exists())
def test_install_can_select_named_profile(self):
args = self.install_args()
named = self.config / 'profiles/work'
named.parent.mkdir(parents=True)
args[3].rename(named)
result = self.cli('--profile', 'work', 'service', 'install', '--xray', args[5],
'--root', self.root / 'stage')
self.assertEqual(0, result.returncode, result.stderr)
self.assertEqual((named / 'config').read_bytes(),
(self.root / 'stage/etc/proxy-client/config').read_bytes())
ambiguous = self.cli('--profile', 'work', *args)
self.assertEqual(1, ambiguous.returncode)
self.assertIn('--config-dir', ambiguous.stderr)
def test_install_resolves_user_paths_before_sudo(self):
args = self.install_args()
for name in ('config', 'default.url'):
(args[3] / name).replace(self.config / name)
local_xray = self.root / '.local/bin/xray'
local_xray.parent.mkdir(parents=True)
args[5].replace(local_xray)
fake = self.root / 'fake-os'
fake.mkdir()
(fake / 'sitecustomize.py').write_text('import os\nos.geteuid = lambda: 1000\n')
sudo = fake / 'sudo'
sudo.write_text('#!/usr/bin/env python3\nimport json, os, sys\n'
'open(os.environ["TEST_SUDO_ARGS"], "w").write(json.dumps(sys.argv[1:]))\n'
'sys.exit(23)\n')
sudo.chmod(0o755)
log = self.root / 'sudo-args.json'
self.env.update(PYTHONPATH=str(fake), TEST_SUDO_ARGS=str(log),
PATH=str(fake) + os.pathsep + os.environ['PATH'])
result = self.cli('service', 'install', '--install-only', '--enable', '--start')
self.assertEqual(23, result.returncode, result.stderr)
command = json.loads(log.read_text())
self.assertEqual(['service', 'install', '--config-dir', str(self.config),
'--xray', str(local_xray), '--enable', '--start'], command[3:])
self.assertEqual('--', command[0])
self.assertEqual(str(CLI.resolve()), command[2])
def test_direct_sudo_uses_invoking_users_home(self):
args = self.install_args()
source = self.root / '.config/proxy-client'
source.parent.mkdir()
args[3].rename(source)
xray = self.root / '.local/bin/xray'
xray.parent.mkdir(parents=True)
args[5].replace(xray)
fake = self.root / 'fake-os'
fake.mkdir()
(fake / 'sitecustomize.py').write_text(
'import os, pwd, types\nos.geteuid = lambda: 0\n'
'def user(uid):\n'
' assert uid == 12345\n'
' return types.SimpleNamespace(pw_dir=os.environ["TEST_USER_HOME"])\n'
'pwd.getpwuid = user\n')
self.env.pop('XDG_CONFIG_HOME')
self.env.update(HOME=str(self.root / 'root-home'), SUDO_UID='12345',
TEST_USER_HOME=str(self.root), PYTHONPATH=str(fake))
result = self.cli('service', 'install', '--root', self.root / 'stage')
self.assertEqual(0, result.returncode, result.stderr)
self.assertEqual((source / 'config').read_bytes(),
(self.root / 'stage/etc/proxy-client/config').read_bytes())
def test_install_refuses_foreign_files_before_writing(self):
args = self.install_args()
target = self.root / 'stage/etc/systemd/system/proxy-client.service'
target.parent.mkdir(parents=True)
target.write_text('foreign unit')
result = self.cli(*args)
self.assertEqual(result.returncode, 1)
self.assertEqual(target.read_text(), 'foreign unit')
self.assertFalse((self.root / 'stage/usr/local/lib/proxy-client/xray').exists())
def test_install_reports_journal_storage_without_changing_policy(self):
args = self.install_args()
policy = self.root / 'stage/etc/systemd/journald.conf.d/storage.conf'
policy.parent.mkdir(parents=True)
policy.write_text('[Journal]\nStorage=volatile\n')
result = self.cli(*args)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('Storage=volatile', result.stdout)
self.assertIn('не сохраняется', result.stdout)
self.assertEqual(policy.read_text(), '[Journal]\nStorage=volatile\n')
def test_update_does_not_require_original_home_profile(self):
args = self.install_args()
first = self.cli(*args)
self.assertEqual(0, first.returncode, first.stderr)
for path in (self.root / 'source').iterdir():
path.unlink()
(self.root / 'source').rmdir()
updated = self.cli(*args)
self.assertEqual(0, updated.returncode, updated.stderr)
def test_default_update_preserves_service_settings_without_home_files(self):
args = self.install_args()
first = self.cli(*args)
self.assertEqual(0, first.returncode, first.stderr)
settings = self.root / 'stage/etc/proxy-client'
saved = {name: (settings / name).read_bytes() for name in ('config', 'default.url')}
for path in args[3].iterdir():
path.unlink()
args[3].rmdir()
args[5].unlink()
# В PATH только Python для shebang тестового CLI; Xray там отсутствует.
path = self.root / 'path'
path.mkdir()
(path / 'python3').symlink_to(sys.executable)
self.env['PATH'] = str(path)
updated = self.cli('service', 'install', '--root', self.root / 'stage')
self.assertEqual(0, updated.returncode, updated.stderr)
for name, data in saved.items():
self.assertEqual(data, (settings / name).read_bytes())
def test_unspecified_journal_storage_is_not_guessed(self):
result = self.cli(*self.install_args())
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn('Storage не задан', result.stdout)
def test_restrictive_umask_keeps_service_executable_accessible(self):
args = self.install_args()
result = subprocess.run(['bash', '-c', 'umask 077; exec "$@"', 'bash',
str(CLI), *map(str, args)], env=self.env,
capture_output=True, text=True, timeout=15)
self.assertEqual(0, result.returncode, result.stderr)
stage = self.root / 'stage'
for name in ('usr', 'usr/local', 'usr/local/lib', 'usr/local/lib/proxy-client'):
self.assertEqual(0o755, (stage / name).stat().st_mode & 0o777, name)
def test_interrupted_install_or_update_can_be_retried(self):
args = self.install_args()
injection = self.root / 'fault'
injection.mkdir()
(injection / 'sitecustomize.py').write_text(
'import os, errno\noriginal = os.replace\n'
'def replace(src, dst):\n'
' if str(dst).endswith("/xray"):\n'
' raise OSError(errno.ENOSPC, "synthetic disk full")\n'
' return original(src, dst)\n'
'os.replace = replace\n')
candidate = self.root / 'new-cli'
candidate.write_bytes(CLI.read_bytes() + b'\n# next release\n')
candidate.chmod(0o755)
for updating in (False, True):
with self.subTest(updating=updating):
args[-1] = self.root / ('update' if updating else 'fresh')
if updating:
result = self.cli(*args)
self.assertEqual(0, result.returncode, result.stderr)
command = [str(candidate), *map(str, args)]
failed = subprocess.run(command, env=dict(self.env, PYTHONPATH=str(injection)),
capture_output=True, text=True, timeout=15)
self.assertNotEqual(0, failed.returncode)
retried = subprocess.run(command, env=self.env, capture_output=True,
text=True, timeout=15)
self.assertEqual(0, retried.returncode, retried.stderr)
def test_system_actions_require_explicit_flags(self):
args = self.install_args()[:-2]
stage = self.root / 'system'
fake = self.root / 'system-api'
fake.mkdir()
log = self.root / 'actions'
# Подмена только ОС: корень файловой системы, root/group API и
# внешние systemd-команды. CLI и установщик выполняются целиком.
(fake / 'sitecustomize.py').write_text('''
import os, pathlib, grp, types
stage = os.environ['TEST_INSTALL_ROOT']
OriginalPath = type(pathlib.Path())
class StagedPath:
home = staticmethod(OriginalPath.home)
def __new__(cls, *parts, **kwargs):
if parts == ('/',):
parts = (stage,)
return OriginalPath(*parts, **kwargs)
pathlib.Path = StagedPath
os.geteuid = lambda: 0
def group(name):
assert os.path.exists(os.environ['TEST_INSTALL_ROOT'] + '/group-created')
return types.SimpleNamespace(gr_gid=os.getgid())
grp.getgrnam = group
def chown(path, uid, gid):
assert str(path).startswith(stage + '/')
assert os.path.exists(stage + '/group-created')
with open(os.environ['TEST_ACTION_LOG'], 'a') as stream:
stream.write('chown ' + os.path.basename(path) + '\\n')
os.chown = chown
''')
for command in ('systemd-sysusers', 'systemctl'):
path = fake / command
path.write_text('#!/bin/sh\n'
'printf "%s %s\\n" "${0##*/}" "$*" >> "$TEST_ACTION_LOG"\n'
'if [ "${0##*/}" = systemd-sysusers ]; then\n'
' touch "$TEST_INSTALL_ROOT/group-created"\nfi\n')
path.chmod(0o755)
env = dict(self.env, PYTHONPATH=str(fake), TEST_INSTALL_ROOT=str(stage),
TEST_ACTION_LOG=str(log), PATH=str(fake) + ':' + os.environ['PATH'])
for flags in ([], ['--enable'], ['--start'], ['--enable', '--start']):
with self.subTest(flags=flags):
log.write_text('')
result = subprocess.run([str(CLI), *map(str, args), *flags], env=env,
capture_output=True, text=True, timeout=15)
self.assertEqual(0, result.returncode, result.stderr)
calls = log.read_text().splitlines()
self.assertTrue(calls[0].startswith('systemd-sysusers '), calls)
self.assertEqual(['chown proxy-client', 'chown config', 'chown default.url'], calls[1:4])
self.assertIn('systemctl daemon-reload', calls)
self.assertEqual('--enable' in flags, 'systemctl enable proxy-client.service' in calls)
self.assertEqual('--start' in flags, 'systemctl start proxy-client.service' in calls)
self.assertFalse(any('restart' in call for call in calls))
def test_invalid_service_requests_do_not_install_files(self):
args = self.install_args()
for extra in (['--enable'], ['--start']):
result = self.cli(*args, *extra)
self.assertNotEqual(0, result.returncode)
self.assertFalse((self.root / 'stage').exists())
result = self.cli('--profile', 'work', *args)
self.assertNotEqual(0, result.returncode)
self.assertFalse((self.root / 'stage').exists())
config = self.root / 'source/config'
original = config.read_text()
for invalid in [original.replace('type=xray', 'type=external')] + [
original.replace('127.0.0.1', address)
for address in ('0.0.0.0', '::', '192.0.2.1', '2001:db8::1', 'localhost')]:
config.write_text(invalid)
result = self.cli(*args)
self.assertNotEqual(0, result.returncode)
self.assertFalse((self.root / 'stage').exists())
def test_manual_and_service_run_reject_non_loopback_before_start(self):
args = self.install_args()
config_path = args[3] / 'config'
original = config_path.read_text()
for name in ('config', 'default.url'):
(args[3] / name).replace(self.config / name)
runtime = self.root / 'service-runtime'
commands = [('start',), ('service', 'run', '--config-dir', self.config,
'--runtime-dir', runtime, '--xray', args[5])]
for address in ('0.0.0.0', '::', '192.0.2.1', '2001:db8::1', 'localhost'):
(self.config / 'config').write_text(original.replace('127.0.0.1', address))
for command in commands:
with self.subTest(address=address, command=command):
result = self.cli(*command)
self.assertEqual(1, result.returncode, result.stderr)
self.assertIn('loopback', result.stderr)
self.assertFalse(runtime.exists())
self.assertFalse((self.root / 'state').exists())
+520
View File
@@ -0,0 +1,520 @@
"""Интерактивная установка через CLI; системные действия подменены."""
import json
import hashlib
import os
import pty
import select
import subprocess
import sys
import tempfile
import time
import unittest
import zipfile
from pathlib import Path
CLI = Path(__file__).resolve().parents[1] / 'proxy-client/.local/bin/proxy-client'
URL = ('vless://11111111-2222-4333-8444-555555555555@cdn.example:443'
'?type=xhttp&security=tls&encryption=none&sni=cdn.example&fp=chrome&path=%2FSECRET')
CONFIG = '[proxy]\ntype=xray\nname=Test\naddress=127.0.0.1\nport=12334\nno_proxy_extra=.example\n'
class ServiceSetupTest(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name)
self.config = self.root / 'config/proxy-client'
self.config.mkdir(parents=True)
self.fake = self.root / 'bin'
self.fake.mkdir()
self.log = self.root / 'actions'
self.stage = self.root / 'system'
self.env = dict(os.environ, HOME=str(self.root / 'home'),
XDG_CONFIG_HOME=str(self.root / 'config'),
XDG_STATE_HOME=str(self.root / 'state'),
PYTHONPATH=str(self.fake), TEST_ROOT=str(self.root),
PATH=str(self.fake) + os.pathsep + os.defpath)
(self.fake / 'python3').symlink_to(sys.executable)
(self.fake / 'sitecustomize.py').write_text('''
import os, sys, pathlib, getpass, grp, types, socket, urllib.request, io
OriginalPath = type(pathlib.Path())
class StagedPath:
home = staticmethod(OriginalPath.home)
def __new__(cls, *parts, **kwargs):
if parts == ('/',):
parts = (os.environ['TEST_ROOT'] + '/system',)
return OriginalPath(*parts, **kwargs)
pathlib.Path = StagedPath
os.geteuid = lambda: 0 if os.environ.get('TEST_PRIVILEGED') else 1000
sys.stdin.isatty = lambda: not os.environ.get('TEST_PRIVILEGED') and not os.environ.get('TEST_NO_TTY')
if not os.environ.get('TEST_REAL_GETPASS'):
getpass.getpass = lambda prompt: input(prompt)
grp.getgrnam = lambda name: types.SimpleNamespace(gr_gid=os.getgid())
def chown(path, uid, gid):
assert str(path).startswith(os.environ['TEST_ROOT'] + '/system/')
os.chown = chown
class LocalSocket:
def __init__(self, *args, **kwargs): pass
def __enter__(self): return self
def __exit__(self, *args): pass
def close(self): pass
def setsockopt(self, *args): pass
def bind(self, address):
busy = os.environ.get('TEST_PORT_BUSY')
if busy == 'manual':
busy = os.path.exists(os.environ['TEST_ROOT'] + '/manual-active')
if busy: raise OSError('busy')
socket.socket = LocalSocket
def connect(*args, **kwargs):
if not os.path.exists(os.environ['TEST_ROOT'] + '/system/active'):
raise OSError('not ready')
return LocalSocket()
socket.create_connection = connect
original_readlink = os.readlink
original_read_bytes = OriginalPath.read_bytes
def readlink(path, *args, **kwargs):
if str(path) == '/proc/4321/exe':
if not os.path.exists(os.environ['TEST_ROOT'] + '/manual-active'): raise FileNotFoundError()
return os.environ['TEST_ROOT'] + '/bin/xray'
return original_readlink(path, *args, **kwargs)
def read_bytes(path):
if str(path) == '/proc/4322/cmdline':
if not os.path.exists(os.environ['TEST_ROOT'] + '/manual-active'): raise FileNotFoundError()
return b'proxy-client start'
return original_read_bytes(path)
os.readlink = readlink
OriginalPath.read_bytes = read_bytes
def kill(pid, signum):
assert not os.environ.get('TEST_PRIVILEGED'), 'нельзя останавливать пользовательский процесс от root'
assert pid == 4322, 'чужой PID'
os.unlink(os.environ['TEST_ROOT'] + '/manual-active')
with open(os.environ['TEST_ROOT'] + '/actions', 'a') as stream:
stream.write('["manual-stop"]\\n')
os.kill = kill
def urlopen(url, **kwargs):
if os.environ.get('TEST_DOWNLOAD_FAIL'): raise OSError('download failed')
with open(os.environ['TEST_ROOT'] + '/actions', 'a') as stream:
stream.write('["download"]\\n')
suffix = 'digest' if url.endswith('.dgst') else 'archive'
return io.BytesIO(open(os.environ['TEST_ROOT'] + '/' + suffix, 'rb').read())
urllib.request.urlopen = urlopen
''')
self.executable('sudo', '''#!/usr/bin/env python3
import json, os, sys
args = sys.argv[1:]
with open(os.environ['TEST_ROOT'] + '/actions', 'a') as stream:
stream.write(json.dumps(['sudo'] + args) + '\\n')
if os.environ.get('TEST_FAIL_SUDO'): sys.exit(23)
if args[0] == '--': args.pop(0)
os.environ['TEST_PRIVILEGED'] = '1'
os.environ['HOME'] = os.environ['TEST_ROOT'] + '/root-home'
os.environ.pop('XDG_CONFIG_HOME', None)
os.environ.pop('XDG_STATE_HOME', None)
os.execvpe(args[0], args, os.environ)
''')
self.executable('systemd-sysusers', '#!/bin/sh\nexit 0\n')
self.executable('systemctl', '''#!/usr/bin/env python3
import json, os, sys
from pathlib import Path
root = Path(os.environ['TEST_ROOT'])
with (root / 'actions').open('a') as stream:
stream.write(json.dumps(['systemctl'] + sys.argv[1:]) + '\\n')
action = sys.argv[1]
active = root / 'system/active'
if action in ('start', 'restart') or (action == 'enable' and '--now' in sys.argv):
if os.environ.get('TEST_FAIL_START'): sys.exit(1)
if not os.environ.get('TEST_NOT_READY'): active.touch()
if os.environ.get('TEST_CHANGE_LINK_AT_START'):
link = root / 'config/proxy-client/default.url'
link.write_text(link.read_text().replace('cdn.example:443', 'after-start.example:443'))
if (action == 'stop' or (action == 'disable' and '--now' in sys.argv)) and active.exists(): active.unlink()
if action == 'is-active': sys.exit(0 if active.exists() else 1)
''')
self.executable('xray', '#!/bin/sh\n'
'[ "$1" != version ] || echo "Xray 26.10.10 (fake)"\nexit 0\n')
def executable(self, name, text):
path = self.fake / name
path.write_text(text)
path.chmod(0o755)
def ready_profile(self):
(self.config / 'config').write_text(CONFIG)
(self.config / 'default.url').write_text(URL + '\n')
(self.config / 'default.url').chmod(0o600)
def running_profile(self):
self.ready_profile()
state = self.root / 'state/proxy-client'
state.mkdir(parents=True, exist_ok=True)
(state / 'state.json').write_text(json.dumps({
'supervisor_pid': 4322, 'xray_pid': 4321, 'xray_path': str(self.fake / 'xray'),
'address': '127.0.0.1', 'port': 12334}))
(self.root / 'manual-active').touch()
def cli(self, answers='', *args):
return subprocess.run([str(CLI), 'service', 'install', *args], input=answers,
env=self.env, capture_output=True, text=True, timeout=20)
def test_declining_plan_changes_nothing(self):
self.ready_profile()
result = self.cli('n\n')
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn('отменена', result.stdout)
self.assertFalse(self.log.exists())
self.assertFalse(self.stage.exists())
self.assertEqual(CONFIG, (self.config / 'config').read_text())
def test_fresh_setup_collects_settings_without_writing_before_confirmation(self):
result = self.cli(URL + '\n\nn\n')
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn('12334', result.stdout)
self.assertIn('отменена', result.stdout)
self.assertNotIn(URL, result.stdout + result.stderr)
self.assertFalse(self.log.exists())
self.assertEqual([], list(self.config.iterdir()))
def test_dangling_profile_symlink_is_reported_before_secret_input(self):
(self.config / 'config').write_text(CONFIG)
(self.config / 'default.url').symlink_to(self.root / 'missing.url')
result = self.cli(URL + '\ny\n')
self.assertEqual(1, result.returncode)
self.assertIn('битый симлинк', result.stderr)
self.assertNotIn('ввод скрыт', result.stdout)
self.assertNotIn('rm -- ', result.stderr)
self.assertFalse(self.log.exists())
self.assertFalse((self.root / 'state/proxy-client/service-setup.json').exists())
def test_fresh_setup_installs_starts_and_configures_shell_profile(self):
result = self.cli(URL + '\n\ny\n')
self.assertEqual(0, result.returncode, result.stderr)
settings = self.stage / 'etc/proxy-client'
self.assertIn('type = xray', (settings / 'config').read_text())
self.assertEqual(URL + '\n', (settings / 'default.url').read_text())
self.assertEqual(0o640, (settings / 'default.url').stat().st_mode & 0o777)
home_config = (self.config / 'config').read_text()
self.assertIn('type = external', home_config)
self.assertIn('address = 127.0.0.1', home_config)
self.assertIn('port = 12334', home_config)
self.assertTrue((self.stage / 'active').exists())
self.assertNotIn(URL, result.stdout + result.stderr + self.log.read_text())
def test_readiness_check_does_not_request_sudo(self):
self.ready_profile()
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
actions = [json.loads(line) for line in self.log.read_text().splitlines()]
self.assertIn(['systemctl', 'is-active', '--quiet', 'proxy-client.service'], actions)
self.assertFalse(any(action[0] == 'sudo' and 'is-active' in action for action in actions))
def test_yes_accepts_file_without_a_terminal(self):
self.env['TEST_NO_TTY'] = '1'
profile = self.root / 'connection.url'
profile.write_text(URL + '\n')
profile.chmod(0o600)
result = self.cli('', '--auto-yes', '--file', str(profile), '--port', '12335')
self.assertEqual(0, result.returncode, result.stderr)
self.assertNotIn('Продолжить?', result.stdout)
self.assertIn('port = 12335', (self.config / 'config').read_text())
self.assertNotIn(URL, result.stdout + result.stderr + self.log.read_text())
def test_no_terminal_requires_an_explicit_installation_mode(self):
self.running_profile()
self.env['TEST_NO_TTY'] = '1'
result = self.cli('')
self.assertEqual(1, result.returncode)
self.assertIn('--yes', result.stderr)
self.assertIn('--install-only', result.stderr)
self.assertFalse(self.log.exists())
self.assertFalse(self.stage.exists())
self.assertTrue((self.root / 'manual-active').exists())
def test_update_keeps_system_settings_without_home_profile(self):
self.ready_profile()
first = self.cli('y\n')
self.assertEqual(0, first.returncode, first.stderr)
settings = self.stage / 'etc/proxy-client'
before = {name: (settings / name).read_bytes() for name in ('config', 'default.url')}
for path in self.config.iterdir():
path.unlink()
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn('обнов', result.stdout.lower())
for name, content in before.items():
self.assertEqual(content, (settings / name).read_bytes())
self.assertFalse((self.config / 'config').exists())
def test_migration_stops_only_own_process_after_files_are_installed(self):
self.running_profile()
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
self.assertFalse((self.root / 'manual-active').exists())
actions = [json.loads(line) for line in self.log.read_text().splitlines()]
self.assertLess(actions.index(['systemctl', 'daemon-reload']), actions.index(['manual-stop']))
self.assertLess(actions.index(['manual-stop']),
actions.index(['systemctl', 'enable', '--now', 'proxy-client.service']))
backups = list(self.config.glob('config.before-service.*'))
self.assertEqual(1, len(backups))
self.assertEqual(CONFIG, backups[0].read_text())
self.assertIn('type = external', (self.config / 'config').read_text())
def test_install_only_can_be_completed_by_the_wizard(self):
self.running_profile()
first = self.cli('', '--install-only')
self.assertEqual(0, first.returncode, first.stderr)
self.env['TEST_PORT_BUSY'] = 'manual'
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
self.assertFalse((self.root / 'manual-active').exists())
self.assertTrue((self.stage / 'active').exists())
self.assertIn('type = external', (self.config / 'config').read_text())
def test_installed_different_profile_is_not_replaced_or_stopped(self):
self.running_profile()
first = self.cli('', '--install-only')
self.assertEqual(0, first.returncode, first.stderr)
system_config = self.stage / 'etc/proxy-client/config'
changed = system_config.read_text().replace('12334', '12335')
system_config.write_text(changed)
result = self.cli('', '--yes')
self.assertEqual(1, result.returncode, result.stderr)
self.assertTrue((self.root / 'manual-active').exists())
self.assertEqual(CONFIG, (self.config / 'config').read_text())
self.assertEqual(changed, system_config.read_text())
self.assertNotIn('["systemctl", "stop"', self.log.read_text())
snapshot = self.root / 'state/proxy-client/service-setup.json'
self.assertTrue(snapshot.exists())
self.assertIn('rm -- ' + str(snapshot), result.stderr)
self.assertIn('/etc/proxy-client', result.stderr)
def test_start_failure_preserves_home_and_explains_recovery(self):
self.running_profile()
self.env['TEST_FAIL_START'] = '1'
result = self.cli('', '--yes')
self.assertEqual(1, result.returncode)
self.assertEqual(CONFIG, (self.config / 'config').read_text())
self.assertIn('proxy-client start', result.stderr)
self.assertIn('disable', self.log.read_text())
self.assertEqual([], list(self.config.glob('config.before-service.*')))
def test_missing_xray_is_downloaded_and_verified_before_installation(self):
binary = (self.fake / 'xray').read_bytes()
(self.fake / 'xray').unlink()
with zipfile.ZipFile(str(self.root / 'archive'), 'w') as archive:
archive.writestr('xray', binary)
digest = hashlib.sha256((self.root / 'archive').read_bytes()).hexdigest()
(self.root / 'digest').write_text('SHA2-256= ' + digest + '\n')
self.ready_profile()
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
self.assertEqual(binary, (self.stage / 'usr/local/lib/proxy-client/xray').read_bytes())
actions = [json.loads(line) for line in self.log.read_text().splitlines()]
self.assertEqual(['download'], actions[0])
self.assertFalse((self.root / 'home/.local/bin/xray').exists())
def test_retry_after_failed_start_completes_the_transition(self):
self.running_profile()
self.env['TEST_FAIL_START'] = '1'
self.assertEqual(1, self.cli('', '--yes').returncode)
self.env.pop('TEST_FAIL_START')
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn('type = external', (self.config / 'config').read_text())
def test_file_option_cannot_be_silently_ignored_in_install_only_mode(self):
self.ready_profile()
result = self.cli('', '--install-only', '--file', str(self.config / 'default.url'))
self.assertEqual(1, result.returncode)
self.assertIn('--file', result.stderr)
self.assertFalse(self.log.exists())
def test_update_downloads_missing_binary_without_asking_for_profile(self):
self.ready_profile()
self.assertEqual(0, self.cli('', '--yes').returncode)
binary = (self.fake / 'xray').read_bytes()
(self.fake / 'xray').unlink()
(self.stage / 'usr/local/lib/proxy-client/xray').unlink()
with zipfile.ZipFile(str(self.root / 'archive'), 'w') as archive:
archive.writestr('xray', binary)
digest = hashlib.sha256((self.root / 'archive').read_bytes()).hexdigest()
(self.root / 'digest').write_text('SHA2-256= ' + digest + '\n')
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
self.assertEqual(binary, (self.stage / 'usr/local/lib/proxy-client/xray').read_bytes())
def test_preparation_failures_do_not_stop_manual_process(self):
self.running_profile()
self.env['TEST_FAIL_SUDO'] = '1'
failed = self.cli('', '--yes')
self.assertEqual(23, failed.returncode, failed.stderr)
self.assertTrue((self.root / 'manual-active').exists())
self.assertEqual(CONFIG, (self.config / 'config').read_text())
pending = self.root / 'state/proxy-client/service-setup.json'
self.assertEqual(0o600, pending.stat().st_mode & 0o777)
self.env.pop('TEST_FAIL_SUDO')
(self.fake / 'xray').unlink()
self.env['TEST_DOWNLOAD_FAIL'] = '1'
failed = self.cli('', '--yes')
self.assertEqual(1, failed.returncode)
self.assertTrue((self.root / 'manual-active').exists())
self.assertFalse(self.stage.exists())
def test_retry_from_scratch_uses_private_snapshot_without_new_secret_input(self):
self.env['TEST_FAIL_START'] = '1'
failed = self.cli(URL + '\n\ny\n')
self.assertEqual(1, failed.returncode, failed.stderr)
self.assertFalse((self.config / 'config').exists())
self.env.pop('TEST_FAIL_START')
self.env['TEST_NO_TTY'] = '1'
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn('type = external', (self.config / 'config').read_text())
self.assertFalse((self.root / 'state/proxy-client/service-setup.json').exists())
def test_changed_link_blocks_stale_snapshot_and_reset_uses_new_settings(self):
self.ready_profile()
self.env['TEST_FAIL_SUDO'] = '1'
self.assertEqual(23, self.cli('', '--yes').returncode)
self.env.pop('TEST_FAIL_SUDO')
changed = URL.replace('cdn.example:443', 'new.example:443')
(self.config / 'default.url').write_text(changed + '\n')
before = self.log.read_bytes()
result = self.cli('', '--yes')
self.assertEqual(1, result.returncode, result.stderr)
self.assertIn('ссылки изменился', result.stderr)
snapshot = self.root / 'state/proxy-client/service-setup.json'
self.assertIn('rm -- ' + str(snapshot), result.stderr)
self.assertFalse(self.stage.exists())
self.assertEqual(before, self.log.read_bytes())
snapshot.unlink() # Пользователь явно сбросил снимок по подсказке CLI.
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
self.assertEqual(changed + '\n', (self.stage / 'etc/proxy-client/default.url').read_text())
def test_link_edit_after_service_start_does_not_abort_the_confirmed_transition(self):
self.running_profile()
self.env['TEST_CHANGE_LINK_AT_START'] = '1'
result = self.cli('', '--yes')
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn('type = external', (self.config / 'config').read_text())
self.assertIn('after-start.example', (self.config / 'default.url').read_text())
self.assertEqual(URL + '\n', (self.stage / 'etc/proxy-client/default.url').read_text())
self.assertFalse((self.root / 'state/proxy-client/service-setup.json').exists())
def test_pending_override_explains_how_to_reset_the_snapshot(self):
self.ready_profile()
self.env['TEST_FAIL_SUDO'] = '1'
self.assertEqual(23, self.cli('', '--yes').returncode)
result = self.cli('', '--yes', '--port', '12335')
self.assertEqual(1, result.returncode)
self.assertIn('rm -- ', result.stderr)
def test_retry_refuses_changed_system_settings_before_stopping_manual_process(self):
self.running_profile()
self.env['TEST_FAIL_START'] = '1'
self.assertEqual(1, self.cli('', '--yes').returncode)
self.env.pop('TEST_FAIL_START')
self.running_profile()
config = self.stage / 'etc/proxy-client/config'
changed = config.read_text().replace('12334', '12335')
config.write_text(changed)
result = self.cli('', '--yes')
self.assertEqual(1, result.returncode)
self.assertIn('отличаются от снимка', result.stderr)
self.assertTrue((self.root / 'manual-active').exists())
self.assertEqual(changed, config.read_text())
self.assertEqual(CONFIG, (self.config / 'config').read_text())
def test_bad_url_and_invalid_port_do_not_install(self):
result = self.cli('not-a-vless-link\n')
self.assertEqual(1, result.returncode)
self.assertNotIn('not-a-vless-link', result.stdout + result.stderr)
result = self.cli('', '--yes', '--port', '0')
self.assertEqual(1, result.returncode)
self.assertFalse(self.log.exists())
self.assertFalse(self.stage.exists())
def test_foreign_port_is_never_stopped(self):
self.ready_profile()
self.env['TEST_PORT_BUSY'] = '1'
result = self.cli('', '--yes')
self.assertEqual(1, result.returncode)
self.assertIn('занят', result.stderr)
self.assertFalse(self.log.exists())
self.assertEqual(CONFIG, (self.config / 'config').read_text())
def test_corrupt_transition_snapshot_gets_clear_error(self):
self.ready_profile()
pending = self.root / 'state/proxy-client/service-setup.json'
pending.parent.mkdir(parents=True)
pending.write_text('{}')
pending.chmod(0o600)
result = self.cli('', '--yes')
self.assertEqual(1, result.returncode)
self.assertIn('снимок', result.stderr)
self.assertNotIn('Traceback', result.stderr)
self.assertFalse(self.log.exists())
def test_retry_does_not_accept_a_foreign_listener_as_the_service(self):
self.ready_profile()
self.env['TEST_FAIL_START'] = '1'
self.assertEqual(1, self.cli('', '--yes').returncode)
self.env.pop('TEST_FAIL_START')
self.env['TEST_PORT_BUSY'] = '1'
result = self.cli('', '--yes')
self.assertEqual(1, result.returncode)
self.assertEqual(CONFIG, (self.config / 'config').read_text())
def test_malformed_config_does_not_expose_its_contents(self):
(self.config / 'config').write_text('PRIVATE_SENTINEL_not_ini\n')
result = self.cli('', '--yes')
self.assertEqual(1, result.returncode)
self.assertNotIn('PRIVATE_SENTINEL', result.stdout + result.stderr)
self.assertNotIn('Traceback', result.stderr)
self.assertFalse(self.log.exists())
def test_retry_keeps_custom_file_in_manual_recovery_command(self):
self.running_profile()
custom = self.root / 'custom.url'
(self.config / 'default.url').rename(custom)
self.env['TEST_FAIL_START'] = '1'
self.assertEqual(1, self.cli('', '--yes', '--file', str(custom)).returncode)
self.running_profile()
(self.config / 'default.url').unlink()
result = self.cli('', '--yes')
self.assertEqual(1, result.returncode)
self.assertIn('proxy-client start -f ' + str(custom), result.stderr)
def test_secret_input_is_not_echoed_in_a_real_terminal(self):
master, slave = pty.openpty()
self.addCleanup(os.close, master)
process = subprocess.Popen([str(CLI), 'service', 'install'], stdin=slave,
stdout=slave, stderr=slave, start_new_session=True,
env=dict(self.env, TEST_REAL_GETPASS='1'))
os.close(slave)
output = bytearray()
def until(prompt):
deadline = time.monotonic() + 10
while prompt.encode() not in output:
if time.monotonic() > deadline:
self.fail('не появился запрос ввода: ' + prompt)
if select.select([master], [], [], 0.1)[0]:
output.extend(os.read(master, 4096))
try:
until('ввод скрыт')
os.write(master, (URL + '\n').encode())
until('Локальный порт')
os.write(master, b'\n')
until('Продолжить?')
os.write(master, b'n\n')
self.assertEqual(0, process.wait(timeout=10))
self.assertNotIn(URL.encode(), output)
self.assertFalse(self.log.exists())
finally:
if process.poll() is None:
process.kill()
process.wait()
if __name__ == '__main__':
unittest.main()