Files
bf-infra/routeros/push-gc-template.py

134 lines
5.1 KiB
Python

#!/usr/bin/env python3
r"""Push nautobot/mikrotik_routeros.j2 to the bf-golden-config Gitea repo.
Reads the Gitea token from ~/.secrets/credentials.env (GITEA_ACCESS_TOKEN).
python3 routeros/push-gc-template.py --dry-run # Diff gegen Remote zeigen
python3 routeros/push-gc-template.py -m "TP22: AP-Zweig" # pushen
SECRET-GUARD (TP22, 2026-08-13)
-------------------------------
Vor jedem PUT wird der Template-Inhalt gegen dieselben Feld-Regexe geprueft, die die
Nautobot-ConfigReplace-Regeln fuers Backup-Scrubbing verwenden. Findet sich ein Wert,
der NICHT die Maske `"<<REDACTED>>"` traegt, bricht das Skript ab.
Grund: genau dieser Weg hat am 2026-07-22 den WG-Hub-private-key und 6 WLAN-PSKs ins
Gitea-Repo getragen (via Backup-Job, aber dieselbe Zielablage). Das Repo musste am
2026-08-13 neu angelegt werden. Der Guard kostet nichts und verhindert die Wiederholung.
"""
import argparse
import base64
import difflib
import json
import pathlib
import re
import subprocess
import sys
import urllib.error
import urllib.request
REPO = "https://gitea.emsenhuber.net/api/v1/repos/admin/bf-golden-config"
FILE = "jinja_templates/mikrotik_routeros.j2"
TOKEN_PATH = pathlib.Path.home() / ".secrets/credentials.env"
SRC = pathlib.Path(__file__).resolve().parent.parent / "nautobot" / "mikrotik_routeros.j2"
MASK = '"<<REDACTED>>"'
# Muss mit nautobot/ingest-gc-scrubbing.py:RULES synchron bleiben.
SECRET_FIELDS = ["passphrase", "private-key", "pre-shared-key", "wpa-pre-shared-key", "password", "secret"]
VAL = r'("(?:[^"\\]|\\.)*"|\S+)'
def get_token():
m = re.search(r"^GITEA_ACCESS_TOKEN=(.+)$", TOKEN_PATH.read_text(), re.MULTILINE)
if not m:
raise SystemExit("GITEA_ACCESS_TOKEN not found")
return m.group(1).strip()
def scan_for_secrets(text):
"""Gibt eine Liste von (zeilennr, feld) zurueck, deren Wert nicht maskiert ist."""
bad = []
for lineno, line in enumerate(text.splitlines(), 1):
# Kommentarzeilen sind Doku (nennen die Feldnamen absichtlich), nicht Config.
if line.lstrip().startswith("#"):
continue
for field in SECRET_FIELDS:
for m in re.finditer(rf"(?<![\w-]){re.escape(field)}=" + VAL, line):
if m.group(1) != MASK:
bad.append((lineno, field))
return bad
def api(path, method="GET", payload=None, token=""):
req = urllib.request.Request(
f"{REPO}/{path}",
data=json.dumps(payload).encode() if payload else None,
method=method,
headers={"Authorization": f"token {token}", "Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read()
return resp.status, (json.loads(raw) if raw else {})
def main():
p = argparse.ArgumentParser()
p.add_argument("-m", "--message", help="Commit-Message (Default: letzter lokaler Commit-Titel)")
p.add_argument("--dry-run", action="store_true", help="nur Diff gegen Remote zeigen")
args = p.parse_args()
content = SRC.read_text()
# --- Secret-Guard: laeuft IMMER, auch im Dry-Run ---
bad = scan_for_secrets(content)
if bad:
print("!! ABBRUCH - unmaskierte Secrets im Template:")
for lineno, field in bad:
print(f" Zeile {lineno}: {field}= traegt nicht die Maske {MASK}")
sys.exit(1)
print(f" Secret-Guard: ok (alle Secret-Felder tragen {MASK})")
token = get_token()
# Remote-Stand holen (sha optional: nach einer Repo-Neuanlage existiert die Datei ggf. nicht)
sha, remote = None, ""
try:
_, meta = api(f"contents/{FILE}?ref=main", token=token)
sha = meta.get("sha")
remote = base64.b64decode(meta.get("content", "")).decode()
except urllib.error.HTTPError as e:
if e.code != 404:
raise
print(f" Remote: {FILE} existiert noch nicht - wird angelegt")
diff = list(difflib.unified_diff(
remote.splitlines(), content.splitlines(),
fromfile=f"remote/{FILE}", tofile=f"local/{SRC.name}", lineterm="", n=2))
if not diff:
print(" Kein Unterschied zum Remote - nichts zu tun.")
return
print(f" Diff: +{sum(1 for l in diff if l.startswith('+') and not l.startswith('+++'))}"
f" / -{sum(1 for l in diff if l.startswith('-') and not l.startswith('---'))} Zeilen")
if args.dry_run:
print("\n".join(diff))
print("\n DRY-RUN - nichts gepusht.")
return
msg = args.message
if not msg:
try:
msg = subprocess.run(["git", "log", "-1", "--pretty=%s"], cwd=SRC.parent.parent,
capture_output=True, text=True, check=True).stdout.strip()
except Exception:
msg = "Update mikrotik_routeros.j2"
payload = {"content": base64.b64encode(content.encode()).decode(), "message": msg, "branch": "main"}
if sha:
payload["sha"] = sha
status, res = api(f"contents/{FILE}", method="PUT" if sha else "POST", payload=payload, token=token)
print(f" HTTP {status} commit={res.get('commit', {}).get('sha', '?')[:8]} msg={msg!r}")
if __name__ == "__main__":
main()