ipa-a-record-update.py

#!/usr/bin/env python3
"""
Bulk-update A records in FreeIPA DNS, driven by a CSV file.

CSV format (no header):
    fqdn,new_ip

Example:
    host1.example.com,10.200.1.50
    host2.example.com,10.200.0.75

Usage:
    # Dry-run — print what would change without executing
    ./ipa-a-record-update.py --dry-run hosts.csv

    # Execute (requires valid Kerberos ticket)
    ./ipa-a-record-update.py hosts.csv

    # Execute without confirmation prompt
    ./ipa-a-record-update.py --yes hosts.csv
"""

import argparse
import csv
import subprocess
import sys


def parse_args():
    p = argparse.ArgumentParser(description="Bulk-update FreeIPA A records from CSV.")
    p.add_argument("csv_file", help="CSV file with fqdn,new_ip (no header)")
    p.add_argument("--dry-run", "-n", action="store_true",
                   help="Print commands without executing")
    p.add_argument("--yes", "-y", action="store_true",
                   help="Skip confirmation prompt")
    return p.parse_args()


def split_fqdn(fqdn):
    """Split FQDN into (hostname, zone). E.g. 'host1.example.com' → ('host1', 'example.com')."""
    parts = fqdn.split(".", 1)
    if len(parts) != 2:
        return None, None
    return parts[0], parts[1]


def load_csv(path):
    """Parse CSV with columns: fqdn,new_ip (no header)."""
    rows = []
    with open(path, newline="") as f:
        reader = csv.reader(f)
        for lineno, row in enumerate(reader, start=1):
            if not row or all(c.strip() == "" for c in row):
                continue
            if len(row) < 2:
                print(f"⚠️  Line {lineno}: expected 2 columns (fqdn,new_ip), got {len(row)}. Skipping.",
                      file=sys.stderr)
                continue
            fqdn, ip = row[0].strip(), row[1].strip()
            if not fqdn or not ip:
                print(f"⚠️  Line {lineno}: empty fqdn or ip. Skipping.", file=sys.stderr)
                continue
            hostname, zone = split_fqdn(fqdn)
            if not hostname or not zone:
                print(f"⚠️  Line {lineno}: could not parse FQDN '{fqdn}'. Skipping.", file=sys.stderr)
                continue
            rows.append((fqdn, hostname, zone, ip))
    return rows


def check_kinit():
    result = subprocess.run(["klist", "-s"], capture_output=True)
    return result.returncode == 0


def lookup_existing_a(zone, hostname):
    """Return current A record IP(s) for a hostname, or None if not found."""
    result = subprocess.run(
        ["ipa", "dnsrecord-show", zone, hostname],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        return None
    ips = []
    for line in result.stdout.splitlines():
        if "A record:" in line or "A Record:" in line.lower():
            val = line.split(":", 1)[1].strip()
            if val:
                ips.append(val)
    return ips if ips else None


def main():
    args = parse_args()

    rows = load_csv(args.csv_file)
    if not rows:
        print("No valid rows found in CSV.", file=sys.stderr)
        sys.exit(1)

    if not args.dry_run:
        if not check_kinit():
            print("❌ No valid Kerberos ticket. Run 'kinit admin' first.", file=sys.stderr)
            sys.exit(1)

    # Show what we're about to do
    print(f"{'DRY RUN — would update' if args.dry_run else 'Will update'} {len(rows)} A record(s):\n")
    for fqdn, hostname, zone, new_ip in rows:
        existing = lookup_existing_a(zone, hostname)
        current = ", ".join(existing) if existing else "NOT FOUND"
        print(f"  {fqdn}: {current} → {new_ip}")

    if not args.dry_run and not args.yes:
        print()
        try:
            resp = input("Proceed? [y/N] ").strip().lower()
        except EOFError:
            resp = "n"
        if resp not in ("y", "yes"):
            print("Aborted.")
            sys.exit(0)

    print()

    ok = fail = skip = 0
    for fqdn, hostname, zone, new_ip in rows:
        if args.dry_run:
            print(f"# {fqdn}: A → {new_ip}")
            print(f"ipa dnsrecord-mod '{zone}' '{hostname}' --a-rec='{new_ip}'")
            print()
            ok += 1
            continue

        existing = lookup_existing_a(zone, hostname)
        if existing is None:
            print(f"⚠️  {fqdn}: no existing A record in zone '{zone}'. Skipping — use dnsrecord-add instead.")
            skip += 1
            continue

        if new_ip in existing:
            print(f"✅ {fqdn}: already points to {new_ip}. Skipping.")
            skip += 1
            continue

        print(f"  {fqdn}: {', '.join(existing)} → {new_ip}", end=" ")

        result = subprocess.run(
            ["ipa", "dnsrecord-mod", zone, hostname, f"--a-rec={new_ip}"],
            capture_output=True, text=True,
        )
        if result.returncode == 0:
            print("✅")
            ok += 1
        else:
            err = result.stderr.strip() or result.stdout.strip()
            print(f"❌ {err}")
            fail += 1

    if not args.dry_run:
        print(f"\nDone.  ✅ {ok} updated  ⚠️  {skip} skipped  ❌ {fail} failed")


if __name__ == "__main__":
    main()