241 lines
12 KiB
Python
241 lines
12 KiB
Python
"""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()
|