171 lines
7.5 KiB
Python
171 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate the SouveraineOS distribution contract without network access."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
import tomllib
|
|
from pathlib import Path
|
|
|
|
|
|
PACKAGE_STATES = {"managed", "blocked", "planned"}
|
|
PROFILE_STATES = {"ready", "blocked", "planned"}
|
|
TARGET_STATES = {"supported", "bringup", "planned"}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("manifest", type=Path)
|
|
parser.add_argument("--ready", metavar="PROFILE", help="require one profile to be installer-ready")
|
|
parser.add_argument("--producer", metavar="NAME", help="validate artifacts emitted by one producer")
|
|
parser.add_argument(
|
|
"--complete",
|
|
action="store_true",
|
|
help="require --artifact to cover every managed artifact declared for --producer",
|
|
)
|
|
parser.add_argument(
|
|
"--artifact",
|
|
metavar="PACKAGE:ARCHITECTURE",
|
|
action="append",
|
|
default=[],
|
|
help="one package artifact emitted by --producer; repeat for every artifact",
|
|
)
|
|
args = parser.parse_args()
|
|
try:
|
|
with args.manifest.open("rb") as source:
|
|
data = tomllib.load(source)
|
|
except (OSError, tomllib.TOMLDecodeError) as error:
|
|
print(f"invalid manifest: {error}", file=sys.stderr)
|
|
return 2
|
|
|
|
errors: list[str] = []
|
|
def error(message: str) -> None:
|
|
errors.append(message)
|
|
|
|
if data.get("schema") != 1:
|
|
error("schema must be 1")
|
|
if data.get("archive", {}).get("publisher_rule") != "additive":
|
|
error("archive publisher_rule must be additive")
|
|
|
|
producers = data.get("producers", {})
|
|
packages = data.get("packages", {})
|
|
for name, package in packages.items():
|
|
if package.get("state") not in PACKAGE_STATES:
|
|
error(f"package {name}: invalid state {package.get('state')!r}")
|
|
package_producers = package.get("producers")
|
|
if package_producers is None:
|
|
package_producers = [package.get("producer")]
|
|
if not package_producers:
|
|
error(f"package {name}: no producer declared")
|
|
elif package.get("state") != "planned":
|
|
for producer in package_producers:
|
|
if producer not in producers:
|
|
error(f"package {name}: unknown producer {producer!r}")
|
|
if not package.get("architectures"):
|
|
error(f"package {name}: no architecture declared")
|
|
|
|
profiles = data.get("profiles", {})
|
|
for profile_name, profile in profiles.items():
|
|
status = profile.get("status")
|
|
architectures = set(profile.get("architectures", []))
|
|
if status not in PROFILE_STATES:
|
|
error(f"profile {profile_name}: invalid status {status!r}")
|
|
if not architectures:
|
|
error(f"profile {profile_name}: no architecture declared")
|
|
for name in profile.get("packages", []):
|
|
package = packages.get(name)
|
|
if package is None:
|
|
error(f"profile {profile_name}: unknown package {name}")
|
|
continue
|
|
if not architectures.intersection(package.get("architectures", [])):
|
|
error(f"profile {profile_name}: {name} has no matching architecture")
|
|
if status == "ready" and package.get("state") != "managed":
|
|
error(f"profile {profile_name}: ready profile includes {name} ({package.get('state')})")
|
|
for name in profile.get("host_tools", []):
|
|
package = packages.get(name)
|
|
if package is None:
|
|
error(f"profile {profile_name}: unknown host tool {name}")
|
|
elif status == "ready" and package.get("state") != "managed":
|
|
error(f"profile {profile_name}: ready profile includes host tool {name} ({package.get('state')})")
|
|
|
|
targets = data.get("targets", {})
|
|
for target_name, target in targets.items():
|
|
profile = profiles.get(target.get("profile"))
|
|
if target.get("state") not in TARGET_STATES:
|
|
error(f"target {target_name}: invalid state {target.get('state')!r}")
|
|
if profile is None:
|
|
error(f"target {target_name}: unknown profile {target.get('profile')!r}")
|
|
continue
|
|
if target.get("architecture") not in profile.get("architectures", []):
|
|
error(f"target {target_name}: architecture does not match its profile")
|
|
if target.get("state") == "supported" and profile.get("status") == "planned":
|
|
error(f"target {target_name}: supported target cannot use a planned profile")
|
|
|
|
if args.producer:
|
|
if args.producer not in producers:
|
|
error(f"unknown producer {args.producer!r}")
|
|
if not args.artifact:
|
|
error(f"producer {args.producer}: no artifacts declared")
|
|
|
|
emitted: set[tuple[str, str]] = set()
|
|
for artifact in args.artifact:
|
|
try:
|
|
package_name, architecture = artifact.rsplit(":", 1)
|
|
except ValueError:
|
|
error(f"invalid artifact {artifact!r}; expected PACKAGE:ARCHITECTURE")
|
|
continue
|
|
package = packages.get(package_name)
|
|
if package is None:
|
|
error(f"producer {args.producer}: undeclared artifact {artifact}")
|
|
continue
|
|
declared_producers = package.get("producers", [package.get("producer")])
|
|
if declared_producers != [args.producer]:
|
|
error(f"producer {args.producer}: does not own {package_name}")
|
|
if package.get("state") != "managed":
|
|
error(f"producer {args.producer}: {package_name} is {package.get('state')}, not managed")
|
|
if architecture not in package.get("architectures", []):
|
|
error(f"producer {args.producer}: {package_name} does not declare {architecture}")
|
|
if architecture not in producers.get(args.producer, {}).get("architectures", []):
|
|
error(f"producer {args.producer}: does not declare {architecture}")
|
|
emitted.add((package_name, architecture))
|
|
|
|
expected = {
|
|
(name, architecture)
|
|
for name, package in packages.items()
|
|
if package.get("state") == "managed"
|
|
and package.get("producer") == args.producer
|
|
for architecture in package.get("architectures", [])
|
|
}
|
|
if args.complete and emitted != expected:
|
|
missing = sorted(expected - emitted)
|
|
unexpected = sorted(emitted - expected)
|
|
if missing:
|
|
error(f"producer {args.producer}: missing artifacts {missing}")
|
|
if unexpected:
|
|
error(f"producer {args.producer}: unexpected artifacts {unexpected}")
|
|
|
|
if errors:
|
|
print("distribution manifest is invalid:", file=sys.stderr)
|
|
print("\n".join(f" - {item}" for item in errors), file=sys.stderr)
|
|
return 1
|
|
|
|
blocked = [name for name, profile in profiles.items() if profile.get("status") != "ready"]
|
|
print(f"distribution manifest valid: {len(targets)} targets, {len(packages)} packages, {len(profiles)} profiles")
|
|
if blocked:
|
|
print("profiles not ready: " + ", ".join(blocked))
|
|
if args.ready:
|
|
profile = profiles.get(args.ready)
|
|
if profile is None:
|
|
print(f"unknown profile: {args.ready}", file=sys.stderr)
|
|
return 2
|
|
if profile.get("status") != "ready":
|
|
print(f"profile {args.ready} is not ready for a public installer", file=sys.stderr)
|
|
return 1
|
|
if args.producer:
|
|
print(f"producer contract valid: {args.producer}, {len(args.artifact)} artifacts")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|