DataOps-platform/scripts/validate_p3_wp14_local_configs.py

"""Validate two non-secret P3-WP14 snapshots through isolated Compose JSON renders.

Every snapshot is read once as bytes and decoded from that same read.  Its two
runtime keys are the entire subprocess configuration (apart from ``PATH``), so
the render cannot inherit parent DataOps configuration.  The script records
only digests and consumed non-secret label/port facts; it never starts Compose.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import subprocess
import unicodedata
from datetime import datetime, timezone
from pathlib import Path

from p3_wp14_secure_io import atomic_write_bytes, exclusive_lock, read_bytes_once, read_json_once


ROOT = Path(__file__).resolve().parents[1]
COMMAND_HOME = ROOT / "work/p3_wp14_config_command_home"
COMPOSE_EXECUTABLE = "/usr/local/bin/docker-compose"
CONFIGS = (
    "docs/validation/P3_WP14_LOCAL_CONFIG_A.json",
    "docs/validation/P3_WP14_LOCAL_CONFIG_B.json",
)
COMPARISON = ROOT / "docs/validation/P3_WP14_LOCAL_CONFIG_COMPARISON.json"
RENDER_RECORD = ROOT / "docs/validation/P3_WP14_LOCAL_CONFIG_RENDER.json"
LOCK = ROOT / "work/p3_wp14_config_lock_scope/lock"
RENDER_COMMAND = [COMPOSE_EXECUTABLE, "--env-file", "/dev/null", "-f", "deploy/docker/docker-compose.yml", "config", "--format", "json"]
ALLOWED_DIFFERENCES = {
    "/environment/label",
    "/runtime/DATAOPS_LOCAL_ENV_LABEL",
    "/runtime/BACKEND_PORT",
}


def _canonical_bytes(value: object) -> bytes:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")


def _flatten(value: object, prefix: str = "") -> dict[str, object]:
    if isinstance(value, dict):
        flattened: dict[str, object] = {}
        for key, child in value.items():
            flattened.update(_flatten(child, f"{prefix}/{key}"))
        return flattened
    if isinstance(value, list):
        flattened = {}
        for index, child in enumerate(value):
            flattened.update(_flatten(child, f"{prefix}/{index}"))
        return flattened
    return {prefix: value}


def _assert_render_diff(actual: set[str], allowed: set[str]) -> None:
    if actual != allowed:
        raise ValueError(f"render difference outside exact allowlist: {sorted(actual)}")


def _ascii(value: object, *, name: str, maximum: int = 80) -> str:
    if not isinstance(value, str) or not value or len(value) > maximum or not value.isascii() or unicodedata.normalize("NFKC", value) != value:
        raise ValueError(f"invalid {name}")
    return value


def _validate_snapshot(snapshot: object) -> None:
    if not isinstance(snapshot, dict) or set(snapshot) != {"schema_version", "environment", "runtime", "compose"}:
        raise ValueError("snapshot schema is not closed")
    if snapshot["schema_version"] != "1.0":
        raise ValueError("unsupported snapshot schema")
    environment = snapshot["environment"]
    runtime = snapshot["runtime"]
    compose = snapshot["compose"]
    if not isinstance(environment, dict) or set(environment) != {"label", "is_enterprise"} or environment["is_enterprise"] is not False:
        raise ValueError("snapshot must be an explicitly non-enterprise configuration")
    if not isinstance(runtime, dict) or set(runtime) != {"DATAOPS_LOCAL_ENV_LABEL", "BACKEND_PORT"}:
        raise ValueError("runtime schema is not closed")
    if not isinstance(compose, dict) or set(compose) != {"file", "render_command"}:
        raise ValueError("compose schema is not closed")
    if compose["file"] != "deploy/docker/docker-compose.yml" or compose["render_command"] != RENDER_COMMAND:
        raise ValueError("unexpected compose render contract")
    _ascii(environment["label"], name="environment label")
    label = _ascii(runtime["DATAOPS_LOCAL_ENV_LABEL"], name="runtime label")
    if label != environment["label"] or not re.fullmatch(r"LOCAL_ISOLATED_[AB]", label):
        raise ValueError("invalid local environment label")
    port = _ascii(runtime["BACKEND_PORT"], name="backend port", maximum=5)
    if not port.isdecimal() or not 1 <= int(port) <= 65535:
        raise ValueError("invalid backend port")


def _render(snapshot: dict[str, object]) -> tuple[str, dict[str, object], dict[str, object]]:
    runtime = snapshot["runtime"]
    assert isinstance(runtime, dict)
    isolated_env = {
        "PATH": os.environ.get("PATH", ""),
        "HOME": str(COMMAND_HOME),
        "DOCKER_CONFIG": str(COMMAND_HOME / "docker"),
        "DATAOPS_LOCAL_ENV_LABEL": str(runtime["DATAOPS_LOCAL_ENV_LABEL"]),
        "BACKEND_PORT": str(runtime["BACKEND_PORT"]),
    }
    completed = subprocess.run(
        RENDER_COMMAND,
        cwd=ROOT,
        env=isolated_env,
        text=True,
        capture_output=True,
        check=False,
        timeout=30,
    )
    rendered = json.loads(completed.stdout) if completed.returncode == 0 else {}
    services = rendered.get("services", {})
    backend = services.get("backend", {}) if isinstance(services, dict) else {}
    labels = backend.get("labels", {}) if isinstance(backend, dict) else {}
    ports = backend.get("ports", []) if isinstance(backend, dict) else []
    published_ports = {str(port.get("published")) for port in ports if isinstance(port, dict)}
    label = str(snapshot["environment"]["label"])
    return label, {
        "exit_code": completed.returncode,
        "status": "PASS" if completed.returncode == 0 else "FAIL",
        "raw_stdout_sha256": hashlib.sha256(completed.stdout.encode("utf-8")).hexdigest(),
        "stderr_sha256": hashlib.sha256(completed.stderr.encode("utf-8")).hexdigest(),
        "canonical_stdout_sha256": hashlib.sha256(_canonical_bytes(rendered)).hexdigest(),
        "services": sorted(services) if isinstance(services, dict) else [],
        "consumed_label": labels.get("com.dataops.local_env") if isinstance(labels, dict) else None,
        "consumed_backend_port": str(runtime["BACKEND_PORT"]) if str(runtime["BACKEND_PORT"]) in published_ports else None,
    }, _flatten(rendered)


def _main_locked() -> int:
    raw = {path: read_bytes_once(ROOT / path, limit=8192) for path in CONFIGS}
    snapshots = {path: json.loads(content.decode("utf-8")) for path, content in raw.items()}
    for snapshot in snapshots.values():
        _validate_snapshot(snapshot)
    labels = {snapshot["environment"]["label"] for snapshot in snapshots.values()}
    if len(labels) != 2:
        raise ValueError("isolated configurations must use distinct labels")
    first, second = (snapshots[path] for path in CONFIGS)
    differences = {
        path
        for path in set(_flatten(first)) | set(_flatten(second))
        if _flatten(first).get(path) != _flatten(second).get(path)
    }
    if differences != ALLOWED_DIFFERENCES:
        raise ValueError(f"configuration difference outside allowlist: {sorted(differences)}")
    snapshot_sha256 = {path: hashlib.sha256(content).hexdigest() for path, content in raw.items()}
    comparison = read_json_once(COMPARISON, limit=8192)
    if not isinstance(comparison, dict):
        raise ValueError("invalid comparison")
    comparison["snapshot_sha256"] = snapshot_sha256
    atomic_write_bytes(COMPARISON, _canonical_bytes(comparison) + b"\n")

    rendered_pairs = [_render(snapshot) for snapshot in snapshots.values()]
    renders = {label: item for label, item, _ in rendered_pairs}
    first_render, second_render = rendered_pairs[0][2], rendered_pairs[1][2]
    render_differences = {
        path for path in set(first_render) | set(second_render)
        if first_render.get(path) != second_render.get(path)
    }
    allowed_render_differences = {
        "/services/backend/labels/com.dataops.local_env",
        "/services/backend/ports/0/published",
    }
    _assert_render_diff(render_differences, allowed_render_differences)
    service_sets = {tuple(item["services"]) for item in renders.values()}
    renders_consumed = all(
        item["exit_code"] == 0
        and item["consumed_label"] == label
        and item["consumed_backend_port"] is not None
        for label, item in renders.items()
    )
    exit_code = 0 if len(service_sets) == 1 and renders_consumed else 1
    record = {
        "schema_version": "1.0",
        "run_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "command_id": "local_config_compose_render",
        "render_command": RENDER_COMMAND,
        "exit_code": exit_code,
        "status": "PASS" if exit_code == 0 else "FAIL",
        "snapshot_sha256": snapshot_sha256,
        "allowed_difference_paths": sorted(ALLOWED_DIFFERENCES),
        "renders": renders,
        "render_difference_paths": sorted(render_differences),
        "compose_input_sha256": hashlib.sha256(read_bytes_once(ROOT / "deploy/docker/docker-compose.yml")).hexdigest(),
        "boundary": "Each snapshot was passed through an isolated subprocess environment for a local Compose JSON render only; no container was started and no enterprise system was contacted.",
    }
    atomic_write_bytes(RENDER_RECORD, _canonical_bytes(record) + b"\n")
    return exit_code


def main() -> int:
    with exclusive_lock(LOCK):
        atomic_write_bytes(RENDER_RECORD, _canonical_bytes({"schema_version": "1.0", "status": "RUNNING", "exit_code": None, "invalidates_prior_pass": True}) + b"\n")
        try:
            return _main_locked()
        except (subprocess.TimeoutExpired, Exception) as error:
            atomic_write_bytes(RENDER_RECORD, _canonical_bytes({"schema_version": "1.0", "status": "FAILED", "exit_code": 1, "invalidates_prior_pass": True, "error_type": type(error).__name__}) + b"\n")
            return 1


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