192 lines
11 KiB
Python
192 lines
11 KiB
Python
"""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()
|