#!/usr/bin/env python3
"""
NetBox prerequisite import for the VR1 DC-DC buildout: six-plane IPAM roles + aggregates.
Closes gaps G1 (5 of 6 six-plane roles missing) and G2 (0 aggregates) identified in
docs/dc-dc-netbox-buildout-scope.md, so that netbox/dc-dc-prefixes-import.py -- which
ASSUMES the roles already exist and does not create aggregates -- can run cleanly.
Creates, idempotently (get-or-create; existing objects detected and skipped):
IPAM roles (slugs MUST match scripts/lib-net.sh SPACES6 exactly; the prefixes
import references them by slug):
provider-public, metal-admin, metal-internal, data-tenant, replication
(`storage` already exists in NetBox and is left untouched.)
RIRs (only if missing): "RFC 4193 ULA", "RFC 1918" (both is_private). "ARIN" is
assumed to exist (it did on 2026-07-10); looked up, not created.
Aggregates (under the right RIR):
2602:f3e2::/36 ARIN (the HELD allocation = region-0 of the planned /32; D-101)
23.157.124.0/24 ARIN
10.0.0.0/8 RFC 1918
172.16.0.0/12 RFC 1918
<ORG_ULA_48> RFC 4193 (ONLY if ORG_ULA_48 is set -- an operator literal, D-101)
NOTHING INVENTED. The six-plane role names are the ADOPTED D-052/D-053 model
(scripts/lib-net.sh). The ARIN /36 and /24 are the real, quoted-in-D-101 allocations.
The RFC1918 blocks are RFC facts. The org ULA /48 is the ONE operator-choice literal
here (D-101 names it NetBox-assigned, not hardcoded): it is read from ORG_ULA_48 and
its format validated; if unset, the ULA aggregate is SKIPPED (with a note), never
guessed. The scope doc recommends fd50:840e:74e2::/48 (CSPRNG, collision-checked vs the
Tailscale fd7a:115c:a1e0::/48 already in NetBox) -- pass it in once ratified.
Note on the /36 vs /32 aggregate: only the /36 is currently allocated by ARIN, so the
aggregate is the /36 (honest for an IPAM apex). D-101 frames it as region-0 of the /32;
once the /36 -> /32 expand completes, update this aggregate to /32.
DELIBERATE default: this script PREVIEWS by default and requires --commit to write
(safer for global roles/aggregates; matches the project's dry-run-first discipline).
This differs from netbox/dc-dc-prefixes-import.py, which commits by default with
--verify-only to preview -- the difference is intentional and documented here.
Usage:
# preview (no writes) -- ULA aggregate skipped unless ORG_ULA_48 is set:
NETBOX_URL=https://netbox.baldurkeep.com NETBOX_TOKEN=<token> \\
python3 roles-aggregates-import.py
# preview including the ULA aggregate:
NETBOX_URL=... NETBOX_TOKEN=... ORG_ULA_48=fd50:840e:74e2::/48 \\
python3 roles-aggregates-import.py
# write:
NETBOX_URL=... NETBOX_TOKEN=... ORG_ULA_48=fd50:840e:74e2::/48 \\
python3 roles-aggregates-import.py --commit
Idempotent: re-running is safe. NetBox 4.x (tested against 4.5.8).
"""
from __future__ import annotations
import argparse
import ipaddress
import os
import sys
try:
import pynetbox
except ImportError:
sys.stderr.write("ERROR: pynetbox not installed. pip install pynetbox\n")
sys.exit(1)
def die(msg: str, code: int = 1) -> None:
sys.stderr.write(f"ERROR: {msg}\n")
sys.exit(code)
def get_nb():
url = os.environ.get("NETBOX_URL")
token = os.environ.get("NETBOX_TOKEN")
if not url:
die("NETBOX_URL environment variable not set")
if not token:
die("NETBOX_TOKEN environment variable not set")
nb = pynetbox.api(url, token=token)
try:
_ = nb.status()
except Exception as exc: # noqa: BLE001
die(f"Could not reach NetBox at {url}: {exc}")
return nb
# --- Data (facts / adopted model; the only operator literal is ORG_ULA_48 via env) ---
ROLES = [
("provider-public", "Provider Public", 100,
"Six-plane: provider public API VIPs + FIP/ext_net (GUA). D-052/D-101."),
("metal-admin", "Metal Admin", 110,
"Six-plane: metal admin API, dual-stack v4+ULA (2026-07-09 amendment). D-052/D-101."),
("metal-internal", "Metal Internal", 120,
"Six-plane: metal internal API endpoints, dual-stack v4+ULA. D-052/D-101."),
("data-tenant", "Data Tenant", 130,
"Six-plane: geneve overlay, ULA-only; tenant GUA delegated separately. D-052/D-101."),
# storage (weight 140) already exists -- intentionally omitted.
("replication", "Replication", 150,
"Six-plane: Ceph cluster incl. cross-DC leg, ULA-only. D-052/D-101/D-108."),
]
RIRS = [
# name, slug, is_private
("RFC 4193 ULA", "rfc-4193-ula", True),
("RFC 1918", "rfc-1918", True),
]
# prefix, rir-name, description. ULA aggregate handled separately (needs ORG_ULA_48).
AGGREGATES = [
("2602:f3e2::/36", "ARIN",
"ARIN IPv6 allocation: the HELD /36 = region-0 of the planned /32 (D-101). "
"Update to /32 after the expand completes."),
("23.157.124.0/24", "ARIN", "ARIN IPv4 allocation."),
("10.0.0.0/8", "RFC 1918", "RFC1918 private IPv4."),
("172.16.0.0/12", "RFC 1918", "RFC1918 private IPv4 (OOB space)."),
]
def validate_ula_48(cidr: str) -> str:
try:
net = ipaddress.ip_network(cidr, strict=True)
except ValueError as exc:
die(f"ORG_ULA_48='{cidr}' is not a valid CIDR: {exc}")
if net.version != 6:
die(f"ORG_ULA_48='{cidr}' must be IPv6")
if net.prefixlen != 48:
die(f"ORG_ULA_48='{cidr}' must be a /48 (got /{net.prefixlen})")
if not str(net.network_address).startswith(("fd", "fc")):
die(f"ORG_ULA_48='{cidr}' is not in the ULA range (fc00::/7)")
return str(net)
def main() -> int:
ap = argparse.ArgumentParser(description="Create VR1 six-plane roles + aggregates (idempotent).")
ap.add_argument("--commit", action="store_true",
help="write to NetBox (default: preview only, no writes)")
args = ap.parse_args()
commit = args.commit
mode = "COMMIT" if commit else "PREVIEW (no writes -- add --commit to apply)"
nb = get_nb()
print(f"NetBox: {os.environ['NETBOX_URL']} | mode: {mode}\n")
created = existed = 0
def act_create(kind: str, label: str, create_fn):
nonlocal created, existed
if commit:
obj = create_fn()
print(f" CREATED {kind:10s} {label}" + (f" (id={obj.id})" if obj else ""))
created += 1
else:
print(f" WOULD {kind:10s} {label}")
created += 1
# --- Roles ---
print("== IPAM roles (six-plane; slugs match lib-net.sh SPACES6) ==")
for slug, name, weight, desc in ROLES:
if nb.ipam.roles.get(slug=slug):
print(f" EXISTS role {slug}")
existed += 1
continue
act_create("role", slug,
lambda s=slug, n=name, w=weight, d=desc:
nb.ipam.roles.create(name=n, slug=s, weight=w, description=d))
# storage sanity note
if nb.ipam.roles.get(slug="storage"):
print(" EXISTS role storage (pre-existing; left untouched)")
else:
print(" WARN role storage MISSING unexpectedly -- verify SPACES6 vs NetBox")
# --- RIRs ---
print("\n== RIRs ==")
rir_ids: dict[str, object] = {}
arin = nb.ipam.rirs.get(slug="arin") or nb.ipam.rirs.get(name="ARIN")
if not arin:
die("RIR 'ARIN' not found (expected pre-existing). Create it first, then re-run.")
rir_ids["ARIN"] = arin.id
print(f" EXISTS rir ARIN (id={arin.id})")
for name, slug, is_private in RIRS:
existing = nb.ipam.rirs.get(slug=slug) or nb.ipam.rirs.get(name=name)
if existing:
rir_ids[name] = existing.id
print(f" EXISTS rir {name}")
existed += 1
continue
rir_ids[name] = None # not yet created (preview) -- resolved on commit below
act_create("rir", name,
lambda n=name, s=slug, p=is_private:
nb.ipam.rirs.create(name=n, slug=s, is_private=p))
if commit:
rir_ids[name] = (nb.ipam.rirs.get(slug=slug)).id
# --- Aggregates ---
print("\n== Aggregates ==")
agg_list = list(AGGREGATES)
ula = os.environ.get("ORG_ULA_48")
if ula:
ula = validate_ula_48(ula)
agg_list.append((ula, "RFC 4193 ULA", "Org ULA /48 (internal-only; D-101 literal)."))
else:
print(" SKIP aggregate <ORG_ULA_48 unset> -- ULA /48 aggregate skipped "
"(set ORG_ULA_48 to include it; nothing guessed).")
for prefix, rir_name, desc in agg_list:
if nb.ipam.aggregates.get(prefix=prefix):
print(f" EXISTS aggregate {prefix} ({rir_name})")
existed += 1
continue
rid = rir_ids.get(rir_name)
act_create("aggregate", f"{prefix} ({rir_name})",
lambda p=prefix, r=rid, d=desc:
nb.ipam.aggregates.create(prefix=p, rir=r, description=d))
print(f"\nSummary: {created} {'created' if commit else 'to create'}, {existed} already present.")
if not commit:
print("Preview only -- re-run with --commit to apply.")
return 0
if __name__ == "__main__":
sys.exit(main())
