From 39924ec4ce2199b33f36170141dcee4513f2172a Mon Sep 17 00:00:00 2001 From: haemka Date: Sun, 23 Aug 2026 13:31:15 +0200 Subject: [PATCH] Added pihole-register.py, updated all scripts for using system trusted CAs --- dhcp-dns-sync.py | 4 +- dynamic-leases.py | 4 +- pihole-register.py | 137 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 141 insertions(+), 4 deletions(-) create mode 100755 pihole-register.py diff --git a/dhcp-dns-sync.py b/dhcp-dns-sync.py index 7da450a..1696b00 100755 --- a/dhcp-dns-sync.py +++ b/dhcp-dns-sync.py @@ -9,8 +9,8 @@ import re import urllib.parse from pathlib import Path -api_url = "https://pi.hole/api" -default_ca_cert = "/etc/pihole/tls_ca.crt" +api_url = "https://pihole.haemka.lan/api" +default_ca_cert = "/etc/ssl/certs/ca-certificates.crt" default_password_file = "/etc/pihole/api_password" default_keep_file = "/etc/pihole/dns_hosts_keep" diff --git a/dynamic-leases.py b/dynamic-leases.py index 7cca758..5e88f3d 100755 --- a/dynamic-leases.py +++ b/dynamic-leases.py @@ -7,8 +7,8 @@ import requests from datetime import datetime from pathlib import Path -api_url = "https://pi.hole/api" -default_ca_cert = "/etc/pihole/tls_ca.crt" +api_url = "https://pihole.haemka.lan/api" +default_ca_cert = "/etc/ssl/certs/ca-certificates.crt" default_password_file = "/etc/pihole/api_password" parser = argparse.ArgumentParser(description="List non-static (dynamic) Pi-hole DHCP leases.") diff --git a/pihole-register.py b/pihole-register.py new file mode 100755 index 0000000..2642c78 --- /dev/null +++ b/pihole-register.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python3 +import argparse +import random +import sys +import urllib.parse +import requests +from pathlib import Path + +api_url = "https://pihole.haemka.lan/api" +default_ca_cert = "/etc/ssl/certs/ca-certificates.crt" +default_password_file = "/etc/pihole/api_password" +mac_oui = "BC:24:11" + +parser = argparse.ArgumentParser( + description="Ensure a static DHCP lease + DNS entry exist for a hostname/IP, " + "generating a MAC if no static lease for that hostname/IP exists yet." +) +parser.add_argument("hostname") +parser.add_argument("ip") +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("-c", "--ca-cert", default=default_ca_cert, + help=f"Path to a CA bundle for verification, defaults to the system trust store " + f"(default: {default_ca_cert}) — override only if pi-hole's cert isn't covered " + f"by a CA the system already trusts") +parser.add_argument("-n", "--dry-run", action="store_true", + help="Show what would change without making any modifying API calls") +parser.add_argument("-d", "--debug", action="store_true", + help="Show detailed processing steps and raw API call results") +args = parser.parse_args() +verify = args.ca_cert + + +def debug(msg): + if args.debug: + print(f"DEBUG: {msg}", file=sys.stderr) + + +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 random_mac(): + return mac_oui + ":" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) + + +auth_payload = {"password": read_password(args.password_file)} + +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() + dhcp_hosts = config_json["config"]["dhcp"]["hosts"] + dns_hosts = config_json["config"]["dns"]["hosts"] + + existing_macs = {h.split(",", 1)[0].lower() for h in dhcp_hosts} + dhcp_by_ip = {} + for h in dhcp_hosts: + entry_mac, entry_ip, entry_name = h.split(",", 2) + dhcp_by_ip.setdefault(entry_ip, []).append((entry_mac, entry_name)) + + mac = None + for entry_mac, entry_name in dhcp_by_ip.get(args.ip, []): + if entry_name == args.hostname: + mac = entry_mac + break + + if mac is None and args.ip in dhcp_by_ip: + conflicting = ", ".join(name for _, name in dhcp_by_ip[args.ip]) + raise SystemExit( + f"IP {args.ip} already has a static DHCP lease for a different hostname " + f"({conflicting}) - refusing to add a second, conflicting lease for '{args.hostname}'. " + f"A DHCP server cannot hand the same static IP to two different MACs; remove the " + f"existing lease first if this is an intentional reassignment." + ) + + dhcp_changed = False + if mac: + debug(f"Existing static lease found for {args.hostname}/{args.ip}: {mac}") + else: + mac = random_mac() + while mac.lower() in existing_macs: + mac = random_mac() + debug(f"No existing static lease for {args.hostname}/{args.ip}, generated {mac}") + + dhcp_entry = f"{mac},{args.ip},{args.hostname}" + dhcp_changed = True + if not args.dry_run: + url = f"{api_url}/config/dhcp/hosts/{urllib.parse.quote(dhcp_entry, safe='')}" + with requests.put(url, headers=headers, verify=verify) as r: + debug(f"PUT dhcp host '{dhcp_entry}': {r.status_code} - {r.reason}") + + dns_by_ip = {} + for entry in dns_hosts: + entry_ip, entry_name = entry.split(" ", 1) + dns_by_ip.setdefault(entry_ip, []).append(entry_name) + + dns_entry = f"{args.ip} {args.hostname}" + if dns_entry not in dns_hosts and args.hostname not in dns_by_ip.get(args.ip, []) and args.ip in dns_by_ip: + conflicting = ", ".join(dns_by_ip[args.ip]) + raise SystemExit( + f"IP {args.ip} already has DNS record(s) for a different hostname " + f"({conflicting}) - refusing to add another DNS entry for '{args.hostname}'." + ) + + dns_changed = dns_entry not in dns_hosts + if dns_changed: + if not args.dry_run: + url = f"{api_url}/config/dns/hosts/{urllib.parse.quote(dns_entry, safe='')}" + with requests.put(url, headers=headers, verify=verify) as r: + debug(f"PUT dns host '{dns_entry}': {r.status_code} - {r.reason}") + else: + debug(f"DNS entry '{dns_entry}' already present") + + if (dhcp_changed or dns_changed) and not args.dry_run: + with requests.post(f"{api_url}/action/restartdns", headers=headers, verify=verify) as r: + debug(f"Restart DNS: {r.status_code} - {r.reason}") + + print(mac) + finally: + requests.delete(f"{api_url}/auth", headers=headers, verify=verify) + debug("Logout completed")