370 lines
19 KiB
Python
370 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Prepare a private UNSIGNED UAT ISO; never signs, publishes or deploys.
|
|
|
|
See docs/unsigned-uat-iso.md for the qualification receipt and resource wrapper.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
|
|
import release_payload as payload
|
|
import uat_oci
|
|
import uat_kernel
|
|
|
|
|
|
def sha(path):
|
|
if path.is_symlink() or not path.is_file():
|
|
raise ValueError(f'not a regular file: {path}')
|
|
with path.open('rb') as stream:
|
|
return hashlib.file_digest(stream, 'sha256').hexdigest()
|
|
|
|
|
|
def git(root, *args):
|
|
return subprocess.check_output(['git', '-C', str(root), *args], text=True).strip()
|
|
|
|
|
|
def source(root, commit, version):
|
|
import tomllib
|
|
if not re.fullmatch('[0-9a-f]{40}', commit) or git(root, 'rev-parse', 'HEAD') != commit:
|
|
raise ValueError('UAT requires exact full source commit at HEAD')
|
|
if git(root, 'status', '--porcelain', '--untracked-files=all'):
|
|
raise ValueError('UAT requires clean frozen source, including untracked files')
|
|
if git(root, 'ls-files', '--stage').find('160000 ') >= 0:
|
|
raise ValueError('submodules require an explicitly qualified materialization')
|
|
if (tomllib.loads((root / 'core/archipelago/Cargo.toml').read_text())['package']['version'] != version
|
|
or json.loads((root / 'neode-ui/package.json').read_text())['version'] != version):
|
|
raise ValueError('source version mismatch')
|
|
return git(root, 'rev-parse', 'HEAD^{tree}')
|
|
|
|
|
|
ARTIFACTS = ('backend', 'frontend', 'reticulum', 'rnodeconf', 'esptool',
|
|
'lnd_tor_oci', 'lnd_tor_identity', 'lnd_tor_sources')
|
|
|
|
|
|
def inputs(receipt):
|
|
if receipt.get('schema') != 'archipelago-unsigned-uat-input-v3':
|
|
raise ValueError('not an unsigned UAT input receipt')
|
|
if receipt.get('qualification') != 'qualified-for-unsigned-uat':
|
|
raise ValueError('candidate not qualified for unsigned UAT')
|
|
if not receipt.get('reviewed_by'):
|
|
raise ValueError('qualification reviewer required')
|
|
kernel_inputs(receipt['kernel'])
|
|
result = {}
|
|
for name in (*ARTIFACTS, 'evidence'):
|
|
item = receipt[name]
|
|
path = Path(item['path'])
|
|
if not path.is_absolute() or sha(path) != item['sha256']:
|
|
raise ValueError(f'input hash/path mismatch: {name}')
|
|
result[name] = path
|
|
return result
|
|
|
|
|
|
def kernel_inputs(kernel):
|
|
if set(kernel) != {'pin', 'archive', 'payload_receipt', 'evidence'}:
|
|
raise ValueError('exact UAT kernel input/provenance fields required')
|
|
uat_kernel.validate(kernel['pin'], 'amd64')
|
|
for key in ('archive', 'payload_receipt', 'evidence'):
|
|
ref = kernel[key]
|
|
path = Path(ref['path'])
|
|
if not path.is_absolute() or sha(path) != ref['sha256']:
|
|
raise ValueError('kernel input hash/path mismatch: ' + key)
|
|
if kernel['archive']['sha256'] != kernel['pin']['archive_sha256']:
|
|
raise ValueError('qualified kernel archive differs from pin')
|
|
uat_kernel.check_receipt(json.loads(Path(kernel['payload_receipt']['path']).read_text()), kernel['pin'])
|
|
return kernel['pin']
|
|
|
|
|
|
def tree(root):
|
|
if not root.is_dir() or root.is_symlink():
|
|
raise ValueError(f'missing/non-directory payload: {root}')
|
|
result = {}
|
|
for path in sorted(root.rglob('*')):
|
|
if path.is_symlink():
|
|
raise ValueError(f'payload symlink: {path}')
|
|
if path.is_dir():
|
|
continue
|
|
result[path.relative_to(root).as_posix()] = sha(path)
|
|
return result
|
|
|
|
|
|
def compare_payload(extracted, staged):
|
|
if sha(extracted / 'bin/archipelago') != sha(staged / 'backend'):
|
|
raise ValueError('ISO backend byte mismatch')
|
|
if tree(extracted / 'web-ui') != tree(staged / 'web-ui'):
|
|
raise ValueError('ISO frontend/Companion byte mismatch')
|
|
if tree(extracted / 'unsigned-uat-oci') != tree(staged / 'unsigned-uat-oci'):
|
|
raise ValueError('ISO OCI candidate bundle byte mismatch')
|
|
for name, installed in [('reticulum', 'archy-reticulum-daemon'),
|
|
('rnodeconf', 'archy-rnodeconf'), ('esptool', 'archy-esptool')]:
|
|
if sha(extracted / 'bin' / installed) != sha(staged / name):
|
|
raise ValueError(f'ISO helper byte mismatch: {name}')
|
|
|
|
|
|
def companion_identity(root):
|
|
return {**payload.check_companion(root), 'metadata_sha256': sha(
|
|
root / 'neode-ui/public/packages/archipelago-companion.json')}
|
|
|
|
|
|
def stage_oci(root, staged, receipt):
|
|
proof = uat_oci.check(root, staged / 'lnd_tor_oci', staged / 'lnd_tor_identity', staged / 'lnd_tor_sources')
|
|
bundle = staged / 'unsigned-uat-oci'
|
|
bundle.mkdir()
|
|
for src, dest in (('lnd_tor_oci', 'lnd-tor.oci.tar'), ('lnd_tor_identity', 'identity.json'),
|
|
('lnd_tor_sources', 'final-source.json')):
|
|
shutil.copyfile(staged / src, bundle / dest)
|
|
for name in sorted(uat_oci.SOURCES | {uat_oci.NATIVE}):
|
|
target = bundle / 'source' / name
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(root / name, target)
|
|
for name in ('uat_oci.py', 'import-uat-oci.py','c01_oci_job.py'):
|
|
shutil.copyfile(root / 'scripts' / name, bundle / name)
|
|
declaration = dict(schema='archipelago-unsigned-uat-oci-v1', source_commit=receipt['source_commit'],
|
|
companion=receipt['companion'], oci=proof, files=uat_oci.inventory(bundle))
|
|
(bundle / 'BUNDLE.json').write_text(json.dumps(declaration, sort_keys=True, indent=2) + '\n')
|
|
uat_oci.bundle(bundle)
|
|
return bundle
|
|
|
|
|
|
def run(command, root, log, env=None):
|
|
with log.open('xb') as stream:
|
|
stream.write((json.dumps([str(x) for x in command]) + '\n').encode())
|
|
stream.flush()
|
|
subprocess.run([str(x) for x in command], cwd=root, env=env,
|
|
stdout=stream, stderr=subprocess.STDOUT, check=True)
|
|
|
|
|
|
def kernel_tar(path, security, pin=None, qualified_payload=None):
|
|
"""Verify every kernel receipt byte without extracting rootfs links/devices."""
|
|
if pin is not None:
|
|
uat_kernel.check_tar(path, pin)
|
|
with tarfile.open(path) as archive:
|
|
members = {}
|
|
for member in archive:
|
|
name = member.name.removeprefix('./').rstrip('/')
|
|
if name.startswith('/') or '..' in name.split('/') or name in members:
|
|
raise ValueError('noncanonical/duplicate rootfs member')
|
|
members[name] = member
|
|
receipts = [m for n, m in members.items()
|
|
if n.startswith('var/lib/archipelago/kernel-security/') and n.endswith('.json')]
|
|
if not receipts:
|
|
raise ValueError('rootfs lacks authenticated kernel receipts')
|
|
for member in receipts:
|
|
if not member.isfile():
|
|
raise ValueError('kernel receipt must be regular')
|
|
receipt = json.load(archive.extractfile(member))
|
|
validate_kernel_receipt(receipt, security, pin, qualified_payload)
|
|
for name, expected in receipt['files'].items():
|
|
key = name.lstrip('/')
|
|
actual = members.get(key)
|
|
if actual is None and key.startswith('lib/'):
|
|
actual = members.get('usr/' + key)
|
|
if actual is None and key.startswith(('usr/share/doc/', 'usr/share/man/',
|
|
'usr/share/lintian/', 'usr/share/linda/')):
|
|
continue
|
|
if actual is None or not actual.isfile():
|
|
raise ValueError(f'rootfs kernel file missing/nonregular: {name}')
|
|
if hashlib.file_digest(archive.extractfile(actual), 'sha256').hexdigest() != expected:
|
|
raise ValueError(f'rootfs kernel hash mismatch: {name}')
|
|
return [m.name for m in receipts]
|
|
|
|
|
|
def validate_kernel_receipt(receipt, security, pin=None, qualified_payload=None):
|
|
if pin is not None:
|
|
uat_kernel.check_receipt(receipt, pin)
|
|
if qualified_payload is not None and receipt.get('files') != qualified_payload.get('files'):
|
|
raise ValueError('kernel file declaration differs from qualified archive payload receipt')
|
|
if security.assess(receipt)['status'] != 'fixed-package':
|
|
raise ValueError('kernel is not a known fixed Debian package')
|
|
release = receipt['package'].removeprefix('linux-image-')
|
|
files = receipt['files']
|
|
if (receipt.get('origin') != 'Debian'
|
|
or not re.fullmatch('[0-9a-f]{64}', receipt.get('archive_sha256', ''))
|
|
or '/boot/vmlinuz-' + release not in files
|
|
or not any(n.startswith(('/lib/modules/' + release + '/', '/usr/lib/modules/' + release + '/')) for n in files)):
|
|
raise ValueError('incomplete authenticated kernel receipt')
|
|
for name, value in files.items():
|
|
if not name.startswith('/') or '..' in name.split('/') or not re.fullmatch('[0-9a-f]{64}', value):
|
|
raise ValueError('invalid kernel file declaration')
|
|
|
|
|
|
def boot_kernel(log, receipt):
|
|
text = log.read_text(errors='replace')
|
|
release = receipt['package'].removeprefix('linux-image-')
|
|
if not re.search(r'Linux version ' + re.escape(release) + r'\s.*Debian ' + re.escape(receipt['version']) + r'(?:\s|\))', text):
|
|
raise ValueError('ISO boot log does not identify the verified fixed kernel build')
|
|
|
|
|
|
def build(root, receipt_path, out):
|
|
receipt_hash = sha(receipt_path)
|
|
receipt = json.loads(receipt_path.read_text())
|
|
commit, version = receipt['source_commit'], receipt['version']
|
|
if not re.fullmatch(r'[0-9]+\.[0-9]+\.[0-9]+(?:-[A-Za-z0-9.-]+)?', version):
|
|
raise ValueError('invalid candidate version')
|
|
source_tree = source(root, commit, version)
|
|
paths = inputs(receipt)
|
|
image_input = receipt['image_inputs']
|
|
image_manifest = Path(image_input['path']).resolve()
|
|
if sha(image_manifest) != image_input['sha256']:
|
|
raise ValueError('reviewed image input manifest changed')
|
|
image_data = json.loads(image_manifest.read_text())
|
|
for name, qualified in [('archipelago', 'backend'), ('web-ui.tar', 'frontend'),
|
|
('archy-reticulum-daemon', 'reticulum'), ('archy-rnodeconf', 'rnodeconf'),
|
|
('archy-esptool', 'esptool')]:
|
|
if image_data['artifacts'][name]['sha256'] != receipt[qualified]['sha256']:
|
|
raise ValueError('image manifest differs from qualified UAT bytes: ' + name)
|
|
pin = kernel_inputs(receipt['kernel'])
|
|
qualified_payload = json.loads(Path(receipt['kernel']['payload_receipt']['path']).read_text())
|
|
payload.check_native(paths['backend'])
|
|
payload.check_archive(root, paths['frontend'], version, commit)
|
|
companion = companion_identity(root)
|
|
if receipt['companion'] != companion:
|
|
raise ValueError('qualified Companion identity differs from exact APK/metadata/signer')
|
|
uat_oci.check(root, paths['lnd_tor_oci'], paths['lnd_tor_identity'], paths['lnd_tor_sources'])
|
|
if out.is_relative_to(root) or not out.is_relative_to(Path('/tmp/opencode')):
|
|
raise ValueError('output must be a fresh directory under /tmp/opencode outside source')
|
|
out.mkdir(exist_ok=False)
|
|
staged = out / 'candidate'
|
|
staged.mkdir()
|
|
(staged / 'kernel-pin.json').write_text(json.dumps(pin, sort_keys=True) + '\n')
|
|
for key in ('archive', 'payload_receipt', 'evidence'):
|
|
shutil.copyfile(receipt['kernel'][key]['path'], staged / ('kernel-' + key))
|
|
if sha(staged / ('kernel-' + key)) != receipt['kernel'][key]['sha256']:
|
|
raise ValueError('kernel input changed during copy')
|
|
for name, path in paths.items():
|
|
shutil.copyfile(path, staged / name)
|
|
if sha(staged / name) != receipt[name]['sha256']:
|
|
raise ValueError('input changed during copy')
|
|
shutil.copyfile(receipt_path, staged / 'qualification.json')
|
|
if sha(staged / 'qualification.json') != receipt_hash:
|
|
raise ValueError('qualification receipt changed')
|
|
oci_bundle = stage_oci(root, staged, receipt)
|
|
oci_hash = sha(oci_bundle / 'BUNDLE.json')
|
|
for name in ('backend', 'reticulum', 'rnodeconf', 'esptool'):
|
|
(staged / name).chmod(0o755)
|
|
payload.check_archive(root, staged / 'frontend', version, commit)
|
|
web = staged / 'web-ui'
|
|
web.mkdir()
|
|
with tarfile.open(staged / 'frontend') as archive:
|
|
archive.extractall(web, filter='data')
|
|
qualified_web = tree(web)
|
|
# A fresh clone excludes ignored stale binaries, build caches and local dist.
|
|
checkout = out / 'source'
|
|
run(['git', 'clone', '--no-local', '--no-hardlinks', '--no-checkout', root, checkout], root, out / 'clone.log')
|
|
run(['git', 'checkout', '--detach', commit], checkout, out / 'checkout.log')
|
|
source(checkout, commit, version)
|
|
for name in ('qemu-system-x86_64', 'xorriso', 'unsquashfs'):
|
|
if not shutil.which(name):
|
|
raise ValueError(f'missing required tool: {name}')
|
|
if not any(Path(p).exists() for p in ('/usr/share/OVMF/OVMF_CODE.fd', '/usr/share/OVMF/OVMF_CODE_4M.fd')):
|
|
raise ValueError('OVMF required for actual UEFI boot')
|
|
env = dict(os.environ, TMPDIR=str(out), PYTHONDONTWRITEBYTECODE='1')
|
|
run(['sudo', '-E', 'env', 'UNBUNDLED=1', 'BUILD_FROM_SOURCE=0', 'DEV_SERVER=localhost',
|
|
'ARCH=x86_64', 'RC=1', f'ARCHIPELAGO_BIN={staged / "backend"}',
|
|
f'ARCHIPELAGO_WEB_UI={web}', f'ARCHY_RETICULUM_DAEMON={staged / "reticulum"}',
|
|
f'ARCHY_RNODECONF={staged / "rnodeconf"}', f'ARCHY_ESPTOOL={staged / "esptool"}',
|
|
'bash', 'image-recipe/build-debian-iso.sh', '--source-ref', commit,
|
|
'--input-manifest', image_manifest, '--unsigned-uat-oci', oci_bundle,
|
|
'--uat-kernel', staged / 'kernel-pin.json'], checkout, out / 'build.log', env)
|
|
isos = list((checkout / 'image-recipe/results').glob('*.iso'))
|
|
if len(isos) != 1:
|
|
raise ValueError('expected exactly one newly built ISO in fresh checkout')
|
|
iso = out / f'archipelago-UNSIGNED-UAT-{version}-{commit[:12]}-x86_64.iso'
|
|
shutil.copyfile(isos[0], iso)
|
|
built_hash = sha(iso)
|
|
run(['bash', 'scripts/iso-smoke-test.sh', iso, version], checkout, out / 'smoke.log', env)
|
|
extracted = out / 'extracted'
|
|
run(['xorriso', '-osirrox', 'on', '-indev', iso, '-extract', '/', extracted], checkout, out / 'extract.log', env)
|
|
compare_payload(extracted / 'archipelago', staged)
|
|
uat_oci.bundle(extracted / 'archipelago/unsigned-uat-oci', oci_hash)
|
|
spec = importlib.util.spec_from_file_location('kernel_security', checkout / 'scripts/kernel-security.py')
|
|
security = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(security)
|
|
rootfs_receipts = kernel_tar(extracted / 'archipelago/rootfs.tar', security, pin, qualified_payload)
|
|
live = out / 'live'
|
|
run(['unsquashfs', '-processors', '2', '-d', live, extracted / 'live/filesystem.squashfs'], checkout, out / 'squashfs.log', env)
|
|
kernels = []
|
|
for path in (live / 'var/lib/archipelago/kernel-security').glob('*.json'):
|
|
item = json.loads(path.read_text())
|
|
validate_kernel_receipt(item, security, pin, qualified_payload)
|
|
security.verify_receipt(item, item, live)
|
|
image = '/boot/vmlinuz-' + item['package'].removeprefix('linux-image-')
|
|
if item['files'][image] == sha(extracted / 'live/vmlinuz'):
|
|
kernels.append(item)
|
|
if len(kernels) != 1:
|
|
raise ValueError('live ISO boot kernel lacks unique fixed package provenance')
|
|
binding_path = live / 'var/lib/archipelago/uat-kernel/boot-binding.json'
|
|
binding = json.loads(binding_path.read_text())
|
|
if (binding.get('pin') != pin or binding.get('release') != uat_kernel.validate(pin)
|
|
or sha(extracted / 'live/vmlinuz') != pin['vmlinuz_sha256']
|
|
or sha(extracted / 'live/initrd.img') != binding['initrd_sha256']):
|
|
raise ValueError('ISO boot image/initrd differs from exact qualified kernel binding')
|
|
uat_kernel.check_initrd(extracted / 'live/initrd.img', binding['release'])
|
|
for mode in ('bios', 'uefi'):
|
|
work = out / mode
|
|
work.mkdir()
|
|
# No host forwarding needed; choose ephemeral ports to avoid other fixtures.
|
|
import socket
|
|
with socket.socket() as a, socket.socket() as b:
|
|
a.bind(('127.0.0.1', 0)); b.bind(('127.0.0.1', 0))
|
|
boot_env = dict(env, TMPDIR=str(work), QEMU_SSH_PORT=str(a.getsockname()[1]), QEMU_HTTP_PORT=str(b.getsockname()[1]))
|
|
command = ['bash', 'image-recipe/_archived/test-iso-qemu.sh', iso, '180', '--nographic']
|
|
if mode == 'bios':
|
|
command.append('--bios')
|
|
run(command, checkout, out / f'{mode}.log', boot_env)
|
|
boot_kernel(work / 'archipelago-qemu-serial.log', kernels[0])
|
|
compare_payload(extracted / 'archipelago', staged)
|
|
if source(root, commit, version) != source_tree or source(checkout, commit, version) != source_tree:
|
|
raise ValueError('source changed during build')
|
|
inputs(receipt)
|
|
for key in ('archive', 'payload_receipt', 'evidence'):
|
|
if sha(staged / ('kernel-' + key)) != receipt['kernel'][key]['sha256']:
|
|
raise ValueError('staged kernel input changed')
|
|
if uat_kernel.read(staged / 'kernel-pin.json') != pin:
|
|
raise ValueError('staged kernel pin changed')
|
|
for name in (*ARTIFACTS, 'evidence'):
|
|
if sha(staged / name) != receipt[name]['sha256']:
|
|
raise ValueError(f'staged input changed during build: {name}')
|
|
if tree(web) != qualified_web:
|
|
raise ValueError('staged web payload changed during build')
|
|
uat_oci.bundle(oci_bundle, oci_hash)
|
|
iso_hash = sha(iso)
|
|
if iso_hash != built_hash or sha(receipt_path) != receipt_hash:
|
|
raise ValueError('ISO or qualification changed during verification')
|
|
(out / (iso.name + '.sha256')).write_text(f'{iso_hash} {iso.name}\n')
|
|
result = dict(schema='archipelago-unsigned-uat-result-v2', status='unsigned-uat-artifact-checks-passed',
|
|
source_commit=commit, source_tree=source_tree, version=version,
|
|
qualification_sha256=receipt_hash, inputs=receipt, companion=companion,
|
|
oci_bundle_sha256=oci_hash, oci_import='deferred until admitted runtime storage is available',
|
|
iso={'name': iso.name, 'sha256': iso_hash, 'size': iso.stat().st_size},
|
|
live_kernel=kernels[0], rootfs_kernel_receipts=rootfs_receipts,
|
|
qualified_kernel=pin, live_boot_binding=binding,
|
|
logs={p.name: sha(p) for p in out.glob('*.log')},
|
|
serial_logs={mode: sha(out / mode / 'archipelago-qemu-serial.log') for mode in ('bios', 'uefi')},
|
|
pending=['Framework deployment/UAT', 'Yaya deployment/UAT', 'physical ISO install/UAT',
|
|
'explicit operator approval before signed OTA/ISO/Companion/publication'])
|
|
(out / 'UNSIGNED-UAT.json').write_text(json.dumps(result, indent=2, sort_keys=True) + '\n')
|
|
print(f'Unsigned UAT artifact checks passed: {iso}; physical install/operator acceptance pending')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--qualification', type=Path, required=True)
|
|
parser.add_argument('--output', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
build(Path(__file__).resolve().parents[1], args.qualification.resolve(), args.output.resolve())
|
|
except (ValueError, OSError, KeyError, TypeError, subprocess.SubprocessError, tarfile.TarError) as error:
|
|
parser.exit(1, f'FAIL (no UAT success receipt): {error}\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|