"""OCI archive, candidate staging and import guards; no host runtime imports.""" import copy import gzip import hashlib import importlib.util import io import json import os from pathlib import Path import re import shutil import subprocess import sys import tarfile import tempfile import unittest from unittest.mock import patch SCRIPTS = Path(__file__).resolve().parents[1] sys.path.insert(0, str(SCRIPTS)) import uat_oci as oci def load(name): spec = importlib.util.spec_from_file_location(name, SCRIPTS / (name + '.py')) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module builder = load('build-iso-uat') importer = load('import-uat-oci') class OCITests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='uat-oci-fixture-', dir='/tmp/opencode') self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.source = self.root / 'source' self.staged = self.root / 'staged' self.staged.mkdir() self.archive = self.staged / 'lnd_tor_oci' self.identity = self.staged / 'lnd_tor_identity' self.sources = self.staged / 'lnd_tor_sources' self.make() def make(self, architecture='amd64', user='100:100', bad_diffid=False): blobs = {} def blob(raw, kind): digest = hashlib.sha256(raw).hexdigest() blobs['blobs/sha256/' + digest] = raw return {'mediaType': 'application/vnd.oci.image.' + kind, 'digest': 'sha256:' + digest, 'size': len(raw)} layer_raw = b'synthetic uncompressed layer' layer = blob(gzip.compress(layer_raw), 'layer.v1.tar+gzip') config = blob(json.dumps({'os': 'linux', 'architecture': architecture, 'config': {'User': user}, 'rootfs': {'type': 'layers', 'diff_ids': ['sha256:' + ('0' * 64 if bad_diffid else hashlib.sha256(layer_raw).hexdigest())]}}).encode(), 'config.v1+json') manifest = blob(json.dumps({'schemaVersion': 2, 'config': config, 'layers': [layer]}).encode(), 'manifest.v1+json') self.contents = {**blobs, 'oci-layout': b'{"imageLayoutVersion":"1.0.0"}', 'index.json': json.dumps({'schemaVersion': 2, 'manifests': [manifest]}).encode()} self.write_tar() ref = oci.REPOSITORY + '@' + manifest['digest'] for name in oci.SOURCES | {oci.NATIVE}: path = self.source / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text('synthetic source') (self.source / 'apps/lnd-tor/manifest.yml').write_text('app:\n container:\n image: ' + ref + '\n') (self.source / oci.NATIVE).write_text(f'const QUALIFIED_IMAGES: &[&str] = &["{ref}"];') self.sources.write_text(json.dumps({n: oci.sha(self.source / n) for n in oci.SOURCES})) self.facts = dict(source_receipt_sha256=oci.sha(self.sources), sources=json.loads(self.sources.read_text()), archive_sha256=oci.sha(self.archive), archive_bytes=self.archive.stat().st_size, oci_manifest_digest=manifest['digest'], config_digest=config['digest'], verified_blobs=[n.rsplit('/', 1)[1] for n in blobs]) self.identity.write_text(json.dumps(self.facts)) for name in ('uat_oci.py', 'import-uat-oci.py','c01_oci_job.py'): target = self.source / 'scripts' / name target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes((SCRIPTS / name).read_bytes()) def write_tar(self, link=False): with tarfile.open(self.archive, 'w') as archive: for name, raw in self.contents.items(): member = tarfile.TarInfo(name) member.size = len(raw) archive.addfile(member, io.BytesIO(raw)) if link: member = tarfile.TarInfo('unsafe') member.type = tarfile.SYMTYPE; member.linkname = '/etc/passwd' archive.addfile(member) def check(self): return oci.check(self.source, self.archive, self.identity, self.sources) def refresh_outer_hash(self): self.facts.update(archive_sha256=oci.sha(self.archive), archive_bytes=self.archive.stat().st_size) self.identity.write_text(json.dumps(self.facts)) def test_complete_blob_and_diffid_validation(self): proof = self.check() self.assertEqual(proof['verified_blobs'], 3) self.assertEqual(proof['platform'], 'linux/amd64') def test_corrupt_blob_even_with_updated_archive_hash(self): name = next(n for n in self.contents if n.startswith('blobs/')) self.contents[name] += b'tamper' self.write_tar(); self.refresh_outer_hash() with self.assertRaisesRegex(ValueError, 'blob hash mismatch'): self.check() def test_old_builder_digest_or_config_id_rejected(self): for key in ('oci_manifest_digest', 'config_digest'): original = self.facts[key] self.facts[key] = 'sha256:' + '0' * 64 self.identity.write_text(json.dumps(self.facts)) with self.subTest(key=key), self.assertRaises(ValueError): self.check() self.facts[key] = original def test_wrong_platform_and_default_identity(self): for arch, user in (('arm64', '100:100'), ('amd64', '0:0')): self.make(arch, user) with self.subTest(arch=arch, user=user), self.assertRaisesRegex(ValueError, 'platform/default user'): self.check() def test_descriptor_size_and_uncompressed_diffid_refused(self): index = json.loads(self.contents['index.json']) index['manifests'][0]['size'] += 1 self.contents['index.json'] = json.dumps(index).encode() self.write_tar(); self.refresh_outer_hash() with self.assertRaisesRegex(ValueError, 'descriptor'): self.check() self.make(bad_diffid=True) with self.assertRaisesRegex(ValueError, 'diffID'): self.check() def test_tag_source_or_missing_native_admission_rejected(self): (self.source / 'apps/lnd-tor/manifest.yml').write_text(' image: ' + oci.REPOSITORY + ':latest\n') with self.assertRaisesRegex(ValueError, 'stale source'): self.check() self.make() (self.source / oci.NATIVE).write_text('const QUALIFIED_IMAGES: &[&str] = &[];') with self.assertRaisesRegex(ValueError, 'native admission'): self.check() def test_archive_hash_and_link_refusal(self): self.archive.write_bytes(b'wrong archive') with self.assertRaisesRegex(ValueError, 'archive hash/size'): self.check() self.write_tar(link=True); self.refresh_outer_hash() with self.assertRaisesRegex(ValueError, 'unsafe/duplicate'): self.check() def test_stage_preserves_bytes_and_defers_without_runtime(self): bundle = builder.stage_oci(self.source, self.staged, {'source_commit': None, 'companion': {'fixture': True}}) target = self.root / 'target' with patch.object(importer.subprocess, 'Popen', side_effect=AssertionError('stage must not invoke runtime')): result = importer.stage(bundle, target) self.assertEqual(result['status'], 'staged-import-deferred') importer.stage(bundle, target) # exact repeat is safe self.assertEqual(oci.inventory(bundle), oci.inventory(target / 'opt/archipelago/unsigned-uat-oci')) (bundle / 'identity.json').write_text('{}') with self.assertRaisesRegex(ValueError, 'files mismatch'): importer.stage(bundle, target) def test_both_actual_installer_hooks_stage_without_runtime_or_intent_change(self): bundle = builder.stage_oci(self.source, self.staged, {'source_commit': None, 'companion': {'fixture': True}}) media = self.root / 'media' shutil.copytree(bundle, media / 'archipelago/unsigned-uat-oci') for index, name in enumerate(('image-recipe/_archived/build-auto-installer-iso.sh', 'image-recipe/archipelago-scripts/install-to-disk.sh')): text = (SCRIPTS.parent / name).read_text() hook = re.search(r' ?if \[ -d "\$BOOT_MEDIA/archipelago/unsigned-uat-oci" \]; then.*?\n ?fi', text, re.S) if hook is None: hook = re.search(r'if \[ -d "\$BOOT_MEDIA/archipelago/unsigned-uat-oci" \]; then.*?\nfi', text, re.S) self.assertIsNotNone(hook) target = self.root / f'installer-{index}' target.mkdir() marker = target / 'user-uninstalled.json' marker.write_text('{"lnd":true,"lnd-tor":true}') before = marker.read_bytes() block = re.sub(r'--target-root /mnt/(target|archipelago)', '--target-root "$TARGET"', hook[0]) result = subprocess.run(['bash', '-ec', block], env=dict(os.environ, BOOT_MEDIA=str(media), TARGET=str(target), PYTHONDONTWRITEBYTECODE='1'), capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn('staged-import-deferred', result.stdout) self.assertEqual(marker.read_bytes(), before) uat_oci_path = target / 'opt/archipelago/unsigned-uat-oci' self.assertEqual(oci.inventory(bundle), oci.inventory(uat_oci_path)) def test_missing_admission_helpers_defer_without_podman(self): with patch.object(importer.os, 'getresuid', return_value=(0, 0, 0)), \ patch.object(importer.Path, 'is_file', return_value=False), \ patch.object(importer.subprocess, 'Popen') as command: with self.assertRaisesRegex(FileNotFoundError, 'defer'): importer.appointed_import(self.root, 'bundle', 'runtime', 'storage') command.assert_not_called() def test_import_only_load_tag_inspect_and_exact_identity(self): proof = self.check() commands = [] actual = dict(Id=proof['config_id'], Digest=proof['manifest_digest'], Os='linux', Architecture='amd64', Config={'User': '100:100'}) def call(args): commands.append(args) return json.dumps([actual]) if args[0] == 'image' else '' importer.import_image(call, self.archive, proof) self.assertEqual([c[0] for c in commands], ['load', 'tag', 'image']) actual['Digest'] = 'sha256:' + '0' * 64 with self.assertRaisesRegex(ValueError, 'imported image'): importer.import_image(call, self.archive, proof) def test_storage_mismatch_root_and_wrong_config_refused(self): config = {'graphroot': '/var/lib/appointed/storage', 'runtime_dir': '/run/user/2200'} path = Path('/home/appointed/.config/containers/storage.conf') storage = {'storage': {'graphroot': config['graphroot'], 'runroot': '/run/user/2200/containers', 'driver': 'vfs'}} info = {'host': {'security': {'rootless': True}}, 'store': {'graphRoot': config['graphroot'], 'runRoot': storage['storage']['runroot'], 'configFile': str(path), 'graphDriverName': 'vfs'}} importer.storage_check(info, config, storage, path) for key, value in [('graphRoot', '/var/lib/containers/storage'), ('runRoot', '/run/containers/storage'), ('configFile', '/etc/containers/storage.conf')]: wrong = copy.deepcopy(info); wrong['store'][key] = value with self.subTest(key=key), self.assertRaises(ValueError): importer.storage_check(wrong, config, storage, path) info['host']['security']['rootless'] = False with self.assertRaises(ValueError): importer.storage_check(info, config, storage, path) storage['storage']['graphroot'] = '/var/lib/containers/storage' with self.assertRaisesRegex(ValueError, 'before invoking Podman'): importer.storage_declaration(config, storage) def test_exact_companion_receipt_contains_metadata_digest(self): path = self.source / 'neode-ui/public/packages/archipelago-companion.json' path.parent.mkdir(parents=True); path.write_text('{"fixture":true}') with patch.object(builder.payload, 'check_companion', return_value={'sha256': 'fixture-apk', 'certificate_sha256': 'fixture-cert'}): value = builder.companion_identity(self.source) self.assertEqual(value['metadata_sha256'], oci.sha(path)) self.assertEqual(value['certificate_sha256'], 'fixture-cert') if __name__ == '__main__': unittest.main()