zurück # CrowdSec Sync & Metabase Dashboard 📊
Dieses Setup synchronisiert automatisch alle CrowdSec-Alerts (inkl. Entscheidungen) in eine eigene MySQL-Datenbank und visualisiert sie über Metabase.
—
## 🔧 Komponenten
- CrowdSec – erkennt Bedrohungen & verwaltet Entscheidungen - MySQL/MariaDB – speichert Alerts & Decisions - Python-Script – sync.py überträgt Daten via CrowdSec Local API - Docker – orchestriert `<DB_USER>-syncer` und Metabase - Metabase – Web-Dashboard zur Analyse
—
## 📁 Verzeichnisstruktur
``` /opt/docker/<DB_USER>-syncer/ ├── app/ │ └── sync.py ├── Dockerfile └── docker-compose.yml ```
—
## 🐳 Dockerfile
FROM python:3.11-slim WORKDIR /app COPY ./app /app RUN pip install --no-cache-dir requests mysql-connector-python schedule CMD ["python", "sync.py"]
—
## 📦 docker-compose.yml
```yaml version: „3.8“
services:
<DB_USER>-syncer:
build: .
container_name: <DB_USER>-syncer
restart: unless-stopped
networks:
- docker_backend
depends_on:
- <DB_USER>
networks:
docker_backend: external: true
```
—
## 🐍 app/sync.py
```python import requests, mysql.connector, schedule, time from datetime import datetime
API_KEY = „DEIN_API_KEY“ API_URL = „http://<DB_USER>:8080/v1/alerts?include_decisions=true“
DB_CONFIG = {
"host": "<MYSQL_HOST>", "user": "<DB_USER>", "password": "<DB_PASSWORD>", "database": "<DB_USER>"
}
def sync():
print(f"[{datetime.now()}] Starte Sync...")
try:
response = requests.get(API_URL, headers={"X-Api-Key": API_KEY})
if response.status_code != 200:
print("API Fehler:", response.status_code, response.text)
return
data = response.json()
conn = mysql.connector.connect(**DB_CONFIG)
cursor = conn.cursor()
for alert in data:
cursor.execute("""
INSERT IGNORE INTO alerts (
id, scenario, scenario_hash, scenario_version, message,
created_at, source_scope, source_value, machine_id, events_count, remediation
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
""", (
alert.get("id"),
alert.get("scenario"),
alert.get("scenario_hash"),
alert.get("scenario_version"),
alert.get("message"),
alert.get("created_at"),
alert.get("source", {}).get("scope"),
alert.get("source", {}).get("value"),
alert.get("machine_id"),
alert.get("events_count"),
alert.get("remediation", True)
))
for decision in alert.get("decisions", []):
cursor.execute("""
INSERT IGNORE INTO decisions (
id, origin, type, scope, value, duration, start_at, stop_at,
simulated, alert_id
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
""", (
decision.get("id"),
decision.get("origin"),
decision.get("type"),
decision.get("scope"),
decision.get("value"),
decision.get("duration"),
decision.get("start_at"),
decision.get("stop_at"),
decision.get("simulated", False),
alert.get("id")
))
conn.commit()
print(f"{cursor.rowcount} Einträge synchronisiert.")
cursor.close()
conn.close()
except Exception as e:
print("Fehler:", e)
schedule.every(5).minutes.do(sync) sync()
while True:
schedule.run_pending() time.sleep(60)
```
—
## 🛠️ SQL Tabellenstruktur
```sql – alerts CREATE TABLE IF NOT EXISTS alerts (
id INT PRIMARY KEY, scenario VARCHAR(255), scenario_hash VARCHAR(255), scenario_version VARCHAR(64), message TEXT, created_at TIMESTAMP, source_scope VARCHAR(64), source_value VARCHAR(64), machine_id VARCHAR(64), events_count INT, remediation BOOLEAN
);
– decisions CREATE TABLE IF NOT EXISTS decisions (
id INT PRIMARY KEY, origin VARCHAR(64), type VARCHAR(64), scope VARCHAR(64), value VARCHAR(64), duration VARCHAR(64), start_at TIMESTAMP, stop_at TIMESTAMP, simulated BOOLEAN, alert_id INT, FOREIGN KEY (alert_id) REFERENCES alerts(id) ON DELETE CASCADE
); ```
—
## 📊 Beispiel-Metabase-Abfragen
```sql – Anzahl Bans pro Tag SELECT DATE(created_at) AS Tag, COUNT(*) AS Anzahl FROM alerts GROUP BY Tag ORDER BY Tag DESC; ```
```sql – Top IPs mit meisten Alerts SELECT source_value AS IP, COUNT(*) AS Hits FROM alerts GROUP BY source_value ORDER BY Hits DESC LIMIT 10; ```
—
## ✅ Ergebnis
- Metabase zeigt strukturierte CrowdSec-Daten - Dashboards für Angriffsquellen, Szenarien und Zeitverlauf - Automatisierte, zyklische Datenaktualisierung
—
## 💡 Tipps
- Achte auf gültige API-Keys (`cscli bouncers add …`) - Container-Rebuild bei Änderungen: `docker compose down && docker compose up -d –build` - Metabase-URL: `https://<DB_USER>.mash4077.dedyn.io`
—
Lars Weiß 05.06.2025 11:45