#!/usr/bin/env python3
"""
Register a D-134 UTILITY-BAND HOST's ip-addresses in office1-netbox (the VR1 IPAM
apex). ONE utility host per invocation, both of its planes (metal-admin +
provider-public). Places the individual host assignment WITHIN the D-134 utility
RANGE that netbox/dc-plane-apex-import.py already loaded -- that range is the
PRECONDITION; a missing one is a hard STOP, never a silent create.
WHY THIS EXISTS (measured 2026-08-07). dc-plane-apex-import.py loaded the D-134 BAND
RANGES (.4-.49) and the ruled VIP ADDRESSES (.50-.62), but NOT the individual
utility-HOST addresses within the band. Against office1-netbox this session: the four
utility ranges exist, yet none of dc0 .5/.6/.7 or dc1 .6 is an ip-address object. This
tool closes that gap so every DC standup records its utility hosts from a tool rather
than by hand (Roosevelt-delta: same at every bare-metal DC).
SCOPE, deliberately narrow (advisor-reviewed 2026-08-07):
* IN : the lib-hosts utility-band hosts -- juju-01 (.5), maas-01 (.6),
tailscale-01 (.7). All 2-plane (metal-admin + provider-public) by the
as-built carve (scripts/lib-hosts.sh: juju-01 "only TWO planes", tailscale-01
"metal-admin + provider-public", maas-01 "SAME 2-plane shape").
* OUT : the .4 ARTIFACT host. It is NOT a lib-hosts host -- it is a metal-admin-ONLY
service alias on the rack bridge (dc0 apt mirror / dc1 caching proxy, D-135),
per-DC divergent, with no host NAME anywhere in the repo. Recording it needs
an operator naming ruling; it is a separate queued finding, not this tool.
DERIVED, NEVER HARDCODED (hard rule 2/3):
* plane CIDRs <- scripts/lib-net.sh via lib_net_select_dc (site-keyed authority)
* host octet <- scripts/lib-hosts.sh via lib_hosts_select_dc (the MAC-pinned carve)
The address is plane_base + octet. The ONLY in-script tables are suffix->role-label
(for the human description) and the 2-plane set -- both cited to lib-hosts, neither an
address. An empty/missing octet from lib-hosts is a hard die (NEVER a .0 default -- a
clean zero that reads as success is this repo's recorded failure mode).
DRY BY DEFAULT -- --commit writes, then READS BACK every object. Upstream write GATED IN
CODE (SANDBOX_HOSTS + --yes-write-upstream). WHOLE-PLAN PREFLIGHT (both planes) runs
BEFORE any create, so a half-write cannot leave one plane recorded and the other not.
Usage (through the office1-netbox tunnel/base-leg, with the sandbox v2 token):
. ~/vr1-office1-creds/vr1-netbox-sandbox.env # NETBOX_URL=http://10.10.1.10:8000 + NETBOX_TOKEN
python3 netbox/dc-util-hosts-import.py --site vr1-dc0 --host vr1-dc0-tailscale-01
... add --commit to write.
Exit: 0 ok | 2 die (bad input / failed precondition).
"""
import argparse
import ipaddress
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
UA = "curl/8.5.0" # upstream 403s the default python UA -- see references/platform-traps.md
# A sandbox is local, or the known Office1 apex 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"}
STATUS = "active" # a real address on the wire
# D-134 utility octet band: .4-.9. A --host whose lib-hosts octet is outside this band is
# a ROLE node (.100+), NOT a utility host -- rejected. (.4 is the artifact host, OUT of
# scope; this tool serves .5/.6/.7 -- see the module docstring.)
UTIL_BAND_LOW = 4
UTIL_BAND_HIGH = 9
# The 2-plane set of the utility hosts, as-built (scripts/lib-hosts.sh). NOT an address --
# the addresses are derived (plane base from lib-net + octet from lib-hosts).
UTIL_PLANES = ("metal-admin", "provider-public")
# host SUFFIX -> role label (description only; cited to lib-hosts). Membership here is also
# the whitelist: a --host whose suffix is absent is refused, so a role node or the .4
# artifact cannot slip through even if its octet somehow landed in the band.
ROLE_LABEL = {
"juju-01": "Juju controller (D-104 dedicated controller VM)",
"maas-01": "MAAS region+rack VM (D-132 q1 per-DC region)",
"tailscale-01": "Tailscale subnet router (D-129(iii) amendment)",
}
def die(msg: str):
print(f"FAIL: {msg}", file=sys.stderr)
sys.exit(2)
def _repo() -> str:
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def lib_net(site: str) -> dict:
"""Plane name -> CIDR from lib-net.sh -- the site-keyed authority (mirrors
netbox/dc-plane-apex-import.py's derivation)."""
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:
die(f"lib-net.sh did not yield planes for {site} (rc={p.returncode}): {p.stderr.strip()[:200]}")
planes = {}
for line in p.stdout.splitlines():
if "\t" in line:
name, cidr = line.split("\t", 1)
planes[name.strip()] = cidr.strip()
if not planes:
die(f"lib-net.sh yielded NO planes for {site} -- refusing to derive an address from nothing.")
return planes
def lib_hosts_octet(site: str, host: str) -> int:
"""The host's last octet from lib-hosts.sh HOST_OCTET (the MAC-pinned carve). An
empty/missing value is a HARD die -- never a .0 default (a clean zero that reads as
success is this repo's recorded failure mode, memory #13)."""
sh = (f'source "{_repo()}/scripts/lib-hosts.sh" >/dev/null 2>&1 || exit 3; '
f'lib_hosts_select_dc "{site}" >/dev/null 2>&1 || exit 3; '
f'printf "%s" "${{HOST_OCTET[{host}]:-}}"')
p = subprocess.run(["bash", "-c", sh], capture_output=True, text=True)
if p.returncode != 0:
die(f"lib-hosts.sh did not select {site} for {host} (rc={p.returncode}): {p.stderr.strip()[:200]}")
raw = p.stdout.strip()
if not raw:
die(f"lib-hosts HOST_OCTET has NO entry for {host} in {site} -- refusing to invent "
f"an octet (an empty/.0 default is the clean-zero failure this repo guards against).")
try:
return int(raw)
except ValueError:
die(f"lib-hosts octet for {host} is not an integer: {raw!r}")
class NB:
"""Stdlib NetBox client -- same shape as netbox/dc-rack-mgmt-import.py; 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 all(self, path, **flt):
q = urllib.parse.urlencode({**flt, "limit": 1000})
res = self._req("GET", f"{path}/?{q}")
return res.get("results", []) if res else []
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 range_covering(nb, addr: ipaddress.IPv4Address):
"""Return the apex ip-range that contains addr, or None. The D-134 utility range is a
PRECONDITION (loaded by dc-plane-apex-import.py); containment in a real range is the
check that can actually fail against live state."""
for r in nb.all("ipam/ip-ranges"):
try:
s = ipaddress.ip_address(r["start_address"].split("/")[0])
e = ipaddress.ip_address(r["end_address"].split("/")[0])
except (KeyError, ValueError):
continue
if s <= addr <= e:
return r
return None
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.split("\n\n", 1)[0])
ap.add_argument("--site", choices=("vr1-dc0", "vr1-dc1"), default=os.environ.get("UTIL_SITE"),
help="REQUIRED. Which DC (env: UTIL_SITE). Explicit -- never inferred -- so "
"one DC's values cannot land scoped to another site.")
ap.add_argument("--host", default=os.environ.get("UTIL_HOST"),
help="REQUIRED. The utility host to record, e.g. vr1-dc0-tailscale-01 "
"(env: UTIL_HOST). One host per run; liveness is operator-asserted "
"(there is no L3 path from the apex host to a DC plane -- SEC-010).")
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.")
# args-or-env, so not argparse required=True (that breaks the env fallback). Hand-roll.
if not args.site:
die("--site (or UTIL_SITE) is REQUIRED -- the target DC is never inferred.")
if args.site not in ("vr1-dc0", "vr1-dc1"):
die(f"--site {args.site!r} is not a known DC (expected vr1-dc0 or vr1-dc1).")
if not args.host:
die("--host (or UTIL_HOST) is REQUIRED -- the utility host is never inferred.")
site = args.site
host = args.host
if not re.fullmatch(r"vr1-dc[01]-[a-z0-9-]+", host):
die(f"--host {host!r} must look like vr1-dc0-<role>-NN (letters/digits/dashes only).")
if not host.startswith(site + "-"):
die(f"--host {host!r} is not in --site {site!r} -- refusing (a dc0 host must not land "
f"scoped to dc1 or vice versa).")
suffix = host[len(site) + 1:]
if suffix not in ROLE_LABEL:
die(f"--host {host!r} (suffix {suffix!r}) is not a utility host this tool records. "
f"IN scope: {', '.join('<site>-' + s for s in sorted(ROLE_LABEL))}. The .4 artifact "
f"host and the .100+ role nodes are OUT of scope (see the module docstring).")
hostname = (urllib.parse.urlparse(url).hostname or url).lower()
is_sandbox = hostname 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 '{hostname}': 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. ***")
# ---- DERIVE the plan (both planes) from lib-net + lib-hosts -- no invented value ----
octet = lib_hosts_octet(site, host)
if not (UTIL_BAND_LOW <= octet <= UTIL_BAND_HIGH):
die(f"{host} has octet .{octet}, outside the D-134 utility band .{UTIL_BAND_LOW}-.{UTIL_BAND_HIGH} "
f"-- this is a role node, not a utility host. Refusing.")
planes = lib_net(site)
plan = [] # (plane, addr_str, addr_obj, description)
for plane in UTIL_PLANES:
cidr = planes.get(plane)
if not cidr:
die(f"lib-net has no {plane} plane for {site} -- refusing to derive its address.")
net = ipaddress.ip_network(cidr)
addr = net.network_address + octet
addr_str = f"{addr}/{net.prefixlen}"
desc = f"D-134 utility .{octet} -- {site} {ROLE_LABEL[suffix]} ({plane} leg)"
plan.append((plane, addr_str, addr, desc))
planned_addr_strs = {a for _, a, _, _ in plan}
print(f"\nHost : {host} (octet .{octet}, planes: {', '.join(UTIL_PLANES)})")
for plane, addr_str, _, _ in plan:
print(f" plan {plane:16s} {addr_str}")
nb = get_nb(url, token)
# ---- WHOLE-PLAN PREFLIGHT (both planes) BEFORE any create -- so a bad plane cannot
# leave the other written. ----
for plane, addr_str, addr, _ in plan:
r = range_covering(nb, addr)
if r is None:
die(f"no apex ip-range covers {addr} ({plane}) -- the D-134 utility range is a "
f"PRECONDITION (run dc-plane-apex-import.py first). Refusing to place a host "
f"address in an unallocated band.")
# dns_name collision: any OTHER address already carrying this dns_name (not one of our
# two planned plane addresses) is a stale/rename conflict, not an idempotent re-run.
for a in nb.all("ipam/ip-addresses", dns_name=host):
if a.get("dns_name") == host and a.get("address") not in planned_addr_strs:
die(f"dns_name {host!r} already on {a.get('address')}, which is not a planned plane "
f"address -- possible stale/rename conflict. Resolve in the apex before writing.")
# present-state (idempotency), resolved up front as part of the whole-plan gate
present = {addr_str: (nb.one("ipam/ip-addresses", address=addr_str) is not None)
for _, addr_str, _, _ in plan}
created = existing = 0
for plane, addr_str, _, desc in plan:
print(f"\n{plane} leg:")
if present[addr_str]:
print(f" EXISTS {addr_str} dns={host}")
existing += 1
elif not args.commit:
print(f" [dry-run] would CREATE {addr_str} dns={host}")
created += 1
else:
o = nb.create("ipam/ip-addresses",
{"address": addr_str, "status": STATUS, "dns_name": host, "description": desc})
print(f" CREATED {addr_str} (id={o['id']}) dns={host}")
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())
