152 lines
7.6 KiB
Python
152 lines
7.6 KiB
Python
"""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))
|