40-validation/validate-bundle-intake.py

#!/usr/bin/env python3
"""Validate repeatable intake of GAP, Source Bundle and Target Bundle archives."""
from __future__ import annotations
import argparse, hashlib, io, json, os, zipfile
from pathlib import Path
from evidence_lifecycle import seal_evidence

VALIDATOR_REF = "engineering-tools:40-validation/validate-bundle-intake.py"
DOC_REF = "engineering-tools:40-validation/BUNDLE_INTAKE_VALIDATION.md"

def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()

def sha256_file(path: Path) -> str:
    h=hashlib.sha256()
    with path.open('rb') as f:
        for chunk in iter(lambda:f.read(1024*1024), b''): h.update(chunk)
    return h.hexdigest()

def atomic_json(path: Path, data: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp=path.with_suffix(path.suffix+'.tmp')
    tmp.write_text(json.dumps(data,indent=2,ensure_ascii=False)+'\n',encoding='utf-8')
    os.replace(tmp,path)

def identity_from_manifest(manifest: dict) -> tuple[str|None,str|None]:
    repo=manifest.get('repository') if isinstance(manifest.get('repository'),dict) else {}
    rid=repo.get('id') or manifest.get('artifactName') or manifest.get('repositoryId') or manifest.get('id')
    ver=repo.get('version') or manifest.get('version') or manifest.get('semanticVersion')
    return (str(rid) if rid is not None else None, str(ver) if ver is not None else None)

def inspect_repository_archive(name: str, data: bytes) -> tuple[dict,list[str]]:
    errors=[]
    try:
        z=zipfile.ZipFile(io.BytesIO(data))
    except zipfile.BadZipFile:
        return ({'archive':name,'sha256':sha256_bytes(data),'status':'FAILED'},[f'invalid nested zip: {name}'])
    files=[n for n in z.namelist() if not n.endswith('/')]
    roots={n.split('/')[0] for n in files if '/' in n}
    if len(roots)!=1: errors.append(f'{name}: expected exactly one repository root, found {sorted(roots)}')
    manifests=[n for n in files if n=='manifest.json' or n.endswith('/manifest.json')]
    root_manifests=[m for m in manifests if m.count('/')<=1]
    if len(root_manifests)!=1:
        errors.append(f'{name}: expected exactly one root manifest, found {root_manifests}')
        manifest={}; mpath=None
    else:
        mpath=root_manifests[0]
        try: manifest=json.loads(z.read(mpath))
        except Exception as exc:
            errors.append(f'{name}: invalid manifest JSON: {exc}'); manifest={}
    rid,ver=identity_from_manifest(manifest)
    if not rid: errors.append(f'{name}: repository identity missing')
    entry_candidates=[n for n in files if n.endswith('/AI_STARTS_HERE.md') or n.endswith('/README.md')]
    versionSource='manifest' if ver is not None else None
    if ver is None:
        ai=[n for n in files if n.endswith('/AI_STARTS_HERE.md')]
        if ai:
            import re
            text=z.read(sorted(ai)[0]).decode('utf-8','replace')
            match=re.search(r'^\*\*Version:\*\*\s*(\S+)',text,re.MULTILINE)
            if match:
                ver=match.group(1); versionSource='AI_STARTS_HERE.md'
    if ver is None: errors.append(f'{name}: repository version missing')
    return ({'archive':name,'archiveSha256':sha256_bytes(data),'root':next(iter(roots),None),'manifestPath':mpath,'repositoryId':rid,'version':ver,'versionSource':versionSource,'entryEvidence':sorted(entry_candidates)[:5],'status':'PASSED' if not errors else 'FAILED'},errors)

def inspect_bundle(path: Path, role: str) -> tuple[dict,list[str]]:
    errors=[]
    try: outer=zipfile.ZipFile(path)
    except zipfile.BadZipFile:
        return ({'path':path.name,'role':role,'sha256':sha256_file(path),'status':'FAILED','repositories':[]},[f'invalid bundle zip: {path}'])
    nested=[n for n in outer.namelist() if not n.endswith('/') and n.lower().endswith('.zip')]
    if not nested: errors.append(f'{role} bundle contains no repository archives')
    repos=[]
    seen={}
    for name in sorted(nested):
        rec, rec_errors=inspect_repository_archive(name,outer.read(name)); repos.append(rec); errors.extend(rec_errors)
        rid=rec.get('repositoryId')
        if rid:
            fingerprint=(rec.get('version'),rec.get('archiveSha256'))
            if rid in seen and seen[rid]!=fingerprint:
                errors.append(f'{role} bundle contains conflicting repository identity: {rid}')
            elif rid in seen:
                errors.append(f'{role} bundle contains duplicate repository identity: {rid}')
            else: seen[rid]=fingerprint
    return ({'path':path.name,'role':role,'sha256':sha256_file(path),'repositoryCount':len(repos),'repositories':repos,'status':'PASSED' if not errors else 'FAILED'},errors)

def validate_intake(gap_zip:Path,source_bundle:Path,target_bundle:Path,expected_source_sha256:str|None=None)->tuple[dict,list[str]]:
    errors=[]
    gap_rec,gerrs=inspect_repository_archive(gap_zip.name,gap_zip.read_bytes()); errors.extend(gerrs)
    if gap_rec.get('repositoryId')!='gap-000': errors.append('GAP archive does not identify gap-000')
    source,serrs=inspect_bundle(source_bundle,'source'); errors.extend(serrs)
    target,terrs=inspect_bundle(target_bundle,'target'); errors.extend(terrs)
    if expected_source_sha256 and source['sha256']!=expected_source_sha256:
        errors.append('Source Bundle SHA-256 differs from expected immutable baseline')
    source_ids={r.get('repositoryId') for r in source['repositories']}
    target_ids={r.get('repositoryId') for r in target['repositories']}
    overlaps=sorted((source_ids & target_ids)-{None})
    result={'validationId':'repeatable-bundle-intake','status':'PASSED' if not errors else 'FAILED','gap':gap_rec,'sourceBundle':source,'targetBundle':target,'crossBundleIdentityOverlap':overlaps,'sourceImmutable':not any('Source Bundle SHA-256' in e for e in errors),'duplicateIdentitiesBlocked':True,'errors':errors,'deploymentPerformed':False,'promotionPerformed':False}
    return result,errors

def main()->int:
    p=argparse.ArgumentParser(); p.add_argument('gap_zip',type=Path); p.add_argument('source_bundle',type=Path); p.add_argument('target_bundle',type=Path); p.add_argument('repository_roots',type=Path); p.add_argument('runtime_root',type=Path); p.add_argument('--expected-source-sha256'); p.add_argument('--work-item',default='KT-0020'); a=p.parse_args()
    result,errors=validate_intake(a.gap_zip.resolve(),a.source_bundle.resolve(),a.target_bundle.resolve(),a.expected_source_sha256)
    mat=a.runtime_root.resolve()/'test/validations/KT-0020/bundle-intake-manifest.json'
    ev=a.runtime_root.resolve()/'test/reports/validation/bundle-intake.json'
    atomic_json(mat,result)
    refs=[DOC_REF,'runtime:test/validations/KT-0020/bundle-intake-manifest.json']
    sealed=seal_evidence(result,a.repository_roots.resolve(),VALIDATOR_REF,refs,work_item=a.work_item,chain='BUNDLE-INTAKE')
    atomic_json(ev,sealed)
    if errors:
        for e in errors: print('ERROR:',e)
        return 1
    print(f"Bundle intake PASSED: source={result['sourceBundle']['repositoryCount']} target={result['targetBundle']['repositoryCount']}")
    return 0
if __name__=='__main__': raise SystemExit(main())