From 33df9f59ec94a2673ee29858c3581be400cb17a9 Mon Sep 17 00:00:00 2001 From: Lyncoln Date: Sat, 19 Sep 2026 15:04:38 -0400 Subject: [PATCH] Extract shared NPM API client --- dns_sync/npm_client.py | 127 ++++++++++++++++++++++++++++++++ dns_sync/plugins/sources/npm.py | 65 +++------------- 2 files changed, 139 insertions(+), 53 deletions(-) create mode 100644 dns_sync/npm_client.py diff --git a/dns_sync/npm_client.py b/dns_sync/npm_client.py new file mode 100644 index 0000000..e01701b --- /dev/null +++ b/dns_sync/npm_client.py @@ -0,0 +1,127 @@ +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 diff --git a/dns_sync/plugins/sources/npm.py b/dns_sync/plugins/sources/npm.py index 78803bb..1df1b53 100644 --- a/dns_sync/plugins/sources/npm.py +++ b/dns_sync/plugins/sources/npm.py @@ -1,10 +1,6 @@ -import os - -import requests -import urllib3 - from ...models import DesiredRecord from ...normalizer import normalize_hostname +from ...npm_client import NpmClient class Plugin: @@ -12,63 +8,25 @@ class Plugin: 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 when PROXY=npm") - if not self.secret: - raise RuntimeError("NPM_SECRET is required when PROXY=npm") - - self.session = requests.Session() - self.token = None - if not config.proxy_verify_tls: - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - @property - def api_url(self): - return f"{self.config.proxy_url}/api" - - def _login(self): - 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(f"NPM login failed: HTTP {response.status_code}: {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 = token - self.session.headers.update({"Authorization": f"Bearer {token}", "Accept": "application/json"}) - - def _proxy_hosts(self): - if not self.token: - self._login() - response = self.session.get( - f"{self.api_url}/nginx/proxy-hosts", - timeout=15, - verify=self.config.proxy_verify_tls, - ) - if response.status_code != 200: - raise RuntimeError(f"Unable to retrieve NPM proxy hosts: HTTP {response.status_code}: {response.text[:300]}") - data = response.json() - if not isinstance(data, list): - raise RuntimeError("NPM proxy-host response was not a list") - return data + self.client = NpmClient(config) def discover(self): records: list[DesiredRecord] = [] - for host in self._proxy_hosts(): + + for host in self.client.proxy_hosts(): enabled = bool(host.get("enabled", False)) + if not enabled and not self.config.include_disabled: continue source_id = str(host.get("id", "")) or None + for raw_domain in host.get("domain_names", []): - hostname, wildcard = normalize_hostname(str(raw_domain), allow_wildcard=True) + hostname, wildcard = normalize_hostname( + str(raw_domain), + allow_wildcard=True, + ) + records.append( DesiredRecord( hostname=hostname, @@ -79,4 +37,5 @@ class Plugin: source_id=source_id, ) ) + return records