344 lines
12 KiB
Python
344 lines
12 KiB
Python
#!/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("--package-lock", type=Path, 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)
|
|
_validate_package_lock(
|
|
_json_object(args.package_lock),
|
|
version=args.version,
|
|
composition=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),
|
|
},
|
|
"package_lock": {
|
|
"url": f"{artifact_base}/package-artifacts.lock.json",
|
|
"sha256": _sha256_file(args.package_lock),
|
|
},
|
|
"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 _validate_package_lock(
|
|
lock: dict[str, Any],
|
|
*,
|
|
version: str,
|
|
composition: dict[str, Any],
|
|
) -> None:
|
|
if lock.get("schema_version") != "1" or lock.get("release_version") != version:
|
|
raise ValueError("package lock schema or release version does not match")
|
|
unsigned = dict(lock)
|
|
expected_hash = unsigned.pop("lock_sha256", None)
|
|
actual_hash = hashlib.sha256(
|
|
json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
).hexdigest()
|
|
if expected_hash != actual_hash:
|
|
raise ValueError("package lock hash does not match its contents")
|
|
rows = lock.get("python")
|
|
if not isinstance(rows, list):
|
|
raise ValueError("package lock Python artifacts are missing")
|
|
locked = _package_identities(rows, name_key="name")
|
|
composed = _package_identities(
|
|
composition["python"]["packages"],
|
|
name_key="package",
|
|
)
|
|
if locked != composed:
|
|
raise ValueError("package lock does not match the runtime wheel composition")
|
|
|
|
|
|
def _package_identities(
|
|
rows: list[object],
|
|
*,
|
|
name_key: str,
|
|
) -> dict[str, tuple[str, str]]:
|
|
identities: dict[str, tuple[str, str]] = {}
|
|
for row in rows:
|
|
if not isinstance(row, dict):
|
|
raise ValueError("package artifact identity is malformed")
|
|
name = row.get(name_key)
|
|
version = row.get("version")
|
|
digest = row.get("sha256")
|
|
if (
|
|
not isinstance(name, str)
|
|
or not isinstance(version, str)
|
|
or not isinstance(digest, str)
|
|
or SHA256.fullmatch(digest) is None
|
|
or name in identities
|
|
):
|
|
raise ValueError("package artifact identity is malformed or duplicated")
|
|
identities[name] = (version, digest)
|
|
return identities
|
|
|
|
|
|
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())
|