verify_assets.py

"""Read-only complete-EOF checks for the exact private HF asset manifest."""
import argparse,hashlib,json,os,stat
from pathlib import Path,PurePosixPath
MANIFEST_SHA256='24ee03260281cc1cbea9555b8d6ae695941bde55ac9fdad77682205615c9cb5d'
def verify_one(root,relative,expected):
 parts=PurePosixPath(relative).parts
 if not parts or relative.startswith('/') or '..' in parts or str(PurePosixPath(relative))!=relative:raise ValueError('Unsafe asset path')
 fd=os.open(root,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW)
 try:
  for part in parts[:-1]:
   nxt=os.open(part,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=fd);os.close(fd);fd=nxt
  src=os.open(parts[-1],os.O_RDONLY|os.O_NOFOLLOW,dir_fd=fd)
  with os.fdopen(src,'rb') as f:
   before=os.fstat(f.fileno())
   if not stat.S_ISREG(before.st_mode) or before.st_size!=expected['bytes']:raise ValueError('Asset type/size differs: '+relative)
   h=hashlib.sha256();n=0
   while chunk:=f.read(8*1024*1024):h.update(chunk);n+=len(chunk)
   after=os.fstat(f.fileno())
   keys=('st_dev','st_ino','st_mode','st_size','st_mtime_ns','st_ctime_ns')
   if any(getattr(before,k)!=getattr(after,k) for k in keys):raise ValueError('Asset changed while reading: '+relative)
   if n!=expected['bytes'] or h.hexdigest()!=expected['sha256']:raise ValueError('Asset EOF identity differs: '+relative)
   return {'repo_path':relative,'bytes':n,'sha256':h.hexdigest(),'full_eof':True}
 finally:os.close(fd)
def verify(root,manifest):
 b=Path(manifest).read_bytes()
 if hashlib.sha256(b).hexdigest()!=MANIFEST_SHA256:raise ValueError('Unreviewed asset manifest')
 m=json.loads(b)
 if m['schema']!='segment-private-hf-serving-assets/1' or m['counts']['total']!=50 or len(m['assets'])!=50:raise ValueError('Manifest scope')
 rows=[verify_one(Path(root),r['repo_path'],r) for r in m['assets']]
 return {'schema':'segment-private-hf-local-assets-eof/1','manifest_sha256':MANIFEST_SHA256,'files':rows,'total_bytes':sum(r['bytes'] for r in rows),'all_complete':True,'scope':'Local model/prior byte identity only; no model execution, upload or access proof.'}
def main():
 p=argparse.ArgumentParser();p.add_argument('--root',required=True);p.add_argument('--manifest',required=True);a=p.parse_args();print(json.dumps(verify(a.root,a.manifest),sort_keys=True,indent=2))
if __name__=='__main__':main()