139 lines
3.4 KiB
Python
139 lines
3.4 KiB
Python
import os
|
|
from typing import Any
|
|
|
|
import requests
|
|
import urllib3
|
|
|
|
|
|
class NpmClient:
|
|
"""Authenticated client for the Nginx Proxy Manager API."""
|
|
|
|
def __init__(self, config):
|
|
self.config = config
|
|
|
|
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_url}/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()
|
|
|
|
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
|