import copy import hashlib import importlib.util import json import os import io from pathlib import Path import subprocess import tempfile import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[3] spec = importlib.util.spec_from_file_location('updater', ROOT / 'scripts/selfhost-update.py') u = importlib.util.module_from_spec(spec) spec.loader.exec_module(u) def data(value): return json.dumps(value).encode() class Http: def __init__(self): self.manifest = {'schemaVersion': 1, 'version': '3.1.0', 'tag': 'v3.1.0', 'commit': 'a' * 40, 'repository': u.REPOSITORY, 'migrations': [], 'images': [{'name': name, 'repository': u.REGISTRY + '/' + name, 'digest': 'sha256:' + char * 64} for name, char in [('backend', 'a'), ('frontend', 'b')]]} self.files = {name: (ROOT / {'docker-compose.selfhost.yml': 'docker-compose.selfhost.yml', '.env.example': '.env.example', 'selfhost-setup.sh': 'scripts/selfhost-setup.sh'}[name]).read_bytes() for name in ['docker-compose.selfhost.yml', '.env.example', 'selfhost-setup.sh']} self.files.update({'release-manifest.json': data(self.manifest), 'CHANGELOG.md': b'New release'}) self.refresh() self.requests = [] self.bad_health = False self.release = {'id': 1, 'tag_name': 'v3.1.0', 'draft': False, 'prerelease': False} self.fail_download = False def refresh(self): self.files['SHA256SUMS'] = ''.join(hashlib.sha256(blob).hexdigest() + ' ' + name + '\n' for name, blob in self.files.items() if name != 'SHA256SUMS').encode() def get(self, url, **kwargs): self.requests.append(url) if url.endswith('/'): return b'FEDEO' if self.fail_download: raise u.UpdateError('Download fehlgeschlagen') return self.files[url.rsplit('/', 1)[-1]] def json(self, url): self.requests.append(url) if url.endswith('/releases/tags/v3.1.0'): return self.release if '/releases?limit=' in url: return [self.release, {'tag_name': 'v9.0.0', 'draft': True}, {'tag_name': 'v4.0.0-rc.1', 'prerelease': True}] if url.endswith('/releases/1/assets'): return [{'name': name, 'browser_download_url': u.REPOSITORY + '/assets/' + name} for name in self.files] if url.endswith('/tags/v3.1.0'): return {'commit': {'sha': 'a' * 40}} if url.endswith('/backend/version'): return {'product': 'FEDEO', 'version': '3.1.0'} if url.endswith('/health'): return {'status': 'ok', 'product': 'FEDEO', 'version': '3.0.0' if self.bad_health else '3.1.0', 'revision': 'a' * 40} raise AssertionError(url) class Runner: """Real Compose parsing with a simulated daemon; no containers are started.""" def __init__(self, log, http): self.log = log self.http = http self.calls = [] self.fail = None self.on_pull = None self.interrupt = None def run(self, args, output=None, input_file=None, **kwargs): self.calls.append(args) if self.interrupt and self.interrupt(args): raise KeyboardInterrupt() if self.fail and self.fail(args): raise u.UpdateError('simulated command failure') if args[:3] == ['docker', 'compose', 'version']: return '2.39.1' if args[:2] == ['docker', 'info']: return json.dumps({'OSType': 'linux', 'Architecture': 'amd64', 'DockerRootDir': '/tmp'}) if args[:3] == ['docker', 'context', 'inspect']: return 'unix:///var/run/docker.sock' if args[:3] == ['docker', 'image', 'inspect']: return 'image-' + args[3].rsplit('/', 1)[-1].split('@')[0] if args[:2] == ['docker', 'inspect']: name = args[2] return json.dumps([{'Image': 'image-' + name, 'Config': {'Labels': {'com.docker.compose.service': name}}, 'State': {'Running': name != 'createbuckets', 'Health': {'Status': 'healthy'}}}]) if 'config' in args: result = subprocess.run(args, env={'PATH': os.environ['PATH'], 'HOME': os.environ['HOME']}, capture_output=True, text=True) if result.returncode: raise u.UpdateError('Compose invalid: ' + result.stderr) return result.stdout if 'ps' in args: if args[:2] == ['docker', 'ps']: return '' if args[-1] in ('backend', 'frontend', 'db', 'minio'): return args[-1] return 'backend\nfrontend\ndb\nminio\ntraefik\nnode-exporter\ncreatebuckets\n' if 'pull' in args and self.on_pull: self.on_pull() if any('pg_database_size' in part for part in args): return '1000' if args[0] == 'du': return '1000\tminio' if output: Path(output).write_bytes(b'PGDMP fake dump') if args[0] == 'tar' and '-cpf' in args: Path(args[2]).write_bytes(b'backup') return '' class UpdateTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) (self.root / 'minio').mkdir() (self.root / 'postgres').mkdir() (self.root / '.env').write_bytes(u.set_version((ROOT / '.env.example').read_bytes(), '3.0.0')) (self.root / 'docker-compose.yml').write_bytes((ROOT / 'docker-compose.selfhost.yml').read_bytes()) self.original_env = (self.root / '.env').read_bytes() self.original_compose = (self.root / 'docker-compose.yml').read_bytes() self.http = Http() self.runner = Runner(self.root / 'log', self.http) self.updater = u.Updater(self.root, http=self.http, runner_factory=lambda log: self.runner) self.updater.check_space = lambda *_: None self.env_patch = patch.dict(os.environ, {'DOCKER_HOST': 'unix:///var/run/docker.sock'}) self.env_patch.start() self.sleep_patch = patch.object(u.time, 'sleep') self.sleep_patch.start() def tearDown(self): self.env_patch.stop() self.sleep_patch.stop() self.temp.cleanup() def run_update(self, **kwargs): return self.updater.execute('apply', '3.1.0', accept_compose=True, **kwargs) def called(self, value): return any(value in args for args in self.runner.calls) def assert_untouched(self): self.assertEqual((self.root / '.env').read_bytes(), self.original_env) self.assertEqual((self.root / 'docker-compose.yml').read_bytes(), self.original_compose) self.assertFalse(self.called('stop')) def test_success_pins_images_preserves_configuration_and_checks_both_public_services(self): result = self.run_update() self.assertEqual(result['status'], 'succeeded') self.assertFalse(result['recoveryRequired']) self.assertEqual((self.root / '.env').read_bytes(), u.set_version(self.original_env, '3.1.0')) self.assertIn(b'backend@sha256:' + b'a' * 64, (self.root / 'docker-compose.yml').read_bytes()) backup = Path(result['backup']) self.assertTrue((backup / 'complete.json').exists()) self.assertEqual((backup / 'config/.env').read_bytes(), self.original_env) self.assertTrue(any(url.endswith('/backend/health') for url in self.http.requests)) self.assertTrue(any(url.endswith('/api/health') for url in self.http.requests)) self.assertEqual((self.root / '.env').stat().st_mode & 0o777, 0o600) self.assertEqual((self.root / '.updates').stat().st_mode & 0o777, 0o700) self.assertFalse(self.called('down')) self.assertFalse(self.called('--remove-orphans')) calls = self.runner.calls self.assertLess(next(i for i, a in enumerate(calls) if 'pull' in a), next(i for i, a in enumerate(calls) if 'stop' in a)) self.assertLess(next(i for i, a in enumerate(calls) if 'pg_restore' in a), next(i for i, a in enumerate(calls) if 'migrate' in a)) def test_check_only_creates_a_plan_and_ignores_unpublished_releases(self): result = self.updater.execute('check') self.assertEqual(result['version'], '3.1.0') self.assertEqual(result['status'], 'checked') self.assertTrue(Path(result['plan']['composeDiff']).exists()) self.assert_untouched() self.assertFalse(self.called('pull')) def test_checksum_failure_does_not_stop_anything(self): self.http.files['docker-compose.selfhost.yml'] += b'\n# tampered' with self.assertRaisesRegex(u.UpdateError, 'Prüfsumme'): self.run_update() self.assert_untouched() def test_download_failure_does_not_stop_anything(self): self.http.fail_download = True with self.assertRaisesRegex(u.UpdateError, 'Download'): self.run_update() self.assert_untouched() def test_draft_is_rejected_even_when_explicitly_selected(self): self.http.release['draft'] = True with self.assertRaisesRegex(u.UpdateError, 'veröffentlichte'): self.run_update() self.assert_untouched() def test_manifest_commit_must_match_git_tag(self): self.http.manifest['commit'] = 'b' * 40 self.http.files['release-manifest.json'] = data(self.http.manifest) self.http.refresh() with self.assertRaisesRegex(u.UpdateError, 'Git-Tag'): self.run_update() self.assert_untouched() def test_invalid_compose_fails_before_stopping(self): self.http.files['docker-compose.selfhost.yml'] += b'\nbroken: [\n' self.http.refresh() with self.assertRaises(u.UpdateError): self.run_update() self.assert_untouched() def test_unknown_base_requires_review(self): with self.assertRaisesRegex(u.UpdateError, 'compose.diff'): self.updater.execute('apply', '3.1.0') self.assert_untouched() def test_override_is_preserved_but_cannot_replace_release_images(self): override = self.root / 'docker-compose.override.yml' contents = 'services:\n frontend:\n environment:\n MY_CUSTOM_SETTING: preserved\n' override.write_text(contents) self.run_update() self.assertEqual(override.read_text(), contents) self.assertEqual(json.loads((self.root / '.updates/installed.json').read_text())['manifest']['version'], '3.1.0') def test_override_cannot_disable_healthcheck(self): (self.root / 'docker-compose.override.yml').write_text('services:\n backend:\n healthcheck:\n disable: true\n') with self.assertRaisesRegex(u.UpdateError, 'Healthcheck'): self.run_update() self.assert_untouched() def test_changed_mount_is_rejected(self): self.http.files['docker-compose.selfhost.yml'] = self.http.files['docker-compose.selfhost.yml'].replace(b'./postgres:', b'./different:') self.http.refresh() with self.assertRaisesRegex(u.UpdateError, 'Datenpfade'): self.run_update() self.assert_untouched() def test_external_storage_fails_before_maintenance(self): (self.root / '.env').write_bytes(self.original_env.replace(b'http://minio:9000', b'https://storage.example.test')) with self.assertRaisesRegex(u.UpdateError, 'Backup-Adapter'): self.run_update() self.assertFalse(self.called('stop')) def test_database_url_must_match_the_database_being_backed_up(self): (self.root / '.env').write_bytes(self.original_env.replace(b'@db:5432/fedeo', b'@db:5432/other_database')) with self.assertRaisesRegex(u.UpdateError, 'gesicherte lokale'): self.run_update() self.assertFalse(self.called('stop')) def test_missing_required_secret_is_rejected(self): (self.root / '.env').write_bytes(self.original_env.replace(b'ENCRYPTION_KEY=', b'IGNORED_KEY=')) with self.assertRaisesRegex(u.UpdateError, 'ENCRYPTION_KEY'): self.run_update() self.assertFalse(self.called('stop')) def test_pull_failure_never_stops_old_stack(self): self.runner.fail = lambda args: 'pull' in args with self.assertRaises(u.UpdateError): self.run_update() self.assert_untouched() def test_configuration_changed_during_pull_is_detected(self): self.runner.on_pull = lambda: (self.root / '.env').write_bytes(self.original_env + b'\nNEW=value\n') with self.assertRaisesRegex(u.UpdateError, 'während'): self.run_update() self.assertFalse(self.called('stop')) def test_backup_failure_restarts_previous_services_and_preserves_configuration(self): self.runner.fail = lambda args: any('pg_dump ' in arg for arg in args) with self.assertRaises(u.UpdateError): self.run_update() self.assertTrue(self.called('start')) self.assertFalse(self.called('migrate')) self.assertFalse(self.updater.status()['recoveryRequired']) self.assertEqual((self.root / '.env').read_bytes(), self.original_env) self.assertEqual((self.root / 'docker-compose.yml').read_bytes(), self.original_compose) def test_migration_failure_keeps_recovery_state_and_never_starts_old_application(self): self.runner.fail = lambda args: 'migrate' in args with self.assertRaises(u.UpdateError): self.run_update() state = self.updater.status() self.assertEqual(state['phase'], 'migrating') self.assertTrue(state['recoveryRequired']) self.assertTrue(Path(state['backup']).exists()) self.assertFalse(self.called('start')) with self.assertRaisesRegex(u.UpdateError, 'Wiederherstellung'): self.run_update() def test_interrupt_during_migration_is_persisted(self): self.runner.interrupt = lambda args: 'migrate' in args with self.assertRaises(KeyboardInterrupt): self.run_update() self.assertEqual(self.updater.status()['status'], 'interrupted') self.assertTrue(self.updater.status()['recoveryRequired']) self.assertFalse(self.called('start')) def test_process_death_is_detected_and_does_not_erase_recovery_state(self): job = self.root / '.updates/interrupted' job.mkdir() u.write_json(self.root / '.updates/status.json', {'status': 'running', 'job': str(job), 'recoveryRequired': True}) self.assertEqual(self.updater.status()['status'], 'interrupted') self.assertTrue(self.updater.status()['recoveryRequired']) with self.assertRaisesRegex(u.UpdateError, 'Wiederherstellung'): self.run_update() def test_lock_prevents_parallel_updates(self): with self.updater.lock(): other = u.Updater(self.root, http=self.http) with self.assertRaisesRegex(u.UpdateError, 'läuft bereits'): other.execute('check') def test_old_migrations_must_not_change(self): u.write_json(self.root / '.updates/installed.json', {'manifest': {'migrations': [{'path': 'old.sql', 'sha256': 'a' * 64}]}}) with self.assertRaisesRegex(u.UpdateError, 'Migrationen'): self.run_update() self.assert_untouched() def test_insufficient_space_does_not_stop_services(self): self.updater.check_space = lambda *_: (_ for _ in ()).throw(u.UpdateError('Zu wenig freier Speicher')) with self.assertRaisesRegex(u.UpdateError, 'Speicher'): self.run_update() self.assert_untouched() def test_failed_health_verification_does_not_mark_release_installed(self): self.http.bad_health = True with self.assertRaisesRegex(u.UpdateError, 'Funktionsprüfung'): self.run_update() self.assertEqual(self.updater.status()['phase'], 'verifying') self.assertTrue(self.updater.status()['recoveryRequired']) self.assertFalse((self.root / '.updates/installed.json').exists()) self.assertEqual(self.runner.calls[-1][-3:], ['backend', 'frontend', 'traefik']) def test_resolve_requires_a_working_manually_recovered_installation(self): self.runner.fail = lambda args: 'migrate' in args with self.assertRaises(u.UpdateError): self.run_update() self.runner.fail = None with self.assertRaisesRegex(u.UpdateError, 'Version stimmt nicht'): self.updater.resolve('3.0.0') self.assertTrue(self.updater.status()['recoveryRequired']) self.assertEqual(self.updater.resolve('3.1.0')['status'], 'recovered') self.assertFalse(self.updater.status()['recoveryRequired']) self.assertEqual(u.read_json(self.root / '.updates/installed.json')['manifest']['version'], '3.1.0') def test_second_managed_update_does_not_need_adoption(self): u.write_json(self.root / '.updates/installed.json', {'composeSha256': u.sha(self.original_compose)}) self.assertEqual(self.updater.execute('apply', '3.1.0')['status'], 'succeeded') def test_downgrade_is_rejected(self): (self.root / '.env').write_bytes(u.set_version(self.original_env, '3.2.0')) with self.assertRaisesRegex(u.UpdateError, 'Downgrades'): self.run_update() self.assertFalse(self.called('stop')) def test_backup_integrity_manifest_is_complete(self): result = self.run_update() backup = Path(result['backup']) manifest = u.read_json(backup / 'complete.json') for name in ('database.dump', 'globals.sql', 'files.tar', 'config/.env', 'config/docker-compose.yml'): self.assertEqual(u.sha((backup / name).read_bytes()), manifest['sha256'][name]) class HttpsTests(unittest.TestCase): def test_token_is_removed_after_cross_origin_redirect(self): requests = [] class Opener: def open(self, request, timeout): requests.append(request) if len(requests) == 1: raise u.urllib.error.HTTPError(request.full_url, 302, 'redirect', {'Location': 'https://objects.example.test/file'}, None) return io.BytesIO(b'asset') with patch.object(u.urllib.request, 'build_opener', return_value=Opener()): self.assertEqual(u.Https('private-token').get(u.REPOSITORY + '/asset'), b'asset') self.assertEqual(requests[0].get_header('Authorization'), 'token private-token') self.assertIsNone(requests[1].get_header('Authorization')) def test_insecure_redirect_is_rejected(self): class Opener: def open(self, request, timeout): raise u.urllib.error.HTTPError(request.full_url, 302, 'redirect', {'Location': 'http://objects.example.test/file'}, None) with patch.object(u.urllib.request, 'build_opener', return_value=Opener()): with self.assertRaisesRegex(u.UpdateError, 'HTTPS'): u.Https().get(u.REPOSITORY + '/asset') def test_download_size_is_limited(self): class Opener: def open(self, request, timeout): return io.BytesIO(b'123456') with patch.object(u.urllib.request, 'build_opener', return_value=Opener()): with self.assertRaisesRegex(u.UpdateError, 'Größe'): u.Https().get(u.REPOSITORY + '/asset', limit=5) if __name__ == '__main__': unittest.main()