Christoph Emsenhuber
b52d6aa49e
Oktett-Klasse .150-.199 (App-Services). Nautobot gibt damit .30/.31 an AGH frei. .157 ist gitea, deshalb nimmt redis .158. - Datenmigration ueber neues nautobot/tp45-ip-move.py: Nautobot verbietet das Aendern einer IP nach dem Anlegen -> anlegen/primary_ip4 umhaengen/ Altobjekt loeschen, in dieser Reihenfolge - Stack-Redeploy (PUT /api/stacks/203), Caddy-Backend, Core-Literal *24 ueber GC deployt (Plan 51a90e25) mit Wirksamkeitspruefung am Geraet - caddy/deploy.sh: push_verified() ersetzt den push||push-Fallback. `pct push` meldet Exit 0 AUCH wenn es die Quelle nicht lesen kann, der Fallback griff deshalb nie und `caddy validate` pruefte die ALTE Datei -> gruene Abnahme bei unveraendertem Zustand (Symptom: 502 bei gesundem Backend) - 39 Referenzen in 28 Dateien mitgezogen; lebende Doku nachgezogen (CLAUDE.md, ip-plan.md, cf-tunnel-mapping §0-Kopf, TP16.3/TP31-Prosa) - GC-Compliance 140/140, 0 Drifts Neue Funde (TP45-Prosa + CLAUDE.md): - Die Core-Adressliste `nautobot` wird von GC nirgends gerendert -> ein Recovery von null erzeugt einen Core, in den Nautobot nicht mehr per SSH kommt. Gleiche Klasse wie die fehlenden DHCP-Reservierungen. - Regel *24 (Literal) und *28 (Liste) sind Dubletten; der Literal steht bei 0 Paketen. Zusammenlegung bleibt offen. - gc-manual-plan.py: /api/extras/job-log-entries/ liefert 404 (Nautobot 3.x) -> bei gescheitertem Deploy fehlt die Fehlermeldung. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
396 lines
16 KiB
Python
396 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
r"""Manuellen Golden-Config-Plan anlegen, freigeben und deployen.
|
|
|
|
Der von der Governance (TP22) vorgeschriebene Weg fuer Geraeteaenderungen:
|
|
feature-/manual-scoped, **nie** "full intended" — ein Voll-Push schriebe
|
|
`<<REDACTED>>`-Sektionen woertlich aufs Geraet (WLAN-PSKs, WG-Privkey).
|
|
|
|
python3 nautobot/gc-manual-plan.py --device crs326-core \
|
|
--commands-file /pfad/zu/commands.rsc --change-id "TP5-Etappe1" --dry-run
|
|
python3 nautobot/gc-manual-plan.py --device crs326-core \
|
|
--commands-file /pfad/zu/commands.rsc --change-id "TP5-Etappe1" --deploy
|
|
|
|
Ablauf: Job "Generate Config Plans" (plan_type=manual) -> Plan auf Approved
|
|
-> Job "Deploy Config Plans". Dazwischen wird jeweils auf den JobResult gewartet.
|
|
|
|
ZWEI TP22-LEHREN SIND HIER EINGEBAUT
|
|
------------------------------------
|
|
1. **Job-Status ist kein Wirksamkeitsnachweis.** Der Deploy-Job meldete schon
|
|
SUCCESS, waehrend der Plan auf Failed stand. Das Skript liest darum nach dem
|
|
Deploy den *Plan*-Status zurueck und richtet seinen Exit-Code danach.
|
|
Die eigentliche Abnahme bleibt trotzdem die Gegenpruefung am Geraet.
|
|
2. **Geraete werden einzeln per Name aufgeloest**, nie positional aus einer
|
|
Sammelabfrage. Nautobot sortiert Sammelabfragen alphabetisch; eine positionale
|
|
Zuordnung hat einen Deploy schon mal ans falsche Geraet geschickt.
|
|
|
|
DRITTE LEHRE (2026-08-17): --assert
|
|
-----------------------------------
|
|
"Plan-Status: Completed" sagt nur, dass die Zeilen abgesetzt wurden — nicht, dass
|
|
sie *gewirkt* haben. Am 2026-08-17 blieb ein `/interface/wifi/set ... !channel`
|
|
wirkungslos, ohne zu werfen; das Selbstprotokoll im Plan meldete trotzdem Erfolg,
|
|
weil es das **Profil** las, waehrend der **Interface-Inline-Override** gewann.
|
|
|
|
`--assert` prueft den Effekt nach dem Deploy per RouterOS-REST **am Geraet** und
|
|
bricht laut ab, wenn er nicht eintritt. Damit ist die Abnahme maschinell statt
|
|
davon abhaengig, dass jemand daran denkt nachzusehen.
|
|
|
|
--assert '/ip/firewall/filter|count=0'
|
|
--assert '/interface/wifi?name=wifi1|channel.width=20/40mhz'
|
|
--assert 'cmd:/interface/wifi/monitor?numbers=wifi1&once=|channel=5180/ax/Ce/I'
|
|
|
|
Format: ``[cmd:]<pfad>[?query]|<feld-oder-count>=<erwartet>`` — mehrfach angebbar.
|
|
Ohne ``cmd:`` ein GET (Lesen), mit ``cmd:`` ein POST (RouterOS-Kommandos wie
|
|
``monitor``, die kein GET sind). Trenner ist ``|``, **nicht** ``:`` — MAC-Adressen
|
|
in Werten enthalten Doppelpunkte.
|
|
|
|
⚠️ **Regel fuer Funkaenderungen:** die Assertion muss ``monitor`` lesen, NIE das
|
|
Profil. RouterOS hat drei Schichten (channel-Profil -> configuration -> Inline am
|
|
Interface); ``print detail`` zeigt aufgeloeste Werte, ``export`` die rohen, und nur
|
|
``monitor`` das, was tatsaechlich auf der Luft ist. Genau diese Verwechslung war
|
|
der Fehler vom 2026-08-17.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
NB = os.environ.get("NAUTOBOT_URL", "http://10.40.20.154:8080")
|
|
GEN_JOB = "f93bfa53-2332-4fa2-a6ab-5921c802fcee" # Generate Config Plans
|
|
DEPLOY_JOB = "31c3f093-eeec-43fa-ba4a-0aed2f3c67aa" # Deploy Config Plans
|
|
|
|
|
|
def token():
|
|
t = os.environ.get("NAUTOBOT_TOKEN")
|
|
if not t:
|
|
raise SystemExit("NAUTOBOT_TOKEN nicht gesetzt "
|
|
"(steht in docs/nas-vlan20/stacks/210-nautobot.yml, gitignored)")
|
|
return t
|
|
|
|
|
|
def api(path, method="GET", payload=None):
|
|
req = urllib.request.Request(
|
|
f"{NB}{path}",
|
|
data=json.dumps(payload).encode() if payload is not None else None,
|
|
method=method,
|
|
headers={"Authorization": f"Token {token()}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else {}
|
|
except urllib.error.HTTPError as e:
|
|
sys.stderr.write(f"HTTP {e.code} auf {method} {path}\n{e.read().decode()[:2000]}\n")
|
|
raise
|
|
|
|
|
|
def resolve_device(name):
|
|
"""Einzeln per Name — siehe Kopfkommentar, Lehre 2."""
|
|
res = api(f"/api/dcim/devices/?name={name}")["results"]
|
|
if len(res) != 1:
|
|
raise SystemExit(f"Device {name!r}: {len(res)} Treffer statt genau 1")
|
|
return res[0]["id"]
|
|
|
|
|
|
def status_id(name):
|
|
res = api(f"/api/extras/statuses/?content_types=nautobot_golden_config.configplan&name={name}")["results"]
|
|
if len(res) != 1:
|
|
raise SystemExit(f"ConfigPlan-Status {name!r}: {len(res)} Treffer statt genau 1")
|
|
return res[0]["id"]
|
|
|
|
|
|
def wait_job(job_result_id, label, timeout=900):
|
|
"""Pollt bis der JobResult terminal ist. Gibt den finalen Status zurueck."""
|
|
deadline = time.time() + timeout
|
|
last = None
|
|
while time.time() < deadline:
|
|
jr = api(f"/api/extras/job-results/{job_result_id}/")
|
|
st = (jr.get("status") or {}).get("value") or jr.get("status")
|
|
if st != last:
|
|
print(f" {label}: {st}")
|
|
last = st
|
|
if st in ("SUCCESS", "FAILURE", "ERRORED", "REVOKED"):
|
|
return st
|
|
time.sleep(5)
|
|
raise SystemExit(f"{label}: Timeout nach {timeout}s")
|
|
|
|
|
|
def job_log(job_result_id, levels=("error", "warning", "failure")):
|
|
try:
|
|
logs = api(f"/api/extras/job-log-entries/?job_result={job_result_id}&limit=200")["results"]
|
|
except Exception:
|
|
return
|
|
for e in logs:
|
|
if (e.get("log_level") or "").lower() in levels:
|
|
print(f" [{e.get('log_level')}] {e.get('message')}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# --assert: Wirksamkeitspruefung am Geraet (RouterOS REST)
|
|
# ---------------------------------------------------------------------------
|
|
# Ein Geraet = ein Passwort-Env. User ist auf allen vier MikroTik `admin`
|
|
# (einheitliches Admin-PW, User-Vorgabe 2026-08-08).
|
|
PASS_ENV = {
|
|
"crs326-core": "CRS326_CORE_PASS",
|
|
"crs326-branch": "CRS326_BRANCH_PASS",
|
|
"ap1": "ROUTER_AP1_PASS",
|
|
"ap2": "ROUTER_AP2_PASS",
|
|
}
|
|
|
|
|
|
SECRETS_FILE = pathlib.Path.home() / ".secrets" / "credentials.env"
|
|
|
|
|
|
def _from_secrets(key):
|
|
"""`credentials.env` ist eine NACHSCHLAGE-Datei, keine sourcebare: sie enthaelt
|
|
u.a. WireGuard-`[Interface]`-Bloecke, an denen `set -a; . datei` in zsh
|
|
abbricht — und zwar *stillschweigend* fuer alle Zeilen danach. Darum hier
|
|
ein simpler KEY=VALUE-Parser statt Shell-Ausfuehrung."""
|
|
if not SECRETS_FILE.exists():
|
|
return None
|
|
for line in SECRETS_FILE.read_text(errors="replace").splitlines():
|
|
line = line.strip()
|
|
if line.startswith("export "):
|
|
line = line[7:]
|
|
k, sep, v = line.partition("=")
|
|
if sep and k.strip() == key:
|
|
return v.strip().strip("'\"")
|
|
return None
|
|
|
|
|
|
def device_credentials(name):
|
|
env = PASS_ENV.get(name)
|
|
if not env:
|
|
raise SystemExit(f"--assert: kein Passwort-Env fuer {name!r} bekannt "
|
|
f"(bekannt: {', '.join(sorted(PASS_ENV))})")
|
|
pw = os.environ.get(env) or _from_secrets(env)
|
|
if not pw:
|
|
raise SystemExit(f"--assert: ${env} weder im Environment noch in "
|
|
f"{SECRETS_FILE} gefunden")
|
|
user = os.environ.get(f"{name.upper().replace('-', '_')}_USER", "admin")
|
|
return user, pw
|
|
|
|
|
|
def device_address(name):
|
|
"""Mgmt-IP aus Nautobot statt hartkodiert — eine Quelle der Wahrheit."""
|
|
dev = api(f"/api/dcim/devices/?name={name}&depth=1")["results"][0]
|
|
ip = (dev.get("primary_ip4") or {}).get("address")
|
|
if not ip:
|
|
raise SystemExit(f"--assert: {name} hat keine primary_ip4 in Nautobot")
|
|
return ip.split("/")[0]
|
|
|
|
|
|
def _redact(text, secret):
|
|
"""Passwoerter nie in Fehlerausgaben (Vorgabe Christoph)."""
|
|
return text.replace(secret, "<<REDACTED>>") if secret else text
|
|
|
|
|
|
def rest_call(name, path, query, post=False):
|
|
host = device_address(name)
|
|
user, pw = device_credentials(name)
|
|
url = f"http://{host}/rest{path}"
|
|
body = None
|
|
if post:
|
|
# ?a=b&c= -> {"a": "b", "c": ""} (RouterOS-Kommandos wie monitor)
|
|
body = json.dumps({
|
|
k: v for k, _, v in (p.partition("=") for p in query.split("&") if p)
|
|
}).encode()
|
|
elif query:
|
|
# Werte URL-encoden. Ein Kommentar wie "TP5-E5: MAIN->IOT" traegt ein
|
|
# Leerzeichen; roh angehaengt wirft http.client InvalidURL ("URL can't
|
|
# contain control characters") — der Deploy ist dann laengst durch und
|
|
# nur die Gegenpruefung crasht, was schlimmer ist als kein Assert.
|
|
# (2026-08-19 bei TP5-E5 aufgeschlagen.) Trennzeichen & und = bleiben.
|
|
parts = []
|
|
for p_ in query.split("&"):
|
|
if not p_:
|
|
continue
|
|
k, sep, v = p_.partition("=")
|
|
parts.append(f"{urllib.parse.quote(k, safe='')}{sep}"
|
|
f"{urllib.parse.quote(v, safe='')}" if sep
|
|
else urllib.parse.quote(k, safe=""))
|
|
url += "?" + "&".join(parts)
|
|
auth = base64.b64encode(f"{user}:{pw}".encode()).decode()
|
|
req = urllib.request.Request(
|
|
url, data=body, method="POST" if post else "GET",
|
|
headers={"Authorization": f"Basic {auth}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else []
|
|
except urllib.error.HTTPError as e:
|
|
detail = _redact(e.read().decode()[:800], pw)
|
|
raise SystemExit(f"--assert: HTTP {e.code} auf {url}\n{detail}")
|
|
except OSError as e:
|
|
raise SystemExit(f"--assert: {host} nicht erreichbar ({e})")
|
|
|
|
|
|
def _field(obj, field):
|
|
"""RouterOS-REST liefert Sub-Properties teils flach ('channel.width'),
|
|
teils verschachtelt. Beides bedienen."""
|
|
if field in obj:
|
|
return obj[field]
|
|
cur = obj
|
|
for part in field.split("."):
|
|
if not isinstance(cur, dict) or part not in cur:
|
|
return None
|
|
cur = cur[part]
|
|
return cur
|
|
|
|
|
|
ASSERT_RE = re.compile(r"^(?P<cmd>cmd:)?(?P<path>[^|?]+)(?:\?(?P<query>[^|]*))?\|"
|
|
r"(?P<field>[^=]+)=(?P<expect>.*)$")
|
|
|
|
|
|
def run_assertions(name, specs):
|
|
print(f"\n -- Wirksamkeitspruefung am Geraet ({len(specs)}) --")
|
|
failures = []
|
|
for spec in specs:
|
|
m = ASSERT_RE.match(spec)
|
|
if not m:
|
|
raise SystemExit(
|
|
f"--assert: Spec nicht lesbar: {spec!r}\n"
|
|
f" Erwartet: [cmd:]<pfad>[?query]|<feld-oder-count>=<erwartet>")
|
|
path, query = m["path"], m["query"] or ""
|
|
field, expect = m["field"].strip(), m["expect"].strip()
|
|
res = rest_call(name, path, query, post=bool(m["cmd"]))
|
|
if isinstance(res, dict):
|
|
res = [res]
|
|
|
|
if field == "count":
|
|
actual = str(len(res))
|
|
elif not res:
|
|
actual = "<keine Treffer>"
|
|
else:
|
|
actual = str(_field(res[0], field))
|
|
|
|
ok = actual == expect
|
|
print(f" {'OK ' if ok else 'FEHL'} {path}"
|
|
f"{'?' + query if query else ''} {field}={actual}"
|
|
f"{'' if ok else f' (erwartet {expect})'}")
|
|
if not ok:
|
|
failures.append(f"{path} {field}: ist {actual!r}, erwartet {expect!r}")
|
|
|
|
if failures:
|
|
raise SystemExit(
|
|
"\n DEPLOY HAT NICHT GEWIRKT — Plan war 'Completed', das Geraet sagt etwas anderes:\n"
|
|
+ "\n".join(f" - {f}" for f in failures)
|
|
+ "\n Ist-Stand am Geraet pruefen (ein gescheiterter Netmiko-Deploy\n"
|
|
" hinterlaesst Teil-Zustand, TP22-Lehre).")
|
|
print(" alle Zusicherungen halten.")
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("--device", required=True)
|
|
p.add_argument("--commands-file", required=True)
|
|
p.add_argument("--change-id", default="")
|
|
p.add_argument("--change-url", default="")
|
|
p.add_argument("--deploy", action="store_true",
|
|
help="nach dem Anlegen freigeben und ausrollen")
|
|
p.add_argument("--assert", dest="assertions", action="append", metavar="SPEC",
|
|
help="Wirksamkeitspruefung nach dem Deploy, am Geraet per REST. "
|
|
"Format: [cmd:]<pfad>[?query]|<feld-oder-count>=<erwartet>. "
|
|
"Mehrfach angebbar. Bei Funkaenderungen 'monitor' lesen, nie das Profil.")
|
|
p.add_argument("--dry-run", action="store_true",
|
|
help="nur zeigen, was gepusht wuerde — legt keinen Plan an")
|
|
args = p.parse_args()
|
|
|
|
# Kommentar- und Leerzeilen fliegen raus: der Plan-config_set geht 1:1 per
|
|
# netmiko an die RouterOS-Konsole, dort haben Doku-Zeilen nichts verloren.
|
|
# Die kommentierte .rsc bleibt das lesbare Artefakt im Repo.
|
|
raw = pathlib.Path(args.commands_file).read_text()
|
|
commands = "\n".join(l for l in (x.rstrip() for x in raw.splitlines())
|
|
if l.strip() and not l.lstrip().startswith("#"))
|
|
if not commands:
|
|
raise SystemExit("commands-file enthaelt keine Kommandozeilen")
|
|
|
|
print(f" Geraet: {args.device}")
|
|
print(f" Change: {args.change_id or '(keine)'}")
|
|
print(f" Kommandos ({len(commands.splitlines())} Zeilen):")
|
|
for line in commands.splitlines():
|
|
print(f" | {line}")
|
|
|
|
if args.dry_run:
|
|
print("\n DRY-RUN — kein Plan angelegt, nichts gepusht.")
|
|
return
|
|
|
|
device_id = resolve_device(args.device)
|
|
print(f"\n Device-UUID: {device_id}")
|
|
|
|
known = {r["id"] for r in api(f"/api/plugins/golden-config/config-plan/?device_id={device_id}&limit=200")["results"]}
|
|
|
|
print(" -> Generate Config Plans (plan_type=manual)")
|
|
run = api(f"/api/extras/jobs/{GEN_JOB}/run/", "POST", {"data": {
|
|
"device": [device_id],
|
|
"plan_type": "manual",
|
|
"commands": commands,
|
|
"change_control_id": args.change_id,
|
|
"change_control_url": args.change_url,
|
|
}})
|
|
jr = run.get("job_result", run).get("id")
|
|
st = wait_job(jr, "generate")
|
|
job_log(jr)
|
|
if st != "SUCCESS":
|
|
raise SystemExit(f"Plan-Erzeugung fehlgeschlagen: {st}")
|
|
|
|
plans = api(f"/api/plugins/golden-config/config-plan/?device_id={device_id}&limit=200")["results"]
|
|
new = [r for r in plans if r["id"] not in known]
|
|
if len(new) != 1:
|
|
raise SystemExit(f"Erwartet genau 1 neuen Plan, gefunden: {len(new)}")
|
|
plan = new[0]
|
|
print(f" Plan angelegt: {plan['id']}")
|
|
|
|
# Gegenprobe, dass der Plan wirklich an diesem Geraet haengt (Lehre 2).
|
|
plan_dev = api(f"/api/plugins/golden-config/config-plan/{plan['id']}/?depth=1")["device"]
|
|
assert plan_dev["name"] == args.device, f"Plan haengt an {plan_dev['name']}, nicht an {args.device}"
|
|
print(f" Plan.device verifiziert: {plan_dev['name']}")
|
|
|
|
if not args.deploy:
|
|
print("\n Plan liegt bereit (nicht freigegeben). Mit --deploy ausrollen.")
|
|
return
|
|
|
|
print(" -> Plan auf Approved")
|
|
api(f"/api/plugins/golden-config/config-plan/{plan['id']}/", "PATCH",
|
|
{"status": status_id("Approved")})
|
|
|
|
print(" -> Deploy Config Plans")
|
|
run = api(f"/api/extras/jobs/{DEPLOY_JOB}/run/", "POST", {"data": {
|
|
"config_plan": [plan["id"]],
|
|
"fail_job_on_task_failure": True,
|
|
}})
|
|
jr = run.get("job_result", run).get("id")
|
|
st = wait_job(jr, "deploy")
|
|
job_log(jr)
|
|
|
|
# Lehre 1: nicht dem Job glauben, den Plan-Status lesen.
|
|
final = api(f"/api/plugins/golden-config/config-plan/{plan['id']}/?depth=1")
|
|
plan_status = (final.get("status") or {}).get("name")
|
|
print(f"\n Job-Status: {st}")
|
|
print(f" Plan-Status: {plan_status} <- der zaehlt")
|
|
if plan_status != "Completed":
|
|
raise SystemExit(f"Deploy nicht sauber durch (Plan: {plan_status})")
|
|
|
|
# Lehre 3: der Plan-Status sagt nichts ueber die Wirkung.
|
|
if args.assertions:
|
|
run_assertions(args.device, args.assertions)
|
|
print("\n Plan Completed UND am Geraet nachgewiesen.")
|
|
else:
|
|
print("\n Plan Completed. Abnahme trotzdem erst nach Gegenpruefung am Geraet.")
|
|
print(" (Tipp: --assert haengt die Gegenpruefung direkt an den Deploy.)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|