ipa-ptr-bulk-del.py

#!/usr/bin/env python3
"""
Bulk-delete PTR records from 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-del.py --dry-run hosts.csv

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

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

import argparse
import csv
import ipaddress
import subprocess
import sys


def parse_args():
    p = argparse.ArgumentParser(description="Bulk-delete 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("--yes", "-y", action="store_true",
                   help="Skip confirmation prompt")
    return p.parse_args()


def reverse_zone_and_record(ip_str):
    """Derive reverse zone and record name from an IPv4 address.

    Assumes /24 reverse zones (same as wipe-ptr.sh):
        192.168.1.50  →  zone='1.168.192.in-addr.arpa', record='50'

    Returns (zone, record) tuple.
    """
    addr = ipaddress.IPv4Address(ip_str)
    octets = str(addr).split(".")
    # /24: zone = <o3>.<o2>.<o1>.in-addr.arpa, record = <o4>
    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):
            # Skip blank lines
            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():
    """Return True if a valid Kerberos ticket exists."""
    result = subprocess.run(["klist", "-s"], capture_output=True)
    return result.returncode == 0


def lookup_existing_ptr(zone, record):
    """Look up the current PTR value for a given reverse zone + record.

    Returns the PTR string (e.g. 'host1.example.com.') or None if not found.
    """
    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)

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

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

        if args.dry_run:
            # Just print the command
            cmd = f"ipa dnsrecord-del '{zone}' '{record}' --ptr-rec='{ptr_value}'"
            print(f"# {fqdn} → {ip}")
            print(cmd)
            print()
        else:
            # Check what's actually there
            existing = lookup_existing_ptr(zone, record)
            if existing is None:
                print(f"ℹ️  {fqdn} ({ip}): no existing PTR in {zone}/{record}. Nothing to delete.")
            elif existing.rstrip(".") + "." != ptr_value:
                print(f"⚠️  {fqdn} ({ip}): PTR mismatch — zone has '{existing}', CSV says '{ptr_value}'. Skipping for safety.")
            else:
                if not args.yes:
                    print(f"🗑  {fqdn} ({ip}) → {ptr_value}")
                result = subprocess.run(
                    ["ipa", "dnsrecord-del", zone, record, f"--ptr-rec={ptr_value}"],
                    capture_output=True, text=True,
                )
                if result.returncode == 0:
                    print(f"   ✅ Deleted.")
                else:
                    err = result.stderr.strip() or result.stdout.strip()
                    print(f"   ❌ Failed: {err}")

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


if __name__ == "__main__":
    main()