165 lines
9.7 KiB
Python
165 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Bounded source/config and unprivileged provisioning tests; never starts services."""
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(sys.argv.pop(1)).resolve() if len(sys.argv) > 1 else Path(__file__).resolve().parents[2]
|
|
UNIT = 'image-recipe/configs/archipelago.service'
|
|
PREP = 'image-recipe/configs/prepare-terminal-workspace.sh'
|
|
BUILDER = 'image-recipe/_archived/build-auto-installer-iso.sh'
|
|
UPDATE = 'scripts/self-update.sh'
|
|
|
|
class WorkspaceContract(unittest.TestCase):
|
|
def test_default_and_legacy_paths_are_narrowly_writable_and_provisioned(self):
|
|
source = (ROOT / 'core/archipelago/src/api/handler/terminal.rs').read_text()
|
|
default = source.split('fn default_workspace()', 1)[1].split('async fn runtime_workspace', 1)[0]
|
|
self.assertIn('std::env::var_os("ARCHIPELAGO_RUNTIME_WORKSPACE")', default)
|
|
self.assertIn('std::env::var_os("HOME")', default)
|
|
self.assertIn('.join("Work")', default)
|
|
self.assertIn('runtime_workspace(&workspace, !workspace_was_requested).await?', source)
|
|
self.assertIn('pane_in_workspace(&tmux_name, &workspace).await', source)
|
|
self.assertNotIn('workspace_path.is_dir()', source, 'manager-local checks cannot validate runtime paths')
|
|
self.assertIn('if user_dir.exists()', source)
|
|
self.assertIn('.local/state/archipelago/sessions', source)
|
|
unit = (ROOT / UNIT).read_text()
|
|
paths = [p.lstrip('-') for line in unit.splitlines() if line.startswith('ReadWritePaths=') for p in line.split('=', 1)[1].split()]
|
|
self.assertIn('/home/archipelago/Work', paths, 'default HOME/Work has no writable exception under ProtectSystem=strict')
|
|
self.assertIn('/home/archipelago/.local/state/archipelago/sessions', paths)
|
|
self.assertNotIn('/home', paths)
|
|
self.assertNotIn('/home/archipelago', paths)
|
|
self.assertIn('ProtectSystem=strict', unit)
|
|
self.assertIn('User=archipelago', unit)
|
|
self.assertNotIn('Environment="HOME=', unit)
|
|
self.assertTrue((ROOT / PREP).is_file(), 'missing unprivileged install-time provisioning')
|
|
builder = (ROOT / BUILDER).read_text()
|
|
self.assertIn('cp "$SCRIPT_DIR/../configs/prepare-terminal-workspace.sh" "$ARCH_DIR/configs/prepare-terminal-workspace.sh"', builder)
|
|
self.assertIn('chroot /mnt/target runuser -u archipelago -- /bin/bash -s -- /home/archipelago < "$BOOT_MEDIA/archipelago/configs/prepare-terminal-workspace.sh"', builder)
|
|
self.assertLess(builder.index('chroot /mnt/target runuser -u archipelago -- /bin/bash -s -- /home/archipelago'), builder.index('cp "$BOOT_MEDIA/archipelago/configs/archipelago.service"'))
|
|
update = (ROOT / UPDATE).read_text()
|
|
self.assertIn('sudo -H -u archipelago -- /bin/bash "$REPO_DIR/image-recipe/configs/prepare-terminal-workspace.sh" /home/archipelago', update)
|
|
self.assertLess(update.index('sudo -H -u archipelago -- /bin/bash "$REPO_DIR/image-recipe/configs/prepare-terminal-workspace.sh"'), update.index('# Build backend'), 'unsafe-layout refusal must precede backend build/install')
|
|
# Execute only the production provisioning script, not either installer.
|
|
with tempfile.TemporaryDirectory(prefix='d09-', dir=os.environ['TMPDIR']) as d:
|
|
home = Path(d) / 'home'
|
|
home.mkdir()
|
|
proc = subprocess.run(['/bin/bash', str(ROOT / PREP), str(home)], capture_output=True, text=True, timeout=3)
|
|
self.assertEqual(proc.returncode, 0, proc.stderr)
|
|
work = home / 'Work'
|
|
self.assertTrue(work.is_dir())
|
|
self.assertEqual(work.stat().st_uid, os.getuid())
|
|
self.assertEqual(work.stat().st_mode & 0o777, 0o700)
|
|
self.assertEqual(sorted(p.name for p in home.iterdir()), ['Work'])
|
|
|
|
def prepare(self, home):
|
|
return subprocess.run(['/bin/bash', str(ROOT / PREP), str(home)], capture_output=True, text=True, timeout=3)
|
|
|
|
def fixture(self):
|
|
return tempfile.TemporaryDirectory(prefix='d09-', dir=os.environ['TMPDIR'])
|
|
|
|
def test_repeated_preparation_preserves_existing_work_and_legacy_metadata(self):
|
|
with self.fixture() as d:
|
|
home = Path(d)
|
|
work = home / 'Work'; work.mkdir(mode=0o750)
|
|
legacy = home / '.local/state/archipelago/sessions'; legacy.mkdir(parents=True, mode=0o700)
|
|
files = [work / 'project.txt', legacy / 'old.json']
|
|
for p in files: p.write_bytes(b'unchanged sentinel\n')
|
|
before = {p: (p.stat().st_ino, p.stat().st_uid, p.stat().st_mode, p.stat().st_mtime_ns, p.read_bytes() if p.is_file() else None) for p in [work, legacy, *files]}
|
|
for _ in range(2):
|
|
p = self.prepare(home); self.assertEqual(p.returncode, 0, p.stderr)
|
|
after = {p: (p.stat().st_ino, p.stat().st_uid, p.stat().st_mode, p.stat().st_mtime_ns, p.read_bytes() if p.is_file() else None) for p in before}
|
|
self.assertEqual(before, after)
|
|
|
|
def test_inaccessible_legacy_components_refused_without_mutation(self):
|
|
self.assertNotEqual(os.geteuid(), 0, 'permission fixtures must be unprivileged')
|
|
for relative in ('.local', '.local/state', '.local/state/archipelago',
|
|
'.local/state/archipelago/sessions'):
|
|
for mode in (0o600, 0o000):
|
|
with self.subTest(component=relative, mode=oct(mode)), self.fixture() as d:
|
|
home = Path(d)
|
|
legacy = home / '.local/state/archipelago/sessions'
|
|
legacy.mkdir(parents=True, mode=0o700)
|
|
sentinel = legacy / 'old.json'
|
|
sentinel.write_bytes(b'unchanged legacy metadata\n')
|
|
paths = [home, home / '.local', home / '.local/state',
|
|
home / '.local/state/archipelago', legacy, sentinel]
|
|
# Open before denying traversal: fstat/pread can observe every
|
|
# retained inode without restoring or repairing permissions.
|
|
fds = [os.open(p, os.O_RDONLY) for p in paths]
|
|
denied = home / relative
|
|
original_mode = denied.stat().st_mode & 0o777
|
|
def snapshot():
|
|
fields = ('st_dev', 'st_ino', 'st_uid', 'st_gid', 'st_mode',
|
|
'st_size', 'st_nlink', 'st_mtime_ns', 'st_ctime_ns')
|
|
return [tuple(getattr(os.fstat(fd), k) for k in fields) for fd in fds]
|
|
try:
|
|
# Fixture-only chmod; the helper must not repair this.
|
|
denied.chmod(mode)
|
|
before = snapshot()
|
|
proc = self.prepare(home)
|
|
after = snapshot()
|
|
content = os.pread(fds[-1], 4096, 0)
|
|
work_created = (home / 'Work').exists()
|
|
self.assertEqual(before, after, proc.stdout + proc.stderr)
|
|
self.assertEqual(content, b'unchanged legacy metadata\n')
|
|
self.assertFalse(work_created, proc.stdout + proc.stderr)
|
|
self.assertNotEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
|
self.assertIn('legacy', proc.stderr)
|
|
finally:
|
|
denied.chmod(original_mode) # owned fixture cleanup only
|
|
for fd in fds:
|
|
os.close(fd)
|
|
|
|
def test_existing_file_is_not_replaced(self):
|
|
with self.fixture() as d:
|
|
home = Path(d); work = home / 'Work'; work.write_text('sentinel')
|
|
self.assertNotEqual(self.prepare(home).returncode, 0)
|
|
self.assertEqual(work.read_text(), 'sentinel')
|
|
|
|
def test_work_symlink_cannot_widen_writable_root(self):
|
|
with self.fixture() as d:
|
|
home = Path(d); work = home / 'Work'; work.symlink_to(home, target_is_directory=True)
|
|
self.assertNotEqual(self.prepare(home).returncode, 0)
|
|
self.assertTrue(work.is_symlink())
|
|
self.assertEqual(list(home.iterdir()), [work])
|
|
|
|
def test_legacy_symlink_refused_before_workspace_creation(self):
|
|
with self.fixture() as d:
|
|
home = Path(d); (home / '.local').symlink_to(home, target_is_directory=True)
|
|
self.assertNotEqual(self.prepare(home).returncode, 0)
|
|
self.assertFalse((home / 'Work').exists())
|
|
self.assertTrue((home / '.local').is_symlink())
|
|
|
|
def test_missing_home_not_created(self):
|
|
with self.fixture() as d:
|
|
home = Path(d) / 'absent'
|
|
self.assertNotEqual(self.prepare(home).returncode, 0)
|
|
self.assertFalse(home.exists())
|
|
|
|
def test_home_symlink_refused(self):
|
|
with self.fixture() as d:
|
|
real = Path(d) / 'real'; real.mkdir()
|
|
alias = Path(d) / 'alias'; alias.symlink_to(real, target_is_directory=True)
|
|
self.assertNotEqual(self.prepare(alias).returncode, 0)
|
|
self.assertFalse((real / 'Work').exists())
|
|
|
|
def test_inaccessible_work_not_chmodded(self):
|
|
with self.fixture() as d:
|
|
home = Path(d); work = home / 'Work'; work.mkdir(mode=0o500)
|
|
self.assertNotEqual(self.prepare(home).returncode, 0)
|
|
self.assertEqual(work.stat().st_mode & 0o777, 0o500)
|
|
|
|
def test_legacy_file_refused_before_workspace_creation(self):
|
|
with self.fixture() as d:
|
|
home = Path(d); (home / '.local').write_text('sentinel')
|
|
self.assertNotEqual(self.prepare(home).returncode, 0)
|
|
self.assertFalse((home / 'Work').exists())
|
|
self.assertEqual((home / '.local').read_text(), 'sentinel')
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|