"""Filesystem transaction regressions; host commands/custody are explicit doubles.""" import hashlib import json import os from pathlib import Path import tempfile import unittest SCRIPTS = Path(__file__).resolve().parents[1] class RuntimeTransaction(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.root = Path(self.directory.name) self.ns = {'__name__': 'runtime_fixture'} for filename in ('public_update_prepare.py', 'public_update_runtime.py', 'public_update_worker.py'): exec(compile((SCRIPTS / filename).read_text(), filename, 'exec'), self.ns) self.commands = [] self.ns['command'] = lambda *args, **kwargs: self.commands.append(args) or '' self.ns['trusted_open'] = lambda path, directory=False: os.open(path, os.O_RDONLY | os.O_NOFOLLOW | (os.O_DIRECTORY if directory else 0)) self.dest = self.root / 'installed' self.dest.mkdir() self.record = self.dest / 'managed.json' self.ns['RUNTIME_RECORD'] = self.record self.ns['RUNTIME_OBJECTS'] = { 'scripts': (str(self.dest / 'scripts'), 'scripts', 'directory', True), 'helper': (str(self.dest / 'helper'), 'helper', 'executable', True), 'optional': (str(self.dest / 'optional'), 'optional', 'file', False), 'routing': (str(self.dest / 'routing'), '', 'file', False), 'managed': (str(self.record), '', 'file', True), } (self.dest / 'scripts').mkdir() (self.dest / 'scripts/local.py').write_text('old operator script\n') (self.dest / 'helper').write_text('old helper\n') (self.dest / 'helper').chmod(0o755) (self.dest / 'routing').write_text('retained operator routing\n') self.tx = self.root / 'transaction' self.tx.mkdir() self.payload = self.tx / 'new-frontend/archipelago-runtime' (self.payload / 'scripts').mkdir(parents=True) (self.payload / 'scripts/new.sh').write_text('#!/bin/sh\nexit 0\n') (self.payload / 'helper').write_text('new helper\n') (self.payload / 'optional').write_text('optional must stay absent\n') (self.payload / 'image-recipe/configs').mkdir(parents=True) (self.payload / 'image-recipe/configs/nginx-archipelago.conf').write_text('generic template\n') (self.tx / 'new-binary').write_text('new executable\n') self.state = {'id': '1' * 32, 'new_version': '2.0.4-alpha', 'new': self.ns['pair_identity'](self.tx / 'new-binary', self.tx / 'new-frontend')} def prepare(self): return self.ns['runtime_prepare'](self.tx, self.state) def test_forward_and_restore_exact_code_routing_and_absence(self): state = self.prepare() self.assertEqual((self.payload / 'image-recipe/configs/nginx-archipelago.conf').read_bytes(), (self.dest / 'routing').read_bytes()) self.ns['runtime_replace'](self.tx, state, 'new') self.assertFalse((self.dest / 'scripts/local.py').exists()) self.assertTrue((self.dest / 'scripts/new.sh').stat().st_mode & 0o111) self.assertEqual((self.dest / 'helper').read_text(), 'new helper\n') self.assertFalse((self.dest / 'optional').exists()) self.assertEqual(json.loads(self.record.read_text())['version'], '2.0.4-alpha') self.ns['runtime_replace'](self.tx, state, 'old') self.assertEqual((self.dest / 'scripts/local.py').read_text(), 'old operator script\n') self.assertEqual((self.dest / 'helper').read_text(), 'old helper\n') self.assertFalse(self.record.exists()) self.assertEqual((self.dest / 'routing').read_text(), 'retained operator routing\n') self.assertTrue(list(self.dest.glob('.public-runtime-retired-*'))) def test_private_app_source_omitted_from_public_payload_is_retained_locally(self): self.ns['RUNTIME_OBJECTS']['apps'] = (str(self.dest / 'apps'), 'apps', 'directory', True) private = self.dest / 'apps/private-app' private.mkdir(parents=True) (private / 'manifest.yaml').write_text('private installed source') public = self.payload / 'apps/public-app' public.mkdir(parents=True) (public / 'manifest.yaml').write_text('new public app') # This unit runs without root. Keep the copied directory writable while # the production root owner composes its protected staged postimage. copy = self.ns['runtime_copy'] def compose(source, target): result = copy(source, target) if target.is_dir(): target.chmod(0o755) return result self.ns['runtime_copy'] = compose state = self.prepare() self.ns['runtime_replace'](self.tx, state, 'new') self.assertEqual((private / 'manifest.yaml').read_text(), 'private installed source') self.assertEqual((self.dest / 'apps/public-app/manifest.yaml').read_text(), 'new public app') self.ns['runtime_replace'](self.tx, state, 'old') self.assertTrue(private.exists()) self.assertFalse((self.dest / 'apps/public-app').exists()) def test_guarded_routes_and_old_template_survive_forward_and_recovery(self): import importlib.util spec = importlib.util.spec_from_file_location('dashboard_guard', SCRIPTS / 'dashboard-public-guard.py') guard = importlib.util.module_from_spec(spec) spec.loader.exec_module(guard) source = ('server { listen 80 default_server; server_name _; location / { return 200 "dashboard"; } }\n' 'server { listen 443 ssl default_server; server_name _; location / { return 200 "dashboard"; } }\n' 'server { listen 80; server_name operator.example; location / { return 200 "private"; } }\n') protected = guard.guarded(source).encode() (self.dest / 'routing').write_bytes(protected) relative = 'archipelago-runtime/image-recipe/configs/nginx-archipelago.conf' old_template = self.tx / 'old-frontend' / relative old_template.parent.mkdir(parents=True) old_template.write_bytes(protected) (self.tx / 'new-frontend' / relative).write_bytes(source.encode()) state = self.prepare() self.assertEqual((self.tx / 'new-frontend' / relative).read_bytes(), protected) for generation in ('new', 'old'): self.ns['runtime_replace'](self.tx, state, generation) self.assertEqual((self.dest / 'routing').read_bytes(), protected) # Even an older bootstrap which copies its restored template keeps # the same management guard and operator routes. self.assertEqual(old_template.read_bytes(), protected) self.assertEqual(guard.guarded(old_template.read_text()), protected.decode()) def test_foreign_configuration_change_refuses_before_other_replacements(self): state = self.prepare() (self.dest / 'routing').write_text('new operator edit\n') with self.assertRaisesRegex(ValueError, 'foreign runtime/configuration'): self.ns['runtime_replace'](self.tx, state, 'new') self.assertEqual((self.dest / 'helper').read_text(), 'old helper\n') self.assertEqual((self.dest / 'routing').read_text(), 'new operator edit\n') def test_corrupt_preimage_refuses_recovery_without_mutating_installed(self): state = self.prepare() self.ns['runtime_replace'](self.tx, state, 'new') (self.tx / 'old-runtime/helper').chmod(0o755) # explicit privileged-corruption fixture (self.tx / 'old-runtime/helper').write_text('corrupt saved preimage\n') with self.assertRaisesRegex(ValueError, 'protected runtime snapshot changed'): self.ns['runtime_replace'](self.tx, state, 'old') self.assertEqual((self.dest / 'helper').read_text(), 'new helper\n') def test_directory_rename_interruption_can_resume_or_restore(self): state = self.prepare() (self.dest / 'scripts').rename(self.dest / '.retained-before-interruption') self.ns['runtime_replace'](self.tx, state, 'new') self.ns['runtime_replace'](self.tx, state, 'old') self.assertEqual((self.dest / 'scripts/local.py').read_text(), 'old operator script\n') self.assertTrue((self.dest / '.retained-before-interruption/local.py').exists()) def test_symlink_refused_without_following_or_mutating_target(self): outside = self.root / 'operator-file' outside.write_text('private fixture\n') (self.dest / 'scripts/foreign-link').symlink_to(outside) with self.assertRaises((ValueError, OSError)): self.prepare() self.assertEqual(outside.read_text(), 'private fixture\n') def test_inactive_and_disabled_maintenance_are_not_started(self): self.ns['command'] = lambda *args, **kwargs: ( 'disabled\n' if '--value' in args else self.commands.append(args) or '') state = {'maintenance_units': { 'archipelago-doctor.timer': {'ActiveState': 'inactive', 'UnitFileState': 'disabled', 'LoadState': 'loaded'}, 'archipelago-npm-bridge.timer': {'ActiveState': 'active', 'UnitFileState': 'disabled', 'LoadState': 'loaded'}, }} self.ns['runtime_resume_units'](state) self.assertEqual(self.commands, [('/usr/bin/systemctl', 'start', 'archipelago-npm-bridge.timer')]) def test_runtime_startup_guards_mask_only_manager_and_hold_maintenance(self): state = {'dropin': '[Unit]\nRequires=recovery.service\n', 'runtime_startup_guard': True} manager = self.ns['owner_dropin'](state, self.ns['DROPIN']).decode() self.assertIn('InaccessiblePaths=-/opt/archipelago/web-ui/archipelago-runtime\n', manager) self.assertNotIn('ConditionPathExists', manager) for path in self.ns['OWNER_DROPINS'][1:]: value = self.ns['owner_dropin'](state, path).decode() self.assertIn('ConditionPathExists=!/var/lib/archipelago-public-transactions/maintenance-paused\n', value) self.assertNotIn('InaccessiblePaths', value) def test_maintenance_pause_cannot_release_another_transaction(self): self.ns['TRANSACTIONS'] = self.root state = {'id': '1' * 32, 'runtime_startup_guard': True} self.ns['runtime_hold_maintenance'](state) marker = self.root / 'maintenance-paused' self.assertEqual(marker.read_text(), state['id']) other = dict(state, id='2' * 32) with self.assertRaisesRegex(ValueError, 'different public transaction'): self.ns['runtime_hold_maintenance'](other) with self.assertRaisesRegex(ValueError, 'different public transaction'): self.ns['runtime_resume_units'](other) self.assertEqual(marker.read_text(), state['id']) self.ns['runtime_resume_units'](state) self.assertFalse(marker.exists()) if __name__ == '__main__': unittest.main()