#!/usr/bin/env python3
"""
Register the D-124 Office1-region <-> DC-rack MANAGEMENT TRANSIT addressing in
office1-netbox (the VR1 IPAM apex): the point-to-point transit on the office1<->dc0
mesh leg, plus the rack's metal-admin static IP. Two objects, nothing else.
D-124 (ADOPTED 2026-07-16, Scheme A -- transit-numbered mesh) rules that the
region<->rack MAAS control path rides a small point-to-point transit on the
office1<->dc0 mesh leg (NOT metal-admin, which D-100 keeps DC-local and node-facing).
The rack (vvr1-dc0) STRADDLES both legs: the transit (region-facing) and metal-admin
(node-facing, 10.12.8.0/22). So this tool registers exactly:
1. the transit prefix role=transit, scope=dcim.site:vr1-dc0 (the /30 or /31)
2. the rack ip-address <rack-ip>/22 in metal-admin, D-120 static band .2-.49
Both LITERALS ARE OPERATOR INPUTS (--transit-cidr / --rack-ip): D-124, like D-115/
D-117, rules the SCHEME and leaves the actual CIDR + IP to office1-netbox (the apex;
operator-held token, this loop cannot query it live). NO literal is invented in-repo.
This tool creates NEITHER the transit ROLE nor any container/site: all are
PRECONDITIONS (a missing one is a hard STOP, never a silent create) -- exactly as
dc-edge-wan-import.py treats the `edge` role + the 172.30.0.0/16 container.
DRY BY DEFAULT -- nothing is written without --commit.
WRITING UPSTREAM IS GATED IN CODE. SANDBOX_HOSTS is the allowlist; a --commit at any
other host REFUSES without --yes-write-upstream. WHOLE-PLAN PREFLIGHT: the transit
role, the container the transit nests under, the transit CIDR shape+placement, the
rack IP's band (metal-admin, static .2-.49, NOT the .1 gateway), and the target site
are ALL checked BEFORE any create -- so a bad plan (e.g. a rack IP that is the .1
gateway) cannot leave the transit prefix written and the apex half-populated (the
failure that bit roles-aggregates-import.py: 4 objects created, then it died on the 5th).
SCOPING / ROLE choices flagged for the operator (see the D-124 return notes):
* ROLE_SLUG = "transit": D-124 calls for the transit's OWN role, "mirroring the
D-115 Edge role" (a DEDICATED role). An `infra` role exists in the apex whose
description reads "ptp transit, device loopbacks, RR loopbacks, anycast" -- it is
NOT reused here because it is explicitly SHARED infrastructure, not the dedicated
per-purpose role D-124 asks for. Operator to seed/confirm the `transit` slug in
office1-netbox (as d115-office-carve.py seeded `edge`). A missing role dies here.
* CONTAINER = "172.31.0.0/24" (D-124 dedicated transit supernet, operator-pinned
2026-07-16): the operator ruled a DEDICATED management/transit supernet (NOT under
Cloud 10.12.0.0/16), mirroring how the D-115 Edge role got its own 172.30.0.0/16.
Sized /24 (holds 64 /30s -- ample for all DC<->region + DC<->DC ptp links). The
transit CIDR must be subnet_of this. This container is a PRECONDITION: the operator
carves 172.31.0.0/24 (a container prefix with the `transit` role) in office1-netbox
BEFORE this runs; a missing/collision container makes the tool DIE (fail-loud is the
backstop for the candidate block being picked against a stale draft). This gates
ACCEPTANCE only -- NetBox auto-nests by CIDR, we set no parent.
* transit prefix SCOPE = dcim.site:vr1-dc0 (RULED 2026-07-16: "mirror the edge" --
site-scoped like the D-115 DC-edge /24s, 172.30.2.0/24 -> vr1-dc0).
* The rack IP band (metal-admin 10.12.8.0/22, static .2-.49, gw .1) is enforced
ARITHMETICALLY (D-120/D-124), NOT via a prefix-object lookup: metal-admin is not
registered as a /22 prefix in the apex draft, and the band is a convention, not a
container. This asymmetry with the transit check is intentional.
Usage (on office1-netbox / through a tunnel, with the sandbox token). D-124 values
operator-ratified 2026-07-16 (transit supernet 172.31.0.0/24, transit /30
172.31.0.0/30, rack IP 10.12.8.2) -- confirm free against the LIVE apex first:
NETBOX_URL=http://10.10.1.10:8000 NETBOX_TOKEN=<tok> \
python3 netbox/dc-rack-mgmt-import.py --transit-cidr 172.31.0.0/30 --rack-ip 10.12.8.2
... same, add --commit, to write.
"""
import argparse
import ipaddress
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
UA = "curl/8.5.0" # upstream 403s the default python UA -- see platform-traps.md
# A sandbox is local, or the known Office1 sandbox address. Anything else is treated
# as the production apex and requires the explicit upstream flag.
SANDBOX_HOSTS = {"localhost", "127.0.0.1", "10.10.1.10"}
ROLE_SLUG = "transit" # the transit's OWN role (D-124); a precondition here
CONTAINER = "172.31.0.0/24" # D-124 dedicated transit supernet (operator-pinned 2026-07-16); transit /30 must be subnet_of this (precondition)
SITE_SLUG = "vr1-dc0" # the transit prefix is site-scoped here (mirrors the D-115 edge /24s)
STATUS = "active" # a real segment on the wire
# metal-admin band arithmetic (D-052/D-124). The rack's metal-admin IP lands in the
# D-120 static band .2-.49 of this /22, and NOT the .1 gateway.
METAL_ADMIN = "10.12.8.0/22"
STATIC_BAND_LOW = 2 # .2 (first static site-services address, D-120)
STATIC_BAND_HIGH = 49 # .49 (last static site-services address, D-120)
RACK_DNS = "vvr1-dc0" # the rack-controller VM (D-124 cloudinit-vm on the two legs)
TRANSIT_DESC = "office1<->dc0 management transit -- region<->rack MAAS control path (D-124 Scheme A)"
RACK_DESC = "vr1-dc0 MAAS rack controller (vvr1-dc0) metal-admin static IP (D-124; D-120 static band)"
def die(msg: str):
print(f"FAIL: {msg}", file=sys.stderr)
sys.exit(2)
class NB:
"""Stdlib NetBox client -- same shape as the other sandbox-loop tools; UA-aware so
it is not 403'd by the upstream User-Agent filter."""
def __init__(self, base, token):
self.base = base.rstrip("/")
self.token = token
def _req(self, method, path, body=None):
data = json.dumps(body).encode() if body is not None else None
req = urllib.request.Request(f"{self.base}/api/{path}", data=data, method=method,
headers={"Authorization": f"Token {self.token}",
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": UA})
try:
with urllib.request.urlopen(req, timeout=45) as r:
return json.load(r) if r.status != 204 else None
except urllib.error.HTTPError as exc:
detail = exc.read().decode(errors="replace")[:300]
if exc.code == 403 and "v1 token" in detail:
die("403 'Invalid v1 token' -- NetBox 4.6 wants the ASSEMBLED v2 token "
"nbt_<key>.<plaintext>, not the API's bare `token` field.")
if exc.code == 403:
die(f"403 on {path}. If curl works with this token, it is the upstream "
f"User-Agent filter, NOT the token (references/platform-traps.md).")
die(f"HTTP {exc.code} {method} {path}: {detail}")
def one(self, path, **flt):
res = self._req("GET", f"{path}/?{urllib.parse.urlencode(flt)}&limit=1")
return res["results"][0] if res["results"] else None
def create(self, path, payload):
return self._req("POST", f"{path}/", payload)
def get_nb(base, token):
"""Client factory -- the injection seam the harness overrides to drive main()
against an in-memory fake without a live NetBox."""
return NB(base, token)
def parse_transit_cidr(raw: str) -> ipaddress.IPv4Network:
"""The transit CIDR must be a valid /30 or /31 network (host bits clear), subnet_of
CONTAINER. Shape is validated here; containment against CONTAINER in preflight."""
try:
net = ipaddress.ip_network(raw, strict=True)
except ValueError as exc:
die(f"--transit-cidr {raw!r} is not a valid network (host bits set?): {exc}")
if net.version != 4:
die(f"--transit-cidr {raw} must be IPv4 (the office1<->dc0 transit is v4).")
if net.prefixlen not in (30, 31):
die(f"--transit-cidr {raw} must be a /30 or /31 point-to-point (got /{net.prefixlen}).")
return net
def parse_rack_ip(raw: str) -> ipaddress.IPv4Address:
"""Accept a bare host (10.12.8.X) or a masked form; if masked, the mask MUST be the
metal-admin /22. Returns the host address; band placement is checked in preflight."""
try:
if "/" in raw:
iface = ipaddress.ip_interface(raw)
if iface.network.prefixlen != 22:
die(f"--rack-ip {raw} carries /{iface.network.prefixlen}; metal-admin is a /22. "
f"Pass a bare host (10.12.8.X) or the /22 form.")
host = iface.ip
else:
host = ipaddress.ip_address(raw)
except ValueError as exc:
die(f"--rack-ip {raw!r} is not a valid IPv4 address: {exc}")
if host.version != 4:
die(f"--rack-ip {raw} must be IPv4 (metal-admin is v4).")
return host
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.split("\n\n", 1)[0])
ap.add_argument("--transit-cidr", default=os.environ.get("TRANSIT_CIDR"),
help="REQUIRED. The /30 or /31 point-to-point for the office1<->dc0 "
"transit leg (operator-supplied, NetBox-assigned; env: TRANSIT_CIDR).")
ap.add_argument("--rack-ip", default=os.environ.get("RACK_IP"),
help="REQUIRED. The rack's metal-admin static IP within 10.12.8.0/22, "
"static band .2-.49 (operator-supplied; env: RACK_IP).")
ap.add_argument("--commit", action="store_true",
help="WRITE. Default is a DRY RUN that writes nothing.")
ap.add_argument("--yes-write-upstream", action="store_true",
help="Required (with --commit) to write to a NON-sandbox NetBox.")
args = ap.parse_args()
url = os.environ.get("NETBOX_URL")
token = os.environ.get("NETBOX_TOKEN")
if not url or not token:
die("NETBOX_URL and NETBOX_TOKEN must be set.")
# Inputs are args-or-env, so they cannot be argparse required=True (that would break
# the env fallback). Hand-roll the missing checks, each with its own die.
if not args.transit_cidr:
die("--transit-cidr (or TRANSIT_CIDR) is REQUIRED -- the NetBox-assigned transit "
"/30 or /31. No literal is invented in-repo (D-124).")
if not args.rack_ip:
die("--rack-ip (or RACK_IP) is REQUIRED -- the rack's NetBox-assigned metal-admin "
"static IP. No literal is invented in-repo (D-124).")
host = (urllib.parse.urlparse(url).hostname or url).lower()
is_sandbox = host in SANDBOX_HOSTS
print(f"Target : {url} ({'SANDBOX' if is_sandbox else 'NOT a known sandbox'})")
if args.commit and not is_sandbox and not args.yes_write_upstream:
die(f"REFUSING to --commit to '{host}': not a known sandbox, so treated as the "
f"PRODUCTION apex. Re-run with --yes-write-upstream if that is intended.")
print("\n*** DRY RUN -- nothing will be written. Re-run with --commit. ***"
if not args.commit else "\n*** COMMITTING. ***")
# ---- WHOLE-PLAN PREFLIGHT: validate the ENTIRE plan (both objects) BEFORE any
# create, so a bad rack IP (e.g. the .1 gateway) cannot leave the transit written. ----
# (a) Transit CIDR shape + containment (local, no NetBox needed).
transit = parse_transit_cidr(args.transit_cidr)
container = ipaddress.ip_network(CONTAINER)
if not transit.subnet_of(container):
die(f"transit {transit} is outside the container {CONTAINER} (Cloud) -- refusing to "
f"place it. If the transit is carved elsewhere, that is an operator/D-124 call.")
# (b) Rack IP band: within metal-admin /22, in the D-120 static band .2-.49, NOT the
# .1 gateway (a distinct die so the ".1 rejected" property is unambiguous).
rack = parse_rack_ip(args.rack_ip)
metal = ipaddress.ip_network(METAL_ADMIN)
if rack not in metal:
die(f"rack IP {rack} is outside metal-admin {METAL_ADMIN} -- refusing to place it.")
gateway = metal.network_address + 1
band_low = metal.network_address + STATIC_BAND_LOW
band_high = metal.network_address + STATIC_BAND_HIGH
if rack == gateway:
die(f"rack IP {rack} is the .1 GATEWAY of {METAL_ADMIN} -- refusing (D-120: .1 is the "
f"site gateway, not a static-service address).")
if not (band_low <= rack <= band_high):
die(f"rack IP {rack} is outside the D-120 static band {band_low}-{band_high} of "
f"{METAL_ADMIN} -- the rack's metal-admin IP must land in .2-.49 (not the dynamic/"
f"node bands).")
rack_addr = f"{rack}/{metal.prefixlen}" # NetBox stores the host WITH the plane mask
nb = get_nb(url, token)
# (c) Preconditions in the apex: the transit role, the container prefix, the site.
role = nb.one("ipam/roles", slug=ROLE_SLUG)
if role is None:
die(f"role '{ROLE_SLUG}' absent -- D-124 calls for the transit's OWN dedicated role "
f"(mirroring the D-115 `edge` role); seed it in office1-netbox first. Refusing to "
f"place the transit prefix with no role.")
if nb.one("ipam/prefixes", prefix=CONTAINER) is None:
die(f"container {CONTAINER} (Cloud) absent in the apex -- refusing to place the transit "
f"in an unallocated supernet.")
site = nb.one("dcim/sites", slug=SITE_SLUG)
if site is None:
die(f"site '{SITE_SLUG}' absent -- cannot scope the transit prefix. The DC site is a "
f"precondition (it already exists in the apex); this tool never creates it.")
# (d) Present-state (idempotency) -- resolved up front, part of the whole-plan gate.
transit_str = str(transit)
transit_present = nb.one("ipam/prefixes", prefix=transit_str) is not None
rack_present = nb.one("ipam/ip-addresses", address=rack_addr) is not None
created = existing = 0
# ---- CREATE (transit prefix, then rack ip-address) -- only reached once the WHOLE
# plan above is viable, so this loop cannot half-write the apex. ----
print("\nTransit prefix (office1<->dc0 mesh leg):")
if transit_present:
print(f" EXISTS {transit_str}")
existing += 1
elif not args.commit:
print(f" [dry-run] would CREATE {transit_str} role={ROLE_SLUG} "
f"scope=dcim.site:{SITE_SLUG}")
created += 1
else:
payload = {"prefix": transit_str, "role": role["id"], "status": STATUS,
"description": TRANSIT_DESC, "scope_type": "dcim.site", "scope_id": site["id"]}
o = nb.create("ipam/prefixes", payload)
print(f" CREATED {transit_str} (id={o['id']}) role={ROLE_SLUG} scope={SITE_SLUG}")
created += 1
print("\nRack metal-admin static IP:")
if rack_present:
print(f" EXISTS {rack_addr}")
existing += 1
elif not args.commit:
print(f" [dry-run] would CREATE {rack_addr} dns={RACK_DNS} (metal-admin static band)")
created += 1
else:
payload = {"address": rack_addr, "status": STATUS,
"dns_name": RACK_DNS, "description": RACK_DESC}
o = nb.create("ipam/ip-addresses", payload)
print(f" CREATED {rack_addr} (id={o['id']}) dns={RACK_DNS}")
created += 1
verb = "would create" if not args.commit else "created"
print(f"\n{'='*66}\n{verb}: {created} already present: {existing}")
if not args.commit:
print("DRY RUN -- nothing was written. Re-run with --commit.")
return 0
if __name__ == "__main__":
sys.exit(main())
