#!/usr/bin/env python3
"""
Seed a SANDBOX NetBox from the JSON draft produced by netbox/prod-draft-dump.py.
Phase 2 of the two-phase seeding loop (see prod-draft-dump.py for the why).
THIS SCRIPT CANNOT WRITE UPSTREAM. It has no upstream URL and no upstream token
-- it only knows the JSON file and the sandbox it is pointed at. That is
deliberate: "the sim never writes upstream" is enforced by STRUCTURE, not by
remembering to pass the right flag.
DRY BY DEFAULT. Nothing is written without --commit. (The repo learned this the
hard way: dc-dc-prefixes-import.py used to write with no flag at all, and would
have silently bound VR1 DC0's prefixes to the wrong site -- D-117.)
Idempotent: existing objects are detected by natural key (slug, or the prefix
itself) and skipped. Re-running is safe.
IDs are never carried across. The draft refs everything by slug; this script
re-resolves those against the SANDBOX's own ids.
Auth: the sandbox (NetBox 4.6) uses v2 tokens -- the wire form is
`nbt_<key>.<plaintext>`. The API's `token` field alone is NOT usable (it parses
as a legacy v1 token and 403s with "Invalid v1 token", which is thoroughly
misleading). Pass the ASSEMBLED value.
Usage (on office1-netbox, where NetBox is on localhost and the token lives):
sudo NETBOX_URL=http://localhost:8000 \
NETBOX_TOKEN="$(cat /root/netbox-secrets/api.token)" \
python3 sandbox-seed.py --draft vr1-draft.json # preview
... --draft vr1-draft.json --commit # apply
"""
import argparse
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
# Seeding order is DEPENDENCY order: a site needs its region; a prefix needs its
# role and its site. Prefixes do NOT need parent-prefix ordering -- NetBox
# derives the hierarchy from the CIDR itself.
ORDER = ["ipam/rirs", "ipam/roles", "dcim/regions", "dcim/sites",
"ipam/aggregates", "ipam/prefixes"]
# Natural key per endpoint -- how we decide "does this already exist?"
NATURAL_KEY = {
"ipam/rirs": "slug",
"ipam/roles": "slug",
"dcim/regions": "slug",
"dcim/sites": "slug",
"ipam/aggregates": "prefix",
"ipam/prefixes": "prefix",
}
def die(msg: str):
print(f"FAIL: {msg}", file=sys.stderr)
sys.exit(2)
class NB:
def __init__(self, base: str, token: str):
self.base = base.rstrip("/")
self.token = token
def _req(self, method: str, path: str, body=None):
url = f"{self.base}/api/{path}"
data = json.dumps(body).encode() if body is not None else None
req = urllib.request.Request(url, data=data, method=method, headers={
"Authorization": f"Token {self.token}",
"Accept": "application/json",
"Content-Type": "application/json",
})
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'. This NetBox is 4.6: the token must be "
"the ASSEMBLED v2 form nbt_<key>.<plaintext>, not the API's bare "
"`token` field. See docs/vr1-office1-as-built.md.")
die(f"HTTP {exc.code} {method} {path}: {detail}")
def get_one(self, path: str, **flt):
q = urllib.parse.urlencode(flt)
res = self._req("GET", f"{path}/?{q}&limit=1")
return res["results"][0] if res["results"] else None
def create(self, path: str, payload: dict):
return self._req("POST", f"{path}/", payload)
def patch(self, path: str, payload: dict):
return self._req("PATCH", f"{path}/", payload)
def resolve(nb: NB, cache: dict, path: str, slug: str):
"""slug -> sandbox id, cached. None if the slug is absent."""
if slug is None:
return None
key = (path, slug)
if key not in cache:
obj = nb.get_one(path, slug=slug)
cache[key] = obj["id"] if obj else None
return cache[key]
def dep(nb: NB, cache: dict, planned: set, path: str, slug: str, commit: bool):
"""Resolve a dependency to (ok, id).
In a DRY RUN the dependency may not exist yet because we are not creating
anything -- but it IS in the plan, and will exist by the time we commit.
Treating that as "missing" made every prefix cascade-skip behind its role,
which reported a plan of 27/129 and looked like the draft was broken. A dry
run must model the END STATE, not the current empty one.
"""
if slug is None:
return True, None
rid = resolve(nb, cache, path, slug)
if rid is not None:
return True, rid
if not commit and (path, slug) in planned:
return True, None # will exist; no id needed, we never POST
return False, None
def build_payload(nb: NB, cache: dict, ep: str, rec: dict, planned: set, commit: bool):
"""Draft record -> sandbox POST body, re-linking every ref by slug."""
p = {k: v for k, v in rec.items()
if k not in ("role", "rir", "region", "parent", "scope_site",
"scope_region", "scope_type", "status") and v is not None}
if rec.get("status"):
p["status"] = rec["status"]
if ep == "dcim/regions" and rec.get("parent"):
ok, rid = dep(nb, cache, planned, "dcim/regions", rec["parent"], commit)
if not ok:
return None, f"parent region '{rec['parent']}' not present"
p["parent"] = rid
if ep == "dcim/sites" and rec.get("region"):
ok, rid = dep(nb, cache, planned, "dcim/regions", rec["region"], commit)
if not ok:
return None, f"region '{rec['region']}' not present"
p["region"] = rid
if ep == "ipam/aggregates":
ok, rid = dep(nb, cache, planned, "ipam/rirs", rec.get("rir"), commit)
if not ok:
return None, f"rir '{rec.get('rir')}' not present"
p["rir"] = rid
if ep == "ipam/prefixes":
if rec.get("role"):
ok, rid = dep(nb, cache, planned, "ipam/roles", rec["role"], commit)
if not ok:
return None, f"role '{rec['role']}' not present"
p["role"] = rid
if rec.get("scope_site"):
ok, sid = dep(nb, cache, planned, "dcim/sites", rec["scope_site"], commit)
if not ok:
return None, f"site '{rec['scope_site']}' not present"
p["scope_type"] = "dcim.site"
if sid is not None:
p["scope_id"] = sid
elif rec.get("scope_region"):
ok, rid = dep(nb, cache, planned, "dcim/regions", rec["scope_region"], commit)
if not ok:
return None, f"region '{rec['scope_region']}' not present"
p["scope_type"] = "dcim.region"
if rid is not None:
p["scope_id"] = rid
return p, None
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.split("\n\n", 1)[0])
ap.add_argument("--draft", required=True, help="JSON produced by prod-draft-dump.py")
ap.add_argument("--commit", action="store_true",
help="WRITE to the sandbox. Default is a DRY RUN that writes nothing.")
ap.add_argument("--update", action="store_true",
help="Also PATCH objects that already exist so they match the draft "
"(otherwise existing objects are left alone). Requires --commit.")
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 (the SANDBOX's, not upstream's).")
with open(args.draft) as fh:
draft = json.load(fh)
src = draft.get("_source", "?")
# -----------------------------------------------------------------------
# SAFETY GUARD -- runs BEFORE anything else touches the network.
#
# Never write to the instance the draft was READ from. This is
# self-configuring: the draft records its own _source, so the guard keeps
# working if upstream ever moves hosts. A hardcoded hostname denylist would
# silently stop protecting the day the URL changed -- which is exactly the
# class of stale-literal bug this repo keeps finding.
#
# An earlier version of this guard sat AFTER json.load() and after the
# banner. It was dead code for any malformed draft, and it read as "safe"
# to a reviewer. Guards go FIRST.
# -----------------------------------------------------------------------
def host(u: str) -> str:
return (urllib.parse.urlparse(u).hostname or u).lower()
if src != "?" and host(url) == host(src):
die(f"REFUSING: NETBOX_URL points at the UPSTREAM NetBox ({host(src)}) -- the "
f"instance this draft was READ FROM. This script seeds a SANDBOX only; "
f"the sim never writes upstream.")
nb = NB(url, token)
print(f"Draft source : {src}")
print(f"Sandbox : {url}")
if not args.commit:
print("\n*** DRY RUN -- nothing will be written. Re-run with --commit to apply. ***")
else:
print("\n*** COMMITTING to the sandbox. ***")
cache, planned = {}, set()
created = existing = skipped = updated = 0
for ep in ORDER:
rows = draft.get(ep, [])
if not rows:
continue
print(f"\n{ep} ({len(rows)} in draft)")
nkey = NATURAL_KEY[ep]
for rec in rows:
nval = rec.get(nkey)
found = nb.get_one(ep, **{nkey: nval})
if found:
existing += 1
cache[(ep, nval)] = found["id"]
if args.update:
payload, why = build_payload(nb, cache, ep, rec, planned, args.commit)
if payload is None:
print(f" SKIP {nval} -- {why}")
continue
if not args.commit:
print(f" [dry-run] would UPDATE {nval}")
continue
nb.patch(f"{ep}/{found['id']}", payload)
print(f" UPDATED {nval} (id={found['id']})")
updated += 1
continue
payload, why = build_payload(nb, cache, ep, rec, planned, args.commit)
if payload is None:
print(f" SKIP {nval} -- {why}")
skipped += 1
continue
if not args.commit:
print(f" [dry-run] would CREATE {nval}")
planned.add((ep, nval))
created += 1
continue
obj = nb.create(ep, payload)
cache[(ep, nval)] = obj["id"]
print(f" CREATED {nval} (id={obj['id']})")
created += 1
verb = "would create" if not args.commit else "created"
print(f"\n{'='*66}")
print(f"{verb}: {created} already present: {existing} updated: {updated} skipped: {skipped}")
if not args.commit:
print("DRY RUN -- the sandbox was NOT modified. Re-run with --commit.")
return 0
if __name__ == "__main__":
sys.exit(main())
