Christoph Emsenhuber
177ba100eb
Template mikrotik_routeros.j2: ether23 auf core+branch als OOB-Port (pvid=99, admit-only-untagged-and-priority-tagged) + VLAN99 untagged=ether23. Branch bekommt einen deny-by-default-Input-Filter (8 Regeln, src-Rahmen spiegelt /ip service available-from), Core eine Regel "OOB->core mgmt". Neue Skripte: ingest-tp38-oob.py (seedet firewall_filter in beiden local_config_context_data) und der gc-configplan.py-Job tp38-oob (Deploy). ConfigPlans angelegt+approved, Deploy steht noch aus. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
106 lines
4.2 KiB
Python
106 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
r"""TP38: OOB-Port + Input-Firewall als Daten nach Nautobot schreiben.
|
|
|
|
Zwei getrennte Aenderungen, ein Skript:
|
|
|
|
1. BRANCH (crs326-branch): `firewall_filter` im local_config_context_data ANLEGEN
|
|
(8 Input-Regeln, deny-by-default). Der src-Rahmen 10.40.0.0/16 + 100.96.0.0/12
|
|
spiegelt bewusst das `available-from` der /ip service-Zeilen — kein neues
|
|
Fenster, nur stateful + deny-by-default + WOULD-DROP-Log.
|
|
2. CORE (crs326-core): EINE Regel "TP38: OOB->core mgmt" in die bestehende
|
|
firewall_filter-Liste EINFUEGEN (vor log/drop), damit der OOB-Port (untagged
|
|
ether23, VLAN99) auch den Core selbst verwalten kann.
|
|
|
|
Attributreihenfolge = alphabetisch, exakt wie RouterOS `/export` sie schreibt
|
|
(gleiche Konvention wie ingest-core-firewall.py).
|
|
|
|
python3 nautobot/ingest-tp38-oob.py # Dry-Run
|
|
python3 nautobot/ingest-tp38-oob.py --apply
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
|
|
NB = os.environ.get("NAUTOBOT_URL", "http://10.40.20.30:8080")
|
|
|
|
BRANCH_RULES = [
|
|
'add action=accept chain=input comment="TP38: established" connection-state=established,related,untracked',
|
|
'add action=drop chain=input comment="TP38: invalid" connection-state=invalid',
|
|
'add action=accept chain=input comment="TP38: icmp" protocol=icmp',
|
|
'add action=accept chain=input comment="TP38: mndp" dst-port=5678 protocol=udp',
|
|
'add action=accept chain=input comment="TP38: mgmt LAN" dst-port=22,80,8291,8728 protocol=tcp src-address=10.40.0.0/16',
|
|
'add action=accept chain=input comment="TP38: mgmt mesh" dst-port=22,80,8291,8728 protocol=tcp src-address=100.96.0.0/12',
|
|
'add action=log chain=input comment="TP38: input log-only" log-prefix=TP38-BRANCH-WOULD-DROP',
|
|
'add action=drop chain=input comment="TP38: input deny-by-default"',
|
|
]
|
|
|
|
CORE_NEW_RULE = 'add action=accept chain=input comment="TP38: OOB->core mgmt" dst-port=22,80,8291,8728 in-interface=vlan99 protocol=tcp'
|
|
|
|
|
|
def nb(path, method="GET", payload=None):
|
|
tok = os.environ.get("NAUTOBOT_TOKEN") or sys.exit("NAUTOBOT_TOKEN nicht gesetzt")
|
|
req = urllib.request.Request(
|
|
f"{NB}{path}", method=method,
|
|
data=json.dumps(payload).encode() if payload is not None else None,
|
|
headers={"Authorization": f"Token {tok}", "Content-Type": "application/json",
|
|
"Accept": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
raw = r.read()
|
|
return json.loads(raw) if raw else {}
|
|
|
|
|
|
def get_device(name):
|
|
return nb(f"/api/dcim/devices/?name={name}&depth=1")["results"][0]
|
|
|
|
|
|
def patch_device(dev, ctx, label):
|
|
dev_id = dev["id"]
|
|
nb(f"/api/dcim/devices/{dev_id}/", "PATCH", {"local_config_context_data": ctx})
|
|
print(f" ~ {label} geschrieben.")
|
|
|
|
|
|
def main():
|
|
apply_ = "--apply" in sys.argv
|
|
|
|
# --- BRANCH ---
|
|
branch = get_device("crs326-branch")
|
|
bctx = dict(branch.get("local_config_context_data") or {})
|
|
if bctx.get("firewall_filter") == BRANCH_RULES:
|
|
print("== branch: firewall_filter unveraendert ==")
|
|
else:
|
|
bctx["firewall_filter"] = BRANCH_RULES
|
|
print(f"== branch: {len(BRANCH_RULES)} Input-Regeln ==")
|
|
for ln in BRANCH_RULES:
|
|
print(" " + ln)
|
|
if apply_:
|
|
patch_device(branch, bctx, "branch firewall_filter")
|
|
|
|
# --- CORE ---
|
|
core = get_device("crs326-core")
|
|
cctx = dict(core.get("local_config_context_data") or {})
|
|
ff = list(cctx.get("firewall_filter") or [])
|
|
if CORE_NEW_RULE in ff:
|
|
print("\n== core: Regel schon vorhanden ==")
|
|
else:
|
|
# vor der Log-Regel einsortieren (die sitzt direkt vor dem finalen drop).
|
|
insert_at = len(ff)
|
|
for i, ln in enumerate(ff):
|
|
if "log-prefix=TP5-INPUT-WOULD-DROP" in ln:
|
|
insert_at = i
|
|
break
|
|
ff.insert(insert_at, CORE_NEW_RULE)
|
|
cctx["firewall_filter"] = ff
|
|
print(f"\n== core: fuege OOB-Regel an Position {insert_at} ein ==")
|
|
print(" " + CORE_NEW_RULE)
|
|
if apply_:
|
|
patch_device(core, cctx, "core firewall_filter")
|
|
|
|
if not apply_:
|
|
print("\nDRY-RUN — mit --apply schreiben. Danach: "
|
|
"push-gc-template.py + gc-configplan.py --refresh-compliance")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|