Load built image before workflow verification

.gitea/workflows/build.yml

@@ -0,0 +1,139 @@
+name: Build and Publish dns-sync Image
+
+on:
+  push:
+    branches: [ main ]
+  workflow_dispatch:
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      # -------------------------------------------------
+      # Checkout (manual, no Node, no external actions)
+      # -------------------------------------------------
+      - name: Checkout repository
+        env:
+          GITEA_TOKEN: ${{ secrets.TEA_TOKEN }}
+        run: |
+          set -euo pipefail
+
+          rm -rf ./* ./.git || true
+
+          BASE="${GITHUB_SERVER_URL%/}"
+          REPO="${GITHUB_REPOSITORY}"
+
+          if [ -n "${GITEA_TOKEN:-}" ]; then
+            CLONE_URL="${BASE}/${REPO}.git"
+            CLONE_URL="$(echo "$CLONE_URL" | sed -E 's#^https://#https://oauth2:'"${GITEA_TOKEN}"'@#')"
+            git clone "$CLONE_URL" .
+          else
+            git clone "${BASE}/${REPO}.git" .
+          fi
+
+          git config --global --add safe.directory "$PWD"
+
+      # -------------------------------------------------
+      # Test source before building
+      # -------------------------------------------------
+      - name: Run test suite
+        run: |
+          set -euo pipefail
+
+          python3 -m venv .venv
+          .venv/bin/python -m pip install --upgrade pip
+          .venv/bin/python -m pip install requests PyYAML
+
+          .venv/bin/python -m unittest discover \
+            -s tests \
+            -v
+
+      # -------------------------------------------------
+      # Build image
+      # -------------------------------------------------
+      - name: Build image
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+        run: |
+          set -euo pipefail
+
+          SHORT_SHA="${GITHUB_SHA::7}"
+          FULL_IMAGE="$REGISTRY/david/dns-sync"
+
+          docker build \
+            -f Containerfile \
+            -t "$FULL_IMAGE:latest" \
+            -t "$FULL_IMAGE:sha-$SHORT_SHA" \
+            --output "type=docker,dest=$RUNNER_TEMP/dns-sync-built.tar" \
+            .
+
+      # -------------------------------------------------
+      # Verify locally before publishing
+      # -------------------------------------------------
+      - name: Load and verify image
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+        run: |
+          set -euo pipefail
+
+          FULL_IMAGE="$REGISTRY/david/dns-sync"
+
+          test -s "$RUNNER_TEMP/dns-sync-built.tar"
+
+          docker load \
+            --input "$RUNNER_TEMP/dns-sync-built.tar"
+
+          docker image inspect "$FULL_IMAGE:latest" >/dev/null
+
+          docker run --rm \
+            --entrypoint python \
+            "$FULL_IMAGE:latest" \
+            -m dns_sync reconcile --help
+
+          echo
+          echo "Loaded image:"
+          docker image ls "$FULL_IMAGE"
+
+      # -------------------------------------------------
+      # Login to container registry
+      # -------------------------------------------------
+      - name: Docker login
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+          REGISTRY_USER: ${{ secrets.REGISTRY_USER }}
+          REGISTRY_TOKEN: ${{ secrets.TEA_TOKEN }}
+        run: |
+          set -euo pipefail
+
+          echo "$REGISTRY_TOKEN" | docker login "$REGISTRY" \
+            -u "$REGISTRY_USER" \
+            --password-stdin
+
+      # -------------------------------------------------
+      # Publish
+      # -------------------------------------------------
+      - name: Publish image
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+        run: |
+          set -euo pipefail
+
+          SHORT_SHA="${GITHUB_SHA::7}"
+          FULL_IMAGE="$REGISTRY/david/dns-sync"
+
+          docker push "$FULL_IMAGE:latest"
+          docker push "$FULL_IMAGE:sha-$SHORT_SHA"
+
+      - name: Report published image
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+        run: |
+          set -euo pipefail
+
+          SHORT_SHA="${GITHUB_SHA::7}"
+          FULL_IMAGE="$REGISTRY/david/dns-sync"
+
+          echo "Published:"
+          echo "  $FULL_IMAGE:latest"
+          echo "  $FULL_IMAGE:sha-$SHORT_SHA"

Containerfile

@@ -2,13 +2,18 @@ FROM docker.io/library/python:3.13-slim
 
 ENV PYTHONDONTWRITEBYTECODE=1 \
     PYTHONUNBUFFERED=1 \
-    PYTHONPATH=/app
+    PYTHONPATH=/app \
+    DNS_SYNC_STATE_DB=/var/lib/dns-sync/state.sqlite3 \
+    DNS_SYNC_PLUGIN_PATH=/plugins
 
 RUN pip install --no-cache-dir requests PyYAML
 
 WORKDIR /app
+
 COPY dns_sync /app/dns_sync
 COPY entrypoint.sh /usr/local/bin/entrypoint.sh
-RUN chmod 0755 /usr/local/bin/entrypoint.sh
+
+RUN mkdir -p /var/lib/dns-sync /plugins \
+    && chmod 0755 /usr/local/bin/entrypoint.sh
 
 ENTRYPOINT ["/usr/local/bin/entrypoint.sh"]

README.md

@@ -2,59 +2,579 @@
 
 A plugin-based DNS reconciliation service.
 
+## Architecture
+
+dns-sync separates generic reconciliation from application-specific endpoint
+behavior.
+
+    native endpoint
+          |
+          v
+    <App>Client
+          |
+          | native objects
+          v
+    Plugin
+          |
+          | DiscoveredObject
+          v
+    dns-sync core
+          |
+          | normalization
+          | observations
+          | desired state
+          | persistence
+          | reconciliation
+          | planning
+          v
+    PlanAction
+          |
+          v
+    Plugin
+          |
+          | native translation
+          v
+    <App>Client
+          |
+          v
+    native endpoint
+
+The core owns orchestration truth.
+
+Plugins own application semantics.
+
+Clients own endpoint-native communication.
+
 ## Plugin contract
 
-Source modules live under `dns_sync/plugins/sources/` and export a class named `Plugin` with:
+The reconciliation runtime is adapter-based.
 
-- `name`
-- `discover() -> list[DesiredRecord]`
+Every adapter package must export:
 
-Provider modules live under `dns_sync/plugins/providers/` and export `Plugin` with:
+- `configured_endpoints(config)`
+- `Plugin`
 
-- `name`
-- `capabilities`
-- `connect()`
-- `list_records()`
-- `add_record()`
-- `update_record()`
-- `delete_record()`
-- `verify_record()`
-- `capacity()`
+The registry calls:
 
-Plugins are convention-loaded from environment variables. No registry edit is needed:
+    configured_endpoints(config)
 
-- `PROXY=npm` loads `dns_sync.plugins.sources.npm.Plugin`
-- `DNS_PROVIDER=verizon` loads `dns_sync.plugins.providers.verizon.Plugin`
+to obtain the endpoints owned by the plugin.
 
-## Current plugins
+For each configured endpoint, the registry constructs:
 
-Sources: `npm`, `file`
+    Plugin(config, endpoint)
 
-Providers: `verizon`
+Every endpoint returned by `configured_endpoints(config)` must identify the
+plugin that returned it. The registry rejects cross-plugin endpoint ownership.
+
+Plugins declare object-specific capabilities using the generic
+`Capability` model from `dns_sync.adapters`.
+
+Capabilities currently describe:
+
+- discovery
+- apply
+- verification
+
+for a particular generic object type.
+
+Built-in adapters live under:
+
+    dns_sync/plugins/adapters/
+
+External adapters are loaded from:
+
+    DNS_SYNC_PLUGIN_PATH
+
+and must be explicitly enabled through:
+
+    ENABLED_PLUGINS
+
+## Canonical plugin structure
+
+Application integrations should use this package structure:
+
+    <plugin_name>/
+    ├── __init__.py
+    └── client.py
+
+The repository contains a copyable example under:
+
+    examples/plugin_template/
+
+The example is documentation only and is not loaded from that location.
+
+### Adapter boundary: __init__.py
+
+`__init__.py` is the dns-sync-facing plugin boundary.
+
+It owns:
+
+- endpoint configuration through `configured_endpoints(config)`
+- generic capability declarations
+- validation of the supplied endpoint
+- native-to-generic discovery translation
+- generic-to-native apply translation
+- verification
+- optional retirement behavior
+- application-specific semantic validation
+
+The adapter communicates with the core using generic types such as:
+
+- `Endpoint`
+- `DiscoveredObject`
+- `NormalizedObject`
+- `ObjectReference`
+- `PlanAction`
+- `ApplyResult`
+- `RetireRequest`
+
+### Native boundary: client.py
+
+`client.py` is private implementation machinery for that application.
+
+It owns:
+
+- authentication
+- HTTP sessions, sockets, commands, or other transport
+- application-native request and response handling
+- application-native models
+- endpoint-local mutation primitives
+- endpoint-specific verification mechanics
+
+A client represents one explicit `Endpoint`.
+
+The client should not independently rediscover which endpoint it is supposed
+to contact from another global endpoint setting.
+
+There is deliberately no generic dns-sync `Client` base class. Different
+applications may communicate through completely different mechanisms. The
+shared contract is the adapter plugin boundary, not the transport layer.
+
+Application-specific models belong inside the plugin package rather than the
+dns-sync core unless they represent a genuinely application-independent core
+concept.
+
+## Discovery
+
+`Plugin.discover(object_type)` reads native endpoint state through the private
+client and returns `DiscoveredObject` instances.
+
+A discovered object carries:
+
+    object_type
+    local_key
+    payload
+    references
+
+`local_key` is endpoint-local identity. It is not assumed to be portable to
+another endpoint.
+
+The plugin owns translation from the native application's representation into
+this generic discovery representation.
+
+Unsupported object types should fail explicitly rather than silently returning
+an empty or misleading result.
+
+## Normalized state
+
+The core converts discoveries into `NormalizedObject` instances and owns the
+reconciliation state built from them.
+
+A normalized object carries:
+
+    object_type
+    logical_key
+    payload
+    local_key
+    references
+
+`logical_key` represents portable identity.
+
+`local_key` represents endpoint-local identity.
+
+Plugins do not persist their own reconciliation truth. The core owns:
+
+- observations
+- observation status
+- desired state
+- applied and verified state
+- reconciliation
+- planning
+- dependency resolution
+
+## Object references
+
+Dependencies between objects use `ObjectReference`.
+
+A reference identifies another object by exactly one of:
+
+    local_key
+
+or:
+
+    logical_key
+
+but never both.
+
+Endpoint discovery may expose local references because those are the identities
+native applications actually use.
+
+Central reconciliation can convert those into portable logical references.
+
+During execution, dns-sync resolves portable references back to the correct
+endpoint-local identity before invoking the target adapter.
+
+Plugins should never treat an endpoint-local ID from one endpoint as portable
+desired state for another endpoint.
+
+## Apply
+
+`Plugin.apply(action)` receives a generic `PlanAction`.
+
+Before mutation, the adapter should validate that the action belongs to its
+endpoint.
+
+Shared validation helpers are available in:
+
+    dns_sync.adapter_support
+
+including:
+
+    require_endpoint_plugin(...)
+    require_action_endpoint(...)
+    require_desired(...)
+
+`NOOP` and `SKIP` actions must not mutate the endpoint.
+
+For a mutating operation, the adapter should:
+
+1. validate the requested operation;
+2. validate the desired normalized object;
+3. translate desired state into an application-native object;
+4. invoke the private client;
+5. re-fetch or rediscover the actual resulting native object;
+6. obtain its real endpoint-local identity;
+7. return an `ApplyResult`.
+
+Do not guess or predict endpoint-local IDs when the endpoint assigns them.
+
+Unsupported operations should fail explicitly.
+
+## Apply results
+
+A successful mutation returns an `ApplyResult` whose normalized object contains
+the actual endpoint-local key.
+
+Conceptually:
+
+    ApplyResult(
+        object=normalized_applied_object,
+        detail=...,
+    )
+
+The returned object's `local_key` is required.
+
+An apply result may also include deferred `RetireRequest` objects when a
+replacement workflow requires the old object to remain temporarily while
+dependencies are rebound.
+
+## Verification
+
+`Plugin.verify(object)` checks that the actual endpoint state matches the
+normalized object dns-sync expects to exist.
+
+Verification is intentionally separate from mutation.
+
+A successful API response is not sufficient evidence that endpoint state
+converged.
+
+The executor requires supported mutations to pass verification before the
+result is treated as successfully applied.
+
+## Retirement
+
+`retire()` is optional.
+
+A plugin should implement retirement only when it can safely remove or retire
+endpoint-local objects.
+
+Retirement is used by deletion and replacement workflows.
+
+The executor performs preflight checks before mutation. If execution requires
+retirement and the adapter does not provide it, the operation fails before the
+unsafe mutation begins.
+
+Retirement should use the endpoint-local identity supplied by dns-sync and fail
+closed when the object cannot be identified safely.
+
+## Fail-closed behavior
+
+Plugins should reject ambiguous or unsupported behavior.
+
+In particular:
+
+- reject endpoints owned by another plugin;
+- reject actions intended for another endpoint;
+- reject unsupported object types;
+- reject unsupported plan operations;
+- reject malformed native state;
+- reject ambiguous endpoint-local identities;
+- never guess missing IDs;
+- never report successful application until the resulting object can be
+  identified;
+- never treat mutation success alone as verification.
+
+The generic executor adds additional preflight and dependency safety checks.
+
+## Configuration ownership
+
+Configuration should live at the lowest layer that actually understands it.
+
+Use this rule:
+
+    Does generic reconciliation need to understand this value?
+
+    yes -> core configuration may be appropriate
+    no  -> plugin/client configuration
+
+Examples of generic reconciliation policy include:
+
+    DNS_TARGET_IP
+    DOMAIN_MATCHING
+    DOMAIN
+
+Examples of plugin/client configuration include:
+
+    NPM_ENDPOINTS
+    PROXY_API_PORT
+    PROXY_API_SCHEME
+    NPM_IDENTITY
+    NPM_SECRET
+    VERIZON_ENDPOINTS
+    ROUTER_PASSWORD
+    ROUTER_TIMEOUT
+
+`configured_endpoints(config)` owns endpoint discovery for the plugin.
+
+The client receives the resulting explicit `Endpoint`; it should not implement
+a second independent endpoint-selection mechanism.
+
+## Secret handling
+
+Secret material should remain below the normalized and persistence boundary
+unless the core has an explicit safe representation for it.
+
+Plugins must not place credentials, authentication tokens, private keys, or
+other secret material into generic persisted payloads.
+
+Runtime credentials should normally be provided through environment variables,
+container secrets, or another external secret-management mechanism.
+
+## Plugins
+
+The generic reconciliation engine uses adapter plugins.
+
+Built-in adapters:
+
+- `npm` - Nginx Proxy Manager proxy hosts and certificates
+
+External adapters are loaded from `DNS_SYNC_PLUGIN_PATH` and must also be
+explicitly enabled through `ENABLED_PLUGINS`.
+
+Example:
+
+```text
+ENABLED_PLUGINS=npm,verizon
+DNS_SYNC_PLUGIN_PATH=/plugins
+```
+
+The Verizon adapter lives under `external_plugins/verizon` and is deliberately
+not copied into the dns-sync container image. The external plugin directory is
+mounted read-only at runtime.
+
+## Plugin loading
+
+Only explicitly enabled plugins are loadable.
+
+The registry first checks for a built-in adapter under:
+
+    dns_sync/plugins/adapters/<plugin_name>/
+
+If no built-in adapter exists, it looks for an external package at:
+
+    $DNS_SYNC_PLUGIN_PATH/<plugin_name>/__init__.py
+
+External plugins are loaded as packages, so they may use relative imports for
+private modules such as:
+
+    from .client import ExampleClient
+
+Imports from dns-sync itself should remain absolute, for example:
+
+    from dns_sync.adapters import Endpoint
+    from dns_sync.adapter_support import require_endpoint_plugin
+
+The external plugin directory does not need to be added globally to
+`PYTHONPATH`.
+
+## Adding a new plugin
+
+Start from:
+
+    examples/plugin_template/
+
+For an external plugin:
+
+1. copy the template into the configured external plugin root;
+2. rename the directory to the plugin name;
+3. rename `ExampleClient` and `ExampleRecord`;
+4. implement `configured_endpoints(config)`;
+5. implement native authentication and transport in `client.py`;
+6. declare only the generic capabilities actually supported;
+7. implement native-to-`DiscoveredObject` translation;
+8. implement `NormalizedObject`-to-native translation;
+9. implement `apply()` and `verify()`;
+10. implement `retire()` only when safe retirement is supported;
+11. add focused client and adapter tests;
+12. enable the plugin through `ENABLED_PLUGINS`;
+13. run read-only reconciliation against the real endpoint before permitting
+    mutations.
+
+Do not begin a new integration by adding application-specific models or
+transport behavior to the core.
+
+## Reference implementations
+
+Use both existing plugins when designing a new integration.
+
+### NPM
+
+The NPM plugin demonstrates:
+
+- a built-in adapter;
+- multiple configured endpoints;
+- proxy-host and certificate object types;
+- endpoint-local object IDs;
+- cross-object certificate references;
+- portable logical references;
+- replacement workflows;
+- deferred retirement;
+- certificate material handling.
+
+Its package lives at:
+
+    dns_sync/plugins/adapters/npm/
+
+### Verizon
+
+The Verizon plugin demonstrates:
+
+- an external adapter;
+- plugin-owned endpoint configuration;
+- a private native record model;
+- a private endpoint client;
+- DNS record discovery;
+- create/update operations;
+- verification;
+- a read-only mounted external plugin package.
+
+Its package lives at:
+
+    external_plugins/verizon/
+
+Together they demonstrate the preferred boundary:
+
+    dns-sync core
+        |
+        +-- generic state and orchestration
+        |
+        +-- Plugin
+              |
+              +-- native / normalized translation
+              |
+              +-- <App>Client
+                    |
+                    +-- application-native implementation
+
+## Plugin development validation
+
+Before committing a plugin change, run at least:
+
+    git diff --check
+    python3 -m compileall -q dns_sync external_plugins tests examples
+    python3 -m unittest discover -s tests -v
+
+Changes involving plugin loading, native clients, endpoint configuration, or
+mutation behavior should also be validated with a real read-only container
+reconciliation against the intended endpoints.
+
+A clean read-only plan does not by itself authorize mutation. Review unexpected
+planned changes before enabling apply mode.
+
+## Container runtime
+
+The container periodically executes the generic reconciliation engine:
+
+```text
+python -m dns_sync reconcile
+```
+
+The default container mode is read-only.
+
+```text
+SYNC_INTERVAL=1
+DNS_SYNC_APPLY=false
+DNS_SYNC_STATE_DB=/var/lib/dns-sync/state.sqlite3
+DNS_SYNC_PLUGIN_PATH=/plugins
+ENABLED_PLUGINS=npm,verizon
+```
+
+`SYNC_INTERVAL` is expressed in minutes. The interval begins after each
+reconciliation completes, so runs do not overlap.
+
+To permit planned changes to be applied and verified:
+
+```text
+DNS_SYNC_APPLY=true
+```
+
+An optional mutation budget can be specified:
+
+```text
+DNS_SYNC_MAX_CHANGES=10
+```
+
+When `DNS_SYNC_APPLY=false`, reconciliation still performs discovery,
+normalization, desired-state reconciliation, planning, and state persistence,
+but no endpoint mutations are executed.
+
+The SQLite database should be stored on persistent storage.
+
+External plugins should be mounted read-only at `/plugins`.
 
 ## Main environment
 
 ```text
-PROXY=npm
-PROXY_IP=192.168.1.101
+NPM_ENDPOINTS=192.168.1.3,192.168.1.4,192.168.1.5
 PROXY_API_PORT=81
 PROXY_API_SCHEME=http
-DNS_PROVIDER=verizon
+DNS_TARGET_IP=192.168.1.101
+
+VERIZON_ENDPOINTS=192.168.1.1
+
 DOMAIN_MATCHING=false
 DOMAIN=example.com
-SYNC_INTERVAL=1
+
 NPM_IDENTITY=...
 NPM_SECRET=...
 ROUTER_PASSWORD=...
 ```
 
-`DNS_TARGET_IP` optionally overrides the default DNS destination, which is `PROXY_IP`.
-
-
 ## Domain filtering
 
-Filtering is performed by the central engine after source discovery and before
-reconciliation, so all source plugins get identical behavior.
+Filtering is performed centrally after adapter discovery and before
+reconciliation, so endpoint-specific adapters get identical behavior.
 
 ```text
 DOMAIN_MATCHING=true

dns-sync.container

@@ -1,5 +1,5 @@
 [Unit]
-Description=Reverse Proxy DNS Sync
+Description=DNS Sync Reconciliation Service
 Wants=network-online.target
 After=network-online.target
 
@@ -7,20 +7,30 @@ After=network-online.target
 Image=localhost/dns-sync:latest
 ContainerName=dns-sync
 
-Environment=PROXY=npm
-Environment=PROXY_IP=192.168.1.101
+Environment=ENABLED_PLUGINS=npm,verizon
+Environment=DNS_SYNC_PLUGIN_PATH=/plugins
+Environment=DNS_SYNC_STATE_DB=/var/lib/dns-sync/state.sqlite3
+
+Environment=NPM_ENDPOINTS=192.168.1.3,192.168.1.4,192.168.1.5
+Environment=DNS_TARGET_IP=192.168.1.101
 Environment=PROXY_API_PORT=81
 Environment=PROXY_API_SCHEME=http
-Environment=DNS_PROVIDER=verizon
+
+Environment=VERIZON_ENDPOINTS=192.168.1.1
 Environment=DOMAIN_MATCHING=false
 Environment=DOMAIN=loxthelion.com
+
 Environment=SYNC_INTERVAL=1
+Environment=DNS_SYNC_APPLY=false
 
 Environment=NPM_IDENTITY=davidfolkesmd@gmail.com
 
 Secret=dns-sync-npm-secret,type=env,target=NPM_SECRET
 Secret=dns-sync-router-password,type=env,target=ROUTER_PASSWORD
 
+Volume=dns-sync-state:/var/lib/dns-sync:Z
+Volume=./external_plugins:/plugins:ro,Z
+
 [Service]
 Restart=on-failure
 RestartSec=30

dns_sync/adapter_support.py

@@ -0,0 +1,41 @@
+from .adapters import Endpoint, NormalizedObject, PlanAction
+
+
+def require_endpoint_plugin(
+    endpoint: Endpoint,
+    plugin_name: str,
+) -> None:
+    expected = plugin_name.strip().lower()
+
+    if not expected:
+        raise ValueError("Adapter plugin name cannot be empty")
+
+    if endpoint.plugin != expected:
+        raise RuntimeError(
+            f"Adapter {expected!r} cannot use endpoint "
+            f"{endpoint.identity!r}"
+        )
+
+
+def require_action_endpoint(
+    action: PlanAction,
+    endpoint: Endpoint,
+) -> None:
+    if action.endpoint_id != endpoint.identity:
+        raise RuntimeError(
+            f"Plan action for {action.endpoint_id!r} does not belong "
+            f"to adapter endpoint {endpoint.identity!r}"
+        )
+
+
+def require_desired(
+    action: PlanAction,
+    *,
+    context: str = "Adapter apply",
+) -> NormalizedObject:
+    if action.desired is None:
+        raise RuntimeError(
+            f"{context} requires desired state"
+        )
+
+    return action.desired

dns_sync/certificates/models.py

@@ -1,5 +1,4 @@
 from dataclasses import dataclass
-from datetime import datetime
 
 
 @dataclass(frozen=True)
@@ -15,21 +14,3 @@ class CertificateIdentity:
     @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/npm.py

@@ -1,79 +0,0 @@
-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/cli.py

@@ -3,17 +3,14 @@ import os
 import sys
 
 from .config import AppConfig
-from .engine import (
-    build_sync_plan,
-    execute_sync,
-    filter_desired_by_domain,
-    summarize_plan,
-    validate_desired,
-)
 from .executor import execute_endpoint_plans
 from .orchestrator import discover_reconcile_plan
 from .persistence import SQLiteStateStore
-from .registry import load_adapter, load_provider, load_source
+from .registry import (
+    configured_adapter_endpoints,
+    enabled_plugins,
+    load_adapter,
+)
 from .reporting import print_endpoint_plans, print_plan
 
 
@@ -35,10 +32,6 @@ def build_parser():
     parser = argparse.ArgumentParser(prog="dns-sync", description="Synchronize reverse-proxy hostnames to a pluggable DNS provider")
     sub = parser.add_subparsers(dest="command", required=True)
 
-    sync = sub.add_parser("sync", help="Discover, normalize, reconcile, and apply DNS records")
-    sync.add_argument("--dry-run", action="store_true", help="Show the plan without modifying DNS")
-    sync.add_argument("--include-disabled", action="store_true", help="Include disabled source entries when supported")
-
     reconcile = sub.add_parser(
         "reconcile",
         help="Run the persistent generic reconciliation engine",
@@ -68,86 +61,26 @@ def build_parser():
         help="Maximum planned mutations allowed with --apply",
     )
 
-    sub.add_parser("plugins", help="Show the configured source/provider plugins")
     return parser
 
 
-def run_sync(args):
-    config = AppConfig.from_env(include_disabled=args.include_disabled)
-    source = load_source(config.source_name, config)
-    provider = load_provider(config.provider_name, config)
-
-    discovered = source.discover()
-    filtered, filtered_count = filter_desired_by_domain(
-        discovered,
-        domain=config.domain,
-        enabled=config.domain_matching,
-    )
-    desired, warnings = validate_desired(filtered)
-    for warning in warnings:
-        print(f"WARNING: {warning}", file=sys.stderr)
-
-    provider.connect()
-    current = provider.list_records()
-    plan = build_sync_plan(desired, current, provider.capabilities)
-    counts = summarize_plan(plan)
-
-    used, free = provider.capacity()
-    print(f"Source: {source.name}:{config.proxy_url} -> {config.target_ip}")
-    print(f"Provider: {provider.name}")
-    if config.domain_matching:
-        print(
-            f"Domain matching: enabled ({config.domain} + subdomains)"
-        )
-        print(f"Filtered out: {filtered_count}")
-    else:
-        print("Domain matching: disabled")
-    print(f"Desired records: {len(desired)}")
-    if free is None:
-        print(f"Provider capacity: {used} used, unlimited/unknown free")
-    else:
-        print(f"Provider capacity: {used} used, {free} free")
-
-    if free is not None and counts["ADD"] > free:
-        raise RuntimeError(f"Sync requires {counts['ADD']} new records but provider has only {free} free slots")
-
-    print("\nReconciliation plan:")
-    print_plan(plan)
-    print()
-    print(
-        "Plan summary: "
-        f"{counts['ADD']} add, {counts['UPDATE']} update, "
-        f"{counts['UNCHANGED']} unchanged, {counts['SKIP']} skipped"
-    )
-    print("Deletion policy: disabled; provider records absent from desired state are untouched.")
-
-    if args.dry_run:
-        print("DRY RUN: no changes were made.")
-        return 0
-
-    if counts["ADD"] == 0 and counts["UPDATE"] == 0:
-        print("Already synchronized. No changes required.")
-        return 0
-
-    print("\nApplying changes...")
-    execute_sync(provider, plan)
-    print(
-        "\nSync complete: "
-        f"{counts['ADD']} added, {counts['UPDATE']} updated, "
-        f"{counts['UNCHANGED']} unchanged, {counts['SKIP']} skipped"
-    )
-    return 0
-
-
 def _configured_adapters(config):
-    endpoints = (
-        *config.npm_endpoints,
-        *config.verizon_endpoints,
+    adapters = []
+
+    for plugin_name in sorted(enabled_plugins()):
+        for endpoint in configured_adapter_endpoints(
+            plugin_name,
+            config,
+        ):
+            adapters.append(
+                load_adapter(
+                    plugin_name,
+                    config,
+                    endpoint,
                 )
-    return [
-        load_adapter(endpoint.plugin, config, endpoint)
-        for endpoint in endpoints
-    ]
+            )
+
+    return adapters
 
 
 def run_reconcile(args):
@@ -261,15 +194,8 @@ def main():
     parser = build_parser()
     args = parser.parse_args()
     try:
-        if args.command == "sync":
-            return run_sync(args)
         if args.command == "reconcile":
             return run_reconcile(args)
-        if args.command == "plugins":
-            config = AppConfig.from_env()
-            print(f"Source plugin:   {config.source_name}")
-            print(f"Provider plugin: {config.provider_name}")
-            return 0
         parser.error("unknown command")
     except Exception as exc:
         print(f"ERROR: {exc}", file=sys.stderr)

dns_sync/config.py

@@ -79,15 +79,6 @@ def parse_endpoint_list(
     return tuple(endpoints)
 
 
-def endpoint_host(endpoint: Endpoint) -> str:
-    parsed = urlsplit(f"//{endpoint.address}")
-    if not parsed.hostname:
-        raise RuntimeError(
-            f"Invalid endpoint address: {endpoint.address!r}"
-        )
-    return parsed.hostname
-
-
 def env_bool(name: str, default: bool) -> bool:
     value = os.getenv(name)
     if value is None:
@@ -97,83 +88,32 @@ def env_bool(name: str, default: bool) -> bool:
 
 @dataclass(frozen=True)
 class AppConfig:
-    source_name: str
-    provider_name: str
-    npm_endpoints: tuple[Endpoint, ...]
-    verizon_endpoints: tuple[Endpoint, ...]
-    proxy_ip: str
-    proxy_api_port: int
-    proxy_api_scheme: str
     target_ip: str
-    proxy_verify_tls: bool
     domain_matching: bool
     domain: str | None
     include_disabled: bool = False
 
-    @property
-    def proxy_url(self) -> str:
-        return (
-            f"{self.proxy_api_scheme}://"
-            f"{self.npm_endpoints[0].address}"
-        )
-
     @classmethod
     def from_env(cls, include_disabled: bool = False) -> "AppConfig":
-        source_name = os.getenv("PROXY", "npm").strip().lower()
-        provider_name = os.getenv("DNS_PROVIDER", "verizon").strip().lower()
-
-        verizon_endpoints = parse_endpoint_list(
-            "verizon",
-            os.getenv("VERIZON_ENDPOINTS", "192.168.1.1"),
-            443,
-        )
-
-        try:
-            port = int(os.getenv("PROXY_API_PORT", "81"))
-        except ValueError as exc:
-            raise RuntimeError("PROXY_API_PORT must be an integer") from exc
-        if not 1 <= port <= 65535:
-            raise RuntimeError("PROXY_API_PORT must be between 1 and 65535")
-
-        scheme = os.getenv("PROXY_API_SCHEME", "http").strip().lower()
-        if scheme not in {"http", "https"}:
-            raise RuntimeError("PROXY_API_SCHEME must be http or https")
-
-        raw_proxy_ip = os.getenv("PROXY_IP")
-        raw_npm_endpoints = os.getenv("NPM_ENDPOINTS")
-
-        if raw_npm_endpoints is None:
-            if not raw_proxy_ip:
-                raise RuntimeError(
-                    "NPM_ENDPOINTS or legacy PROXY_IP is required"
-                )
-            proxy_ip = normalize_ipv4(raw_proxy_ip, "PROXY_IP")
-            npm_endpoints = parse_endpoint_list("npm", proxy_ip, port)
-        else:
-            npm_endpoints = parse_endpoint_list(
-                "npm",
-                raw_npm_endpoints,
-                port,
-            )
-            if raw_proxy_ip:
-                proxy_ip = normalize_ipv4(raw_proxy_ip, "PROXY_IP")
-            else:
-                proxy_ip = endpoint_host(npm_endpoints[0])
-
         raw_target_ip = os.getenv("DNS_TARGET_IP")
+        raw_proxy_ip = os.getenv("PROXY_IP")
+
         if raw_target_ip:
-            target_ip = normalize_ipv4(raw_target_ip, "DNS_TARGET_IP")
-        elif raw_proxy_ip:
-            target_ip = proxy_ip
-        elif len(npm_endpoints) == 1:
             target_ip = normalize_ipv4(
-                endpoint_host(npm_endpoints[0]),
+                raw_target_ip,
                 "DNS_TARGET_IP",
             )
+        elif raw_proxy_ip:
+            # Deprecated compatibility fallback. Core no longer treats
+            # PROXY_IP as an NPM endpoint.
+            target_ip = normalize_ipv4(
+                raw_proxy_ip,
+                "PROXY_IP",
+            )
         else:
             raise RuntimeError(
-                "DNS_TARGET_IP is required when multiple NPM_ENDPOINTS "
-                "are configured without legacy PROXY_IP"
+                "DNS_TARGET_IP is required "
+                "(legacy PROXY_IP is also accepted)"
             )
 
         domain_matching = env_bool("DOMAIN_MATCHING", False)
@@ -192,15 +132,7 @@ class AppConfig:
             )
 
         return cls(
-            source_name=source_name,
-            provider_name=provider_name,
-            npm_endpoints=npm_endpoints,
-            verizon_endpoints=verizon_endpoints,
-            proxy_ip=proxy_ip,
-            proxy_api_port=port,
-            proxy_api_scheme=scheme,
             target_ip=target_ip,
-            proxy_verify_tls=env_bool("PROXY_VERIFY_TLS", True),
             domain_matching=domain_matching,
             domain=domain,
             include_disabled=include_disabled,

dns_sync/engine.py

@@ -1,110 +0,0 @@
-from .models import DesiredRecord, ProviderRecord, SyncAction
-
-
-def hostname_matches_domain(hostname: str, domain: str) -> bool:
-    """Match a domain apex and any subdomain, without suffix false positives."""
-    candidate = hostname.lower().rstrip(".")
-    if candidate.startswith("*."):
-        candidate = candidate[2:]
-
-    base = domain.lower().rstrip(".")
-    return candidate == base or candidate.endswith("." + base)
-
-
-def filter_desired_by_domain(
-    records: list[DesiredRecord],
-    domain: str | None,
-    enabled: bool,
-) -> tuple[list[DesiredRecord], int]:
-    if not enabled:
-        return records, 0
-
-    if not domain:
-        raise RuntimeError(
-            "Domain filtering enabled without a configured DOMAIN"
-        )
-
-    matched = [
-        record
-        for record in records
-        if hostname_matches_domain(record.hostname, domain)
-    ]
-    return matched, len(records) - len(matched)
-
-
-
-
-def validate_desired(records: list[DesiredRecord]) -> tuple[dict[str, DesiredRecord], list[str]]:
-    desired: dict[str, DesiredRecord] = {}
-    warnings: list[str] = []
-
-    for record in records:
-        key = record.hostname.lower()
-        if key in desired:
-            warnings.append(
-                f"DUPLICATE {record.hostname} from {record.source}"
-                + (f" id={record.source_id}" if record.source_id else "")
-                + "; keeping first occurrence"
-            )
-            continue
-        desired[key] = record
-
-    return dict(sorted(desired.items())), warnings
-
-
-def build_sync_plan(desired, current: dict[str, ProviderRecord], capabilities):
-    plan: list[SyncAction] = []
-
-    for hostname, record in desired.items():
-        if record.record_type not in capabilities.record_types:
-            plan.append(SyncAction("SKIP", record, reason=f"record type {record.record_type} unsupported"))
-            continue
-
-        if record.wildcard and not capabilities.wildcard_records:
-            plan.append(SyncAction("SKIP", record, reason="wildcard records unsupported"))
-            continue
-
-        existing = current.get(hostname.lower())
-        if existing is None:
-            if not capabilities.supports_create:
-                plan.append(SyncAction("SKIP", record, reason="provider cannot create records"))
-            else:
-                plan.append(SyncAction("ADD", record))
-        elif existing.ip == record.ip and existing.record_type == record.record_type:
-            plan.append(SyncAction("UNCHANGED", record, current=existing))
-        elif not capabilities.supports_update:
-            plan.append(SyncAction("SKIP", record, current=existing, reason="provider cannot update records"))
-        else:
-            plan.append(SyncAction("UPDATE", record, current=existing))
-
-    return plan
-
-
-def summarize_plan(plan):
-    counts = {name: 0 for name in ("ADD", "UPDATE", "UNCHANGED", "SKIP")}
-    for item in plan:
-        counts[item.action] = counts.get(item.action, 0) + 1
-    return counts
-
-
-def execute_sync(provider, plan):
-    counts = summarize_plan(plan)
-
-    for item in plan:
-        if item.action in {"UNCHANGED", "SKIP"}:
-            continue
-
-        if item.action == "ADD":
-            detail = provider.add_record(item.desired)
-        elif item.action == "UPDATE":
-            detail = provider.update_record(item.desired)
-        else:
-            raise RuntimeError(f"Unknown sync action: {item.action}")
-
-        if not provider.verify_record(item.desired):
-            raise RuntimeError(f"{item.desired.hostname}: {item.action.lower()} did not verify")
-
-        suffix = f" ({detail})" if detail else ""
-        print(f"APPLIED    {item.action:<6} {item.desired.hostname} -> {item.desired.ip}{suffix}")
-
-    return counts

dns_sync/interfaces.py

@@ -8,7 +8,6 @@ from .adapters import (
     NormalizedObject,
     PlanAction,
 )
-from .models import DesiredRecord, ProviderCapabilities, ProviderRecord
 
 
 class AdapterPlugin(Protocol):
@@ -20,24 +19,3 @@ class AdapterPlugin(Protocol):
     def discover(self, object_type: str) -> list[DiscoveredObject]: ...
     def apply(self, action: PlanAction) -> ApplyResult | None: ...
     def verify(self, object: NormalizedObject) -> bool: ...
-
-
-# Compatibility protocols for the existing DNS-only execution path. They remain
-# in place until the current plugins and CLI are migrated to AdapterPlugin.
-class SourcePlugin(Protocol):
-    name: str
-
-    def discover(self) -> list[DesiredRecord]: ...
-
-
-class ProviderPlugin(Protocol):
-    name: str
-    capabilities: ProviderCapabilities
-
-    def connect(self) -> None: ...
-    def list_records(self) -> dict[str, ProviderRecord]: ...
-    def add_record(self, record: DesiredRecord) -> str | None: ...
-    def update_record(self, record: DesiredRecord) -> str | None: ...
-    def delete_record(self, hostname: str) -> str | None: ...
-    def verify_record(self, record: DesiredRecord) -> bool: ...
-    def capacity(self) -> tuple[int, int | None]: ...

dns_sync/models.py

@@ -1,38 +0,0 @@
-from dataclasses import dataclass, field
-from typing import Any
-
-
-@dataclass(frozen=True)
-class DesiredRecord:
-    hostname: str
-    ip: str
-    record_type: str = "A"
-    wildcard: bool = False
-    source: str = ""
-    source_id: str | None = None
-
-
-@dataclass(frozen=True)
-class ProviderRecord:
-    hostname: str
-    ip: str
-    record_type: str = "A"
-    metadata: dict[str, Any] = field(default_factory=dict)
-
-
-@dataclass(frozen=True)
-class ProviderCapabilities:
-    record_types: frozenset[str] = frozenset({"A"})
-    wildcard_records: bool = False
-    supports_create: bool = True
-    supports_update: bool = True
-    supports_delete: bool = False
-    max_records: int | None = None
-
-
-@dataclass(frozen=True)
-class SyncAction:
-    action: str
-    desired: DesiredRecord
-    current: ProviderRecord | None = None
-    reason: str | None = None

dns_sync/normalizer.py

@@ -1,6 +1,16 @@
 import ipaddress
 import re
 
+def hostname_matches_domain(hostname: str, domain: str) -> bool:
+    """Match a domain apex and any subdomain, without suffix false positives."""
+    candidate = hostname.lower().rstrip(".")
+    if candidate.startswith("*."):
+        candidate = candidate[2:]
+
+    base = domain.lower().rstrip(".")
+    return candidate == base or candidate.endswith("." + base)
+
+
 _LABEL_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
 
 

dns_sync/orchestrator.py

@@ -15,7 +15,7 @@ from .adapters import (
     NormalizedObject,
     Observation,
 )
-from .engine import hostname_matches_domain
+from .normalizer import hostname_matches_domain
 from .object_normalizer import normalize_observations
 from .persistence import SQLiteStateStore
 from .projections import project_proxy_hosts_to_dns

dns_sync/plugins/adapters/npm/__init__.py

@@ -1,6 +1,7 @@
+import os
 from typing import Any
 
-from ...adapters import (
+from ....adapters import (
     CERTIFICATES,
     PROXY_HOSTS,
     ApplyResult,
@@ -13,9 +14,49 @@ from ...adapters import (
     PlanOperation,
     RetireRequest,
 )
-from ...certificates.material import CertificateMaterial
-from ...npm_client import NpmClient
-from ...object_normalizer import normalize_discovered_object
+from ....adapter_support import (
+    require_action_endpoint,
+    require_desired,
+    require_endpoint_plugin,
+)
+from ....certificates.material import CertificateMaterial
+from ....config import parse_endpoint_list
+from ....normalizer import normalize_ipv4
+from .client import NpmClient
+from ....object_normalizer import normalize_discovered_object
+
+
+def configured_endpoints(config):
+    try:
+        port = int(os.getenv("PROXY_API_PORT", "81"))
+    except ValueError as exc:
+        raise RuntimeError(
+            "PROXY_API_PORT must be an integer"
+        ) from exc
+
+    if not 1 <= port <= 65535:
+        raise RuntimeError(
+            "PROXY_API_PORT must be between 1 and 65535"
+        )
+
+    raw_endpoints = os.getenv("NPM_ENDPOINTS")
+    if raw_endpoints is None:
+        raw_proxy_ip = os.getenv("PROXY_IP")
+        if not raw_proxy_ip:
+            raise RuntimeError(
+                "NPM_ENDPOINTS or legacy PROXY_IP is required"
+            )
+
+        raw_endpoints = normalize_ipv4(
+            raw_proxy_ip,
+            "PROXY_IP",
+        )
+
+    return parse_endpoint_list(
+        "npm",
+        raw_endpoints,
+        port,
+    )
 
 
 _PROXY_HOST_FIELDS = (
@@ -67,9 +108,10 @@ class Plugin:
     )
 
     def __init__(self, config, endpoint):
-        self.config = config
+        require_endpoint_plugin(endpoint, self.name)
+        self.endpoints = configured_endpoints(config)
         self.endpoint = endpoint
-        self.client = NpmClient(config, endpoint)
+        self.client = NpmClient(endpoint)
 
     @staticmethod
     def _local_key(row: dict[str, Any], label: str) -> str:
@@ -236,16 +278,13 @@ class Plugin:
         source_errors: list[str] = []
 
         for endpoint in sorted(
-            self.config.npm_endpoints,
+            self.endpoints,
             key=lambda item: item.identity,
         ):
             if endpoint.identity == self.endpoint.identity:
                 continue
 
-            client = NpmClient(
-                self.config,
-                endpoint,
-            )
+            client = NpmClient(endpoint)
 
             try:
                 rows = client.certificates()
@@ -593,16 +632,14 @@ class Plugin:
         self,
         action: PlanAction,
     ) -> ApplyResult | None:
-        desired = action.desired
-
-        if desired is None:
-            raise RuntimeError(
-                "NPM apply requires desired state"
+        desired = require_desired(
+            action,
+            context="NPM apply",
         )
 
-        if action.endpoint_id != self.endpoint.identity:
-            raise RuntimeError(
-                "NPM apply action endpoint does not match adapter"
+        require_action_endpoint(
+            action,
+            self.endpoint,
         )
 
         if (

dns_sync/plugins/adapters/npm/client.py

@@ -7,8 +7,9 @@ from zipfile import BadZipFile, ZipFile
 import requests
 import urllib3
 
-from .adapters import Endpoint
-from .certificates.material import CertificateMaterial
+from ....adapters import Endpoint
+from ....certificates.material import CertificateMaterial
+from ....config import env_bool
 
 
 class NpmClient:
@@ -16,11 +17,9 @@ class NpmClient:
 
     def __init__(
         self,
-        config,
-        endpoint: Endpoint | None = None,
+        endpoint: Endpoint,
     ):
-        self.config = config
-        self.endpoint = endpoint or config.npm_endpoints[0]
+        self.endpoint = endpoint
 
         if self.endpoint.plugin != "npm":
             raise RuntimeError("NpmClient requires an npm endpoint")
@@ -34,10 +33,25 @@ class NpmClient:
         if not self.secret:
             raise RuntimeError("NPM_SECRET is required")
 
+        self.api_scheme = os.getenv(
+            "PROXY_API_SCHEME",
+            "http",
+        ).strip().lower()
+
+        if self.api_scheme not in {"http", "https"}:
+            raise RuntimeError(
+                "PROXY_API_SCHEME must be http or https"
+            )
+
+        self.verify_tls = env_bool(
+            "PROXY_VERIFY_TLS",
+            True,
+        )
+
         self.session = requests.Session()
         self.token: str | None = None
 
-        if not config.proxy_verify_tls:
+        if not self.verify_tls:
             urllib3.disable_warnings(
                 urllib3.exceptions.InsecureRequestWarning
             )
@@ -45,7 +59,7 @@ class NpmClient:
     @property
     def api_url(self) -> str:
         return (
-            f"{self.config.proxy_api_scheme}://"
+            f"{self.api_scheme}://"
             f"{self.endpoint.address}/api"
         )
 
@@ -57,7 +71,7 @@ class NpmClient:
                 "secret": self.secret,
             },
             timeout=15,
-            verify=self.config.proxy_verify_tls,
+            verify=self.verify_tls,
         )
 
         if response.status_code != 200:
@@ -107,7 +121,7 @@ class NpmClient:
             method,
             f"{self.api_url}{path}",
             timeout=kwargs.pop("timeout", 15),
-            verify=self.config.proxy_verify_tls,
+            verify=self.verify_tls,
             **kwargs,
         )
 

dns_sync/plugins/sources/file.py

@@ -1,39 +0,0 @@
-import os
-
-import yaml
-
-from ...models import DesiredRecord
-from ...normalizer import normalize_hostname, normalize_ipv4
-
-
-class Plugin:
-    name = "file"
-
-    def __init__(self, config):
-        self.config = config
-        self.path = os.getenv("DNS_SYNC_FILE")
-        if not self.path:
-            raise RuntimeError("DNS_SYNC_FILE is required when PROXY=file")
-
-    def discover(self):
-        try:
-            with open(self.path, "r", encoding="utf-8") as fh:
-                data = yaml.safe_load(fh)
-        except (OSError, yaml.YAMLError) as exc:
-            raise RuntimeError(f"Could not read desired-state file {self.path}: {exc}") from exc
-
-        if not isinstance(data, dict) or not isinstance(data.get("records"), dict):
-            raise RuntimeError("Desired-state YAML must contain a top-level records mapping")
-
-        result = []
-        for raw_hostname, raw_ip in data["records"].items():
-            hostname, wildcard = normalize_hostname(str(raw_hostname), allow_wildcard=True)
-            result.append(
-                DesiredRecord(
-                    hostname=hostname,
-                    ip=normalize_ipv4(str(raw_ip), hostname),
-                    wildcard=wildcard,
-                    source=self.name,
-                )
-            )
-        return result

dns_sync/plugins/sources/npm.py

@@ -1,41 +0,0 @@
-from ...models import DesiredRecord
-from ...normalizer import normalize_hostname
-from ...npm_client import NpmClient
-
-
-class Plugin:
-    name = "npm"
-
-    def __init__(self, config):
-        self.config = config
-        self.client = NpmClient(config)
-
-    def discover(self):
-        records: list[DesiredRecord] = []
-
-        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,
-                )
-
-                records.append(
-                    DesiredRecord(
-                        hostname=hostname,
-                        ip=self.config.target_ip,
-                        record_type="A",
-                        wildcard=wildcard,
-                        source=self.name,
-                        source_id=source_id,
-                    )
-                )
-
-        return records

dns_sync/registry.py

@@ -1,34 +1,267 @@
 import importlib
+import importlib.util
+import os
+import sys
+from pathlib import Path
+from types import ModuleType
 
 
-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}")
+_EXTERNAL_NAMESPACE = "dns_sync_external_plugins"
 
+
+def _validate_plugin_name(kind: str, name: str) -> str:
+    normalized = name.strip().lower()
+
+    if (
+        not normalized
+        or not normalized.replace("_", "").replace("-", "").isalnum()
+    ):
+        raise RuntimeError(
+            f"Invalid {kind} plugin name: {name!r}"
+        )
+
+    return normalized
+
+
+def enabled_plugins() -> frozenset[str]:
+    raw = os.getenv("ENABLED_PLUGINS", "npm")
+
+    names = [
+        item.strip().lower()
+        for item in raw.split(",")
+    ]
+
+    if not names or any(not name for name in names):
+        raise RuntimeError(
+            "ENABLED_PLUGINS contains an empty plugin name"
+        )
+
+    validated = tuple(
+        _validate_plugin_name("adapter", name)
+        for name in names
+    )
+
+    if len(set(validated)) != len(validated):
+        raise RuntimeError(
+            "ENABLED_PLUGINS contains duplicate plugin names"
+        )
+
+    return frozenset(validated)
+
+
+def _external_plugin_root() -> Path:
+    return Path(
+        os.getenv(
+            "DNS_SYNC_PLUGIN_PATH",
+            "/plugins",
+        )
+    )
+
+
+def _ensure_external_namespace() -> None:
+    if _EXTERNAL_NAMESPACE in sys.modules:
+        return
+
+    namespace = ModuleType(_EXTERNAL_NAMESPACE)
+    namespace.__package__ = _EXTERNAL_NAMESPACE
+    namespace.__path__ = [
+        str(_external_plugin_root())
+    ]
+    sys.modules[_EXTERNAL_NAMESPACE] = namespace
+
+
+def _load_external_adapter_module(name: str):
+    module_name = name.replace("-", "_")
+    package_path = (
+        _external_plugin_root()
+        / module_name
+        / "__init__.py"
+    )
+
+    if not package_path.is_file():
+        return None
+
+    _ensure_external_namespace()
+
+    qualified_name = (
+        f"{_EXTERNAL_NAMESPACE}.{module_name}"
+    )
+
+    existing = sys.modules.get(qualified_name)
+    if existing is not None:
+        existing_file = getattr(
+            existing,
+            "__file__",
+            None,
+        )
+
+        if existing_file is not None:
+            try:
+                same_file = (
+                    Path(existing_file).resolve()
+                    == package_path.resolve()
+                )
+            except OSError:
+                same_file = False
+        else:
+            same_file = False
+
+        if same_file:
+            return existing
+
+        # The same plugin name is being resolved from a different
+        # external plugin root. Do not reuse stale module state.
+        sys.modules.pop(qualified_name, None)
+
+    spec = importlib.util.spec_from_file_location(
+        qualified_name,
+        package_path,
+        submodule_search_locations=[
+            str(package_path.parent)
+        ],
+    )
+
+    if spec is None or spec.loader is None:
+        raise RuntimeError(
+            f"Could not load external adapter plugin "
+            f"{name!r} from {package_path}"
+        )
+
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[qualified_name] = module
+
+    try:
+        spec.loader.exec_module(module)
+    except Exception:
+        sys.modules.pop(qualified_name, None)
+        raise
+
+    return module
+
+
+def _load_builtin(kind: str, name: str):
     module_name = name.replace("-", "_")
     path = f"dns_sync.plugins.{kind}.{module_name}"
 
     try:
-        module = importlib.import_module(path)
+        return importlib.import_module(path)
     except ModuleNotFoundError as exc:
         if exc.name == path:
-            raise RuntimeError(f"Unknown {kind[:-1]} plugin: {name}") from exc
+            return None
         raise
 
-    plugin_class = getattr(module, "Plugin", None)
-    if plugin_class is None:
-        raise RuntimeError(f"Plugin module {path} does not export Plugin")
 
+def _adapter_module(name: str):
+    normalized = _validate_plugin_name("adapter", name)
+
+    enabled = enabled_plugins()
+    if normalized not in enabled:
+        raise RuntimeError(
+            f"Adapter plugin {normalized!r} is not enabled"
+        )
+
+    module = _load_builtin("adapters", normalized)
+
+    if module is not None:
+        return (
+            normalized,
+            module,
+            f"dns_sync.plugins.adapters.{normalized}",
+        )
+
+    module = _load_external_adapter_module(normalized)
+
+    if module is not None:
+        return (
+            normalized,
+            module,
+            str(
+                _external_plugin_root()
+                / normalized
+                / "__init__.py"
+            ),
+        )
+
+    raise RuntimeError(
+        f"Unknown adapter plugin: {normalized}"
+    )
+
+
+def configured_adapter_endpoints(
+    name: str,
+    config,
+):
+    normalized, module, display_path = _adapter_module(name)
+
+    loader = getattr(
+        module,
+        "configured_endpoints",
+        None,
+    )
+
+    if loader is None:
+        raise RuntimeError(
+            f"Adapter plugin {display_path} does not export "
+            "configured_endpoints"
+        )
+
+    endpoints = tuple(loader(config))
+
+    for endpoint in endpoints:
+        if endpoint.plugin != normalized:
+            raise RuntimeError(
+                f"Adapter plugin {normalized!r} returned endpoint "
+                f"for plugin {endpoint.plugin!r}"
+            )
+
+    return endpoints
+
+
+def _plugin_class(module, display_path: str):
+    plugin_class = getattr(module, "Plugin", None)
+
+    if plugin_class is None:
+        raise RuntimeError(
+            f"Plugin module {display_path} does not export Plugin"
+        )
+
+    return plugin_class
+
+
+def _load(kind: str, name: str, config, *args):
+    normalized = _validate_plugin_name(kind, name)
+
+    if kind == "adapters":
+        (
+            normalized,
+            module,
+            display_path,
+        ) = _adapter_module(normalized)
+
+        plugin_class = _plugin_class(
+            module,
+            display_path,
+        )
         return plugin_class(config, *args)
 
+    module = _load_builtin(kind, normalized)
 
-def load_source(name: str, config):
-    return _load("sources", name, config)
+    if module is not None:
+        plugin_class = _plugin_class(
+            module,
+            f"dns_sync.plugins.{kind}.{normalized}",
+        )
+        return plugin_class(config, *args)
 
-
-def load_provider(name: str, config):
-    return _load("providers", name, config)
+    raise RuntimeError(
+        f"Unknown {kind[:-1]} plugin: {normalized}"
+    )
 
 
 def load_adapter(name: str, config, endpoint):
-    return _load("adapters", name, config, endpoint)
+    return _load(
+        "adapters",
+        name,
+        config,
+        endpoint,
+    )

entrypoint.sh

@@ -2,17 +2,44 @@
 set -u
 
 INTERVAL_MINUTES="${SYNC_INTERVAL:-1}"
+APPLY_MODE="${DNS_SYNC_APPLY:-false}"
+STATE_DB="${DNS_SYNC_STATE_DB:-/var/lib/dns-sync/state.sqlite3}"
+MAX_CHANGES="${DNS_SYNC_MAX_CHANGES:-}"
+
 case "$INTERVAL_MINUTES" in
     ''|*[!0-9]*)
         echo "ERROR: SYNC_INTERVAL must be an integer number of minutes" >&2
         exit 1
         ;;
 esac
+
 if [ "$INTERVAL_MINUTES" -lt 1 ]; then
     echo "ERROR: SYNC_INTERVAL must be at least 1 minute" >&2
     exit 1
 fi
 
+case "$(printf '%s' "$APPLY_MODE" | tr '[:upper:]' '[:lower:]')" in
+    1|true|yes|on)
+        APPLY_MODE=true
+        ;;
+    0|false|no|off)
+        APPLY_MODE=false
+        ;;
+    *)
+        echo "ERROR: DNS_SYNC_APPLY must be true or false" >&2
+        exit 1
+        ;;
+esac
+
+if [ -n "$MAX_CHANGES" ]; then
+    case "$MAX_CHANGES" in
+        *[!0-9]*)
+            echo "ERROR: DNS_SYNC_MAX_CHANGES must be a non-negative integer" >&2
+            exit 1
+            ;;
+    esac
+fi
+
 INTERVAL_SECONDS=$((INTERVAL_MINUTES * 60))
 sleep_pid=""
 
@@ -21,22 +48,45 @@ shutdown() {
     [ -z "$sleep_pid" ] || kill "$sleep_pid" 2>/dev/null || true
     exit 0
 }
+
 trap shutdown INT TERM
 
+run_reconcile() {
+    if [ "$APPLY_MODE" = true ]; then
+        if [ -n "$MAX_CHANGES" ]; then
+            python -m dns_sync reconcile \
+                --state-db "$STATE_DB" \
+                --apply \
+                --max-changes "$MAX_CHANGES"
+        else
+            python -m dns_sync reconcile \
+                --state-db "$STATE_DB" \
+                --apply
+        fi
+    else
+        python -m dns_sync reconcile \
+            --state-db "$STATE_DB"
+    fi
+}
+
 echo "DNS sync service started"
 echo "Sync interval: ${INTERVAL_MINUTES} minute(s)"
-echo "Source plugin: ${PROXY:-npm}"
-echo "DNS provider: ${DNS_PROVIDER:-verizon}"
+echo "Reconciliation mode: $([ "$APPLY_MODE" = true ] && echo APPLY || echo READ-ONLY)"
+echo "State database: ${STATE_DB}"
+echo "Enabled plugins: ${ENABLED_PLUGINS:-npm}"
+echo "External plugin path: ${DNS_SYNC_PLUGIN_PATH:-/plugins}"
 
 while :; do
     echo
-    echo "=== DNS sync: $(date -Iseconds) ==="
-    if python -m dns_sync sync; then
-        echo "Sync run completed successfully"
+    echo "=== DNS reconciliation: $(date -Iseconds) ==="
+
+    if run_reconcile; then
+        echo "Reconciliation run completed successfully"
     else
         rc=$?
-        echo "ERROR: Sync run failed with exit code ${rc}" >&2
+        echo "ERROR: Reconciliation run failed with exit code ${rc}" >&2
     fi
+
     sleep "$INTERVAL_SECONDS" &
     sleep_pid=$!
     wait "$sleep_pid" || true

examples/plugin_template/README.md

@@ -0,0 +1,44 @@
+# dns-sync plugin template
+
+This directory is a documentation/example package. dns-sync does not load it
+from this location.
+
+Copy it to one of these locations and rename it for the application:
+
+    dns_sync/plugins/adapters/<plugin_name>/
+
+for a built-in plugin, or:
+
+    $DNS_SYNC_PLUGIN_PATH/<plugin_name>/
+
+for an external plugin.
+
+The canonical package layout is:
+
+    <plugin_name>/
+    ├── __init__.py
+    └── client.py
+
+`__init__.py` is the dns-sync adapter boundary. It owns endpoint discovery,
+generic capabilities, native-to-normalized translation, apply, verify, and
+optional retirement behavior.
+
+`client.py` is private to the plugin. It owns authentication, transport,
+application-specific models, native request/response handling, and endpoint
+operations.
+
+## Before enabling a copied plugin
+
+1. Rename `example`, `ExampleClient`, and `ExampleRecord`.
+2. Implement `configured_endpoints(config)`.
+3. Implement the native client methods.
+4. Declare only capabilities the plugin really supports.
+5. Implement discovery for those object types.
+6. Translate normalized desired state into native state in `apply()`.
+7. Re-discover mutations and return the real endpoint-local key.
+8. Implement `verify()`.
+9. Add `retire()` only when retirement/deletion is safely supported.
+10. Add focused adapter/client tests.
+11. Validate using read-only reconciliation before enabling mutations.
+
+See the root `README.md` for the complete plugin contract and architecture.

examples/plugin_template/__init__.py

@@ -0,0 +1,208 @@
+"""Canonical dns-sync adapter plugin template."""
+
+import os
+
+from dns_sync.adapter_support import (
+    require_action_endpoint,
+    require_desired,
+    require_endpoint_plugin,
+)
+from dns_sync.adapters import (
+    DNS_RECORDS,
+    ApplyResult,
+    Capability,
+    CapabilityOperation,
+    DiscoveredObject,
+    NormalizedObject,
+    PlanAction,
+    PlanOperation,
+)
+from dns_sync.config import parse_endpoint_list
+
+from .client import ExampleClient, ExampleRecord
+
+
+PLUGIN_NAME = "example"
+
+
+def configured_endpoints(config):
+    """Return every endpoint owned by this plugin."""
+    return parse_endpoint_list(
+        PLUGIN_NAME,
+        os.getenv(
+            "EXAMPLE_ENDPOINTS",
+            "example.test",
+        ),
+        443,
+    )
+
+
+class Plugin:
+    """dns-sync adapter for one configured endpoint."""
+
+    name = PLUGIN_NAME
+
+    capabilities = frozenset(
+        {
+            Capability(
+                DNS_RECORDS,
+                CapabilityOperation.DISCOVER,
+            ),
+            Capability(
+                DNS_RECORDS,
+                CapabilityOperation.APPLY,
+            ),
+            Capability(
+                DNS_RECORDS,
+                CapabilityOperation.VERIFY,
+            ),
+        }
+    )
+
+    def __init__(self, config, endpoint):
+        require_endpoint_plugin(
+            endpoint,
+            self.name,
+        )
+
+        self.endpoint = endpoint
+        self.client = ExampleClient(endpoint)
+
+    def connect(self) -> None:
+        self.client.connect()
+
+    def discover(
+        self,
+        object_type: str,
+    ) -> list[DiscoveredObject]:
+        if object_type != DNS_RECORDS:
+            raise RuntimeError(
+                f"{self.endpoint.identity} cannot discover "
+                f"object type {object_type!r}"
+            )
+
+        return [
+            DiscoveredObject(
+                object_type=DNS_RECORDS,
+                local_key=record.local_key,
+                payload={
+                    "hostname": record.hostname,
+                    "ip": record.ip,
+                    "record_type": record.record_type,
+                },
+            )
+            for record in self.client.list_records()
+        ]
+
+    def _native_record(
+        self,
+        object: NormalizedObject,
+    ) -> ExampleRecord:
+        if object.object_type != DNS_RECORDS:
+            raise RuntimeError(
+                f"{self.endpoint.identity} cannot handle "
+                f"object type {object.object_type!r}"
+            )
+
+        payload = object.payload
+        hostname = str(payload.get("hostname") or "")
+
+        if hostname != object.logical_key:
+            raise RuntimeError(
+                "DNS object hostname does not match its logical key"
+            )
+
+        record_type = str(
+            payload.get("record_type") or ""
+        ).upper()
+
+        if record_type != "A":
+            raise RuntimeError(
+                f"Example plugin does not support "
+                f"{record_type!r} records"
+            )
+
+        if payload.get("wildcard") is True:
+            raise RuntimeError(
+                "Example plugin does not support wildcard records"
+            )
+
+        return ExampleRecord(
+            local_key=object.local_key or "",
+            hostname=hostname,
+            ip=str(payload.get("ip") or ""),
+            record_type=record_type,
+        )
+
+    def apply(
+        self,
+        action: PlanAction,
+    ) -> ApplyResult | None:
+        require_action_endpoint(
+            action,
+            self.endpoint,
+        )
+
+        if action.operation in {
+            PlanOperation.NOOP,
+            PlanOperation.SKIP,
+        }:
+            return None
+
+        desired = require_desired(
+            action,
+            context="Example apply action",
+        )
+
+        record = self._native_record(desired)
+
+        if action.operation == PlanOperation.CREATE:
+            detail = self.client.add_record(record)
+        elif action.operation == PlanOperation.UPDATE:
+            detail = self.client.update_record(record)
+        else:
+            raise RuntimeError(
+                f"Example adapter does not support "
+                f"{action.operation.value} actions"
+            )
+
+        records = self.client.list_records()
+        applied = next(
+            (
+                candidate
+                for candidate in records
+                if candidate.hostname.lower()
+                == record.hostname.lower()
+            ),
+            None,
+        )
+
+        if applied is None:
+            raise RuntimeError(
+                "Example mutation succeeded but the object "
+                "could not be rediscovered"
+            )
+
+        normalized = NormalizedObject(
+            object_type=desired.object_type,
+            logical_key=desired.logical_key,
+            local_key=applied.local_key,
+            payload=dict(desired.payload),
+            references=desired.references,
+        )
+
+        return ApplyResult(
+            object=normalized,
+            detail=detail,
+        )
+
+    def verify(
+        self,
+        object: NormalizedObject,
+    ) -> bool:
+        return self.client.verify_record(
+            self._native_record(object)
+        )
+
+    # Add retire(self, request) only when this plugin supports safe
+    # deletion/replacement retirement.

examples/plugin_template/client.py

@@ -0,0 +1,56 @@
+"""Private native client for the example dns-sync plugin."""
+
+from dataclasses import dataclass
+
+from dns_sync.adapters import Endpoint
+
+
+@dataclass(frozen=True)
+class ExampleRecord:
+    """Endpoint-native record used only by this plugin."""
+
+    local_key: str
+    hostname: str
+    ip: str
+    record_type: str = "A"
+
+
+class ExampleClient:
+    """Client for one explicit example endpoint."""
+
+    def __init__(self, endpoint: Endpoint):
+        if endpoint.plugin != "example":
+            raise RuntimeError(
+                "ExampleClient requires an example endpoint"
+            )
+
+        self.endpoint = endpoint
+
+    def connect(self) -> None:
+        """Authenticate or initialize the native connection."""
+        raise NotImplementedError
+
+    def list_records(self) -> list[ExampleRecord]:
+        """Return endpoint-native records."""
+        raise NotImplementedError
+
+    def add_record(
+        self,
+        record: ExampleRecord,
+    ) -> str | None:
+        """Create one native record."""
+        raise NotImplementedError
+
+    def update_record(
+        self,
+        record: ExampleRecord,
+    ) -> str | None:
+        """Update one native record."""
+        raise NotImplementedError
+
+    def verify_record(
+        self,
+        record: ExampleRecord,
+    ) -> bool:
+        """Verify endpoint-native state."""
+        raise NotImplementedError

external_plugins/verizon/__init__.py

@@ -1,4 +1,6 @@
-from ...adapters import (
+import os
+
+from dns_sync.adapters import (
     DNS_RECORDS,
     ApplyResult,
     Capability,
@@ -8,8 +10,24 @@ from ...adapters import (
     PlanAction,
     PlanOperation,
 )
-from ...models import DesiredRecord
-from ..providers.verizon import Plugin as VerizonProvider
+from dns_sync.adapter_support import (
+    require_action_endpoint,
+    require_desired,
+    require_endpoint_plugin,
+)
+from dns_sync.config import parse_endpoint_list
+from .client import VerizonClient, VerizonRecord
+
+
+def configured_endpoints(config):
+    return parse_endpoint_list(
+        "verizon",
+        os.getenv(
+            "VERIZON_ENDPOINTS",
+            "192.168.1.1",
+        ),
+        443,
+    )
 
 
 class Plugin:
@@ -32,17 +50,13 @@ class Plugin:
     )
 
     def __init__(self, config, endpoint):
-        if endpoint.plugin != self.name:
-            raise RuntimeError(
-                "Verizon adapter requires a verizon endpoint"
-            )
+        require_endpoint_plugin(endpoint, self.name)
 
-        self.config = config
         self.endpoint = endpoint
-        self.provider = VerizonProvider(config, endpoint)
+        self.client = VerizonClient(endpoint)
 
     def connect(self) -> None:
-        self.provider.connect()
+        self.client.connect()
 
     def discover(
         self,
@@ -56,7 +70,7 @@ class Plugin:
 
         discovered: list[DiscoveredObject] = []
         records = sorted(
-            self.provider.list_records().values(),
+            self.client.list_records().values(),
             key=lambda record: record.hostname.lower(),
         )
 
@@ -85,7 +99,7 @@ class Plugin:
     def _desired_record(
         self,
         object: NormalizedObject,
-    ) -> DesiredRecord:
+    ) -> VerizonRecord:
         if object.object_type != DNS_RECORDS:
             raise RuntimeError(
                 f"{self.endpoint.identity} cannot handle "
@@ -113,17 +127,16 @@ class Plugin:
                 "Verizon DNS does not support wildcard records"
             )
 
-        return DesiredRecord(
+        return VerizonRecord(
             hostname=hostname,
             ip=str(payload.get("ip") or ""),
             record_type=record_type,
-            wildcard=False,
         )
 
     def apply(self, action: PlanAction) -> ApplyResult | None:
-        if action.endpoint_id != self.endpoint.identity:
-            raise RuntimeError(
-                "Plan action belongs to a different endpoint"
+        require_action_endpoint(
+            action,
+            self.endpoint,
         )
 
         if action.operation in {
@@ -132,24 +145,24 @@ class Plugin:
         }:
             return None
 
-        if action.desired is None:
-            raise RuntimeError(
-                "Verizon apply action requires desired state"
+        desired = require_desired(
+            action,
+            context="Verizon apply action",
         )
 
-        record = self._desired_record(action.desired)
+        record = self._desired_record(desired)
 
         if action.operation == PlanOperation.CREATE:
-            detail = self.provider.add_record(record)
+            detail = self.client.add_record(record)
         elif action.operation == PlanOperation.UPDATE:
-            detail = self.provider.update_record(record)
+            detail = self.client.update_record(record)
         else:
             raise RuntimeError(
                 f"Verizon adapter does not support "
                 f"{action.operation.value} actions"
             )
 
-        records = self.provider.list_records()
+        records = self.client.list_records()
         applied_record = records.get(record.hostname.lower())
         if applied_record is None:
             applied_record = records.get(record.hostname)
@@ -168,11 +181,11 @@ class Plugin:
             )
 
         applied = NormalizedObject(
-            object_type=action.desired.object_type,
-            logical_key=action.desired.logical_key,
+            object_type=desired.object_type,
+            logical_key=desired.logical_key,
             local_key=str(slot),
-            payload=dict(action.desired.payload),
-            references=action.desired.references,
+            payload=dict(desired.payload),
+            references=desired.references,
         )
 
         return ApplyResult(
@@ -181,6 +194,6 @@ class Plugin:
         )
 
     def verify(self, object: NormalizedObject) -> bool:
-        return self.provider.verify_record(
+        return self.client.verify_record(
             self._desired_record(object)
         )

external_plugins/verizon/client.py

@@ -1,14 +1,15 @@
 import hashlib
+from dataclasses import dataclass, field
 import os
 import re
 import time
+from typing import Any
 
 import requests
 import urllib3
 
-from ...adapters import Endpoint
-from ...models import ProviderCapabilities, ProviderRecord
-from ...normalizer import normalize_hostname, normalize_ipv4
+from dns_sync.adapters import Endpoint
+from dns_sync.normalizer import normalize_hostname, normalize_ipv4
 
 LUCI_USERNAME = (
     "edbd881f1ee2f76ba0bd70fd184f87711be991a0401fd07ccd4b199665f00761"
@@ -17,6 +18,15 @@ LUCI_USERNAME = (
 CFG_RE = re.compile(r'addCfg\("dns_(ip|name)(\d+)",\s*"([^"]*)",\s*"([^"]*)"\);')
 
 
+
+@dataclass(frozen=True)
+class VerizonRecord:
+    hostname: str
+    ip: str
+    record_type: str = "A"
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
 class _Slot:
     def __init__(self, slot):
         self.slot = slot
@@ -26,40 +36,23 @@ class _Slot:
         self.ip_token = ""
 
 
-class Plugin:
-    name = "verizon"
-    capabilities = ProviderCapabilities(
-        record_types=frozenset({"A"}),
-        wildcard_records=False,
-        supports_create=True,
-        supports_update=True,
-        supports_delete=True,
-        max_records=64,
-    )
-
+class VerizonClient:
     def __init__(
         self,
-        config,
-        endpoint: Endpoint | None = None,
+        endpoint: Endpoint,
     ):
-        self.config = config
         self.endpoint = endpoint
 
-        if endpoint is None:
-            self.router_url = os.getenv(
-                "ROUTER_URL",
-                "https://192.168.1.1",
-            ).rstrip("/")
-        else:
         if endpoint.plugin != "verizon":
             raise RuntimeError(
-                    "Verizon provider requires a verizon endpoint"
+                "VerizonClient requires a verizon endpoint"
             )
+
         self.router_url = f"https://{endpoint.address}"
 
         self.password = os.getenv("ROUTER_PASSWORD")
         if not self.password:
-            raise RuntimeError("ROUTER_PASSWORD is required when DNS_PROVIDER=verizon")
+            raise RuntimeError("ROUTER_PASSWORD is required by the Verizon adapter")
 
         self.timeout = int(os.getenv("ROUTER_TIMEOUT", "10"))
         self.session = requests.Session()
@@ -161,7 +154,7 @@ class Plugin:
                 # The firmware UI has no wildcard support, but normalize defensively.
                 pass
             ip = normalize_ipv4(slot.ip, hostname)
-            result[hostname.lower()] = ProviderRecord(
+            result[hostname.lower()] = VerizonRecord(
                 hostname=hostname,
                 ip=ip,
                 record_type="A",
@@ -176,7 +169,7 @@ class Plugin:
 
     def _write_slot(self, slot, hostname, ip):
         if not self.auth_token:
-            raise RuntimeError("Verizon provider is not authenticated")
+            raise RuntimeError("Verizon client is not authenticated")
         if not slot.ip_token or not slot.name_token:
             raise RuntimeError(f"Slot {slot.slot} is missing dynamic field tokens")
 

tests/test_adapter_support.py

@@ -0,0 +1,106 @@
+import unittest
+
+from dns_sync.adapter_support import (
+    require_action_endpoint,
+    require_desired,
+    require_endpoint_plugin,
+)
+from dns_sync.adapters import (
+    Endpoint,
+    NormalizedObject,
+    PlanAction,
+    PlanOperation,
+)
+
+
+class AdapterSupportTests(unittest.TestCase):
+    def test_endpoint_plugin_accepts_matching_plugin(self):
+        endpoint = Endpoint("npm", "node.example:81")
+
+        require_endpoint_plugin(endpoint, "npm")
+
+
+    def test_endpoint_plugin_rejects_wrong_plugin(self):
+        endpoint = Endpoint("npm", "node.example:81")
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "cannot use endpoint",
+        ):
+            require_endpoint_plugin(
+                endpoint,
+                "verizon",
+            )
+
+
+    def test_endpoint_plugin_rejects_empty_expected_name(self):
+        endpoint = Endpoint("npm", "node.example:81")
+
+        with self.assertRaisesRegex(
+            ValueError,
+            "cannot be empty",
+        ):
+            require_endpoint_plugin(endpoint, " ")
+
+
+    def test_action_endpoint_accepts_matching_endpoint(self):
+        endpoint = Endpoint("npm", "node.example:81")
+        action = PlanAction(
+            endpoint_id=endpoint.identity,
+            operation=PlanOperation.CREATE,
+        )
+
+        require_action_endpoint(action, endpoint)
+
+
+    def test_action_endpoint_rejects_other_endpoint(self):
+        endpoint = Endpoint("npm", "node.example:81")
+        action = PlanAction(
+            endpoint_id="npm://other.example:81",
+            operation=PlanOperation.CREATE,
+        )
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "does not belong",
+        ):
+            require_action_endpoint(action, endpoint)
+
+
+    def test_require_desired_returns_desired_object(self):
+        desired = NormalizedObject(
+            object_type="example",
+            logical_key="object",
+            payload={},
+        )
+
+        action = PlanAction(
+            endpoint_id="example://endpoint",
+            operation=PlanOperation.CREATE,
+            desired=desired,
+        )
+
+        self.assertIs(
+            require_desired(action),
+            desired,
+        )
+
+
+    def test_require_desired_fails_closed(self):
+        action = PlanAction(
+            endpoint_id="example://endpoint",
+            operation=PlanOperation.CREATE,
+        )
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "Test mutation requires desired state",
+        ):
+            require_desired(
+                action,
+                context="Test mutation",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_cli_reconcile.py

@@ -150,6 +150,15 @@ class ReconcileCliTests(unittest.TestCase):
             verizon_endpoint.identity: verizon,
         }
 
+        def endpoint_loader(name, config):
+            if name == "npm":
+                return (npm_endpoint,)
+            if name == "verizon":
+                return (verizon_endpoint,)
+            raise AssertionError(
+                f"Unexpected adapter plugin: {name}"
+            )
+
         def loader(name, config, endpoint):
             return adapters[endpoint.identity]
 
@@ -164,8 +173,12 @@ class ReconcileCliTests(unittest.TestCase):
                     "NPM_ENDPOINTS": "192.168.1.3",
                     "VERIZON_ENDPOINTS": "192.168.1.1",
                     "DNS_TARGET_IP": "192.168.1.101",
+                    "ENABLED_PLUGINS": "npm,verizon",
                 },
                 clear=True,
+            ), patch(
+                "dns_sync.cli.configured_adapter_endpoints",
+                side_effect=endpoint_loader,
             ), patch(
                 "dns_sync.cli.load_adapter",
                 side_effect=loader,

tests/test_config.py

@@ -47,29 +47,27 @@ class EndpointParsingTests(unittest.TestCase):
             )
 
 
-class AppConfigEndpointTests(unittest.TestCase):
-    def test_legacy_proxy_ip_remains_supported(self):
+class AppConfigPolicyTests(unittest.TestCase):
+    def test_explicit_dns_target_is_used(self):
         with patch.dict(
             os.environ,
-            {"PROXY_IP": "192.168.1.101"},
+            {
+                "DNS_TARGET_IP": "192.168.1.101",
+                "NPM_ENDPOINTS": "192.168.1.3,192.168.1.4",
+            },
             clear=True,
         ):
             config = AppConfig.from_env()
 
         self.assertEqual(
-            tuple(endpoint.identity for endpoint in config.npm_endpoints),
-            ("npm://192.168.1.101:81",),
+            config.target_ip,
+            "192.168.1.101",
         )
-        self.assertEqual(config.proxy_url, "http://192.168.1.101:81")
-        self.assertEqual(config.target_ip, "192.168.1.101")
 
-    def test_endpoint_list_controls_api_and_proxy_ip_remains_target(self):
+    def test_legacy_proxy_ip_remains_target_fallback(self):
         with patch.dict(
             os.environ,
             {
-                "NPM_ENDPOINTS": (
-                    "192.168.1.3,192.168.1.4:8081,192.168.1.5"
-                ),
                 "PROXY_IP": "192.168.1.101",
             },
             clear=True,
@@ -77,35 +75,15 @@ class AppConfigEndpointTests(unittest.TestCase):
             config = AppConfig.from_env()
 
         self.assertEqual(
-            tuple(endpoint.identity for endpoint in config.npm_endpoints),
-            (
-                "npm://192.168.1.3:81",
-                "npm://192.168.1.4:8081",
-                "npm://192.168.1.5:81",
-            ),
+            config.target_ip,
+            "192.168.1.101",
         )
-        self.assertEqual(config.proxy_url, "http://192.168.1.3:81")
-        self.assertEqual(config.target_ip, "192.168.1.101")
 
-    def test_explicit_dns_target_supports_multiple_endpoints(self):
+    def test_dns_target_is_required_without_legacy_fallback(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,
-        ):
-            config = AppConfig.from_env()
-
-        self.assertEqual(config.proxy_url, "http://192.168.1.3:81")
-        self.assertEqual(config.target_ip, "192.168.1.101")
-
-    def test_multiple_endpoints_require_target_without_proxy_ip(self):
-        with patch.dict(
-            os.environ,
-            {
-                "NPM_ENDPOINTS": "192.168.1.3,192.168.1.4",
+                "NPM_ENDPOINTS": "192.168.1.3",
             },
             clear=True,
         ):
@@ -115,49 +93,6 @@ class AppConfigEndpointTests(unittest.TestCase):
             ):
                 AppConfig.from_env()
 
-    def test_verizon_endpoint_list_is_plugin_specific(self):
-        with patch.dict(
-            os.environ,
-            {
-                "NPM_ENDPOINTS": "192.168.1.3",
-                "VERIZON_ENDPOINTS": (
-                    "192.168.1.1,192.168.1.254:8443"
-                ),
-            },
-            clear=True,
-        ):
-            config = AppConfig.from_env()
-
-        self.assertEqual(
-            tuple(
-                endpoint.identity
-                for endpoint in config.verizon_endpoints
-            ),
-            (
-                "verizon://192.168.1.1:443",
-                "verizon://192.168.1.254:8443",
-            ),
-        )
-
-    def test_router_url_does_not_configure_generic_endpoint(self):
-        with patch.dict(
-            os.environ,
-            {
-                "NPM_ENDPOINTS": "192.168.1.3",
-                "ROUTER_URL": "https://192.168.1.254",
-            },
-            clear=True,
-        ):
-            config = AppConfig.from_env()
-
-        self.assertEqual(
-            tuple(
-                endpoint.identity
-                for endpoint in config.verizon_endpoints
-            ),
-            ("verizon://192.168.1.1:443",),
-        )
-
 
 if __name__ == "__main__":
     unittest.main()

tests/test_legacy_dns_engine.py

@@ -1,54 +0,0 @@
-import unittest
-
-from dns_sync.engine import build_sync_plan
-from dns_sync.models import (
-    DesiredRecord,
-    ProviderCapabilities,
-    ProviderRecord,
-)
-
-
-class LegacyDnsEngineTests(unittest.TestCase):
-    def test_existing_dns_plan_still_adds_updates_and_preserves_matches(self):
-        desired = {
-            "add.example.com": DesiredRecord(
-                hostname="add.example.com",
-                ip="192.168.1.101",
-            ),
-            "keep.example.com": DesiredRecord(
-                hostname="keep.example.com",
-                ip="192.168.1.101",
-            ),
-            "update.example.com": DesiredRecord(
-                hostname="update.example.com",
-                ip="192.168.1.101",
-            ),
-        }
-        current = {
-            "keep.example.com": ProviderRecord(
-                hostname="keep.example.com",
-                ip="192.168.1.101",
-            ),
-            "update.example.com": ProviderRecord(
-                hostname="update.example.com",
-                ip="192.168.1.50",
-            ),
-        }
-
-        plan = build_sync_plan(
-            desired,
-            current,
-            ProviderCapabilities(),
-        )
-
-        actions = {
-            item.desired.hostname: item.action
-            for item in plan
-        }
-        self.assertEqual(actions["add.example.com"], "ADD")
-        self.assertEqual(actions["keep.example.com"], "UNCHANGED")
-        self.assertEqual(actions["update.example.com"], "UPDATE")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_npm_adapter.py

@@ -6,6 +6,7 @@ from dns_sync.adapters import (
     CERTIFICATES,
     PROXY_HOSTS,
     Capability,
+    Endpoint,
     CapabilityOperation,
     NormalizedObject,
     ObjectReference,
@@ -15,23 +16,38 @@ from dns_sync.adapters import (
 from dns_sync.certificates.material import CertificateMaterial
 from dns_sync.config import AppConfig
 from dns_sync.object_normalizer import normalize_discovered_object
-from dns_sync.registry import load_adapter
+from dns_sync.registry import (
+    configured_adapter_endpoints,
+    load_adapter,
+)
 
 
 class NpmAdapterTests(unittest.TestCase):
-    def config(self):
-        with patch.dict(
+    @staticmethod
+    def npm_env(
+        endpoints="192.168.1.3,192.168.1.4",
+    ):
+        return patch.dict(
             os.environ,
             {
-                "NPM_ENDPOINTS": "192.168.1.3,192.168.1.4",
+                "NPM_ENDPOINTS": endpoints,
                 "DNS_TARGET_IP": "192.168.1.101",
             },
             clear=True,
-        ):
+        )
+
+    def config(self):
+        with self.npm_env():
             return AppConfig.from_env()
 
     def adapter(self, client):
-        config = self.config()
+        with self.npm_env():
+            config = AppConfig.from_env()
+            endpoints = configured_adapter_endpoints(
+                "npm",
+                config,
+            )
+
             with patch(
                 "dns_sync.plugins.adapters.npm.NpmClient",
                 return_value=client,
@@ -39,7 +55,7 @@ class NpmAdapterTests(unittest.TestCase):
                 return load_adapter(
                     "npm",
                     config,
-                config.npm_endpoints[0],
+                    endpoints[0],
                 )
 
     def target_adapter_with_sources(
@@ -47,7 +63,12 @@ class NpmAdapterTests(unittest.TestCase):
         target_client,
         source_clients,
     ):
-        config = self.config()
+        with self.npm_env():
+            config = AppConfig.from_env()
+            endpoints = configured_adapter_endpoints(
+                "npm",
+                config,
+            )
 
             with patch(
                 "dns_sync.plugins.adapters.npm.NpmClient",
@@ -56,7 +77,7 @@ class NpmAdapterTests(unittest.TestCase):
                 adapter = load_adapter(
                     "npm",
                     config,
-                config.npm_endpoints[1],
+                    endpoints[1],
                 )
 
         client_by_identity = {
@@ -64,23 +85,33 @@ class NpmAdapterTests(unittest.TestCase):
             for endpoint_id, client in source_clients.items()
         }
 
-        def client_factory(config, endpoint):
+        def client_factory(endpoint):
             return client_by_identity[endpoint.identity]
 
         return adapter, client_factory
 
     def test_registry_creates_one_adapter_per_endpoint(self):
-        config = self.config()
         first_client = Mock()
         second_client = Mock()
 
+        with self.npm_env():
+            config = AppConfig.from_env()
+            endpoints = configured_adapter_endpoints(
+                "npm",
+                config,
+            )
+
             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
+                    load_adapter(
+                        "npm",
+                        config,
+                        endpoint,
+                    )
+                    for endpoint in endpoints
                 ]
 
         self.assertEqual(
@@ -94,6 +125,29 @@ class NpmAdapterTests(unittest.TestCase):
         self.assertIs(adapters[0].client, first_client)
         self.assertIs(adapters[1].client, second_client)
 
+    def test_adapter_rejects_endpoint_for_other_plugin(self):
+        config = self.config()
+        endpoint = Endpoint(
+            "verizon",
+            "192.168.1.1:443",
+        )
+
+        with patch(
+            "dns_sync.plugins.adapters.npm.NpmClient"
+        ) as client_class:
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "cannot use endpoint",
+            ):
+                load_adapter(
+                    "npm",
+                    config,
+                    endpoint,
+                )
+
+        client_class.assert_not_called()
+
+
     def test_capabilities_cover_discovery_apply_and_verify(self):
         adapter = self.adapter(Mock())
 
@@ -493,24 +547,13 @@ class NpmAdapterTests(unittest.TestCase):
         source_client.certificate_material.assert_not_called()
 
     def test_certificate_material_source_rejects_conflicting_material(self):
-        config = self.config()
-
-        # Add a third endpoint for this ambiguity test.
-        endpoint_type = type(config.npm_endpoints[0])
-        third_endpoint = endpoint_type(
+        with self.npm_env(
+            "192.168.1.3,192.168.1.4,192.168.1.5"
+        ):
+            config = AppConfig.from_env()
+            endpoints = configured_adapter_endpoints(
                 "npm",
-            "192.168.1.5:81",
-        )
-
-        config = __import__(
-            "dataclasses"
-        ).replace(
                 config,
-            npm_endpoints=(
-                config.npm_endpoints[0],
-                config.npm_endpoints[1],
-                third_endpoint,
-            ),
             )
 
         target_client = Mock()
@@ -544,6 +587,9 @@ class NpmAdapterTests(unittest.TestCase):
             )
         )
 
+        with self.npm_env(
+            "192.168.1.3,192.168.1.4,192.168.1.5"
+        ):
             with patch(
                 "dns_sync.plugins.adapters.npm.NpmClient",
                 return_value=target_client,
@@ -551,7 +597,7 @@ class NpmAdapterTests(unittest.TestCase):
                 adapter = load_adapter(
                     "npm",
                     config,
-                config.npm_endpoints[1],
+                    endpoints[1],
                 )
 
         clients = {
@@ -559,7 +605,7 @@ class NpmAdapterTests(unittest.TestCase):
             "npm://192.168.1.5:81": source_b,
         }
 
-        def client_factory(config, endpoint):
+        def client_factory(endpoint):
             return clients[endpoint.identity]
 
         desired = NormalizedObject(
@@ -583,23 +629,13 @@ class NpmAdapterTests(unittest.TestCase):
                 adapter._certificate_material_source(desired)
 
     def test_certificate_material_source_accepts_identical_replicas(self):
-        config = self.config()
-
-        endpoint_type = type(config.npm_endpoints[0])
-        third_endpoint = endpoint_type(
+        with self.npm_env(
+            "192.168.1.3,192.168.1.4,192.168.1.5"
+        ):
+            config = AppConfig.from_env()
+            endpoints = configured_adapter_endpoints(
                 "npm",
-            "192.168.1.5:81",
-        )
-
-        config = __import__(
-            "dataclasses"
-        ).replace(
                 config,
-            npm_endpoints=(
-                config.npm_endpoints[0],
-                config.npm_endpoints[1],
-                third_endpoint,
-            ),
             )
 
         target_client = Mock()
@@ -636,6 +672,9 @@ class NpmAdapterTests(unittest.TestCase):
         source_a.certificate_material.return_value = common_material
         source_b.certificate_material.return_value = common_material
 
+        with self.npm_env(
+            "192.168.1.3,192.168.1.4,192.168.1.5"
+        ):
             with patch(
                 "dns_sync.plugins.adapters.npm.NpmClient",
                 return_value=target_client,
@@ -643,7 +682,7 @@ class NpmAdapterTests(unittest.TestCase):
                 adapter = load_adapter(
                     "npm",
                     config,
-                config.npm_endpoints[1],
+                    endpoints[1],
                 )
 
         clients = {
@@ -651,7 +690,7 @@ class NpmAdapterTests(unittest.TestCase):
             "npm://192.168.1.5:81": source_b,
         }
 
-        def client_factory(config, endpoint):
+        def client_factory(endpoint):
             return clients[endpoint.identity]
 
         desired = NormalizedObject(
@@ -675,23 +714,13 @@ class NpmAdapterTests(unittest.TestCase):
         self.assertEqual(resolved, common_material)
 
     def test_certificate_material_source_tolerates_unavailable_unrelated_source(self):
-        config = self.config()
-
-        endpoint_type = type(config.npm_endpoints[0])
-        third_endpoint = endpoint_type(
+        with self.npm_env(
+            "192.168.1.3,192.168.1.4,192.168.1.5"
+        ):
+            config = AppConfig.from_env()
+            endpoints = configured_adapter_endpoints(
                 "npm",
-            "192.168.1.5:81",
-        )
-
-        config = __import__(
-            "dataclasses"
-        ).replace(
                 config,
-            npm_endpoints=(
-                config.npm_endpoints[0],
-                config.npm_endpoints[1],
-                third_endpoint,
-            ),
             )
 
         target_client = Mock()
@@ -719,6 +748,9 @@ class NpmAdapterTests(unittest.TestCase):
         )
         available.certificate_material.return_value = material
 
+        with self.npm_env(
+            "192.168.1.3,192.168.1.4,192.168.1.5"
+        ):
             with patch(
                 "dns_sync.plugins.adapters.npm.NpmClient",
                 return_value=target_client,
@@ -726,7 +758,7 @@ class NpmAdapterTests(unittest.TestCase):
                 adapter = load_adapter(
                     "npm",
                     config,
-                config.npm_endpoints[1],
+                    endpoints[1],
                 )
 
         clients = {
@@ -734,7 +766,7 @@ class NpmAdapterTests(unittest.TestCase):
             "npm://192.168.1.5:81": available,
         }
 
-        def client_factory(config, endpoint):
+        def client_factory(endpoint):
             return clients[endpoint.identity]
 
         desired = NormalizedObject(

tests/test_npm_client.py

@@ -5,48 +5,49 @@ from zipfile import ZipFile
 from unittest.mock import Mock, patch
 
 from dns_sync.adapters import Endpoint
-from dns_sync.config import AppConfig
 from dns_sync.certificates.material import CertificateMaterial
-from dns_sync.npm_client import NpmClient
+from dns_sync.plugins.adapters.npm.client import NpmClient
 
 
 class NpmClientEndpointTests(unittest.TestCase):
-    def config(self):
-        return AppConfig.from_env()
-
-    def test_default_client_uses_first_configured_endpoint(self):
+    def test_client_uses_explicit_endpoint(self):
         with patch.dict(
             os.environ,
             {
-                "NPM_ENDPOINTS": "192.168.1.3,192.168.1.4:8081",
-                "DNS_TARGET_IP": "192.168.1.101",
                 "NPM_IDENTITY": "test@example.com",
                 "NPM_SECRET": "not-a-real-secret",
             },
             clear=True,
         ):
-            config = self.config()
+            endpoint = Endpoint(
+                "npm",
+                "192.168.1.3:81",
+            )
             with patch(
-                "dns_sync.npm_client.requests.Session.request"
+                "dns_sync.plugins.adapters.npm.client.requests.Session.request"
             ) as request:
-                client = NpmClient(config)
+                client = NpmClient(endpoint)
 
         request.assert_not_called()
-        self.assertEqual(client.api_url, "http://192.168.1.3:81/api")
+        self.assertEqual(
+            client.api_url,
+            "http://192.168.1.3:81/api",
+        )
 
-    def test_client_can_select_specific_configured_endpoint(self):
+    def test_client_can_select_specific_endpoint(self):
         with patch.dict(
             os.environ,
             {
-                "NPM_ENDPOINTS": "192.168.1.3,192.168.1.4:8081",
-                "DNS_TARGET_IP": "192.168.1.101",
                 "NPM_IDENTITY": "test@example.com",
                 "NPM_SECRET": "not-a-real-secret",
             },
             clear=True,
         ):
-            config = self.config()
-            client = NpmClient(config, config.npm_endpoints[1])
+            endpoint = Endpoint(
+                "npm",
+                "192.168.1.4:8081",
+            )
+            client = NpmClient(endpoint)
 
         self.assertEqual(
             client.api_url,
@@ -61,19 +62,66 @@ class NpmClientEndpointTests(unittest.TestCase):
         with patch.dict(
             os.environ,
             {
-                "PROXY_IP": "192.168.1.101",
                 "NPM_IDENTITY": "test@example.com",
                 "NPM_SECRET": "not-a-real-secret",
             },
             clear=True,
         ):
-            config = self.config()
-            endpoint = Endpoint("verizon", "192.168.1.1")
+            endpoint = Endpoint(
+                "verizon",
+                "192.168.1.1:443",
+            )
 
-            with self.assertRaisesRegex(RuntimeError, "npm endpoint"):
-                NpmClient(config, endpoint)
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "npm endpoint",
+            ):
+                NpmClient(endpoint)
 
+    def test_client_rejects_invalid_api_scheme(self):
+        with patch.dict(
+            os.environ,
+            {
+                "NPM_IDENTITY": "test@example.com",
+                "NPM_SECRET": "not-a-real-secret",
+                "PROXY_API_SCHEME": "ftp",
+            },
+            clear=True,
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "PROXY_API_SCHEME",
+            ):
+                NpmClient(
+                    Endpoint(
+                        "npm",
+                        "192.168.1.3:81",
+                    )
+                )
 
+    def test_client_owns_tls_verification_setting(self):
+        with patch.dict(
+            os.environ,
+            {
+                "NPM_IDENTITY": "test@example.com",
+                "NPM_SECRET": "not-a-real-secret",
+                "PROXY_API_SCHEME": "https",
+                "PROXY_VERIFY_TLS": "false",
+            },
+            clear=True,
+        ):
+            client = NpmClient(
+                Endpoint(
+                    "npm",
+                    "192.168.1.3:81",
+                )
+            )
+
+        self.assertEqual(
+            client.api_url,
+            "https://192.168.1.3:81/api",
+        )
+        self.assertFalse(client.verify_tls)
 
 
 class NpmClientMutationTests(unittest.TestCase):
@@ -81,17 +129,16 @@ class NpmClientMutationTests(unittest.TestCase):
         with patch.dict(
             os.environ,
             {
-                "NPM_ENDPOINTS": "192.168.1.3",
-                "DNS_TARGET_IP": "192.168.1.101",
                 "NPM_IDENTITY": "test@example.com",
                 "NPM_SECRET": "not-a-real-secret",
             },
             clear=True,
         ):
-            config = AppConfig.from_env()
             return NpmClient(
-                config,
-                config.npm_endpoints[0],
+                Endpoint(
+                    "npm",
+                    "192.168.1.3:81",
+                )
             )
 
     @staticmethod

tests/test_plugin_template.py

@@ -0,0 +1,73 @@
+import os
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from dns_sync.registry import (
+    configured_adapter_endpoints,
+    load_adapter,
+)
+
+
+class PluginTemplateTests(unittest.TestCase):
+    def test_template_loads_as_external_plugin(self):
+        source = Path("examples/plugin_template")
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            plugin = root / "example"
+
+            shutil.copytree(
+                source,
+                plugin,
+            )
+
+            with patch.dict(
+                os.environ,
+                {
+                    "ENABLED_PLUGINS": "example",
+                    "DNS_SYNC_PLUGIN_PATH": directory,
+                    "EXAMPLE_ENDPOINTS": "example.test",
+                },
+                clear=True,
+            ):
+                config = SimpleNamespace()
+
+                endpoints = configured_adapter_endpoints(
+                    "example",
+                    config,
+                )
+
+                self.assertEqual(
+                    tuple(
+                        endpoint.identity
+                        for endpoint in endpoints
+                    ),
+                    ("example://example.test:443",),
+                )
+
+                adapter = load_adapter(
+                    "example",
+                    config,
+                    endpoints[0],
+                )
+
+                self.assertEqual(
+                    adapter.name,
+                    "example",
+                )
+                self.assertEqual(
+                    adapter.endpoint,
+                    endpoints[0],
+                )
+                self.assertEqual(
+                    adapter.client.endpoint,
+                    endpoints[0],
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_registry.py

@@ -0,0 +1,511 @@
+import os
+import tempfile
+import textwrap
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from dns_sync.adapters import Endpoint
+from dns_sync.registry import (
+    configured_adapter_endpoints,
+    enabled_plugins,
+    load_adapter,
+)
+
+
+class PluginRegistryTests(unittest.TestCase):
+    def test_default_enabled_plugin_is_npm(self):
+        with patch.dict(os.environ, {}, clear=True):
+            self.assertEqual(
+                enabled_plugins(),
+                frozenset({"npm"}),
+            )
+
+    def test_enabled_plugins_are_explicit(self):
+        with patch.dict(
+            os.environ,
+            {
+                "ENABLED_PLUGINS": (
+                    "npm,verizon"
+                )
+            },
+            clear=True,
+        ):
+            self.assertEqual(
+                enabled_plugins(),
+                frozenset(
+                    {
+                        "npm",
+                        "verizon",
+                    }
+                ),
+            )
+
+    def test_duplicate_enabled_plugin_is_rejected(self):
+        with patch.dict(
+            os.environ,
+            {
+                "ENABLED_PLUGINS": "npm,npm",
+            },
+            clear=True,
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "duplicate plugin",
+            ):
+                enabled_plugins()
+
+    def test_disabled_adapter_is_not_loaded(self):
+        config = SimpleNamespace()
+        endpoint = Endpoint(
+            "verizon",
+            "192.168.1.1:443",
+        )
+
+        with patch.dict(
+            os.environ,
+            {
+                "ENABLED_PLUGINS": "npm",
+            },
+            clear=True,
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "is not enabled",
+            ):
+                load_adapter(
+                    "verizon",
+                    config,
+                    endpoint,
+                )
+
+    def test_external_adapter_is_loaded(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            plugin = root / "example"
+            plugin.mkdir()
+
+            (plugin / "__init__.py").write_text(
+                textwrap.dedent(
+                    """
+                    class Plugin:
+                        name = "example"
+
+                        def __init__(
+                            self,
+                            config,
+                            endpoint,
+                        ):
+                            self.config = config
+                            self.endpoint = endpoint
+                    """
+                )
+            )
+
+            config = SimpleNamespace()
+            endpoint = Endpoint(
+                "example",
+                "example.test:443",
+            )
+
+            with patch.dict(
+                os.environ,
+                {
+                    "ENABLED_PLUGINS": (
+                        "example"
+                    ),
+                    "DNS_SYNC_PLUGIN_PATH": (
+                        directory
+                    ),
+                },
+                clear=True,
+            ):
+                adapter = load_adapter(
+                    "example",
+                    config,
+                    endpoint,
+                )
+
+            self.assertEqual(
+                adapter.name,
+                "example",
+            )
+            self.assertEqual(
+                adapter.endpoint,
+                endpoint,
+            )
+
+    def test_external_adapter_can_import_plugin_submodule(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            plugin = root / "example_package"
+            plugin.mkdir()
+
+            (plugin / "helper.py").write_text(
+                "VALUE = 'loaded-from-submodule'\n"
+            )
+
+            (plugin / "__init__.py").write_text(
+                textwrap.dedent(
+                    """
+                    from .helper import VALUE
+
+                    class Plugin:
+                        name = "example_package"
+
+                        def __init__(
+                            self,
+                            config,
+                            endpoint,
+                        ):
+                            self.config = config
+                            self.endpoint = endpoint
+                            self.value = VALUE
+                    """
+                )
+            )
+
+            config = SimpleNamespace()
+            endpoint = Endpoint(
+                "example_package",
+                "example.test:443",
+            )
+
+            with patch.dict(
+                os.environ,
+                {
+                    "ENABLED_PLUGINS": (
+                        "example_package"
+                    ),
+                    "DNS_SYNC_PLUGIN_PATH": directory,
+                },
+                clear=True,
+            ):
+                adapter = load_adapter(
+                    "example_package",
+                    config,
+                    endpoint,
+                )
+
+            self.assertEqual(
+                adapter.value,
+                "loaded-from-submodule",
+            )
+
+    def test_missing_enabled_adapter_fails_closed(self):
+        with tempfile.TemporaryDirectory() as directory:
+            config = SimpleNamespace()
+            endpoint = Endpoint(
+                "missing",
+                "example.test:443",
+            )
+
+            with patch.dict(
+                os.environ,
+                {
+                    "ENABLED_PLUGINS": "missing",
+                    "DNS_SYNC_PLUGIN_PATH": (
+                        directory
+                    ),
+                },
+                clear=True,
+            ):
+                with self.assertRaisesRegex(
+                    RuntimeError,
+                    "Unknown adapter plugin",
+                ):
+                    load_adapter(
+                        "missing",
+                        config,
+                        endpoint,
+                    )
+
+
+
+    def test_builtin_adapter_supplies_configured_endpoints(self):
+        endpoint = Endpoint(
+            "npm",
+            "192.168.1.3:81",
+        )
+        config = SimpleNamespace()
+
+        with patch.dict(
+            os.environ,
+            {
+                "ENABLED_PLUGINS": "npm",
+                "NPM_ENDPOINTS": "192.168.1.3",
+            },
+            clear=True,
+        ):
+            endpoints = configured_adapter_endpoints(
+                "npm",
+                config,
+            )
+
+        self.assertEqual(
+            endpoints,
+            (endpoint,),
+        )
+
+
+    def test_external_adapter_cache_respects_plugin_root(self):
+        with tempfile.TemporaryDirectory() as first_directory, \
+             tempfile.TemporaryDirectory() as second_directory:
+
+            first_root = Path(first_directory)
+            second_root = Path(second_directory)
+
+            first_plugin = first_root / "example_cache"
+            second_plugin = second_root / "example_cache"
+
+            first_plugin.mkdir()
+            second_plugin.mkdir()
+
+            first_plugin.joinpath("__init__.py").write_text(
+                textwrap.dedent(
+                    """
+                    from dns_sync.adapters import Endpoint
+
+
+                    def configured_endpoints(config):
+                        return (
+                            Endpoint(
+                                "example_cache",
+                                "first.example:443",
+                            ),
+                        )
+
+
+                    class Plugin:
+                        name = "example_cache"
+
+                        def __init__(
+                            self,
+                            config,
+                            endpoint,
+                        ):
+                            self.endpoint = endpoint
+                    """
+                )
+            )
+
+            second_plugin.joinpath("__init__.py").write_text(
+                textwrap.dedent(
+                    """
+                    from dns_sync.adapters import Endpoint
+
+
+                    def configured_endpoints(config):
+                        return (
+                            Endpoint(
+                                "example_cache",
+                                "second.example:443",
+                            ),
+                        )
+
+
+                    class Plugin:
+                        name = "example_cache"
+
+                        def __init__(
+                            self,
+                            config,
+                            endpoint,
+                        ):
+                            self.endpoint = endpoint
+                    """
+                )
+            )
+
+            with patch.dict(
+                os.environ,
+                {
+                    "ENABLED_PLUGINS": "example_cache",
+                    "DNS_SYNC_PLUGIN_PATH": first_directory,
+                },
+                clear=True,
+            ):
+                first = configured_adapter_endpoints(
+                    "example_cache",
+                    SimpleNamespace(),
+                )
+
+            with patch.dict(
+                os.environ,
+                {
+                    "ENABLED_PLUGINS": "example_cache",
+                    "DNS_SYNC_PLUGIN_PATH": second_directory,
+                },
+                clear=True,
+            ):
+                second = configured_adapter_endpoints(
+                    "example_cache",
+                    SimpleNamespace(),
+                )
+
+        self.assertEqual(
+            tuple(endpoint.identity for endpoint in first),
+            ("example_cache://first.example:443",),
+        )
+        self.assertEqual(
+            tuple(endpoint.identity for endpoint in second),
+            ("example_cache://second.example:443",),
+        )
+
+
+    def test_external_adapter_supplies_configured_endpoints(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            plugin = root / "example"
+            plugin.mkdir()
+
+            (plugin / "__init__.py").write_text(
+                textwrap.dedent(
+                    """
+                    import os
+
+                    from dns_sync.adapters import Endpoint
+
+
+                    def configured_endpoints(config):
+                        address = os.environ["EXAMPLE_ENDPOINT"]
+                        return (
+                            Endpoint("example", address),
+                        )
+
+
+                    class Plugin:
+                        name = "example"
+
+                        def __init__(
+                            self,
+                            config,
+                            endpoint,
+                        ):
+                            self.config = config
+                            self.endpoint = endpoint
+                    """
+                )
+            )
+
+            with patch.dict(
+                os.environ,
+                {
+                    "ENABLED_PLUGINS": "example",
+                    "DNS_SYNC_PLUGIN_PATH": directory,
+                    "EXAMPLE_ENDPOINT": "example.test:9443",
+                },
+                clear=True,
+            ):
+                endpoints = configured_adapter_endpoints(
+                    "example",
+                    SimpleNamespace(),
+                )
+
+            self.assertEqual(
+                tuple(
+                    endpoint.identity
+                    for endpoint in endpoints
+                ),
+                ("example://example.test:9443",),
+            )
+
+
+    def test_adapter_missing_endpoint_hook_fails_closed(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            plugin = root / "example_missing_hook"
+            plugin.mkdir()
+
+            (plugin / "__init__.py").write_text(
+                textwrap.dedent(
+                    """
+                    class Plugin:
+                        name = "example_missing_hook"
+
+                        def __init__(
+                            self,
+                            config,
+                            endpoint,
+                        ):
+                            self.config = config
+                            self.endpoint = endpoint
+                    """
+                )
+            )
+
+            with patch.dict(
+                os.environ,
+                {
+                    "ENABLED_PLUGINS": "example_missing_hook",
+                    "DNS_SYNC_PLUGIN_PATH": directory,
+                },
+                clear=True,
+            ):
+                with self.assertRaisesRegex(
+                    RuntimeError,
+                    "does not export configured_endpoints",
+                ):
+                    configured_adapter_endpoints(
+                        "example_missing_hook",
+                        SimpleNamespace(),
+                    )
+
+
+    def test_adapter_endpoint_hook_cannot_cross_plugin_boundary(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            plugin = root / "example_wrong_endpoint"
+            plugin.mkdir()
+
+            (plugin / "__init__.py").write_text(
+                textwrap.dedent(
+                    """
+                    from dns_sync.adapters import Endpoint
+
+
+                    def configured_endpoints(config):
+                        return (
+                            Endpoint(
+                                "other",
+                                "example.test:443",
+                            ),
+                        )
+
+
+                    class Plugin:
+                        name = "example_wrong_endpoint"
+
+                        def __init__(
+                            self,
+                            config,
+                            endpoint,
+                        ):
+                            self.config = config
+                            self.endpoint = endpoint
+                    """
+                )
+            )
+
+            with patch.dict(
+                os.environ,
+                {
+                    "ENABLED_PLUGINS": "example_wrong_endpoint",
+                    "DNS_SYNC_PLUGIN_PATH": directory,
+                },
+                clear=True,
+            ):
+                with self.assertRaisesRegex(
+                    RuntimeError,
+                    "returned endpoint for plugin",
+                ):
+                    configured_adapter_endpoints(
+                        "example_wrong_endpoint",
+                        SimpleNamespace(),
+                    )
+
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_verizon_adapter.py

@@ -5,14 +5,18 @@ from unittest.mock import Mock, patch
 from dns_sync.adapters import (
     DNS_RECORDS,
     Capability,
+    Endpoint,
     CapabilityOperation,
     NormalizedObject,
     PlanAction,
     PlanOperation,
 )
 from dns_sync.config import AppConfig
-from dns_sync.models import ProviderRecord
-from dns_sync.registry import load_adapter
+from external_plugins.verizon.client import VerizonRecord
+from dns_sync.registry import (
+    configured_adapter_endpoints,
+    load_adapter,
+)
 
 
 class VerizonAdapterTests(unittest.TestCase):
@@ -20,23 +24,52 @@ class VerizonAdapterTests(unittest.TestCase):
         with patch.dict(
             os.environ,
             {
+                "DNS_TARGET_IP": "192.168.1.101",
                 "NPM_ENDPOINTS": "192.168.1.3",
                 "VERIZON_ENDPOINTS": "192.168.1.1",
+                "ENABLED_PLUGINS": "npm,verizon",
+                "DNS_SYNC_PLUGIN_PATH": "external_plugins",
             },
             clear=True,
         ):
             return AppConfig.from_env()
 
-    def adapter(self, provider):
+    def adapter(self, client):
         config = self.config()
-        with patch(
-            "dns_sync.plugins.adapters.verizon.VerizonProvider",
-            return_value=provider,
+
+        import dns_sync.registry as registry
+
+        with patch.dict(
+            os.environ,
+            {
+                "ENABLED_PLUGINS": "npm,verizon",
+                "DNS_SYNC_PLUGIN_PATH": "external_plugins",
+            },
+            clear=False,
         ):
+            module = registry._load_external_adapter_module(
+                "verizon"
+            )
+
+            if module is None:
+                raise AssertionError(
+                    "external Verizon adapter was not found"
+                )
+
+            with patch.object(
+                module,
+                "VerizonClient",
+                return_value=client,
+            ):
+                endpoint = configured_adapter_endpoints(
+                    "verizon",
+                    config,
+                )[0]
+
                 return load_adapter(
                     "verizon",
                     config,
-                config.verizon_endpoints[0],
+                    endpoint,
                 )
 
     @staticmethod
@@ -56,6 +89,32 @@ class VerizonAdapterTests(unittest.TestCase):
             },
         )
 
+    def test_adapter_rejects_endpoint_for_other_plugin(self):
+        config = self.config()
+        endpoint = Endpoint(
+            "npm",
+            "192.168.1.3:81",
+        )
+
+        with patch.dict(
+            os.environ,
+            {
+                "ENABLED_PLUGINS": "npm,verizon",
+                "DNS_SYNC_PLUGIN_PATH": "external_plugins",
+            },
+            clear=False,
+        ):
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "cannot use endpoint",
+            ):
+                load_adapter(
+                    "verizon",
+                    config,
+                    endpoint,
+                )
+
+
     def test_capabilities_cover_dns_discovery_apply_and_verify(self):
         adapter = self.adapter(Mock())
 
@@ -79,22 +138,30 @@ class VerizonAdapterTests(unittest.TestCase):
             ),
         )
 
-    def test_generic_adapter_uses_endpoint_not_router_url(self):
+    def test_client_uses_explicit_endpoint(self):
         with patch.dict(
             os.environ,
             {
+                "DNS_TARGET_IP": "192.168.1.101",
                 "NPM_ENDPOINTS": "192.168.1.3",
                 "VERIZON_ENDPOINTS": "192.168.1.1:8443",
+                "ENABLED_PLUGINS": "npm,verizon",
+                "DNS_SYNC_PLUGIN_PATH": "external_plugins",
                 "ROUTER_URL": "https://192.168.1.254",
                 "ROUTER_PASSWORD": "not-a-real-password",
             },
             clear=True,
         ):
             config = AppConfig.from_env()
+            endpoint = configured_adapter_endpoints(
+                "verizon",
+                config,
+            )[0]
+
             adapter = load_adapter(
                 "verizon",
                 config,
-                config.verizon_endpoints[0],
+                endpoint,
             )
 
         self.assertEqual(
@@ -102,33 +169,33 @@ class VerizonAdapterTests(unittest.TestCase):
             "verizon://192.168.1.1:8443",
         )
         self.assertEqual(
-            adapter.provider.router_url,
+            adapter.client.router_url,
             "https://192.168.1.1:8443",
         )
 
-    def test_connect_delegates_to_provider(self):
-        provider = Mock()
-        adapter = self.adapter(provider)
+    def test_connect_delegates_to_client(self):
+        client = Mock()
+        adapter = self.adapter(client)
 
         adapter.connect()
 
-        provider.connect.assert_called_once_with()
+        client.connect.assert_called_once_with()
 
     def test_discovers_dns_records_with_slot_local_keys(self):
-        provider = Mock()
-        provider.list_records.return_value = {
-            "second.example.com": ProviderRecord(
+        client = Mock()
+        client.list_records.return_value = {
+            "second.example.com": VerizonRecord(
                 hostname="second.example.com",
                 ip="192.168.1.101",
                 metadata={"slot": 42},
             ),
-            "first.example.com": ProviderRecord(
+            "first.example.com": VerizonRecord(
                 hostname="first.example.com",
                 ip="192.168.1.101",
                 metadata={"slot": 41},
             ),
         }
-        adapter = self.adapter(provider)
+        adapter = self.adapter(client)
 
         discovered = adapter.discover(DNS_RECORDS)
 
@@ -146,17 +213,17 @@ class VerizonAdapterTests(unittest.TestCase):
         )
 
     def test_create_and_update_delegate_without_backend_ip(self):
-        provider = Mock()
-        provider.add_record.return_value = "created"
-        provider.update_record.return_value = "updated"
-        provider.list_records.return_value = {
-            "app.example.com": ProviderRecord(
+        client = Mock()
+        client.add_record.return_value = "created"
+        client.update_record.return_value = "updated"
+        client.list_records.return_value = {
+            "app.example.com": VerizonRecord(
                 hostname="app.example.com",
                 ip="192.168.1.101",
                 metadata={"slot": 41},
             ),
         }
-        adapter = self.adapter(provider)
+        adapter = self.adapter(client)
         desired = self.dns_object(ip="192.168.1.101")
 
         create_result = adapter.apply(
@@ -191,17 +258,17 @@ class VerizonAdapterTests(unittest.TestCase):
             "updated",
         )
         self.assertEqual(
-            provider.add_record.call_args.args[0].ip,
+            client.add_record.call_args.args[0].ip,
             "192.168.1.101",
         )
         self.assertEqual(
-            provider.update_record.call_args.args[0].ip,
+            client.update_record.call_args.args[0].ip,
             "192.168.1.101",
         )
 
     def test_noop_does_not_mutate_router(self):
-        provider = Mock()
-        adapter = self.adapter(provider)
+        client = Mock()
+        adapter = self.adapter(client)
 
         result = adapter.apply(
             PlanAction(
@@ -212,29 +279,29 @@ class VerizonAdapterTests(unittest.TestCase):
         )
 
         self.assertIsNone(result)
-        provider.add_record.assert_not_called()
-        provider.update_record.assert_not_called()
+        client.add_record.assert_not_called()
+        client.update_record.assert_not_called()
 
     def test_verify_delegates_expected_dns_record(self):
-        provider = Mock()
-        provider.verify_record.return_value = True
-        adapter = self.adapter(provider)
+        client = Mock()
+        client.verify_record.return_value = True
+        adapter = self.adapter(client)
 
         verified = adapter.verify(self.dns_object())
 
         self.assertTrue(verified)
-        record = provider.verify_record.call_args.args[0]
+        record = client.verify_record.call_args.args[0]
         self.assertEqual(record.hostname, "app.example.com")
         self.assertEqual(record.ip, "192.168.1.101")
 
     def test_unsupported_objects_actions_and_wildcards_fail(self):
-        provider = Mock()
-        adapter = self.adapter(provider)
+        client = Mock()
+        adapter = self.adapter(client)
 
         with self.assertRaisesRegex(RuntimeError, "cannot discover"):
             adapter.discover("certificates")
 
-        with self.assertRaisesRegex(RuntimeError, "different endpoint"):
+        with self.assertRaisesRegex(RuntimeError, "does not belong"):
             adapter.apply(
                 PlanAction(
                     endpoint_id="verizon://different:443",