#!/usr/bin/env python3
"""Verify and render a reviewable Hermes chat-router image release."""

from __future__ import annotations

import argparse
import base64
import difflib
import json
import os
import re
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any, Callable


DEFAULT_IMAGE = "registry.bstein.dev/bstein/hermes-chat-router"
HARBOR_API_ORIGIN = "https://registry.bstein.dev/api/v2.0"
HARBOR_PROJECT = "bstein"
HARBOR_REPOSITORY = "hermes-chat-router"
IMMUTABLE_REPOSITORY_PATTERN = "hermes-chat-router"
IMMUTABLE_TAG_PATTERN = "git-*-build-*"
DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
BUILD_PATTERN = re.compile(r"^[1-9][0-9]*$")
DESTINATION_PATTERN = re.compile(
    r"^registry\.bstein\.dev/bstein/hermes-chat-router:"
    r"git-([0-9a-f]{40})-build-([1-9][0-9]*)$"
)


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """Never forward registry credentials to a redirect target."""

    def redirect_request(self, _request, _file, _code, _message, _headers, _url):
        return None


def _validated(value: str, pattern: re.Pattern[str], label: str) -> str:
    """Return a normalized value only when it matches the release contract."""
    normalized = value.strip()
    if not pattern.fullmatch(normalized):
        raise ValueError(f"invalid {label}: expected {pattern.pattern}")
    return normalized


def validate_destination(
    destination: str, source_revision: str, build_number: str
) -> tuple[str, str]:
    """Bind the unique build tag to one reviewed source and Jenkins build."""
    revision = _validated(source_revision, REVISION_PATTERN, "source revision")
    build = _validated(build_number, BUILD_PATTERN, "build number")
    match = DESTINATION_PATTERN.fullmatch(destination.strip())
    if not match or match.groups() != (revision, build):
        raise ValueError("destination does not match the reviewed revision and build")
    return revision, build


def validate_kaniko_evidence(
    *, digest_text: str, image_text: str, destination: str
) -> str:
    """Cross-check Kaniko's two independent output files."""
    digests = digest_text.splitlines()
    images = image_text.splitlines()
    if len(digests) != 1 or len(images) != 1:
        raise ValueError("Kaniko evidence must contain exactly one line per file")
    digest = _validated(digests[0], DIGEST_PATTERN, "image digest")
    if images[0].strip() != f"{destination}@{digest}":
        raise ValueError("Kaniko image evidence does not match destination and digest")
    return digest


def _registry_request(request: urllib.request.Request, timeout: int) -> Any:
    """Return same-origin registry responses without following redirects."""
    opener = urllib.request.build_opener(_NoRedirect())
    try:
        return opener.open(request, timeout=timeout)
    except urllib.error.HTTPError as exc:
        return exc


def _authorization(username: str, password: str) -> str:
    """Build a Basic header without placing credentials in a URL."""
    if not username or not password:
        raise RuntimeError("Harbor credentials are unavailable")
    token = base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
    return f"Basic {token}"


def _artifact_response(
    destination: str,
    *,
    username: str,
    password: str,
    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
) -> tuple[int, bytes]:
    """Read one exact Harbor artifact by candidate tag."""
    if not DESTINATION_PATTERN.fullmatch(destination):
        raise ValueError("invalid destination")
    tag = urllib.parse.quote(destination.rsplit(":", 1)[1], safe="")
    request = urllib.request.Request(
        f"{HARBOR_API_ORIGIN}/projects/{HARBOR_PROJECT}/repositories/"
        f"{HARBOR_REPOSITORY}/artifacts/{tag}?with_immutable_status=true",
        headers={
            "Accept": "application/json",
            "Authorization": _authorization(username, password),
        },
        method="GET",
    )
    with opener(request, 20) as response:
        body = response.read(1_048_577)
        if len(body) > 1_048_576:
            raise RuntimeError("Harbor artifact response exceeded the size limit")
        return int(response.status), body


def _rules_response(
    *,
    username: str,
    password: str,
    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
) -> tuple[int, bytes, dict[str, str]]:
    """Read the complete bounded Harbor immutable-rule page."""
    request = urllib.request.Request(
        f"{HARBOR_API_ORIGIN}/projects/{HARBOR_PROJECT}/immutabletagrules"
        "?page=1&page_size=100",
        headers={
            "Accept": "application/json",
            "Authorization": _authorization(username, password),
        },
        method="GET",
    )
    with opener(request, 20) as response:
        body = response.read(1_048_577)
        if len(body) > 1_048_576:
            raise RuntimeError("Harbor immutable rule response exceeded the size limit")
        return int(response.status), body, dict(response.headers)


def _normalized_rule(rule: dict[str, Any]) -> dict[str, Any]:
    """Select only immutable-policy fields used by this lane."""
    return {
        "disabled": bool(rule.get("disabled", False)),
        "action": rule.get("action"),
        "template": rule.get("template"),
        "tag_selectors": [
            {key: item.get(key) for key in ("kind", "decoration", "pattern")}
            for item in rule.get("tag_selectors") or []
            if isinstance(item, dict)
        ],
        "scope_selectors": {
            "repository": [
                {key: item.get(key) for key in ("kind", "decoration", "pattern")}
                for item in (rule.get("scope_selectors") or {}).get("repository", [])
                if isinstance(item, dict)
            ]
        },
    }


def verify_immutable_policy(
    *,
    username: str,
    password: str,
    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
) -> None:
    """Fail closed unless one exact active router immutability rule exists."""
    status, body, headers = _rules_response(
        username=username, password=password, opener=opener
    )
    if status != 200:
        raise RuntimeError(f"Harbor immutable policy preflight returned HTTP {status}")
    try:
        rules = json.loads(body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise RuntimeError("Harbor returned invalid immutable rule JSON") from exc
    if not isinstance(rules, list) or not all(isinstance(item, dict) for item in rules):
        raise RuntimeError("Harbor immutable rule list has an invalid shape")
    total = next(
        (value for key, value in headers.items() if key.lower() == "x-total-count"),
        None,
    )
    if total is None or not str(total).isdecimal() or int(total) != len(rules):
        raise RuntimeError("Harbor immutable rule page is incomplete")
    expected = {
        "disabled": False,
        "action": "immutable",
        "template": "immutable_template",
        "tag_selectors": [
            {
                "kind": "doublestar",
                "decoration": "matches",
                "pattern": IMMUTABLE_TAG_PATTERN,
            }
        ],
        "scope_selectors": {
            "repository": [
                {
                    "kind": "doublestar",
                    "decoration": "repoMatches",
                    "pattern": IMMUTABLE_REPOSITORY_PATTERN,
                }
            ]
        },
    }
    matches = [
        value
        for value in map(_normalized_rule, rules)
        if value["tag_selectors"] == expected["tag_selectors"]
        and value["scope_selectors"] == expected["scope_selectors"]
    ]
    if matches != [expected]:
        raise RuntimeError("Harbor router immutable build-tag policy is not exact")


def assert_tag_absent(
    destination: str,
    *,
    username: str,
    password: str,
    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
) -> None:
    """Reject replay before Kaniko can target an already-used build tag."""
    status, _ = _artifact_response(
        destination, username=username, password=password, opener=opener
    )
    if status == 404:
        return
    if status == 200:
        raise RuntimeError("Harbor destination tag already exists")
    raise RuntimeError(f"Harbor destination preflight returned HTTP {status}")


def verify_registry_digest(
    destination: str,
    digest: str,
    source_revision: str,
    *,
    username: str,
    password: str,
    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
) -> None:
    """Verify Harbor's digest, immutable tag, and persisted source label."""
    digest = _validated(digest, DIGEST_PATTERN, "image digest")
    revision = _validated(source_revision, REVISION_PATTERN, "source revision")
    status, body = _artifact_response(
        destination, username=username, password=password, opener=opener
    )
    if status != 200:
        raise RuntimeError(f"Harbor manifest verification returned HTTP {status}")
    try:
        artifact = json.loads(body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise RuntimeError("Harbor returned invalid artifact JSON") from exc
    tag = destination.rsplit(":", 1)[1]
    matching = [
        item
        for item in artifact.get("tags") or []
        if isinstance(item, dict) and item.get("name") == tag
    ]
    labels = ((artifact.get("extra_attrs") or {}).get("config") or {}).get("Labels")
    if artifact.get("digest") != digest:
        raise RuntimeError("Harbor digest does not match Kaniko evidence")
    if len(matching) != 1 or matching[0].get("immutable") is not True:
        raise RuntimeError("Harbor did not enforce the candidate tag as immutable")
    if not isinstance(labels, dict) or labels.get(
        "org.opencontainers.image.revision"
    ) != revision:
        raise RuntimeError("Harbor OCI source-revision label does not match")


def render_workload(source: str, digest: str) -> str:
    """Replace the single exact router image while preserving the Flux marker."""
    digest = _validated(digest, DIGEST_PATTERN, "image digest")
    identity = "apiVersion: apps/v1\nkind: Deployment\nmetadata:\n  name: hermes-chat-router\n"
    if not source.startswith("# services/hermes/chat-router.yaml\n" + identity):
        raise ValueError("Flux target identity changed")
    lines = source.splitlines(keepends=True)
    matches: list[int] = []
    for index, line in enumerate(lines):
        stripped = line.strip()
        if not stripped.startswith("image: "):
            continue
        value = stripped.removeprefix("image: ").split(" #", 1)[0]
        image, separator, current_digest = value.rpartition("@")
        if separator and re.fullmatch(
            rf"{re.escape(DEFAULT_IMAGE)}(?::[A-Za-z0-9_][A-Za-z0-9_.-]{{0,127}})?",
            image,
        ):
            _validated(current_digest, DIGEST_PATTERN, "current Flux image digest")
            matches.append(index)
    if len(matches) != 1:
        raise ValueError(f"expected exactly one router image; found {len(matches)}")
    index = matches[0]
    indent = lines[index][: len(lines[index]) - len(lines[index].lstrip())]
    comment = ""
    if " #" in lines[index]:
        comment = " #" + lines[index].split(" #", 1)[1].rstrip("\n")
    newline = "\n" if lines[index].endswith("\n") else ""
    lines[index] = f"{indent}image: {DEFAULT_IMAGE}@{digest}{comment}{newline}"
    return "".join(lines)


def _metadata(
    digest: str, revision: str, build: str, destination: str
) -> dict[str, Any]:
    return {
        "build_number": build,
        "digest": digest,
        "flux_image": f"{DEFAULT_IMAGE}@{digest}",
        "flux_targets": ["apps/Deployment/hermes/hermes-chat-router"],
        "image": DEFAULT_IMAGE,
        "published_tag": destination,
        "source_revision": revision,
    }


def _expected_artifacts(
    *,
    digest: str,
    source_revision: str,
    build_number: str,
    destination: str,
    manifest: Path,
) -> dict[str, str]:
    source = manifest.read_text(encoding="utf-8")
    rendered = render_workload(source, digest)
    patch = "".join(
        difflib.unified_diff(
            source.splitlines(keepends=True),
            rendered.splitlines(keepends=True),
            fromfile="a/services/hermes/chat-router.yaml",
            tofile="b/services/hermes/chat-router.yaml",
        )
    )
    if not patch:
        raise ValueError("published digest already matches the Flux target")
    metadata = _metadata(digest, source_revision, build_number, destination)
    return {
        "hermes-chat-router-deployment.yaml": rendered,
        "hermes-chat-router-image-update.patch": patch,
        "hermes-chat-router-image.json": json.dumps(
            metadata, indent=2, sort_keys=True
        )
        + "\n",
    }


def write_release_artifacts(
    *,
    digest: str,
    source_revision: str,
    build_number: str,
    destination: str,
    manifest: Path,
    output_dir: Path,
) -> None:
    """Write deterministic, credential-free Flux handoff evidence."""
    digest = _validated(digest, DIGEST_PATTERN, "image digest")
    revision, build = validate_destination(destination, source_revision, build_number)
    expected = _expected_artifacts(
        digest=digest,
        source_revision=revision,
        build_number=build,
        destination=destination,
        manifest=manifest,
    )
    output_dir.mkdir(parents=True, exist_ok=True)
    for name, content in expected.items():
        (output_dir / name).write_text(content, encoding="utf-8")


def validate_release_artifacts(
    *,
    digest: str,
    source_revision: str,
    build_number: str,
    destination: str,
    manifest: Path,
    output_dir: Path,
) -> None:
    """Recompute and compare every archived handoff byte."""
    digest = _validated(digest, DIGEST_PATTERN, "image digest")
    revision, build = validate_destination(destination, source_revision, build_number)
    expected = _expected_artifacts(
        digest=digest,
        source_revision=revision,
        build_number=build,
        destination=destination,
        manifest=manifest,
    )
    entries = list(output_dir.iterdir())
    if {entry.name for entry in entries} != set(expected) or not all(
        entry.is_file() and not entry.is_symlink() for entry in entries
    ):
        raise ValueError("release output must contain exactly three evidence files")
    for name, content in expected.items():
        if (output_dir / name).read_text(encoding="utf-8") != content:
            raise ValueError(f"release evidence is incomplete or mismatched: {name}")


def _credentials() -> tuple[str, str]:
    username = os.environ.get("HARBOR_USER", "")
    password = os.environ.get("HARBOR_PASSWORD", "")
    if not username or not password:
        raise RuntimeError("Harbor credentials are unavailable")
    return username, password


def _add_common(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--source-revision", required=True)
    parser.add_argument("--build-number", required=True)
    parser.add_argument("--destination", required=True)


def main() -> int:
    """Run one fail-closed candidate or evidence operation."""
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest="command", required=True)
    absent = commands.add_parser("assert-absent")
    _add_common(absent)
    for name in ("render", "verify-evidence"):
        command = commands.add_parser(name)
        _add_common(command)
        command.add_argument("--digest-file", required=True, type=Path)
        command.add_argument("--image-file", required=True, type=Path)
        command.add_argument("--manifest", required=True, type=Path)
        command.add_argument("--output-dir", required=True, type=Path)
    args = parser.parse_args()
    validate_destination(args.destination, args.source_revision, args.build_number)
    if args.command == "verify-evidence":
        digest = validate_kaniko_evidence(
            digest_text=args.digest_file.read_text(encoding="utf-8"),
            image_text=args.image_file.read_text(encoding="utf-8"),
            destination=args.destination,
        )
        validate_release_artifacts(
            digest=digest,
            source_revision=args.source_revision,
            build_number=args.build_number,
            destination=args.destination,
            manifest=args.manifest,
            output_dir=args.output_dir,
        )
        return 0
    username, password = _credentials()
    if args.command == "assert-absent":
        verify_immutable_policy(username=username, password=password)
        assert_tag_absent(
            args.destination, username=username, password=password
        )
        return 0
    digest = validate_kaniko_evidence(
        digest_text=args.digest_file.read_text(encoding="utf-8"),
        image_text=args.image_file.read_text(encoding="utf-8"),
        destination=args.destination,
    )
    verify_registry_digest(
        args.destination,
        digest,
        args.source_revision,
        username=username,
        password=password,
    )
    write_release_artifacts(
        digest=digest,
        source_revision=args.source_revision,
        build_number=args.build_number,
        destination=args.destination,
        manifest=args.manifest,
        output_dir=args.output_dir,
    )
    return 0


if __name__ == "__main__":  # pragma: no cover - exercised through main()
    raise SystemExit(main())