#!/usr/bin/env python3
"""Prepare pinned Swagger assets for build.rs. No Cargo config or runtime CDN needed.

Only the upstream tag archive (or an operator-supplied local copy of that tag) is
accepted. ZIP paths are never extracted: bounded, exact members are read and
written to fixed output names. No application data or credentials are accessed.
"""
from __future__ import annotations

import argparse
import hashlib
import io
import json
import os
from pathlib import Path
import re
import shutil
import sys
import tempfile
import time
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener, url2pathname
import zipfile
import zlib

VERSION = "5.33.1"
DOWNLOAD_URL = f"https://github.com/swagger-api/swagger-ui/archive/refs/tags/v{VERSION}.zip"
PREFIX = f"swagger-ui-{VERSION}/"
VENDORED = Path(__file__).resolve().parents[2] / "third_party" / f"swagger-ui-{VERSION}"
MAX_ARCHIVE = 64 * 1024 * 1024
ASSETS = {
    "swagger-ui-bundle.js": ("dist/swagger-ui-bundle.js", 16 * 1024 * 1024),
    "swagger-ui.css": ("dist/swagger-ui.css", 2 * 1024 * 1024),
}
LICENSE_MEMBERS = ("LICENSE", "NOTICE", "dist/swagger-ui-bundle.js.LICENSE.txt")
HELP = ("Install Python 3.10+ and allow HTTPS access to github.com/codeload.github.com, "
        "or set SWAGGER_UI_DOWNLOAD_URL=file:///absolute/path/swagger-ui-5.33.1.zip "
        "to a reviewed copy of the same tag archive. See docs/INSTALL.md.")


class AssetError(ValueError):
    """A controlled, actionable source/archive error (never an implicit fallback)."""


def source_value(value: str | None) -> str | Path:
    """Unset/empty means the pin, never the adapter's historical default."""
    if value is None or not value.strip():
        return DOWNLOAD_URL
    if any(c in value for c in "\r\n\0"):
        raise AssetError("Swagger source contains invalid control characters.")
    value = value.strip()
    if value == DOWNLOAD_URL:
        return value
    try:
        parts = urlsplit(value)
        if (parts.scheme != "file" or parts.netloc not in ("", "localhost")
                or parts.query or parts.fragment):
            raise ValueError()
        # url2pathname decodes percent escapes (including spaces) on this platform.
        path = Path(url2pathname(parts.path))
        if not path.is_absolute() or any(c in str(path) for c in "\r\n\0"):
            raise ValueError()
        return path
    except (ValueError, OSError):
        raise AssetError("Swagger source must be the pinned HTTPS URL or an absolute file:// URL.") from None


def expected_digest(value: str | None) -> str | None:
    if value is None or not value.strip():
        return None
    value = value.strip().lower()
    if not re.fullmatch(r"[0-9a-f]{64}", value):
        raise AssetError("SWAGGER_UI_ARCHIVE_SHA256 must contain exactly 64 hexadecimal digits.")
    return value


class HttpsRedirects(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        target = urlsplit(newurl)
        if (target.scheme != "https" or target.hostname not in ("github.com", "codeload.github.com")
                or target.username or target.password or target.port not in (None, 443)):
            raise AssetError("Swagger download refused an unexpected redirect.")
        return super().redirect_request(req, fp, code, msg, headers, newurl)


def download(destination: Path) -> None:
    """Download with a few retries; github.com occasionally times out or returns a gateway error."""
    last: Exception | None = None
    for attempt in range(4):
        try:
            _download_once(destination)
            return
        except AssetError:
            raise
        except (OSError, ValueError) as error:  # URLError/HTTPError/timeouts are OSError subclasses
            last = error
            if attempt < 3:
                delay = (3, 6, 12)[attempt]
                print(f"Swagger download failed ({error}); retrying in {delay}s ({attempt + 1}/3).", file=sys.stderr)
                time.sleep(delay)
    raise AssetError(f"Swagger download failed after 4 attempts: {last}")


def _download_once(destination: Path) -> None:
    """TLS verification uses Python's default trust store; never executes a shell."""
    started = time.monotonic()
    opener = build_opener(HttpsRedirects())
    request = Request(DOWNLOAD_URL, headers={"User-Agent": "RustProxy-build"})
    with opener.open(request, timeout=60) as response, destination.open("wb") as output:
        length = response.headers.get("Content-Length")
        if length and int(length) > MAX_ARCHIVE:
            raise AssetError("Swagger archive exceeds the 64 MiB limit.")
        received = 0
        while True:
            block = response.read(128 * 1024)
            if not block:
                break
            received += len(block)
            if received > MAX_ARCHIVE or time.monotonic() - started > 180:
                raise AssetError("Swagger download exceeded the size or time limit.")
            output.write(block)


def read_assets(path: Path, checksum: str | None) -> tuple[dict[str, bytes], str]:
    if not path.is_file() or not 0 < path.stat().st_size <= MAX_ARCHIVE:
        raise AssetError("Swagger archive must be a nonempty regular file of at most 64 MiB.")
    with path.open("rb") as stream:
        snapshot = stream.read(MAX_ARCHIVE + 1)
    if not 0 < len(snapshot) <= MAX_ARCHIVE:
        raise AssetError("Swagger archive exceeds the 64 MiB limit.")

    digest = hashlib.sha256(snapshot).hexdigest()
    if checksum and digest != checksum:
        raise AssetError("Swagger archive SHA-256 mismatch; no assets were published.")
    with zipfile.ZipFile(io.BytesIO(snapshot)) as archive:
        names = archive.namelist()
        if len(names) > 20000:
            raise AssetError("Swagger archive contains too many entries.")

        def member(relative: str, limit: int, *, required: bool = True) -> bytes:
            name = PREFIX + relative
            count = names.count(name)
            if count == 0 and not required:
                return b""
            if count != 1:
                raise AssetError(f"Swagger archive must contain exactly one {name}.")
            info = archive.getinfo(name)
            mode = (info.external_attr >> 16) & 0o170000
            if info.is_dir() or mode not in (0, 0o100000) or not 0 < info.file_size <= limit:
                raise AssetError(f"Swagger archive member is not a bounded regular file: {name}.")
            # Bounded read as well as metadata validation; CRC checked by ZipFile.
            with archive.open(info) as stream:
                data = stream.read(limit + 1)
            if not 0 < len(data) <= limit or len(data) != info.file_size:
                raise AssetError(f"Swagger archive member has an invalid size: {name}.")
            data.decode("utf-8")
            return data

        package = json.loads(member("package.json", 256 * 1024))
        if not isinstance(package, dict) or package.get("name") != "swagger-ui" or package.get("version") != VERSION:
            raise AssetError(f"Swagger archive must be swagger-ui {VERSION}; older versions are not a fallback.")
        assets = {name: member(source, limit) for name, (source, limit) in ASSETS.items()}
        licenses = []
        for name in LICENSE_MEMBERS:
            text = member(name, 512 * 1024, required=name == "LICENSE")
            if text:
                licenses.append(name.encode() + b"\n" + b"-" * 40 + b"\n" + text)
        assets["licenses.txt"] = b"Swagger UI " + VERSION.encode() + b"\n\n" + b"\n\n".join(licenses)
        return assets, digest


def atomic_write(path: Path, data: bytes) -> None:
    if path.is_file() and path.read_bytes() == data:
        return  # Preserve mtime for Cargo's rerun-if-changed tracking.
    fd, name = tempfile.mkstemp(prefix=".swagger-", dir=path.parent)
    temporary = Path(name)
    try:
        with os.fdopen(fd, "wb") as stream:
            stream.write(data)
        os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)


def cached_archives(out_dir: Path, cache: Path) -> list[Path]:
    """Previously downloaded copies of the pinned archive.

    Cargo gives every package version its own OUT_DIR, so a version bump would download the same
    pinned archive again. Reuse an archive from the shared cache or from another build directory,
    after the usual validation below.
    """
    candidates: list[Path] = []
    override = os.environ.get("SWAGGER_UI_CACHE_DIR", "").strip()
    if override:
        candidates.append(Path(override) / cache.name)
    parents = out_dir.parents
    if len(parents) > 3 and parents[1].name == "build":
        candidates.append(parents[3] / "swagger-ui-cache" / cache.name)
        try:
            siblings = sorted(parents[1].glob("*/out/" + cache.name), key=lambda item: item.stat().st_mtime, reverse=True)
        except OSError:
            siblings = []
        candidates.extend(siblings)
    return [item for item in dict.fromkeys(candidates) if item != cache and item.is_file()]


def remember_archive(out_dir: Path, cache: Path) -> None:
    """Best-effort copy of a validated archive into the shared cache directory."""
    parents = out_dir.parents
    if len(parents) > 3 and parents[1].name == "build":
        shared = parents[3] / "swagger-ui-cache"
        try:
            shared.mkdir(parents=True, exist_ok=True)
            if not (shared / cache.name).exists():
                shutil.copyfile(cache, shared / cache.name)
        except OSError:
            pass


def read_vendored() -> tuple[dict[str, bytes], str, Path] | None:
    """Assets vendored in third_party/ (official swagger-ui-dist files); verified against SHA256SUMS.

    Returns None when the directory is absent so the pinned download still works for older trees.
    A present but inconsistent directory is an error, never a silent fallback.
    """
    manifest = VENDORED / "SHA256SUMS"
    if not manifest.is_file():
        return None
    expected: dict[str, str] = {}
    for line in manifest.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        digest, _, name = line.partition("  ")
        if not re.fullmatch(r"[0-9a-f]{64}", digest) or not name or "/" in name or name.startswith("."):
            raise AssetError("Vendored Swagger SHA256SUMS is malformed.")
        expected[name] = digest
    files: dict[str, bytes] = {}
    for name in ("swagger-ui-bundle.js", "swagger-ui.css", "LICENSE", "NOTICE", "swagger-ui-bundle.js.LICENSE.txt"):
        path = VENDORED / name
        if name not in expected or not path.is_file() or path.is_symlink():
            raise AssetError(f"Vendored Swagger UI is incomplete: {name} is missing or not listed in SHA256SUMS.")
        data = path.read_bytes()
        if not 0 < len(data) <= 16 * 1024 * 1024 or hashlib.sha256(data).hexdigest() != expected[name]:
            raise AssetError(f"Vendored Swagger UI file failed its SHA-256 check: {name}.")
        data.decode("utf-8")
        files[name] = data
    package = {name: files[name] for name in ASSETS}
    licenses = [label.encode() + b"\n" + b"-" * 40 + b"\n" + files[name] for label, name in (("LICENSE", "LICENSE"), ("NOTICE", "NOTICE"), ("dist/swagger-ui-bundle.js.LICENSE.txt", "swagger-ui-bundle.js.LICENSE.txt"))]
    assets = {"swagger-ui-bundle.js": package["swagger-ui-bundle.js"], "swagger-ui.css": package["swagger-ui.css"],
              "licenses.txt": b"Swagger UI " + VERSION.encode() + b"\n\n" + b"\n\n".join(licenses)}
    return assets, hashlib.sha256(manifest.read_bytes()).hexdigest(), manifest


def prepare(out_dir: Path, source: str | Path, checksum: str | None = None,
            *, offline: bool = False) -> list[Path]:
    out_dir.mkdir(parents=True, exist_ok=True)
    cache = out_dir / f"swagger-ui-{VERSION}.zip"
    if isinstance(source, Path):
        archive = source
        assets, digest = read_assets(archive, checksum)
    elif source == DOWNLOAD_URL and checksum is None and (vendored := read_vendored()) is not None:
        assets, digest, archive = vendored
        print(f"Using the vendored Swagger UI {VERSION} from {VENDORED}.", file=sys.stderr)
    elif source == DOWNLOAD_URL:
        archive = cache
        reused = None
        if cache.exists():
            assets, digest = read_assets(cache, checksum)
        else:
            for candidate in cached_archives(out_dir, cache):
                try:
                    reused = (read_assets(candidate, checksum), candidate)
                    break
                except (OSError, ValueError, KeyError, RuntimeError, zipfile.BadZipFile, zlib.error):
                    continue
        if reused is not None:
            (assets, digest), candidate = reused
            shutil.copyfile(candidate, cache)
            print(f"Reusing the cached Swagger UI archive from {candidate}.", file=sys.stderr)
        elif cache.exists():
            pass
        else:
            if offline:
                raise AssetError("Swagger assets are not cached and CARGO_NET_OFFLINE is set. " + HELP)
            fd, name = tempfile.mkstemp(prefix=".swagger-download-", suffix=".zip", dir=out_dir)
            os.close(fd)
            temporary = Path(name)
            try:
                download(temporary)
                assets, digest = read_assets(temporary, checksum)
                os.replace(temporary, cache)  # Cache only an already-validated archive.
            finally:
                temporary.unlink(missing_ok=True)
        remember_archive(out_dir, cache)
    else:
        raise AssetError("Unrecognized Swagger source.")
    destination = out_dir / "swagger-ui"
    destination.mkdir(exist_ok=True)
    outputs = []
    for name, data in assets.items():
        target = destination / name
        atomic_write(target, data)
        outputs.append(target)
    receipt = {
        "version": VERSION, "source": str(source), "archive_sha256": digest,
        "assets_sha256": {key: hashlib.sha256(data).hexdigest() for key, data in assets.items()},
    }
    receipt_path = destination / "source.json"
    atomic_write(receipt_path, (json.dumps(receipt, indent=2, sort_keys=True) + "\n").encode())
    # Watch both a supplied symlink and its resolved target, plus cached/generated files.
    inputs = [archive.absolute(), archive.resolve(), *outputs, receipt_path]
    return list(dict.fromkeys(inputs))


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--out-dir", required=True, type=Path)
    args = parser.parse_args()
    if sys.version_info < (3, 10):
        parser.error("Python 3.10+ is required to prepare the embedded Swagger UI.")
    try:
        source = source_value(os.environ.get("SWAGGER_UI_DOWNLOAD_URL"))
        checksum = expected_digest(os.environ.get("SWAGGER_UI_ARCHIVE_SHA256"))
        offline = os.environ.get("CARGO_NET_OFFLINE", "").lower() in ("1", "true")
        paths = prepare(args.out_dir.resolve(), source, checksum, offline=offline)

        import secrets
        atomic_write(args.out_dir.resolve() / "asset-build-id.txt", secrets.token_hex(8).encode("ascii"))
        for path in paths:
            if any(c in str(path) for c in "\r\n\0"):
                raise AssetError("Build input/output path contains an invalid control character.")
            print(f"cargo:rerun-if-changed={path}")
    except (OSError, ValueError, RuntimeError, zipfile.BadZipFile, zlib.error) as error:
        print(f"Swagger UI asset preparation failed: {error}\n{HELP}", file=sys.stderr)
        return 1
    return 0


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