#!/usr/bin/env python3 # -*- coding: utf-8 -*- # ############################################################################# # TOOL : FRITZ!Box Enterprise Monitor (Lua-Based) # VERSION : 1.2.0 # AUTHOR : [Dein Name/Kürzel] # CONTACT : [Deine E-Mail / Abteilung] # REPOSITORY : https://github.com[DeinAccount]/[Projekt] # # DOCUMENTATION: # - WIKI : https://wiki.deinefirma.internal # - ISMS : Asset-ID [z.B. FB-MON-2026] / Risiko-Klasse: Medium # - MONITORING: Graylog-Stream "FritzBox-Satelliten" # # # ISMS CONTEXT: # - ASSET-ID : HOME-NET-MON-01 # - CLASS : Internal / Private Setup # - WIKI : https://dein-privates-wiki.internal # # HISTORY: # - 2026-02-04: v1.2.0 - Refactoring auf Lua-API, JSON-Support & Persistenz # - REVISION : Quarterly (Next: 2026-05-04) # - Created : 2026-02-04 # - Last Rev. : 2026-02-04 (Initial Release nach Provider-Wartung Hamburg) # - Next Rev. : 2026-08-04 (Halbjährlicher Security-Audit & API-Check) # # DESCRIPTION: # Extrahiert via Lua-Session das System-Log und VPN-Statusdaten aus # FRITZ!Boxen und sendet diese strukturiert (JSON) an einen Syslog-Server. # ############################################################################# import sys import os import time import requests import hashlib import logging import logging.handlers import json import re import signal import xml.etree.ElementTree as ET from datetime import datetime from dotenv import load_dotenv VER="8.55_rc3 mit json, mixed, s2s, caching, errorlevels-spezial" # Initialisierung print(f"SKRIPT-START: Initialisiere Fritz-Enterprise-Monitor v {VER} :...", flush=True) load_dotenv() # Konfiguration aus .env FB_IP = os.getenv("FRITZ_IP") FB_USER = os.getenv("FRITZ_USER") FB_PWD = os.getenv("FRITZ_PWD") SENDER_NAME = os.getenv("SYSLOG_SENDER_NAME", "fritzbox") SYSLOG_SERVER = os.getenv("SYSLOG_SERVER") SYSLOG_PORT = int(os.getenv("SYSLOG_PORT", 514)) INTERVAL = int(os.getenv("CHECK_INTERVAL", 60)) # Schalter DEBUG_MODE = os.getenv("DEBUG_MODE", "False").lower() == "true" SEND_INITIAL = os.getenv("SEND_INITIAL_LOG", "True").lower() == "true" EXTRACT_TIME = os.getenv("SYSLOG_EXTRACT_TIME", "True").lower() == "true" MIXED_JSON = os.getenv("SYSLOG_MIXED_JSON", "False").lower() == "true" JSON_ONLY = os.getenv("SYSLOG_JSON_ONLY", "False").lower() == "true" SEPARATOR = os.getenv("JSON_SEPARATOR", " -_- ") # Persistenz-Konfiguration HASH_PERSISTENT = os.getenv("HASH_PERSISTENT", "False").lower() == "true" HASH_FILE = os.getenv("HASH_FILE", "/data/hashes.db.txt") HASH_LIMIT = int(os.getenv("HASH_LIMIT", 2000)) HASH_RETAIN = int(os.getenv("HASH_RETAIN", 1000)) last_hashes = set() def save_hashes(): """ Speichert das Hash-Set in eine Datei """ if not HASH_PERSISTENT: return try: os.makedirs(os.path.dirname(HASH_FILE), exist_ok=True) with open(HASH_FILE, "w") as f: for h in last_hashes: f.write(f"{h}\n") if DEBUG_MODE: print(f"[DEBUG] {len(last_hashes)} Hashes persistent gespeichert.", flush=True) except Exception as e: print(f"!!! Fehler beim Speichern der Hashes: {e}", flush=True) def load_hashes(): """ Lädt Hashes beim Start """ global last_hashes if not HASH_PERSISTENT or not os.path.exists(HASH_FILE): return try: with open(HASH_FILE, "r") as f: lines = f.read().splitlines() last_hashes = set(lines) print(f"[*] {len(last_hashes)} Hashes aus Persistenz geladen.", flush=True) except Exception as e: print(f"!!! Fehler beim Laden der Hashes: {e}", flush=True) def signal_handler(sig, frame): """ Fängt Docker Stop Signale ab """ print(f"[*] Signal {sig} empfangen. Beende sicher...", flush=True) save_hashes() sys.exit(0) # Signale registrieren (für sauberes Docker Down) signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGINT, signal_handler) # Syslog Handler handler = logging.handlers.SysLogHandler(address=(SYSLOG_SERVER, SYSLOG_PORT)) class VPNErrorLookup: """ Übersetzt kryptische Fritzbox VPN Hex-Codes in Klartext """ # ERRORS = { # "0x203e": "IKE-Error: Timeout. Gegenstelle antwortet nicht (DNS prüfen oder Host down).", # "0x2027": "IKE-Error: Connection refused. Gegenstelle lehnt Verbindung aktiv ab (IP-Mismatch?).", # "0x201c": "IKE-Error: Authentisierungsfehler. Shared Secret (PSK) prüfen!", # "0x2026": "IKE-Error: Remote-ID mismatch. Identität passt nicht zur Konfiguration.", # "0x2020": "IKE-Error: Proposal mismatch. Verschlüsselungs-SAs passen nicht zusammen.", # "0x2028": "IKE-Error: Initial-Contact fehlgeschlagen (Gegenstelle instabil).", # "0x2033": "IKE-Error: Lifetime-Timeout (Re-Keying fehlgeschlagen).", # "0x1": "VPN-Fehler: Interner Fehler oder DNS-Auflösung fehlgeschlagen.", # "3 IKE server": "VPN-Trennung durch lokalen Server (z.B. Reconnect/Zwangstrennung)." # } ERRORS = { # Verbindungsprobleme "0x203e": "IKE-Error: Timeout. Gegenstelle antwortet nicht (DNS prüfen oder Host down).", "0x2027": "IKE-Error: Connection refused. Gegenstelle lehnt Verbindung aktiv ab.", "0x1": "VPN-Fehler: DNS-Fehler oder interner Fehler beim Verbindungsaufbau.", "3 IKE server": "VPN-Trennung durch lokalen Server (z.B. Zwangstrennung/Reconnect).", # Authentifizierung & Schlüssel "0x201c": "IKE-Error: Authentisierungsfehler. Shared Secret (PSK) prüfen!", "0x2026": "IKE-Error: Remote-ID mismatch. Identität der Gegenstelle passt nicht zur Konfiguration.", # Protokoll- & Konfigurationskonflikte "0x2020": "IKE-Error: Proposal mismatch. Verschlüsselungs-Algorithmen (IKE/IPsec SAs) passen nicht zusammen.", "0x2005": "IKE-Error: Keine virtuelle IP-Adresse für VPN-Nutzer festlegbar.", "0x203d": "IKE-Error: Peer-Authentifizierungs-Problem (Zertifikat oder ID fehlerhaft).", "0x200A": "IKE-Error: Invalid exchange type (Protokollfehler).", "0x200D": "IKE-Error: Invalid minor version.", "0x200E": "IKE-Error: Invalid protocol version.", "0x000F": "IKE-Error: Payloads not encrypted.", "0x0010": "IKE-Error: Payloads are encrypted (obwohl unverschlüsselt erwartet).", "0x2028": "IKE-Error: Initial-Contact fehlgeschlagen (Gegenstelle evtl. instabil).", "0x2033": "IKE-Error: Lifetime-Timeout (Re-Keying fehlgeschlagen)." } @classmethod def translate(cls, msg): for code, description in cls.ERRORS.items(): if code.lower() in msg.lower(): return f"{msg} | INFO: {description}" return msg def get_meta(msg): """ Kategorisierung und Severity-Mapping """ m = msg.lower() cat = "SYSTEM" if any(x in m for x in ["anmeldung", "zugang", "benutzeroberfläche"]): cat = "LOGIN" #elif any(x in m for x in ["dyndns", "wlan", "funknetz"]): cat = "DynDNS" elif any(x in m for x in ["ghz", "wlan", "funknetz"]): cat = "WLAN" elif any(x in m for x in ["vpn", "ike", "ipsec", "wireguard"]): cat = "VPN" elif any(x in m for x in ["dsl", "internetverbindung", "synchronisierung", "breitband"]): cat = "UPLINK" elif any(x in m for x in ["einstellung", "konfiguration", "geändert"]): cat = "CONFIG" elif "dyndns" in m: cat = "DynDNS" # Severity Mapping (RFC 5424) sev, lvl_name = 6, "INFO" ## Kritisch: VPN-Fehler während Site-to-Site Phase ## noch etwas zu hart gefasst #if "vpn" in m and any(x in m for x in ["0x", "fehler", "error", "ike"]): # sev, lvl_name = 2, "CRITICAL" #elif any(x in m for x in ["fehler", "error", "failed", "abbruch", "gescheitert"]): # sev, lvl_name = 3, "ERROR" #elif (any(x in m for x in ["warnung", "warning", "ursache", "seit"]) and "[" in m): # sev, lvl_name = 4, "WARNING" # # hier eine entschärftere version gegen false positive bei VPN # KRITISCH: VPN-Fehler, aber NUR wenn NICHT "erfolgreich" im Text steht # Das verhindert, dass der Tunnel-Aufbau als Fehler markiert wird. if "vpn" in m and any(x in m for x in ["0x", "fehler", "error", "ike"]): if "erfolgreich" not in m: sev, lvl_name = 2, "CRITICAL" else: sev, lvl_name = 6, "INFO" # Erfolgreicher Aufbau ist nur eine Info elif any(x in m for x in ["fehler", "error", "failed", "abbruch", "gescheitert"]): sev, lvl_name = 3, "ERROR" elif any(x in m for x in ["warnung", "warning", "ursache", "seit"]) and "[" in m: sev, lvl_name = 4, "WARNING" # DynDNS-Logik (Dein neues Beispiel) elif "dyndns" in m: # Fehlerszenarien: "nicht aufgelöst werden" oder klassische Fehler-Keywords if any(x in m for x in ["fehler", "error", "nicht aufgelöst", "deaktiviert", "gescheitert"]): sev, lvl_name = 3, "ERROR" elif "erfolgreich" in m: sev, lvl_name = 6, "INFO" else: sev, lvl_name = 4, "WARNING" # Fallback für unklare Meldungen # Allgemeine Fehlerprüfung für andere Kategorien elif any(x in m for x in ["fehler", "error", "failed", "abbruch", "gescheitert"]): sev, lvl_name = 3, "ERROR" elif any(x in m for x in ["warnung", "warning", "ursache", "seit"]) and "[" in m: sev, lvl_name = 4, "WARNING" # ausgabe der ergebnisse return cat, sev, lvl_name def send_payload(msg, date_str=None, time_str=None, force_cat=None): """ Baut das finale Paket und sendet an Syslog """ enriched_msg = VPNErrorLookup.translate(msg) category, severity, lvl_name = get_meta(enriched_msg) if force_cat: category = force_cat # Zeitstempel if EXTRACT_TIME and date_str and time_str: try: dt = datetime.strptime(f"{date_str} {time_str}", "%d.%m.%y %H:%M:%S") ts = dt.strftime("%b %d %H:%M:%S") except: ts = datetime.now().strftime("%b %d %H:%M:%S") else: ts = datetime.now().strftime("%b %d %H:%M:%S") # Basis JSON Objekt log_obj = { "timestamp_raw": f"{date_str} {time_str}" if date_str else ts, "hostname": SENDER_NAME, "category": category, "level": lvl_name, "message": enriched_msg.strip() } # Erweiterte VPN-Metadaten if category == "VPN": p_ip, p_name = "0.0.0.0", "unknown" ip_match = re.search(r"\[(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\]", msg) if ip_match: p_ip = ip_match.group(1) name_match = re.search(r"zu\s+([^\s\[]+)", msg) if name_match: p_name = name_match.group(1).strip(',').strip() log_obj.update({ "vpn_peer_ip": p_ip, "vpn_peer_name": p_name, "vpn_error_code": re.search(r"0x[0-9a-fA-F]+", msg).group(0) if "0x" in msg else "None", "vpn_phase": "Phase 1 (IKE)" if "IKE SA" in msg else "Phase 2 (IPsec)" }) # Paketbau pri = (1 * 8) + severity header = f"<{pri}>{ts} {SENDER_NAME} FRITZ_{category}: " if JSON_ONLY: packet = f"{header}{json.dumps(log_obj)}" elif MIXED_JSON: packet = f"{header}{enriched_msg.strip()}{SEPARATOR}{json.dumps(log_obj)}" else: packet = f"{header}{enriched_msg.strip()}" if DEBUG_MODE: print(f"[DEBUG] OUT: {packet}", flush=True) try: handler.socket.sendto(packet.encode('utf-8'), (SYSLOG_SERVER, SYSLOG_PORT)) except Exception as e: print(f"!!! Sendefehler: {e}", flush=True) class FritzLuaSession: def __init__(self, host, user, pwd): self.host, self.user, self.pwd, self.sid = host, user, pwd, "0000000000000000" def get_sid(self): url = f"http://{self.host}/login_sid.lua" try: r = requests.get(url, timeout=5) challenge = ET.fromstring(r.text).find('Challenge').text response = f"{challenge}-{hashlib.md5(f'{challenge}-{self.pwd}'.encode('utf-16-le')).hexdigest()}" r = requests.get(f"{url}?username={self.user}&response={response}", timeout=5) self.sid = ET.fromstring(r.text).find('SID').text if DEBUG_MODE: print(f"[DEBUG] Login OK, SID: {self.sid}", flush=True) return self.sid except Exception as e: print(f"!!! Login-Fehler: {e}", flush=True); return "0000000000000000" def fetch_data(self, page): if self.sid == "0000000000000000": self.get_sid() try: r = requests.post(f"http://{self.host}/data.lua", data={"sid": self.sid, "page": page}, timeout=10) d = r.json() if 'data' not in d and 'sid' in d: self.get_sid(); return None return d except: return None def main(): global last_hashes print(f"=== Enterprise Monitor aktiv: {SENDER_NAME} ===", flush=True) # 1. Hashes aus Persistenz laden (Zustand vor dem Stopp/Absturz) if HASH_PERSISTENT: load_hashes() # 2. Verhalten bei SEND_INITIAL_LOG steuern # Wenn True, wird der lokale Cache ignoriert/gelöscht, um alles in der Box neu zu pushen if SEND_INITIAL: print("[!] SEND_INITIAL_LOG aktiv: Lokaler Cache wird ignoriert -> Kompletter Neu-Push aller Box-Einträge.", flush=True) last_hashes = set() session = FritzLuaSession(FB_IP, FB_USER, FB_PWD) vpn_states = {} is_initial = True new_hashes_since_backup = 0 # Backup-Intervall aus ENV (z.B. alle 20 neuen Hashes sichern) backup_every = int(os.getenv("HASH_BACKUP_EVERY", 20)) while True: # --- TEIL 1: SYSTEM LOGS (Chronologischer Rückwärts-Sync) --- log_data = session.fetch_data("log") if log_data and 'data' in log_data and 'log' in log_data['data']: entries = log_data['data']['log'] entries.reverse() # Wichtig: Alt nach Neu verarbeiten for e in entries: m, d, t = e.get('msg', ''), e.get('date', ''), e.get('time', '') h = hashlib.sha1(f"{d}{t}{m}".encode()).hexdigest() if h not in last_hashes: # Senden wenn: # a) Es kein Initial-Lauf ist (Normalbetrieb) # b) Es der Initial-Lauf ist UND SEND_INITIAL_LOG auf True steht if not is_initial or SEND_INITIAL: send_payload(m, d, t) # Neuen Hash in Cache aufnehmen und Zähler erhöhen last_hashes.add(h) new_hashes_since_backup += 1 # Inkrementelles Backup während des Laufs (beugt Datenverlust bei Absturz vor) if HASH_PERSISTENT and new_hashes_since_backup >= backup_every: save_hashes() new_hashes_since_backup = 0 # --- TEIL 2: VPN MONITORING (Site-to-Site Fokus) --- vpn_data = session.fetch_data("vpn") if vpn_data and 'data' in vpn_data and 'vpnList' in vpn_data['data']: for t in vpn_data['data']['vpnList']: n, c = t.get('name'), t.get('connected', False) # Statusänderung erkennen if vpn_states.get(n) != c: # VPN-Alarme nur nach dem ersten Scan senden (Spamschutz beim Boot) if not is_initial: v_msg = f"VPN Tunnel {n} ist {'UP' if c else 'DOWN'}" send_payload(v_msg, force_cat="VPN") vpn_states[n] = c # --- TEIL 3: INITIAL-ABSCHLUSS & CACHE-PFLEGE --- # Nach dem allerersten Durchlauf: Inkrementelle Hashes sofort sichern if is_initial: if HASH_PERSISTENT and new_hashes_since_backup > 0: save_hashes() new_hashes_since_backup = 0 is_initial = False # Speichermanagement: Cache-Limit prüfen und ggf. rotieren if len(last_hashes) > HASH_LIMIT: # Behalte nur die neuesten X Hashes laut HASH_RETAIN sorted_hashes = list(last_hashes) last_hashes = set(sorted_hashes[-HASH_RETAIN:]) if DEBUG_MODE: print(f"[DEBUG] Cache-Limit ({HASH_LIMIT}) erreicht. Bereinigt auf {HASH_RETAIN}.", flush=True) # Nach der Bereinigung den neuen kompakten Stand sichern if HASH_PERSISTENT: save_hashes() # Pause bis zum nächsten Polling-Intervall time.sleep(INTERVAL) if __name__ == "__main__": main()