import hashlib import importlib.util import io import os from pathlib import Path import stat import tarfile import tempfile import unittest from unittest.mock import patch spec = importlib.util.spec_from_file_location('public_update_prepare', Path(__file__).resolve().parents[1] / 'public_update_prepare.py') prepare = importlib.util.module_from_spec(spec) spec.loader.exec_module(prepare) class PublicUpdatePreparationTests(unittest.TestCase): def component(self, raw=b'qualified fixture bytes'): return {'name': 'archipelago', 'new_version': '2.0.4-alpha', 'size_bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()} def test_copy_matches_signed_size_and_hash_and_makes_read_only_snapshot(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) raw = b'qualified fixture bytes' (root / 'archipelago').write_bytes(raw) fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY) try: prepare.copy_component(fd, root / 'snapshot', self.component(raw)) finally: os.close(fd) self.assertEqual((root / 'snapshot').read_bytes(), raw) self.assertEqual(stat.S_IMODE((root / 'snapshot').stat().st_mode), 0o444) def test_source_symlink_is_refused_before_copy(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / 'outside').write_bytes(b'qualified fixture bytes') (root / 'archipelago').symlink_to(root / 'outside') fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY) try: with self.assertRaises(OSError): prepare.copy_component(fd, root / 'snapshot', self.component()) finally: os.close(fd) self.assertFalse((root / 'snapshot').exists()) def test_corrupt_copy_is_not_published_read_only(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / 'archipelago').write_bytes(b'X' * self.component()['size_bytes']) fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY) try: with self.assertRaises(ValueError): prepare.copy_component(fd, root / 'snapshot', self.component()) finally: os.close(fd) self.assertEqual(stat.S_IMODE((root / 'snapshot').stat().st_mode), 0o600) def archive(self, path, extras): with tarfile.open(path, 'w:gz') as archive: for name, kind, mode in [('index.html', tarfile.REGTYPE, 0o644), ('archipelago-runtime/RELEASE-PAYLOAD.json', tarfile.REGTYPE, 0o644), *extras]: entry = tarfile.TarInfo(name) entry.type = kind entry.mode = mode entry.size = 1 if kind == tarfile.REGTYPE else 0 if kind == tarfile.SYMTYPE: entry.linkname = '/etc/shadow' archive.addfile(entry, io.BytesIO(b'x') if entry.size else None) def test_archive_refuses_escape_links_duplicates_and_setuid(self): with tempfile.TemporaryDirectory() as directory: for index, extra in enumerate([ ('../outside', tarfile.REGTYPE, 0o644), ('link', tarfile.SYMTYPE, 0o777), ('index.html', tarfile.REGTYPE, 0o644), ('unsafe', tarfile.REGTYPE, 0o4755)]): path = Path(directory) / str(index) self.archive(path, [extra]) with self.assertRaises(ValueError): prepare.validate_archive(path) good = Path(directory) / 'good' self.archive(good, []) prepare.validate_archive(good) def test_archive_refuses_file_directory_collisions_in_either_order(self): with tempfile.TemporaryDirectory() as directory: for index, entries in enumerate([ [('parent', tarfile.REGTYPE, 0o644), ('parent/child', tarfile.REGTYPE, 0o644)], [('parent/child', tarfile.REGTYPE, 0o644), ('parent', tarfile.REGTYPE, 0o644)], [('archipelago-runtime', tarfile.REGTYPE, 0o644)]]): path = Path(directory) / str(index) self.archive(path, entries) with self.assertRaises(ValueError): prepare.validate_archive(path) def test_archive_required_entries_must_be_files(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'archive' with tarfile.open(path, 'w:gz') as archive: for name in ['index.html', 'archipelago-runtime/RELEASE-PAYLOAD.json']: entry = tarfile.TarInfo(name) entry.type = tarfile.DIRTYPE archive.addfile(entry) with self.assertRaises(ValueError): prepare.validate_archive(path) def test_partial_or_dangling_protected_owner_refuses_before_service_probe(self): for marker in ['/etc/archipelago-d09', '/etc/archipelago-c01', '/var/lib/archipelago-d09', '/run/archipelago-d09']: def lstat(path): if path == marker: return object() # Any inode, including a dangling symlink. raise FileNotFoundError(path) with self.subTest(marker=marker), patch.object(prepare.os, 'lstat', side_effect=lstat), \ patch.object(prepare.subprocess, 'check_output') as command: with self.assertRaises(ValueError): prepare.require_legacy_service() command.assert_not_called() def test_ownership_probe_errors_fail_closed(self): with patch.object(prepare.os, 'lstat', side_effect=PermissionError('fixture')), \ patch.object(prepare.subprocess, 'check_output') as command: with self.assertRaises(PermissionError): prepare.require_legacy_service() command.assert_not_called() def test_service_probe_accepts_only_standard_legacy_command(self): command = 'LoadState=loaded\nUser=archipelago\nExecStart={ path=/usr/local/bin/archipelago ; argv[]=/usr/local/bin/archipelago ; }\n' for output, accepted in [(command, True), (command.replace('argv[]=/usr/local/bin/archipelago ;', 'argv[]=/usr/local/bin/archipelago --managed ;'), False), (command.replace('User=archipelago', 'User=root'), False), (command.replace('LoadState=loaded', 'LoadState=masked'), False)]: with self.subTest(output=output), \ patch.object(prepare.os, 'lstat', side_effect=FileNotFoundError), \ patch.object(prepare.subprocess, 'check_output', return_value=output): if accepted: prepare.require_legacy_service() else: with self.assertRaises(ValueError): prepare.require_legacy_service() def test_manifest_write_is_exclusive_and_readonly(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'manifest' prepare.write_manifest(path, b'original signed bytes') self.assertEqual(path.read_bytes(), b'original signed bytes') self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o444) with self.assertRaises(FileExistsError): prepare.write_manifest(path, b'replacement') def test_duplicate_json_fields_are_rejected(self): import json with self.assertRaises(ValueError): json.loads('{"version": "2.0.4-alpha", "version": "2.0.3-alpha"}', object_pairs_hook=prepare.pairs) def test_document_shape_refuses_os_policy_unsigned_and_boolean_version(self): raw = self.component() frontend = dict(raw, name='archipelago-frontend-2.0.4-alpha.tar.gz') document = {'version': '2.0.4-alpha', 'application_update': dict(prepare.POLICY), 'signed_by': prepare.RELEASE_ROOT, 'signature': '0' * 128, 'components': [raw, frontend]} # Shape validation only. authenticate() separately performs real signature verification. prepare.validate_document(document, '2.0.3-alpha') for bad in [dict(document, os_update={}), dict(document, signature=None), dict(document, application_update={'version': True, 'mode': 'signed-application-only'})]: with self.assertRaises(ValueError): prepare.validate_document(bad, '2.0.3-alpha') with self.assertRaises(ValueError): prepare.validate_document(document, '2.0.4-alpha') if __name__ == '__main__': unittest.main()