fix(proxy-client): исправлено восстановление установки службы

Заданы права системных каталогов независимо от umask, добавлен журнал незавершённой установки и исправлена подсказка прав профиля. Проверены явные флаги systemd через подставные API ОС.

Проверка: 72 теста на Python 3.8 и 3.14, оба shell-набора. Настоящий boot/systemd остаётся непроверенным.
This commit is contained in:
ddadmin committed 2026-11-02 22:30:43 +01:00
1 parent b34eb6ac4c
commit 384b4f215f
5 files changed
+170 -29

No files matched your search

+9
View File
@@ -143,6 +143,15 @@ class ManagedXrayTest(unittest.TestCase):
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"):
self.config_dir.mkdir(parents=True, exist_ok=True)
(self.config_dir / "config").write_text(
+109
View File
@@ -101,3 +101,112 @@ class ProfilesServiceTest(unittest.TestCase):
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:
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', '192.0.2.1')):
config.write_text(invalid)
result = self.cli(*args)
self.assertNotEqual(0, result.returncode)
self.assertFalse((self.root / 'stage').exists())