ddadmin
384b4f215f
Заданы права системных каталогов независимо от umask, добавлен журнал незавершённой установки и исправлена подсказка прав профиля. Проверены явные флаги systemd через подставные API ОС. Проверка: 72 теста на Python 3.8 и 3.14, оба shell-набора. Настоящий boot/systemd остаётся непроверенным.
213 lines
11 KiB
Python
213 lines
11 KiB
Python
"""Публичный CLI профилей и установка в изолированный корень."""
|
|
import os
|
|
import subprocess
|
|
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,
|
|
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_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_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:
|
|
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())
|