#!/usr/bin/env python3 """ Populate the NetBox APEX with the per-DC D-134 BANDS and the ruled VIP ADDRESSES. WHY THIS EXISTS. The apex holds the entire IP PREFIX layer and is drift-free there, but NOTHING below it: measured 2026-07-27, `ipam/ip-ranges` carried 3 objects (all Office1's D-120 bands) and `ipam/ip-addresses` carried 4 (netbox, tailscale, two rack controllers -- none a VIP). The D-134 bands and the 33->39 VIPs lived only in prose and hand-maintained overlays. D-136 option (D) (ADOPTED 2026-07-27) rules the apex be populated FIRST and the renderer generated from it; this is that population. Pattern and write-guard mirror netbox/d120-compose-bands.py, which already loaded the D-120 bands into this apex -- proven, not invented. DERIVED, NOT HARDCODED (hard rule 2). The only table below is the ruled app->octet map: * plane CIDRs <- scripts/lib-net.sh via lib_net_select_dc (site-keyed) * v6 plane /64s <- the apex's own prefix layer (read back before writing) * VIP legs <- the three plane bases + the app octet. Verified against the measured live values: dc0 keystone .50 is 10.12.4.50 / 10.12.8.50 / 10.12.12.50 = provider-public / metal-admin / metal-internal. NOT taken from VIP_PREFIX_*, which lib-net deliberately UNSETS for dc1 (R9 rules that group generated). * v6 host part <- mirrors the v4 octet TEXTUALLY (RULED 2026-07-27). Note the provider v6 leg lives in the DEDICATED GUA VIP /64 (f0X:11::), not the node /64 (f0X:10::) -- that asymmetry is why the host-part convention needed a ruling at all. DRY BY DEFAULT. --commit writes, then READS BACK every object. Exit: 0 ok | 1 write/verify error | 2 could not evaluate. """ import argparse import json import os import subprocess import sys import urllib.error import urllib.parse import urllib.request UA = "curl/8.5.0" # upstream 403s the default python UA -- platform-traps.md SANDBOX_HOSTS = {"localhost", "127.0.0.1", "10.10.1.10"} # ---- the ruled app -> VIP octet map ------------------------------------------- # Octets .50-.60 are the measured as-built set (bundle.yaml + overlays/vr1-dc1-vips.yaml, # frozen at tests/render-baseline/). vault .61 and designate .62 are RULED BUT NOT BUILT # (R11, a D-020 AMENDMENT 2026-07-27) -- recording them in the apex is exactly what an # IPAM apex is for: reserving a planned address so nothing else takes it. APP_OCTET = [ ("keystone", 50), ("barbican", 51), ("cinder", 52), ("glance", 53), ("magnum", 54), ("neutron-api", 55), ("nova-cloud-controller", 56), ("octavia", 57), ("openstack-dashboard", 58), ("placement", 59), ("ceph-radosgw", 60), ("vault", 61), # R11: ruled, not yet built ("designate", 62), # R11: ruled, not yet built -- new to D-020's enumeration ] BAND_UTIL = (4, 49) BAND_VIP = (50, 99) def die(msg, rc=2): print(f"FAIL: {msg}", file=sys.stderr) sys.exit(rc) class NB: def __init__(self, url, token): self.url = url.rstrip("/") self.token = token def _req(self, method, path, body=None): data = json.dumps(body).encode() if body is not None else None r = urllib.request.Request( f"{self.url}/api{path}", data=data, method=method, headers={"Authorization": f"Token {self.token}", "User-Agent": UA, "Accept": "application/json", "Content-Type": "application/json"}) try: with urllib.request.urlopen(r, timeout=30) as resp: return json.loads(resp.read() or "null") except urllib.error.HTTPError as e: raise RuntimeError(f"{method} {path} -> {e.code} {e.read()[:400]!r}") def get_all(self, path): out, nxt = [], f"{path}?limit=500" while nxt: d = self._req("GET", nxt) out += d.get("results", []) n = d.get("next") nxt = n.split("/api", 1)[1] if n else None return out def post(self, path, body): return self._req("POST", path, body) def lib_net(site): """Plane CIDRs + names from lib-net.sh -- the site-keyed authority.""" repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sh = (f'source "{repo}/scripts/lib-net.sh" >/dev/null 2>&1 || exit 3; ' f'lib_net_select_dc "{site}" >/dev/null 2>&1 || exit 3; ' 'for c in "${PLANE_CIDRS[@]}"; do printf "%s\\t%s\\n" "${PLANE_NAME[$c]}" "$c"; done') p = subprocess.run(["bash", "-c", sh], capture_output=True, text=True) if p.returncode != 0 or not p.stdout.strip(): die(f"lib-net.sh did not yield planes for {site} (rc={p.returncode})") return dict(l.split("\t") for l in p.stdout.strip().splitlines()) def main(): ap = argparse.ArgumentParser() ap.add_argument("--dc", required=True, choices=["vr1-dc0", "vr1-dc1"]) ap.add_argument("--commit", action="store_true") a = ap.parse_args() url = os.environ.get("NETBOX_URL", "") token = os.environ.get("NETBOX_TOKEN", "") if not url or not token: die("set NETBOX_URL and NETBOX_TOKEN (token from the env, never argv -- it would " "land in shell history and process listings)") host = urllib.parse.urlsplit(url).hostname or "" if host not in SANDBOX_HOSTS: die(f"refusing to write to '{host}': not a known apex/sandbox host. During VR1, " f"office1-netbox (10.10.1.10) takes ALL writes and netbox.baldurkeep.com stays " f"READ-ONLY (DOCFIX-195).") planes = lib_net(a.dc) nb = NB(url, token) prefixes = nb.get_all("/ipam/prefixes/") by_cidr = {p["prefix"]: p for p in prefixes} v6 = {} for p in prefixes: pre = p["prefix"] if ":" not in pre or not pre.endswith("/64"): continue sc = (p.get("scope") or {}) # MEASURED 2026-07-27: the LIVE API returns scope.name as the DISPLAY name # ("VR1 DC0") and scope.slug as the site key ("vr1-dc0"), while the repo's # dump files normalise `name` to the slug. Matching only `name` -- as a first # version of this did, having been written against the dump -- silently found # ZERO prefixes live and refused. Match the SLUG first, name as fallback, so # the tool works against both shapes. This is the same class as the audit's # own "used a two-day-old dump instead of polling live" finding. if a.dc not in ((sc.get("slug") or ""), (sc.get("name") or "")): continue role = (p.get("role") or {}).get("slug") kind = "vip" if "VIP" in (p.get("description") or "") else "plane" v6[(role, kind)] = pre if not v6: die(f"the apex carries no /64 prefixes scoped to {a.dc} -- populate the prefix " f"layer first (netbox/dc-dc-prefixes-import.py)") existing_ranges = {(r["start_address"].split("/")[0], r["end_address"].split("/")[0]) for r in nb.get_all("/ipam/ip-ranges/")} existing_addrs = {x["address"].split("/")[0] for x in nb.get_all("/ipam/ip-addresses/")} planned_r, planned_a, skipped = [], [], 0 # ---- D-134 bands, per plane, v4 ------------------------------------------- for name, cidr in sorted(planes.items()): net, mask = cidr.split("/") b = ".".join(net.split(".")[:3]) for label, (lo, hi) in (("utility", BAND_UTIL), ("VIP", BAND_VIP)): s, e = f"{b}.{lo}", f"{b}.{hi}" if (s, e) in existing_ranges: skipped += 1 continue planned_r.append({ "start_address": f"{s}/{mask}", "end_address": f"{e}/{mask}", "status": "reserved", "description": f"D-134 {label} band ({a.dc} {name}) -- .{lo}-.{hi}", }) # ---- VIP addresses: 13 apps x 3 legs x 2 families ------------------------- leg_planes = ["provider-public", "metal-admin", "metal-internal"] missing = [l for l in leg_planes if l not in planes] if missing: die(f"{a.dc} is missing plane(s) {missing} in lib-net -- refusing to guess a VIP leg") for app, oct_ in APP_OCTET: for leg in leg_planes: v4net, v4mask = planes[leg].split("/") b = ".".join(v4net.split(".")[:3]) addr = f"{b}.{oct_}" if addr not in existing_addrs: planned_a.append({ "address": f"{addr}/{v4mask}", "status": "reserved", "description": f"VIP {app} {leg} ({a.dc}) -- D-020 octet .{oct_}", }) else: skipped += 1 # v6 twin. The PROVIDER leg uses the DEDICATED GUA VIP /64, not the node /64. key = ("provider-public", "vip") if leg == "provider-public" else (leg, "plane") pre = v6.get(key) if not pre: die(f"no apex v6 prefix for {key} in {a.dc} -- refusing to invent one") base = pre.split("::/")[0] v6addr = f"{base}::{oct_}" # TEXTUAL octet mirror, RULED 2026-07-27 if v6addr not in existing_addrs: planned_a.append({ "address": f"{v6addr}/64", "status": "reserved", "description": f"VIP {app} {leg} v6 ({a.dc}) -- octet mirror ::{oct_}", }) else: skipped += 1 print(f"== dc-plane-apex-import {a.dc} " f"{'--commit' if a.commit else '(DRY RUN)'} ==") print(f" apex: {url}") print(f" planned ip-ranges : {len(planned_r)}") print(f" planned ip-addresses: {len(planned_a)} " f"({len(APP_OCTET)} apps x 3 legs x 2 families = {len(APP_OCTET)*6})") print(f" already present : {skipped}") for r in planned_r[:4]: print(f" [range] {r['start_address']} - {r['end_address']} {r['description']}") if len(planned_r) > 4: print(f" ... and {len(planned_r)-4} more range(s)") for x in planned_a[:6]: print(f" [addr] {x['address']:44} {x['description']}") if len(planned_a) > 6: print(f" ... and {len(planned_a)-6} more address(es)") if not a.commit: print("\nDRY RUN -- nothing was written. Re-run with --commit to apply.") return 0 errs = 0 for r in planned_r: try: nb.post("/ipam/ip-ranges/", r) except RuntimeError as e: print(f" [ERR] range {r['start_address']}: {e}", file=sys.stderr); errs += 1 for x in planned_a: try: nb.post("/ipam/ip-addresses/", x) except RuntimeError as e: print(f" [ERR] addr {x['address']}: {e}", file=sys.stderr); errs += 1 # READ BACK -- a POST that reports success but did not land is the class this repo # has been bitten by (opnsense-plugins.sh apply always silently dry-ran). now_r = {(r["start_address"].split("/")[0], r["end_address"].split("/")[0]) for r in nb.get_all("/ipam/ip-ranges/")} now_a = {x["address"].split("/")[0] for x in nb.get_all("/ipam/ip-addresses/")} miss_r = [r for r in planned_r if (r["start_address"].split("/")[0], r["end_address"].split("/")[0]) not in now_r] miss_a = [x for x in planned_a if x["address"].split("/")[0] not in now_a] print(f"\n READ-BACK: ranges {len(planned_r)-len(miss_r)}/{len(planned_r)}, " f"addresses {len(planned_a)-len(miss_a)}/{len(planned_a)}") for m in (miss_r + miss_a)[:5]: print(f" [MISSING AFTER WRITE] {m}", file=sys.stderr) errs += len(miss_r) + len(miss_a) print(f"\nRESULT: errors={errs}") return 1 if errs else 0 if __name__ == "__main__": sys.exit(main())