pihole-scripts/dhcp-dns-sync.py

#!/usr/bin/env python3
import argparse
import csv
import ipaddress
import json
import sys
import requests
import re
import urllib.parse
from pathlib import Path

api_url = "https://pi.hole/api"
verify = "/etc/pihole/tls_ca.crt"
default_password_file = "/etc/pihole/api_password"
default_keep_file = "/etc/pihole/dns_hosts_keep"

parser = argparse.ArgumentParser(description="Sync static Pi-hole DHCP leases to local DNS entries.")
parser.add_argument("-p", "--password-file", default=default_password_file,
                     help=f"Path to file containing the Pi-hole API password (default: {default_password_file})")
parser.add_argument("-k", "--keep-file", default=default_keep_file,
                     help=f"Path to file listing full 'ip name' DNS entries to keep regardless of DHCP state (default: {default_keep_file})")
parser.add_argument("-n", "--dry-run", action="store_true",
                     help="Show what would change without making any API calls that modify state")
parser.add_argument("-s", "--terse", action="store_true",
                     help="Only output rows that represent an actual change (new, ip changed, removed) - omit unchanged/kept rows")
parser.add_argument("-d", "--debug", action="store_true",
                     help="Show detailed processing steps and raw API call results")
parser.add_argument("-t", "--table", action="store_true",
                     help="Print a human-readable aligned table instead of the default CSV")
parser.add_argument("-j", "--json", action="store_true",
                     help="Print output as JSON instead of the default CSV")
args = parser.parse_args()

if args.table and args.json:
	parser.error("--table and --json are mutually exclusive")


def read_password(path):
	try:
		return Path(path).read_text().strip()
	except OSError as e:
		raise SystemExit(f"Could not read password file '{path}': {e}")


def read_keep_list(path):
	p = Path(path)
	if not p.exists():
		return set()
	entries = set()
	for line in p.read_text().splitlines():
		line = line.strip()
		if line and not line.startswith("#"):
			entries.add(line)
	return entries


def ip_sort_key(ip_str):
	try:
		return (0, ipaddress.ip_address(ip_str))
	except ValueError:
		return (1, ip_str)


def debug(msg):
	if args.debug:
		print(f"DEBUG: {msg}", file=sys.stderr)


auth_payload = {"password": read_password(args.password_file)}
keep_entries = read_keep_list(args.keep_file)
hosts_changed = False

debug(f"Password file: {args.password_file}")
debug(f"Keep file: {args.keep_file} ({len(keep_entries)} entries loaded)")

with requests.post(f"{api_url}/auth", json=auth_payload, verify=verify) as auth:
	debug(f"POST /auth -> {auth.status_code}")
	auth_json = auth.json()
	session = auth_json.get('session')
	if not session or not session.get('sid'):
		raise SystemExit(f"Authentication failed: {auth_json}")

	headers = {
		"X-FTL-SID": session.get('sid'),
		"X-FTL-CSRF": session.get('sid')
	}

	try:
		with requests.get(f"{api_url}/config", headers=headers, verify=verify) as config:
			debug(f"GET /config -> {config.status_code}")
			config_json = config.json()

			raw_dhcp_hosts = config_json.get('config').get('dhcp').get('hosts')
			host_mappings = []
			for dhcp_host_mapping in raw_dhcp_hosts:
				converted = re.sub("([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2},(?P<ip>.*),(?P<name>.*)", r"\g<ip> \g<name>", dhcp_host_mapping)
				debug(f"DHCP raw '{dhcp_host_mapping}' -> '{converted}'")
				host_mappings.append(converted)
			dhcp_names = {hm.split(" ", 1)[1] for hm in host_mappings}

			dns_host_mappings = config_json.get('config').get('dns').get('hosts')
			dns_by_name = {}
			for entry in dns_host_mappings:
				parts = entry.split(" ", 1)
				if len(parts) == 2:
					ip, name = parts
					dns_by_name[name] = (entry, ip)
				else:
					debug(f"Skipping malformed DNS entry: '{entry}'")

			# records: dicts with ip, hostname, status, old_ip (only set for "ip changed")
			records = []
			actions = []  # (ip, fn) executed in IP order, fn performs the real API calls

			for host_mapping in host_mappings:
				ip, name = host_mapping.split(" ", 1)

				if host_mapping in dns_host_mappings:
					records.append({"ip": ip, "hostname": name, "status": "unchanged", "old_ip": ""})
					continue

				if name in dns_by_name:
					old_entry, old_ip = dns_by_name[name]
					records.append({"ip": ip, "hostname": name, "status": "ip changed", "old_ip": old_ip})
				else:
					old_entry = None
					records.append({"ip": ip, "hostname": name, "status": "new", "old_ip": ""})

				def do_add_or_change(host_mapping=host_mapping, ip=ip, name=name, old_entry_ref=dns_by_name.get(name)):
					global hosts_changed
					hosts_changed = True
					if old_entry_ref:
						old_entry, old_ip = old_entry_ref
						debug(f"IP change for '{name}': {old_ip} -> {ip}")
						if not args.dry_run:
							delete_url = f"{api_url}/config/dns/hosts/{urllib.parse.quote(old_entry)}"
							with requests.delete(delete_url, headers=headers, verify=verify) as del_result:
								debug(f"DELETE '{old_entry}': {del_result.status_code} - {del_result.reason}")
					if not args.dry_run:
						target_url = f"{api_url}/config/dns/hosts/{urllib.parse.quote(host_mapping)}"
						with requests.put(target_url, headers=headers, verify=verify) as result:
							debug(f"PUT '{host_mapping}': {result.status_code} - {result.reason}")

				actions.append((ip, do_add_or_change))

			for name, (entry, ip) in dns_by_name.items():
				if name in dhcp_names:
					continue

				if entry in keep_entries:
					records.append({"ip": ip, "hostname": name, "status": "kept", "old_ip": ""})
					continue

				records.append({"ip": ip, "hostname": name, "status": "removed", "old_ip": ""})

				def do_remove(entry=entry, ip=ip):
					global hosts_changed
					hosts_changed = True
					debug(f"Removing '{entry}' (no longer in DHCP)")
					if not args.dry_run:
						delete_url = f"{api_url}/config/dns/hosts/{urllib.parse.quote(entry)}"
						with requests.delete(delete_url, headers=headers, verify=verify) as del_result:
							debug(f"DELETE '{entry}': {del_result.status_code} - {del_result.reason}")

				actions.append((ip, do_remove))

			actions.sort(key=lambda item: ip_sort_key(item[0]))
			debug(f"Executing {len(actions)} action(s) in IP order")
			for _, action_fn in actions:
				action_fn()

			if hosts_changed and not args.dry_run:
				with requests.post(f"{api_url}/action/restartdns", headers=headers, verify=verify) as result:
					debug(f"Restart DNS: {result.status_code} - {result.reason}")
			elif hosts_changed:
				debug("Would restart DNS")

			# --- Output ---
			records.sort(key=lambda r: ip_sort_key(r["ip"]))
			if args.terse:
				records = [r for r in records if r["status"] not in ("unchanged", "kept")]

			if args.table:
				if not records:
					print("No changes." if args.terse else "No entries.")
				else:
					ip_w = max(len(r["ip"]) for r in records + [{"ip": "IP"}])
					name_w = max(len(r["hostname"]) for r in records + [{"hostname": "Hostname"}])
					status_w = max(len(r["status"]) for r in records + [{"status": "Status"}])
					print(f"{'IP':<{ip_w}}  {'Hostname':<{name_w}}  {'Status':<{status_w}}  Old IP")
					print(f"{'-' * ip_w}  {'-' * name_w}  {'-' * status_w}  {'-' * 6}")
					for r in records:
						print(f"{r['ip']:<{ip_w}}  {r['hostname']:<{name_w}}  {r['status']:<{status_w}}  {r['old_ip']}")
			elif args.json:
				print(json.dumps(records, indent=2))
			else:
				writer = csv.DictWriter(sys.stdout, fieldnames=["ip", "hostname", "status", "old_ip"])
				writer.writeheader()
				writer.writerows(records)
	finally:
		requests.delete(f"{api_url}/auth", headers=headers, verify=verify)
		debug("Logout completed")