david/dns-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add per-endpoint NPM discovery adapters

Browse Source
This commit is contained in:
Lyncoln
2026-09-19 17:02:51 -04:00
parent 79342590ba
commit da53a9f1e1
4 changed files with 306 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dns_sync/plugins/adapters/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Generic endpoint adapter plugins."""
dns_sync/plugins/adapters/npm.py
+136
View File
@@ -0,0 +1,136 @@
from typing import Any
from ...adapters import (
CERTIFICATES,
PROXY_HOSTS,
Capability,
CapabilityOperation,
DiscoveredObject,
)
from ...npm_client import NpmClient
_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(
CERTIFICATES,
CapabilityOperation.DISCOVER,
),
}
)
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 _discover_proxy_hosts(self) -> list[DiscoveredObject]:
discovered: list[DiscoveredObject] = []
for row in self.client.proxy_hosts():
payload = {
field: row.get(field)
for field in _PROXY_HOST_FIELDS
}
discovered.append(
DiscoveredObject(
object_type=PROXY_HOSTS,
local_key=self._local_key(row, "proxy host"),
payload=payload,
)
)
return discovered
def _discover_certificates(self) -> list[DiscoveredObject]:
discovered: list[DiscoveredObject] = []
for row in self.client.certificates():
meta = row.get("meta")
if not isinstance(meta, dict):
meta = {}
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
),
}
discovered.append(
DiscoveredObject(
object_type=CERTIFICATES,
local_key=self._local_key(row, "certificate"),
payload=payload,
)
)
return discovered
def apply(self, action):
raise RuntimeError(
f"{self.endpoint.identity} does not support apply yet"
)
def verify(self, object):
raise RuntimeError(
f"{self.endpoint.identity} does not support verify yet"
)
dns_sync/registry.py
+6 -2
View File
@@ -1,7 +1,7 @@
import importlib
def _load(kind: str, name: str, config):
def _load(kind: str, name: str, config, *args):
if not name or not name.replace("_", "").replace("-", "").isalnum():
raise RuntimeError(f"Invalid {kind} plugin name: {name!r}")
@@ -19,7 +19,7 @@ def _load(kind: str, name: str, config):
if plugin_class is None:
raise RuntimeError(f"Plugin module {path} does not export Plugin")
return plugin_class(config)
return plugin_class(config, *args)
def load_source(name: str, config):
@@ -28,3 +28,7 @@ def load_source(name: str, config):
def load_provider(name: str, config):
return _load("providers", name, config)
def load_adapter(name: str, config, endpoint):
return _load("adapters", name, config, endpoint)
tests/test_npm_adapter.py
+163
View File
@@ -0,0 +1,163 @@
import os
import unittest
from unittest.mock import Mock, patch
from dns_sync.adapters import (
CERTIFICATES,
PROXY_HOSTS,
Capability,
CapabilityOperation,
)
from dns_sync.config import AppConfig
from dns_sync.registry import load_adapter
class NpmAdapterTests(unittest.TestCase):
def config(self):
with patch.dict(
os.environ,
{
"NPM_ENDPOINTS": "192.168.1.3,192.168.1.4",
"DNS_TARGET_IP": "192.168.1.101",
},
clear=True,
):
return AppConfig.from_env()
def adapter(self, client):
config = self.config()
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=client,
):
return load_adapter(
"npm",
config,
config.npm_endpoints[0],
)
def test_registry_creates_one_adapter_per_endpoint(self):
config = self.config()
first_client = Mock()
second_client = Mock()
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
side_effect=[first_client, second_client],
) as client_class:
adapters = [
load_adapter("npm", config, endpoint)
for endpoint in config.npm_endpoints
]
self.assertEqual(
tuple(adapter.endpoint.identity for adapter in adapters),
(
"npm://192.168.1.3:81",
"npm://192.168.1.4:81",
),
)
self.assertEqual(client_class.call_count, 2)
self.assertIs(adapters[0].client, first_client)
self.assertIs(adapters[1].client, second_client)
def test_capabilities_are_discovery_only(self):
adapter = self.adapter(Mock())
self.assertEqual(
adapter.capabilities,
frozenset(
{
Capability(
PROXY_HOSTS,
CapabilityOperation.DISCOVER,
),
Capability(
CERTIFICATES,
CapabilityOperation.DISCOVER,
),
}
),
)
def test_discovers_proxy_hosts_with_endpoint_local_key(self):
client = Mock()
client.proxy_hosts.return_value = [
{
"id": 12,
"domain_names": ["app.example.com"],
"forward_scheme": "http",
"forward_host": "192.168.1.20",
"forward_port": 8080,
"certificate_id": 67,
"enabled": 1,
"unexpected_secret": "must-not-be-copied",
}
]
adapter = self.adapter(client)
discovered = adapter.discover(PROXY_HOSTS)
self.assertEqual(len(discovered), 1)
self.assertEqual(discovered[0].object_type, PROXY_HOSTS)
self.assertEqual(discovered[0].local_key, "12")
self.assertEqual(
discovered[0].payload["domain_names"],
["app.example.com"],
)
self.assertNotIn(
"unexpected_secret",
discovered[0].payload,
)
def test_certificate_discovery_excludes_secret_material(self):
client = Mock()
client.certificates.return_value = [
{
"id": 67,
"domain_names": ["app.example.com"],
"provider": "other",
"nice_name": "cluster copy",
"valid_from": "2026-09-01T00:00:00Z",
"expires_on": "2026-12-01T00:00:00Z",
"private_key": "must-not-be-copied",
"certificate": "must-not-be-copied",
"meta": {
"cluster_managed": True,
"acme_renew": True,
"dns_challenge": True,
"dns_provider_credentials": "must-not-be-copied",
},
}
]
adapter = self.adapter(client)
discovered = adapter.discover(CERTIFICATES)
self.assertEqual(len(discovered), 1)
certificate = discovered[0]
self.assertEqual(certificate.local_key, "67")
self.assertTrue(certificate.payload["cluster_managed"])
self.assertTrue(certificate.payload["acme_renew"])
self.assertNotIn("private_key", certificate.payload)
self.assertNotIn("certificate", certificate.payload)
self.assertNotIn(
"dns_provider_credentials",
certificate.payload,
)
def test_unsupported_operations_fail_explicitly(self):
adapter = self.adapter(Mock())
with self.assertRaisesRegex(RuntimeError, "cannot discover"):
adapter.discover("unsupported_objects")
with self.assertRaisesRegex(RuntimeError, "does not support apply"):
adapter.apply(Mock())
with self.assertRaisesRegex(RuntimeError, "does not support verify"):
adapter.verify(Mock())
if __name__ == "__main__":
unittest.main()