Add read-only NPM certificate discovery

dns_sync/certificates/__init__.py

@@ -0,0 +1 @@
+"""Certificate discovery and reconciliation support."""

dns_sync/certificates/models.py

@@ -0,0 +1,35 @@
+from dataclasses import dataclass
+from datetime import datetime
+
+
+@dataclass(frozen=True)
+class CertificateIdentity:
+    """
+    Cluster identity for a certificate.
+
+    NPM numeric IDs are intentionally excluded because they are node-local.
+    """
+
+    domains: tuple[str, ...]
+
+    @property
+    def key(self) -> str:
+        return ",".join(self.domains)
+
+
+@dataclass(frozen=True)
+class CertificateState:
+    """
+    Non-secret certificate state suitable for discovery, comparison, and logs.
+    """
+
+    identity: CertificateIdentity
+    local_id: int
+    provider: str
+    nice_name: str
+    valid_from: datetime | None
+    expires_on: datetime | None
+    cluster_managed: bool
+    acme_renew: bool
+    dns_challenge: bool
+    source: str

dns_sync/certificates/normalizer.py

@@ -0,0 +1,47 @@
+from datetime import datetime, timezone
+
+from ..normalizer import normalize_hostname
+from .models import CertificateIdentity
+
+
+def normalize_certificate_domains(values) -> CertificateIdentity:
+    domains: set[str] = set()
+
+    for raw in values or []:
+        hostname, _ = normalize_hostname(
+            str(raw),
+            allow_wildcard=True,
+        )
+        domains.add(hostname)
+
+    if not domains:
+        raise RuntimeError(
+            "Certificate has no usable domain names"
+        )
+
+    return CertificateIdentity(
+        domains=tuple(sorted(domains))
+    )
+
+
+def parse_npm_datetime(value) -> datetime | None:
+    if not value:
+        return None
+
+    raw = str(value).strip()
+
+    # NPM normally emits ISO timestamps ending in Z.
+    if raw.endswith("Z"):
+        raw = raw[:-1] + "+00:00"
+
+    try:
+        result = datetime.fromisoformat(raw)
+    except ValueError as exc:
+        raise RuntimeError(
+            f"Invalid NPM certificate timestamp: {value!r}"
+        ) from exc
+
+    if result.tzinfo is None:
+        result = result.replace(tzinfo=timezone.utc)
+
+    return result.astimezone(timezone.utc)

dns_sync/certificates/npm.py

@@ -0,0 +1,79 @@
+from typing import Any
+
+from ..npm_client import NpmClient
+from .models import CertificateState
+from .normalizer import (
+    normalize_certificate_domains,
+    parse_npm_datetime,
+)
+
+
+class NpmCertificateRepository:
+    """
+    Read-only NPM certificate repository.
+
+    This milestone intentionally discovers metadata only.
+    Certificate/private-key material is not requested or logged.
+    """
+
+    def __init__(self, config):
+        self.config = config
+        self.client = NpmClient(config)
+
+    def _rows(self) -> list[dict[str, Any]]:
+        data = self.client.certificates()
+
+        if not isinstance(data, list):
+            raise RuntimeError(
+                "NPM certificate response was not a list"
+            )
+
+        return data
+
+    def discover(self) -> list[CertificateState]:
+        certificates: list[CertificateState] = []
+
+        for row in self._rows():
+            local_id = row.get("id")
+
+            if not isinstance(local_id, int):
+                raise RuntimeError(
+                    f"NPM certificate has invalid id: {local_id!r}"
+                )
+
+            meta = row.get("meta")
+            if not isinstance(meta, dict):
+                meta = {}
+
+            certificates.append(
+                CertificateState(
+                    identity=normalize_certificate_domains(
+                        row.get("domain_names")
+                    ),
+                    local_id=local_id,
+                    provider=str(
+                        row.get("provider") or ""
+                    ),
+                    nice_name=str(
+                        row.get("nice_name") or ""
+                    ),
+                    valid_from=parse_npm_datetime(
+                        row.get("valid_from")
+                    ),
+                    expires_on=parse_npm_datetime(
+                        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
+                    ),
+                    source=self.config.proxy_url,
+                )
+            )
+
+        return certificates

dns_sync/npm_client.py

@@ -126,6 +126,15 @@ class NpmClient:
 
         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 certificates(self) -> list[dict[str, Any]]:
         data = self.get_json("/nginx/certificates")