gc-manual-plan: --assert prueft die Wirkung am Geraet, nicht den Plan-Status

"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:]<pfad>[?query]|<feld-oder-count>=<erwartet>, 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 <noreply@anthropic.com>

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:]<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
@@ -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, "<<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:
+        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"^(?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)
@@ -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:]<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()
@@ -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__":