Add signed runtime distribution pipeline
This commit is contained in:
@@ -0,0 +1,282 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Create runtime SBOM, provenance, and an unsigned distribution descriptor."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
from typing import Any
|
||||
from urllib.parse import urlsplit
|
||||
import uuid
|
||||
|
||||
|
||||
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
||||
DIGEST_IMAGE = re.compile(r"^[^@\s]+@sha256:[0-9a-f]{64}$")
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--composition", type=Path, required=True)
|
||||
parser.add_argument("--api-metadata", type=Path, required=True)
|
||||
parser.add_argument("--web-metadata", type=Path, required=True)
|
||||
parser.add_argument("--deployer", type=Path, required=True)
|
||||
parser.add_argument("--deployer-url", required=True)
|
||||
parser.add_argument("--artifact-base-url", required=True)
|
||||
parser.add_argument("--source-commit", required=True)
|
||||
parser.add_argument("--version", required=True)
|
||||
parser.add_argument("--channel", default="stable")
|
||||
parser.add_argument("--sequence", type=int, required=True)
|
||||
parser.add_argument("--expires-days", type=int, default=90)
|
||||
parser.add_argument(
|
||||
"--dependency",
|
||||
action="append",
|
||||
default=[],
|
||||
metavar="NAME=IMAGE@SHA256",
|
||||
)
|
||||
parser.add_argument("--output-directory", type=Path, required=True)
|
||||
parser.add_argument("--descriptor", type=Path, required=True)
|
||||
return parser
|
||||
|
||||
|
||||
def finalize(args: argparse.Namespace) -> dict[str, Any]:
|
||||
composition = _json_object(args.composition)
|
||||
api = _image_metadata(_json_object(args.api_metadata), "api")
|
||||
web = _image_metadata(_json_object(args.web_metadata), "web")
|
||||
dependencies = dict(_dependency(value) for value in args.dependency)
|
||||
if not dependencies:
|
||||
raise ValueError("at least one --dependency is required")
|
||||
_https_url(args.deployer_url, "deployer URL")
|
||||
artifact_base = _https_url(args.artifact_base_url, "artifact base URL").rstrip("/")
|
||||
if args.sequence < 1 or not 1 <= args.expires_days <= 365:
|
||||
raise ValueError("sequence and expiry window are out of bounds")
|
||||
output = args.output_directory.expanduser().resolve()
|
||||
output.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
api_sbom = _api_sbom(composition, version=args.version)
|
||||
web_sbom = _web_sbom(composition, version=args.version)
|
||||
api_provenance = _provenance(
|
||||
subject=api["index"],
|
||||
source_commit=args.source_commit,
|
||||
composition=composition,
|
||||
)
|
||||
web_provenance = _provenance(
|
||||
subject=web["index"],
|
||||
source_commit=args.source_commit,
|
||||
composition=composition,
|
||||
)
|
||||
artifact_values = {
|
||||
"api-sbom.cdx.json": api_sbom,
|
||||
"web-sbom.cdx.json": web_sbom,
|
||||
"api-provenance.json": api_provenance,
|
||||
"web-provenance.json": web_provenance,
|
||||
}
|
||||
artifacts: dict[str, dict[str, str]] = {}
|
||||
for filename, value in artifact_values.items():
|
||||
path = output / filename
|
||||
encoded = _canonical_json(value)
|
||||
path.write_bytes(encoded)
|
||||
artifacts[filename] = {
|
||||
"url": f"{artifact_base}/{filename}",
|
||||
"sha256": hashlib.sha256(encoded).hexdigest(),
|
||||
}
|
||||
composition_encoded = _canonical_json(composition)
|
||||
packages = _manifest_packages(composition)
|
||||
issued = datetime.now(UTC).replace(microsecond=0)
|
||||
descriptor: dict[str, Any] = {
|
||||
"schema_version": "1",
|
||||
"channel": args.channel,
|
||||
"sequence": args.sequence,
|
||||
"version": args.version,
|
||||
"issued_at": issued.isoformat(),
|
||||
"expires_at": (issued + timedelta(days=args.expires_days)).isoformat(),
|
||||
"revoked": False,
|
||||
"deployer": {
|
||||
"url": args.deployer_url,
|
||||
"sha256": _sha256_file(args.deployer),
|
||||
},
|
||||
"images": {
|
||||
"api": {
|
||||
**api,
|
||||
"sbom": artifacts["api-sbom.cdx.json"],
|
||||
"provenance": artifacts["api-provenance.json"],
|
||||
},
|
||||
"web": {
|
||||
**web,
|
||||
"sbom": artifacts["web-sbom.cdx.json"],
|
||||
"provenance": artifacts["web-provenance.json"],
|
||||
},
|
||||
},
|
||||
"dependencies": dict(sorted(dependencies.items())),
|
||||
"composition": {
|
||||
"sha256": hashlib.sha256(composition_encoded).hexdigest(),
|
||||
"module_ids": list(composition["python"]["module_ids"]),
|
||||
"packages": packages,
|
||||
},
|
||||
}
|
||||
args.descriptor.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.descriptor.write_bytes(_canonical_json(descriptor))
|
||||
return descriptor
|
||||
|
||||
|
||||
def _api_sbom(composition: dict[str, Any], *, version: str) -> dict[str, Any]:
|
||||
components = []
|
||||
for package in composition["python"]["packages"]:
|
||||
components.append(
|
||||
{
|
||||
"type": "library",
|
||||
"name": package["package"],
|
||||
"version": package["version"],
|
||||
"hashes": [{"alg": "SHA-256", "content": package["sha256"]}],
|
||||
"purl": f"pkg:pypi/{package['package']}@{package['version']}",
|
||||
}
|
||||
)
|
||||
return {
|
||||
"bomFormat": "CycloneDX",
|
||||
"specVersion": "1.6",
|
||||
"serialNumber": f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, 'govoplan-api:' + version)}",
|
||||
"version": 1,
|
||||
"metadata": {"component": {"type": "application", "name": "govoplan-api", "version": version}},
|
||||
"components": components,
|
||||
}
|
||||
|
||||
|
||||
def _web_sbom(composition: dict[str, Any], *, version: str) -> dict[str, Any]:
|
||||
return {
|
||||
"bomFormat": "CycloneDX",
|
||||
"specVersion": "1.6",
|
||||
"serialNumber": f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, 'govoplan-web:' + version)}",
|
||||
"version": 1,
|
||||
"metadata": {"component": {"type": "application", "name": "govoplan-web", "version": version}},
|
||||
"components": [
|
||||
{
|
||||
"type": "file",
|
||||
"name": "govoplan-web-dist",
|
||||
"version": version,
|
||||
"hashes": [
|
||||
{
|
||||
"alg": "SHA-256",
|
||||
"content": composition["web"]["sha256"],
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _provenance(
|
||||
*,
|
||||
subject: str,
|
||||
source_commit: str,
|
||||
composition: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
digest = subject.rsplit("@sha256:", 1)[1]
|
||||
return {
|
||||
"_type": "https://in-toto.io/Statement/v1",
|
||||
"subject": [{"name": subject.split("@", 1)[0], "digest": {"sha256": digest}}],
|
||||
"predicateType": "https://slsa.dev/provenance/v1",
|
||||
"predicate": {
|
||||
"buildDefinition": {
|
||||
"buildType": "https://govoplan.add-ideas.de/build/runtime-oci/v1",
|
||||
"externalParameters": {
|
||||
"source_commit": source_commit,
|
||||
"network_free_image_assembly": True,
|
||||
},
|
||||
"resolvedDependencies": [
|
||||
{
|
||||
"uri": "govoplan:runtime-composition",
|
||||
"digest": {
|
||||
"sha256": hashlib.sha256(_canonical_json(composition)).hexdigest()
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
"runDetails": {
|
||||
"builder": {"id": "https://git.add-ideas.de/GovOPlaN/govoplan/actions"},
|
||||
"metadata": {"invocationId": source_commit},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _manifest_packages(composition: dict[str, Any]) -> list[dict[str, str]]:
|
||||
values = []
|
||||
for package in composition["python"]["packages"]:
|
||||
values.append(
|
||||
{
|
||||
"name": str(package["package"]),
|
||||
"version": str(package["version"]),
|
||||
"wheel_sha256": str(package["sha256"]),
|
||||
}
|
||||
)
|
||||
return sorted(values, key=lambda item: item["name"])
|
||||
|
||||
|
||||
def _image_metadata(value: dict[str, Any], label: str) -> dict[str, Any]:
|
||||
if set(value) != {"index", "platforms"}:
|
||||
raise ValueError(f"{label} image metadata has invalid fields")
|
||||
if not isinstance(value["index"], str) or DIGEST_IMAGE.fullmatch(value["index"]) is None:
|
||||
raise ValueError(f"{label} index is not digest-pinned")
|
||||
platforms = value["platforms"]
|
||||
if not isinstance(platforms, dict) or set(platforms) != {"linux/amd64", "linux/arm64"}:
|
||||
raise ValueError(f"{label} image does not cover amd64 and arm64")
|
||||
if any(not isinstance(item, str) or DIGEST_IMAGE.fullmatch(item) is None for item in platforms.values()):
|
||||
raise ValueError(f"{label} platform image is not digest-pinned")
|
||||
return {"index": value["index"], "platforms": dict(sorted(platforms.items()))}
|
||||
|
||||
|
||||
def _dependency(value: str) -> tuple[str, str]:
|
||||
if "=" not in value:
|
||||
raise ValueError("--dependency must use NAME=IMAGE@SHA256")
|
||||
name, reference = value.split("=", 1)
|
||||
if re.fullmatch(r"[a-z][a-z0-9_]{1,63}", name) is None:
|
||||
raise ValueError(f"invalid dependency name: {name!r}")
|
||||
if DIGEST_IMAGE.fullmatch(reference) is None:
|
||||
raise ValueError(f"dependency {name!r} is not digest-pinned")
|
||||
return name, reference
|
||||
|
||||
|
||||
def _json_object(path: Path) -> dict[str, Any]:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"JSON root must be an object: {path}")
|
||||
return value
|
||||
|
||||
|
||||
def _https_url(value: str, label: str) -> str:
|
||||
parsed = urlsplit(value)
|
||||
if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password:
|
||||
raise ValueError(f"{label} must be an HTTPS URL without credentials")
|
||||
return value
|
||||
|
||||
|
||||
def _sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as handle:
|
||||
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _canonical_json(value: object) -> bytes:
|
||||
return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = build_parser().parse_args()
|
||||
try:
|
||||
finalize(args)
|
||||
except (KeyError, OSError, TypeError, ValueError, json.JSONDecodeError) as exc:
|
||||
print(f"error: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
print(f"Runtime release evidence written below {args.output_directory}")
|
||||
print(f"Unsigned descriptor written to {args.descriptor}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user