138 lines
5.6 KiB
Python
Executable File
138 lines
5.6 KiB
Python
Executable File
#!/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")
|