Custom DDNS for Your Home Lab with OPNsense BIND (Python + cron)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import requests, ipaddress, sys

# ===== Settings =====
HOST        = "https://opnsense.maksonlee.com:8444"  # OPNsense base URL
API_KEY     = "OPNSENSE_API_KEY_HERE"
API_SECRET  = "OPNSENSE_API_SECRET_HERE"
ZONE        = "maksonlee.com"                        # primary zone name
NAME        = "home"                                 # "@", "home", or "home.maksonlee.com"
TTL         = 300                                    # seconds
VERIFY_TLS  = True                                    # set False if self-signed
# ====================

def get_public_ipv4():
    for u in ("https://api.ipify.org","https://ipv4.icanhazip.com","https://ifconfig.me/ip"):
        try:
            ip = requests.get(u, timeout=6).text.strip()
            if ip and ipaddress.ip_address(ip).version == 4:
                return ip
        except Exception:
            pass
    sys.exit("Failed to determine public IPv4.")

def api_req(method, path, json=None):
    url = f"{HOST}{path}"
    r = requests.request(method, url, json=json, auth=(API_KEY, API_SECRET), timeout=15, verify=VERIFY_TLS)
    if r.status_code == 401:
        raise PermissionError(f"401 Unauthorized for {method} {path}")
    r.raise_for_status()
    return r.json()

def rows_of(d):
    return d.get("rows") or d.get("result") or d.get("data") or (d if isinstance(d, list) else [])

def to_label(name, zone):
    n = name.strip().rstrip(".").lower()
    z = zone.strip().rstrip(".").lower()
    if n in ("", "@", z): return "@"
    if n.endswith("." + z): return n[:-(len(z)+1)]
    if "." not in n: return n
    sys.exit(f"NAME '{name}' is not in zone '{zone}'")

def fetch_primary_zone_rows():
    for path in (
        "/api/bind/domain/search_primary_domain",
        "/api/bind/domain/search_master_domain",
    ):
        r = rows_of(api_req("GET", path))
        if r: return r
    return []

def find_zone_uuid(zone):
    z = zone.strip().rstrip(".").lower()
    # 1) Normal path: query domain list
    try:
        rows = fetch_primary_zone_rows()
        if not rows:
            print("No zones returned by search_*_domain (check privileges / click Apply in UI).")
        else:
            for row in rows:
                if str(row.get("domainname","")).strip().rstrip(".").lower() == z and row.get("uuid"):
                    return row["uuid"]
    except PermissionError as e:
        print(str(e))
    # 2) Fallback: derive from existing records if the zone already contains entries
    try:
        rec_rows = rows_of(api_req("GET", "/api/bind/record/search_record"))
        for r in rec_rows:
            if str(r.get("domainname","")).strip().rstrip(".").lower() == z and r.get("domain"):
                return r["domain"]
    except PermissionError as e:
        print(str(e))
    sys.exit(f"Unable to determine zone UUID for '{zone}'. Ensure the user has Services: BIND or the zone already has records.")

def find_a_records(domain_uuid, label):
    rows = rows_of(api_req("GET", "/api/bind/record/search_record"))
    out = []
    for r in rows:
        if (str(r.get("type","")).upper() == "A"
            and str(r.get("name","")).strip().rstrip(".").lower() == label
            and str(r.get("domain") or "").strip().lower() == str(domain_uuid).lower()):
            out.append(r)
    return out

def get_record(uuid):
    d = api_req("GET", f"/api/bind/record/get_record/{uuid}")
    return d.get("record") or d

def add_a_record(domain_uuid, label, ip, ttl):
    rec = {
        "type": "A",
        "name": label,
        "ttl": str(ttl),
        "enabled": "1",
        "value": ip,           # content
        "domain": domain_uuid  # zone UUID
    }
    return api_req("POST", "/api/bind/record/add_record", {"record": rec})

def set_a_record(uuid, domain_uuid, label, ip, ttl):
    rec = {
        "type": "A",
        "name": label,
        "ttl": str(ttl),
        "enabled": "1",
        "value": ip,
        "domain": domain_uuid
    }
    return api_req("POST", f"/api/bind/record/set_record/{uuid}", {"record": rec})

def reconfigure():
    return api_req("POST", "/api/bind/service/reconfigure")

def main():
    ip = get_public_ipv4()
    print("IPv4:", ip)
    label = to_label(NAME, ZONE)

    zone_uuid = find_zone_uuid(ZONE)
    print("Zone UUID:", zone_uuid)

    recs = find_a_records(zone_uuid, label)
    changed = False

    if not recs:
        resp = add_a_record(zone_uuid, label, ip, TTL)
        if resp.get("validations"):
            print("Add failed:", resp["validations"]); sys.exit(1)
        print("Add response:", resp)
        changed = True
    else:
        for r in recs:
            rid = r.get("uuid")
            cur = get_record(rid)
            cur_ip  = str(cur.get("value") or "")
            cur_ttl = str(cur.get("ttl") or "")
            cur_en  = str(cur.get("enabled") or "1") in ("1","true","True")
            if cur_ip == ip and cur_ttl == str(TTL) and cur_en:
                print(f"No change (uuid={rid})")
                continue
            resp = set_a_record(rid, zone_uuid, label, ip, TTL)
            if resp.get("validations"):
                print("Update failed:", resp["validations"]); sys.exit(1)
            print("Update response:", resp)
            changed = True

    if changed:
        reconfigure()
        # Confirm
        recs = find_a_records(zone_uuid, label)
        print("Post-reconfigure records found:", recs)
        if not recs:
            sys.exit("Record not visible after reconfigure. Check privileges or UI Save/Apply.")
        print("Applied changes.")
    else:
        print("No updates were necessary.")

if __name__ == "__main__":
    main()

crontab -e
# add:
*/5 * * * * /home/administrator/scripts/bind_ddns.py >/dev/null 2>&1