Config-Transaktionen: begin/commit/rollback fuer Testaenderungen

Eine Aenderung, die am Geraet steht, aber nicht in der SoT, ist Drift -- und Drift
sieht bisher immer gleich aus, egal ob er ein Versehen ist oder ein laufender
Versuch. Wer testen wollte, hatte die Wahl zwischen "Compliance kaputt" und "gar
nicht testen".

gc-transaktion.py macht die Absicht sichtbar, ohne die Governance aufzuweichen:
der Change geht weiterhin ausschliesslich ueber Golden Config, die Compliance
meldet den Drift weiterhin -- neu ist, dass `status` ihn zuordnen kann,
"erwartet (offene Transaktion)" statt "unerklaert". Und dass er ein Ablaufdatum
hat.

Drei Regeln erzwingt das Werkzeug: --expires ist Pflicht (eine unbefristete
Ausnahme ist undokumentierter Drift mit besserem Gewissen), der Scope ist eine
ComplianceRule und kein Freibrief (Drift ausserhalb wird ausdruecklich als nicht
abgedeckt gemeldet), und es gibt genau eine offene Transaktion je Geraet -- sonst
ist beim Rollback nicht mehr entscheidbar, wohin zurueck. Der Commit wird
verweigert, solange der Scope noch driftet: dann waere die SoT gar nicht
nachgezogen worden.

Zwei bewusste Grenzen: der Rollback pusht nicht selbst, sondern schreibt einen
Vorschlag zum Pruefen (add-Zeilen aus einem /export legen neu an, statt zu
aendern), und der Snapshot enthaelt nur den Scope-Abschnitt. Der Vollexport
enthielt Firewall, Adressen, Benutzerliste und Skriptquellen -- nichts davon
braucht der Rueckbau, und die Akten liegen im Git.

Erste Akte ist der ether3-Trunk: ap1 haengt testweise am Sideboard auf core
ether3, und dieser Port traegt bisher keines der sieben VLANs. Deploy steht noch
aus -- er wurde in dieser Session vom Berechtigungs-Klassifikator blockiert.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-21 20:15:05 +02:00
parent fcd221388f
commit 3f938e8389
5 changed files with 422 additions and 0 deletions

View File

@@ -0,0 +1,47 @@
# Transaktionen — Teständerungen mit Ablaufdatum
Hier liegen die Akten offener und geschlossener **Config-Transaktionen**. Werkzeug:
[`nautobot/gc-transaktion.py`](../../nautobot/gc-transaktion.py).
## Wozu
Eine Änderung, die am Gerät steht, aber nicht in der SoT, ist **Drift** — und Drift sieht bisher immer
gleich aus, egal ob er ein Versehen ist oder ein laufender Versuch. Wer testen wollte, hatte die Wahl
zwischen „Compliance kaputt" und „gar nicht testen".
Eine Transaktion macht die Absicht sichtbar, ohne die Governance aufzuweichen: **der Change geht
weiterhin ausschließlich über Golden Config** (`gc-manual-plan.py`), und die Compliance meldet den
Drift weiterhin. Neu ist nur, dass `status` ihn **zuordnen** kann.
| SQL | hier |
|---|---|
| `BEGIN` | Snapshot des Scope-Abschnitts + Akte mit Gerät, Scope, Grund und **Ablaufdatum** |
| *uncommitted* | Change ist am Gerät wirksam und im Testumfeld sichtbar, aber nicht in der SoT |
| `COMMIT` | Template/Config-Context nachziehen → Compliance grün → Akte schließen |
| `ROLLBACK` | Rückbau-**Vorschlag** aus dem Snapshot, prüfen, deployen, Akte schließen |
## Drei Regeln, die das Werkzeug erzwingt
1. **`--expires` ist Pflicht.** Eine offene Transaktion ohne Frist ist undokumentierter Drift mit
besserem Gewissen. `status` markiert überfällige Akten.
2. **Der Scope ist eine ComplianceRule**, kein Freibrief. Driftet etwas *außerhalb* des Scopes, meldet
`status` das ausdrücklich als **nicht abgedeckt**.
3. **Eine offene Transaktion je Gerät** — sonst ist beim Rollback nicht mehr entscheidbar, wohin zurück.
## Zwei bewusste Grenzen
- **Der Rollback pusht nicht selbst.** Er schreibt einen Vorschlag aus dem Snapshot; `add`-Zeilen aus
einem `/export` legen neu an, statt zu ändern, und brauchen meist ein `set [find …]`. Ein Rückbau,
den niemand gelesen hat, ist genau die Sorte Automatik, die hier schon zweimal teuer war.
- **Der Snapshot enthält nur den Scope-Abschnitt.** Ein voller `/export` enthält Firewall, Adressen,
Benutzerliste und Skriptquellen — nichts davon wird für den Rückbau gebraucht, und diese Akten liegen
im Git. Für den Ernstfall gibt es das binäre `/system backup` auf dem Gerät (`--device-backup`).
## Dateien je Transaktion
| Datei | Inhalt |
|---|---|
| `<change-id>.json` | die Akte: Gerät, Scope, Grund, Fristen, Status |
| `<change-id>.snapshot.rsc` | der Scope-Abschnitt **vor** der Änderung |
| `<change-id>.commands.rsc` | was deployt wurde |
| `<change-id>.rollback.rsc` | erst nach `rollback` — der Vorschlag zum Prüfen |

View File

@@ -0,0 +1,7 @@
/interface bridge vlan set [find where vlan-ids=10 and !dynamic] tagged=ether1,ether3,ether8,ether13,ether15
/interface bridge vlan set [find where vlan-ids=20 and !dynamic] tagged=ether3,ether8,ether13,ether15
/interface bridge vlan set [find where vlan-ids=30 and !dynamic] tagged=ether1,ether3,ether8,ether15
/interface bridge vlan set [find where vlan-ids=40 and !dynamic] tagged=ether1,ether3,ether8,ether13,ether15
/interface bridge vlan set [find where vlan-ids=50 and !dynamic] tagged=ether1,ether3,ether8,ether15
/interface bridge vlan set [find where vlan-ids=90 and !dynamic] tagged=ether1,ether3,ether8,ether15
/interface bridge vlan set [find where vlan-ids=99 and !dynamic] tagged=ether3,ether8,ether15,bridge

View File

@@ -0,0 +1,14 @@
{
"change_id": "TEST-ether3-trunk-ap1-sideboard",
"device": "crs326-core",
"scope": "bridge-vlan",
"match_config": [
"/interface bridge vlan"
],
"reason": "ap1 testweise am Sideboard, Uplink auf core ether3 statt ether15 (TP25 Standorttest)",
"started": "2026-08-21",
"expires": "2026-09-04",
"state": "open",
"snapshot": "docs/transaktionen/TEST-ether3-trunk-ap1-sideboard.snapshot.rsc",
"device_backup": "vor-ether3-trunk-20260821.backup"
}

View File

@@ -0,0 +1,14 @@
# Snapshot des Scopes 'bridge-vlan' auf crs326-core, 2026-08-21 (Transaktion TEST-ether3-trunk-ap1-sideboard)
# Bewusst NUR dieser Abschnitt — der Vollexport gehoert nicht ins Repo.
/interface bridge vlan
add bridge=bridge untagged=ether1,ether2 vlan-ids=1
add bridge=bridge tagged=ether1,ether8,ether13,ether15 vlan-ids=10
add bridge=bridge tagged=ether8,ether13,ether15 vlan-ids=20
add bridge=bridge tagged=ether1,ether8,ether15 untagged=ether14 vlan-ids=30
add bridge=bridge tagged=ether1,ether8,ether13,ether15 untagged=\
ether2,ether3,ether4,ether5 vlan-ids=40
add bridge=bridge tagged=ether1,ether8,ether15 vlan-ids=50
add bridge=bridge tagged=ether1,ether8,ether15 vlan-ids=90
add bridge=bridge tagged=ether8,ether15,bridge vlan-ids=99
add bridge=bridge tagged=ether1 untagged=ether16 vlan-ids=100

340
nautobot/gc-transaktion.py Normal file
View File

@@ -0,0 +1,340 @@
#!/usr/bin/env python3
r"""Testaenderungen als Transaktion fuehren: begin / status / commit / rollback.
Das Problem, das dieses Skript loest
------------------------------------
Eine Aenderung, die am Geraet steht, aber nicht in der SoT, ist heute schlicht
**Drift** — und Drift sieht immer gleich aus, egal ob er ein Versehen ist oder ein
laufender Versuch. Wer testen will, hat also die Wahl zwischen "Compliance kaputt"
und "gar nicht testen". Beides ist falsch.
Die Uebersetzung von BEGIN/COMMIT/ROLLBACK auf diesen Stack:
BEGIN Snapshot des Geraets + Akte mit Geraet, Scope und ABLAUFDATUM.
Die Aenderung wird ganz normal per gc-manual-plan.py deployt und
ist am Geraet sofort wirksam — nur eben nicht in der SoT.
offen Compliance meldet den Drift weiter, aber `status` ordnet ihn zu:
"erwartet (offene Transaktion)" statt "unerklaert".
COMMIT Erst das Template/den Config-Context nachziehen, dann committen.
Das Skript verweigert den Commit, solange der Scope noch driftet —
denn dann waere die SoT gar nicht nachgezogen worden.
ROLLBACK Erzeugt aus dem Snapshot einen Rueckbau-Vorschlag fuer den Scope.
Bewusst NICHT automatisch gepusht: ein Rueckbau, den niemand
gelesen hat, ist genau die Sorte Automatik, die dieses Projekt
schon zweimal teuer bezahlt hat.
Drei Regeln, fest eingebaut
---------------------------
1. **Ablaufdatum ist Pflicht.** Eine offene Transaktion ohne Frist ist nichts
anderes als undokumentierter Drift, nur mit besserem Gewissen.
2. **Der Scope ist eine ComplianceRule**, kein Freibrief. `--scope bridge-vlan`
heisst: Abweichungen in `/interface bridge vlan` sind erwartet, alles andere
bleibt scharf.
3. **Eine offene Transaktion je Geraet.** Sonst ist beim Rollback nicht mehr
entscheidbar, wohin zurueck.
Aufruf
------
python3 nautobot/gc-transaktion.py begin --device crs326-core \
--scope bridge-vlan --change-id TEST-ether3 --expires 2026-08-28 \
--reason "ap1 testweise am Sideboard, Uplink auf ether3"
python3 nautobot/gc-transaktion.py status
python3 nautobot/gc-transaktion.py commit --change-id TEST-ether3
python3 nautobot/gc-transaktion.py rollback --change-id TEST-ether3
"""
import argparse
import datetime
import json
import os
import re
import subprocess
import sys
import urllib.request
HIER = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.dirname(HIER)
AKTEN = os.path.join(REPO, "docs", "transaktionen")
STACK = os.path.join(REPO, "docs", "nas-vlan20", "stacks", "210-nautobot.yml")
NB = os.environ.get("NAUTOBOT_URL", "http://10.40.20.30:8080")
PW_ENV = {"crs326-core": "CRS326_CORE_PASS", "crs326-branch": "CRS326_BRANCH_PASS",
"ap1": "ROUTER_AP1_PASS", "ap2": "ROUTER_AP2_PASS"}
IPS = {"crs326-core": "10.40.99.1", "crs326-branch": "10.40.99.2",
"ap1": "10.40.99.11", "ap2": "10.40.99.12"}
# --------------------------------------------------------------------------- Nautobot
def token():
m = re.search(r"NAUTOBOT_SUPERUSER_API_TOKEN=([A-Za-z0-9]+)", open(STACK).read())
if not m:
sys.exit("Token nicht in %s gefunden." % STACK)
return m.group(1)
def nb_get(pfad):
req = urllib.request.Request(NB + pfad, headers={
"Authorization": "Token " + token(), "Accept": "application/json"})
return json.load(urllib.request.urlopen(req, timeout=30))
def regeln():
"""ComplianceRule-Name -> match_config (die Abschnitte, die sie prueft)."""
d = nb_get("/api/plugins/golden-config/compliance-rule/?limit=100&depth=1")
out = {}
for r in d["results"]:
feat = r.get("feature") or {}
name = feat.get("name") or feat.get("display")
out[name] = [z.strip() for z in (r.get("match_config") or "").splitlines() if z.strip()]
return out
def compliance(device):
"""Feature -> compliant? fuer ein Geraet."""
d = nb_get("/api/plugins/golden-config/config-compliance/?limit=200&depth=1"
"&device=%s" % device)
out = {}
for r in d["results"]:
feat = r.get("rule") or {}
feat = (feat.get("feature") or {}) if isinstance(feat, dict) else {}
name = feat.get("name") or feat.get("display") or r.get("rule")
out[str(name)] = bool(r.get("compliance"))
return out
# --------------------------------------------------------------------------- Geraet
def geheim(device):
env = PW_ENV.get(device)
if not env:
sys.exit("Kein Passwort-Env fuer %r bekannt." % device)
wert = os.environ.get(env)
if wert:
return wert
pfad = os.path.expanduser("~/.secrets/credentials.env")
m = re.search(r"^%s=(.*)$" % env, open(pfad).read(), re.M)
if not m:
sys.exit("$%s weder im Environment noch in %s." % (env, pfad))
return m.group(1).strip().strip("\"'")
def export(device):
"""/export vom Geraet holen — der Snapshot, gegen den spaeter zurueckgebaut wird."""
ip = IPS.get(device) or sys.exit("Keine IP fuer %r." % device)
p = subprocess.run(["sshpass", "-p", geheim(device), "ssh",
"-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=10",
"admin@" + ip, "/export"],
capture_output=True, text=True, timeout=120)
if p.returncode != 0:
sys.exit("Export fehlgeschlagen (rc=%d). %s" % (p.returncode, p.stderr[:200]))
return p.stdout
def abschnitt(text, praefixe):
"""Die Zeilen aus einem /export, die zu den match_config-Praefixen gehoeren.
RouterOS-Exporte schreiben die Sektion einmal als Kopfzeile und darunter die
Eintraege ohne Pfad — beides muss mit.
"""
raus, aktiv = [], False
for zeile in text.splitlines():
s = zeile.strip()
if s.startswith("/"):
aktiv = any(s.startswith(p) for p in praefixe)
if aktiv:
raus.append(s)
continue
if aktiv and s and not s.startswith("#"):
raus.append(s)
return raus
# --------------------------------------------------------------------------- Akten
def akte_pfad(change_id):
return os.path.join(AKTEN, "%s.json" % change_id)
def akten():
if not os.path.isdir(AKTEN):
return []
out = []
for n in sorted(os.listdir(AKTEN)):
if n.endswith(".json"):
out.append(json.load(open(os.path.join(AKTEN, n))))
return out
def schreibe(akte):
os.makedirs(AKTEN, exist_ok=True)
with open(akte_pfad(akte["change_id"]), "w") as fh:
json.dump(akte, fh, indent=2, ensure_ascii=False)
fh.write("\n")
def heute():
return datetime.date.today().isoformat()
# --------------------------------------------------------------------------- Befehle
def cmd_begin(a):
if os.path.exists(akte_pfad(a.change_id)):
sys.exit("Es gibt bereits eine Akte %s." % a.change_id)
offen = [t for t in akten() if t["state"] == "open" and t["device"] == a.device]
if offen:
sys.exit("Auf %s ist bereits %r offen — erst commit oder rollback.\n"
"(Regel 3: eine offene Transaktion je Geraet.)"
% (a.device, offen[0]["change_id"]))
verfuegbar = regeln()
if a.scope not in verfuegbar:
sys.exit("Scope %r ist keine ComplianceRule. Bekannt sind:\n %s"
% (a.scope, "\n ".join(sorted(verfuegbar))))
if a.expires <= heute():
sys.exit("--expires muss in der Zukunft liegen (Regel 1).")
os.makedirs(AKTEN, exist_ok=True)
snap = os.path.join(AKTEN, "%s.snapshot.rsc" % a.change_id)
# 🔴 NUR den Scope-Abschnitt sichern, nicht den ganzen /export. Der Vollexport
# enthaelt Firewall, Adressen, Benutzerliste und Skriptquellen — nichts davon
# wird fuer den Rueckbau gebraucht, und die Akten liegen im Git. Fuer den
# Ernstfall gibt es das binaere /system backup auf dem Geraet (--device-backup).
zeilen = abschnitt(export(a.device), verfuegbar[a.scope])
with open(snap, "w") as fh:
fh.write("# Snapshot des Scopes %r auf %s, %s (Transaktion %s)\n"
"# Bewusst NUR dieser Abschnitt — der Vollexport gehoert nicht ins Repo.\n\n"
% (a.scope, a.device, heute(), a.change_id))
fh.write("\n".join(zeilen) + "\n")
akte = {
"change_id": a.change_id, "device": a.device, "scope": a.scope,
"match_config": verfuegbar[a.scope], "reason": a.reason,
"started": heute(), "expires": a.expires, "state": "open",
"snapshot": os.path.relpath(snap, REPO),
"device_backup": a.device_backup or "",
}
schreibe(akte)
print("BEGIN %s auf %s — Scope %s, laeuft ab am %s"
% (a.change_id, a.device, a.scope, a.expires))
print(" Snapshot: %s" % akte["snapshot"])
print("\nJetzt die Aenderung wie gewohnt deployen:")
print(" python3 nautobot/gc-manual-plan.py --device %s --commands-file <datei.rsc> \\"
% a.device)
print(" --change-id %s --deploy --assert ..." % a.change_id)
def cmd_status(a):
ts = akten()
if not ts:
print("Keine Transaktionen.")
return
for t in ts:
marke = {"open": "OFFEN", "committed": "committed", "rolledback": "zurueckgebaut"}[t["state"]]
ueberfaellig = t["state"] == "open" and t["expires"] < heute()
print("%-34s %-12s %-14s Scope %s%s"
% (t["change_id"], marke, t["device"], t["scope"],
" ⚠️ ABGELAUFEN am " + t["expires"] if ueberfaellig else ""))
if t.get("reason"):
print(" %s" % t["reason"])
if t["state"] != "open":
continue
try:
c = compliance(t["device"])
except Exception as e:
print(" (Compliance nicht lesbar: %s)" % e)
continue
drifts = [f for f, ok in c.items() if not ok]
erwartet = [f for f in drifts if f == t["scope"]]
fremd = [f for f in drifts if f != t["scope"]]
print(" Drift im Scope: %s" % (", ".join(erwartet) or "keiner — SoT und Geraet gleich"))
if fremd:
print(" ⚠️ Drift AUSSERHALB des Scopes: %s" % ", ".join(fremd))
print(" Das deckt diese Transaktion NICHT ab.")
def cmd_commit(a):
pfad = akte_pfad(a.change_id)
if not os.path.exists(pfad):
sys.exit("Keine Akte %s." % a.change_id)
t = json.load(open(pfad))
if t["state"] != "open":
sys.exit("%s ist bereits %s." % (a.change_id, t["state"]))
c = compliance(t["device"])
if not c.get(t["scope"], True):
sys.exit("COMMIT verweigert: %s driftet auf %s noch.\n"
"Das heisst, die SoT ist nicht nachgezogen — erst Template/Config-Context\n"
"aendern, dann `gc-configplan.py --refresh-compliance`, dann committen."
% (t["scope"], t["device"]))
t["state"] = "committed"
t["closed"] = heute()
schreibe(t)
print("COMMIT %s — %s ist auf %s compliant, die Aenderung steht jetzt in der SoT."
% (a.change_id, t["scope"], t["device"]))
def cmd_rollback(a):
pfad = akte_pfad(a.change_id)
if not os.path.exists(pfad):
sys.exit("Keine Akte %s." % a.change_id)
t = json.load(open(pfad))
if t["state"] != "open":
sys.exit("%s ist bereits %s." % (a.change_id, t["state"]))
alt = abschnitt(open(os.path.join(REPO, t["snapshot"])).read(), t["match_config"])
neu = abschnitt(export(t["device"]), t["match_config"])
ziel = os.path.join(AKTEN, "%s.rollback.rsc" % a.change_id)
with open(ziel, "w") as fh:
fh.write("# Rueckbau-VORSCHLAG fuer %s (%s, Scope %s)\n"
"# Aus dem Snapshot vom %s. NICHT ungeprueft deployen:\n"
"# `add`-Zeilen legen neu an, statt zu aendern — was hier gebraucht wird,\n"
"# ist in der Regel ein `set [find ...]`. Zeilen entsprechend anpassen.\n\n"
% (a.change_id, t["device"], t["scope"], t["started"]))
fh.write("\n".join(alt) + "\n")
print("Snapshot (%d Zeilen) gegen Ist-Stand (%d Zeilen) im Scope %s:\n"
% (len(alt), len(neu), t["scope"]))
for z in sorted(set(neu) - set(alt)):
print(" JETZT NUR AM GERAET %s" % z)
for z in sorted(set(alt) - set(neu)):
print(" IM SNAPSHOT, FEHLT %s" % z)
print("\nVorschlag geschrieben: %s" % os.path.relpath(ziel, REPO))
print("Pruefen, anpassen, dann deployen:")
print(" python3 nautobot/gc-manual-plan.py --device %s --commands-file %s \\"
% (t["device"], os.path.relpath(ziel, REPO)))
print(" --change-id %s-rollback --deploy --assert ..." % a.change_id)
print("\nDanach die Akte schliessen: %s rollback --change-id %s --erledigt"
% (os.path.basename(__file__), a.change_id))
if a.erledigt:
t["state"] = "rolledback"
t["closed"] = heute()
schreibe(t)
print("\nAkte %s geschlossen (zurueckgebaut)." % a.change_id)
def main():
p = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = p.add_subparsers(dest="cmd", required=True)
b = sub.add_parser("begin", help="Transaktion oeffnen (Snapshot + Akte)")
b.add_argument("--device", required=True)
b.add_argument("--scope", required=True, help="Name einer ComplianceRule, z.B. bridge-vlan")
b.add_argument("--change-id", required=True)
b.add_argument("--expires", required=True, help="YYYY-MM-DD, Pflicht")
b.add_argument("--reason", default="")
b.add_argument("--device-backup", default="", help="Name eines vorher angelegten /system backup")
b.set_defaults(func=cmd_begin)
s = sub.add_parser("status", help="offene Transaktionen und ihre Drifts")
s.set_defaults(func=cmd_status)
c = sub.add_parser("commit", help="Aenderung in die SoT uebernommen -> Akte schliessen")
c.add_argument("--change-id", required=True)
c.set_defaults(func=cmd_commit)
r = sub.add_parser("rollback", help="Rueckbau-Vorschlag aus dem Snapshot erzeugen")
r.add_argument("--change-id", required=True)
r.add_argument("--erledigt", action="store_true", help="Akte schliessen (nach dem Deploy)")
r.set_defaults(func=cmd_rollback)
a = p.parse_args()
a.func(a)
if __name__ == "__main__":
main()