src/open_range/lint.py

"""Manifest authoring lint for OpenRange.

Validates YAML manifest files with cross-field consistency checks beyond
what Pydantic schema validation covers.

Usage::

    python -m open_range.lint manifests/tier1_basic.yaml
    python -m open_range.lint manifests/*.yaml
"""

from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path
from typing import Any

from open_range.manifest_schema import Manifest, load_manifest


# ---------------------------------------------------------------------------
# Cross-field checks
# ---------------------------------------------------------------------------


def _check_connects_to(manifest: Manifest) -> list[str]:
    """All hosts referenced in connects_to must exist."""
    host_names = {h.name for h in manifest.topology.hosts}
    errors: list[str] = []
    for host in manifest.topology.hosts:
        for target in host.connects_to:
            if target not in host_names:
                errors.append(
                    f"Host '{host.name}' connects_to '{target}' "
                    f"which is not a defined host. "
                    f"Valid hosts: {sorted(host_names)}"
                )
    return errors


def _check_firewall_zones(manifest: Manifest) -> list[str]:
    """All firewall rules reference valid zones."""
    zone_names = {n.name for n in manifest.topology.networks}
    errors: list[str] = []
    for rule in manifest.topology.firewall_rules:
        if rule.from_zone not in zone_names:
            errors.append(
                f"Firewall rule from_zone '{rule.from_zone}' "
                f"is not a defined zone. Valid zones: {sorted(zone_names)}"
            )
        if rule.to_zone not in zone_names:
            errors.append(
                f"Firewall rule to_zone '{rule.to_zone}' "
                f"is not a defined zone. Valid zones: {sorted(zone_names)}"
            )
    return errors


def _check_user_hosts(manifest: Manifest) -> list[str]:
    """All user host references must be valid hosts."""
    host_names = {h.name for h in manifest.topology.hosts}
    errors: list[str] = []
    for user in manifest.users:
        for host in user.hosts:
            if host not in host_names:
                errors.append(
                    f"User '{user.username}' references host '{host}' "
                    f"which is not a defined host. "
                    f"Valid hosts: {sorted(host_names)}"
                )
    return errors


def _check_npc_usernames(manifest: Manifest) -> list[str]:
    """All NPC personas must reference valid usernames from the users list."""
    user_names = {u.username for u in manifest.users}
    errors: list[str] = []
    for npc in manifest.npc_personas:
        if npc.username not in user_names:
            errors.append(
                f"NPC persona references username '{npc.username}' "
                f"which is not in the users list. "
                f"Valid usernames: {sorted(user_names)}"
            )
    return errors


def _check_data_inventory_hosts(manifest: Manifest) -> list[str]:
    """All data inventory items must reference valid hosts."""
    host_names = {h.name for h in manifest.topology.hosts}
    errors: list[str] = []
    for asset in manifest.data_inventory:
        if asset.host not in host_names:
            errors.append(
                f"Data asset '{asset.name}' references host '{asset.host}' "
                f"which is not a defined host. "
                f"Valid hosts: {sorted(host_names)}"
            )
    return errors


def _check_business_process_flows(manifest: Manifest) -> list[str]:
    """All business process data_flow entries must reference valid host:service pairs."""
    # Build a set of valid host:service pairs
    host_services: dict[str, set[str]] = {}
    for host in manifest.topology.hosts:
        host_services[host.name] = set(host.services)

    errors: list[str] = []
    for process in manifest.business_processes:
        for entry in process.data_flow:
            if ":" not in entry:
                errors.append(
                    f"Business process '{process.name}' has data_flow entry "
                    f"'{entry}' that is not in host:service format"
                )
                continue
            host_part, service_part = entry.split(":", 1)
            if host_part not in host_services:
                errors.append(
                    f"Business process '{process.name}' references host "
                    f"'{host_part}' in data_flow entry '{entry}' "
                    f"which is not a defined host. "
                    f"Valid hosts: {sorted(host_services.keys())}"
                )
            elif service_part not in host_services[host_part]:
                errors.append(
                    f"Business process '{process.name}' references service "
                    f"'{service_part}' on host '{host_part}' in data_flow "
                    f"entry '{entry}', but host '{host_part}' only has "
                    f"services: {sorted(host_services[host_part])}"
                )
    return errors


_PRINCIPAL_RE = re.compile(r"^[A-Za-z0-9._@-]+$")


def _check_trust_relationships(manifest: Manifest) -> list[str]:
    """Trust principals must be well-formed identifiers.

    Trust edges may reference people who are not login accounts. Those are
    normalized into the canonical principal catalog at build time, so lint
    should validate identifier quality rather than requiring every principal to
    appear in ``users``.
    """
    errors: list[str] = []
    for rel in manifest.trust_relationships:
        if rel.source and not _PRINCIPAL_RE.match(rel.source):
            errors.append(
                f"Trust relationship source '{rel.source}' is not a valid "
                "principal identifier"
            )
        if rel.target and not _PRINCIPAL_RE.match(rel.target):
            errors.append(
                f"Trust relationship target '{rel.target}' is not a valid "
                "principal identifier"
            )
    return errors


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

ALL_CHECKS = [
    ("connects_to references", _check_connects_to),
    ("firewall zone references", _check_firewall_zones),
    ("user host references", _check_user_hosts),
    ("NPC persona usernames", _check_npc_usernames),
    ("data inventory hosts", _check_data_inventory_hosts),
    ("business process data flows", _check_business_process_flows),
    ("trust relationship principals", _check_trust_relationships),
]


def lint_manifest(manifest: Manifest) -> dict[str, list[str]]:
    """Run all cross-field checks on a validated manifest.

    Returns:
        Dict mapping check name to list of error messages.
        Empty list means the check passed.
    """
    results: dict[str, list[str]] = {}
    for name, check_fn in ALL_CHECKS:
        results[name] = check_fn(manifest)
    return results


def lint_file(path: str | Path) -> dict[str, Any]:
    """Load and lint a manifest file.

    Returns:
        Dict with 'path', 'valid' (bool), 'schema_error' (str or None),
        and 'checks' (dict of check name -> error list).
    """
    path = Path(path)
    result: dict[str, Any] = {
        "path": str(path),
        "valid": True,
        "schema_error": None,
        "checks": {},
    }

    try:
        manifest = load_manifest(path)
    except FileNotFoundError as exc:
        result["valid"] = False
        result["schema_error"] = str(exc)
        return result
    except Exception as exc:
        result["valid"] = False
        result["schema_error"] = str(exc)
        return result

    checks = lint_manifest(manifest)
    result["checks"] = checks

    for errors in checks.values():
        if errors:
            result["valid"] = False
            break

    return result


# ---------------------------------------------------------------------------
# Terminal output
# ---------------------------------------------------------------------------

# ANSI color codes
_GREEN = "\033[32m"
_RED = "\033[31m"
_BOLD = "\033[1m"
_RESET = "\033[0m"


def _format_result(result: dict[str, Any]) -> str:
    """Format a lint result as colored terminal output."""
    lines: list[str] = []
    path = result["path"]

    if result["schema_error"]:
        lines.append(f"{_BOLD}{path}{_RESET}")
        lines.append(f"  {_RED}SCHEMA ERROR: {result['schema_error']}{_RESET}")
        return "\n".join(lines)

    lines.append(f"{_BOLD}{path}{_RESET}")
    all_passed = True
    for check_name, errors in result["checks"].items():
        if errors:
            all_passed = False
            lines.append(f"  {_RED}FAIL{_RESET} {check_name}")
            for err in errors:
                lines.append(f"    - {err}")
        else:
            lines.append(f"  {_GREEN}PASS{_RESET} {check_name}")

    if all_passed:
        lines.append(f"  {_GREEN}All checks passed.{_RESET}")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Lint OpenRange manifest YAML files",
    )
    parser.add_argument(
        "manifests",
        nargs="+",
        help="One or more manifest YAML file paths",
    )
    args = parser.parse_args()

    any_failed = False
    for manifest_path in args.manifests:
        result = lint_file(manifest_path)
        print(_format_result(result))
        print()
        if not result["valid"]:
            any_failed = True

    sys.exit(1 if any_failed else 0)


if __name__ == "__main__":
    main()