112 lines
8.0 KiB
Python
112 lines
8.0 KiB
Python
"""Capture the generated source bundle; delegate unchanged semantic oracle to check.py."""
|
|
from pathlib import Path, PurePosixPath
|
|
import argparse, datetime, hashlib, json, re, shutil, subprocess, sys
|
|
HERE=Path(__file__).resolve().parent
|
|
ROOT=HERE.parent
|
|
WORK=ROOT/'working'
|
|
def digest(p):
|
|
b=p.read_bytes(); return {'bytes':len(b),'sha256':hashlib.sha256(b).hexdigest()}
|
|
def read(p): return json.loads(p.read_text(encoding='utf-8-sig'))
|
|
def write(p,v):
|
|
with p.open('x',encoding='utf-8') as f: json.dump(v,f,indent=2);f.write('\n')
|
|
def contained(path):
|
|
p=path.absolute()
|
|
if not p.is_relative_to(HERE): raise RuntimeError('Outside supplement')
|
|
current=p
|
|
while True:
|
|
if current.exists() and (current.is_symlink() or current.is_junction() or getattr(current.lstat(),'st_file_attributes',0)&1024): raise RuntimeError('Linked ancestor: '+str(current))
|
|
if current==HERE: break
|
|
current=current.parent
|
|
if not p.resolve().is_relative_to(HERE.resolve()): raise RuntimeError('Resolved path escapes supplement')
|
|
def verify_record(r):
|
|
if digest(Path(r['path']))!={k:r[k] for k in ('bytes','sha256')}: raise RuntimeError('Admission artifact changed')
|
|
def inventory(root):
|
|
if root.is_symlink() or root.is_junction() or getattr(root.lstat(),'st_file_attributes',0)&1024: raise RuntimeError('Linked bundle root forbidden')
|
|
out={}
|
|
for p in root.rglob('*'):
|
|
if p.is_symlink() or p.is_junction() or getattr(p.lstat(), 'st_file_attributes', 0)&1024: raise RuntimeError('Links forbidden: '+str(p))
|
|
if p.is_file() and p.suffix.lower()=='.arcadia': out[p.relative_to(root).as_posix()]=digest(p)
|
|
return out
|
|
def closure(entry, inv):
|
|
# Bounded generated fixture: includes only in entry, with simple normalized paths.
|
|
text=entry.read_text(encoding='utf-8-sig')
|
|
paths=re.findall(r'\binclude\s+"([^"\r\n]+)"\s*;',text)
|
|
if len(re.findall(r'\binclude\b',text))!=len(paths): raise RuntimeError('Unsupported include syntax')
|
|
for name in paths:
|
|
p=PurePosixPath(name)
|
|
if p.is_absolute() or '..' in p.parts or ':' in name or '\\' in name or str(p)!=name or name not in inv: raise RuntimeError('Unsafe/missing include: '+name)
|
|
if re.search(r'\binclude\b',(entry.parent/name).read_text(encoding='utf-8-sig')): raise RuntimeError('Nested includes outside this fixed assay')
|
|
if set(inv)!={entry.name,*paths}: raise RuntimeError('Unexpected source file outside declared bundle')
|
|
def main():
|
|
ap=argparse.ArgumentParser();ap.add_argument('--expected',choices=['B','P'],required=True);ap.add_argument('--checkpoint',required=True);ap.add_argument('--calibration',type=Path);a=ap.parse_args()
|
|
if not re.fullmatch(r'[0-9]{2}-[a-z0-9-]+',a.checkpoint): raise RuntimeError('Use NN-name checkpoint')
|
|
if (HERE/'UNACCEPTED.json').exists(): raise RuntimeError('Prior outer failure: trial quarantined, no binding reuse or relearning')
|
|
dest=HERE/'checkpoints'/a.checkpoint
|
|
contained(dest)
|
|
if dest.exists(): raise RuntimeError('Fresh checkpoint required')
|
|
manifest=read(ROOT/'input-manifest.json');entry=Path(manifest['saved_working_path'])
|
|
if entry.parent.resolve()!=WORK.resolve() or entry.is_symlink(): raise RuntimeError('Unexpected entry location')
|
|
prior=ROOT/'fresh-node-binding.json'
|
|
if prior.exists():
|
|
binding=read(prior); first=Path(binding['first_checkpoint_receipt']['path']);outer=first.parent/'bundle-receipt.json'
|
|
accepted=read(outer)
|
|
if not accepted['accepted'] or accepted['inner_receipt']!=digest(first) or accepted['binding_after']!=digest(prior): raise RuntimeError('Binding lacks successful outer acceptance')
|
|
admission=read(HERE/'converter-admission.json')
|
|
if admission.get('approved') is not True or admission['source_commit']!=manifest['source_commit']: raise RuntimeError('Converter not admitted')
|
|
for r in admission['records']:verify_record(r)
|
|
converter=admission['binary'];verify_record(converter)
|
|
dest.mkdir(parents=True)
|
|
contained(dest)
|
|
receipt={'started_utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),'accepted':False,'failure':None,'expected':a.expected,'wrapper':digest(Path(__file__)),'plan':digest(HERE/'README.md'),'manifest':digest(ROOT/'input-manifest.json'),'checker':digest(ROOT/'check.py'),'converter_admission':digest(HERE/'converter-admission.json'),'entry':str(entry),'binding_before':digest(prior) if prior.exists() else None}
|
|
try:
|
|
before=inventory(WORK);closure(entry,before)
|
|
if any(x in before for x in ('state.arcadia','converted.arcadia')): raise RuntimeError('Snapshot entry collision')
|
|
receipt['original_before']=before
|
|
for name in before:
|
|
target=dest/name;target.parent.mkdir(parents=True,exist_ok=True)
|
|
with target.open('xb') as f:f.write((WORK/name).read_bytes())
|
|
if inventory(dest)!=before or inventory(WORK)!=before: raise RuntimeError('Bundle changed during capture')
|
|
converted=dest/'converted.arcadia'
|
|
with (dest/'converter.stdout.log').open('xb') as out,(dest/'converter.stderr.log').open('xb') as err:
|
|
conversion=subprocess.run([converter['path'],str(dest/entry.name),str(converted)],cwd=ROOT/'unrelated-cwd',stdout=out,stderr=err)
|
|
receipt['converter_argv']=[converter['path'],str(dest/entry.name),str(converted)]
|
|
receipt['converter_cwd']=str(ROOT/'unrelated-cwd')
|
|
receipt['converter_stdout']=digest(dest/'converter.stdout.log');receipt['converter_stderr']=digest(dest/'converter.stderr.log')
|
|
receipt['converter_exit_code']=conversion.returncode
|
|
if conversion.returncode or (dest/'converter.stderr.log').stat().st_size: raise RuntimeError('Converter failed')
|
|
receipt['converted']=digest(converted)
|
|
argv=[sys.executable,str(ROOT/'check.py'),'--expected',a.expected,'--actual',str(converted),'--output-prefix',str(dest/'state')]
|
|
if a.calibration: argv+=['--calibration',str(a.calibration.resolve())]
|
|
receipt['argv']=argv
|
|
with (dest/'wrapper.stdout.log').open('xb') as out,(dest/'wrapper.stderr.log').open('xb') as err:
|
|
result=subprocess.run(argv,cwd=ROOT,stdout=out,stderr=err)
|
|
receipt['exit_code']=result.returncode
|
|
receipt['checker_stdout']=digest(dest/'wrapper.stdout.log');receipt['checker_stderr']=digest(dest/'wrapper.stderr.log')
|
|
inner=dest/'state.receipt.json';receipt['inner_receipt']=digest(inner) if inner.exists() else None
|
|
receipt['original_after']=inventory(WORK);receipt['snapshot_after']=inventory(dest)
|
|
expected=dict(before);expected['converted.arcadia']=receipt['converted'];expected['state.arcadia']=receipt['converted']
|
|
if result.returncode or not inner.exists():raise RuntimeError('Inner checker failed')
|
|
r=read(inner)
|
|
if r['failure'] or not r['full_project_exact'] or not r['inputs_unchanged']:raise RuntimeError('Inner comparison not accepted')
|
|
if receipt['original_after']!=before or receipt['snapshot_after']!=expected:raise RuntimeError('Original or copied bundle changed')
|
|
if digest(Path(__file__))!=receipt['wrapper'] or digest(HERE/'README.md')!=receipt['plan']:raise RuntimeError('Supplement changed')
|
|
contained(dest)
|
|
for r in admission['records']:verify_record(r)
|
|
verify_record(converter)
|
|
if digest(HERE/'converter-admission.json')!=receipt['converter_admission']:raise RuntimeError('Converter admission changed')
|
|
receipt['binding_after']=digest(prior) if prior.exists() else None
|
|
if receipt['binding_before'] and receipt['binding_after']!=receipt['binding_before']:raise RuntimeError('Existing binding changed')
|
|
receipt['accepted']=True
|
|
except Exception as e:
|
|
receipt['failure']=str(e)
|
|
write(HERE/'UNACCEPTED.json',{'checkpoint':str(dest),'failure':str(e),'binding_unaccepted':True,'instruction':'Stop this trial; preserve binding and failures, never delete or relearn.'})
|
|
finally:
|
|
receipt['finished_utc']=datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
try: write(dest/'bundle-receipt.json',receipt)
|
|
except Exception as e:
|
|
if not (HERE/'UNACCEPTED.json').exists(): write(HERE/'UNACCEPTED.json',{'checkpoint':str(dest),'failure':'Outer receipt write failed: '+str(e),'binding_unaccepted':True})
|
|
raise
|
|
if not receipt['accepted']:raise SystemExit('Outer checkpoint failed; trial quarantined')
|
|
print('Complete source bundle and unchanged full-Project oracle accepted: '+a.checkpoint)
|
|
if __name__=='__main__': main()
|