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",
            # Der Executor läuft unter pythonw (kein eigenes Konsolenfenster) — ohne dieses
            # Flag bekäme JEDE gespawnte powershell ein sichtbares Fenster (beim 10-s-Polling
            # der Lucy-Annahme blitzte das im Sekundentakt auf dem Desktop, 12.07.).
            creationflags=subprocess.CREATE_NO_WINDOW,
        )
        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))


# ── Medien & Lautstärke (A3: Lucy steuert den PC per Sprache) ──────────────
# Media-Keys wirken auf den AKTIVEN Player (Spotify, YouTube, VLC …) — genau wie
# die Tasten auf einer Multimedia-Tastatur. Lautstärke = System-Master (2 % je Schritt).
MEDIA_KEYS = {"play_pause": "playpause", "next": "nexttrack", "prev": "prevtrack", "stop": "stop"}
VOLUME_KEYS = {"up": "volumeup", "down": "volumedown", "mute": "volumemute"}


@app.post("/media", dependencies=[Depends(require_auth)])
async def media_control(req: dict):
    action = (req.get("action") or "").strip().lower()
    key = MEDIA_KEYS.get(action)
    if not key:
        raise HTTPException(400, f"Unbekannte Aktion '{action}'. Erlaubt: {', '.join(MEDIA_KEYS)}")
    try:
        import pyautogui
        pyautogui.press(key)
        return {"ok": True, "action": action}
    except Exception as e:
        raise HTTPException(500, str(e))


@app.post("/volume", dependencies=[Depends(require_auth)])
async def volume_control(req: dict):
    action = (req.get("action") or "").strip().lower()
    key = VOLUME_KEYS.get(action)
    if not key:
        raise HTTPException(400, f"Unbekannte Aktion '{action}'. Erlaubt: {', '.join(VOLUME_KEYS)}")
    steps = 1 if action == "mute" else max(1, min(25, int(req.get("steps", 5))))
    try:
        import pyautogui
        pyautogui.press(key, presses=steps, interval=0.02)
        return {"ok": True, "action": action, "steps": steps}
    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)