#!/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.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__":
import urllib.parse # noqa: E402 (kept local; parse used only in main)
sys.exit(main())
