106 lines
5.7 KiB
Python
106 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate and stage explicitly reviewed ISO inputs; never discover installed artifacts."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
|
|
REQUIRED = {'archipelago', 'archy-reticulum-daemon', 'archy-rnodeconf', 'archy-esptool',
|
|
'nostr-rs-relay', 'fips.deb', 'tailscale.deb', 'web-ui.tar'}
|
|
SHA = re.compile(r'[0-9a-f]{64}')
|
|
REF = re.compile(r'[0-9a-f]{40}')
|
|
|
|
def digest(path):
|
|
h = hashlib.sha256()
|
|
with path.open('rb') as stream:
|
|
for block in iter(lambda: stream.read(1024 * 1024), b''): h.update(block)
|
|
return h.hexdigest()
|
|
|
|
def validate(manifest, repo, source_ref, architecture):
|
|
data = json.loads(manifest.read_text())
|
|
if data.get('schema') != 1 or not REF.fullmatch(source_ref):
|
|
raise ValueError('schema 1 and full source commit required')
|
|
if data.get('source_ref') != source_ref or data.get('architecture') != architecture:
|
|
raise ValueError('input source/architecture differs from requested build')
|
|
actual = subprocess.check_output(['git', '-C', str(repo), 'rev-parse', 'HEAD'], text=True).strip()
|
|
if actual != source_ref or subprocess.check_output(['git', '-C', str(repo), 'status', '--porcelain', '--untracked-files=no']):
|
|
raise ValueError('build requires the exact clean reviewed source checkout')
|
|
if type(data.get('source_date_epoch')) is not int or data['source_date_epoch'] < 1:
|
|
raise ValueError('explicit source_date_epoch required')
|
|
if type(data.get('build_number')) is not int or data['build_number'] < 1:
|
|
raise ValueError('explicit build_number required')
|
|
artifacts = data.get('artifacts', {})
|
|
if set(artifacts) != REQUIRED:
|
|
raise ValueError('all required artifact inputs must be selected explicitly')
|
|
containers = data.get('container_archives', {})
|
|
if not isinstance(containers, dict) or not containers:
|
|
raise ValueError('explicit offline container archives required')
|
|
if set(containers) & set(artifacts):
|
|
raise ValueError('container and artifact names must be distinct')
|
|
for name in containers:
|
|
if not re.fullmatch(r'[a-z0-9][a-z0-9-]*\.tar(?:\.zst)?', name):
|
|
raise ValueError('invalid container archive name')
|
|
if not any(n in containers for n in ('filebrowser.tar', 'filebrowser.tar.zst')) or not any(n in containers for n in ('fmcd.tar', 'fmcd.tar.zst')):
|
|
raise ValueError('offline baseline requires filebrowser and fmcd archives')
|
|
verified = {}
|
|
for name, entry in {**artifacts, **containers}.items():
|
|
if not isinstance(entry, dict) or not SHA.fullmatch(entry.get('sha256', '')) or not entry.get('provenance'):
|
|
raise ValueError('artifact hash and reviewable provenance required: ' + name)
|
|
path = Path(entry['path'])
|
|
if not path.is_absolute(): path = manifest.parent / path
|
|
if path.is_symlink() or not path.is_file() or digest(path) != entry['sha256']:
|
|
raise ValueError('artifact hash mismatch or unsafe file: ' + name)
|
|
verified[name] = path
|
|
return data, verified
|
|
|
|
def unpack_web(archive, dest):
|
|
# Explicitly avoid links, device entries, traversal and tar metadata surprises.
|
|
with tarfile.open(archive, 'r:*') as tar:
|
|
members = tar.getmembers()
|
|
seen = set()
|
|
for member in members:
|
|
name = PurePosixPath(member.name)
|
|
if str(name) == '.' and member.isdir():
|
|
continue
|
|
if (name.is_absolute() or '..' in name.parts or not name.parts
|
|
or not (member.isfile() or member.isdir()) or str(name) in seen):
|
|
raise ValueError('unsafe web archive member: ' + member.name)
|
|
seen.add(str(name))
|
|
required = {'index.html', 'packages/archipelago-companion.apk', 'packages/archipelago-companion.json',
|
|
'archipelago-runtime/scripts/c01_network_owner.py'}
|
|
if not required.issubset(seen) or 'aiui/index.html' in seen:
|
|
raise ValueError('qualified native dashboard required; standalone AIUI is forbidden')
|
|
tar.extractall(dest, members=members, filter='data')
|
|
|
|
def stage(manifest, repo, source_ref, architecture, dest):
|
|
data, verified = validate(manifest, repo, source_ref, architecture)
|
|
dest.mkdir(mode=0o700) # a reused or partially populated staging area is refused
|
|
for name, path in verified.items():
|
|
target = dest / ('container-images' if name in data['container_archives'] else 'artifacts') / name
|
|
target.parent.mkdir(exist_ok=True)
|
|
shutil.copyfile(path, target)
|
|
# Bind the actual staged bytes too, not just the earlier source read.
|
|
entry = data['container_archives'].get(name, data['artifacts'].get(name))
|
|
if digest(target) != entry['sha256']: raise ValueError('input changed during copy: ' + name)
|
|
unpack_web(dest/'artifacts/web-ui.tar', dest/'web-ui')
|
|
subprocess.run(['git', '-C', str(repo), 'archive', '--format=tar', '-o', str(dest/'source.tar'), source_ref], check=True)
|
|
(dest/'inputs.json').write_text(json.dumps(data, indent=2)+'\n')
|
|
(dest/'build-number').write_text(str(data['build_number']))
|
|
(dest/'source-date-epoch').write_text(str(data['source_date_epoch']))
|
|
return data
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--manifest', type=Path, required=True)
|
|
parser.add_argument('--repo', type=Path, required=True)
|
|
parser.add_argument('--source-ref', required=True)
|
|
parser.add_argument('--architecture', choices=['amd64', 'arm64'], required=True)
|
|
parser.add_argument('--stage', type=Path, required=True)
|
|
a = parser.parse_args()
|
|
stage(a.manifest.resolve(), a.repo.resolve(), a.source_ref, a.architecture, a.stage.resolve())
|