40-validation/validate-cross-repository-references.py

#!/usr/bin/env python3
"""Validate repository-qualified references declared in repository manifests."""
from __future__ import annotations
import argparse, json, re, sys
from pathlib import Path

REF_RE = re.compile(r"^(?P<repo>[a-z][a-z0-9-]*):(?P<path>[^\\:]+)$")
ROLES = {"canonical-source","specialization","realization","materialization","validation","evidence"}
CHAIN_REQUIRED_ROLES = {"canonical-source","specialization","realization","materialization","evidence"}

def find_roots(container: Path) -> dict[str, Path]:
    roots: dict[str, Path] = {}
    for manifest in container.rglob("manifest.json"):
        root = manifest.parent
        try: data=json.loads(manifest.read_text())
        except Exception: continue
        rid=data.get("artifactName") or data.get("repository",{}).get("id")
        if not isinstance(rid,str): continue
        if rid in roots and roots[rid] != root:
            raise ValueError(f"duplicate repository identity: {rid}")
        roots[rid]=root
    return roots

def validate(container: Path) -> list[str]:
    errors=[]
    try: roots=find_roots(container)
    except ValueError as exc: return [str(exc)]
    for rid,root in sorted(roots.items()):
        data=json.loads((root/"manifest.json").read_text())
        refs=data.get("crossRepositoryReferences",[])
        if not isinstance(refs,list):
            errors.append(f"{rid}: crossRepositoryReferences must be a list"); continue
        for i,item in enumerate(refs):
            if not isinstance(item,dict): errors.append(f"{rid}[{i}]: declaration must be an object"); continue
            ref=item.get("reference"); role=item.get("role"); required=item.get("required",True)
            if role not in ROLES: errors.append(f"{rid}[{i}]: unknown role {role!r}")
            if not isinstance(ref,str) or not (m:=REF_RE.fullmatch(ref)):
                errors.append(f"{rid}[{i}]: invalid reference syntax {ref!r}"); continue
            target_repo=m.group("repo"); rel=m.group("path")
            p=Path(rel)
            if p.is_absolute() or ".." in p.parts or not rel or rel.endswith("/"):
                errors.append(f"{rid}[{i}]: unsafe reference path {rel!r}"); continue
            if required:
                if target_repo not in roots: errors.append(f"{rid}[{i}]: unknown target repository {target_repo}")
                elif not (roots[target_repo]/p).is_file(): errors.append(f"{rid}[{i}]: missing target {ref}")
    # Validate distributed realization-chain declarations.
    chains: dict[str, dict[str, object]] = {}
    for rid, root in sorted(roots.items()):
        data=json.loads((root/"manifest.json").read_text())
        declarations=data.get("realizationChains",[])
        if not isinstance(declarations,list):
            errors.append(f"{rid}: realizationChains must be a list")
            continue
        for i,decl in enumerate(declarations):
            if not isinstance(decl,dict):
                errors.append(f"{rid}.realizationChains[{i}]: declaration must be an object")
                continue
            cid=decl.get("id"); registry=decl.get("registry"); artifacts=decl.get("artifacts",[])
            if not isinstance(cid,str) or not cid:
                errors.append(f"{rid}.realizationChains[{i}]: missing chain id")
                continue
            entry=chains.setdefault(cid,{"roles":set(),"registries":set(),"members":set()})
            entry["members"].add(rid)
            if not isinstance(registry,str) or not (m:=REF_RE.fullmatch(registry)):
                errors.append(f"{rid}.{cid}: invalid registry reference {registry!r}")
            else:
                target_repo=m.group("repo"); rel=m.group("path"); pp=Path(rel)
                entry["registries"].add(registry)
                if target_repo not in roots or not (roots[target_repo]/pp).is_file():
                    errors.append(f"{rid}.{cid}: missing registry {registry}")
            if not isinstance(artifacts,list):
                errors.append(f"{rid}.{cid}: artifacts must be a list")
                continue
            for j,item in enumerate(artifacts):
                if not isinstance(item,dict):
                    errors.append(f"{rid}.{cid}[{j}]: artifact must be an object")
                    continue
                rel=item.get("reference"); role=item.get("role"); required=item.get("required",True)
                if role not in ROLES:
                    errors.append(f"{rid}.{cid}[{j}]: unknown role {role!r}")
                else:
                    entry["roles"].add(role)
                if not isinstance(rel,str):
                    errors.append(f"{rid}.{cid}[{j}]: invalid local reference {rel!r}")
                    continue
                pp=Path(rel)
                if pp.is_absolute() or ".." in pp.parts or not rel or rel.endswith("/"):
                    errors.append(f"{rid}.{cid}[{j}]: unsafe local path {rel!r}")
                elif required and not (root/pp).is_file():
                    errors.append(f"{rid}.{cid}[{j}]: missing local artifact {rel}")
    for cid,entry in sorted(chains.items()):
        if len(entry["registries"]) != 1:
            errors.append(f"{cid}: declarations must use exactly one registry")
        missing=CHAIN_REQUIRED_ROLES-entry["roles"]
        if missing:
            errors.append(f"{cid}: missing required roles {sorted(missing)}")
    return errors

def main() -> int:
    ap=argparse.ArgumentParser(); ap.add_argument("container",type=Path); args=ap.parse_args()
    errors=validate(args.container)
    if errors:
        for e in errors: print(f"ERROR: {e}",file=sys.stderr)
        return 1
    print("Cross-repository references valid.")
    return 0
if __name__=="__main__": raise SystemExit(main())