net-kingdom/tools/tenancy-posture/validate.py

#!/usr/bin/env python3
# /// script
# dependencies = ["jsonschema>=4.23,<5", "PyYAML>=6,<7"]
# ///
"""Validate tenancy.yaml files against the canonical syntax and semantics."""

from __future__ import annotations

import argparse
import datetime as dt
import json
import pathlib
import sys
from typing import Any

import jsonschema
import yaml


ROOT = pathlib.Path(__file__).resolve().parents[2]
SCHEMA = ROOT / "canon/schemas/tenancy-posture_v0.1.schema.json"
AXES = ("I", "A", "E", "P", "R", "V")
FLOORS = {"I": 1, "A": 1, "E": 0, "P": 0, "R": 1, "V": 0}


def services(document: dict[str, Any]) -> list[dict[str, Any]]:
    if "services" in document:
        return document["services"]
    return [document]


def validate_semantics(document: dict[str, Any], path: pathlib.Path) -> list[str]:
    errors: list[str] = []
    entries = services(document)
    names = [entry["service"] for entry in entries]
    if len(names) != len(set(names)):
        errors.append("service names must be unique")

    for entry in entries:
        name = entry["service"]
        workload_identity = entry.get("workload_identity")
        if workload_identity:
            if workload_identity["name"] != name:
                errors.append(
                    f"{name}: workload_identity.name must equal service"
                )
            bindings = workload_identity["identity_bindings"]
            binding_keys = [
                (
                    binding["scheme"],
                    binding["authority"],
                    binding["subject"],
                    binding.get("environment"),
                )
                for binding in bindings
            ]
            if len(binding_keys) != len(set(binding_keys)):
                errors.append(f"{name}: workload identity bindings must be unique")
        zones = entry.get("zones")
        if zones:
            zone_reviewed = dt.date.fromisoformat(zones["reviewed"])
            zone_review_due = dt.date.fromisoformat(zones["review_due"])
            if zone_review_due <= zone_reviewed:
                errors.append(f"{name}: zones.review_due must be after zones.reviewed")
        posture = entry["tenancy"]
        current = posture["current"]
        reason = posture.get("reason", {})
        gap = posture.get("gap", {})
        evidence = entry.get("evidence", {})
        evidence_freshness = entry.get("evidence_freshness", {})

        reviewed = dt.date.fromisoformat(posture["reviewed"])
        review_due = dt.date.fromisoformat(posture["review_due"])
        if review_due < reviewed:
            errors.append(f"{name}: review_due precedes reviewed")

        for axis in AXES:
            level = current[axis]
            explained = bool(reason.get(axis) or gap.get(axis))
            if level == "n/a" or (isinstance(level, int) and level <= FLOORS[axis]):
                if not explained:
                    errors.append(f"{name}: {axis}{level} needs reason or gap text")
                continue
            key = f"{axis}{level}"
            if key not in evidence:
                errors.append(f"{name}: current {key} has no evidence entry")

        implemented = posture.get("implemented", {})
        for axis, level in implemented.items():
            current_level = current[axis]
            if current_level != "n/a" and level != "n/a" and level <= current_level:
                errors.append(
                    f"{name}: implemented {axis}{level} must be above current {axis}{current_level}"
                )

        for key, freshness in evidence_freshness.items():
            if key not in evidence:
                errors.append(f"{name}: evidence_freshness {key} has no evidence entry")
            observed_at = dt.datetime.fromisoformat(
                freshness["observed_at"].replace("Z", "+00:00")
            )
            valid_until_raw = freshness.get("valid_until")
            if valid_until_raw:
                valid_until = dt.datetime.fromisoformat(
                    valid_until_raw.replace("Z", "+00:00")
                )
                if valid_until <= observed_at:
                    errors.append(
                        f"{name}: evidence_freshness {key} valid_until must be after observed_at"
                    )

        provider = entry.get("provider", {})
        for axis, reach in provider.get("axes", {}).items():
            available = reach["available"]
            maximum = reach["maximum"]
            if available != "n/a" and maximum != "n/a" and available > maximum:
                errors.append(
                    f"{name}: provider {axis} available {available} exceeds maximum {maximum}"
                )

    return [f"{path}: {error}" for error in errors]


def validate(path: pathlib.Path, schema: dict[str, Any]) -> list[str]:
    document = yaml.safe_load(path.read_text(encoding="utf-8"))
    validator = jsonschema.Draft202012Validator(
        schema, format_checker=jsonschema.FormatChecker()
    )
    errors = [
        f"{path}: {'/'.join(str(part) for part in error.absolute_path) or '<root>'}: {error.message}"
        for error in sorted(validator.iter_errors(document), key=lambda item: list(item.path))
    ]
    if not errors:
        errors.extend(validate_semantics(document, path))
    return errors


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("declarations", nargs="+", type=pathlib.Path)
    args = parser.parse_args()
    schema = json.loads(SCHEMA.read_text(encoding="utf-8"))
    errors = [error for path in args.declarations for error in validate(path, schema)]
    if errors:
        print("\n".join(errors), file=sys.stderr)
        return 1
    for path in args.declarations:
        print(f"{path}: valid")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())