dns-sync/dns_sync/cli.py

import argparse
import os
import sys

from .config import AppConfig
from .executor import execute_endpoint_plans
from .orchestrator import discover_reconcile_plan
from .persistence import SQLiteStateStore
from .registry import (
    configured_adapter_endpoints,
    enabled_plugins,
    load_adapter,
)
from .reporting import print_endpoint_plans, print_plan


def _nonnegative_int(value: str) -> int:
    try:
        parsed = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            "must be a non-negative integer"
        ) from exc
    if parsed < 0:
        raise argparse.ArgumentTypeError(
            "must be a non-negative integer"
        )
    return parsed


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)

    reconcile = sub.add_parser(
        "reconcile",
        help="Run the persistent generic reconciliation engine",
    )
    reconcile.add_argument(
        "--state-db",
        default=os.getenv(
            "DNS_SYNC_STATE_DB",
            "dns-sync-state.sqlite3",
        ),
        help="SQLite state database path",
    )
    reconcile.add_argument(
        "--include-disabled",
        action="store_true",
        help="Include disabled proxy hosts in desired DNS state",
    )
    reconcile.add_argument(
        "--apply",
        action="store_true",
        help="Apply and verify planned changes; default is read-only",
    )
    reconcile.add_argument(
        "--max-changes",
        type=_nonnegative_int,
        default=None,
        help="Maximum planned mutations allowed with --apply",
    )

    return parser


def _configured_adapters(config):
    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 adapters


def run_reconcile(args):
    config = AppConfig.from_env(
        include_disabled=args.include_disabled
    )
    adapters = _configured_adapters(config)

    with SQLiteStateStore(args.state_db) as store:
        result = discover_reconcile_plan(
            adapters,
            store,
            target_ip=config.target_ip,
            include_disabled=config.include_disabled,
            domain_matching=config.domain_matching,
            domain=config.domain,
        )

    for failure in result.failures:
        print(
            "WARNING: discovery failed for "
            f"{failure.endpoint_id} "
            f"{failure.object_type}: {failure.error}",
            file=sys.stderr,
        )

    mode = "APPLY" if args.apply else "READ ONLY"
    print(f"Generic reconciliation mode: {mode}")
    print(f"State database: {args.state_db}")
    print(f"Configured endpoints: {len(adapters)}")
    print(f"Cached/live observations: {len(result.observations)}")
    print(f"Canonical DNS records: {len(result.desired)}")
    print(f"Discovery failures: {len(result.failures)}")
    print()
    print("Per-endpoint plans:")
    print_endpoint_plans(result.plans)

    create_count = sum(
        action.operation.value == "create"
        for plan in result.plans
        for action in plan.actions
    )
    update_count = sum(
        action.operation.value == "update"
        for plan in result.plans
        for action in plan.actions
    )
    change_count = create_count + update_count
    noop_count = sum(
        action.operation.value == "noop"
        for plan in result.plans
        for action in plan.actions
    )

    retirement_count = sum(
        len(plan.retirements)
        for plan in result.plans
    )
    mutation_count = (
        change_count
        + retirement_count
    )

    print()
    print(
        "Plan summary: "
        f"{create_count} create, "
        f"{update_count} update, "
        f"{retirement_count} retirement, "
        f"{noop_count} unchanged"
    )

    if not args.apply:
        print("READ ONLY: no endpoint changes were applied.")
        return 0

    if result.failures:
        raise RuntimeError(
            "Refusing to apply while discovery failures are present"
        )

    if (
        args.max_changes is not None
        and mutation_count > args.max_changes
    ):
        raise RuntimeError(
            f"Refusing to apply {mutation_count} mutation(s); "
            f"--max-changes allows {args.max_changes}"
        )

    if args.max_changes is not None:
        print(f"Apply mutation budget: {args.max_changes}")

    with SQLiteStateStore(args.state_db) as store:
        execution = execute_endpoint_plans(
            adapters,
            result.plans,
            store,
        )

    print(
        "APPLY COMPLETE: "
        f"{len(execution.applied)} action(s) applied and verified; "
        f"{len(execution.retired)} retirement(s) completed; "
        f"{len(execution.skipped)} unchanged/skipped."
    )
    return 0


def main():
    parser = build_parser()
    args = parser.parse_args()
    try:
        if args.command == "reconcile":
            return run_reconcile(args)
        parser.error("unknown command")
    except Exception as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1