139 lines
5.4 KiB
Python
139 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Resolve source snapshot and AMD64 OCI metadata into a Gate 2 candidate lock."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import gzip
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tarfile
|
|
from pathlib import Path
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def deterministic_source_archive(source: Path, output: Path) -> dict:
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
excluded_parts = {".git", "__pycache__", ".pytest_cache", "node_modules", ".venv", "venv"}
|
|
files = sorted(
|
|
path for path in source.rglob("*")
|
|
if path.is_file()
|
|
and not excluded_parts.intersection(path.relative_to(source).parts)
|
|
and path.suffix != ".iso"
|
|
and path.name != "how-it-flows.gif"
|
|
)
|
|
with output.open("wb") as raw:
|
|
with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed:
|
|
with tarfile.open(fileobj=compressed, mode="w") as archive:
|
|
for path in files:
|
|
relative = path.relative_to(source)
|
|
info = archive.gettarinfo(str(path), arcname=str(Path("aipackage") / relative))
|
|
info.uid = info.gid = 0
|
|
info.uname = info.gname = "root"
|
|
info.mtime = 0
|
|
with path.open("rb") as stream:
|
|
archive.addfile(info, stream)
|
|
return {
|
|
"id": "nexus-one-ai-source-snapshot",
|
|
"version": os.environ.get("OFFLINE_SOURCE_COMMIT", "exported-snapshot"),
|
|
"architecture": "all",
|
|
"source": "local-reviewed-source-snapshot",
|
|
"target": "payload/source/aipackage.tar.gz",
|
|
"sha256": sha256_file(output),
|
|
"bytes": output.stat().st_size,
|
|
"bytes_basis": "measured",
|
|
"license": "Proprietary-Cezen-Technologies",
|
|
"file_count": len(files),
|
|
}
|
|
|
|
|
|
def manifest_records(reference: str) -> list[dict]:
|
|
result = subprocess.run(
|
|
["docker", "manifest", "inspect", "--verbose", reference],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
parsed = json.loads(result.stdout)
|
|
return parsed if isinstance(parsed, list) else [parsed]
|
|
|
|
|
|
def resolve_container(item: dict) -> dict:
|
|
records = manifest_records(item["source"])
|
|
matches = [
|
|
record for record in records
|
|
if record.get("Descriptor", {}).get("platform", {}).get("os") == "linux"
|
|
and record.get("Descriptor", {}).get("platform", {}).get("architecture") == "amd64"
|
|
]
|
|
if len(matches) != 1:
|
|
raise RuntimeError(f"{item['source']}: expected one linux/amd64 manifest, found {len(matches)}")
|
|
record = matches[0]
|
|
descriptor = record["Descriptor"]
|
|
digest = descriptor["digest"]
|
|
manifest = record.get("SchemaV2Manifest") or {}
|
|
content = [manifest.get("config", {})] + list(manifest.get("layers", []))
|
|
declared_bytes = sum(entry.get("size", 0) for entry in content)
|
|
return {
|
|
"id": item["id"],
|
|
"version": item["version"],
|
|
"architecture": "amd64",
|
|
"source": f"{item['source'].split(':')[0]}@{digest}",
|
|
"resolved_from": item["source"],
|
|
"target": f"payload/containers/{item['id']}.oci.tar",
|
|
"sha256": digest.removeprefix("sha256:"),
|
|
"bytes": declared_bytes,
|
|
"bytes_basis": "registry-declared-content",
|
|
"license": item["license"],
|
|
"manifest_bytes": descriptor.get("size"),
|
|
"config": manifest.get("config"),
|
|
"layers": manifest.get("layers", []),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--candidate", type=Path, required=True)
|
|
parser.add_argument("--source-root", type=Path, required=True)
|
|
parser.add_argument("--container-sources", type=Path, required=True)
|
|
parser.add_argument("--staging", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
|
|
lock = json.loads(args.candidate.read_text(encoding="utf-8"))
|
|
source_object = deterministic_source_archive(
|
|
args.source_root.resolve(), args.staging / "source" / "aipackage.tar.gz"
|
|
)
|
|
declarations = json.loads(args.container_sources.read_text(encoding="utf-8"))
|
|
containers = [resolve_container(item) for item in declarations]
|
|
lock["objects"]["source"] = [source_object]
|
|
lock["objects"]["containers"] = containers
|
|
lock["unresolved"] = [
|
|
item for item in lock["unresolved"]
|
|
if item not in {"source snapshot archive bytes and digest", "OCI image manifests and layers by digest"}
|
|
]
|
|
lock["unresolved"].extend([
|
|
"OCI archive bytes and SHA-256 after digest-pinned acquisition",
|
|
"license review for container entries marked REVIEW-REQUIRED",
|
|
"replace mutable container source tags with approved release tags before final review"
|
|
])
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(lock, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
print(f"Wrote Gate 2 candidate lock: {args.output}")
|
|
print(f"Source files: {source_object['file_count']}; source bytes: {source_object['bytes']}")
|
|
print(f"AMD64 OCI manifests resolved: {len(containers)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|