"""Strict local OCI candidate verification. Never invokes a container runtime.""" import gzip import hashlib import json from pathlib import Path import re import tarfile REPOSITORY = 'source.archipelago-foundation.org/chaum/lnd-tor' NATIVE = 'core/archipelago/src/container/lnd_dependency.rs' SOURCES = {'apps/lnd-tor/Dockerfile', 'apps/lnd-tor/torrc', 'apps/lnd-tor/manifest.yml', 'apps/lnd-tor/test_namespace.py', 'apps/lnd/Dockerfile', 'apps/lnd/manifest.yml'} 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 exact(pairs): result = {} for key, value in pairs: if key in result: raise ValueError('duplicate JSON key') result[key] = value return result def decode(raw): return json.loads(raw, object_pairs_hook=exact) def check(root, archive, identity_path, sources_path): identity = decode(identity_path.read_bytes()) sources = decode(sources_path.read_bytes()) if (sha(sources_path) != identity['source_receipt_sha256'] or sources != identity['sources'] or set(sources) != SOURCES): raise ValueError('source receipt mismatch') for name, expected in sources.items(): if sha(root / name) != expected: raise ValueError(f'stale source: {name}') if sha(archive) != identity['archive_sha256'] or archive.stat().st_size != identity['archive_bytes']: raise ValueError('OCI archive hash/size mismatch') reference = REPOSITORY + '@' + identity['oci_manifest_digest'] manifest_source = (root / 'apps/lnd-tor/manifest.yml').read_text() if re.findall(r'^\s+image:\s*(\S+)\s*$', manifest_source, re.M) != [reference]: raise ValueError('source image pin is not exact imported digest') native = (root / NATIVE).read_text() qualified = re.search(r'const QUALIFIED_IMAGES:.*?=\s*&\[(.*?)\];', native, re.S) if not qualified or reference not in re.findall(r'"([^"]+)"', qualified[1]): raise ValueError('imported digest absent from native admission source') with tarfile.open(archive) as tar: members = {} blobs = {} for member in tar: name = member.name.removeprefix('./').rstrip('/') if (not name or name.startswith('/') or any(p in ('', '.', '..') for p in name.split('/')) or name in members or not (member.isfile() or member.isdir())): raise ValueError('unsafe/duplicate OCI member') members[name] = member if member.isfile() and name.startswith('blobs/'): if not re.fullmatch(r'blobs/sha256/[0-9a-f]{64}', name): raise ValueError('unsupported OCI blob name') with tar.extractfile(member) as stream: actual = hashlib.file_digest(stream, 'sha256').hexdigest() if actual != name.rsplit('/', 1)[1]: raise ValueError('OCI blob hash mismatch') blobs['sha256:' + actual] = member elif member.isfile() and name not in ('index.json', 'oci-layout'): raise ValueError('unexpected OCI archive file') def document(name): member = members[name] if not member.isfile() or member.size > 4 * 1024 * 1024: raise ValueError('OCI document type/size') return decode(tar.extractfile(member).read()) if document('oci-layout') != {'imageLayoutVersion': '1.0.0'}: raise ValueError('OCI layout version') index = document('index.json') if index.get('schemaVersion') != 2 or len(index['manifests']) != 1: raise ValueError('one exact OCI platform manifest required') used = set() def descriptor(value, media): digest = value['digest'] if value['mediaType'] != media or digest not in blobs or blobs[digest].size != value['size']: raise ValueError('OCI descriptor type/digest/size mismatch') used.add(digest) return blobs[digest] desc = index['manifests'][0] descriptor(desc, 'application/vnd.oci.image.manifest.v1+json') if desc['digest'] != identity['oci_manifest_digest']: raise ValueError('stale builder/import manifest digest') if 'platform' in desc and (desc['platform'].get('os'), desc['platform'].get('architecture')) != ('linux', 'amd64'): raise ValueError('OCI descriptor platform mismatch') manifest = document(blobs[desc['digest']].name.removeprefix('./')) if manifest.get('schemaVersion') != 2: raise ValueError('OCI manifest schema') conf = manifest['config'] descriptor(conf, 'application/vnd.oci.image.config.v1+json') if conf['digest'] != identity['config_digest']: raise ValueError('OCI config identity mismatch') config = document(blobs[conf['digest']].name.removeprefix('./')) if (config.get('os'), config.get('architecture'), config['config'].get('User')) != ('linux', 'amd64', '100:100'): raise ValueError('OCI platform/default user mismatch') diffids = config['rootfs']['diff_ids'] if config['rootfs']['type'] != 'layers' or len(diffids) != len(manifest['layers']): raise ValueError('OCI layer/diffID count mismatch') for layer, diffid in zip(manifest['layers'], diffids): member = descriptor(layer, 'application/vnd.oci.image.layer.v1.tar+gzip') with tar.extractfile(member) as raw, gzip.GzipFile(fileobj=raw) as stream: if 'sha256:' + hashlib.file_digest(stream, 'sha256').hexdigest() != diffid: raise ValueError('OCI uncompressed layer diffID mismatch') if used != set(blobs) or sorted(d.removeprefix('sha256:') for d in blobs) != sorted(identity['verified_blobs']): raise ValueError('unreferenced/missing OCI blob') return dict(reference=reference, manifest_digest=desc['digest'], config_id=conf['digest'], archive_sha256=identity['archive_sha256'], platform='linux/amd64', user='100:100', source_receipt_sha256=sha(sources_path), verified_blobs=len(blobs), diff_ids=diffids) def inventory(root): result = {} for path in sorted(root.rglob('*')): if path.is_symlink(): raise ValueError('bundle symlink') if not path.is_dir() and path != root / 'BUNDLE.json': result[path.relative_to(root).as_posix()] = sha(path) return result def bundle(root, expected=None): if not root.is_dir() or root.is_symlink(): raise ValueError('bundle directory missing or symlinked') if expected is not None and sha(root / 'BUNDLE.json') != expected: raise ValueError('candidate bundle receipt hash mismatch') receipt = decode((root / 'BUNDLE.json').read_bytes()) if receipt.get('schema') != 'archipelago-unsigned-uat-oci-v1' or receipt['files'] != inventory(root): raise ValueError('candidate bundle files mismatch') proof = check(root / 'source', root / 'lnd-tor.oci.tar', root / 'identity.json', root / 'final-source.json') if proof != receipt['oci']: raise ValueError('candidate OCI proof mismatch') return receipt if __name__ == '__main__': import argparse parser = argparse.ArgumentParser(description=__doc__) for name in ('root', 'archive', 'identity', 'sources'): parser.add_argument('--' + name, type=Path, required=True) args = parser.parse_args() print(json.dumps(check(args.root, args.archive, args.identity, args.sources), sort_keys=True))