#!/usr/bin/env python3 """Proof Orchard: bounded source/wheel license-byte inventory, no extraction/execution. Python stdlib only. Usage: python3 tools/package-license-audit.py > ledger.json --test runs offline controls; --ledger ledger.json rechecks its pinned artifacts. All remote content is data, never imported/executed. """ import datetime import email.parser import hashlib import io import json import pathlib import sys import tarfile import urllib.parse import urllib.request import zipfile PACKAGES = ('httpx', 'attrs', 'packaging', 'platformdirs') PRIMARY = {'license', 'license.txt', 'license.md', 'licence', 'licence.txt', 'licence.md', 'copying', 'copying.txt'} MAX_TOTAL = 20_000_000 MAX_MEMBER = 2_000_000 MAX_EXPANDED = 100_000_000 spent = 0 def sha(data): return hashlib.sha256(data).hexdigest() class RestrictedRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ValueError('Redirect rejected: inspect before widening network scope') def fetch(url, limit): global spent parsed = urllib.parse.urlparse(url) if parsed.scheme != 'https' or parsed.hostname not in {'pypi.org', 'files.pythonhosted.org'} or parsed.username or parsed.port: raise ValueError('URL outside fixed registry/file hosts') req = urllib.request.Request(url, headers={'User-Agent': 'ProofOrchard-AI-source-audit/0.1'}) with urllib.request.build_opener(RestrictedRedirect).open(req, timeout=30) as response: data = response.read(min(limit, MAX_TOTAL - spent) + 1) if len(data) > limit or spent + len(data) > MAX_TOTAL: raise ValueError('Download cap exceeded') spent += len(data) return data def primary(path): parts = pathlib.PurePosixPath(path).parts return len(parts) == 2 and parts[-1].lower() in PRIMARY def wheel_license(path): parts = pathlib.PurePosixPath(path).parts return (len(parts) >= 2 and parts[0].endswith('.dist-info') and (parts[1] == 'licenses' or (len(parts) == 2 and parts[1].lower() in PRIMARY))) def describe(path, data): return {'path': path, 'bytes': len(data), 'sha256': sha(data)} def compare(source, wheel): return [{**describe(p, b), 'identical_wheel_paths': [q for q, c in wheel.items() if c == b]} for p, b in source.items()] def members(data, kind): # Bound declared expanded sizes and member count, never write archive paths to disk. selected = {} if kind == 'sdist': with tarfile.open(fileobj=io.BytesIO(data), mode='r:gz') as archive: total = 0 for i, member in enumerate(archive): total += member.size if i >= 10000 or total > MAX_EXPANDED: raise ValueError('Archive expansion/member cap') if member.isfile() and primary(member.name): if member.size > MAX_MEMBER or member.name in selected: raise ValueError('Oversize/duplicate selected member') selected[member.name] = archive.extractfile(member).read(MAX_MEMBER + 1) else: with zipfile.ZipFile(io.BytesIO(data)) as archive: infos = archive.infolist() if len(infos) > 10000 or sum(i.file_size for i in infos) > MAX_EXPANDED: raise ValueError('Archive expansion/member cap') for member in infos: if not member.is_dir() and (wheel_license(member.filename) or member.filename.endswith('.dist-info/METADATA')): if member.file_size > MAX_MEMBER or member.filename in selected: raise ValueError('Oversize/duplicate selected member') selected[member.filename] = archive.read(member) return selected def audit(name, pinned=None): if pinned is None: registry = fetch(f'https://pypi.org/pypi/{name}/json', 2_000_000) data = json.loads(registry) files = data['urls'] sources = sorted([f for f in files if f['packagetype'] == 'sdist' and f['filename'].endswith('.tar.gz') and not f.get('yanked')], key=lambda f: f['filename']) wheels = sorted([f for f in files if f['filename'].endswith('-py3-none-any.whl') and not f.get('yanked')], key=lambda f: f['filename']) if not sources or not wheels: raise ValueError(f'{name}: expected source and universal wheel unavailable') version, registry_hash = data['info']['version'], sha(registry) else: sources, wheels = [[f for f in pinned['artifacts'] if f['filename'].endswith(ext)] for ext in ('.tar.gz', '.whl')] if len(sources) != 1 or len(wheels) != 1: raise ValueError('Pinned ledger requires exactly one source and wheel') version, registry_hash = pinned['version'], pinned['registry_response_sha256'] record = {'project': name, 'version': version, 'registry_response_sha256': registry_hash, 'artifacts': []} loaded = {} for kind, artifact in [('sdist', sources[0]), ('wheel', wheels[0])]: if artifact['size'] > MAX_TOTAL - spent: raise ValueError('Remaining download budget insufficient') body = fetch(artifact['url'], artifact['size']) if len(body) != artifact['size'] or sha(body) != artifact['digests']['sha256']: raise ValueError('Registry size/digest mismatch') record['artifacts'].append({k: artifact[k] for k in ('filename', 'url', 'size', 'upload_time_iso_8601', 'digests')}) loaded[kind] = members(body, kind) metadata = {p: b for p, b in loaded['wheel'].items() if p.endswith('.dist-info/METADATA')} licenses = {p: b for p, b in loaded['wheel'].items() if wheel_license(p)} record['source_primary_licenses'] = compare(loaded['sdist'], licenses) record['wheel_license_files'] = [describe(p, b) for p, b in licenses.items()] record['wheel_metadata'] = [] for path, raw in metadata.items(): message = email.parser.BytesParser().parsebytes(raw) record['wheel_metadata'].append({**describe(path, raw), 'fields': {key: message.get_all(key, []) for key in ('Metadata-Version', 'Name', 'Version', 'License', 'License-Expression', 'License-File')}}) record['status'] = ('all-primary-texts-identical' if record['source_primary_licenses'] and all(f['identical_wheel_paths'] for f in record['source_primary_licenses']) else 'manual-review-required') return record def test(): a = {'pkg/LICENSE': b'license\n'} assert compare(a, {'pkg.dist-info/licenses/LICENSE': b'license\n'})[0]['identical_wheel_paths'] assert not compare(a, {'pkg.dist-info/LICENSE': b'license'})[0]['identical_wheel_paths'] assert not compare(a, {})[0]['identical_wheel_paths'] assert primary('pkg/LICENSE') and not primary('pkg/vendor/LICENSE') assert wheel_license('pkg.dist-info/licenses/LICENSE') and not wheel_license('pkg/LICENSE') buf = io.BytesIO() with tarfile.open(fileobj=buf, mode='w:gz') as archive: for path in ('pkg/LICENSE', '../../LICENSE', 'pkg/vendor/LICENSE'): member = tarfile.TarInfo(path); member.size = 4 archive.addfile(member, io.BytesIO(b'test')) assert members(buf.getvalue(), 'sdist') == {'pkg/LICENSE': b'test'} buf = io.BytesIO() with zipfile.ZipFile(buf, 'w') as archive: archive.writestr('pkg.dist-info/licenses/LICENSE', b'test') archive.writestr('../../LICENSE', b'ignored') assert members(buf.getvalue(), 'wheel') == {'pkg.dist-info/licenses/LICENSE': b'test'} print('7 offline controls passed; archives never extracted') if __name__ == '__main__': if sys.argv[1:] == ['--test']: test() else: results = [] pinned = {} if len(sys.argv) == 3 and sys.argv[1] == '--ledger': pinned = {r['project']: r for r in json.loads(pathlib.Path(sys.argv[2]).read_text())['results']} elif sys.argv[1:]: raise SystemExit('Usage: package-license-audit.py [--test | --ledger ledger.json]') for name in PACKAGES: try: if pinned and name not in pinned: raise ValueError('Missing fixed-sample record in pinned ledger') results.append(audit(name, pinned.get(name))) except Exception as error: results.append({'project': name, 'status': 'coverage-failed', 'error': str(error)}) print(json.dumps({'observed_at_utc': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'agent': 'Proof Orchard (AI)', 'scope': 'primary sdist license text byte inclusion in one universal wheel; not legal compliance', 'download_bytes_including_metadata': spent, 'results': results}, indent=2))