lazaret/tools/lazaret_dev.py

#!/usr/bin/env python3
"""Repository-owned generation, policy, probe, model, and verification commands."""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import platform
import pwd
import re
import shutil
import signal
import stat
import subprocess
import sys
import tempfile
import tomllib

ROOT = Path(__file__).resolve().parent.parent
SUPPORT = ROOT / "support"
LEVELS = ("V1", "V2", "V3", "V4", "V5")


class RegistryError(RuntimeError):
    pass


def load_toml(path: Path) -> dict:
    with path.open("rb") as source:
        return tomllib.load(source)


def require(condition: bool, message: str) -> None:
    if not condition:
        raise RegistryError(message)


def validate_fields(item: object, required: set[str], optional: set[str], owner: str) -> dict:
    require(isinstance(item, dict), f"{owner}: entry must be a table")
    fields = set(item)
    missing = required - fields
    extra = fields - required - optional
    require(not missing, f"{owner}: missing fields {sorted(missing)}")
    require(not extra, f"{owner}: unexpected fields {sorted(extra)}")
    return item


def validate_levels(levels: object, owner: str) -> None:
    require(isinstance(levels, list) and levels, f"{owner}: levels must be nonempty")
    require(all(isinstance(level, str) for level in levels), f"{owner}: levels must be strings")
    require(len(levels) == len(set(levels)), f"{owner}: duplicate levels")
    require(all(level in LEVELS for level in levels), f"{owner}: unknown level")


def validate_packages(data: dict) -> list[dict]:
    require(data.get("schema") == 1, "work-package registry schema must be 1")
    packages = data.get("package")
    require(isinstance(packages, list) and packages, "work-package registry is empty")
    by_id = {}
    for package in packages:
        package_id = package.get("id")
        require(isinstance(package_id, str) and re.fullmatch(r"IP-\d{2}", package_id) is not None, "invalid package ID")
        require(package_id not in by_id, f"duplicate package {package_id}")
        require(isinstance(package.get("title"), str) and package["title"], f"{package_id}: title is empty")
        require(isinstance(package.get("depends_on"), list), f"{package_id}: depends_on must be a list")
        validate_levels(package.get("levels"), package_id)
        by_id[package_id] = package
    for package_id, package in by_id.items():
        dependencies = package["depends_on"]
        require(len(dependencies) == len(set(dependencies)), f"{package_id}: duplicate dependency")
        for dependency in dependencies:
            require(dependency in by_id, f"{package_id}: unknown dependency {dependency}")
            require(dependency != package_id, f"{package_id}: self dependency")
    topological_packages(packages)
    return packages


def topological_packages(packages: list[dict]) -> list[dict]:
    by_id = {package["id"]: package for package in packages}
    indegree = {package_id: 0 for package_id in by_id}
    consumers = {package_id: [] for package_id in by_id}
    for package_id, package in by_id.items():
        for dependency in package["depends_on"]:
            indegree[package_id] += 1
            consumers[dependency].append(package_id)
    ready = sorted(package_id for package_id, degree in indegree.items() if degree == 0)
    ordered = []
    while ready:
        package_id = ready.pop(0)
        ordered.append(by_id[package_id])
        for consumer in sorted(consumers[package_id]):
            indegree[consumer] -= 1
            if indegree[consumer] == 0:
                ready.append(consumer)
                ready.sort()
    require(len(ordered) == len(packages), "work-package dependency cycle")
    return ordered


def operator_blocks(source: str) -> dict[str, str]:
    definitions = list(re.finditer(r"(?m)^([A-Za-z][A-Za-z0-9_]*)(?:\([^\n=]*\))? ==", source))
    blocks = {}
    for index, definition in enumerate(definitions):
        end = definitions[index + 1].start() if index + 1 < len(definitions) else len(source)
        blocks[definition.group(1)] = source[definition.end():end]
    return blocks


def action_symbols(source: str, aggregates: list[str]) -> set[str]:
    blocks = operator_blocks(source)
    aggregate_set = set(aggregates)
    require("Next" in aggregate_set, "action aggregates must include Next")
    found = set()
    pending = ["Next"]
    visited = set()
    while pending:
        aggregate = pending.pop()
        require(aggregate in blocks, f"missing aggregate operator {aggregate}")
        if aggregate in visited:
            continue
        visited.add(aggregate)
        body = blocks[aggregate]
        calls = set(re.findall(r"\b([A-Z][A-Za-z0-9_]*)\s*\(", body))
        calls.update(re.findall(r"(?m)^\s*\\/\s+([A-Z][A-Za-z0-9_]*)\s*$", body))
        for symbol in calls:
            if symbol in aggregate_set:
                pending.append(symbol)
            elif symbol in blocks:
                found.add(symbol)
    require(visited == aggregate_set, f"unused action aggregates: {sorted(aggregate_set - visited)}")
    return found


def configured_symbols(source: str, section: str) -> set[str]:
    match = re.search(rf"(?ms)^{section}\s*$\n(.*?)(?=^[A-Z]+\s*$|\Z)", source)
    if match is None:
        return set()
    return set(re.findall(r"(?m)^\s+([A-Za-z][A-Za-z0-9_]*)\s*$", match.group(1)))


def validate_refinements(data: dict, package_ids: set[str]) -> None:
    require(data.get("schema") == 1, "model-refinement registry schema must be 1")
    models = data.get("model")
    coverage = data.get("coverage")
    require(isinstance(models, list) and models, "model registry is empty")
    require(isinstance(coverage, list) and coverage, "model coverage is empty")
    by_name = {}
    expected = {}
    for model in models:
        name = model.get("name")
        require(isinstance(name, str) and name not in by_name, "duplicate or invalid model name")
        module = ROOT / model.get("module", "")
        config = ROOT / model.get("config", "")
        require(module.is_file(), f"{name}: missing model module")
        require(config.is_file(), f"{name}: missing model configuration")
        aggregates = model.get("action_aggregates")
        require(isinstance(aggregates, list) and aggregates, f"{name}: action aggregates are empty")
        module_source = module.read_text()
        config_source = config.read_text()
        expected[(name, "action")] = action_symbols(module_source, aggregates)
        expected[(name, "invariant")] = configured_symbols(config_source, "INVARIANTS")
        expected[(name, "property")] = configured_symbols(config_source, "PROPERTIES")
        by_name[name] = model
    actual = {key: set() for key in expected}
    for group in coverage:
        model = group.get("model")
        kind = group.get("kind")
        key = (model, kind)
        require(key in actual, f"unknown model coverage kind {model}:{kind}")
        symbols = group.get("symbols")
        owners = group.get("owners")
        require(isinstance(symbols, list) and symbols, f"{model}:{kind}: symbols are empty")
        require(isinstance(owners, list) and owners, f"{model}:{kind}: owners are empty")
        require(all(owner in package_ids for owner in owners), f"{model}:{kind}: unknown owner")
        validate_levels(group.get("levels"), f"{model}:{kind}")
        require(isinstance(group.get("verifier"), str) and group["verifier"], f"{model}:{kind}: verifier is empty")
        for symbol in symbols:
            require(symbol not in actual[key], f"duplicate refinement {model}!{symbol}")
            actual[key].add(symbol)
    for key, symbols in expected.items():
        missing = symbols - actual[key]
        extra = actual[key] - symbols
        require(not missing and not extra, f"{key[0]} {key[1]} coverage mismatch; missing={sorted(missing)}, extra={sorted(extra)}")


def validate_external_tools(data: dict) -> None:
    validate_fields(data, {"schema", "levels", "command", "feature"}, set(), "external manifest")
    require(data.get("schema") == 1, "external-tool manifest schema must be 1")
    require(data.get("levels") == list(LEVELS), "external-tool levels must be V1 through V5")
    commands = data.get("command")
    features = data.get("feature")
    require(isinstance(commands, list) and commands, "external command manifest is empty")
    require(isinstance(features, list) and features, "external feature manifest is empty")
    ids = set()
    for item in commands + features:
        require(isinstance(item, dict), "external capability entry must be a table")
        item_id = item.get("id")
        require(isinstance(item_id, str) and item_id not in ids, "duplicate or invalid external capability ID")
        ids.add(item_id)
        require(item.get("policy") in {"fail", "skip"}, f"{item_id}: policy must be fail or skip")
        validate_levels(item.get("levels"), item_id)
        require(isinstance(item.get("diagnostic"), str) and item["diagnostic"], f"{item_id}: diagnostic is empty")
    required = {
        "rustc",
        "cargo",
        "cargo-deny",
        "tlc",
        "runsc",
        "fuse-overlayfs",
        "slirp4netns",
        "linux-x86_64",
        "kernel",
        "user-namespaces",
        "subordinate-ids",
        "fuse-device",
    }
    require(required <= ids, f"external manifest missing {sorted(required - ids)}")
    for command in commands:
        command_id = command["id"]
        validate_fields(
            command,
            {"id", "program", "args", "version_pattern", "policy", "levels", "diagnostic"},
            {"accepted_exit_codes"},
            command_id,
        )
        require(isinstance(command.get("program"), str) and command["program"], f"{command['id']}: program is empty")
        require(isinstance(command.get("args"), list) and all(isinstance(arg, str) for arg in command["args"]), f"{command['id']}: args must be strings")
        exit_codes = command.get("accepted_exit_codes", [0])
        require(isinstance(exit_codes, list) and exit_codes and all(isinstance(code, int) and not isinstance(code, bool) for code in exit_codes), f"{command['id']}: invalid accepted exit codes")
        pattern = command.get("version_pattern")
        require(isinstance(pattern, str), f"{command['id']}: version pattern is missing")
        try:
            re.compile(pattern)
        except re.error as error:
            raise RegistryError(f"{command['id']}: invalid version pattern: {error}") from error
    common = {"id", "probe", "policy", "levels", "diagnostic"}
    probe_fields = {
        "linux-x86_64": set(),
        "kernel-at-least": {"minimum"},
        "user-namespace": {"program", "sysctl_path"},
        "subordinate-ids": {"minimum_count", "uidmap_program", "gidmap_program", "unshare_program"},
        "open-character-device": {"path", "access"},
    }
    for feature in features:
        feature_id = feature["id"]
        probe_name = feature.get("probe")
        require(probe_name in probe_fields, f"{feature_id}: unknown feature probe")
        validate_fields(feature, common | probe_fields[probe_name], set(), feature_id)
        if probe_name == "kernel-at-least":
            minimum = feature["minimum"]
            require(isinstance(minimum, str) and re.fullmatch(r"\d+(?:\.\d+)+", minimum) is not None, f"{feature_id}: minimum must be a dotted numeric version")
        elif probe_name == "user-namespace":
            require(isinstance(feature["program"], str) and feature["program"], f"{feature_id}: program is empty")
            require(isinstance(feature["sysctl_path"], str) and Path(feature["sysctl_path"]).is_absolute(), f"{feature_id}: sysctl_path must be absolute")
        elif probe_name == "subordinate-ids":
            count = feature["minimum_count"]
            require(isinstance(count, int) and not isinstance(count, bool) and count >= 65536, f"{feature_id}: minimum_count must be at least 65536")
            for field in ("uidmap_program", "gidmap_program", "unshare_program"):
                require(isinstance(feature[field], str) and feature[field], f"{feature_id}: {field} is empty")
        elif probe_name == "open-character-device":
            require(isinstance(feature["path"], str) and Path(feature["path"]).is_absolute(), f"{feature_id}: path must be absolute")
            require(feature["access"] == "read-write", f"{feature_id}: access must be read-write")


def validate_integrations(data: dict, capability_ids: set[str]) -> None:
    require(data.get("schema") == 1, "integration manifest schema must be 1")
    require(data.get("levels") == list(LEVELS[1:]), "integration levels must be V2 through V5")
    scenarios = data.get("scenarios")
    require(isinstance(scenarios, list), "integration scenarios must be a list")
    names = set()
    for scenario in scenarios:
        name = scenario.get("name")
        require(isinstance(name, str) and name and name not in names, "duplicate or invalid integration scenario")
        names.add(name)
        require(scenario.get("level") in LEVELS[1:], f"{name}: invalid integration level")
        command = scenario.get("command")
        require(isinstance(command, list) and command and all(isinstance(part, str) for part in command), f"{name}: invalid command")
        capabilities = scenario.get("capabilities", [])
        require(isinstance(capabilities, list), f"{name}: capabilities must be a list")
        require(all(capability in capability_ids for capability in capabilities), f"{name}: unknown capability")


def validated_data() -> tuple[list[dict], dict, dict]:
    packages = validate_packages(load_toml(SUPPORT / "work-packages.toml"))
    package_ids = {package["id"] for package in packages}
    refinements = load_toml(SUPPORT / "model-refinements.toml")
    validate_refinements(refinements, package_ids)
    external = load_toml(SUPPORT / "external-tools.toml")
    validate_external_tools(external)
    capability_ids = {item["id"] for kind in ("command", "feature") for item in external[kind]}
    integrations = load_toml(SUPPORT / "integration-tests.toml")
    validate_integrations(integrations, capability_ids)
    return packages, external, integrations


def dependency_view(packages: list[dict]) -> str:
    ordered = topological_packages(packages)
    lines = [
        "# Implementation dependencies",
        "",
        "Generated from `support/work-packages.toml`; do not edit this file directly.",
        "",
        "| Package | Deliverable | Depends on | Verification |",
        "| --- | --- | --- | --- |",
    ]
    for package in ordered:
        dependencies = ", ".join(f"`{item}`" for item in package["depends_on"]) or "—"
        levels = ", ".join(package["levels"])
        lines.append(f"| `{package['id']}` | {package['title']} | {dependencies} | {levels} |")
    lines.extend(["", "```mermaid", "graph TD"])
    for package in ordered:
        label = package["title"].replace('"', "'")
        lines.append(f"    {package['id'].replace('-', '_')}[\"{package['id']} {label}\"]")
    for package in ordered:
        for dependency in package["depends_on"]:
            lines.append(f"    {dependency.replace('-', '_')} --> {package['id'].replace('-', '_')}")
    lines.extend(["```", ""])
    return "\n".join(lines)


def generate(check: bool) -> None:
    packages, _, _ = validated_data()
    target = ROOT / "docs/implementation-dependencies.md"
    content = dependency_view(packages)
    if check:
        require(target.is_file(), f"generated file missing: {target.relative_to(ROOT)}")
        require(target.read_text() == content, f"generated file drift: {target.relative_to(ROOT)}")
    else:
        target.write_text(content)


def version_tuple(value: str) -> tuple[int, ...]:
    return tuple(int(part) for part in value.split("."))


def subordinate_id_range(path: Path, principals: set[str], minimum_count: int) -> int | None:
    try:
        rows = path.read_text().splitlines()
    except OSError:
        return None
    for row in rows:
        fields = row.split(":")
        if len(fields) != 3 or fields[0] not in principals:
            continue
        try:
            start = int(fields[1])
            count = int(fields[2])
        except ValueError:
            continue
        if start > 0 and count >= minimum_count:
            return start
    return None


def functional_user_namespace(program: str) -> bool:
    executable = shutil.which(program)
    if executable is None:
        return False
    try:
        result = subprocess.run(
            [executable, "--user", "--map-root-user", "true"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=5,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return False
    return result.returncode == 0


def mapping_helpers_usable(feature: dict, uid_start: int, gid_start: int) -> bool:
    unshare = shutil.which(feature["unshare_program"])
    newuidmap = shutil.which(feature["uidmap_program"])
    newgidmap = shutil.which(feature["gidmap_program"])
    if None in {unshare, newuidmap, newgidmap}:
        return False
    try:
        process = subprocess.Popen(
            [unshare, "--user", "--", "sh", "-c", "kill -STOP $$; exec sleep 5"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
    except OSError:
        return False
    try:
        try:
            _, status = os.waitpid(process.pid, os.WUNTRACED)
        except OSError:
            return False
        if not os.WIFSTOPPED(status):
            return False
        count = feature["minimum_count"]
        uid_result = subprocess.run(
            [newuidmap, str(process.pid), "0", str(os.getuid()), "1", "1", str(uid_start), str(count)],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
        gid_result = subprocess.run(
            [newgidmap, str(process.pid), "0", str(os.getgid()), "1", "1", str(gid_start), str(count)],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
        return uid_result.returncode == 0 and gid_result.returncode == 0
    finally:
        if process.poll() is None:
            os.kill(process.pid, signal.SIGKILL)
        process.wait()


def subordinate_ids_available(feature: dict) -> bool:
    try:
        user = pwd.getpwuid(os.getuid())
    except KeyError:
        return False
    principals = {user.pw_name, str(user.pw_uid)}
    minimum_count = feature["minimum_count"]
    uid_start = subordinate_id_range(Path("/etc/subuid"), principals, minimum_count)
    gid_start = subordinate_id_range(Path("/etc/subgid"), principals, minimum_count)
    if uid_start is None or gid_start is None:
        return False
    return mapping_helpers_usable(feature, uid_start, gid_start)


def probe_feature(feature: dict) -> tuple[bool, str]:
    probe = feature["probe"]
    if probe == "linux-x86_64":
        passed = platform.system() == "Linux" and platform.machine() == "x86_64"
        return passed, f"{platform.system()} {platform.machine()}"
    if probe == "kernel-at-least":
        release = platform.release().split("-", 1)[0]
        try:
            passed = platform.system() == "Linux" and version_tuple(release) >= version_tuple(feature["minimum"])
        except ValueError:
            passed = False
        return passed, release
    if probe == "user-namespace":
        try:
            value = int(Path(feature["sysctl_path"]).read_text().strip())
        except (OSError, ValueError):
            value = 0
        passed = value > 0 and functional_user_namespace(feature["program"])
        return passed, "functional" if passed else "unavailable or nonfunctional"
    if probe == "subordinate-ids":
        passed = subordinate_ids_available(feature)
        return passed, "configured and usable" if passed else "unavailable or unusable"
    if probe == "open-character-device":
        try:
            path = Path(feature["path"])
            passed = stat.S_ISCHR(path.stat().st_mode)
            descriptor = os.open(path, os.O_RDWR | os.O_CLOEXEC | os.O_NONBLOCK) if passed else None
        except OSError:
            passed = False
            descriptor = None
        if descriptor is not None:
            os.close(descriptor)
        return passed, "openable read-write" if passed else "unavailable or inaccessible"
    raise RegistryError(f"unknown feature probe {probe}")


def probe(level: str, external: dict | None = None) -> tuple[set[str], set[str]]:
    if external is None:
        _, external, _ = validated_data()
    available = set()
    skipped = set()
    failed = False
    for command in external["command"]:
        if level not in command["levels"]:
            continue
        program = shutil.which(command["program"])
        output = ""
        passed = program is not None
        if passed:
            result = subprocess.run([program, *command["args"]], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False)
            output = result.stdout.strip()
            passed = result.returncode in command.get("accepted_exit_codes", [0]) and re.search(command["version_pattern"], output) is not None
        if passed:
            available.add(command["id"])
            print(f"PASS {command['id']}: {output.splitlines()[0]}")
        else:
            status = command["policy"].upper()
            print(f"{status} {command['id']}: {command['diagnostic']}")
            if command["policy"] == "fail":
                failed = True
            else:
                skipped.add(command["id"])
    for feature in external["feature"]:
        if level not in feature["levels"]:
            continue
        passed, detail = probe_feature(feature)
        if passed:
            available.add(feature["id"])
            print(f"PASS {feature['id']}: {detail}")
        else:
            status = feature["policy"].upper()
            print(f"{status} {feature['id']}: {feature['diagnostic']}")
            if feature["policy"] == "fail":
                failed = True
            else:
                skipped.add(feature["id"])
    require(not failed, f"{level} external capability probe failed")
    return available, skipped


def run(command: list[str], cwd: Path = ROOT) -> None:
    print(f"RUN {' '.join(command)}")
    subprocess.run(command, cwd=cwd, check=True)


def dependency_audit() -> None:
    policy = load_toml(ROOT / "deny.toml")
    allowed_licenses = set(policy["licenses"]["allow"])
    result = subprocess.run(["cargo", "metadata", "--locked", "--offline", "--format-version", "1"], cwd=ROOT, text=True, stdout=subprocess.PIPE, check=True)
    metadata = json.loads(result.stdout)
    workspace_members = set(metadata["workspace_members"])
    for package in metadata["packages"]:
        if package["id"] in workspace_members:
            continue
        source = package.get("source") or ""
        require(source.startswith("registry+https://github.com/rust-lang/crates.io-index"), f"unapproved dependency source: {package['name']}")
        license_expression = package.get("license") or ""
        tokens = set(re.findall(r"[A-Za-z0-9.-]+", license_expression)) - {"AND", "OR", "WITH"}
        require(bool(tokens) and tokens <= allowed_licenses, f"unapproved dependency license: {package['name']} {license_expression}")
    for package in metadata["packages"]:
        for dependency in package["dependencies"]:
            source = dependency.get("source")
            require(source is None or source.startswith("registry+https://github.com/rust-lang/crates.io-index"), f"git or unknown dependency: {package['name']} -> {dependency['name']}")
            if source is None:
                dependency_path = dependency.get("path")
                require(dependency_path is not None and Path(dependency_path).resolve().is_relative_to(ROOT), f"external path dependency: {dependency['name']}")
            else:
                require(dependency["req"] != "*", f"wildcard dependency: {package['name']} -> {dependency['name']}")
    run(["cargo", "deny", "--frozen", "check", "bans", "licenses", "sources"])
    print("PASS dependency source, wildcard, and license audit")


def refresh_advisories() -> None:
    run(["cargo", "deny", "fetch"])
    run(["cargo", "deny", "--frozen", "check", "advisories"])


def run_models() -> None:
    refinements = load_toml(SUPPORT / "model-refinements.toml")
    with tempfile.TemporaryDirectory(prefix="lazaret-tlc-") as directory:
        for model in refinements["model"]:
            module = ROOT / model["module"]
            config = ROOT / model["config"]
            metadata = Path(directory) / model["name"]
            run(["tlc", "-workers", "1", "-metadir", str(metadata), "-config", str(config), str(module)], Path(directory))


def run_integrations(level: str, external: dict, integrations: dict) -> None:
    scenarios = [scenario for scenario in integrations["scenarios"] if scenario["level"] == level]
    if not scenarios:
        print(f"SKIP {level} integration harness: no scenarios registered")
        return
    available, skipped = probe(level, external)
    by_id = {item["id"]: item for kind in ("command", "feature") for item in external[kind]}
    for scenario in scenarios:
        missing = set(scenario.get("capabilities", [])) - available
        if missing:
            require(missing <= skipped and all(by_id[item]["policy"] == "skip" for item in missing), f"{scenario['name']}: undeclared capability failure")
            print(f"SKIP {scenario['name']}: missing {', '.join(sorted(missing))}")
            continue
        run(scenario["command"])


def verify(level: str) -> None:
    _, external, integrations = validated_data()
    generate(check=True)
    probe("V1", external)
    run(["cargo", "fmt", "--all", "--", "--check"])
    run(["cargo", "clippy", "--offline", "--workspace", "--all-targets", "--all-features", "--locked", "--", "-D", "warnings"])
    run(["cargo", "test", "--offline", "--workspace", "--all-features", "--locked"])
    run(["cargo", "test", "--offline", "--workspace", "--all-features", "--locked", "--doc"])
    run([sys.executable, "-m", "unittest", "discover", "-s", "tests/tooling"])
    dependency_audit()
    run_models()
    target_index = LEVELS.index(level)
    for current in LEVELS[1:target_index + 1]:
        probe(current, external)
        run_integrations(current, external, integrations)


def main() -> int:
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command", required=True)
    generate_parser = subparsers.add_parser("generate")
    generate_parser.add_argument("--check", action="store_true")
    probe_parser = subparsers.add_parser("probe")
    probe_parser.add_argument("level", choices=LEVELS)
    verify_parser = subparsers.add_parser("verify")
    verify_parser.add_argument("level", choices=LEVELS)
    subparsers.add_parser("validate")
    subparsers.add_parser("dependencies")
    subparsers.add_parser("refresh-advisories")
    subparsers.add_parser("models")
    args = parser.parse_args()
    try:
        if args.command == "generate":
            generate(args.check)
        elif args.command == "probe":
            probe(args.level)
        elif args.command == "verify":
            verify(args.level)
        elif args.command == "validate":
            validated_data()
        elif args.command == "dependencies":
            dependency_audit()
        elif args.command == "refresh-advisories":
            validated_data()
            refresh_advisories()
        elif args.command == "models":
            validated_data()
            run_models()
    except (RegistryError, OSError, subprocess.CalledProcessError) as error:
        print(f"ERROR: {error}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())