| """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() |
|
|
