dns-sync/dns_sync/npm_client.py

import os
from typing import Any

import requests
import urllib3

from .adapters import Endpoint


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

    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,
    ) -> dict[str, Any]:
        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,
            )

        response = self.request(
            "POST",
            f"/nginx/certificates/{certificate_id}/upload",
            expected_status=200,
            files=files,
        )

        return self._json_object(
            response,
            "certificate upload",
        )