#!/usr/bin/env python3
"""
Push the D-139 ruling-B IPv6 plane carve into the VR1 IPAM apex and REPORT the ULA
rows it retires. Prefixes only: no addresses, no MAAS, and NO deletion ever.
. ~/vr1-office1-creds/vr1-netbox-sandbox.env # the WORKING VR1 apex (DOCFIX-195)
python3 netbox/d139-gua-carve.py --dc vr1-dc0 # DRY RUN (the default)
python3 netbox/d139-gua-carve.py --dc vr1-dc0 --commit # writes, then reads back
Retirement is REPORTED ONLY -- D-139 sequences the ULA/v4 removals LAST, after each
plane is proven, and each is separately gated.
Exit: 0 ok | 1 write/read-back error | 2 REFUSE (could not evaluate).
"""
import argparse
import ipaddress
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
UA = "curl/8.5.0" # the apex 403s other UAs and it reads as an auth failure -- platform-traps.md
SANDBOX_HOSTS = {"localhost", "127.0.0.1", "10.10.1.10"}
# --- D-139 ruling B, table verbatim ---
DC_GUA = {"vr1-dc0": "2602:f3e2:f02::/48", "vr1-dc1": "2602:f3e2:f03::/48"}
# (4th hextet, apex role slug, D-139 plane label, owns a /60 parent)
CARVE = [
(0x10, "provider-public", "provider-public", True),
(0x11, "provider-public", "provider-public VIP", False),
(0x20, "metal-admin", "metal-admin", True),
(0x21, "metal-internal", "metal-internal", False),
(0x30, "data-tenant", "data-tenant", True),
(0x40, "storage", "storage", True),
(0x50, "replication", "replication", True),
(0x80, "lbaas-mgmt", "lb-mgmt", True),
# OOB added 2026-08-01 by operator ruling ("make sure to include oob in the dual stack
# recordings" / "Dual-stack; rule 10.12.60.0/22 back into force for OOB"). It was absent
# because design-decisions.md:3066 declared lbaas-mgmt(:80), vpn(:e0) and oob(:f0) all out
# of scope of the six-plane tool; D-139 brought :80 back and left the other two. VR0 DC0
# and Willamette both carve :f0::/60 + :f0::/64 with role `oob`, so this CONFORMS on the v6
# side. Its IPv4 half (10.12.60.0/22) is NOT this tool's business -- prefixes only, v6 only.
# VPN (:e0) is the SAME omission and is deliberately NOT added: it was never ruled.
(0xf0, "oob", "oob", True),
]
RETIRED_ULA_48 = "fd50:840e:74e2::/48" # retired FOR VR1; still a valid org aggregate
STATUS = "active" # measured on all 44 VR1 plane rows, incl. the ULA rows being replaced
def die(msg):
print("REFUSE: %s" % msg, file=sys.stderr)
sys.exit(2)
class NB:
def __init__(self, url, token):
self.url, self.token = url.rstrip("/"), token
def _req(self, method, path, body=None):
data = json.dumps(body).encode() if body is not None else None
r = urllib.request.Request(
self.url + "/api" + path, data=data, method=method,
headers={"Authorization": "Token " + self.token, "User-Agent": UA,
"Accept": "application/json", "Content-Type": "application/json"})
try:
with urllib.request.urlopen(r, timeout=45) as resp:
return json.loads(resp.read() or "null")
except urllib.error.HTTPError as e:
raise RuntimeError("%s %s -> %d %r" % (method, path, e.code, e.read()[:300]))
except OSError as e: # unreachable/timeout is UNEVALUABLE (rc 2), not a write error (rc 1)
raise RuntimeError("%s %s -> %s" % (method, path, e))
def get_all(self, path):
out, nxt = [], 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 scope_slug(p):
"""The live API returns scope.slug ('vr1-dc0') and scope.name ('VR1 DC0'); repo dumps
normalise name to the slug. Matching name alone found ZERO live (dc-plane-apex-import)."""
sc = p.get("scope") or {}
return sc.get("slug") or sc.get("name") or ""
def sub_at(base48, nn, plen):
"""Direct arithmetic, never list(subnets()) -- the DOCFIX-181 no-hang property."""
if nn % 0x10 and plen == 60:
die("internal: /60 parent at hextet %#x is not /60-aligned" % nn)
return ipaddress.ip_network((int(base48.network_address) + (nn << 64), plen))
def build_plan(dc, site, prefixes):
base48 = ipaddress.ip_network(DC_GUA[dc])
by_cidr = {}
for p in prefixes:
by_cidr.setdefault(str(ipaddress.ip_network(p["prefix"])), []).append(p)
create, exists, conflict = [], [], []
for nn, role, label, parent in CARVE:
for plen in ((60, 64) if parent else (64,)):
cidr = str(sub_at(base48, nn, plen))
rows = by_cidr.get(cidr, [])
mine = [p for p in rows
if scope_slug(p) == dc and ((p.get("role") or {}).get("slug")) == role]
if mine:
exists.append((cidr, role, label))
elif rows:
conflict.append((cidr, role, [(scope_slug(p) or "-",
(p.get("role") or {}).get("slug") or "-")
for p in rows]))
else:
create.append({"prefix": cidr, "status": STATUS, "role": role,
"scope_type": "dcim.site", "scope_id": site["id"],
"description": "%s %s (GUA /%d; D-139)"
% (site["name"], label, plen)})
return create, exists, conflict
def build_retire(dc, prefixes, addrs, ranges):
ula = ipaddress.ip_network(RETIRED_ULA_48)
scoped = [p for p in prefixes if scope_slug(p) == dc]
if not scoped:
die("the apex returned ZERO prefixes scoped to %s. A filtered or failed read looks "
"exactly like a virgin DC -- refusing to call every plane 'missing' on that "
"evidence (MEMORY: instrument currency before negatives)." % dc)
retire = []
for p in scoped:
net = ipaddress.ip_network(p["prefix"])
if net.version == 6 and net.subnet_of(ula):
retire.append(p)
nets = [ipaddress.ip_network(p["prefix"]) for p in retire]
dep_a = sum(1 for a in addrs
if any(ipaddress.ip_interface(a["address"]).ip in n for n in nets))
dep_r = sum(1 for r in ranges
if any(ipaddress.ip_interface(r["start_address"]).ip in n for n in nets))
return sorted(retire, key=lambda p: ipaddress.ip_network(p["prefix"])), dep_a, dep_r, len(scoped)
def main():
ap = argparse.ArgumentParser(description=__doc__.split("\n\n", 1)[0])
ap.add_argument("--dc", required=True, choices=sorted(DC_GUA))
ap.add_argument("--commit", action="store_true",
help="WRITE the CREATE set. Default is a DRY RUN that writes nothing.")
a = ap.parse_args()
url, token = os.environ.get("NETBOX_URL", ""), os.environ.get("NETBOX_TOKEN", "")
if not url or not token:
die("set NETBOX_URL and NETBOX_TOKEN from the env (never argv -- it would land in "
"shell history): . ~/vr1-office1-creds/vr1-netbox-sandbox.env")
host = urllib.parse.urlsplit(url).hostname or ""
if host not in SANDBOX_HOSTS:
die("'%s' is not the VR1 working apex. office1-netbox (10.10.1.10) takes ALL VR1 "
"reads and writes; netbox.baldurkeep.com is the v1 REFERENCE and stays "
"untouched (DOCFIX-195)." % host)
nb = NB(url, token)
try:
sites = nb.get_all("/dcim/sites/")
roles = nb.get_all("/ipam/roles/")
prefixes = nb.get_all("/ipam/prefixes/")
addrs = nb.get_all("/ipam/ip-addresses/")
ranges = nb.get_all("/ipam/ip-ranges/")
except RuntimeError as e:
die("apex read failed, so NOTHING here is evaluable: %s -- a 403 is usually the "
"User-Agent trap (this tool pins %s), not a bad token." % (e, UA))
site = [s for s in sites if s["slug"] == a.dc]
if len(site) != 1:
die("expected exactly 1 apex site with slug %s, found %d" % (a.dc, len(site)))
site = site[0]
role_ids = {r["slug"]: r["id"] for r in roles}
missing = sorted({r for _n, r, _l, _p in CARVE if r not in role_ids})
if missing:
die("apex IPAM role(s) %s do not exist. This tool does not create roles "
"(netbox/roles-aggregates-import.py does); refusing to half-carve a DC." % missing)
create, exists, conflict = build_plan(a.dc, site, prefixes)
retire, dep_a, dep_r, n_scoped = build_retire(a.dc, prefixes, addrs, ranges)
print("== d139-gua-carve %s %s ==" % (a.dc, "--commit" if a.commit else "(DRY RUN)"))
print(" apex: %s site: %s (id=%d)" % (url, site["name"], site["id"]))
print(" apex prefixes scoped to %s: %d (read this run)" % (a.dc, n_scoped))
print(" CREATE %d | EXISTS %d | RETIRE-REPORT %d" % (len(create), len(exists), len(retire)))
if conflict:
for cidr, role, rows in conflict:
print(" [CONFLICT] %s wants role=%s scope=%s but the apex holds %s"
% (cidr, role, a.dc, rows), file=sys.stderr)
die("%d D-139 prefix(es) exist under a DIFFERENT site or role. That is an "
"unrecognised state, not a no-op -- resolve it by hand." % len(conflict))
for title, rows in (
("CREATE", [(c["prefix"], c["role"], c["description"]) for c in create]),
("EXISTS (idempotent -- left untouched)", exists),
("RETIRE -- REPORTED ONLY, this tool never deletes",
[(p["prefix"], (p.get("role") or {}).get("slug"), p.get("description", ""))
for p in retire])):
print("\n %s:" % title)
for cidr, role, note in rows:
print(" %-24s role=%-16s %s" % (cidr, role, note))
print(" dependent objects inside those prefixes: %d ip-address(es), %d ip-range(s)"
% (dep_a, dep_r))
if not a.commit:
print("\nDRY RUN -- nothing was written. Re-run with --commit to apply.")
return 0
errs = 0
for c in create:
body = dict(c, role=role_ids[c["role"]])
try:
nb.post("/ipam/prefixes/", body)
except RuntimeError as e:
print(" [ERR] %s: %s" % (c["prefix"], e), file=sys.stderr)
errs += 1
# READ BACK: a POST that reports success but did not land is a class this repo has
# been bitten by (opnsense-plugins.sh apply always silently dry-ran).
now = {str(ipaddress.ip_network(p["prefix"])) for p in nb.get_all("/ipam/prefixes/")
if scope_slug(p) == a.dc}
gone = [c["prefix"] for c in create if c["prefix"] not in now]
for g in gone:
print(" [MISSING AFTER WRITE] %s" % g, file=sys.stderr)
print("\n READ-BACK: %d/%d present" % (len(create) - len(gone), len(create)))
errs += len(gone)
print("RESULT: errors=%d" % errs)
return 1 if errs else 0
if __name__ == "__main__":
sys.exit(main())
