scripts/m2image-manifest.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Validate and query the single source of truth for M2Image releases."""
from __future__ import annotations
import argparse
import datetime as dt
import hashlib
import json
import os
import re
import sys
from pathlib import Path, PurePosixPath
from typing import Any
ALIAS_RE = re.compile(r"^[a-z0-9]+(?:[.-][a-z0-9]+)*$")
ENV_RE = re.compile(r"^[A-Z][A-Z0-9_]*$")
SUPPORTED_ARCHITECTURES = ("x86_64", "aarch64")
class ManifestError(ValueError):
pass
def fail(message: str) -> None:
raise ManifestError(message)
def require_string(value: Any, location: str) -> str:
if not isinstance(value, str) or not value:
fail(f"{location} must be a non-empty string")
return value
def require_relative_path(value: Any, location: str) -> str:
value = require_string(value, location)
path = PurePosixPath(value)
if path.is_absolute() or any(part in ("", ".", "..") for part in path.parts):
fail(f"{location} must be a safe relative path")
return value
def load_manifest(path: Path) -> dict[str, Any]:
try:
data = json.loads(path.read_text(encoding="utf-8"))
except OSError as error:
fail(f"cannot read manifest {path}: {error}")
except json.JSONDecodeError as error:
fail(f"invalid JSON in {path}: {error}")
validate_manifest(data)
return data
def validate_manifest(data: Any) -> None:
if not isinstance(data, dict):
fail("manifest root must be an object")
if data.get("schemaVersion") != 1:
fail("schemaVersion must be 1")
registry = data.get("registry")
if not isinstance(registry, dict):
fail("registry must be an object")
require_relative_path(registry.get("catalogKey"), "registry.catalogKey")
architectures = data.get("architectures")
if architectures != list(SUPPORTED_ARCHITECTURES):
fail("architectures must be exactly [\"x86_64\", \"aarch64\"]")
images = data.get("images")
if not isinstance(images, list) or not images:
fail("images must be a non-empty array")
seen_aliases: set[str] = set()
seen_keys: set[str] = set()
for index, image in enumerate(images):
location = f"images[{index}]"
if not isinstance(image, dict):
fail(f"{location} must be an object")
alias = require_string(image.get("alias"), f"{location}.alias")
if not ALIAS_RE.fullmatch(alias):
fail(f"{location}.alias has an invalid format: {alias}")
if alias in seen_aliases:
fail(f"duplicate alias: {alias}")
seen_aliases.add(alias)
for field in ("distribution", "series", "version", "templateVersion"):
require_string(image.get(field), f"{location}.{field}")
for field in ("revision", "minDiskGb"):
value = image.get(field)
if not isinstance(value, int) or isinstance(value, bool) or value < 1:
fail(f"{location}.{field} must be a positive integer")
expected_template_version = (
f"{image['distribution']}-{image['version']}-v{image['revision']}"
)
if image["templateVersion"] != expected_template_version:
fail(f"{location}.templateVersion must be {expected_template_version}")
builder = image.get("builder")
if not isinstance(builder, dict):
fail(f"{location}.builder must be an object")
require_relative_path(builder.get("script"), f"{location}.builder.script")
requires = builder.get("requires")
if not isinstance(requires, list) or not all(isinstance(item, str) and item for item in requires):
fail(f"{location}.builder.requires must be an array of command names")
environment = builder.get("environment")
if not isinstance(environment, dict):
fail(f"{location}.builder.environment must be an object")
for key, value in environment.items():
if not ENV_RE.fullmatch(key) or not isinstance(value, str):
fail(f"{location}.builder.environment contains an invalid entry: {key}")
artifacts = image.get("artifacts")
if not isinstance(artifacts, dict) or set(artifacts) != set(SUPPORTED_ARCHITECTURES):
fail(f"{location}.artifacts must define x86_64 and aarch64 only")
for architecture in SUPPORTED_ARCHITECTURES:
artifact = artifacts[architecture]
artifact_location = f"{location}.artifacts.{architecture}"
if not isinstance(artifact, dict):
fail(f"{artifact_location} must be an object")
kernel = require_relative_path(artifact.get("kernel"), f"{artifact_location}.kernel")
rootfs = require_relative_path(artifact.get("rootfs"), f"{artifact_location}.rootfs")
initrd = artifact.get("initrd")
if initrd is not None:
require_relative_path(initrd, f"{artifact_location}.initrd")
require_string(artifact.get("bootArgs"), f"{artifact_location}.bootArgs")
registry_key = require_relative_path(
artifact.get("registryKey"), f"{artifact_location}.registryKey"
)
if not kernel.startswith("kernel/") or (initrd and not initrd.startswith("kernel/")):
fail(f"{artifact_location} kernel artifacts must be below kernel/")
if not rootfs.startswith("rootfs/"):
fail(f"{artifact_location}.rootfs must be below rootfs/")
if not registry_key.endswith(f"/{alias}.tar.zst"):
fail(f"{artifact_location}.registryKey must end with /{alias}.tar.zst")
immutable_suffix = (
f"/{image['templateVersion']}/r{image['revision']}/"
f"{architecture}/{alias}.tar.zst"
)
if not registry_key.endswith(immutable_suffix):
fail(
f"{artifact_location}.registryKey must end with "
f"{immutable_suffix}"
)
if registry_key in seen_keys:
fail(f"duplicate registry key: {registry_key}")
seen_keys.add(registry_key)
def find_image(manifest: dict[str, Any], alias: str) -> dict[str, Any]:
for image in manifest["images"]:
if image["alias"] == alias:
return image
fail(f"unknown alias: {alias}")
def get_field(value: Any, dotted_field: str) -> Any:
for part in dotted_field.split("."):
if not isinstance(value, dict) or part not in value:
fail(f"unknown field: {dotted_field}")
value = value[part]
return value
def source_registry_key(manifest: dict[str, Any], alias: str, architecture: str) -> str:
artifact = find_image(manifest, alias)["artifacts"][architecture]
package_key = str(artifact["registryKey"])
suffix = f"/{alias}.tar.zst"
if not package_key.endswith(suffix):
fail(f"registry key for {alias}/{architecture} has an unexpected package suffix")
return package_key[: -len(suffix)] + f"/{alias}.sources.tar.zst"
def sha256(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 published_at(path: Path) -> str:
epoch = int(os.environ.get("SOURCE_DATE_EPOCH", str(int(path.stat().st_mtime))))
return dt.datetime.fromtimestamp(epoch, tz=dt.timezone.utc).isoformat().replace("+00:00", "Z")
def build_catalog(
manifest: dict[str, Any],
dist_dir: Path,
aliases: list[str],
architectures: list[str],
) -> dict[str, Any]:
entries: list[dict[str, Any]] = []
for alias in aliases:
image = find_image(manifest, alias)
for architecture in architectures:
artifact = image["artifacts"][architecture]
package = dist_dir / architecture / f"{alias}.tar.zst"
source_package = dist_dir / architecture / f"{alias}.sources.tar.zst"
if not package.is_file():
fail(f"package not found: {package}")
if not source_package.is_file():
fail(f"source package not found: {source_package}")
entries.append(
{
"alias": alias,
"distribution": image["distribution"],
"series": image["series"],
"architecture": architecture,
"version": image["templateVersion"],
"distributionVersion": image["version"],
"package": artifact["registryKey"],
"sha256": sha256(package),
"sizeBytes": package.stat().st_size,
"source": source_registry_key(manifest, alias, architecture),
"sourceSha256": sha256(source_package),
"sourceSizeBytes": source_package.stat().st_size,
"minDiskGb": image["minDiskGb"],
"publishedAt": published_at(package),
}
)
return {"schemaVersion": 1, "images": entries}
def parser() -> argparse.ArgumentParser:
repo_dir = Path(__file__).resolve().parent.parent
result = argparse.ArgumentParser(description=__doc__)
result.add_argument(
"--manifest",
type=Path,
default=repo_dir / "packaging" / "m2images.json",
help="release manifest path",
)
subparsers = result.add_subparsers(dest="command", required=True)
subparsers.add_parser("validate")
subparsers.add_parser("aliases")
subparsers.add_parser("architectures")
field = subparsers.add_parser("field")
field.add_argument("alias")
field.add_argument("field")
environment = subparsers.add_parser("environment")
environment.add_argument("alias")
requires = subparsers.add_parser("requires")
requires.add_argument("alias")
artifacts = subparsers.add_parser("artifacts")
artifacts.add_argument("alias")
artifacts.add_argument("architecture", choices=SUPPORTED_ARCHITECTURES)
registry_key = subparsers.add_parser("registry-key")
registry_key.add_argument("alias")
registry_key.add_argument("architecture", choices=SUPPORTED_ARCHITECTURES)
source_key = subparsers.add_parser("source-registry-key")
source_key.add_argument("alias")
source_key.add_argument("architecture", choices=SUPPORTED_ARCHITECTURES)
catalog = subparsers.add_parser("catalog")
catalog.add_argument("--dist-dir", type=Path, required=True)
catalog.add_argument("--alias", action="append", dest="aliases")
catalog.add_argument(
"--architecture",
action="append",
dest="architectures",
choices=SUPPORTED_ARCHITECTURES,
)
catalog.add_argument("--output", type=Path)
return result
def main() -> int:
args = parser().parse_args()
manifest = load_manifest(args.manifest)
if args.command == "validate":
print(f"valid: {args.manifest}")
elif args.command == "aliases":
for image in manifest["images"]:
print(image["alias"])
elif args.command == "architectures":
print("\n".join(manifest["architectures"]))
elif args.command == "field":
value = get_field(find_image(manifest, args.alias), args.field)
if isinstance(value, (dict, list)):
print(json.dumps(value, separators=(",", ":")))
else:
print(value)
elif args.command == "environment":
environment = find_image(manifest, args.alias)["builder"]["environment"]
for key in sorted(environment):
print(f"{key}\t{environment[key]}")
elif args.command == "requires":
print("\n".join(find_image(manifest, args.alias)["builder"]["requires"]))
elif args.command == "artifacts":
artifact = find_image(manifest, args.alias)["artifacts"][args.architecture]
for key in ("kernel", "initrd", "rootfs"):
if artifact[key]:
print(artifact[key])
elif args.command == "registry-key":
print(find_image(manifest, args.alias)["artifacts"][args.architecture]["registryKey"])
elif args.command == "source-registry-key":
print(source_registry_key(manifest, args.alias, args.architecture))
elif args.command == "catalog":
aliases = args.aliases or [image["alias"] for image in manifest["images"]]
architectures = args.architectures or manifest["architectures"]
catalog = build_catalog(manifest, args.dist_dir, aliases, architectures)
rendered = json.dumps(catalog, indent=2, ensure_ascii=False) + "\n"
if args.output:
args.output.parent.mkdir(parents=True, exist_ok=True)
temporary = args.output.with_name(f".{args.output.name}.tmp")
temporary.write_text(rendered, encoding="utf-8")
temporary.replace(args.output)
else:
print(rendered, end="")
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except ManifestError as error:
print(f"[FAIL] {error}", file=sys.stderr)
raise SystemExit(2)