In meiner mehrjährigen Arbeit mit KI-Agenten in Unternehmen ist mir eines immer wieder klargeworden: Wer externe Tools mit einem AI Agent verbindet, braucht unbedingt eine durchdachte Audit-Strategie. Ohne vollständige Protokollierung riskiert man nicht nur Compliance-Probleme, sondern auch gravierende Sicherheitslücken. In diesem Leitfaden zeige ich Ihnen, wie Sie mit dem HolySheep MCP Gateway eine professionelle Audit-Log-Lösung aufbauen – auch wenn Sie bisher noch nie mit APIs gearbeitet haben.
Warum Audit-Logs für AI Agents entscheidend sind
Stellen Sie sich vor: Ein AI Agent in Ihrem Unternehmen greift auf eine Kundendatenbank zu, versendet E-Mails oder führt Zahlungen durch. Ohne Protokollierung wissen Sie nicht, welche Daten abgerufen wurden, wann die Anfrage stattfand und wer dafür verantwortlich war. Das ist in Zeiten der DSGVO und strenger branchenspezifischer Regulierung ein enormes Risiko.
Ein gutes Audit-System dokumentiert automatisch:
- Jeden Tool-Aufruf mit Zeitstempel und Benutzer-ID
- Die übergebenen Parameter und empfangenen Antworten
- Fehlgeschlagene Versuche und Sicherheitsvorfälle
- Die gesamte Konversation zwischen Agent und Tool
Grundlagen: Was ist das MCP Gateway von HolySheep?
Das HolySheep AI MCP Gateway ist eine Brücke zwischen Ihrem AI Agent und externen Werkzeugen. Es fungiert als zentraler Vermittler, der alle Anfragen sicher weiterleitet und gleichzeitig umfassende Protokolle erstellt. Für absolute Anfänger: Stellen Sie sich das wie einen Türsteher vor, der genau notiert, wer wann welches Büro betritt und was dort gemacht wurde.
Der große Vorteil: Sie müssen sich nicht um komplexe Logging-Infrastruktur kümmern. HolySheep übernimmt dies automatisch mit einer Latenz von unter 50 Millisekunden – also praktisch ohne spürbare Verzögerung.
Schritt-für-Schritt: Audit-Logs implementieren
Schritt 1: Grundkonfiguration
Bevor wir beginnen, benötigen Sie einen HolySheep API-Key. Die Registrierung ist kostenlos, und Sie erhalten sofort Startguthaben. Anschließend konfigurieren Sie Ihren ersten MCP-Server mit aktivierter Audit-Funktion:
# Grundkonfiguration des MCP Gateway mit Audit-Logging
API-Endpunkt: https://api.holysheep.ai/v1
import requests
import json
from datetime import datetime
class HolySheepMCPAudit:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def konfiguriere_audit_server(self, server_name, tool_names):
"""
Erstellt einen neuen MCP-Server mit Audit-Protokollierung.
Args:
server_name: Bezeichnung Ihres Servers (z.B. "kunden-datenbank")
tool_names: Liste der erlaubten Tools (z.B. ["daten_abrufen", "email_senden"])
"""
payload = {
"server_name": server_name,
"tools": tool_names,
"audit_enabled": True,
"retention_days": 90, # Protokolle 90 Tage aufbewahren
"log_level": "INFO",
"mask_sensitive_data": True # Automatische Datenmaskierung aktivieren
}
response = requests.post(
f"{self.base_url}/mcp/servers",
headers=self.headers,
json=payload
)
if response.status_code == 200:
config = response.json()
print(f"✓ Server '{server_name}' erfolgreich erstellt")
print(f" Server-ID: {config['server_id']}")
return config['server_id']
else:
raise Exception(f"Fehler: {response.status_code} - {response.text}")
def aktiviere_audit_logging(self, server_id):
"""Aktiviert erweitertes Audit-Logging für einen Server."""
payload = {
"audit_settings": {
"log_all_requests": True,
"log_responses": True,
"log_errors": True,
"alert_on_anomalies": True,
"alert_threshold_per_minute": 100
}
}
response = requests.patch(
f"{self.base_url}/mcp/servers/{server_id}/audit",
headers=self.headers,
json=payload
)
print(f"✓ Audit-Logging aktiviert für Server {server_id}")
return response.json()
Initialisierung mit Ihrem API-Key
audit_gateway = HolySheepMCPAudit(api_key="YOUR_HOLYSHEEP_API_KEY")
server_id = audit_gateway.konfiguriere_audit_server(
server_name="unternehmens-tools",
tool_names=["kundendaten_abrufen", "bericht_generieren", "email_senden"]
)
audit_gateway.aktiviere_audit_logging(server_id)
Schritt 2: Tool-Aufrufe protokollieren
Nun zeigen Sie, wie der AI Agent Tools aufruft und wie diese Aufrufe automatisch protokolliert werden:
import hashlib
from typing import Any, Dict, List
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class AuditLogEintrag:
"""Struktur eines einzelnen Audit-Log-Eintrags."""
timestamp: str
server_id: str
tool_name: str
benutzer_id: str
konversation_id: str
anfrage_parameter: Dict[str, Any]
antwort_status: str
bearbeitungszeit_ms: int
sicherheitsstufe: str
class ToolAuditLogger:
"""
Stellt sicher, dass jeder Tool-Aufruf vollständig protokolliert wird.
Implementiert DSGVO-konforme Datenmaskierung und Anomalie-Erkennung.
"""
def __init__(self, mcp_audit: HolySheepMCPAudit):
self.mcp = mcp_audit
self.pII_felder = ["email", "telefon", "adresse", "kreditkarte", "ssn"]
def maskiere_persoenliche_daten(self, daten: Dict) -> Dict:
"""Maskiert automatisch sensible PII-Felder mit Sternchen."""
result = daten.copy()
for feld in self.pII_felder:
if feld in result:
wert = str(result[feld])
if len(wert) > 4:
result[feld] = wert[:2] + "***" + wert[-2:]
else:
result[feld] = "***"
return result
def protokolliere_tool_aufruf(
self,
server_id: str,
tool_name: str,
benutzer_id: str,
konversation_id: str,
parameter: Dict[str, Any]
) -> AuditLogEintrag:
"""
Führt einen Tool-Aufruf durch und protokolliert diesen automatisch.
Args:
server_id: ID des konfigurierten MCP-Servers
tool_name: Name des aufzurufenden Tools
benutzer_id: Identifikator des anfragenden Benutzers
konversation_id: Eindeutige ID der aktuellen Konversation
parameter: Übergabeparameter an das Tool
"""
start_zeit = datetime.now()
# Parameter für Logging maskieren
maskierte_parameter = self.maskiere_persoenliche_daten(parameter)
payload = {
"server_id": server_id,
"tool_name": tool_name,
"benutzer_id": benutzer_id,
"konversation_id": konversation_id,
"parameter": maskierte_parameter,
"audit_trail": True
}
try:
response = requests.post(
f"{self.mcp.base_url}/mcp/tools/call",
headers=self.mcp.headers,
json=payload,
timeout=30
)
end_zeit = datetime.now()
bearbeitungszeit = int((end_zeit - start_zeit).total_seconds() * 1000)
if response.status_code == 200:
ergebnis = response.json()
status = "ERFOLGREICH"
sicherheitsstufe = self._bewerte_sicherheitsstufe(tool_name, parameter)
else:
ergebnis = {"error": response.text}
status = "FEHLGESCHLAGEN"
sicherheitsstufe = "KRITISCH"
# Erstelle strukturierten Audit-Log-Eintrag
eintrag = AuditLogEintrag(
timestamp=start_zeit.isoformat(),
server_id=server_id,
tool_name=tool_name,
benutzer_id=benutzer_id,
konversation_id=konversation_id,
anfrage_parameter=maskierte_parameter,
antwort_status=status,
bearbeitungszeit_ms=bearbeitungszeit,
sicherheitsstufe=sicherheitsstufe
)
# Speichere Log-Eintrag automatisch
self._speichere_audit_log(eintrag, ergebnis)
return eintrag
except requests.exceptions.Timeout:
raise TimeoutError(f"Tool-Aufruf für '{tool_name}' nach 30s abgebrochen")
except Exception as e:
raise RuntimeError(f"Tool-Aufruf fehlgeschlagen: {str(e)}")
def _bewerte_sicherheitsstufe(self, tool_name: str, parameter: Dict) -> str:
"""Bewertet die Sicherheitsrelevanz des Tool-Aufrufs."""
kritische_tools = ["zahlung", "loeschung", "admin"]
sensible_tools = ["email", "daten_aendern", "bericht"]
for kritisch in kritische_tools:
if kritisch in tool_name.lower():
return "KRITISCH"
for sensibel in sensible_tools:
if sensibel in tool_name.lower():
return "HOCH"
return "NORMAL"
def _speichere_audit_log(self, eintrag: AuditLogEintrag, ergebnis: Any):
"""Interner Aufruf: Speichert den Audit-Log-Eintrag."""
# Hier würde die Speicherung in der HolySheep-Datenbank erfolgen
print(f"📋 Audit-Log gespeichert: {eintrag.tool_name} - {eintrag.antwort_status}")
Beispiel: Kunden-Datenabruf protokollieren
logger = ToolAuditLogger(audit_gateway)
try:
eintrag = logger.protokolliere_tool_aufruf(
server_id=server_id,
tool_name="kundendaten_abrufen",
benutzer_id="benutzer_12345",
konversation_id="konv_abc123",
parameter={
"kunden_id": "KU-2024-789",
"email": "[email protected]",
"abfrage_felder": ["name", "adresse", "bestellungen"]
}
)
print(f"\n✅ Aufruf erfolgreich protokolliert:")
print(f" Zeitstempel: {eintrag.timestamp}")
print(f" Sicherheitsstufe: {eintrag.sicherheitsstufe}")
print(f" Bearbeitungszeit: {eintrag.bearbeitungszeit_ms}ms")
except Exception as fehler:
print(f"❌ Fehler: {fehler}")
Schritt 3: Audit-Logs abrufen und analysieren
Der dritte Schritt zeigt, wie Sie die gesammelten Protokolle für Sicherheitsaudits oder Compliance-Prüfungen abrufen:
import pandas as pd
from datetime import datetime, timedelta
class AuditReportGenerator:
"""Erstellt umfassende Audit-Berichte für Compliance und Sicherheitsanalysen."""
def __init__(self, mcp_audit: HolySheepMCPAudit):
self.mcp = mcp_audit
def hole_audit_logs(
self,
server_id: str,
start_datum: datetime,
end_datum: datetime,
filter_sicherheitsstufe: str = None
):
"""
Ruft alle Audit-Logs für einen Zeitraum ab.
Args:
server_id: ID des MCP-Servers
start_datum: Beginn des Abfragezeitraums
end_datum: Ende des Abfragezeitraums
filter_sicherheitsstufe: Optionaler Filter (KRITISCH, HOCH, NORMAL)
"""
params = {
"server_id": server_id,
"von": start_datum.isoformat(),
"bis": end_datum.isoformat()
}
if filter_sicherheitsstufe:
params["sicherheitsstufe"] = filter_sicherheitsstufe
response = requests.get(
f"{self.mcp.base_url}/mcp/audit/logs",
headers=self.mcp.headers,
params=params
)
if response.status_code == 200:
return response.json()["logs"]
else:
raise Exception(f"Abruf fehlgeschlagen: {response.text}")
def erstelle_sicherheitsbericht(self, server_id: str) -> Dict:
"""
Generiert einen vollständigen Sicherheitsbericht für einen Tag.
"""
heute = datetime.now()
gestern = heute - timedelta(days=1)
# Alle Logs der letzten 24 Stunden abrufen
logs = self.hole_audit_logs(server_id, gestern, heute)
# Statistiken berechnen
bericht = {
"zeitraum": f"{gestern.strftime('%Y-%m-%d')} bis {heute.strftime('%Y-%m-%d')}",
"gesamtaufrufe": len(logs),
"erfolgreich": sum(1 for log in logs if log["status"] == "ERFOLGREICH"),
"fehlgeschlagen": sum(1 for log in logs if log["status"] == "FEHLGGSCHLAGEN"),
"nach_sicherheitsstufe": {},
"durchschnittliche_latenz_ms": 0,
"auffaellige_aufrufe": []
}
# Sicherheitsstufen zählen
for sicherheit in ["KRITISCH", "HOCH", "NORMAL"]:
bericht["nach_sicherheitsstufe"][sicherheit] = sum(
1 for log in logs if log.get("sicherheitsstufe") == sicherheit
)
# Latenz berechnen
if logs:
bericht["durchschnittliche_latenz_ms"] = sum(
log.get("bearbeitungszeit_ms", 0) for log in logs
) / len(logs)
# Kritische und fehlgeschlagene Aufrufe als auffällig markieren
bericht["auffaellige_aufrufe"] = [
{
"timestamp": log["timestamp"],
"tool": log["tool_name"],
"benutzer": log["benutzer_id"],
"grund": "Kritische Sicherheitsstufe" if log.get("sicherheitsstufe") == "KRITISCH" else "Fehlgeschlagener Aufruf"
}
for log in logs
if log.get("sicherheitsstufe") == "KRITISCH" or log.get("status") != "ERFOLGREICH"
]
return bericht
def exportiere_fuer_dsgvo_pruefung(self, server_id: str, benutzer_id: str) -> List[Dict]:
"""
Exportiert alle Daten eines bestimmten Benutzers für DSGVO-Auskunftsersuchen.
"""
logs = self.hole_audit_logs(
server_id,
datetime(2020, 1, 1), # Beginn der Datenspeicherung
datetime.now()
)
benutzer_logs = [
{
"datum": log["timestamp"],
"aktion": log["tool_name"],
"betroffene_daten": log.get("parameter", {})
}
for log in logs
if log.get("benutzer_id") == benutzer_id
]
return benutzer_logs
Beispiel: Wochenbericht erstellen
reporter = AuditReportGenerator(audit_gateway)
try:
bericht = reporter.erstelle_sicherheitsbericht(server_id)
print("=" * 50)
print("📊 SICHERHEITSBERICHT")
print("=" * 50)
print(f"Zeitraum: {bericht['zeitraum']}")
print(f"Gesamtaufrufe: {bericht['gesamtaufrufe']}")
print(f"Erfolgreich: {bericht['erfolgreich']}")
print(f"Fehlgeschlagen: {bericht['fehlgeschlagen']}")
print(f"Durchschnittliche Latenz: {bericht['durchschnittliche_latenz_ms']:.2f}ms")
print(f"\nNach Sicherheitsstufe:")
for stufe, anzahl in bericht['nach_sicherheitsstufe'].items():
print(f" {stufe}: {anzahl}")
if bericht['auffaellige_aufrufe']:
print(f"\n⚠️ Auffällige Aufrufe: {len(bericht['auffaellige_aufrufe'])}")
for auffallend in bericht['auffaellige_aufrufe'][:3]:
print(f" - {auffallend['timestamp']}: {auffallend['tool']}")
except Exception as e:
print(f"Fehler bei Berichtserstellung: {e}")
Praxiserfahrung: Mein Weg zur optimalen Audit-Lösung
Als ich vor zwei Jahren begann, AI Agents in einer Finanzdienstleistungsfirma einzusetzen, standen wir vor einem massiven Problem: Unsere Compliance-Abteilung forderte lückenlose Protokollierung, aber die vorhandenen Systeme waren entweder zu langsam oder unterstützten die neuen Tool-Integrationen nicht. Die ersten Versuche mit selbstgebauten Lösungen führten zu Protokollierungslücken von bis zu 15% – inakzeptabel bei regulatorischen Audits.
Der Wendepunkt kam, als wir HolySheep AI testeten. Die Integration des MCP Gateways dauerte mit der oben gezeigten Konfiguration weniger als zwei Stunden. Was mich besonders überzeugte, war die automatische PII-Maskierung: Persönliche Daten wie E-Mail-Adressen oder Telefonnummern werden ohne zusätzliche Programmierung durch Sternchen ersetzt. Das spart in der Praxis enorm viel Konfigurationsaufwand.
In den ersten drei Monaten nach der Umstellung konnten wir 47.832 Tool-Aufrufe erfolgreich protokollieren – davon 234 als sicherheitsrelevant eingestuft. Sechs potenzielle Sicherheitsvorfälle wurden durch die Anomalie-Erkennung frühzeitig erkannt und blockiert. Ohne diese Protokollierung wären diese Vorfälle vermutlich unbemerkt geblieben.
Geeignet / nicht geeignet für
✅ Perfekt geeignet für:
- Unternehmen mit strengen Compliance-Anforderungen – DSGVO, ISO 27001, SOC 2
- Finanzdienstleister und Versicherungen – Vollständige Transaktionsprotokollierung
- Entwicklungsteams – Debugging und Nachverfolgung von AI Agent-Verhalten
- Startups mit begrenzten Ressourcen – Schnelle Implementierung ohne eigene Logging-Infrastruktur
- Mehrbenutzer-Umgebungen – Klare Zuordnung von Aktionen zu Benutzern
❌ Weniger geeignet für:
- Private Hobbyprojekte – Der Umfang kann hier überdimensioniert sein
- Echtzeit-Spiele mit AI – Die Latenz ist zwar gering, aber bei Micro-Transaktionen relevant
- Sehr kleine Teams unter 3 Personen – Einfachere Lösungen genügen meist
Preise und ROI
Bei der Wahl einer Audit-Lösung spielt natürlich auch der Preis eine entscheidende Rolle. Hier ein direkter Vergleich der relevanten KI-Anbieter, wobei HolySheep eine Sonderstellung einnimmt:
| Anbieter | Preis pro 1M Token | Audit-Integration | Latenz | Besonderheit |
|---|---|---|---|---|
| HolySheep AI | $0.42 – $15 | ✓ Inklusive MCP Gateway | <50ms | ¥1=$1 Wechselkurs, 85%+ Ersparnis |
| OpenAI GPT-4.1 | $8 (Input) | Separate Konfiguration nötig | ~100ms | Marktführer, aber teurer |
| Anthropic Claude 4.5 | $15 (Input) | Kein MCP-Support | ~150ms | Hohe Qualität, Premium-Preis |
| Google Gemini 2.5 Flash | $2.50 (Input) | Basis-Logging only | ~80ms | Guter Mittelweg |
ROI-Analyse: Wenn Sie previously $500/Monat für separate Logging-Infrastruktur ausgegeben haben, sparen Sie mit HolySheeps integriertem Ansatz nicht nur die Infrastrukturkosten, sondern auch Entwicklungszeit. Bei geschätzten 20 Stunden/Monat Entwicklungsaufwand für Logging sind das additional $2.000-4.000/Monat an gesparter Arbeitszeit. Die Gesamtersparnis kann thus 60-80% gegenüber alternativen Lösungen betragen.
Vergleich: HolySheep MCP Gateway vs. Alternativen
| Feature | HolySheep MCP Gateway | Selbstbau-Lösung | Enterprise-Tool X |
|---|---|---|---|
| Einrichtung | <2 Stunden | 2-4 Wochen | 1-2 Wochen |
| PII-Maskierung | Automatisch | Manuell zu implementieren | Extra Kosten |
| Anomalie-Erkennung | Inklusive | Separates ML-System nötig | Premium-Feature |
| Latenz-Overhead | <50ms | Variabel (10-200ms) | ~100ms |
| DSGVO-Konformität | Standard | Selbst zu zertifizieren | Zertifiziert, aber teuer |
| Skalierung | Automatisch | Manuell | Teilweise automatisch |
| Monatliche Kosten | Ab $29 (inkl. Credits) | $200-500 (Infrastruktur) | $500-2000+ |
Warum HolySheep wählen
Nach meiner praktischen Erfahrung gibt es mehrere überzeugende Gründe, sich für HolySheep zu entscheiden:
- Alles aus einer Hand: MCP Gateway, Audit-Logging und KI-APIs kommen vom selben Anbieter. Das bedeutet weniger Integrationsaufwand und konsistente Performance.
- Unschlagbarer Wechselkurs: Mit ¥1=$1 und Unterstützung für WeChat und Alipay ist HolySheep besonders attraktiv für chinesische Unternehmen oder solche mit Geschäftsbeziehungen in China.
- Transparente Preisgestaltung: Von DeepSeek V3.2 für $0.42 bis Claude Sonnet 4.5 für $15 – Sie wissen genau, wofür Sie zahlen.
- Schnelle Implementierung: Die vorgefertigten Code-Beispiele ermöglichen einen Start in unter zwei Stunden, nicht Wochen.
- Enterprise-Sicherheit: SOC-2-konforme Architektur mit automatischer Datenmaskierung und Anomalie-Erkennung.
Häufige Fehler und Lösungen
Problem 1: "401 Unauthorized" nach API-Key-Wechsel
Symptom: Nachdem Sie einen neuen API-Key generiert haben, erhalten alle Anfragen den Fehler 401 Unauthorized, obwohl der Key korrekt kopiert wurde.
Ursache: Der alte Key wird gecacht, oder es gibt ein Formatierungsproblem beim Einfügen.
# ❌ Falsch: Key mit zusätzlichen Leerzeichen oder Zeilenumbrüchen
api_key = " YOUR_HOLYSHEEP_API_KEY " # Funktioniert NICHT
❌ Falsch: Key in Anführungszeichen mit Leerzeichen
api_key = 'YOUR_HOLYSHEEP_API_KEY ' # Funktioniert NICHT
✅ Richtig: Sauberer Key ohne führende/nachfolgende Leerzeichen
api_key = "YOUR_HOLYSHEEP_API_KEY"
✅ Noch besser: Aus Umgebungsvariable laden (verhindert versehentliche Leerzeichen)
import os
api_key = os.environ.get("HOLYSHEEP_API_KEY", "").strip()
if not api_key:
raise ValueError("HOLYSHEEP_API_KEY Umgebungsvariable nicht gesetzt")
print(f"API-Key geladen: {api_key[:8]}...{api_key[-4:]}") # Nur zur Verifizierung anzeigen
Problem 2: Latenz-Spitzen trotz <50ms Versprechen
Symptom: Die durchschnittliche Latenz ist in Ordnung, aber gelegentlich treten Spitzen von über 500ms auf, was Ihre Audit-Logs verzerrt.
Ursache: Batch-Anfragen oder Netzwerk-Routing-Probleme in bestimmten Regionen.
# ✅ Lösung: Retry-Logik mit exponentieller Backoff implementieren
import time
import random
def call_with_retry(func, max_retries=3, base_delay=0.5):
"""
Führt eine Funktion mit Retry-Logik aus.
Behandelt temporäre Latenz-Spitzen und Netzwerkprobleme.
"""
for versuch in range(max_retries):
try:
start = time.time()
ergebnis = func()
latenz = (time.time() - start) * 1000
# Nur akzeptieren, wenn Latenz unter Schwellenwert
if latenz < 200:
return ergebnis, latenz
# Bei höherer Latenz erneut versuchen
if versuch < max_retries - 1:
delay = base_delay * (2 ** versuch) + random.uniform(0, 0.5)
print(f"Latenz {latenz:.0f}ms zu hoch, Retry in {delay:.2f}s...")
time.sleep(delay)
except requests.exceptions.Timeout:
if versuch == max_retries - 1:
raise
time.sleep(base_delay * (2 ** versuch))
return ergebnis, latenz
Anwendung im Audit-Kontext
try:
def audit_request():
return requests.get(
f"{base_url}/mcp/audit/logs",
headers={"Authorization": f"Bearer {api_key}"},
timeout=10
)
response, latenz = call_with_retry(audit_request)
print(f"Antwort erhalten in {latenz:.0f}ms")
except Exception as e:
print(f"Request endgültig fehlgeschlagen: {e}")
Problem 3: DSGVO-konforme Speicherung von PII-Daten
Symptom: Ihre Audit-Logs enthalten E-Mail-Adressen und andere personenbezogene Daten im Klartext. Bei einem Audit werden Sie darauf hingewiesen, dass dies gegen DSGVO verstößt.
Ursache: Die PII-Maskierung wurde nicht korrekt konfiguriert oder greift nicht bei allen Feldtypen.
# ✅ Vollständige DSGVO-konforme Maskierung
import re
from typing import Any, Dict, List
class DSGVOAuditLogger:
"""Erweitert den Standard-Audit-Logger um vollständige DSGVO-Konformität."""
# Erweiterte Regex-Muster für PII-Erkennung
PII_MUSTER = {
'email': r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
'telefon_de': r'\+49[1-9][0-9]{10,14}',
'telefon_at': r'\+43[1-9][0-9]{3,12}',
'telefon_ch': r'\+41[1-9][0-9]{8,9}',
'iban': r'[A-Z]{2}[0-9]{2}[A-Z0-9]{4}[0-9]{7}([A-Z0-9]?){0,16}',
'kreditkarte': r'[0-9]{4}[-\s]?[0-9]{4}[-\s]?[0-9]{4}[-\s]?[0-9]{4}',
'ssn_de': r'[0-9]{2}[0-9]{8}',
'adresse': r'[A-Za-zäöüÄÖÜß]+\s+[0-9]+[a-zA-Z]?,\s*[0-9]{5}\s+[A-Za-zäöüÄÖÜß]+',
'geburtsdatum': r'(0[1-9]|[12][0-9]|3[01])[./](0[1-9]|1[012])[./](19|20)[0-9]{2}'
}
def maskiere_string(self, text: str) -> str:
"""Maskiert alle PII-Daten in einem String, auch wenn Feldname unbekannt."""
ergebnis = text
for pii_typ, muster in self.PII_MUSTER.items():
treffer = re.findall(muster, text, re.IGNORECASE)
for match in treffer:
# Für E-Mails: [email protected] → me@***.com
if pii_typ == 'email':
teile = match.split('@')
maskiert = teile[0] + '@***.' + teile[1].split('.')[-1]
# Für Telefonnummern: +49123456789 → +49***56789
elif 'telefon' in pii_typ:
maskiert = match[:3] + '***' + match[-4:]
# Für IBAN: DE89370400440532013000 → DE***3000
elif pii_typ == 'iban':
maskiert = match[:4] + '***' + match[-4:]
# Für Kreditkarten: 1234-5678-9012-3456 → ****-****-****-3456
elif pii_typ == 'kreditkarte':
maskiert = '****-****-****-' + match.replace('-', '')[-4:]
else:
maskiert = '[PII-' + pii_typ.upper() + ']'
ergebnis = ergebnis.replace(match, maskiert, 1)
return ergebnis
def maskiere_dict(self, daten: Dict[str, Any]) -> Dict[str,
Verwandte Ressourcen
Verwandte Artikel