RustProxy/scripts/migrate_npm_api.py

"""RustProxy administration API helpers for NPM migration."""
from __future__ import annotations

from migrate_npm_core import *

class RustProxyClient:
    def __init__(self, base_url: str, *, ca_file: Path | None = None, insecure_tls: bool = False):
        parsed = urllib.parse.urlsplit(base_url.rstrip("/"))
        if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.path not in {"", "/"}:
            raise MigrationError("--rustproxy must be an http(s) origin, e.g. https://rustproxy.example:9090")
        if parsed.query or parsed.fragment or parsed.username or parsed.password:
            raise MigrationError("--rustproxy must not contain credentials, query or fragment")
        self.base_url = urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, "", "", ""))
        jar = http.cookiejar.CookieJar()
        handlers: list[Any] = [urllib.request.HTTPCookieProcessor(jar)]
        if parsed.scheme == "https":
            if insecure_tls:
                context = ssl._create_unverified_context()  # Explicit operator opt-in for local/self-signed admin TLS.
            else:
                context = ssl.create_default_context(cafile=str(ca_file) if ca_file else None)
            handlers.append(urllib.request.HTTPSHandler(context=context))
        self.opener = urllib.request.build_opener(*handlers)
        self.csrf = ""

    def request(self, method: str, path: str, body: dict[str, Any] | None = None, *, authenticated_write: bool = False) -> Any:
        data = None if body is None else json.dumps(body, separators=(",", ":")).encode()
        headers = {"Accept": "application/json", "X-RustProxy-Request": "1"}
        if data is not None:
            headers["Content-Type"] = "application/json"
        if method != "GET":
            headers["Origin"] = self.base_url
        if authenticated_write:
            headers["X-CSRF-Token"] = self.csrf
        request = urllib.request.Request(self.base_url + path, data=data, headers=headers, method=method)
        try:
            with self.opener.open(request, timeout=10) as response:
                payload = response.read(1024 * 1024)
        except urllib.error.HTTPError as error:
            payload = error.read(65536)
            try:
                detail = json.loads(payload).get("error", payload.decode("utf-8", "replace"))
            except Exception:
                detail = payload.decode("utf-8", "replace")
            raise MigrationError(f"RustProxy API {method} {path} failed with HTTP {error.code}: {detail}") from error
        except (OSError, urllib.error.URLError) as error:
            raise MigrationError(f"RustProxy API connection failed: {error}") from error
        try:
            return json.loads(payload)
        except json.JSONDecodeError as error:
            raise MigrationError(f"RustProxy API returned invalid JSON for {path}") from error

    def login(self, username: str, password: str) -> None:
        info = self.request("POST", "/api/auth/login", {"username": username, "password": password})
        token = info.get("csrf_token") if isinstance(info, dict) else None
        if not isinstance(token, str) or not token:
            session = self.request("GET", "/api/auth/session")
            token = session.get("csrf_token") if isinstance(session, dict) else None
        if not isinstance(token, str) or not token:
            raise MigrationError("RustProxy login succeeded but no CSRF token was returned")
        self.csrf = token

    def config(self) -> dict[str, Any]:
        value = self.request("GET", "/api/config")
        if not isinstance(value, dict) or not isinstance(value.get("revision"), int) or not isinstance(value.get("config"), dict):
            raise MigrationError("unexpected RustProxy /api/config response")
        return value

    def apply(self, revision: int, config: dict[str, Any]) -> dict[str, Any]:
        value = self.request("PUT", "/api/config", {"expected_revision": revision, "config": config}, authenticated_write=True)
        if not isinstance(value, dict):
            raise MigrationError("unexpected RustProxy PUT /api/config response")
        return value



def write_private_json(path: Path, value: dict[str, Any]) -> None:
    path = path.expanduser()
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.is_symlink():
        raise MigrationError(f"refusing to write migration plan through symlink: {path}")
    flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    fd = os.open(path, flags, 0o600)
    try:
        os.fchmod(fd, 0o600)
        payload = (json.dumps(value, indent=2, ensure_ascii=False) + "\n").encode("utf-8")
        with os.fdopen(fd, "wb", closefd=False) as stream:
            stream.write(payload)
            stream.flush()
            os.fsync(stream.fileno())
    finally:
        os.close(fd)

def split_ids(values: list[str] | None) -> list[str] | None:
    if values is None:
        return None
    out: list[str] = []
    for value in values:
        for item in value.split(","):
            item = item.strip()
            if item and item not in out:
                out.append(item)
    return out


def print_report(decisions: list[Decision]) -> None:
    imported = sum(1 for d in decisions if d.action == "import")
    skipped = sum(1 for d in decisions if d.action == "skip")
    print(f"NPM proxy hosts: {len(decisions)}; import: {imported}; skip: {skipped}")
    for decision in decisions:
        print(f"\n[{decision.action.upper()}] NPM #{decision.npm_id}")
        if decision.hosts:
            print("  hosts:   " + ", ".join(decision.hosts))
        if decision.backend:
            print("  backend: " + decision.backend)
        if decision.rustproxy_ids:
            print("  ids:     " + ", ".join(decision.rustproxy_ids))
        if decision.reason:
            print("  reason:  " + decision.reason)
        for warning in decision.warnings or []:
            print("  warning: " + warning)