#!/usr/bin/env python3 """Create or update a local DNS A record on every Waters Church UniFi console. Usage: docs/examples/waters-church-na/unifi-dns-record.py repo.watersmedia.org 10.3.8.90 docs/examples/waters-church-na/unifi-dns-record.py repo.watersmedia.org 10.3.8.90 --sites apollo-beach,norwood docs/examples/waters-church-na/unifi-dns-record.py repo.watersmedia.org 10.3.8.90 --dry-run docs/examples/waters-church-na/unifi-dns-record.py --list # show every DNS policy per site Each console needs its own Network API key (Site Manager keys from api.ui.com are rejected locally). Create one per console in the UniFi UI: Settings -> Control Plane -> Integrations -> Create API Key and save it, one line, mode 600, as: ~/.config/unifi/-network-api-key where is one of the names in SITES below. The script talks to the console's Network Integration API (/proxy/network/integration/v1/sites/{siteId}/dns/policies) over the LAN / site-to-site links, so it must run from inside the church network. Gateways use self-signed certificates, so certificate verification is disabled for them. Keys are read from disk and sent only in the X-API-KEY header; they are never printed. """ from __future__ import annotations import argparse import json import os import shutil import ssl import subprocess import sys import urllib.error import urllib.parse import urllib.request # Console name -> gateway LAN address (UDM Pro at each site, 2026-10). SITES = { "apollo-beach": "10.3.0.1", "north-attleboro": "10.1.0.1", "woonsocket": "10.2.0.1", "norwood": "10.4.0.1", "southcoast": "10.5.0.1", } KEY_DIR = os.path.expanduser("~/.config/unifi") TIMEOUT = 15 class ApiError(RuntimeError): pass class Console: def __init__(self, name: str, gateway: str): self.name = name self.gateway = gateway self.base = f"https://{gateway}/proxy/network/integration/v1" self.key_path = os.path.join(KEY_DIR, f"{name}-network-api-key") self.key = self._load_key() self.ctx = ssl.create_default_context() self.ctx.check_hostname = False self.ctx.verify_mode = ssl.CERT_NONE def _load_key(self) -> str | None: try: with open(self.key_path) as fh: key = fh.read().strip() except FileNotFoundError: return None return key or None def request(self, method: str, path: str, body: dict | None = None, query: dict | None = None): url = self.base + path if query: url += "?" + urllib.parse.urlencode(query) data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("X-API-KEY", self.key or "") req.add_header("Accept", "application/json") if data is not None: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=TIMEOUT, context=self.ctx) as resp: raw = resp.read() return json.loads(raw) if raw else None except urllib.error.HTTPError as e: detail = e.read().decode(errors="replace")[:300] raise ApiError(f"{method} {path} -> HTTP {e.code}: {detail}") from None except (urllib.error.URLError, TimeoutError, OSError) as e: raise ApiError(f"{method} {path} -> {e}") from None def site_id(self) -> str: sites = self.request("GET", "/sites").get("data", []) if not sites: raise ApiError("console reports no sites") for s in sites: if s.get("internalReference") == "default" or s.get("name") == "default": return s["id"] return sites[0]["id"] def dns_policies(self, site_id: str) -> list[dict]: out: list[dict] = [] offset, limit = 0, 100 while True: page = self.request("GET", f"/sites/{site_id}/dns/policies", query={"offset": offset, "limit": limit}) data = page.get("data", []) out.extend(data) total = page.get("totalCount") offset += len(data) if not data or (total is not None and offset >= total) or len(data) < limit: return out def ensure_a_record(self, site_id: str, fqdn: str, ipv4: str, ttl: int | None, dry_run: bool) -> str: existing = [p for p in self.dns_policies(site_id) if p.get("domain", "").lower() == fqdn.lower()] a_records = [p for p in existing if p.get("type") == "A_RECORD"] other = [p for p in existing if p.get("type") != "A_RECORD"] if other: kinds = ", ".join(sorted({p.get("type", "?") for p in other})) return f"SKIPPED: {fqdn} already has a non-A policy here ({kinds}); fix by hand" if len(a_records) > 1: return f"SKIPPED: {len(a_records)} A records already exist for {fqdn}; fix by hand" body = {"type": "A_RECORD", "enabled": True, "domain": fqdn, "ipv4Address": ipv4} if ttl: body["ttlSeconds"] = ttl if a_records: cur = a_records[0] same = cur.get("ipv4Address") == ipv4 and cur.get("enabled", True) and (ttl is None or cur.get("ttlSeconds") == ttl) if same: return f"unchanged: {fqdn} -> {ipv4} already present (id {cur.get('id')})" if dry_run: return f"would UPDATE {fqdn}: {cur.get('ipv4Address')} -> {ipv4} (id {cur.get('id')})" self.request("PUT", f"/sites/{site_id}/dns/policies/{cur['id']}", body=body) return f"UPDATED {fqdn}: {cur.get('ipv4Address')} -> {ipv4}" if dry_run: return f"would CREATE {fqdn} -> {ipv4}" created = self.request("POST", f"/sites/{site_id}/dns/policies", body=body) return f"CREATED {fqdn} -> {ipv4} (id {created.get('id') if isinstance(created, dict) else '?'})" def resolve_via(gateway: str, fqdn: str) -> str: """Ask the gateway's resolver directly; returns the answer or a short status.""" dig = shutil.which("dig") if not dig: return "(dig not installed)" try: res = subprocess.run([dig, "+short", "+time=3", "+tries=1", f"@{gateway}", fqdn, "A"], capture_output=True, text=True, timeout=10) except subprocess.TimeoutExpired: return "(timeout)" answer = res.stdout.strip().replace("\n", " ") return answer or "(no answer)" def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("fqdn", nargs="?", help="host name, e.g. repo.watersmedia.org") ap.add_argument("ipv4", nargs="?", help="IPv4 address the name should resolve to") ap.add_argument("--sites", default=",".join(SITES), help="comma-separated subset of: " + ", ".join(SITES)) ap.add_argument("--ttl", type=int, default=300, help="TTL in seconds (default 300; the API rejects a null TTL)") ap.add_argument("--dry-run", action="store_true", help="show what would change without writing") ap.add_argument("--list", action="store_true", help="print every DNS policy on each console and exit") ap.add_argument("--no-verify", action="store_true", help="skip the dig check after writing") args = ap.parse_args() if not args.list and (not args.fqdn or not args.ipv4): ap.error("fqdn and ipv4 are required unless --list is given") names = [s.strip() for s in args.sites.split(",") if s.strip()] unknown = [n for n in names if n not in SITES] if unknown: ap.error(f"unknown site(s): {', '.join(unknown)}") failures = 0 for name in names: console = Console(name, SITES[name]) label = f"{name:16s} {console.gateway:10s}" if not console.key: print(f"{label} NO KEY: save the console's API key to {console.key_path}") failures += 1 continue try: site_id = console.site_id() if args.list: policies = console.dns_policies(site_id) print(f"{label} {len(policies)} DNS policies") for p in sorted(policies, key=lambda p: p.get("domain", "")): target = p.get("ipv4Address") or p.get("ipv6Address") or p.get("targetDomain") or p.get("text") or "" flag = "" if p.get("enabled", True) else " (disabled)" print(f" {p.get('type', '?'):12s} {p.get('domain', ''):40s} -> {target}{flag}") continue result = console.ensure_a_record(site_id, args.fqdn, args.ipv4, args.ttl, args.dry_run) line = f"{label} {result}" if not args.dry_run and not args.no_verify and not result.startswith("SKIPPED"): line += f" | dig @{console.gateway}: {resolve_via(console.gateway, args.fqdn)}" print(line) if result.startswith("SKIPPED"): failures += 1 except ApiError as e: print(f"{label} FAILED: {e}") failures += 1 return 1 if failures else 0 if __name__ == "__main__": sys.exit(main())