"""Fail closed when an RC Compose plan can overlap a protected installation.

The public-candidate workflow proves a disposable API/Web/Database stack.  P12 adds a
GPU-capable Node Agent and Runtime Worker, which makes an accidental shared volume, host
network, port, or resource name materially more dangerous.  This helper validates the rendered
Compose JSON and the resources observed after startup.  It deliberately accepts no implicit
production defaults: callers must provide the candidate root and every protected port/name they
want to keep outside the validation project namespace.
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Any


PROJECT_PATTERN = re.compile(
    r"modelforge-validation-[a-z0-9](?:[a-z0-9-]{0,39}[a-z0-9])?\Z"
)
VOLUME_NAME_PATTERN = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9_.-]*")
CONFIG_TARGET = "/app/config"
CONFIG_MANIFEST = "models/initial-candidates.yaml"


class IsolationError(ValueError):
    """A candidate can share state or reach outside its owned RC namespace."""


def _mapping(value: object, label: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise IsolationError(f"{label} must be an object")
    return value


def _sequence(value: object, label: str) -> Sequence[Any]:
    if not isinstance(value, Sequence) or isinstance(value, (str, bytes, bytearray)):
        raise IsolationError(f"{label} must be a list")
    return value


def validate_project(project: str) -> str:
    if PROJECT_PATTERN.fullmatch(project) is None:
        raise IsolationError(
            "validation project must use a bounded modelforge-validation-* namespace"
        )
    return project


def _is_relative_to(path: Path, root: Path) -> bool:
    try:
        path.relative_to(root)
    except ValueError:
        return False
    return True


def _published_port(port: Mapping[str, Any], service_name: str, public_tls_port: int | None) -> int | None:
    published = port.get("published")
    if published in (None, ""):
        return None
    try:
        result = int(str(published))
    except ValueError as exc:
        raise IsolationError(f"{service_name} has an invalid published port") from exc
    if not 0 <= result <= 65535:
        raise IsolationError(f"{service_name} has an invalid published port")
    host_ip = str(port.get("host_ip") or "")
    if host_ip not in {"127.0.0.1", "::1"} and not (
        service_name == "tls"
        and public_tls_port == result == 18444
        and host_ip == "0.0.0.0"
        and str(port.get("target")) == "8443"
        and port.get("protocol", "tcp") == "tcp"
    ):
        raise IsolationError(f"{service_name} publishes a non-loopback port")
    return result


def _validate_service_mount(
    mount: Mapping[str, Any], *, service_name: str, candidate_root: Path, declared_volumes: set[str]
) -> None:
    mount_type = mount.get("type")
    source = mount.get("source")
    target = str(mount.get("target") or "")
    if target == "/var/run/docker.sock":
        raise IsolationError(f"{service_name} mounts the Docker socket")
    if mount_type == "bind":
        if not isinstance(source, str) or not source:
            raise IsolationError(f"{service_name} has an invalid bind source")
        resolved = Path(source)
        if not resolved.is_absolute():
            resolved = candidate_root / resolved
        try:
            resolved = resolved.resolve()
        except (OSError, RuntimeError) as exc:
            raise IsolationError(f"{service_name} has an invalid bind source") from exc
        if not _is_relative_to(resolved, candidate_root):
            raise IsolationError(f"{service_name} bind source escapes the candidate root")
        try:
            resolved = resolved.resolve(strict=True)
        except FileNotFoundError as exc:
            raise IsolationError(f"{service_name} has a missing bind source") from exc
        except (OSError, RuntimeError) as exc:
            raise IsolationError(f"{service_name} has an invalid bind source") from exc
        if not _is_relative_to(resolved, candidate_root):
            raise IsolationError(f"{service_name} bind source escapes the candidate root")
        if mount.get("read_only") is not True:
            raise IsolationError(f"{service_name} has a writable bind mount")
        if target == CONFIG_TARGET:
            manifest = resolved / CONFIG_MANIFEST
            try:
                resolved_manifest = manifest.resolve(strict=True)
            except FileNotFoundError as exc:
                raise IsolationError(
                    f"{service_name} {CONFIG_TARGET} bind is missing {CONFIG_MANIFEST} under its source"
                ) from exc
            except (OSError, RuntimeError) as exc:
                raise IsolationError(
                    f"{service_name} {CONFIG_TARGET} bind has an invalid {CONFIG_MANIFEST}"
                ) from exc
            if not _is_relative_to(resolved_manifest, resolved) or not resolved_manifest.is_file():
                raise IsolationError(
                    f"{service_name} {CONFIG_TARGET} bind is missing {CONFIG_MANIFEST} under its source"
                )
        return
    if mount_type == "volume":
        if not isinstance(source, str) or VOLUME_NAME_PATTERN.fullmatch(source) is None:
            raise IsolationError(f"{service_name} has an invalid named volume")
        if source not in declared_volumes:
            raise IsolationError(f"{service_name} uses an undeclared volume {source}")
        return
    if mount_type in {"tmpfs", "npipe"}:
        return
    raise IsolationError(f"{service_name} has unsupported mount type {mount_type!r}")


def validate_compose(
    document: Mapping[str, Any],
    *,
    project: str,
    candidate_root: Path,
    protected_ports: set[int],
    protected_resource_names: set[str],
    public_tls_port: int | None = None,
) -> dict[str, object]:
    """Validate a rendered Compose document before any RC container starts."""

    validate_project(project)
    candidate_root = candidate_root.resolve()
    services = _mapping(document.get("services"), "services")
    if not services:
        raise IsolationError("Compose plan has no services")
    top_volumes = _mapping(document.get("volumes", {}), "volumes")
    top_networks = _mapping(document.get("networks", {}), "networks")
    declared_volumes = set(top_volumes)
    declared_networks = set(top_networks)
    if not declared_volumes or not declared_networks:
        raise IsolationError("validation plan requires its own declared volumes and network")
    if public_tls_port is not None and public_tls_port != 18444:
        raise IsolationError("the only admitted public TLS test port is 18444")
    if public_tls_port is not None and "tls" not in services:
        raise IsolationError("public TLS test port requires the isolated tls service")

    for name, configuration in top_volumes.items():
        if VOLUME_NAME_PATTERN.fullmatch(name) is None:
            raise IsolationError(f"invalid declared volume {name!r}")
        config = _mapping(configuration, f"volume {name}")
        if config.get("external"):
            raise IsolationError(f"volume {name} is external")
        actual_name = config.get("name")
        if actual_name is not None and actual_name != f"{project}_{name}":
            raise IsolationError(f"volume {name} has a non-project name")

    for name, configuration in top_networks.items():
        if VOLUME_NAME_PATTERN.fullmatch(name) is None:
            raise IsolationError(f"invalid declared network {name!r}")
        config = _mapping(configuration, f"network {name}")
        if config.get("external"):
            raise IsolationError(f"network {name} is external")
        actual_name = config.get("name")
        if actual_name is not None and actual_name != f"{project}_{name}":
            raise IsolationError(f"network {name} has a non-project name")

    published_ports: set[int] = set()
    for name, configuration in services.items():
        service = _mapping(configuration, f"service {name}")
        if service.get("container_name"):
            raise IsolationError(f"{name} declares a global container name")
        if service.get("network_mode") == "host":
            raise IsolationError(f"{name} uses the host network")
        configured_networks = service.get("networks", [])
        if isinstance(configured_networks, Mapping):
            service_networks: Sequence[Any] = list(configured_networks)
        else:
            service_networks = _sequence(configured_networks, f"{name}.networks")
        network_names = []
        for network in service_networks:
            network_name = network if isinstance(network, str) else _mapping(network, f"{name}.network").get("target")
            if network_name not in declared_networks:
                raise IsolationError(f"{name} uses an undeclared network")
            network_names.append(network_name)
        if name == "tls" and public_tls_port is not None:
            if set(network_names) != {"ingress"}:
                raise IsolationError("tls service must join only the candidate ingress network")
            if service.get("image") != "caddy:2.10-alpine@sha256:4c6e91c6ed0e2fa03efd5b44747b625fec79bc9cd06ac5235a779726618e530d":
                raise IsolationError("tls service must use the pinned Caddy image")
            if service.get("read_only") is not True or service.get("cap_drop") != ["ALL"]:
                raise IsolationError("tls service must have read-only root and no capabilities")
            if service.get("privileged") or service.get("cap_add") or service.get("build"):
                raise IsolationError("tls service cannot be privileged or locally built")
            if "no-new-privileges:true" not in _sequence(service.get("security_opt", []), "tls.security_opt"):
                raise IsolationError("tls service must refuse new privileges")
            mounts = _sequence(service.get("volumes", []), "tls.volumes")
            targets = {str(_mapping(mount, "tls.volume").get("target")) for mount in mounts}
            if targets != {"/etc/caddy/Caddyfile", "/run/tls/fullchain.pem", "/run/tls/privkey.pem"} or len(mounts) != 3:
                raise IsolationError("tls service must mount only its config and certificate pair")
            tls_ports = _sequence(service.get("ports", []), "tls.ports")
            if len(tls_ports) != 1:
                raise IsolationError("tls service must publish exactly one port")
            tls_port = _mapping(tls_ports[0], "tls.port")
            if str(tls_port.get("published")) != "18444" or tls_port.get("host_ip") != "0.0.0.0":
                raise IsolationError("tls service must bind only public test port 18444")
        for mount in _sequence(service.get("volumes", []), f"{name}.volumes"):
            _validate_service_mount(
                _mapping(mount, f"{name}.volume"),
                service_name=name,
                candidate_root=candidate_root,
                declared_volumes=declared_volumes,
            )
        for port in _sequence(service.get("ports", []), f"{name}.ports"):
            published = _published_port(_mapping(port, f"{name}.port"), name, public_tls_port)
            if published and published in protected_ports:
                raise IsolationError(f"{name} reuses protected host port {published}")
            if published:
                published_ports.add(published)

    serialized = json.dumps(document, sort_keys=True)
    referenced = sorted(name for name in protected_resource_names if name and name in serialized)
    if referenced:
        raise IsolationError(f"Compose plan references protected resources: {', '.join(referenced)}")
    return {
        "project": project,
        "services": sorted(services),
        "declared_volumes": sorted(declared_volumes),
        "declared_networks": sorted(declared_networks),
        "published_ports": sorted(published_ports),
    }


def validate_observed_resources(
    document: Mapping[str, Any], *, project: str
) -> dict[str, list[str]]:
    """Ensure resources created by Compose retain the exact validation project label and name."""

    validate_project(project)
    expected_label = project
    prefixes = {"containers": f"{project}-", "volumes": f"{project}_", "networks": f"{project}_"}
    result: dict[str, list[str]] = {}
    for kind, prefix in prefixes.items():
        items = _sequence(document.get(kind), kind)
        if not items:
            raise IsolationError(f"no observed {kind} for {project}")
        names: list[str] = []
        for item in items:
            resource = _mapping(item, kind)
            name = resource.get("name")
            labels = _mapping(resource.get("labels"), f"{kind}.labels")
            if not isinstance(name, str) or not name.startswith(prefix):
                raise IsolationError(f"{kind} resource has a non-project name")
            if labels.get("com.docker.compose.project") != expected_label:
                raise IsolationError(f"{kind} resource has no exact project label")
            names.append(name)
        result[kind] = sorted(names)
    return result


def _docker(*args: str) -> str:
    completed = subprocess.run(
        ("docker", *args),
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    if completed.returncode:
        raise IsolationError(f"Docker inspection failed: {' '.join(args)}")
    return completed.stdout


def collect_observed_resources(project: str) -> dict[str, list[dict[str, object]]]:
    """Collect only resources carrying the exact Compose project label.

    ``docker inspect`` is used instead of parsing ``docker * ls`` label strings: a label value can
    itself contain commas or equals signs, and P12 must not mistake a truncated display value for
    resource ownership evidence.
    """

    validate_project(project)
    kinds = {
        "containers": ("container", ("ls", "--all", "--quiet")),
        "volumes": ("volume", ("ls", "--quiet")),
        "networks": ("network", ("ls", "--quiet")),
    }
    result: dict[str, list[dict[str, object]]] = {}
    for plural, (singular, list_args) in kinds.items():
        identifiers = [
            value for value in _docker(
                singular, *list_args, "--filter", f"label=com.docker.compose.project={project}"
            ).splitlines() if value
        ]
        resources: list[dict[str, object]] = []
        for identifier in identifiers:
            try:
                inspected = json.loads(_docker(singular, "inspect", identifier))
            except json.JSONDecodeError as exc:
                raise IsolationError(f"Docker returned invalid {singular} inspection JSON") from exc
            if not isinstance(inspected, list) or len(inspected) != 1 or not isinstance(inspected[0], Mapping):
                raise IsolationError(f"Docker returned ambiguous {singular} inspection data")
            item = inspected[0]
            labels = item.get("Labels")
            if singular == "container":
                config = _mapping(item.get("Config"), "container.Config")
                labels = config.get("Labels")
            resources.append(
                {
                    "name": str(item.get("Name") or "").lstrip("/"),
                    "labels": dict(_mapping(labels or {}, f"{singular}.Labels")),
                }
            )
        result[plural] = resources
    return result


def _read_json(path: Path) -> Mapping[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise IsolationError(f"cannot read JSON evidence {path}") from exc
    return _mapping(value, str(path))


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--project", required=True)
    parser.add_argument("--candidate-root", required=True, type=Path)
    parser.add_argument("--compose-json", required=True, type=Path)
    parser.add_argument("--observed-json", type=Path)
    parser.add_argument("--inspect-docker", action="store_true")
    parser.add_argument("--protected-port", action="append", default=[], type=int)
    parser.add_argument("--public-tls-port", type=int)
    parser.add_argument("--protected-resource", action="append", default=[])
    args = parser.parse_args(argv)
    try:
        if args.observed_json is not None and args.inspect_docker:
            raise IsolationError("choose observed JSON or Docker inspection, not both")
        result: dict[str, object] = {
            "preflight": validate_compose(
                _read_json(args.compose_json),
                project=args.project,
                candidate_root=args.candidate_root,
                protected_ports=set(args.protected_port),
                protected_resource_names=set(args.protected_resource),
                public_tls_port=args.public_tls_port,
            )
        }
        if args.observed_json is not None:
            result["observed"] = validate_observed_resources(
                _read_json(args.observed_json), project=args.project
            )
        elif args.inspect_docker:
            result["observed"] = validate_observed_resources(
                collect_observed_resources(args.project), project=args.project
            )
    except IsolationError as exc:
        print(json.dumps({"result": "FAIL", "error": str(exc)}, sort_keys=True))
        return 2
    print(json.dumps({"result": "PASS", **result}, sort_keys=True))
    return 0


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