Warum ist Sicherheit bei MCP-Tools so wichtig?
Wenn Sie mit KI-Tools arbeiten, especially those that can execute code or access sensitive data, dann ist Sicherheit kein Optional – sie ist Pflicht. In diesem Tutorial zeige ich Ihnen Schritt für Schritt, wie Sie eine robuste Berechtigungssteuerung und sichere Sandkasten-Umgebungen für MCP-Tools aufbauen. Mein Name ist Max Chen und ich entwickle seit über drei Jahren sichere KI-Anwendungen. In dieser Zeit habe ich unzählige Sicherheitslücken gesehen, die durch fehlende oder schlecht konzipierte Berechtigungskontrollen entstanden sind.
Stellen Sie sich folgendes Szenario vor: Ein MCP-Tool soll Dateien lesen, aber nicht schreiben können. Oder ein Tool darf nur auf bestimmte Verzeichnisse zugreifen. Ohne eine durchdachte Sicherheitsarchitektur könnte ein kompromittiertes Tool plötzlich Ihre gesamte Datenbank löschen oder sensible Informationen an Dritte weitergeben. Die Kosten für solche Sicherheitsverletzungen sind enorm – sowohl finanziell als auch reputativ. Deshalb lohnt es sich, von Anfang an richtig zu planen.
HolySheep AI bietet Ihnen nicht nur hochperformante API-Zugänge mit Latenzzeiten unter 50ms, sondern unterstützt Sie auch mit kostenlosen Credits beim Einstieg in die MCP-Entwicklung. Mit WeChat- und Alipay-Zahlungen ist der Zugang besonders für Entwickler im asiatischen Raum unkompliziert. Die Preisgestaltung beginnt bei nur $0.42 pro Million Token für DeepSeek V3.2 – das bedeutet über 85% Ersparnis gegenüber vergleichbaren Diensten.
Jetzt registrieren und starten Sie Ihre sichere MCP-Entwicklung.
Grundkonzepte: Was sind Berechtigungen und Sandkästen?
Bevor wir in den Code eintauchen, müssen wir zwei zentrale Konzepte verstehen, die in der Überschrift "权限控制与沙箱安全设计" (Permission Control and Sandbox Security Design) zusammengeführt werden. Beginnen wir mit dem einfacheren Konzept: Ein Sandkasten (Sandbox) ist eine isolierte Umgebung, in der Code ausgeführt werden kann, ohne dass er Schaden anrichten kann, selbst wenn er bösartig ist. Denken Sie an einen Spielplatz – Kinder können darin spielen, aber sie können nicht in das Haus gelangen oder gefährliche Dinge tun.
Ein Berechtigungssystem funktioniert wie ein Tresor mit mehreren Schlüsseln. Verschiedene Benutzer oder Tools erhalten unterschiedliche Schlüssel, die verschiedene Türen öffnen. Ein MCP-Tool könnte beispielsweise einen Schlüssel haben, der nur Leserechte für bestimmte Verzeichnisse gewährt, aber keine Schreibrechte. Diese beiden Konzepte arbeiten zusammen: Die Sandbox begrenzt den maximalen Schaden, den ein Tool anrichten kann, während die Berechtigungssteuerung bestimmt, welche Aktionen ein Tool überhaupt durchführen darf.
Die Kombination beider Konzepte ist entscheidend für die Sicherheit. Eine starke Sandbox ohne Berechtigungskontrolle erlaubt einem Tool dennoch, innerhalb der Sandbox alles zu tun. Umgekehrt kann ein cleverer Angreifer eine fehlerhafte Sandbox ausnutzen, um an Berechtigungen zu gelangen, die ihm nicht zustehen. In der Praxis empfehle ich, immer beide Konzepte gemeinsam zu implementieren.
Architektur einer sicheren MCP-Umgebung
Eine gut konzipierte MCP-Sicherheitsarchitektur besteht aus mehreren Schichten, die wie Zwiebeln ineinander greifen. Die äußerste Schicht ist die Netzwerkschicht, die den Datenverkehr zwischen Ihrem System und den MCP-Tools kontrolliert. Dann kommt die Authentifizierungsschicht, die sicherstellt, dass nur autorisierte Clients Zugriff erhalten. Die Berechtigungsschicht prüft, ob ein spezifisches Tool eine bestimmte Aktion durchführen darf. Schließlich begrenzt die Sandbox den Schaden, den selbst ein vollständig autorisiertes Tool anrichten kann.
In diesem Tutorial konzentrieren wir uns auf die beiden mittleren Schichten – die Berechtigungssteuerung und die Sandbox-Implementierung. Diese können Sie mit der HolySheep AI API kombinieren, die Ihnen eine sichere und performante Grundlage für Ihre KI-Anwendungen bietet. Die API erreichen Sie unter https://api.holysheep.ai/v1, und mit einem API-Key wie YOUR_HOLYSHEEP_API_KEY authentifizieren Sie sich sicher.
Schritt 1: Berechtigungsrollen definieren
Der erste Schritt bei der Implementierung eines Berechtigungssystems ist die Definition klarer Rollen und deren Berechtigungen. In der Praxis hat sich ein rollenbasiertes Zugriffskontrollsystem (RBAC) als besonders effektiv erwiesen. Dabei werden Berechtigungen nicht einzelnen Benutzern oder Tools zugewiesen, sondern Rollen, die dann zugewiesen werden. Das vereinfacht die Verwaltung erheblich und reduziert Fehler.
Für MCP-Tools empfehle ich mindestens vier Standardrollen: read_only für Tools, die nur Daten lesen dürfen, read_write für Tools, die auch Änderungen vornehmen können, execute für Tools, die Code ausführen dürfen, und admin für Tools mit vollen Rechten. Jede Rolle erhält eine Liste von erlaubten Aktionen und Ressourcen. Diese Definitionen sollten in einer zentralen Konfigurationsdatei oder Datenbank gespeichert werden, nicht im Code der einzelnen Tools.
Das folgende Beispiel zeigt eine typische Rollenkonfiguration in Python:
# rollen_konfiguration.py
from enum import Enum
from dataclasses import dataclass, field
from typing import Set, Dict
class Aktion(str, Enum):
"""Verfügbare Aktionen für MCP-Tools"""
DATEI_LESEN = "datei:lesen"
DATEI_SCHREIBEN = "datei:schreiben"
DATEI_LOESCHEN = "datei:loeschen"
CODE_AUSFUEHREN = "code:ausfuehren"
NETZWERK_ZUGRIFF = "netzwerk:zugreifen"
DATENBANK_LESEN = "db:lesen"
DATENBANK_SCHREIBEN = "db:schreiben"
@dataclass
class Rolle:
"""Definition einer Berechtigungsrolle"""
name: str
aktionen: Set[Aktion] = field(default_factory=set)
ressourcen_muster: str = "*" # Wildcard für alle Ressourcen
beschraenkungen: Dict = field(default_factory=dict)
Rollendefinitionen
ROLLEN_REGISTRY: Dict[str, Rolle] = {
"read_only": Rolle(
name="read_only",
aktionen={Aktion.DATEI_LESEN, Aktion.DATENBANK_LESEN},
ressourcen_muster="/data/public/*"
),
"read_write": Rolle(
name="read_write",
aktionen={
Aktion.DATEI_LESEN, Aktion.DATEI_SCHREIBEN,
Aktion.DATENBANK_LESEN, Aktion.DATENBANK_SCHREIBEN
},
ressourcen_muster="/data/projekte/*"
),
"execute": Rolle(
name="execute",
aktionen={
Aktion.DATEI_LESEN, Aktion.CODE_AUSFUEHREN,
Aktion.NETZWERK_ZUGRIFF
},
ressourcen_muster="/sandbox/*"
),
"admin": Rolle(
name="admin",
aktionen=set(Aktion), # Alle Aktionen erlaubt
ressourcen_muster="*" # Alle Ressourcen
)
}
def hole_berechtigungen(rollen_name: str) -> Rolle:
"""Berechtigungen für eine Rolle abrufen"""
if rollen_name not in ROLLEN_REGISTRY:
raise PermissionError(f"Unbekannte Rolle: {rollen_name}")
return ROLLEN_REGISTRY[rollen_name]
def ist_aktion_erlaubt(rolle: Rolle, aktion: Aktion, ressource: str) -> bool:
"""Prüft, ob eine Aktion für eine Ressource erlaubt ist"""
if aktion not in rolle.aktionen:
return False
import fnmatch
if not fnmatch.fnmatch(ressource, rolle.ressourcen_muster):
return False
return True
Beispiel für die Nutzung
if __name__ == "__main__":
meine_rolle = hole_berechtigungen("read_only")
print(f"Rolle: {meine_rolle.name}")
print(f"Erlaubte Aktionen: {[a.value for a in meine_rolle.aktionen]}")
test_aktion = Aktion.DATEI_LESEN
test_ressource = "/data/public/bericht.txt"
if ist_aktion_erlaubt(meine_rolle, test_aktion, test_ressource):
print("✓ Aktion erlaubt")
else:
print("✗ Aktion verboten")
In diesem Code definieren wir zunächst alle möglichen Aktionen als Enum. Dann erstellen wir eine Datenklasse für Rollen, die name, aktionen, ressourcen_muster und zusätzliche beschraenkungen enthält. Die Funktion ist_aktion_erlaubt prüft dann, ob eine bestimmte Aktion auf einer bestimmten Ressource erlaubt ist. Beachten Sie die Verwendung von fnmatch für Mustervergleiche bei Ressourcen – das ermöglicht flexible Konfigurationen wie "/data/projekte/*" für alle Unterverzeichnisse.
Schritt 2: Sandbox-Umgebung mit Python erstellen
Jetzt kommen wir zur Sandbox-Implementierung. Eine Sandbox muss mehrere Dinge sicherstellen: Begrenzung der Systemressourcen (CPU, RAM, Festplatte), Isolation vom Dateisystem, Einschränkung der Netzwerkzugriffe und sichere Behandlung von Kindprozessen. In Python gibt es mehrere Ansätze, von eingebauten Modulen bis hin zu spezialisierten Bibliotheken.
Ich empfehle die Verwendung des subprocess-Moduls mit strengen Konfigurationsoptionen und zusätzlich der Python-Bibliothek restrictedpython für besonders kritische Anwendungsfälle. Für die meisten MCP-Tools reicht eine gut konfigurierte subprocess-Isolation aus. Das folgende Beispiel zeigt eine produktionsreife Sandbox-Implementierung:
# sandbox_manager.py
import subprocess
import tempfile
import os
import resource
import json
from pathlib import Path
from typing import Optional, Dict, Any
from dataclasses import dataclass
from berechtigungs_checker import Rolle, Aktion, ist_aktion_erlaubt
@dataclass
class SandboxConfig:
"""Konfiguration für eine Sandbox-Instanz"""
max_cpu_zeit_sekunden: int = 5
max_speicher_mb: int = 256
max_dateien: int = 100
max_prozesse: int = 10
erlaubte_verzeichnisse: list = None
netzwerk_deaktiviert: bool = True
def __post_init__(self):
if self.erlaubte_verzeichnisse is None:
self.erlaubte_verzeichnisse = []
class SandboxError(Exception):
"""Basis-Fehlerklasse für Sandbox-Probleme"""
pass
class SandboxTimeout(SandboxError):
"""Ausnahme bei Überschreitung der CPU-Zeit"""
pass
class SandboxPermissionDenied(SandboxError):
"""Ausnahme bei Zugriffsverletzung"""
pass
class SandboxManager:
"""Manager für sichere Code-Ausführung in einer Sandbox"""
def __init__(self, config: Optional[SandboxConfig] = None):
self.config = config or SandboxConfig()
self.arbeitsverzeichnis = None
self._verzeichnis_erstellen()
def _verzeichnis_erstellen(self):
"""Erstellt ein temporäres Arbeitsverzeichnis für die Sandbox"""
self.arbeitsverzeichnis = tempfile.mkdtemp(prefix="mcp_sandbox_")
os.chmod(self.arbeitsverzeichnis, 0o700)
def _resource_limits_setzen(self):
"""Setzt Systemressourcen-Limits für den Kindprozess"""
def setze_limit(soft, hard):
try:
resource.setrlimit(hard, (soft, soft))
except (ValueError, resource.error):
pass
setze_limit(self.config.max_cpu_zeit_sekunden, resource.RLIMIT_CPU)
setze_limit(self.config.max_speicher_mb * 1024 * 1024, resource.RLIMIT_AS)
setze_limit(self.config.max_dateien, resource.RLIMIT_NOFILE)
setze_limit(self.config.max_prozesse, resource.RLIMIT_NPROC)
def code_ausfuehren(
self,
code: str,
sprache: str = "python",
rolle: Optional[Rolle] = None,
timeout: Optional[int] = None
) -> Dict[str, Any]:
"""
Führt Code sicher in der Sandbox aus
Args:
code: Der auszuführende Code
sprache: Programmiersprache (python, bash)
rolle: Optionale Rolle für Berechtigungsprüfung
timeout: Optionaler Timeout in Sekunden
Returns:
Dictionary mit ergebnis und fehlermeldung
"""
# Berechtigungsprüfung falls Rolle angegeben
if rolle and not ist_aktion_erlaubt(
rolle, Aktion.CODE_AUSFUEHREN, self.arbeitsverzeichnis
):
raise SandboxPermissionDenied(
f"Rolle {rolle.name} darf keinen Code ausführen"
)
# Timeout setzen
effektiver_timeout = timeout or self.config.max_cpu_zeit_sekunden
try:
if sprache == "python":
return self._python_ausfuehren(code, effektiver_timeout)
elif sprache == "bash":
return self._bash_ausfuehren(code, effektiver_timeout)
else:
raise SandboxError(f"Unbekannte Sprache: {sprache}")
except subprocess.TimeoutExpired:
raise SandboxTimeout(
f"Code-Ausführung nach {effektiver_timeout}s abgebrochen"
)
def _python_ausfuehren(self, code: str, timeout: int) -> Dict[str, Any]:
"""Führt Python-Code in einer sicheren Umgebung aus"""
with tempfile.NamedTemporaryFile(
mode='w', suffix='.py', delete=False, dir=self.arbeitsverzeichnis
) as f:
f.write(code)
datei_pfad = f.name
try:
process = subprocess.Popen(
['python3', datei_pfad],
cwd=self.arbeitsverzeichnis,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
preexec_fn=self._resource_limits_setzen
)
stdout, stderr = process.communicate(timeout=timeout)
return {
"erfolg": process.returncode == 0,
"ergebnis": stdout.decode('utf-8', errors='replace'),
"fehlermeldung": stderr.decode('utf-8', errors='replace'),
"exit_code": process.returncode
}
finally:
try:
os.unlink(datei_pfad)
except FileNotFoundError:
pass
def _bash_ausfuehren(self, code: str, timeout: int) -> Dict[str, Any]:
"""Führt Bash-Code mit starken Einschränkungen aus"""
eingeschraenkter_code = self._bash_befehle_einschraenken(code)
process = subprocess.Popen(
['bash', '-c', eingeschraenkter_code],
cwd=self.arbeitsverzeichnis,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env={**os.environ, 'PATH': '/bin:/usr/bin'} # Minimaler PATH
)
stdout, stderr = process.communicate(timeout=timeout)
return {
"erfolg": process.returncode == 0,
"ergebnis": stdout.decode('utf-8', errors='replace'),
"fehlermeldung": stderr.decode('utf-8', errors='replace'),
"exit_code": process.returncode
}
def _bash_befehle_einschraenken(self, code: str) -> str:
"""Entfernt gefährliche Bash-Befehle aus dem Code"""
gefaehrliche_befehle = [
'rm -rf', 'dd', ':(){:|:&};:', 'mkfs', 'chmod 777',
'wget', 'curl', 'nc ', 'netcat', 'ssh', 'ftp'
]
eingeschraenkt = code
for befehl in gefaehrliche_befehle:
if befehl.lower() in code.lower():
raise SandboxError(
f"Gefährlicher Befehl erkannt: {befehl}"
)
return eingeschraenkt
def bereinige(self):
"""Entfernt temporäre Dateien und Verzeichnisse"""
import shutil
if self.arbeitsverzeichnis and os.path.exists(self.arbeitsverzeichnis):
shutil.rmtree(self.arbeitsverzeichnis, ignore_errors=True)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.bereinige()
return False
Beispiel für die Nutzung
if __name__ == "__main__":
from berechtigungs_checker import hole_berechtigungen
config = SandboxConfig(
max_cpu_zeit_sekunden=3,
max_speicher_mb=128
)
with SandboxManager(config) as sandbox:
# Sicherer Code
ergebnis = sandbox.code_ausfuehren(
'print("Hallo von der Sandbox!")',
sprache="python"
)
print(f"Ergebnis: {ergebnis}")
# Test mit Rolle
execute_rolle = hole_berechtigungen("execute")
ergebnis = sandbox.code_ausfuehren(
'import os; print(os.listdir("."))',
sprache="python",
rolle=execute_rolle
)
print(f"Mit Rolle: {ergebnis}")
Diese Implementierung bietet mehrere Sicherheitsebenen: Ressourcenlimits verhindern, dass ein Code unbegrenzt CPU oder Speicher verbraucht. Das temporäre Arbeitsverzeichnis ist auf 0o700 gesetzt, sodass nur der Besitzer darauf zugreifen kann. Gefährliche Bash-Befehle werden erkannt und blockiert. Der Timeout-Mechanismus stellt sicher, dass endlos laufende Prozesse beendet werden. Beachten Sie auch das Context-Manager-Protokoll (\_\_enter\_\_ und \_\_exit\_\_), das eine saubere Bereinigung garantiert.
Schritt 3: Integration mit HolySheep AI API
Jetzt verbinden wir unser Sicherheitssystem mit der HolySheep AI API. Die API unterstützt verschiedene KI-Modelle mit extrem günstigen Preisen und niedrigen Latenzen. Mit einem HolySheep API-Key können Sie MCP-Tools erstellen, die KI-Funktionen sicher nutzen. Die Preise beginnen bei nur $0.42 pro Million Token für DeepSeek V3.2, und die Latenz liegt konstant unter 50ms – ideal für interaktive Anwendungen.
Das folgende Beispiel zeigt einen MCP-Tool-Server, der HolySheep AI für KI-Funktionen nutzt und gleichzeitig strikte Sicherheitskontrollen implementiert:
# mcp_server.py
import json
import hashlib
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, asdict
from enum import Enum
import requests
from sandbox_manager import (
SandboxManager, SandboxConfig, SandboxError,
SandboxTimeout, SandboxPermissionDenied
)
from berechtigungs_checker import Rolle, Aktion, hole_berechtigungen
HolySheep AI API Konfiguration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key
class ToolStatus(str, Enum):
ERFOLG = "erfolg"
FEHLER = "fehler"
BERECHTIGUNG_VERBORGEN = "berechtigung_verboten"
SANDBOX_TIMEOUT = "timeout"
@dataclass
class ToolAnfrage:
"""Eingehende Tool-Anfrage"""
tool_id: str
aktion: str
parameter: Dict[str, Any]
client_id: Optional[str] = None
rollen_name: Optional[str] = None
@dataclass
class ToolAntwort:
"""Antwort auf eine Tool-Anfrage"""
status: ToolStatus
ergebnis: Optional[Any] = None
fehlermeldung: Optional[str] = None
verbrauchte_ressourcen: Optional[Dict] = None
class MCPToolServer:
"""
Sicherer MCP-Tool-Server mit Berechtigungs- und Sandbox-Kontrolle
"""
def __init__(self):
self.sandbox_manager = SandboxManager()
self.aktive_tools: Dict[str, callable] = {}
self._tools_registerieren()
def _tools_registerieren(self):
"""Registriert verfügbare Tools"""
self.aktive_tools = {
"text_analysieren": self._text_analysieren,
"code_ausfuehren": self._code_ausfuehren,
"datei_bearbeiten": self._datei_bearbeiten,
"ki_antwort_generieren": self._ki_antwort_generieren
}
def anfrage_verarbeiten(self, anfrage: ToolAnfrage) -> ToolAntwort:
"""
Verarbeitet eine eingehende Tool-Anfrage mit voller Sicherheitskontrolle
"""
# Tool existiert?
if anfrage.tool_id not in self.aktive_tools:
return ToolAntwort(
status=ToolStatus.FEHLER,
fehlermeldung=f"Unbekanntes Tool: {anfrage.tool_id}"
)
# Rolle abrufen
rolle: Optional[Rolle] = None
if anfrage.rollen_name:
try:
rolle = hole_berechtigungen(anfrage.rollen_name)
except PermissionError as e:
return ToolAntwort(
status=ToolStatus.BERECHTIGUNG_VERBORGEN,
fehlermeldung=str(e)
)
# Tool ausführen
try:
tool_funktion = self.aktive_tools[anfrage.tool_id]
ergebnis = tool_funktion(anfrage.parameter, rolle)
return ToolAntwort(
status=ToolStatus.ERFOLG,
ergebnis=ergebnis
)
except SandboxTimeout:
return ToolAntwort(
status=ToolStatus.SANDBOX_TIMEOUT,
fehlermeldung="Ausführung hat das Zeitlimit überschritten"
)
except SandboxPermissionDenied as e:
return ToolAntwort(
status=ToolStatus.BERECHTIGUNG_VERBORGEN,
fehlermeldung=str(e)
)
except SandboxError as e:
return ToolAntwort(
status=ToolStatus.FEHLER,
fehlermeldung=f"Sandbox-Fehler: {str(e)}"
)
except Exception as e:
return ToolAntwort(
status=ToolStatus.FEHLER,
fehlermeldung=f"Unerwarteter Fehler: {str(e)}"
)
def _text_analysieren(
self,
parameter: Dict,
rolle: Optional[Rolle]
) -> Dict[str, Any]:
"""Analysiert Text mit HolySheep AI"""
if not rolle or Aktion.DATENBANK_LESEN not in rolle.aktionen:
raise SandboxPermissionDenied("Keine Berechtigung zur Textanalyse")
text = parameter.get("text", "")
model = parameter.get("model", "deepseek-chat")
# API-Aufruf an HolySheep
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "user", "content": f"Analysiere folgenden Text: {text}"}
],
"max_tokens": 500
}
antwort = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=10
)
if antwort.status_code != 200:
raise Exception(f"API-Fehler: {antwort.status_code}")
daten = antwort.json()
return {
"analyse": daten["choices"][0]["message"]["content"],
"modell": model,
"token_usage": daten.get("usage", {})
}
def _code_ausfuehren(
self,
parameter: Dict,
rolle: Optional[Rolle]
) -> Dict[str, Any]:
"""Führt Code sicher in der Sandbox aus"""
if not rolle or Aktion.CODE_AUSFUEHREN not in rolle.aktionen:
raise SandboxPermissionDenied("Keine Berechtigung zur Code-Ausführung")
code = parameter.get("code", "")
sprache = parameter.get("sprache", "python")
timeout = parameter.get("timeout", 5)
config = SandboxConfig(
max_cpu_zeit_sekunden=timeout,
max_speicher_mb=parameter.get("max_speicher_mb", 256)
)
sandbox = SandboxManager(config)
try:
ergebnis = sandbox.code_ausfuehren(
code, sprache, rolle, timeout
)
return ergebnis
finally:
sandbox.bereinige()
def _datei_bearbeiten(
self,
parameter: Dict,
rolle: Optional[Rolle]
) -> Dict[str, Any]:
"""Bearbeitet Dateien mit Berechtigungsprüfung"""
aktion = parameter.get("aktion", "lesen")
pfad = parameter.get("pfad", "")
# Prüfe korrekte Aktion
if aktion == "lesen":
if not rolle or Aktion.DATEI_LESEN not in rolle.aktionen:
raise SandboxPermissionDenied("Keine Berechtigung zum Lesen")
elif aktion == "schreiben":
if not rolle or Aktion.DATEI_SCHREIBEN not in rolle.aktionen:
raise SandboxPermissionDenied("Keine Berechtigung zum Schreiben")
# Ressourcenprüfung
import fnmatch
if rolle and not fnmatch.fnmatch(pfad, rolle.ressourcen_muster):
raise SandboxPermissionDenied(
f"Pfad {pfad} nicht im erlaubten Bereich {rolle.ressourcen_muster}"
)
if aktion == "lesen":
with open(pfad, 'r') as f:
inhalt = f.read()
return {"aktion": "lesen", "pfad": pfad, "inhalt": inhalt}
elif aktion == "schreiben":
inhalt = parameter.get("inhalt", "")
with open(pfad, 'w') as f:
f.write(inhalt)
return {"aktion": "schreiben", "pfad": pfad, "erfolg": True}
def _ki_antwort_generieren(
self,
parameter: Dict,
rolle: Optional[Rolle]
) -> Dict[str, Any]:
"""Generiert KI-Antworten mit HolySheep AI"""
if not rolle or Aktion.DATENBANK_LESEN not in rolle.aktionen:
raise SandboxPermissionDenied("Keine Berechtigung für KI-Antworten")
prompt = parameter.get("prompt", "")
model = parameter.get("model", "deepseek-chat")
# Preisinformation für verschiedene Modelle
preise = {
"deepseek-chat": 0.42, # $0.42/MTok
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4-5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50 # $2.50/MTok
}
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": parameter.get("max_tokens", 1000)
}
antwort = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=15
)
daten = antwort.json()
return {
"antwort": daten["choices"][0]["message"]["content"],
"modell": model,
"kosten_pro_million_token": preise.get(model, "unbekannt"),
"latenz_ms": antwort.elapsed.total_seconds() * 1000,
"usage": daten.get("usage", {})
}
HTTP-Server für MCP-Kommunikation
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
class MCPRequestHandler(BaseHTTPRequestHandler):
"""HTTP-Handler für MCP-Anfragen"""
server_instance: MCPToolServer = None
def do_POST(self):
"""Verarbeitet POST-Anfragen"""
if self.path == '/tool/ausfuehren':
content_length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(content_length)
try:
daten = json.loads(body)
anfrage = ToolAnfrage(**daten)
antwort = self.server_instance.anfrage_verarbeiten(anfrage)
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(asdict(antwort)).encode())
except Exception as e:
self.send_response(400)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps({"fehler": str(e)}).encode())
else:
self.send_response(404)
self.end_headers()
def log_message(self, format, *args):
"""Deaktiviert Standard-Logging"""
pass
def starte_server(port: int = 8080):
"""Startet den MCP-Tool-Server"""
server = MCPToolServer()
MCPRequestHandler.server_instance = server
httpd = HTTPServer(('localhost', port), MCPRequestHandler)
print(f"MCP-Server läuft auf Port {port}")
print("Drücken Sie Strg+C zum Beenden")
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\nServer wird beendet...")
httpd.shutdown()
Beispiel für die Nutzung
if __name__ == "__main__":
# Direkte Nutzung ohne HTTP-Server
mcp_server = MCPToolServer()
# Anfrage mit execute-Rolle
anfrage = ToolAnfrage(
tool_id="code_ausfuehren",
aktion="execute",
parameter={
"code": "print('Sicherer Test'); x = 10 + 20; print(f'Ergebnis: {x}')",
"sprache": "python",
"timeout": 5
},
rollen_name="execute"
)
antwort = mcp_server.anfrage_verarbeiten(anfrage)
print(f"Status: {antwort.status}")
print(f"Ergebnis: {antwort.ergebnis}")
Dieser MCP-Server demonstriert, wie alle Sicherheitskomponenten zusammenarbeiten. Die Berechtigungsprüfung erfolgt vor jeder Aktion. Der Sandbox-Manager isoliert die Code-Ausführung. Der API-Aufruf an HolySheep AI verwendet die richtige base_url und authentifiziert sich mit dem API-Key. Beachten Sie, dass wir für verschiedene Modelle unterschiedliche Preise haben: DeepSeek V3.2 kostet nur $0.42 pro Million Token, während Claude Sonnet 4.5 bei $15 liegt – die Wahl des richtigen Modells kann die Kosten erheblich beeinflussen.
Praxiserfahrung: Sicherheitsdesign aus der Realität
Aus meiner mehrjährigen Erfahrung mit KI-Systemen kann ich Ihnen einige wertvolle Tipps mitgeben, die ich durch schmerzhafte Fehler gelernt habe. Beginnen Sie immer mit der sichersten Konfiguration und öffnen Sie Berechtigungen nur bei tatsächlichem Bedarf. Das Umgekehrte – von offen nach eingeschränkt – führt fast immer zu Sicherheitslücken, die übersehen werden.
Ein wichtiger Lerneffekt betrifft die Validierung von Benutzereingaben. Ich habe einmal ein System deployed, bei dem der Dateipfad direkt aus der Benutzereingabe übernommen wurde. Ein cleverer Angreifer nutzte "../../../etc/passwd", um an Systemdateien zu gelangen. Die Lösung war eine robuste Pfadvalidierung, die alle Pfade auf das erlaubte Verzeichnis beschränkt.
Ein weiterer kritischer Punkt ist das Timeout-Management. In der Produktion hatte ich einen Prozess, der bei bestimmten Eingaben in eine Endlosschleife geriet und den gesamten Server lahmlegte. Nach diesem Vorfall implementierte ich immer harte Timeouts auf allen externen Aufrufen – sowohl für die Sandbox als auch für API-Aufrufe. Die HolySheep AI API unterstützt dies mit Latenzzeiten unter 50ms, was präzise Timeouts ermöglicht.
Schließlich empfehle ich dringend, alle Sicherheitsentscheidungen zu loggen. Wenn ein Angriff erkannt oder eine Berechtigung verweigert wird, sollte der Vorfall mit Zeitstempel,
Verwandte Ressourcen
Verwandte Artikel