"""First-generation admission: real files, explicit crypto/network/PID1 doubles.""" import contextlib import hashlib import io import json import os from pathlib import Path import shutil import tarfile import tempfile import unittest SCRIPTS = Path(__file__).resolve().parents[1] class LegacyAdoption(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.ns = {'__name__': 'adoption_fixture'} for name in ('prepare', 'runtime', 'adopt', 'worker'): exec(compile((SCRIPTS / f'public_update_{name}.py').read_text(), name, 'exec'), self.ns) for name in ('TRANSACTIONS', 'BINARY', 'FRONTEND', 'LEGACY_MARKER', 'LEGACY_BINARY', 'LEGACY_FRONTEND', 'RUNTIME_RECORD', 'DROPIN', 'STAGING'): self.ns[name] = self.root / name.lower() self.ns['OWNER_DROPINS'] = (self.ns['DROPIN'],) self.ns['TRANSACTIONS'].mkdir() self.ns['owner_lock'] = contextlib.nullcontext self.ns['require_legacy_service'] = lambda: None self.ns['manager_adoptable'] = lambda: True self.ns['current'] = lambda: (None, None) self.ns['custody_frontend_parent'] = lambda: None 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.ns['RUNTIME_OBJECTS'] = {'routing': (str(self.root / 'routing'), '', 'file', False)} # Root production can modify its readonly staging trees. This ordinary # user fixture changes only directory modes, never identity/file modes. extract = self.ns['extract'] def readable_extract(*args, **kwargs): result = extract(*args, **kwargs) for directory, _, _ in os.walk(args[1]): Path(directory).chmod(0o755) return result self.ns['extract'] = readable_extract durable_mode = self.ns['durable_mode'] self.ns['durable_mode'] = lambda path, mode: durable_mode(path, 0o755 if path.is_dir() else mode) self.documents = {} self.bytes = {} self.downloads = [] def download(url, path, size=None, sha256=None): raw = self.bytes[url] if size is not None: self.assertEqual(len(raw), size) if sha256 is not None: self.assertEqual(hashlib.sha256(raw).hexdigest(), sha256) path.write_bytes(raw); self.downloads.append(url) self.ns['public_download'] = download def authenticate(raw): if raw not in self.documents.values(): raise ValueError('fixture signature refusal') return '2.0.4-alpha' self.ns['authenticate'] = authenticate self.ns['schedule_transaction'] = lambda tx, state, bundle: dict(state, transaction_path=str(tx)) self.marker = {'previous_version': '1.9.1-alpha', 'new_version': '2.0.4-alpha'} self.ns['LEGACY_MARKER'].write_text(json.dumps(self.marker)) self.install_release('1.9.1-alpha', b'old binary', {'index.html': b'old UI', 'aiui/extra.html': b'accepted UI extra'}) self.install_release('2.0.4-alpha', b'new binary', {'index.html': b'new UI', 'archipelago-runtime/RELEASE-PAYLOAD.json': b'{}'}) self.ns['LEGACY_BINARY'].write_bytes(b'old binary') self.ns['BINARY'].write_bytes(b'new binary') for version, destination in [('1.9.1-alpha', self.ns['LEGACY_FRONTEND']), ('2.0.4-alpha', self.ns['FRONTEND'])]: archive = self.root / (version + '.tar.gz'); archive.write_bytes(self.bytes[self.frontend_url(version)]) self.ns['extract'](archive, destination, version != '1.9.1-alpha') self.ns['historical_frontend_modes'](destination) # Reproduce the historical installer's documented static-extra merge. self.ns['FRONTEND'].chmod(0o755) shutil.copytree(self.ns['LEGACY_FRONTEND'] / 'aiui', self.ns['FRONTEND'] / 'aiui') def frontend_url(self, version): return self.ns['PUBLIC_RELEASE_BASE'] + f'/v{version}/archipelago-frontend-{version}.tar.gz' def install_release(self, version, binary, files): data = io.BytesIO() with tarfile.open(fileobj=data, mode='w:gz') as archive: for name, raw in files.items(): info = tarfile.TarInfo(name); info.size = len(raw); info.mode = 0o644 archive.addfile(info, io.BytesIO(raw)) frontend = data.getvalue() components = [] for name, raw in [('archipelago', binary), (f'archipelago-frontend-{version}.tar.gz', frontend)]: url = self.ns['PUBLIC_RELEASE_BASE'] + f'/v{version}/{name}' self.bytes[url] = raw components.append({'name': name, 'new_version': version, 'download_url': url, 'sha256': hashlib.sha256(raw).hexdigest(), 'size_bytes': len(raw)}) doc = {'version': version, 'signed_by': self.ns['RELEASE_ROOT'], 'signature': 'a' * 128, 'components': components} if version == '2.0.4-alpha': doc['application_update'] = self.ns['RUNTIME_POLICY'] raw = json.dumps(doc).encode(); self.documents[version] = raw name = 'manifest.json' if version == '2.0.4-alpha' else f'legacy-manifest-{version}.json' self.bytes[self.ns['PUBLIC_RELEASE_BASE'] + '/v2.0.4-alpha/' + name] = raw def adopt(self): return self.ns['adopt_legacy_pair']('installed fixture bundle') def test_accepts_historical_staging_cleanup_and_freezes_authenticated_pair(self): self.assertFalse(self.ns['STAGING'].exists()) state = self.adopt() self.assertTrue(state['adoption'] and state['runtime_requested']) tx = Path(state['transaction_path']) self.assertEqual((tx / 'old-binary').read_bytes(), b'old binary') self.assertEqual((tx / 'new-frontend/aiui/extra.html').read_bytes(), b'accepted UI extra') self.assertEqual(self.ns['pair_identity'](self.ns['BINARY'], self.ns['FRONTEND']), state['initial']) self.assertEqual(len(self.downloads), 5) def test_rejects_unauthenticated_old_binary_before_scheduling(self): self.ns['LEGACY_BINARY'].write_bytes(b'evil binary') with self.assertRaisesRegex(ValueError, 'untrusted historical backend'): self.adopt() self.assertEqual(self.ns['BINARY'].read_bytes(), b'new binary') def test_rejects_unsigned_runtime_changes_in_old_frontend(self): path = self.ns['LEGACY_FRONTEND'] / 'index.html'; path.chmod(0o644); path.write_bytes(b'foreign mutation') with self.assertRaisesRegex(ValueError, 'historical frontend backup contains'): self.adopt() self.assertTrue(self.ns['LEGACY_MARKER'].exists()) def iso_runtime_fixture(self, old_version='1.9.1-alpha'): # The authenticated OTA archive bundles runtime assets, whereas the # historical ISO could install the qualified UI dist directly. self.marker['previous_version'] = old_version self.ns['LEGACY_MARKER'].write_text(json.dumps(self.marker)) self.install_release(old_version, b'old binary', { 'index.html': b'old UI', 'aiui/extra.html': b'accepted UI extra', 'archipelago-runtime/apps/fixture/manifest.yml': b'signed app', 'archipelago-runtime/scripts/fixture.sh': b'signed script', }) def test_accepts_exact_191_iso_ui_without_ota_runtime_subtree(self): self.iso_runtime_fixture() state = self.adopt() old = Path(state['transaction_path']) / 'old-frontend' self.assertFalse((old / 'archipelago-runtime').exists()) self.assertEqual((old / 'index.html').read_bytes(), b'old UI') def test_accepts_oldest_supported_iso_ui_without_ota_runtime_subtree(self): self.iso_runtime_fixture('1.8.13-alpha') state = self.adopt() self.assertEqual(state['old_version'], '1.8.13-alpha') self.assertFalse((Path(state['transaction_path']) / 'old-frontend/archipelago-runtime').exists()) def test_accepts_last_18_iso_ui_without_ota_runtime_subtree(self): self.iso_runtime_fixture('1.8.22-alpha') state = self.adopt() self.assertEqual(state['old_version'], '1.8.22-alpha') def test_accepts_first_19_iso_ui_without_ota_runtime_subtree(self): self.iso_runtime_fixture('1.9.0-alpha') state = self.adopt() self.assertEqual(state['old_version'], '1.9.0-alpha') def test_rejects_partially_retained_ota_runtime_subtree(self): self.iso_runtime_fixture() partial = self.ns['LEGACY_FRONTEND'] / 'archipelago-runtime/apps/fixture/manifest.yml' partial.parent.mkdir(parents=True) partial.write_bytes(b'signed app') with self.assertRaisesRegex(ValueError, 'historical frontend backup contains'): self.adopt() def test_rejects_runtime_symlink_without_touching_its_target(self): self.iso_runtime_fixture() outside = self.root / 'untrusted-runtime-target' outside.mkdir() sentinel = outside / 'sentinel' sentinel.write_bytes(b'keep this external tree unchanged') (self.ns['LEGACY_FRONTEND'] / 'archipelago-runtime').symlink_to(outside, target_is_directory=True) with self.assertRaises((OSError, ValueError)): self.adopt() self.assertEqual(sentinel.read_bytes(), b'keep this external tree unchanged') self.assertTrue(self.ns['LEGACY_MARKER'].exists()) def test_rejects_iso_projection_with_other_missing_file(self): self.iso_runtime_fixture() (self.ns['LEGACY_FRONTEND'] / 'aiui/extra.html').unlink() with self.assertRaisesRegex(ValueError, 'historical frontend backup contains'): self.adopt() def test_rejects_iso_projection_with_changed_existing_file(self): self.iso_runtime_fixture() path = self.ns['LEGACY_FRONTEND'] / 'index.html' path.chmod(0o644) path.write_bytes(b'changed UI') with self.assertRaisesRegex(ValueError, 'historical frontend backup contains'): self.adopt() def test_iso_projection_requires_exact_historical_version(self): self.iso_runtime_fixture() expected = self.root / 'expected' expected.mkdir() (expected / 'archipelago-runtime').mkdir() self.assertFalse(self.ns['project_legacy_iso_frontend']( self.ns['LEGACY_FRONTEND'], expected, '1.8.12-alpha')) self.assertTrue((expected / 'archipelago-runtime').exists()) def test_iso_projection_allowlist_is_exact_supported_range(self): allowed = self.ns['ISO_FRONTEND_OLD_RELEASES'] self.assertEqual(allowed, { *(f'1.8.{minor}-alpha' for minor in range(13, 23)), '1.9.0-alpha', '1.9.1-alpha', }) self.assertNotIn('1.8.12-alpha', allowed) self.assertNotIn('1.9.2-alpha', allowed) def test_rejects_modified_incoming_pair(self): self.ns['BINARY'].write_bytes(b'foreign current binary') with self.assertRaisesRegex(ValueError, 'exact signed incoming pair'): self.adopt() def routing_fixture(self, retained): relative = 'archipelago-runtime/image-recipe/configs/nginx-archipelago.conf' self.install_release('1.9.1-alpha', b'old binary', { 'index.html': b'old UI', 'aiui/extra.html': b'accepted UI extra', relative: b'signed original routing\n'}) path = self.ns['LEGACY_FRONTEND'] / relative path.parent.mkdir(parents=True) path.write_bytes(retained) self.ns['historical_frontend_modes'](self.ns['LEGACY_FRONTEND']) (self.root / 'routing').write_bytes(b'root authorized live routing\n') (self.root / 'routing').chmod(0o644) return relative def test_accepts_original_signed_template_when_live_routing_differs(self): relative = self.routing_fixture(b'signed original routing\n') state = self.adopt() self.assertEqual((Path(state['transaction_path']) / 'old-frontend' / relative).read_bytes(), b'signed original routing\n') def test_accepts_exact_root_authorized_routing_in_retained_template(self): relative = self.routing_fixture(b'root authorized live routing\n') state = self.adopt() self.assertEqual((Path(state['transaction_path']) / 'old-frontend' / relative).read_bytes(), b'root authorized live routing\n') def test_rejects_template_that_matches_neither_signed_nor_live_routing(self): self.routing_fixture(b'foreign routing mutation\n') with self.assertRaisesRegex(ValueError, 'historical frontend backup contains'): self.adopt() def test_rejects_previous_manifest_signature_failure(self): self.documents.pop('1.9.1-alpha') with self.assertRaisesRegex(ValueError, 'signature refusal'): self.adopt() def test_refuses_existing_runtime_ownership(self): self.ns['RUNTIME_RECORD'].write_text('{}') with self.assertRaisesRegex(ValueError, 'already has an owner'): self.adopt() self.assertEqual(self.downloads, []) def test_keeps_changed_legacy_marker_for_reconciliation(self): state = self.adopt() self.ns['LEGACY_MARKER'].write_text('{}') with self.assertRaisesRegex(ValueError, 'marker changed'): self.ns['retain_legacy_marker'](Path(state['transaction_path']), state) self.assertEqual(self.ns['LEGACY_MARKER'].read_text(), '{}') if __name__ == '__main__': unittest.main()