net-kingdom/tools/posture-feedback/posture_feedback.py

#!/usr/bin/env python3
# /// script
# dependencies = ["jsonschema>=4.23,<5", "PyYAML>=6,<7"]
# ///
"""Emit deterministic, proposal-only feedback for tenancy posture declarations."""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import importlib.util
import json
import pathlib
import sys
from collections import Counter
from typing import Any

import jsonschema
import yaml


ROOT = pathlib.Path(__file__).resolve().parents[2]
TENANCY_SCHEMA = ROOT / "canon/schemas/tenancy-posture_v0.1.schema.json"
REPORT_SCHEMA = ROOT / "canon/schemas/posture-feedback-report_v0.1.schema.json"
TENANCY_VALIDATOR_PATH = ROOT / "tools/tenancy-posture/validate.py"
API_VERSION = "netkingdom.io/posture-feedback/v0.1"
KIND = "PostureFeedbackReport"
ADVERSARIAL_LEVELS = {"E2", "R4", "V2", "V3", "V4"}
SEVERITY_RANK = {"low": 1, "medium": 2, "high": 3}

SPEC = importlib.util.spec_from_file_location(
    "tenancy_posture_validator_for_feedback", TENANCY_VALIDATOR_PATH
)
assert SPEC and SPEC.loader
TENANCY_VALIDATOR = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(TENANCY_VALIDATOR)


def parse_timestamp(value: str) -> dt.datetime:
    parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        raise ValueError("timestamp must include an explicit UTC offset")
    return parsed.astimezone(dt.timezone.utc)


def format_timestamp(value: dt.datetime) -> str:
    return value.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z")


def load_document(path: pathlib.Path) -> dict[str, Any]:
    document = yaml.safe_load(path.read_text(encoding="utf-8"))
    if not isinstance(document, dict):
        raise ValueError("document must be a YAML object")
    return document


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


def source_identifier(path: pathlib.Path) -> str:
    """Return a portable declaration id without treating the path as ownership."""
    resolved = path.resolve()
    try:
        return resolved.relative_to(ROOT.parent).as_posix()
    except ValueError:
        return resolved.as_posix()


def finding_id(
    source: str,
    service: str,
    finding_class: str,
    control: str,
    due: str | None,
) -> str:
    identity = "\x1f".join((source, service, finding_class, control, due or ""))
    digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:16]
    return f"NKFB-{digest}"


def finding(
    *,
    source: str,
    service: str,
    finding_class: str,
    severity: str,
    control: str,
    owner: str,
    evidence_state: str,
    reason: str,
    recommended_action: str,
    due: str | None = None,
) -> dict[str, Any]:
    result = {
        "id": finding_id(source, service, finding_class, control, due),
        "class": finding_class,
        "severity": severity,
        "source": source,
        "service": service,
        "control": control,
        "owner": owner,
        "evidence_state": evidence_state,
        "reason": reason,
        "recommended_action": recommended_action,
    }
    if due is not None:
        result["due"] = due
    return result


def review_finding(
    *,
    source: str,
    service: str,
    owner: str,
    control: str,
    due: dt.date,
    as_of: dt.datetime,
    horizon_days: int,
    prefix: str,
) -> dict[str, Any] | None:
    due_text = due.isoformat()
    if as_of.date() > due:
        return finding(
            source=source,
            service=service,
            finding_class=f"{prefix}-review-overdue",
            severity="high",
            control=control,
            owner=owner,
            evidence_state="overdue",
            due=due_text,
            reason=f"{control} review was due on {due_text} and is overdue at {format_timestamp(as_of)}.",
            recommended_action=f"Review {control}, update its declaration and evidence, and retain the review record.",
        )
    days = (due - as_of.date()).days
    if days <= horizon_days:
        return finding(
            source=source,
            service=service,
            finding_class=f"{prefix}-review-due-soon",
            severity="medium",
            control=control,
            owner=owner,
            evidence_state="due-soon",
            due=due_text,
            reason=f"{control} review is due in {days} day(s), within the {horizon_days}-day horizon.",
            recommended_action=f"Schedule and evidence the {control} review before {due_text}.",
        )
    return None


def current_evidence_keys(entry: dict[str, Any]) -> set[str]:
    current = entry["tenancy"]["current"]
    return {
        f"{axis}{level}"
        for axis, level in current.items()
        if isinstance(level, int)
    }


def evidence_findings(
    *,
    source: str,
    entry: dict[str, Any],
    as_of: dt.datetime,
    horizon_days: int,
) -> list[dict[str, Any]]:
    service = entry["service"]
    freshness = entry.get("evidence_freshness", {})
    current_keys = current_evidence_keys(entry)
    results: list[dict[str, Any]] = []

    for key in sorted(ADVERSARIAL_LEVELS & current_keys):
        if key not in freshness:
            results.append(
                finding(
                    source=source,
                    service=service,
                    finding_class="evidence-freshness-unknown",
                    severity="high",
                    control=key,
                    owner="unknown",
                    evidence_state="unknown",
                    reason=f"Current adversarial claim {key} has no authoritative evidence_freshness entry.",
                    recommended_action=(
                        f"Declare the {key} observation, expiry, bounded scope, responsible repository, "
                        "and replacement action; do not infer freshness from prose."
                    ),
                )
            )

    for key in sorted(current_keys & set(freshness)):
        metadata = freshness[key]
        valid_until_raw = metadata.get("valid_until")
        if not valid_until_raw:
            continue
        valid_until = parse_timestamp(valid_until_raw)
        owner = metadata["responsible_repo"]
        if as_of > valid_until:
            results.append(
                finding(
                    source=source,
                    service=service,
                    finding_class="evidence-expired",
                    severity="high",
                    control=key,
                    owner=owner,
                    evidence_state="expired",
                    due=format_timestamp(valid_until),
                    reason=f"Evidence for {key} expired before {format_timestamp(as_of)}.",
                    recommended_action=metadata["remediation"],
                )
            )
            continue
        remaining = valid_until - as_of
        if remaining <= dt.timedelta(days=horizon_days):
            results.append(
                finding(
                    source=source,
                    service=service,
                    finding_class="evidence-due-soon",
                    severity="medium",
                    control=key,
                    owner=owner,
                    evidence_state="due-soon",
                    due=format_timestamp(valid_until),
                    reason=(
                        f"Evidence for {key} expires within the {horizon_days}-day horizon "
                        f"at {format_timestamp(valid_until)}."
                    ),
                    recommended_action=metadata["remediation"],
                )
            )
    return results


def evaluate_entry(
    source: str,
    entry: dict[str, Any],
    as_of: dt.datetime,
    horizon_days: int,
) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []
    service = entry["service"]
    posture = entry["tenancy"]
    posture_owner = entry.get("responsible_repo", "unknown")

    posture_review = review_finding(
        source=source,
        service=service,
        owner=posture_owner,
        control="tenancy-posture",
        due=dt.date.fromisoformat(posture["review_due"]),
        as_of=as_of,
        horizon_days=horizon_days,
        prefix="posture",
    )
    if posture_review:
        results.append(posture_review)

    zones = entry.get("zones")
    if zones:
        zone_review = review_finding(
            source=source,
            service=service,
            owner=zones["responsible_party"],
            control="security-zones",
            due=dt.date.fromisoformat(zones["review_due"]),
            as_of=as_of,
            horizon_days=horizon_days,
            prefix="zone",
        )
        if zone_review:
            results.append(zone_review)

    results.extend(
        evidence_findings(
            source=source,
            entry=entry,
            as_of=as_of,
            horizon_days=horizon_days,
        )
    )

    current = posture["current"]
    for axis, level in sorted(posture.get("implemented", {}).items()):
        current_level = current[axis]
        results.append(
            finding(
                source=source,
                service=service,
                finding_class="implemented-not-evidenced",
                severity="medium",
                control=f"{axis}{level}",
                owner=posture_owner,
                evidence_state="implemented",
                reason=f"{axis}{level} is implemented while the current evidenced level is {axis}{current_level}.",
                recommended_action=f"Produce the canon-required evidence for {axis}{level} or remove the implemented claim.",
            )
        )

    for axis, description in sorted(posture.get("gap", {}).items()):
        results.append(
            finding(
                source=source,
                service=service,
                finding_class="declared-gap",
                severity="low",
                control=axis,
                owner=posture_owner,
                evidence_state="gap",
                reason=description,
                recommended_action=f"Review the declared {axis}-axis gap and retain, schedule, or resolve it explicitly.",
            )
        )
    return results


def report_digest(report_without_digest: dict[str, Any]) -> str:
    encoded = json.dumps(
        report_without_digest,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return f"sha256:{hashlib.sha256(encoded).hexdigest()}"


def build_report(
    paths: list[pathlib.Path],
    *,
    as_of: dt.datetime,
    horizon_days: int,
    fail_on: str,
) -> dict[str, Any]:
    tenancy_schema = json.loads(TENANCY_SCHEMA.read_text(encoding="utf-8"))
    validation_errors: list[str] = []
    findings: list[dict[str, Any]] = []

    for path in paths:
        source = source_identifier(path)
        try:
            document = load_document(path)
        except Exception as exc:
            validation_errors.append(f"{path}: {exc}")
            continue
        errors = TENANCY_VALIDATOR.validate(path, tenancy_schema)
        if errors:
            validation_errors.extend(errors)
            continue
        for entry in service_entries(document):
            findings.extend(evaluate_entry(source, entry, as_of, horizon_days))

    findings.sort(
        key=lambda item: (
            item["source"],
            item["service"],
            item["class"],
            item["control"],
            item.get("due", ""),
        )
    )
    counts = Counter(item["severity"] for item in findings)
    threshold_failed = False
    if fail_on != "none":
        threshold = SEVERITY_RANK[fail_on]
        threshold_failed = any(
            SEVERITY_RANK[item["severity"]] >= threshold for item in findings
        )

    report: dict[str, Any] = {
        "apiVersion": API_VERSION,
        "kind": KIND,
        "ok": not validation_errors and not threshold_failed,
        "as_of": format_timestamp(as_of),
        "horizon_days": horizon_days,
        "fail_on": fail_on,
        "automation": {
            "mode": "proposal-only",
            "external_write_permitted": False,
            "policy_mutation_permitted": False,
            "declaration_mutation_permitted": False,
        },
        "summary": {
            "total": len(findings),
            "high": counts["high"],
            "medium": counts["medium"],
            "low": counts["low"],
            "unknown_owner": sum(item["owner"] == "unknown" for item in findings),
        },
        "validation_errors": sorted(validation_errors),
        "findings": findings,
    }
    report["report_digest"] = report_digest(report)
    report_schema = json.loads(REPORT_SCHEMA.read_text(encoding="utf-8"))
    jsonschema.Draft202012Validator(
        report_schema,
        format_checker=jsonschema.FormatChecker(),
    ).validate(report)
    return report


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Emit deterministic, proposal-only NetKingdom posture feedback."
    )
    parser.add_argument("declarations", nargs="+", type=pathlib.Path)
    parser.add_argument("--as-of", required=True, help="RFC 3339 timestamp with explicit offset")
    parser.add_argument("--horizon-days", type=int, default=30)
    parser.add_argument("--fail-on", choices=("none", "low", "medium", "high"), default="high")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    if args.horizon_days < 0:
        raise SystemExit("--horizon-days must be non-negative")
    try:
        as_of = parse_timestamp(args.as_of)
    except ValueError as exc:
        raise SystemExit(f"invalid --as-of: {exc}") from exc

    report = build_report(
        args.declarations,
        as_of=as_of,
        horizon_days=args.horizon_days,
        fail_on=args.fail_on,
    )
    print(json.dumps(report, indent=2, sort_keys=True))
    if report["validation_errors"]:
        return 2
    return 0 if report["ok"] else 1


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