Remove legacy source provider sync path
This commit is contained in:
@@ -3,13 +3,6 @@ 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
|
||||
@@ -17,8 +10,6 @@ from .registry import (
|
||||
configured_adapter_endpoints,
|
||||
enabled_plugins,
|
||||
load_adapter,
|
||||
load_provider,
|
||||
load_source,
|
||||
)
|
||||
from .reporting import print_endpoint_plans, print_plan
|
||||
|
||||
@@ -41,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",
|
||||
@@ -74,77 +61,9 @@ 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):
|
||||
adapters = []
|
||||
|
||||
@@ -275,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)
|
||||
|
||||
@@ -97,8 +97,6 @@ def env_bool(name: str, default: bool) -> bool:
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AppConfig:
|
||||
source_name: str
|
||||
provider_name: str
|
||||
npm_endpoints: tuple[Endpoint, ...]
|
||||
proxy_ip: str
|
||||
proxy_api_port: int
|
||||
@@ -118,9 +116,6 @@ class AppConfig:
|
||||
|
||||
@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()
|
||||
|
||||
try:
|
||||
port = int(os.getenv("PROXY_API_PORT", "81"))
|
||||
except ValueError as exc:
|
||||
@@ -185,8 +180,6 @@ class AppConfig:
|
||||
)
|
||||
|
||||
return cls(
|
||||
source_name=source_name,
|
||||
provider_name=provider_name,
|
||||
npm_endpoints=npm_endpoints,
|
||||
proxy_ip=proxy_ip,
|
||||
proxy_api_port=port,
|
||||
|
||||
@@ -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
|
||||
@@ -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]: ...
|
||||
|
||||
@@ -28,11 +28,3 @@ class ProviderCapabilities:
|
||||
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
|
||||
|
||||
@@ -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])?$")
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -258,14 +258,6 @@ def _load(kind: str, name: str, config, *args):
|
||||
)
|
||||
|
||||
|
||||
def load_source(name: str, config):
|
||||
return _load("sources", name, config)
|
||||
|
||||
|
||||
def load_provider(name: str, config):
|
||||
return _load("providers", name, config)
|
||||
|
||||
|
||||
def load_adapter(name: str, config, endpoint):
|
||||
return _load(
|
||||
"adapters",
|
||||
|
||||
Reference in New Issue
Block a user