import concurrent.futures import http.client import importlib.util import json from pathlib import Path import socket import subprocess import tempfile import threading import unittest ROOT = Path(__file__).resolve().parents[3] spec = importlib.util.spec_from_file_location('update_service', ROOT / 'scripts/selfhost-update-service.py') s = importlib.util.module_from_spec(spec) spec.loader.exec_module(s) class Factory: def __init__(self): self.calls = [] self.started = threading.Event() self.release = threading.Event() self.release.set() self.failure = None self.adoption = False self.available = True self.fingerprint = 'a' * 64 def __call__(self, directory): factory = self class FakeUpdater(s.engine.Updater): def execute(self, command, version, *, expected_fingerprint, request_id): with self.lock(): factory.calls.append((command, version, expected_fingerprint, request_id)) self.job = self.home / ('job-' + request_id) self.job.mkdir() self.state = {'requestId': request_id, 'status': 'running', 'phase': 'validating', 'job': str(self.job), 'recoveryRequired': False} self.save() factory.started.set() if not factory.release.wait(10): raise AssertionError('test worker timed out') if factory.failure: self.save(status='failed') raise factory.failure if expected_fingerprint and expected_fingerprint != factory.fingerprint: self.save(status='failed') raise s.engine.PlanChangedError('private changed configuration') plan = {'version': version or '3.2.0', 'revision': 'b' * 40, 'updateAvailable': factory.available, 'requiresComposeAdoption': factory.adoption, 'estimatedBackupBytes': 1000, 'approvalFingerprint': factory.fingerprint, 'composeDiff': '/private/path/secret.diff', 'retiredServices': ['matrix-db']} self.save(status='checked' if command == 'check' else 'succeeded', phase='complete') return {**self.state, 'plan': plan} return FakeUpdater(directory) class UnixConnection(http.client.HTTPConnection): def __init__(self, path): super().__init__('localhost', timeout=3) self.path = str(path) def connect(self): self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.sock.settimeout(self.timeout) self.sock.connect(self.path) class ServiceTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.factory = Factory() self.controller = s.Controller(self.root, updater_factory=self.factory) self.socket = self.root / 'control.sock' self.server = s.Server(self.socket, self.controller) self.thread = threading.Thread(target=self.server.serve_forever, kwargs={'poll_interval': 0.01}, daemon=True) self.thread.start() def tearDown(self): self.factory.release.set() self.server.shutdown() self.server.server_close() self.thread.join(3) self.controller.close() self.temp.cleanup() def request(self, method, path, body=None, headers=None): connection = UnixConnection(self.socket) connection.request(method, path, None if body is None else json.dumps(body), headers or ({'Content-Type': 'application/json'} if body is not None else {})) response = connection.getresponse() result = response.status, json.loads(response.read()) self.assertEqual(response.getheader('Cache-Control'), 'no-store') connection.close() return result def finish(self): self.controller.worker.join(3) self.assertFalse(self.controller.worker.is_alive()) def check(self, request_id='check-request-01'): status, _ = self.request('POST', '/v1/check', {'requestId': request_id, 'version': '3.2.0'}) self.assertEqual(status, 202) self.finish() return request_id def test_async_job_survives_client_disconnect_and_status_stays_responsive(self): self.factory.release.clear() connection = UnixConnection(self.socket) connection.request('POST', '/v1/check', json.dumps({'requestId': 'check-disconnected'}), {'Content-Type': 'application/json'}) connection.close() self.assertTrue(self.factory.started.wait(3)) status, result = self.request('GET', '/v1/status') self.assertEqual(status, 200) self.assertEqual(result['request']['status'], 'running') self.assertEqual(result['request']['progress']['phase'], 'validating') self.factory.release.set() self.finish() self.assertEqual(self.request('GET', '/v1/requests/check-disconnected')[1]['status'], 'checked') def test_duplicate_request_is_idempotent_even_after_completion_and_restart(self): request_id = self.check() status, result = self.request('POST', '/v1/check', {'requestId': request_id, 'version': '3.2.0'}) self.assertEqual(status, 200) self.assertEqual(result['status'], 'checked') self.controller.close() self.controller = s.Controller(self.root, updater_factory=self.factory) self.server.controller = self.controller self.assertEqual(self.request('GET', '/v1/status')[1]['request']['requestId'], request_id) status, _ = self.request('POST', '/v1/check', {'requestId': request_id, 'version': '3.2.0'}) self.assertEqual(status, 200) self.assertEqual(len(self.factory.calls), 1) status, result = self.request('POST', '/v1/check', {'requestId': request_id, 'version': '3.3.0'}) self.assertEqual(status, 409) self.assertEqual(result['error']['code'], 'REQUEST_CONFLICT') def test_concurrent_submissions_start_exactly_one_worker(self): self.factory.release.clear() with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: responses = list(pool.map(lambda i: self.request('POST', '/v1/check', {'requestId': f'parallel-{i}'}), range(2))) self.assertEqual(sorted(r[0] for r in responses), [202, 409]) self.assertTrue(self.factory.started.wait(3)) self.assertEqual(len(self.factory.calls), 1) def test_apply_requires_reviewed_version_and_passes_private_fingerprint(self): body = {'requestId': 'apply-request-01', 'version': '3.2.0', 'checkRequestId': 'check-request-01'} self.assertEqual(self.request('POST', '/v1/apply', body)[1]['error']['code'], 'CHECK_REQUIRED') self.check() self.assertEqual(self.request('POST', '/v1/apply', {**body, 'version': '3.3.0'})[0], 409) self.assertEqual(self.request('POST', '/v1/apply', body)[0], 202) self.finish() self.assertEqual(self.factory.calls[-1], ('apply', '3.2.0', 'a' * 64, 'apply-request-01')) self.assertEqual(self.request('GET', '/v1/requests/apply-request-01')[1]['status'], 'succeeded') def test_unmanaged_compose_and_changed_plan_cannot_be_applied_remotely(self): self.factory.adoption = True self.check() body = {'requestId': 'apply-request-01', 'version': '3.2.0', 'checkRequestId': 'check-request-01'} self.assertEqual(self.request('POST', '/v1/apply', body)[1]['error']['code'], 'COMPOSE_REVIEW_REQUIRED') self.factory.adoption = False self.check('check-request-02') self.factory.fingerprint = 'b' * 64 self.assertEqual(self.request('POST', '/v1/apply', {**body, 'checkRequestId': 'check-request-02'})[0], 202) self.finish() self.assertEqual(self.request('GET', '/v1/requests/apply-request-01')[1]['error']['code'], 'PLAN_CHANGED') def test_already_current_is_a_successful_check_without_apply(self): self.factory.available = False self.check() self.assertEqual(self.request('GET', '/v1/requests/check-request-01')[1]['status'], 'up_to_date') body = {'requestId': 'apply-request-01', 'version': '3.2.0', 'checkRequestId': 'check-request-01'} self.assertEqual(self.request('POST', '/v1/apply', body)[0], 409) def test_restart_marks_pending_requests_interrupted_without_replaying(self): for status in ('queued', 'running'): s.engine.write_json(self.controller.path('abandoned-' + status), {'requestId': 'abandoned-' + status, 'command': 'apply', 'status': status}) self.controller.close() self.controller = s.Controller(self.root, updater_factory=self.factory) self.server.controller = self.controller for status in ('queued', 'running'): record = self.request('GET', '/v1/requests/abandoned-' + status)[1] self.assertEqual(record['status'], 'interrupted') self.assertEqual(self.factory.calls, []) def test_shutdown_drains_worker_and_rejects_new_work(self): self.factory.release.clear() self.assertEqual(self.request('POST', '/v1/check', {'requestId': 'check-request-01'})[0], 202) self.assertTrue(self.factory.started.wait(3)) closing = threading.Thread(target=self.controller.close) closing.start() # Acquire the gate after close has entered it, without relying on a wall-clock sleep. for _ in range(1000): with self.controller.gate: if self.controller.closing: break threading.Event().wait(0.001) self.assertTrue(closing.is_alive()) self.assertEqual(self.request('POST', '/v1/check', {'requestId': 'check-request-02'})[1]['error']['code'], 'SHUTTING_DOWN') self.factory.release.set() closing.join(3) self.assertFalse(closing.is_alive()) self.assertEqual(self.controller.request_status('check-request-01')['status'], 'checked') def test_cli_lock_and_recovery_block_new_requests(self): updater = self.factory(self.root) job = updater.home / 'cli-job' job.mkdir() s.engine.write_json(updater.home / 'status.json', {'status': 'running', 'job': str(job)}) with updater.lock(): self.assertEqual(self.request('POST', '/v1/check', {'requestId': 'check-request-01'})[1]['error']['code'], 'BUSY') s.engine.write_json(updater.home / 'status.json', {'status': 'failed', 'recoveryRequired': True}) self.assertEqual(self.request('POST', '/v1/check', {'requestId': 'check-request-01'})[1]['error']['code'], 'RECOVERY_REQUIRED') def test_status_never_returns_raw_errors_paths_logs_or_configuration_fingerprints(self): self.factory.failure = ValueError('password-secret /opt/private/.env') self.check() result = self.request('GET', '/v1/requests/check-request-01')[1] self.assertEqual(result['error']['code'], 'UPDATE_FAILED') self.assertNotIn('password-secret', json.dumps(result)) self.factory.failure = None self.check('check-request-02') payload = json.dumps(self.request('GET', '/v1/requests/check-request-02')[1]) self.assertNotIn('/private', payload) self.assertNotIn('approvalFingerprint', payload) self.assertNotIn('a' * 64, payload) self.assertEqual(self.request('GET', '/v1/requests/../../secret')[0], 400) self.assertEqual(self.request('GET', '/v1/logs')[0], 404) self.assertEqual((self.controller.home / 'check-request-01.log').stat().st_mode & 0o777, 0o600) def test_api_rejects_unsupported_commands_options_bodies_and_browser_requests(self): for body in [[], {'requestId': 'bad/path'}, {'requestId': 'check-request-01', 'version': ';reboot'}, {'requestId': 'check-request-01', 'directory': '/etc'}, {'requestId': 'check-request-01', 'acceptComposeReplacement': True}]: with self.subTest(body=body): self.assertEqual(self.request('POST', '/v1/check', body)[0], 400) self.assertEqual(self.request('POST', '/v1/resolve', {'requestId': 'resolve-request'})[0], 404) self.assertEqual(self.request('POST', '/v1/check', {'requestId': 'check-request-01'}, {'Content-Type': 'application/json', 'Origin': 'https://example.test'})[0], 400) self.assertEqual(self.request('POST', '/v1/check', {'requestId': 'check-request-01', 'padding': 'x' * 5000})[0], 400) self.assertEqual(self.factory.calls, []) def test_second_controller_cannot_take_over_active_instance(self): self.assertEqual(self.socket.stat().st_mode & 0o777, 0o660) with self.assertRaisesRegex(s.engine.UpdateError, 'bereits ein Update-Dienst'): s.Controller(self.root, updater_factory=self.factory) class InstallerTests(unittest.TestCase): def test_unit_render_is_read_only_and_escapes_systemd_expansions(self): installer = ROOT / 'scripts/selfhost-update-service-install.sh' subprocess.run(['bash', '-n', str(installer)], check=True) result = subprocess.run(['bash', str(installer), '--print-unit', '--directory', '/opt/fedeo test%$"'], text=True, capture_output=True, check=True) self.assertIn('serve --directory "/opt/fedeo test%%$$\\\""', result.stdout) self.assertIn('KillMode=mixed', result.stdout) self.assertIn('TimeoutStopSec=infinity', result.stdout) self.assertNotIn('@FEDEO_DIRECTORY@', result.stdout) invalid = subprocess.run(['bash', str(installer), '--print-unit', '--directory', '/opt/fedeo\nExecStart=reboot'], capture_output=True) self.assertNotEqual(invalid.returncode, 0) if __name__ == '__main__': unittest.main()