450 lines
12 KiB
Python
450 lines
12 KiB
Python
import os
|
|
from io import BytesIO
|
|
from pathlib import PurePosixPath
|
|
from typing import Any
|
|
from zipfile import BadZipFile, ZipFile
|
|
|
|
import requests
|
|
import urllib3
|
|
|
|
from .adapters import Endpoint
|
|
from .certificates.material import CertificateMaterial
|
|
|
|
|
|
class NpmClient:
|
|
"""Authenticated client for the Nginx Proxy Manager API."""
|
|
|
|
def __init__(
|
|
self,
|
|
config,
|
|
endpoint: Endpoint | None = None,
|
|
):
|
|
self.config = config
|
|
self.endpoint = endpoint or config.npm_endpoints[0]
|
|
|
|
if self.endpoint.plugin != "npm":
|
|
raise RuntimeError("NpmClient requires an npm endpoint")
|
|
|
|
self.identity = os.getenv("NPM_IDENTITY")
|
|
self.secret = os.getenv("NPM_SECRET")
|
|
|
|
if not self.identity:
|
|
raise RuntimeError("NPM_IDENTITY is required")
|
|
|
|
if not self.secret:
|
|
raise RuntimeError("NPM_SECRET is required")
|
|
|
|
self.session = requests.Session()
|
|
self.token: str | None = None
|
|
|
|
if not config.proxy_verify_tls:
|
|
urllib3.disable_warnings(
|
|
urllib3.exceptions.InsecureRequestWarning
|
|
)
|
|
|
|
@property
|
|
def api_url(self) -> str:
|
|
return (
|
|
f"{self.config.proxy_api_scheme}://"
|
|
f"{self.endpoint.address}/api"
|
|
)
|
|
|
|
def login(self) -> None:
|
|
response = self.session.post(
|
|
f"{self.api_url}/tokens",
|
|
json={
|
|
"identity": self.identity,
|
|
"secret": self.secret,
|
|
},
|
|
timeout=15,
|
|
verify=self.config.proxy_verify_tls,
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
raise RuntimeError(
|
|
"NPM login failed: "
|
|
f"HTTP {response.status_code}: "
|
|
f"{response.text[:300]}"
|
|
)
|
|
|
|
data = response.json()
|
|
token = data.get("token")
|
|
|
|
if not token:
|
|
raise RuntimeError(
|
|
"NPM login succeeded but no token was returned"
|
|
)
|
|
|
|
self.token = str(token)
|
|
|
|
self.session.headers.update(
|
|
{
|
|
"Authorization": f"Bearer {self.token}",
|
|
"Accept": "application/json",
|
|
}
|
|
)
|
|
|
|
def _ensure_login(self) -> None:
|
|
if not self.token:
|
|
self.login()
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
expected_status: int | tuple[int, ...] = 200,
|
|
**kwargs,
|
|
) -> requests.Response:
|
|
self._ensure_login()
|
|
|
|
if isinstance(expected_status, int):
|
|
expected = (expected_status,)
|
|
else:
|
|
expected = expected_status
|
|
|
|
response = self.session.request(
|
|
method,
|
|
f"{self.api_url}{path}",
|
|
timeout=kwargs.pop("timeout", 15),
|
|
verify=self.config.proxy_verify_tls,
|
|
**kwargs,
|
|
)
|
|
|
|
if response.status_code not in expected:
|
|
raise RuntimeError(
|
|
f"NPM {method.upper()} {path} failed: "
|
|
f"HTTP {response.status_code}: "
|
|
f"{response.text[:300]}"
|
|
)
|
|
|
|
return response
|
|
|
|
def get_json(
|
|
self,
|
|
path: str,
|
|
*,
|
|
expected_status: int | tuple[int, ...] = 200,
|
|
) -> Any:
|
|
response = self.request(
|
|
"GET",
|
|
path,
|
|
expected_status=expected_status,
|
|
)
|
|
return response.json()
|
|
|
|
@staticmethod
|
|
def _json_object(
|
|
response: requests.Response,
|
|
label: str,
|
|
) -> dict[str, Any]:
|
|
data = response.json()
|
|
if not isinstance(data, dict):
|
|
raise RuntimeError(
|
|
f"NPM {label} response was not an object"
|
|
)
|
|
return data
|
|
|
|
@staticmethod
|
|
def _local_id(value: str | int, label: str) -> str:
|
|
if isinstance(value, bool):
|
|
raise RuntimeError(f"NPM {label} id is invalid")
|
|
|
|
local_id = str(value).strip()
|
|
if not local_id or not local_id.isdigit():
|
|
raise RuntimeError(f"NPM {label} id is invalid")
|
|
|
|
return local_id
|
|
|
|
def proxy_hosts(self) -> list[dict[str, Any]]:
|
|
data = self.get_json("/nginx/proxy-hosts")
|
|
|
|
if not isinstance(data, list):
|
|
raise RuntimeError(
|
|
"NPM proxy-host response was not a list"
|
|
)
|
|
|
|
return data
|
|
|
|
def certificates(self) -> list[dict[str, Any]]:
|
|
data = self.get_json("/nginx/certificates")
|
|
|
|
if not isinstance(data, list):
|
|
raise RuntimeError(
|
|
"NPM certificate response was not a list"
|
|
)
|
|
|
|
return data
|
|
|
|
|
|
def proxy_host(
|
|
self,
|
|
local_id: str | int,
|
|
) -> dict[str, Any]:
|
|
host_id = self._local_id(local_id, "proxy host")
|
|
data = self.get_json(
|
|
f"/nginx/proxy-hosts/{host_id}"
|
|
)
|
|
if not isinstance(data, dict):
|
|
raise RuntimeError(
|
|
"NPM proxy-host response was not an object"
|
|
)
|
|
return data
|
|
|
|
def create_proxy_host(
|
|
self,
|
|
payload: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
response = self.request(
|
|
"POST",
|
|
"/nginx/proxy-hosts",
|
|
expected_status=201,
|
|
json=payload,
|
|
)
|
|
return self._json_object(
|
|
response,
|
|
"proxy-host create",
|
|
)
|
|
|
|
def update_proxy_host(
|
|
self,
|
|
local_id: str | int,
|
|
payload: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
host_id = self._local_id(local_id, "proxy host")
|
|
response = self.request(
|
|
"PUT",
|
|
f"/nginx/proxy-hosts/{host_id}",
|
|
expected_status=200,
|
|
json=payload,
|
|
)
|
|
return self._json_object(
|
|
response,
|
|
"proxy-host update",
|
|
)
|
|
|
|
def certificate(
|
|
self,
|
|
local_id: str | int,
|
|
) -> dict[str, Any]:
|
|
certificate_id = self._local_id(
|
|
local_id,
|
|
"certificate",
|
|
)
|
|
data = self.get_json(
|
|
f"/nginx/certificates/{certificate_id}"
|
|
)
|
|
if not isinstance(data, dict):
|
|
raise RuntimeError(
|
|
"NPM certificate response was not an object"
|
|
)
|
|
return data
|
|
|
|
@staticmethod
|
|
def _pem_bytes(value: Any, label: str) -> bytes:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise RuntimeError(
|
|
f"NPM certificate is missing {label}"
|
|
)
|
|
return value.encode("utf-8")
|
|
|
|
@staticmethod
|
|
def _zip_pem_member(
|
|
archive: ZipFile,
|
|
prefix: str,
|
|
) -> bytes:
|
|
matches = []
|
|
|
|
for info in archive.infolist():
|
|
if info.is_dir():
|
|
continue
|
|
|
|
basename = PurePosixPath(info.filename).name
|
|
|
|
if not basename.endswith(".pem"):
|
|
continue
|
|
|
|
stem = basename[:-4]
|
|
|
|
if not stem.startswith(prefix):
|
|
continue
|
|
|
|
suffix = stem[len(prefix):]
|
|
|
|
if not suffix.isdigit():
|
|
continue
|
|
|
|
matches.append(info)
|
|
|
|
if len(matches) != 1:
|
|
raise RuntimeError(
|
|
"NPM certificate download must contain exactly one "
|
|
f"{prefix}N.pem member"
|
|
)
|
|
|
|
data = archive.read(matches[0])
|
|
|
|
if not data:
|
|
raise RuntimeError(
|
|
f"NPM certificate download {prefix} member is empty"
|
|
)
|
|
|
|
return data
|
|
|
|
def certificate_material(
|
|
self,
|
|
local_id: str | int,
|
|
) -> CertificateMaterial:
|
|
certificate_id = self._local_id(
|
|
local_id,
|
|
"certificate",
|
|
)
|
|
|
|
detail = self.certificate(certificate_id)
|
|
provider = str(
|
|
detail.get("provider") or ""
|
|
).strip().lower()
|
|
|
|
if provider == "other":
|
|
meta = detail.get("meta")
|
|
if not isinstance(meta, dict):
|
|
raise RuntimeError(
|
|
"NPM custom certificate has invalid metadata"
|
|
)
|
|
|
|
intermediate = meta.get(
|
|
"intermediate_certificate"
|
|
)
|
|
|
|
return CertificateMaterial(
|
|
certificate=self._pem_bytes(
|
|
meta.get("certificate"),
|
|
"certificate",
|
|
),
|
|
certificate_key=self._pem_bytes(
|
|
meta.get("certificate_key"),
|
|
"certificate_key",
|
|
),
|
|
intermediate_certificate=(
|
|
None
|
|
if intermediate in (None, "")
|
|
else self._pem_bytes(
|
|
intermediate,
|
|
"intermediate_certificate",
|
|
)
|
|
),
|
|
)
|
|
|
|
if provider == "letsencrypt":
|
|
response = self.request(
|
|
"GET",
|
|
(
|
|
"/nginx/certificates/"
|
|
f"{certificate_id}/download"
|
|
),
|
|
expected_status=200,
|
|
)
|
|
|
|
try:
|
|
with ZipFile(BytesIO(response.content)) as archive:
|
|
return CertificateMaterial(
|
|
certificate=self._zip_pem_member(
|
|
archive,
|
|
"cert",
|
|
),
|
|
certificate_key=self._zip_pem_member(
|
|
archive,
|
|
"privkey",
|
|
),
|
|
intermediate_certificate=(
|
|
self._zip_pem_member(
|
|
archive,
|
|
"chain",
|
|
)
|
|
),
|
|
)
|
|
except BadZipFile as exc:
|
|
raise RuntimeError(
|
|
"NPM certificate download was not a valid ZIP archive"
|
|
) from exc
|
|
|
|
raise RuntimeError(
|
|
f"Unsupported NPM certificate provider: {provider!r}"
|
|
)
|
|
|
|
def delete_certificate(
|
|
self,
|
|
local_id: str | int,
|
|
) -> None:
|
|
certificate_id = self._local_id(
|
|
local_id,
|
|
"certificate",
|
|
)
|
|
|
|
self.request(
|
|
"DELETE",
|
|
f"/nginx/certificates/{certificate_id}",
|
|
expected_status=200,
|
|
)
|
|
|
|
def create_certificate(
|
|
self,
|
|
payload: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
response = self.request(
|
|
"POST",
|
|
"/nginx/certificates",
|
|
expected_status=201,
|
|
json=payload,
|
|
)
|
|
return self._json_object(
|
|
response,
|
|
"certificate create",
|
|
)
|
|
|
|
def upload_certificate(
|
|
self,
|
|
local_id: str | int,
|
|
*,
|
|
certificate: bytes,
|
|
certificate_key: bytes,
|
|
intermediate_certificate: bytes | None = None,
|
|
) -> None:
|
|
certificate_id = self._local_id(
|
|
local_id,
|
|
"certificate",
|
|
)
|
|
|
|
if not certificate:
|
|
raise RuntimeError(
|
|
"NPM certificate upload requires certificate data"
|
|
)
|
|
if not certificate_key:
|
|
raise RuntimeError(
|
|
"NPM certificate upload requires certificate key data"
|
|
)
|
|
|
|
files: dict[str, tuple[str, bytes]] = {
|
|
"certificate": (
|
|
"certificate.pem",
|
|
certificate,
|
|
),
|
|
"certificate_key": (
|
|
"certificate_key.pem",
|
|
certificate_key,
|
|
),
|
|
}
|
|
|
|
if intermediate_certificate:
|
|
files["intermediate_certificate"] = (
|
|
"intermediate_certificate.pem",
|
|
intermediate_certificate,
|
|
)
|
|
|
|
self.request(
|
|
"POST",
|
|
f"/nginx/certificates/{certificate_id}/upload",
|
|
expected_status=200,
|
|
files=files,
|
|
)
|