170 lines
9.2 KiB
Python
170 lines
9.2 KiB
Python
"""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 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()
|