ipa-ptr-bulk-add.py

#!/usr/bin/env python3
"""
Bulk-add PTR records to FreeIPA DNS, driven by a CSV file.

CSV format (no header):
    fqdn,ip

Example:
    host1.example.com,192.168.1.50
    host2.example.com,10.0.0.75

Usage:
    # Dry-run — print commands without executing
    ./ipa-ptr-bulk-add.py --dry-run hosts.csv

    # Execute (requires valid Kerberos ticket)
    ./ipa-ptr-bulk-add.py hosts.csv

    # Execute + auto-create missing reverse zones
    ./ipa-ptr-bulk-add.py --create-zones hosts.csv

    # Execute without confirmation prompt
    ./ipa-ptr-bulk-add.py --yes hosts.csv
"""

import argparse
import csv
import ipaddress
import subprocess
import sys


def parse_args():
    p = argparse.ArgumentParser(description="Bulk-add FreeIPA PTR records from CSV.")
    p.add_argument("csv_file", help="CSV file with fqdn,ip (no header)")
    p.add_argument("--dry-run", "-n", action="store_true",
                   help="Print commands without executing")
    p.add_argument("--create-zones", "-z", action="store_true",
                   help="Auto-create missing reverse zones (/24 only)")
    p.add_argument("--yes", "-y", action="store_true",
                   help="Skip confirmation prompt")
    return p.parse_args()


def reverse_zone_and_record(ip_str):
    """Derive /24 reverse zone and record name from an IPv4 address."""
    addr = ipaddress.IPv4Address(ip_str)
    octets = str(addr).split(".")
    zone = f"{octets[2]}.{octets[1]}.{octets[0]}.in-addr.arpa"
    record = octets[3]
    return zone, record


def load_csv(path):
    """Parse CSV with columns: fqdn,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,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
            rows.append((fqdn, ip))
    return rows


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


def zone_exists(zone):
    result = subprocess.run(
        ["ipa", "dnszone-show", zone],
        capture_output=True, text=True,
    )
    return result.returncode == 0


def ensure_zone(zone, dry_run=False):
    """Create a /24 reverse zone if it doesn't exist."""
    if dry_run:
        # Can't check whether zone exists without hitting IPA — always show the add
        print(f"# ensure zone exists: ipa dnszone-add '{zone}'")
        return True, None

    if zone_exists(zone):
        return True, None  # already exists

    parts = zone.split(".")
    if len(parts) != 5 or parts[3:] != ["in-addr", "arpa"]:
        return False, f"cannot derive subnet from zone '{zone}'"
    cmd = ["ipa", "dnszone-add", zone]

    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode == 0:
        return True, None
    return False, result.stderr.strip() or result.stdout.strip()


def lookup_existing_ptr(zone, record):
    result = subprocess.run(
        ["ipa", "dnsrecord-find", zone, "--name", record],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        return None
    for line in result.stdout.splitlines():
        if "PTR record:" in line or "PTR Record:" in line.lower():
            val = line.split(":", 1)[1].strip()
            return val if val else None
    return 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)

    mode = "DRY RUN —" if args.dry_run else "Will add"
    print(f"{mode} {len(rows)} PTR record(s)\n")

    # Track zones we've already ensured exist
    known_zones = set()

    for fqdn, ip in rows:
        zone, record = reverse_zone_and_record(ip)
        ptr_value = f"{fqdn}."

        # --- zone creation ---
        if args.create_zones and zone not in known_zones:
            ok, err = ensure_zone(zone, dry_run=args.dry_run)
            if ok:
                known_zones.add(zone)
            else:
                print(f"❌ Failed to create zone '{zone}': {err}", file=sys.stderr)
                continue
        else:
            known_zones.add(zone)  # don't re-check each time

        if args.dry_run:
            if args.create_zones and zone not in known_zones:
                known_zones.add(zone)
                print(f"# ensure zone exists: ipa dnszone-add '{zone}'")
            print(f"# {fqdn} → {ip}")
            print(f"ipa dnsrecord-add '{zone}' '{record}' --ptr-rec='{ptr_value}'")
            print()
        else:
            # Check for existing PTR
            existing = lookup_existing_ptr(zone, record)
            if existing is not None:
                if existing.rstrip(".") + "." == ptr_value:
                    print(f"✅ {fqdn} ({ip}): PTR already set. Skipping.")
                else:
                    print(f"⚠️  {fqdn} ({ip}): PTR conflict — zone has '{existing}', not overwriting. Skipping.")
                continue

            # Add
            result = subprocess.run(
                ["ipa", "dnsrecord-add", zone, record, f"--ptr-rec={ptr_value}"],
                capture_output=True, text=True,
            )
            if result.returncode == 0:
                print(f"✅ {fqdn} → {ptr_value}")
            else:
                err = result.stderr.strip() or result.stdout.strip()
                print(f"❌ {fqdn}: {err}")

    if not args.dry_run:
        print("\nDone.")


if __name__ == "__main__":
    main()