#!/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_.`. 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())