233 lines
10 KiB
Python
Executable File
233 lines
10 KiB
Python
Executable File
#!/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 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.32.11"
|
|
DOWNLOAD_URL = f"https://github.com/swagger-api/swagger-ui/archive/refs/tags/v{VERSION}.zip"
|
|
PREFIX = 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.32.11.zip "
|
|
"to a reviewed copy of the same tag archive. See docs/API_ACCESS.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:
|
|
"""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/0.26.6"})
|
|
with opener.open(request, timeout=30) 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.")
|
|
# Parse exactly the same bounded bytes that were hashed, not a reopened file
|
|
# that could have changed between verification and ZIP member reads.
|
|
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 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:
|
|
archive = cache
|
|
if cache.exists():
|
|
assets, digest = read_assets(cache, checksum)
|
|
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)
|
|
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)
|
|
for path in paths:
|
|
# These values are Cargo directives, so never allow extra output lines.
|
|
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())
|