Andreas.O.Schmidt/fbsyslogsender
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
707e8fce6ecade5f7e9238b31de30a481d38fa6c
fbsyslogsender/syslog-fb-monitor.py
T
Hacklab1-hh 707e8fce6e initialer upload des projektes
2026-02-05 06:11:47 +01:00

410 lines
16 KiB
Python
Raw Blame History

#!/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()
Reference in New Issue View Git Blame Copy Permalink