319 lines
13 KiB
Python
319 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import quote, urljoin, urlsplit, urlunsplit
|
|
from urllib.request import HTTPRedirectHandler, Request, build_opener
|
|
|
|
from packaging.version import InvalidVersion, Version
|
|
|
|
from .validation import (
|
|
ValidationError,
|
|
validate_distribution_name,
|
|
validate_import_name,
|
|
validate_public_url,
|
|
validate_sha256,
|
|
validate_slug,
|
|
)
|
|
|
|
|
|
class StoreClientError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class URLPolicy:
|
|
"""Exact scheme/host/port allowlist with an optional path prefix."""
|
|
|
|
def __init__(self, allowed_roots: tuple[str, ...] | list[str]):
|
|
if not allowed_roots:
|
|
raise ValueError("At least one allowed URL is required.")
|
|
if any(urlsplit(root).query or urlsplit(root).fragment for root in allowed_roots):
|
|
raise ValueError("Allowlist roots may not contain a query string or fragment.")
|
|
self._roots = tuple(self._normalize(validate_public_url(root)) for root in allowed_roots)
|
|
|
|
@staticmethod
|
|
def _normalize(value: str) -> tuple[str, str, int, str]:
|
|
parsed = urlsplit(value)
|
|
scheme = parsed.scheme.lower()
|
|
default_port = 443 if scheme == "https" else 80
|
|
port = parsed.port or default_port
|
|
prefix = parsed.path.rstrip("/") or "/"
|
|
return scheme, parsed.hostname.lower(), port, prefix
|
|
|
|
def check(self, value: str) -> str:
|
|
validate_public_url(value)
|
|
parsed = urlsplit(value)
|
|
candidate = self._normalize(value)
|
|
scheme, host, port, path = candidate
|
|
for allowed_scheme, allowed_host, allowed_port, prefix in self._roots:
|
|
path_ok = prefix == "/" or path == prefix or path.startswith(prefix + "/")
|
|
if (scheme, host, port) == (allowed_scheme, allowed_host, allowed_port) and path_ok:
|
|
return value
|
|
raise StoreClientError("URL is outside the configured allowlist.")
|
|
|
|
|
|
class _PolicyRedirectHandler(HTTPRedirectHandler):
|
|
def __init__(self, policy: URLPolicy):
|
|
self.policy = policy
|
|
super().__init__()
|
|
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
self.policy.check(newurl)
|
|
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Release:
|
|
version: str
|
|
download_url: str
|
|
sha256: str
|
|
min_netbox_version: str
|
|
max_netbox_version: str
|
|
published_at: str
|
|
approved: bool
|
|
immutable: bool
|
|
approved_payload_sha256: str
|
|
artifact_kind: str = "wheel"
|
|
artifact_filename: str = ""
|
|
|
|
@classmethod
|
|
def from_mapping(cls, value: dict[str, Any]) -> "Release":
|
|
version = str(value.get("version", "")).strip()
|
|
try:
|
|
Version(version)
|
|
except InvalidVersion as exc:
|
|
raise StoreClientError("Store returned an invalid release version.") from exc
|
|
digest = str(value.get("sha256") or "").strip()
|
|
if digest:
|
|
digest = validate_sha256(digest)
|
|
approved_payload_sha256 = str(value.get("approved_payload_sha256") or "").strip()
|
|
if approved_payload_sha256:
|
|
approved_payload_sha256 = validate_sha256(approved_payload_sha256)
|
|
download_url = str(value.get("download_url") or value.get("artifact_url") or "").strip()
|
|
if download_url:
|
|
validate_public_url(download_url)
|
|
return cls(
|
|
version=version,
|
|
download_url=download_url,
|
|
sha256=digest,
|
|
min_netbox_version=str(value.get("min_netbox_version") or "").strip(),
|
|
max_netbox_version=str(value.get("max_netbox_version") or "").strip(),
|
|
published_at=str(value.get("published_at") or "").strip(),
|
|
approved=value.get("approved") is True or value.get("status") == "approved",
|
|
immutable=value.get("immutable") is True,
|
|
approved_payload_sha256=approved_payload_sha256,
|
|
artifact_kind=str(value.get("artifact_kind") or "wheel").strip(),
|
|
artifact_filename=str(value.get("artifact_filename") or "").strip(),
|
|
)
|
|
|
|
def supports(self, netbox_version: str, plugin: "CatalogPlugin") -> bool:
|
|
try:
|
|
current = Version(netbox_version)
|
|
minimum = Version(self.min_netbox_version or plugin.min_netbox_version or "0")
|
|
maximum = Version(self.max_netbox_version or plugin.max_netbox_version or "999999")
|
|
except InvalidVersion:
|
|
return False
|
|
return minimum <= current <= maximum
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CatalogPlugin:
|
|
slug: str
|
|
name: str
|
|
summary: str
|
|
description: str
|
|
repository_url: str
|
|
latest_version: str
|
|
package_name: str
|
|
import_name: str
|
|
min_netbox_version: str
|
|
max_netbox_version: str
|
|
approved: bool
|
|
releases: tuple[Release, ...]
|
|
|
|
@classmethod
|
|
def from_mapping(cls, value: dict[str, Any]) -> "CatalogPlugin":
|
|
try:
|
|
slug = validate_slug(str(value.get("slug", "")))
|
|
package_name = validate_distribution_name(str(value.get("package_name", "")))
|
|
import_name = validate_import_name(str(value.get("import_name", "")))
|
|
except ValidationError as exc:
|
|
raise StoreClientError(str(exc)) from exc
|
|
releases_raw = value.get("releases") or []
|
|
if not isinstance(releases_raw, list):
|
|
raise StoreClientError("Store returned an invalid releases collection.")
|
|
releases = tuple(Release.from_mapping(item) for item in releases_raw if isinstance(item, dict))
|
|
repository_url = str(value.get("repository_url") or "").strip()
|
|
if repository_url:
|
|
validate_public_url(repository_url)
|
|
return cls(
|
|
slug=slug,
|
|
name=str(value.get("name") or slug)[:200],
|
|
summary=str(value.get("summary") or "")[:2_000],
|
|
description=str(value.get("description") or "")[:250_000],
|
|
repository_url=repository_url,
|
|
latest_version=str(value.get("latest_version") or "").strip(),
|
|
package_name=package_name,
|
|
import_name=import_name,
|
|
min_netbox_version=str(value.get("min_netbox_version") or "").strip(),
|
|
max_netbox_version=str(value.get("max_netbox_version") or "").strip(),
|
|
approved=value.get("approved") is True or value.get("status") == "approved",
|
|
releases=releases,
|
|
)
|
|
|
|
def select_release(self, netbox_version: str, requested_version: str = "") -> Release:
|
|
compatible = list(self.installable_releases(netbox_version))
|
|
if requested_version:
|
|
compatible = [release for release in compatible if release.version == requested_version]
|
|
elif self.latest_version:
|
|
latest = [release for release in compatible if release.version == self.latest_version]
|
|
if latest:
|
|
compatible = latest
|
|
if not compatible:
|
|
raise StoreClientError("No compatible release is available for this NetBox version.")
|
|
return max(compatible, key=lambda release: Version(release.version))
|
|
|
|
def installable_releases(
|
|
self,
|
|
netbox_version: str,
|
|
*,
|
|
require_approval_marker: bool = False,
|
|
) -> tuple[Release, ...]:
|
|
return tuple(
|
|
release
|
|
for release in self.releases
|
|
if (
|
|
release.supports(netbox_version, self)
|
|
and release.approved
|
|
and release.immutable
|
|
and bool(release.download_url)
|
|
and bool(release.sha256)
|
|
and release.artifact_kind in {"wheel", "source_archive"}
|
|
and (not require_approval_marker or bool(release.approved_payload_sha256))
|
|
)
|
|
)
|
|
|
|
|
|
class StoreClient:
|
|
API_LIMIT = 5 * 1024 * 1024
|
|
MAX_PAGES = 50
|
|
|
|
def __init__(
|
|
self,
|
|
store_url: str,
|
|
allowed_store_urls: tuple[str, ...],
|
|
allowed_artifact_urls: tuple[str, ...],
|
|
*,
|
|
api_token: str = "",
|
|
timeout: int = 15,
|
|
):
|
|
self.store_policy = URLPolicy(allowed_store_urls)
|
|
self.artifact_policy = URLPolicy(allowed_artifact_urls)
|
|
if urlsplit(store_url).query or urlsplit(store_url).fragment:
|
|
raise StoreClientError("Store base URL may not contain a query string or fragment.")
|
|
self.store_url = self.store_policy.check(store_url.rstrip("/"))
|
|
# API authentication may only follow redirects below the configured base URL.
|
|
self.api_policy = URLPolicy([self.store_url])
|
|
self.api_token = api_token
|
|
self.timeout = timeout
|
|
|
|
def _request_json(self, url: str) -> Any:
|
|
self.api_policy.check(url)
|
|
headers = {"Accept": "application/json", "User-Agent": "netbox-plugin-store/0.1"}
|
|
if self.api_token:
|
|
headers["Authorization"] = f"Bearer {self.api_token}"
|
|
request = Request(url, headers=headers, method="GET")
|
|
opener = build_opener(_PolicyRedirectHandler(self.api_policy))
|
|
try:
|
|
with opener.open(request, timeout=self.timeout) as response:
|
|
length = response.headers.get("Content-Length")
|
|
if length and int(length) > self.API_LIMIT:
|
|
raise StoreClientError("Store response exceeds the size limit.")
|
|
body = response.read(self.API_LIMIT + 1)
|
|
except (HTTPError, URLError, TimeoutError, OSError) as exc:
|
|
raise StoreClientError(f"Store request failed: {type(exc).__name__}") from exc
|
|
if len(body) > self.API_LIMIT:
|
|
raise StoreClientError("Store response exceeds the size limit.")
|
|
try:
|
|
return json.loads(body.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise StoreClientError("Store returned invalid JSON.") from exc
|
|
|
|
def list_plugins(self) -> list[CatalogPlugin]:
|
|
next_url: str | None = self.store_url + "/api/v1/plugins/"
|
|
plugins: list[CatalogPlugin] = []
|
|
pages = 0
|
|
while next_url:
|
|
pages += 1
|
|
if pages > self.MAX_PAGES:
|
|
raise StoreClientError("Store pagination limit exceeded.")
|
|
payload = self._request_json(next_url)
|
|
if isinstance(payload, list):
|
|
items = payload
|
|
next_url = None
|
|
elif isinstance(payload, dict):
|
|
items = payload.get("results", payload.get("plugins", []))
|
|
raw_next = payload.get("next")
|
|
next_url = urljoin(next_url, str(raw_next)) if raw_next else None
|
|
if next_url:
|
|
self.api_policy.check(next_url)
|
|
else:
|
|
raise StoreClientError("Store returned an invalid catalog payload.")
|
|
if not isinstance(items, list):
|
|
raise StoreClientError("Store returned an invalid plugin collection.")
|
|
plugins.extend(CatalogPlugin.from_mapping(item) for item in items if isinstance(item, dict))
|
|
return plugins
|
|
|
|
def get_plugin(self, slug: str) -> CatalogPlugin:
|
|
slug = validate_slug(slug)
|
|
payload = self._request_json(self.store_url + f"/api/v1/plugins/{quote(slug)}/")
|
|
if not isinstance(payload, dict):
|
|
raise StoreClientError("Store returned an invalid plugin payload.")
|
|
return CatalogPlugin.from_mapping(payload)
|
|
|
|
def download_artifact(
|
|
self,
|
|
url: str,
|
|
destination: Path,
|
|
*,
|
|
expected_sha256: str,
|
|
timeout: int,
|
|
max_bytes: int,
|
|
) -> tuple[str, int]:
|
|
self.artifact_policy.check(url)
|
|
expected = validate_sha256(expected_sha256)
|
|
request = Request(
|
|
url,
|
|
headers={"Accept": "application/octet-stream", "User-Agent": "netbox-plugin-store/0.1"},
|
|
method="GET",
|
|
)
|
|
opener = build_opener(_PolicyRedirectHandler(self.artifact_policy))
|
|
digest = hashlib.sha256()
|
|
written = 0
|
|
try:
|
|
with opener.open(request, timeout=timeout) as response, destination.open("xb") as target:
|
|
length = response.headers.get("Content-Length")
|
|
if length and int(length) > max_bytes:
|
|
raise StoreClientError("Artifact exceeds the configured size limit.")
|
|
while chunk := response.read(1024 * 1024):
|
|
written += len(chunk)
|
|
if written > max_bytes:
|
|
raise StoreClientError("Artifact exceeds the configured size limit.")
|
|
digest.update(chunk)
|
|
target.write(chunk)
|
|
except StoreClientError:
|
|
destination.unlink(missing_ok=True)
|
|
raise
|
|
except (HTTPError, URLError, TimeoutError, OSError) as exc:
|
|
destination.unlink(missing_ok=True)
|
|
raise StoreClientError(f"Artifact download failed: {type(exc).__name__}") from exc
|
|
actual = digest.hexdigest()
|
|
if actual != expected:
|
|
destination.unlink(missing_ok=True)
|
|
raise StoreClientError("Artifact SHA-256 verification failed.")
|
|
return actual, written
|