tegwick
e0359bcb6c
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
125 lines
8.2 KiB
Python
125 lines
8.2 KiB
Python
"""Read a public policy image without executing it or retaining registry tokens."""
|
|
import argparse
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import tarfile
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
ORIGIN = 'https://forgejo.coulomb.social'
|
|
BASE = ORIGIN + '/v2/coulomb/informed-decision/'
|
|
LIMIT = 300 * 1024 * 1024
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, request, response, code, message, headers, new_url):
|
|
raise ValueError('registry redirects refused')
|
|
OPENER = urllib.request.build_opener(NoRedirect())
|
|
|
|
|
|
def fetch(url, accept=None):
|
|
parsed_url = urllib.parse.urlparse(url)
|
|
if parsed_url.scheme != 'https' or parsed_url.netloc != 'forgejo.coulomb.social':
|
|
raise ValueError('unexpected registry origin')
|
|
headers = {'Accept': accept} if accept else {}
|
|
try:
|
|
response = OPENER.open(urllib.request.Request(url, headers=headers), timeout=30)
|
|
except urllib.error.HTTPError as error:
|
|
if error.code != 401:
|
|
raise
|
|
fields = dict(re.findall(r'(\w+)="([^"]*)"', error.headers.get('WWW-Authenticate', '')))
|
|
realm = fields['realm']
|
|
parsed = urllib.parse.urlparse(realm)
|
|
if parsed.scheme != 'https' or parsed.netloc != 'forgejo.coulomb.social':
|
|
raise ValueError('unexpected registry challenge origin')
|
|
query = urllib.parse.urlencode({k: fields[k] for k in ('service', 'scope') if k in fields})
|
|
with OPENER.open(realm + '?' + query, timeout=30) as auth:
|
|
token = json.load(auth)['token']
|
|
headers['Authorization'] = 'Bearer ' + token
|
|
response = OPENER.open(urllib.request.Request(url, headers=headers), timeout=60)
|
|
with response:
|
|
body = response.read(LIMIT + 1)
|
|
if len(body) > LIMIT:
|
|
raise ValueError('registry artifact exceeds local verification bound')
|
|
return body
|
|
|
|
|
|
def sha(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def blob(descriptor):
|
|
digest = descriptor['digest']
|
|
if not re.fullmatch(r'sha256:[a-f0-9]{64}', digest):
|
|
raise ValueError('unknown image digest')
|
|
data = fetch(BASE + 'blobs/' + digest)
|
|
if 'sha256:' + sha(data) != digest or len(data) != descriptor['size']:
|
|
raise ValueError('image blob hash or size mismatch')
|
|
return data
|
|
|
|
|
|
def manifest(digest):
|
|
raw = fetch(BASE + 'manifests/' + digest, 'application/vnd.oci.image.index.v1+json, application/vnd.docker.distribution.manifest.list.v2+json, application/vnd.oci.image.manifest.v1+json, application/vnd.docker.distribution.manifest.v2+json')
|
|
if 'sha256:' + sha(raw) != digest:
|
|
raise ValueError('manifest digest mismatch')
|
|
value = json.loads(raw)
|
|
if 'manifests' in value:
|
|
selected = [d for d in value['manifests'] if d.get('platform',{}).get('os') == 'linux' and d.get('platform',{}).get('architecture') == 'amd64']
|
|
if len(selected) != 1:
|
|
raise ValueError('one linux/amd64 platform required')
|
|
child, config, layerlist = manifest(selected[0]['digest'])
|
|
return child, config, layerlist
|
|
return value, json.loads(blob(value['config'])), value['layers']
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(); p.add_argument('--digest',required=True);p.add_argument('--revision',required=True);p.add_argument('--output',type=Path,required=True);a=p.parse_args()
|
|
if not re.fullmatch(r'sha256:[a-f0-9]{64}',a.digest) or not re.fullmatch(r'[a-f0-9]{40}',a.revision):
|
|
raise ValueError('exact immutable revision and image required')
|
|
base_digest='sha256:7f5c6cb29c9f70a026bfe30e72934274ddc69261676cf6d476f81dd32812d173'
|
|
m,c,l=manifest(a.digest);bm,bc,bl=manifest(base_digest)
|
|
allowed_media={'application/vnd.docker.image.rootfs.diff.tar.gzip','application/vnd.oci.image.layer.v1.tar+gzip'}
|
|
if any(d['mediaType'] not in allowed_media for d in l+bl): raise ValueError('unsupported layer encoding')
|
|
if [(d['digest'],d['size']) for d in l[:-1]] != [(d['digest'],d['size']) for d in bl] or len(l) != len(bl)+1:
|
|
raise ValueError('candidate must retain all base descriptors plus exactly one layer')
|
|
for key in ('architecture','os'):
|
|
if c[key] != bc[key]: raise ValueError('platform changed')
|
|
if c['rootfs']['diff_ids'][:-1] != bc['rootfs']['diff_ids'] or len(c['rootfs']['diff_ids']) != len(bc['rootfs']['diff_ids'])+1:
|
|
raise ValueError('base rootfs changed')
|
|
defaults={'Hostname':'','Domainname':'','AttachStdin':False,'AttachStdout':False,'AttachStderr':False,'Tty':False,'OpenStdin':False,'StdinOnce':False,'Cmd':None,'Volumes':None,'OnBuild':None}
|
|
extra_defaults={}
|
|
for key,value in defaults.items():
|
|
if key not in bc['config'] and key in c['config']:
|
|
if c['config'][key] != value: raise ValueError('legacy builder runtime default changed')
|
|
extra_defaults[key]=value
|
|
if 'Image' in c['config'] and 'Image' not in bc['config']:
|
|
if not re.fullmatch(r'sha256:[a-f0-9]{64}',c['config']['Image']):raise ValueError('unexpected legacy builder parent-image metadata')
|
|
extra_defaults['Image']=c['config']['Image']
|
|
actual_config={k:v for k,v in c['config'].items() if k != 'Labels' and k not in extra_defaults}
|
|
base_config={k:v for k,v in bc['config'].items() if k != 'Labels'}
|
|
if actual_config != base_config: raise ValueError('runtime image config changed')
|
|
labels=c['config']['Labels']; expected={'org.opencontainers.image.revision':a.revision,'org.opencontainers.image.base.digest':base_digest,'social.coulomb.source.policy.sha256':'9d7c1be4cb6199d14ef0aaead5f85951f45b0ccbd7c800a7fde922a253566b62','social.coulomb.release.scope':'INFD-WP-0009-T04/purpose-overlay-v1'}
|
|
for k,v in expected.items():
|
|
if labels.get(k)!=v:raise ValueError('exact source/scope label mismatch')
|
|
if {k:v for k,v in labels.items() if k not in expected} != {k:v for k,v in bc['config'].get('Labels',{}).items() if k not in expected}:
|
|
raise ValueError('unrelated labels changed')
|
|
files=[]; dirs=[];content=None
|
|
with tarfile.open(fileobj=io.BytesIO(blob(l[-1])),mode='r:*') as archive:
|
|
for item in archive:
|
|
path=PurePosixPath(item.name.removeprefix('./'))
|
|
if path.is_absolute() or '..' in path.parts or any(p.startswith('.wh.') for p in path.parts):raise ValueError('unsafe overlay entry')
|
|
if item.isdir():
|
|
allowed_dirs={'opt','opt/venv','opt/venv/lib','opt/venv/lib/python3.12','opt/venv/lib/python3.12/site-packages','opt/venv/lib/python3.12/site-packages/informed_decision'}
|
|
if str(path) not in allowed_dirs: raise ValueError('unexpected overlay directory')
|
|
dirs.append(str(path));continue
|
|
if not item.isfile():raise ValueError('unexpected overlay link or device')
|
|
files.append(str(path))
|
|
if str(path)!='opt/venv/lib/python3.12/site-packages/informed_decision/policy.py' or item.mode!=0o644 or item.uid!=0 or item.gid!=0 or item.size>100000:raise ValueError('unexpected overlay file metadata')
|
|
content=archive.extractfile(item).read()
|
|
if len(files)!=1 or sha(content)!=expected['social.coulomb.source.policy.sha256']:raise ValueError('single installed source file not exact')
|
|
a.output.mkdir(parents=True,exist_ok=True)
|
|
(a.output/'image-manifest.json').write_text(json.dumps(m,indent=2)+'\n');(a.output/'image-config.json').write_text(json.dumps(c,indent=2)+'\n');(a.output/'base-image-config.json').write_text(json.dumps(bc,indent=2)+'\n')
|
|
receipt={'image_digest':a.digest,'source_revision':a.revision,'base_image_digest':base_digest,'base_layer_bytes_identical':True,'layer_media_type_conversion':'OCI gzip to Docker gzip, same digest and size','base_runtime_config_semantics_identical':True,'legacy_builder_config_additions':extra_defaults,'base_rootfs_prefix_identical':True,'overlay_regular_files':files,'overlay_directories':dirs,'policy_sha256':sha(content),'policy_mode':'0644','base_user':bc['config'].get('User'),'source_scope_labels':expected,'registry_read_route':'anonymous public scoped bearer challenge, no token retention','image_executed':False,'native_workload_changed':False,'native_backup_restore_executed':False}
|
|
(a.output/'registry-proof.json').write_text(json.dumps(receipt,indent=2)+'\n');print(json.dumps(receipt,sort_keys=True))
|
|
if __name__=='__main__': main()
|