mission-control-v2/client/hermes-pc/executor.py

"""
Hermes PC Executor — läuft auf dem Windows PC.
Empfängt Tool-Befehle vom MCP-Server der AI Box und führt sie lokal aus.
"""
import base64
import hmac
import io
import os
import socket
import subprocess
import webbrowser
from urllib.parse import quote

import uvicorn
from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI(title="Hermes PC Executor")

MY_PORT = int(os.environ.get("HERMES_PC_PORT", "7777"))
# Auf welchem Interface lauschen. Default 0.0.0.0 (LAN), per Env einschränkbar (z.B. die LAN-IP des PCs).
MY_HOST = os.environ.get("HERMES_PC_HOST", "0.0.0.0")

# ── Auth ───────────────────────────────────────────────────────────────────
# Shared Secret. OHNE Token sind die gefährlichen Endpunkte (Shell/Eingabe/Öffnen/Screenshot)
# fail-closed gesperrt → ein un-konfigurierter Executor ist KEINE offene Remote-Code-Execution mehr.
# Der Token muss identisch auf der Box (Hermes-Env PC_EXECUTOR_TOKEN → mcp_pc.py) gesetzt sein.
AUTH_TOKEN = os.environ.get("HERMES_PC_TOKEN", "").strip()


def require_auth(authorization: str | None = Header(default=None)) -> None:
    """Bearer-Token-Prüfung (konstante Zeit). 503 wenn der Executor ohne Token läuft (fail-closed),
    401 bei fehlendem/falschem Token."""
    if not AUTH_TOKEN:
        raise HTTPException(
            503,
            "Executor ohne HERMES_PC_TOKEN gestartet — Steuer-Endpunkte sind aus Sicherheitsgründen "
            "gesperrt. Setze die Env-Variable HERMES_PC_TOKEN (identisch zur Box) und starte neu.",
        )
    expected = f"Bearer {AUTH_TOKEN}"
    if not authorization or not hmac.compare_digest(authorization, expected):
        raise HTTPException(401, "Ungültiges oder fehlendes Bearer-Token.")


# ── Shell ──────────────────────────────────────────────────────────────────

@app.post("/shell", dependencies=[Depends(require_auth)])
async def run_shell(req: dict):
    cmd = req.get("command", "")
    try:
        result = subprocess.run(
            ["powershell", "-NoProfile", "-NonInteractive", "-Command", cmd],
            capture_output=True, text=True, timeout=60,
            encoding="utf-8", errors="replace",
        )
        return {
            "stdout": result.stdout.strip(),
            "stderr": result.stderr.strip(),
            "returncode": result.returncode,
        }
    except subprocess.TimeoutExpired:
        raise HTTPException(408, "Timeout nach 60s")
    except Exception as e:
        raise HTTPException(500, str(e))


# ── Screen ─────────────────────────────────────────────────────────────────

@app.post("/screenshot", dependencies=[Depends(require_auth)])
async def take_screenshot():
    try:
        import mss
        from PIL import Image

        with mss.mss() as sct:
            monitor = sct.monitors[1]
            raw = sct.grab(monitor)
        img = Image.frombytes("RGB", raw.size, raw.bgra, "raw", "BGRX")
        img.thumbnail((1280, 720))
        buf = io.BytesIO()
        img.save(buf, format="JPEG", quality=75)
        return {"screenshot": base64.b64encode(buf.getvalue()).decode()}
    except Exception as e:
        raise HTTPException(500, str(e))


# ── Input ──────────────────────────────────────────────────────────────────

@app.post("/type", dependencies=[Depends(require_auth)])
async def type_text(req: dict):
    text = req.get("text", "")
    try:
        import pyautogui
        pyautogui.write(text, interval=0.03)
        return {"ok": True}
    except Exception as e:
        raise HTTPException(500, str(e))


@app.post("/key", dependencies=[Depends(require_auth)])
async def press_keys(req: dict):
    keys = req.get("keys", "")
    try:
        import pyautogui
        parts = [k.strip() for k in keys.split("+")]
        pyautogui.hotkey(*parts)
        return {"ok": True}
    except Exception as e:
        raise HTTPException(500, str(e))


# ── Apps / Browser ─────────────────────────────────────────────────────────

@app.post("/open", dependencies=[Depends(require_auth)])
async def open_target(req: dict):
    target = req.get("target", "")
    try:
        if target.startswith("http://") or target.startswith("https://"):
            webbrowser.open(target)
        else:
            os.startfile(target)
        return {"ok": True}
    except Exception as e:
        raise HTTPException(500, str(e))


@app.post("/search", dependencies=[Depends(require_auth)])
async def search_web(req: dict):
    query = req.get("query", "")
    url = f"https://www.google.com/search?q={quote(query)}"
    webbrowser.open(url)
    return {"ok": True, "url": url}


# ── Health ─────────────────────────────────────────────────────────────────

@app.get("/health")
async def health():
    return {"status": "ok", "host": socket.gethostname()}


def _get_local_ip() -> str:
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
            s.connect(("8.8.8.8", 80))
            return s.getsockname()[0]
    except Exception:
        return socket.gethostbyname(socket.gethostname())


def _ensure_streams() -> None:
    """Unter pythonw (kein Konsolenfenster) sind sys.stdout/stderr = None →
    uvicorns Logging crasht beim Start. Dann auf eine Logdatei umbiegen."""
    import sys
    if sys.stdout is not None and sys.stderr is not None:
        return
    logdir = os.path.join(
        os.environ.get("LOCALAPPDATA", os.path.dirname(os.path.abspath(__file__))),
        "HermesPCExecutor",
    )
    os.makedirs(logdir, exist_ok=True)
    f = open(os.path.join(logdir, "executor.log"), "a", buffering=1, encoding="utf-8")
    sys.stdout = f
    sys.stderr = f


if __name__ == "__main__":
    _ensure_streams()
    my_ip = _get_local_ip()
    auth_state = "Token AKTIV" if AUTH_TOKEN else "KEIN Token → Steuer-Endpunkte GESPERRT (fail-closed)"
    print(f"Hermes PC Executor läuft auf http://{my_ip}:{MY_PORT} (bind {MY_HOST}) — {auth_state}")
    uvicorn.run(app, host=MY_HOST, port=MY_PORT)