254 lines
15 KiB
Python
254 lines
15 KiB
Python
"""Exact UAT binding: real tar/initrd/APT-policy fixtures, no installed mutation."""
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
import uat_kernel as kernel
|
|
|
|
|
|
def load(name):
|
|
spec = importlib.util.spec_from_file_location(name, SCRIPTS / (name + '.py'))
|
|
value = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(value)
|
|
return value
|
|
|
|
|
|
uat = load('build-iso-uat')
|
|
security = load('kernel-security')
|
|
|
|
|
|
class KernelTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='uat-kernel-')
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.pin = dict(architecture='amd64', package='linux-image-6.12.113+deb13-amd64',
|
|
version='6.12.113-1', source='linux-signed-amd64',
|
|
archive_sha256=hashlib.sha256(b'fixture archive').hexdigest(),
|
|
vmlinuz_sha256=hashlib.sha256(b'fixture kernel').hexdigest())
|
|
self.release = kernel.validate(self.pin)
|
|
|
|
def test_strict_pin_and_architecture(self):
|
|
for key, bad in [('package', 'linux-image-amd64'), ('version', '6.12.115-1'),
|
|
('package', 'linux-image-6.12.113+deb13-amd64;true'),
|
|
('source', 'linux'), ('archive_sha256', 'a' * 63), ('architecture', 'x86_64')]:
|
|
value = dict(self.pin, **{key: bad})
|
|
with self.subTest(key=key, value=bad), self.assertRaises(ValueError):
|
|
kernel.validate(value)
|
|
with self.assertRaises(ValueError):
|
|
kernel.validate(self.pin, 'arm64')
|
|
with self.assertRaises(ValueError):
|
|
kernel.validate({})
|
|
|
|
def receipt(self, pin=None):
|
|
pin = pin or self.pin
|
|
return {k: pin[k] for k in ('package', 'version', 'source', 'archive_sha256')} | {
|
|
'origin': 'Debian', 'files': {'/boot/vmlinuz-' + kernel.validate(pin): pin['vmlinuz_sha256'],
|
|
'/usr/lib/modules/' + kernel.validate(pin) + '/kernel/test.ko': hashlib.sha256(b'module').hexdigest()}}
|
|
|
|
def test_higher_fixed_kernel_still_fails_exact_binding(self):
|
|
higher = dict(self.pin, package='linux-image-6.12.115+deb13-amd64', version='6.12.115-1')
|
|
receipt = self.receipt(higher)
|
|
self.assertEqual(security.assess(receipt)['status'], 'fixed-package')
|
|
with self.assertRaisesRegex(ValueError, 'exact qualified'):
|
|
uat.validate_kernel_receipt(receipt, security, self.pin)
|
|
uat.validate_kernel_receipt(self.receipt(), security, self.pin)
|
|
|
|
def test_kernel_module_declaration_must_match_qualified_archive(self):
|
|
expected = self.receipt()
|
|
actual = copy.deepcopy(expected)
|
|
module = next(key for key in actual['files'] if key.endswith('.ko'))
|
|
actual['files'][module] = hashlib.sha256(b'replaced module').hexdigest()
|
|
with self.assertRaisesRegex(ValueError, 'qualified archive payload'):
|
|
uat.validate_kernel_receipt(actual, security, self.pin, expected)
|
|
uat.validate_kernel_receipt(expected, security, self.pin, expected)
|
|
|
|
def test_apt_requires_current_authenticated_exact_archive(self):
|
|
version = SimpleNamespace(version=self.pin['version'], source_name=self.pin['source'], architecture='amd64',
|
|
record={'SHA256': self.pin['archive_sha256']},
|
|
origins=[SimpleNamespace(archive='stable-security', trusted=True, origin='Debian')])
|
|
package = SimpleNamespace(versions={self.pin['version']: version}, candidate=version)
|
|
cache = {self.pin['package']: package}
|
|
self.assertIs(kernel.qualified_apt_version(self.pin, cache), version)
|
|
for key, bad in [('source_name', 'unknown'), ('architecture', 'arm64'), ('record', {'SHA256': '0' * 64}),
|
|
('origins', [SimpleNamespace(archive='now', trusted=True, origin='Debian')]),
|
|
('origins', [SimpleNamespace(archive='stable-security', trusted=False, origin='Debian')])]:
|
|
old = getattr(version, key); setattr(version, key, bad)
|
|
with self.subTest(key=key), self.assertRaises(ValueError):
|
|
kernel.qualified_apt_version(self.pin, cache)
|
|
setattr(version, key, old)
|
|
package.candidate = None
|
|
with self.assertRaises(ValueError):
|
|
kernel.qualified_apt_version(self.pin, cache)
|
|
with self.assertRaisesRegex(ValueError, 'absent'):
|
|
kernel.qualified_apt_version(self.pin, {})
|
|
|
|
def test_old_schema_and_missing_pin_refused(self):
|
|
for schema in ('archipelago-unsigned-uat-input-v1', 'archipelago-unsigned-uat-input-v2'):
|
|
with self.assertRaises(ValueError):
|
|
uat.inputs({'schema': schema})
|
|
with self.assertRaises((KeyError, ValueError)):
|
|
uat.inputs({'schema': 'archipelago-unsigned-uat-input-v3',
|
|
'qualification': 'qualified-for-unsigned-uat', 'reviewed_by': 'fixture'})
|
|
|
|
def cache(self, pin=None, wrong_initrd=False, selected=None):
|
|
pin = pin or self.pin
|
|
release = kernel.validate(pin)
|
|
image = b'fixture kernel'
|
|
initrd = self.initrd(release)
|
|
binding = {'pin': pin, 'release': release, 'vmlinuz_sha256': hashlib.sha256(image).hexdigest(),
|
|
'initrd_sha256': hashlib.sha256(initrd).hexdigest()}
|
|
files = {'var/lib/dpkg/status': f'Package: {pin["package"]}\nVersion: {pin["version"]}\nStatus: install ok installed\n'.encode(),
|
|
'var/lib/archipelago/kernel-security/' + pin['package'] + '.json': json.dumps(self.receipt(pin)).encode(),
|
|
'var/lib/archipelago/uat-kernel/boot-binding.json': json.dumps(binding).encode(),
|
|
'boot/vmlinuz-' + release: image, 'boot/initrd.img-' + release: b'tampered' if wrong_initrd else initrd,
|
|
'usr/lib/modules/' + release + '/kernel/test.ko': b'module'}
|
|
path = self.root / 'rootfs.tar'
|
|
with tarfile.open(path, 'w') as archive:
|
|
for name, raw in files.items():
|
|
info = tarfile.TarInfo(name); info.size = len(raw)
|
|
archive.addfile(info, io.BytesIO(raw))
|
|
for alias, target in [('vmlinuz', 'boot/vmlinuz-' + (selected or release)),
|
|
('initrd.img', 'boot/initrd.img-' + release)]:
|
|
info = tarfile.TarInfo(alias); info.type = tarfile.SYMTYPE; info.linkname = target
|
|
archive.addfile(info)
|
|
return path
|
|
|
|
def initrd(self, release):
|
|
with tempfile.TemporaryDirectory(dir=self.root) as temp:
|
|
work = Path(temp)
|
|
(work / 'usr/lib/modules' / release).mkdir(parents=True)
|
|
(work / 'usr/lib/modules' / release / 'module.ko').write_bytes(b'module fixture')
|
|
names = '\n'.join(str(p.relative_to(work)) for p in sorted(work.rglob('*'))) + '\n'
|
|
return subprocess.run(['cpio', '-o', '-H', 'newc'], input=names.encode(), cwd=work,
|
|
capture_output=True, check=True).stdout
|
|
|
|
def test_actual_cache_hashes_selection_and_higher_kernel(self):
|
|
kernel.check_tar(self.cache(), self.pin)
|
|
with self.assertRaisesRegex(ValueError, 'boot bytes'):
|
|
kernel.check_tar(self.cache(wrong_initrd=True), self.pin)
|
|
with self.assertRaisesRegex(ValueError, 'boot selection'):
|
|
kernel.check_tar(self.cache(selected='6.12.115+deb13-amd64'), self.pin)
|
|
higher = dict(self.pin, package='linux-image-6.12.115+deb13-amd64', version='6.12.115-1')
|
|
with self.assertRaises(ValueError):
|
|
kernel.check_tar(self.cache(higher), self.pin)
|
|
|
|
def test_actual_initrd_module_binding(self):
|
|
for release in (self.release, '6.12.115+deb13-amd64'):
|
|
work = self.root / release
|
|
(work / 'usr/lib/modules' / release).mkdir(parents=True)
|
|
(work / 'usr/lib/modules' / release / 'module.ko').write_bytes(b'module fixture')
|
|
names = '\n'.join(str(p.relative_to(work)) for p in sorted(work.rglob('*'))) + '\n'
|
|
initrd = work / 'initrd'
|
|
with initrd.open('wb') as stream:
|
|
subprocess.run(['cpio', '-o', '-H', 'newc'], input=names.encode(), cwd=work,
|
|
stdout=stream, stderr=subprocess.PIPE, check=True)
|
|
if release == self.release:
|
|
kernel.check_initrd(initrd, self.release)
|
|
else:
|
|
with self.assertRaises(ValueError):
|
|
kernel.check_initrd(initrd, self.release)
|
|
|
|
def test_real_apt_policy_selects_exact_and_forbids_newer_meta(self):
|
|
# apt-cache consumes synthetic Packages without apt update or trust
|
|
# overrides. This tests resolver policy only, not APT authentication.
|
|
lists = self.root / 'lists'; lists.mkdir(); (lists / 'partial').mkdir()
|
|
prefs = self.root / 'preferences'; prefs.write_text(kernel.preferences(self.pin))
|
|
parts = self.root / 'parts'; parts.mkdir()
|
|
sources = self.root / 'sources.list'
|
|
sources.write_text('deb https://fixture.invalid/debian trixie main\n')
|
|
status = self.root / 'status'; status.write_text('')
|
|
packages = ''
|
|
for name, version in [(self.pin['package'], self.pin['version']),
|
|
(self.pin['package'], '6.12.113-2'),
|
|
('linux-image-6.12.115+deb13-amd64', '6.12.115-1'),
|
|
('linux-image-amd64', '6.12.115-1')]:
|
|
packages += f'Package: {name}\nVersion: {version}\nArchitecture: amd64\nDescription: synthetic\n\n'
|
|
(lists / 'fixture.invalid_debian_dists_trixie_main_binary-amd64_Packages').write_text(packages)
|
|
args = ['apt-cache', '-o', f'Dir::State::lists={lists}', '-o', f'Dir::State::status={status}',
|
|
'-o', f'Dir::Etc::sourcelist={sources}', '-o', f'Dir::Etc::sourceparts={parts}',
|
|
'-o', f'Dir::Etc::preferences={prefs}', '-o', f'Dir::Etc::preferencesparts={parts}',
|
|
'-o', 'Dir::Cache::pkgcache=', '-o', 'Dir::Cache::srcpkgcache=', 'policy']
|
|
exact = subprocess.check_output(args + [self.pin['package']], text=True)
|
|
self.assertIn('Candidate: 6.12.113-1', exact)
|
|
for name in ('linux-image-amd64', 'linux-image-6.12.115+deb13-amd64'):
|
|
output = subprocess.check_output(args + [name], text=True)
|
|
self.assertIn('Candidate: (none)', output)
|
|
|
|
def test_both_recipes_order_exact_policy_before_upgrade(self):
|
|
text = (SCRIPTS.parent / 'image-recipe/_archived/build-auto-installer-iso.sh').read_text()
|
|
docker = text[text.index('FROM docker.io/library/debian:trixie@sha256:'):text.index('# Strip docs, man pages')]
|
|
self.assertLess(docker.index('uat_kernel.py prepare'), docker.index('-y full-upgrade'))
|
|
self.assertIn('"$LINUX_IMAGE_PKG"', docker)
|
|
self.assertIn('LINUX_IMAGE_PKG="$UAT_KERNEL_SPEC"', text)
|
|
live = text.split("<<'INSTALLER_ENV_EOF'", 1)[1].split('\nINSTALLER_ENV_EOF', 1)[0]
|
|
self.assertLess(live.index('KERNEL_INCLUDE=""'), live.index('debootstrap --variant=minbase --arch='))
|
|
self.assertIn('${KERNEL_INCLUDE}grub-efi-', live)
|
|
self.assertLess(live.index('uat_kernel.py prepare'), live.index('full-upgrade'))
|
|
self.assertIn('"$KERNEL_INSTALL"', live)
|
|
self.assertIn('KVER=${KERNEL_INSTALL%%=*}', live)
|
|
self.assertLess(live.index('update-initramfs -c -k $KVER'), live.index('uat_kernel.py verify'))
|
|
cache = text[text.index('RECIPE_HASH=$('):text.index('if [ ! -f "$ROOTFS_TAR" ]')]
|
|
self.assertIn('cat "$WORK_DIR/uat-kernel.json"', cache)
|
|
self.assertIn('cat "$WORK_DIR/uat_kernel.py"', cache)
|
|
self.assertIn('cache "$WORK_DIR/uat-kernel.json" --archive "$ROOTFS_TAR"', text)
|
|
|
|
def test_actual_builder_cli_pin_cannot_be_inherited_or_defaulted(self):
|
|
builder = SCRIPTS.parent / 'image-recipe/_archived/build-auto-installer-iso.sh'
|
|
prefix = builder.read_text().split('# Every build starts from explicitly reviewed input bytes', 1)[0]
|
|
bundle = self.root / 'oci'; bundle.mkdir(); (bundle / 'BUNDLE.json').write_text('{}')
|
|
path = self.root / 'pin.json'; path.write_text(json.dumps(self.pin))
|
|
def invoke(args, **env):
|
|
return subprocess.run(['bash', '-c', prefix + '\nprintf "%s" "$UAT_KERNEL_SPEC"\n', str(builder), *args],
|
|
env=dict(os.environ, ARCH='amd64', **env), capture_output=True, text=True)
|
|
self.assertEqual(invoke(['--unsigned-uat-oci', str(bundle)]).returncode, 2)
|
|
self.assertEqual(invoke(['--uat-kernel', str(path)]).returncode, 2)
|
|
good = invoke(['--unsigned-uat-oci', str(bundle), '--uat-kernel', str(path)])
|
|
self.assertEqual(good.returncode, 0, good.stderr)
|
|
self.assertEqual(good.stdout, self.pin['package'] + '=' + self.pin['version'])
|
|
generic = invoke([], UAT_KERNEL=str(path), UAT_KERNEL_SPEC='injected')
|
|
self.assertEqual((generic.returncode, generic.stdout), (0, ''))
|
|
path.write_text(json.dumps(dict(self.pin, version='6.12.115-1')))
|
|
self.assertNotEqual(invoke(['--unsigned-uat-oci', str(bundle), '--uat-kernel', str(path)]).returncode, 0)
|
|
|
|
def test_rendered_rootfs_and_executed_live_bootstrap_select_exact(self):
|
|
text = (SCRIPTS.parent / 'image-recipe/_archived/build-auto-installer-iso.sh').read_text()
|
|
fragment = text[text.index(' cat > "$WORK_DIR/Dockerfile.rootfs"'):text.index('\nDOCKERFILE_TAIL') + len('\nDOCKERFILE_TAIL')]
|
|
env = dict(os.environ, WORK_DIR=str(self.root), LINUX_IMAGE_PKG=self.pin['package'] + '=' + self.pin['version'],
|
|
GRUB_EFI_PKG='grub-efi-amd64', GRUB_EFI_SIGNED_PKG='grub-efi-amd64-signed', GRUB_PC_PKG='grub-pc-bin')
|
|
subprocess.run(['bash', '-ec', fragment], env=env, check=True)
|
|
docker = (self.root / 'Dockerfile.rootfs').read_text()
|
|
self.assertIn(' linux-image-6.12.113+deb13-amd64=6.12.113-1 \\', docker)
|
|
self.assertNotIn(' linux-image-amd64 \\', docker)
|
|
logical = docker.replace('\\\n', '')
|
|
for line in logical.splitlines():
|
|
if line.startswith('RUN '):
|
|
subprocess.run(['bash', '-n'], input=line[4:].encode(), check=True)
|
|
live = text.split("<<'INSTALLER_ENV_EOF'", 1)[1].split('\nINSTALLER_ENV_EOF', 1)[0]
|
|
fragment = live[live.index('KERNEL_INCLUDE='):live.index('\n# Install live-boot via chroot')]
|
|
fragment = fragment.replace('/output/', str(self.root) + '/')
|
|
(self.root / 'uat_kernel.py').write_bytes((SCRIPTS / 'uat_kernel.py').read_bytes())
|
|
(self.root / 'uat-kernel.json').write_text(json.dumps(self.pin))
|
|
command = 'debootstrap() { printf "%s\\n" "$@"; };\n' + fragment + '\nprintf "INSTALL=%s\\n" "$KERNEL_INSTALL"'
|
|
result = subprocess.check_output(['bash', '-ec', command], env=dict(os.environ, DEB_ARCH='amd64'), text=True)
|
|
self.assertNotIn('linux-image-amd64', result)
|
|
self.assertIn('INSTALL=' + self.pin['package'] + '=' + self.pin['version'], result)
|
|
self.assertIn('--include=systemd,', result)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|