124 lines
9.3 KiB
Python
124 lines
9.3 KiB
Python
"""Build a NEW portable evidence directory; originals are read-only. No publication."""
|
|
from pathlib import Path
|
|
import argparse, hashlib, json, subprocess
|
|
|
|
def digest(data): return hashlib.sha256(data).hexdigest()
|
|
def read(p): return json.loads(Path(p).read_text(encoding='utf-8-sig'))
|
|
def write(p,v):
|
|
p.parent.mkdir(parents=True,exist_ok=True)
|
|
with p.open('x',encoding='utf-8',newline='\n') as f: json.dump(v,f,indent=2,ensure_ascii=False); f.write('\n')
|
|
def git_blobs(repo,pin,paths):
|
|
proc=subprocess.run(['git','-c','core.longpaths=true','cat-file','--batch'],cwd=repo,input=''.join(pin+':'+p+'\n' for p in paths).encode(),stdout=subprocess.PIPE,check=True)
|
|
data=proc.stdout; cursor=0
|
|
for path in paths:
|
|
end=data.index(b'\n',cursor); header=data[cursor:end].split(); assert len(header)==3 and header[1]==b'blob',path
|
|
size=int(header[2]); cursor=end+1; blob=data[cursor:cursor+size]; cursor+=size+1
|
|
yield path,blob
|
|
assert cursor==len(data)
|
|
def records(value):
|
|
if isinstance(value,dict):
|
|
if all(k in value for k in ('path','bytes','sha256')): yield value
|
|
for v in value.values(): yield from records(v)
|
|
elif isinstance(value,list):
|
|
for v in value: yield from records(v)
|
|
def build(a):
|
|
native=Path(a.native).resolve(); manifest=read(native/'input-manifest.json'); repo=Path(a.repo).resolve(); out=Path(a.output).resolve()
|
|
if out.exists(): raise RuntimeError('Output must be new; never append into a frozen archive')
|
|
out.mkdir(parents=True); inventory=[]; sources=[]
|
|
def copy(src,rel):
|
|
src=Path(src).resolve(); data=src.read_bytes(); dest=out/rel; dest.parent.mkdir(parents=True,exist_ok=True)
|
|
with dest.open('xb') as f:f.write(data)
|
|
assert dest.read_bytes()==src.read_bytes(),str(src)
|
|
inventory.append(dict(original=str(src),archive=rel,bytes=len(data),sha256=digest(data)))
|
|
def tree(folder,prefix):
|
|
for p in sorted(folder.rglob('*')):
|
|
relative=p.relative_to(folder)
|
|
if not p.is_file() or p.suffix.lower() in ('.exe','.dll','.pdb') or any(s in ('profile','working','package','unrelated-cwd','__pycache__') for s in relative.parts):continue
|
|
copy(p,prefix+'/'+relative.as_posix())
|
|
gates=[Path(g).resolve() for g in a.gate]
|
|
for n,gate in enumerate(gates):
|
|
if not (gate/'result.json').is_file(): raise RuntimeError('Closed gate result required: '+str(gate))
|
|
tree(gate,f'gates/{n:02d}-{gate.name}')
|
|
result=read(gate/'result.json'); pin=result['source_commit']; context=read(gate/'context.json')
|
|
runner=result.get('runner')
|
|
if runner:
|
|
raw=Path(runner['path']).read_bytes()
|
|
assert len(raw)==runner['bytes'] and digest(raw)==runner['sha256'], 'Gate runner changed'
|
|
copy(Path(runner['path']),f'gates/{n:02d}-{gate.name}/bound-runner.py')
|
|
# Failed gates remain failures and get their own pin/input mapping.
|
|
for filename in ('inputs-before.json','inputs-after.json'):
|
|
p=gate/filename
|
|
if not p.exists():continue
|
|
items=read(p); paths=[r['path'] for r in items]
|
|
blobs=dict(git_blobs(repo,pin,paths))
|
|
for item in items:
|
|
key=(pin,item['path'],item['sha256'])
|
|
if any((r['commit'],r['git_path'],r['sha256'])==key for r in sources):continue
|
|
blob=blobs[item['path']]; row={**item,'git_path':item['path'],'commit':pin,'original':str(Path(context['cwd'])/item['path'])}; row.pop('path')
|
|
if len(blob)==item['bytes'] and digest(blob)==item['sha256']:row['storage']='git'
|
|
else:
|
|
raw=Path(row['original']); data=raw.read_bytes() if raw.is_file() else b''
|
|
if len(data)!=item['bytes'] or digest(data)!=item['sha256']:
|
|
data=blob.replace(b'\r\n',b'\n').replace(b'\n',b'\r\n')
|
|
if len(data)!=item['bytes'] or digest(data)!=item['sha256']:raise RuntimeError('Cannot recover exact captured source: '+str(key))
|
|
rel='source-overrides/'+pin+'/'+item['path']; dest=out/rel; dest.parent.mkdir(parents=True,exist_ok=True)
|
|
with dest.open('xb') as f:f.write(data)
|
|
row.update(storage='override',archive=rel)
|
|
sources.append(row)
|
|
for p in sorted(native.iterdir()):
|
|
if p.is_file() and p.suffix in ('.py','.ps1','.js','.md','.json'):copy(p,'native/'+p.name)
|
|
for folder in ('baseline','expected','staging-failed-01','supplemental-reconnect'):
|
|
if (native/folder).is_dir():tree(native/folder,'native/'+folder)
|
|
for number in a.run:
|
|
run=native/'runs'/f'run-{number:02d}'
|
|
if not (run/'normal-close.json').is_file():raise RuntimeError('Refuse active/unclosed run: '+str(run))
|
|
close=read(run/'normal-close.json'); launch=read(run/'launch.json')
|
|
assert close['launch_sha256']==digest((run/'launch.json').read_bytes()) and close['pid']==launch['pid'] and close['started_utc']==launch['started_utc']
|
|
tree(run,'native/runs/'+run.name) # includes failed comparison receipts, without editing/reclassifying them
|
|
for n,p in enumerate(a.extra):copy(Path(p),f'reviews/{n:02d}-'+Path(p).name)
|
|
copy(Path(__file__),'verify-build.py')
|
|
write(out/'source-map.json',sources)
|
|
write(out/'archive-manifest.json',{'original_receipts_unchanged':True,'files':inventory,'source_map':'source-map.json','source_map_sha256':digest((out/'source-map.json').read_bytes()),'excluded':'Binaries, private profiles, mutable working directory and active runs; their original receipts/hashes remain for provenance.'})
|
|
(out/'README.md').write_text('# Edge attachment evidence\n\nPrepared immutable evidence, not blanket issue acceptance. Original receipts retain absolute machine paths; archive-manifest.json maps copied originals and source-map.json maps captured source to exact Git blobs or raw byte overrides. Failed gate/staging/comparison evidence remains failed. Only explicitly selected closed runs are included.\n\nRun `python verify-build.py verify --archive . --repo /path/to/repository` with every listed source commit fetched. This verifies archived bytes, exact captured source bytes, and inventories receipt references; unavailable excluded runtime binaries remain explicitly external, not falsely verified. Native captures prove only recorded actions/results; typed comparisons prove complete saved Projects, not held-motion or camera behavior. Consult native/native-plan.md and independent reviews for scope and limitations.\n',encoding='utf-8')
|
|
(out/'.gitattributes').write_text('* -text\n',encoding='ascii')
|
|
print(json.dumps({'archive':str(out),'copied_originals':len(inventory),'source_records':len(sources),'git_exact':sum(r['storage']=='git' for r in sources),'overrides':sum(r['storage']=='override' for r in sources)},indent=2))
|
|
def verify(a):
|
|
root=Path(a.archive).resolve(); m=read(root/'archive-manifest.json'); available=set(); checked=0
|
|
def check(data,r):
|
|
assert len(data)==r['bytes'] and digest(data)==r['sha256'],r
|
|
available.add((r['bytes'],r['sha256']))
|
|
for r in m['files']:check((root/r['archive']).read_bytes(),r);checked+=1
|
|
assert digest((root/'source-map.json').read_bytes())==m['source_map_sha256']
|
|
sources=read(root/'source-map.json')
|
|
for pin in sorted({r['commit'] for r in sources}):
|
|
rows=[r for r in sources if r['commit']==pin and r['storage']=='git']; blobs=dict(git_blobs(a.repo,pin,[r['git_path'] for r in rows]))
|
|
for r in rows:check(blobs[r['git_path']],r)
|
|
for r in sources:
|
|
if r['storage']=='override':check((root/r['archive']).read_bytes(),r)
|
|
for dependency in m.get('prerequisite_archives',[]):
|
|
matches=[Path(p).resolve() for p in getattr(a,'related_archive',[]) if digest((Path(p)/'archive-manifest.json').read_bytes())==dependency['manifest_sha256']]
|
|
if len(matches)!=1:raise RuntimeError('Provide exact prerequisite with --related-archive: '+dependency['name'])
|
|
dep=matches[0]; dm=read(dep/'archive-manifest.json')
|
|
assert digest((dep/'source-map.json').read_bytes())==dm['source_map_sha256']
|
|
for r in dm['files']:check((dep/r['archive']).read_bytes(),r)
|
|
ds=read(dep/'source-map.json')
|
|
for pin in sorted({r['commit'] for r in ds}):
|
|
rows=[r for r in ds if r['commit']==pin and r['storage']=='git']; blobs=dict(git_blobs(a.repo,pin,[r['git_path'] for r in rows]))
|
|
for r in rows:check(blobs[r['git_path']],r)
|
|
for r in ds:
|
|
if r['storage']=='override':check((dep/r['archive']).read_bytes(),r)
|
|
references=[]
|
|
for r in m['files']:
|
|
if r['archive'].endswith('.json'):
|
|
for ref in records(read(root/r['archive'])):
|
|
references.append(dict(receipt=r['archive'],**ref,available=(ref['bytes'],ref['sha256']) in available))
|
|
external=[r for r in references if not r['available']]
|
|
# Explicitly report external references; do not equate inventory coverage with re-executing a gate.
|
|
print(json.dumps({'archived_originals_verified':checked,'source_records_verified':len(sources),'receipt_references':len(references),'available_receipt_references':len(references)-len(external),'external_reference_count':len(external),'external_paths':sorted({r['path'] for r in external})},indent=2))
|
|
def main():
|
|
p=argparse.ArgumentParser(description=__doc__); sub=p.add_subparsers(dest='cmd',required=True)
|
|
b=sub.add_parser('build');b.add_argument('--repo',required=True);b.add_argument('--native',required=True);b.add_argument('--gate',action='append',required=True);b.add_argument('--run',type=int,action='append',required=True);b.add_argument('--extra',action='append',default=[]);b.add_argument('--output',required=True)
|
|
v=sub.add_parser('verify');v.add_argument('--archive',required=True);v.add_argument('--repo',required=True);v.add_argument('--related-archive',action='append',default=[])
|
|
a=p.parse_args();(build if a.cmd=='build' else verify)(a)
|
|
if __name__=='__main__':main()
|