From 2f102b48de0c61e240ed48c77ddec2693ab699a8 Mon Sep 17 00:00:00 2001 From: Christoph Emsenhuber Date: Mon, 17 Aug 2026 15:44:13 +0200 Subject: [PATCH] gc-manual-plan: --assert prueft die Wirkung am Geraet, nicht den Plan-Status MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit "Plan-Status: Completed" sagt nur, dass die Zeilen abgesetzt wurden. Am 2026-08-17 blieb ein `/interface/wifi/set ... !channel` wirkungslos ohne zu werfen, und das Selbstprotokoll meldete trotzdem Erfolg — weil es das Profil las, waehrend der Interface-Inline-Override gewann. Die Abnahme haengt seither daran, dass jemand daran denkt nachzusehen. Das ist kein Mechanismus. --assert prueft nach dem Deploy per RouterOS-REST am Geraet und bricht laut ab: --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:][?query]|=, mehrfach angebbar. GET fuers Lesen, cmd: fuer POST-Kommandos wie monitor. Trenner ist |, nicht :, weil MAC-Adressen in Werten Doppelpunkte haben. Dokumentierte Regel dazu: bei Funkaenderungen muss die Assertion monitor lesen, NIE das Profil — print detail zeigt aufgeloeste Werte, export die rohen, nur monitor das, was auf der Luft ist. Genau diese Verwechslung war der Fehler. Mgmt-IP kommt aus Nautobot (primary_ip4) statt hartkodiert. Passwoerter aus $ENV, sonst aus ~/.secrets/credentials.env — mit eigenem KEY=VALUE-Parser, weil die Datei WireGuard-[Interface]-Bloecke enthaelt, an denen `set -a; . datei` in zsh still abbricht und alle Zeilen danach verschluckt. Fehlerausgaben redacten das Passwort. Getestet: count-GET, field-GET mit Query, monitor-POST (inkl. tx-power=18) und der Fehlerfall, der korrekt abbricht. Co-Authored-By: Claude Opus 5 --- nautobot/gc-manual-plan.py | 186 ++++++++++++++++++++++++++++++++++++- 1 file changed, 185 insertions(+), 1 deletion(-) diff --git a/nautobot/gc-manual-plan.py b/nautobot/gc-manual-plan.py index 665694e..c226a82 100644 --- a/nautobot/gc-manual-plan.py +++ b/nautobot/gc-manual-plan.py @@ -22,11 +22,39 @@ ZWEI TP22-LEHREN SIND HIER EINGEBAUT 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:][?query]|=`` — 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 @@ -104,6 +132,151 @@ def job_log(job_result_id, levels=("error", "warning", "failure")): 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, "<>") 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: + url += f"?{query}" + 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"^(?Pcmd:)?(?P[^|?]+)(?:\?(?P[^|]*))?\|" + r"(?P[^=]+)=(?P.*)$") + + +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:][?query]|=") + 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 = "" + 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) @@ -112,6 +285,10 @@ def main(): 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:][?query]|=. " + "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() @@ -190,7 +367,14 @@ def main(): print(f" Plan-Status: {plan_status} <- der zaehlt") if plan_status != "Completed": raise SystemExit(f"Deploy nicht sauber durch (Plan: {plan_status})") - print("\n Plan Completed. Abnahme trotzdem erst nach Gegenpruefung am Geraet.") + + # 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__":