Hitonabi
b31736cde7
executor.py erzwingt jetzt HERMES_PC_TOKEN auf allen Steuer-Endpunkten (/shell,/screenshot,/type,/key,/open,/search), fail-closed (503) wenn kein Token gesetzt ist; /health bleibt offen für den Reachability-Check. CORS- Wildcard entfernt, Bind-Host konfigurierbar (HERMES_PC_HOST). mcp_pc.py sendet PC_EXECUTOR_TOKEN als Authorization-Bearer mit. Schließt die unauthentifizierte Remote-Code-Execution auf dem Windows-PC (host=0.0.0.0, kein Token) — Teil von Stufe 0 (Security) des Stack-Reviews. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
118 lines
4.3 KiB
Python
118 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mission Control 2.0 — PC Control MCP Server.
|
|
|
|
Stdio-MCP-Server. Läuft als Subprocess von Hermes Gateway (AI Box).
|
|
Proxied Befehle an den Hermes PC Executor (Windows PC, Port 7777).
|
|
|
|
Env: PC_EXECUTOR_URL (default http://192.168.178.XXX:7777)
|
|
"""
|
|
|
|
import os
|
|
|
|
import httpx
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
PC_URL = os.environ.get("PC_EXECUTOR_URL", "").rstrip("/")
|
|
# Shared Secret — muss identisch zum HERMES_PC_TOKEN des Executors auf dem Windows-PC sein.
|
|
PC_TOKEN = os.environ.get("PC_EXECUTOR_TOKEN", "").strip()
|
|
|
|
mcp = FastMCP("hermes-pc-control")
|
|
|
|
|
|
def _pc(path: str, data: dict | None = None) -> dict:
|
|
if not PC_URL:
|
|
return {"error": "PC_EXECUTOR_URL nicht gesetzt. Setze die Env-Variable mit der IP des Windows PCs."}
|
|
headers = {"Authorization": f"Bearer {PC_TOKEN}"} if PC_TOKEN else {}
|
|
try:
|
|
with httpx.Client(timeout=90) as c:
|
|
if data is None:
|
|
r = c.get(f"{PC_URL}{path}", headers=headers)
|
|
else:
|
|
r = c.post(f"{PC_URL}{path}", json=data, headers=headers)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except httpx.ConnectError:
|
|
return {"error": f"PC Executor nicht erreichbar ({PC_URL}). Läuft executor.py auf dem Windows PC?"}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
|
|
@mcp.tool()
|
|
def pc_shell(command: str) -> str:
|
|
"""Führt einen PowerShell-Befehl auf dem Windows-PC des Nutzers aus.
|
|
Nutze dies um Dateien zu lesen/schreiben, Programme zu starten oder Systeminfos abzufragen.
|
|
Beispiele: 'Get-Date', 'ls C:\\Users', 'notepad.exe test.txt'
|
|
Gibt stdout, stderr und returncode zurück."""
|
|
result = _pc("/shell", {"command": command})
|
|
if "error" in result:
|
|
return f"FEHLER: {result['error']}"
|
|
parts = []
|
|
if result.get("stdout"):
|
|
parts.append(f"stdout:\n{result['stdout']}")
|
|
if result.get("stderr"):
|
|
parts.append(f"stderr:\n{result['stderr']}")
|
|
parts.append(f"returncode: {result.get('returncode', '?')}")
|
|
return "\n".join(parts)
|
|
|
|
|
|
@mcp.tool()
|
|
def pc_screenshot() -> str:
|
|
"""Macht einen Screenshot des Windows-PC-Bildschirms.
|
|
Nutze dies um zu sehen was der Nutzer gerade auf dem Bildschirm hat.
|
|
Gibt einen base64-kodierten JPEG-String zurück."""
|
|
result = _pc("/screenshot", {})
|
|
if "error" in result:
|
|
return f"FEHLER: {result['error']}"
|
|
b64 = result.get("screenshot", "")
|
|
if b64:
|
|
return f"[Screenshot aufgenommen, {len(b64)} Zeichen base64-JPEG]"
|
|
return "Kein Screenshot-Daten erhalten."
|
|
|
|
|
|
@mcp.tool()
|
|
def pc_type(text: str) -> str:
|
|
"""Tippt Text in das aktive Fenster auf dem Windows-PC.
|
|
Der Nutzer muss das Zielfenster vorher aktiviert haben.
|
|
Nützlich um Text in Editoren, Terminals oder Formulare einzugeben."""
|
|
result = _pc("/type", {"text": text})
|
|
if "error" in result:
|
|
return f"FEHLER: {result['error']}"
|
|
return "Text getippt." if result.get("ok") else str(result)
|
|
|
|
|
|
@mcp.tool()
|
|
def pc_key(keys: str) -> str:
|
|
"""Drückt eine Tastenkombination auf dem Windows-PC.
|
|
Format: Tasten mit '+' verbinden. Beispiele: 'ctrl+s', 'alt+F4', 'ctrl+shift+esc', 'win+d'
|
|
Nützlich um Tastenkürzel auszuführen."""
|
|
result = _pc("/key", {"keys": keys})
|
|
if "error" in result:
|
|
return f"FEHLER: {result['error']}"
|
|
return f"Tastenkombination '{keys}' gedrückt." if result.get("ok") else str(result)
|
|
|
|
|
|
@mcp.tool()
|
|
def pc_open(target: str) -> str:
|
|
"""Öffnet eine URL im Browser oder eine Datei/Programm auf dem Windows-PC.
|
|
Für URLs: 'https://...' → öffnet im Standard-Browser.
|
|
Für Dateien/Programme: 'C:\\Pfad\\zur\\Datei.txt' oder 'notepad.exe'"""
|
|
result = _pc("/open", {"target": target})
|
|
if "error" in result:
|
|
return f"FEHLER: {result['error']}"
|
|
return f"'{target}' geöffnet." if result.get("ok") else str(result)
|
|
|
|
|
|
@mcp.tool()
|
|
def pc_status() -> str:
|
|
"""Prüft ob der Windows-PC Executor erreichbar ist und gibt Hostname zurück.
|
|
Rufe dies auf um zu prüfen ob PC-Steuerung möglich ist, bevor du andere pc_* Tools nutzt."""
|
|
result = _pc("/health")
|
|
if "error" in result:
|
|
return f"PC Executor NICHT erreichbar: {result['error']}"
|
|
return f"PC Executor erreichbar. Hostname: {result.get('host', '?')}, Status: {result.get('status', '?')}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run()
|