#!/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)