dns-sync/dns_sync/plugins/adapters/npm.py

from typing import Any

from ...adapters import (
    CERTIFICATES,
    PROXY_HOSTS,
    Capability,
    CapabilityOperation,
    DiscoveredObject,
    NormalizedObject,
    ObjectReference,
)
from ...npm_client import NpmClient
from ...object_normalizer import normalize_discovered_object


_PROXY_HOST_FIELDS = (
    "domain_names",
    "forward_scheme",
    "forward_host",
    "forward_port",
    "certificate_id",
    "ssl_forced",
    "caching_enabled",
    "block_exploits",
    "allow_websocket_upgrade",
    "http2_support",
    "hsts_enabled",
    "hsts_subdomains",
    "enabled",
)


class Plugin:
    name = "npm"
    capabilities = frozenset(
        {
            Capability(
                PROXY_HOSTS,
                CapabilityOperation.DISCOVER,
            ),
            Capability(
                PROXY_HOSTS,
                CapabilityOperation.VERIFY,
            ),
            Capability(
                CERTIFICATES,
                CapabilityOperation.DISCOVER,
            ),
            Capability(
                CERTIFICATES,
                CapabilityOperation.VERIFY,
            ),
        }
    )

    def __init__(self, config, endpoint):
        self.config = config
        self.endpoint = endpoint
        self.client = NpmClient(config, endpoint)

    @staticmethod
    def _local_key(row: dict[str, Any], label: str) -> str:
        local_id = row.get("id")
        if isinstance(local_id, bool) or not isinstance(local_id, int):
            raise RuntimeError(
                f"NPM {label} has invalid id: {local_id!r}"
            )
        return str(local_id)

    def connect(self) -> None:
        self.client.login()

    def discover(
        self,
        object_type: str,
    ) -> list[DiscoveredObject]:
        if object_type == PROXY_HOSTS:
            return self._discover_proxy_hosts()
        if object_type == CERTIFICATES:
            return self._discover_certificates()

        raise RuntimeError(
            f"{self.endpoint.identity} cannot discover "
            f"object type {object_type!r}"
        )

    def _proxy_host_from_row(
        self,
        row: dict[str, Any],
    ) -> DiscoveredObject:
        payload = {
            field: row.get(field)
            for field in _PROXY_HOST_FIELDS
        }

        certificate_id = row.get("certificate_id")

        if certificate_id in (None, "", 0, "0"):
            references = ()
        else:
            if isinstance(certificate_id, bool):
                raise RuntimeError(
                    "NPM proxy host has invalid certificate_id"
                )

            certificate_local_key = str(
                certificate_id
            ).strip()

            if not certificate_local_key:
                raise RuntimeError(
                    "NPM proxy host has invalid certificate_id"
                )

            references = (
                ObjectReference(
                    relation="certificate",
                    object_type=CERTIFICATES,
                    local_key=certificate_local_key,
                ),
            )

        return DiscoveredObject(
            object_type=PROXY_HOSTS,
            local_key=self._local_key(
                row,
                "proxy host",
            ),
            payload=payload,
            references=references,
        )

    def _certificate_from_row(
        self,
        row: dict[str, Any],
    ) -> DiscoveredObject:
        meta = row.get("meta")
        if not isinstance(meta, dict):
            meta = {}

        return DiscoveredObject(
            object_type=CERTIFICATES,
            local_key=self._local_key(
                row,
                "certificate",
            ),
            payload={
                "domain_names": list(
                    row.get("domain_names") or []
                ),
                "provider": str(
                    row.get("provider") or ""
                ),
                "nice_name": str(
                    row.get("nice_name") or ""
                ),
                "valid_from": row.get("valid_from"),
                "expires_on": row.get("expires_on"),
                "cluster_managed": (
                    meta.get("cluster_managed") is True
                ),
                "acme_renew": (
                    meta.get("acme_renew") is True
                ),
                "dns_challenge": (
                    meta.get("dns_challenge") is True
                ),
            },
        )

    def _discover_proxy_hosts(self) -> list[DiscoveredObject]:
        return [
            self._proxy_host_from_row(row)
            for row in self.client.proxy_hosts()
        ]

    def _discover_certificates(self) -> list[DiscoveredObject]:
        return [
            self._certificate_from_row(row)
            for row in self.client.certificates()
        ]

    def apply(self, action):
        raise RuntimeError(
            f"{self.endpoint.identity} does not support apply yet"
        )

    def verify(self, object: NormalizedObject) -> bool:
        if object.local_key is None:
            return False

        if object.object_type == PROXY_HOSTS:
            row = self.client.proxy_host(
                object.local_key
            )
            actual = normalize_discovered_object(
                self._proxy_host_from_row(row)
            )

            return actual == object

        if object.object_type == CERTIFICATES:
            row = self.client.certificate(
                object.local_key
            )
            actual = normalize_discovered_object(
                self._certificate_from_row(row)
            )

            if actual.local_key != object.local_key:
                return False

            if actual.logical_key != object.logical_key:
                return False

            # Provider, display name, and renewal flags describe how
            # this NPM endpoint manages the certificate. They are not
            # portable certificate identity. A copied Let's Encrypt
            # certificate becomes provider="other" on the recipient.
            #
            # The portable verification state is the domain identity
            # plus the certificate validity period exposed by NPM.
            for field in (
                "domain_names",
                "valid_from",
                "expires_on",
            ):
                if (
                    actual.payload.get(field)
                    != object.payload.get(field)
                ):
                    return False

            return True

        raise RuntimeError(
            f"{self.endpoint.identity} cannot verify "
            f"object type {object.object_type!r}"
        )