"""Curated openHop plugin catalogue client (static JSON over HTTPS).""" from __future__ import annotations import hashlib import json import logging import os import re import socket import ssl import tempfile import threading import time import urllib.error import urllib.request from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Optional from urllib.parse import urlsplit logger = logging.getLogger("PluginCatalogue") DEFAULT_CATALOGUE_URL = "https://repeater-plugins.openhop.dev/catalogue.json" DEFAULT_CACHE_TTL_SECONDS = 600 # 10 minutes SUPPORTED_SCHEMAS = (1, 2) USER_AGENT = "openhop-repeater-plugin-manager/1.0" GITHUB_RELEASE_ORIGIN = "https://github.com" GITHUB_RELEASE_ASSET_HOST = "release-assets.githubusercontent.com" MAX_WHEEL_BYTES = 100 * 1024 * 1024 MAX_JSON_BYTES = 4 * 1024 * 1024 JSON_DOWNLOAD_TIMEOUT_SECONDS = 30.0 WHEEL_DOWNLOAD_TIMEOUT_SECONDS = 120.0 def _bounded_chunks(response, *, max_bytes: int, deadline: float): """Stream a bounded body, sharing the budget started before opening the URL. HTTPResponse.read(n) can wait for n bytes despite a trickling peer. read1 returns after one buffered/socket read; resetting the socket timeout bounds an idle final read by the *remaining* budget. The shutdown watchdog also interrupts read1's internal chunk-header/trailer readline loops. urllib's DNS, connect, TLS and response-header/redirect processing happen before the response is exposed here: their timeout is not a strict total deadline. Injected non-socket openers must themselves provide bounded reads. """ def remaining(): budget = deadline - time.monotonic() if budget <= 0: raise TimeoutError("download deadline exceeded") return budget raw = getattr(getattr(response, "fp", None), "raw", None) sock = getattr(raw, "_sock", None) watchdog = None if isinstance(sock, socket.socket): def interrupt(): try: sock.shutdown(socket.SHUT_RDWR) except OSError: pass # Response may already have closed its socket at EOF. watchdog = threading.Timer(remaining(), interrupt) watchdog.daemon = True watchdog.start() read = getattr(response, "read1", None) or response.read total = 0 try: while True: budget = remaining() if isinstance(sock, socket.socket) and sock.fileno() >= 0: sock.settimeout(budget) chunk = read(min(64 * 1024, max_bytes - total + 1)) remaining() if not chunk: if getattr(response, "length", None): raise ValueError("download ended before Content-Length was received") return total += len(chunk) if total > max_bytes: raise ValueError(f"download exceeds {max_bytes} bytes") yield chunk finally: if watchdog is not None: watchdog.cancel() watchdog.join() _REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") _ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$") _SHA256_RE = re.compile(r"^[a-f0-9]{64}$") _ssl_ctx: Optional[ssl.SSLContext] = None class CatalogueError(Exception): def __init__(self, message: str, *, code: int = 502): super().__init__(message) self.code = code def _ssl_context() -> ssl.SSLContext: global _ssl_ctx if _ssl_ctx is None: _ssl_ctx = ssl.create_default_context() return _ssl_ctx def _validate_approved_wheel_url(repository: str, version: str, wheel_url: str) -> None: parsed = urlsplit(wheel_url) origin = f"{parsed.scheme}://{parsed.netloc}" expected_prefix = f"/{repository}/releases/download/" path_tail = ( parsed.path[len(expected_prefix) :] if parsed.path.startswith(expected_prefix) else "" ) path_parts = path_tail.split("/") filename = path_parts[-1] if path_parts else "" if ( origin != GITHUB_RELEASE_ORIGIN or len(path_parts) != 2 or not path_parts[0] or not filename.endswith(".whl") or f"-{version}-" not in filename or parsed.query or parsed.fragment ): raise CatalogueError( f"wheel_url must be a version-matched GitHub Release wheel in {repository}", code=400, ) def _validate_redirect_target(source_url: str, target_url: str) -> None: source = urlsplit(source_url) target = urlsplit(target_url) same_origin = ( source.scheme == "https" and target.scheme == "https" and source.netloc == target.netloc ) github_asset_redirect = ( source.scheme == "https" and source.hostname in {"github.com", GITHUB_RELEASE_ASSET_HOST} and target.scheme == "https" and target.hostname == GITHUB_RELEASE_ASSET_HOST and target.port is None ) if not (same_origin or github_asset_redirect): raise CatalogueError( f"download redirect is not allowed: {source_url} -> {target_url}", code=502, ) def _validate_download_response_url(source_url: str, final_url: str) -> None: if final_url == source_url: return _validate_redirect_target(source_url, final_url) class _ApprovedRedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): _validate_redirect_target(req.full_url, newurl) return super().redirect_request(req, fp, code, msg, headers, newurl) @dataclass(frozen=True) class CataloguePlugin: id: str name: str description: str repository: str category: str = "" logo: str = "" homepage: str = "" version: Optional[str] = None wheel_url: Optional[str] = None sha256: Optional[str] = None @property def has_approved_wheel(self) -> bool: return bool(self.version and self.wheel_url and self.sha256) def to_dict(self) -> dict[str, Any]: out: dict[str, Any] = { "id": self.id, "name": self.name, "description": self.description, "repository": self.repository, } if self.category: out["category"] = self.category if self.logo: out["logo"] = self.logo if self.homepage: out["homepage"] = self.homepage if self.version: out["version"] = self.version if self.wheel_url: out["wheel_url"] = self.wheel_url if self.sha256: out["sha256"] = self.sha256 return out @dataclass(frozen=True) class Catalogue: schema: int plugins: tuple[CataloguePlugin, ...] def get(self, plugin_id: str) -> Optional[CataloguePlugin]: for plugin in self.plugins: if plugin.id == plugin_id: return plugin return None def to_dict(self) -> dict[str, Any]: return { "schema": self.schema, "plugins": [p.to_dict() for p in self.plugins], } def parse_catalogue(data: Any) -> Catalogue: if not isinstance(data, dict): raise CatalogueError("catalogue must be a JSON object", code=400) schema = data.get("schema") if schema not in SUPPORTED_SCHEMAS: raise CatalogueError( f"unsupported catalogue schema {schema!r}; expected one of {SUPPORTED_SCHEMAS}", code=400, ) raw_plugins = data.get("plugins") if not isinstance(raw_plugins, list): raise CatalogueError("catalogue.plugins must be an array", code=400) plugins: list[CataloguePlugin] = [] seen_ids: set[str] = set() seen_repos: set[str] = set() seen_wheel_urls: set[str] = set() for idx, item in enumerate(raw_plugins): if not isinstance(item, dict): raise CatalogueError(f"plugins[{idx}] must be an object", code=400) plugin_id = item.get("id") name = item.get("name") description = item.get("description", "") repository = item.get("repository") category = item.get("category", "") or "" if not isinstance(plugin_id, str) or not plugin_id.strip(): raise CatalogueError(f"plugins[{idx}].id is required", code=400) plugin_id = plugin_id.strip() if not _ID_RE.match(plugin_id): raise CatalogueError(f"plugins[{idx}].id is invalid: {plugin_id!r}", code=400) if plugin_id in seen_ids: raise CatalogueError(f"duplicate plugin id: {plugin_id}", code=400) if not isinstance(name, str) or not name.strip(): raise CatalogueError(f"plugins[{idx}].name is required", code=400) if description is None: description = "" if not isinstance(description, str): raise CatalogueError(f"plugins[{idx}].description must be a string", code=400) if not isinstance(repository, str) or not repository.strip(): raise CatalogueError(f"plugins[{idx}].repository is required", code=400) repository = repository.strip() if not _REPO_RE.match(repository): raise CatalogueError( f"plugins[{idx}].repository must be owner/repo: {repository!r}", code=400, ) if repository in seen_repos: raise CatalogueError(f"duplicate repository: {repository}", code=400) if category is None: category = "" if not isinstance(category, str): raise CatalogueError(f"plugins[{idx}].category must be a string", code=400) logo = item.get("logo", "") or "" if logo is None: logo = "" if not isinstance(logo, str): raise CatalogueError(f"plugins[{idx}].logo must be a string", code=400) logo = logo.strip() if logo and not logo.startswith("https://"): raise CatalogueError( f"plugins[{idx}].logo must be an https:// URL", code=400, ) homepage = item.get("homepage", "") or "" if homepage is None: homepage = "" if not isinstance(homepage, str): raise CatalogueError(f"plugins[{idx}].homepage must be a string", code=400) homepage = homepage.strip() if homepage and not homepage.startswith("https://"): raise CatalogueError( f"plugins[{idx}].homepage must be an https:// URL", code=400, ) version = None wheel_url = None sha256 = None if schema == 2: raw_version = item.get("version") raw_wheel_url = item.get("wheel_url") raw_sha256 = item.get("sha256") if not isinstance(raw_version, str) or not raw_version.strip(): raise CatalogueError(f"plugins[{idx}].version is required", code=400) if not isinstance(raw_wheel_url, str) or not raw_wheel_url.strip(): raise CatalogueError(f"plugins[{idx}].wheel_url is required", code=400) if not isinstance(raw_sha256, str) or not _SHA256_RE.fullmatch(raw_sha256): raise CatalogueError( f"plugins[{idx}].sha256 must be a lowercase SHA-256 digest", code=400, ) version = raw_version.strip() wheel_url = raw_wheel_url.strip() sha256 = raw_sha256 try: _validate_approved_wheel_url(repository, version, wheel_url) except CatalogueError as exc: raise CatalogueError(f"plugins[{idx}].{exc}", code=exc.code) from exc if wheel_url in seen_wheel_urls: raise CatalogueError(f"duplicate wheel URL: {wheel_url}", code=400) seen_wheel_urls.add(wheel_url) seen_ids.add(plugin_id) seen_repos.add(repository) plugins.append( CataloguePlugin( id=plugin_id, name=name.strip(), description=description.strip(), repository=repository, category=category.strip(), logo=logo, homepage=homepage, version=version, wheel_url=wheel_url, sha256=sha256, ) ) return Catalogue(schema=int(schema), plugins=tuple(plugins)) class CatalogueClient: """Fetch, cache, and download from the curated plugin catalogue. Configured catalogue URLs must use HTTPS; HTTP and other schemes are rejected with code 400, not silently upgraded. Custom HTTPS origins, ports, paths and query strings remain supported. """ def __init__( self, url: str = DEFAULT_CATALOGUE_URL, *, cache_ttl: float = DEFAULT_CACHE_TTL_SECONDS, opener: Optional[Callable[..., Any]] = None, user_agent: str = USER_AGENT, ): self.url = str(url or DEFAULT_CATALOGUE_URL).strip() or DEFAULT_CATALOGUE_URL # Reject insecure configuration; never silently rewrite HTTP to HTTPS. parsed_url = urlsplit(self.url) if parsed_url.scheme != "https" or not parsed_url.hostname: raise CatalogueError("Plugin catalogue URL must use HTTPS", code=400) self.cache_ttl = cache_ttl self._opener = opener or self._default_open self.user_agent = user_agent self._cached_at: float = 0.0 self._cached: Optional[Catalogue] = None def _default_open(self, request: urllib.request.Request, timeout: float = 30.0): opener = urllib.request.build_opener( urllib.request.HTTPSHandler(context=_ssl_context()), _ApprovedRedirectHandler(), ) return opener.open(request, timeout=timeout) # nosec B310 - validated HTTPS URLs def clear_cache(self) -> None: self._cached = None self._cached_at = 0.0 def fetch(self, *, force_refresh: bool = False) -> Catalogue: now = time.monotonic() if ( not force_refresh and self._cached is not None and (now - self._cached_at) < self.cache_ttl ): return self._cached request = urllib.request.Request( self.url, headers={ "User-Agent": self.user_agent, "Accept": "application/json", }, method="GET", ) deadline = time.monotonic() + JSON_DOWNLOAD_TIMEOUT_SECONDS try: with self._opener(request, timeout=JSON_DOWNLOAD_TIMEOUT_SECONDS) as resp: body = bytearray() for chunk in _bounded_chunks(resp, max_bytes=MAX_JSON_BYTES, deadline=deadline): body.extend(chunk) except (ValueError, TimeoutError) as exc: raise CatalogueError(f"Plugin catalogue download failed: {exc}", code=502) from exc except urllib.error.HTTPError as exc: raise CatalogueError( f"Plugin catalogue is currently unavailable (HTTP {exc.code})", code=502, ) from exc except urllib.error.URLError as exc: raise CatalogueError( "Plugin catalogue is currently unavailable", code=502, ) from exc except Exception as exc: raise CatalogueError( "Plugin catalogue is currently unavailable", code=502, ) from exc try: data = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise CatalogueError(f"invalid catalogue JSON: {exc}", code=502) from exc catalogue = parse_catalogue(data) self._cached = catalogue self._cached_at = now return catalogue def get_plugin(self, plugin_id: str, *, force_refresh: bool = False) -> CataloguePlugin: catalogue = self.fetch(force_refresh=force_refresh) plugin = catalogue.get(plugin_id) if plugin is None: raise CatalogueError(f"plugin not in catalogue: {plugin_id}", code=404) return plugin def download_wheel(self, plugin: CataloguePlugin, dest_dir: Path | str) -> Path: """Download and verify one schema-2 approved GitHub Release wheel.""" if not plugin.version or not plugin.wheel_url or not plugin.sha256: raise CatalogueError( f"catalogue entry {plugin.id} does not contain approved wheel metadata", code=400, ) _validate_approved_wheel_url(plugin.repository, plugin.version, plugin.wheel_url) if not _SHA256_RE.fullmatch(plugin.sha256): raise CatalogueError(f"invalid approved checksum for {plugin.id}", code=400) destination_dir = Path(dest_dir) destination_dir.mkdir(parents=True, exist_ok=True) destination = destination_dir / Path(urlsplit(plugin.wheel_url).path).name fd, temporary_name = tempfile.mkstemp(prefix=".wheel-", dir=str(destination_dir)) digest = hashlib.sha256() total = 0 request = urllib.request.Request( plugin.wheel_url, headers={ "User-Agent": self.user_agent, "Accept": "application/octet-stream", }, method="GET", ) try: with os.fdopen(fd, "wb") as output: try: deadline = time.monotonic() + WHEEL_DOWNLOAD_TIMEOUT_SECONDS with self._opener(request, timeout=WHEEL_DOWNLOAD_TIMEOUT_SECONDS) as response: final_url = getattr(response, "geturl", lambda: plugin.wheel_url)() _validate_download_response_url( plugin.wheel_url, final_url or plugin.wheel_url, ) for chunk in _bounded_chunks( response, max_bytes=MAX_WHEEL_BYTES, deadline=deadline ): total += len(chunk) digest.update(chunk) output.write(chunk) except (ValueError, TimeoutError) as exc: raise CatalogueError( f"approved plugin wheel download failed: {exc}", code=502 ) from exc except CatalogueError: raise except urllib.error.HTTPError as exc: raise CatalogueError( f"approved plugin wheel is unavailable (HTTP {exc.code})", code=502 ) from exc except urllib.error.URLError as exc: raise CatalogueError("approved plugin wheel is unavailable", code=502) from exc except Exception as exc: raise CatalogueError("approved plugin wheel is unavailable", code=502) from exc if total == 0: raise CatalogueError("approved plugin wheel is empty", code=502) if digest.hexdigest() != plugin.sha256: raise CatalogueError( f"approved plugin wheel checksum mismatch for {plugin.id}", code=502 ) os.replace(temporary_name, destination) return destination finally: try: Path(temporary_name).unlink(missing_ok=True) except OSError as exc: logger.debug("Could not remove temporary wheel %s: %s", temporary_name, exc)