"""Approval-gated remote MacBook GUI control over SSH. The server-side tool never talks to cua-driver directly. It validates one fixed computer_use action, obtains approval, and invokes a Mac-local bridge over SSH with a single JSON argv payload. """ from __future__ import annotations import json import logging import os import re import shlex import subprocess import time from typing import Any from hermes_cli.config import load_config from hermes_constants import get_hermes_home from tools.registry import registry logger = logging.getLogger(__name__) _ALLOWED_ACTIONS = { "health", "capture", "click", "double_click", "right_click", "middle_click", "drag", "scroll", "type", "key", "set_value", "wait", "list_apps", "focus_app", "launch_app", "copy_text", } _APPROVAL_ACTIONS = { "capture", "click", "double_click", "right_click", "middle_click", "drag", "scroll", "type", "key", "set_value", "wait", "list_apps", "focus_app", "launch_app", "copy_text", } _MUTATING_ACTIONS = { "click", "double_click", "right_click", "middle_click", "drag", "scroll", "type", "key", "set_value", "focus_app", "launch_app", } _BLOCKED_KEY_COMBOS = { frozenset({"cmd", "shift", "backspace"}), frozenset({"cmd", "option", "backspace"}), frozenset({"cmd", "ctrl", "q"}), frozenset({"cmd", "shift", "q"}), frozenset({"cmd", "option", "shift", "q"}), } _KEY_ALIASES = {"command": "cmd", "control": "ctrl", "alt": "option", "⌘": "cmd", "⌥": "option"} _BLOCKED_TYPE_PATTERNS = [ re.compile(r"curl\s+[^|]*\|\s*bash", re.IGNORECASE), re.compile(r"curl\s+[^|]*\|\s*sh", re.IGNORECASE), re.compile(r"wget\s+[^|]*\|\s*bash", re.IGNORECASE), re.compile(r"\bsudo\s+rm\s+-[rf]", re.IGNORECASE), re.compile(r"\brm\s+-rf\s+/\s*$", re.IGNORECASE), re.compile(r":\s*\(\)\s*\{\s*:\|:\s*&\s*\}", re.IGNORECASE), ] _PAYLOAD_KEYS = { "action", "mode", "app", "max_elements", "element", "coordinate", "button", "modifiers", "from_element", "to_element", "from_coordinate", "to_coordinate", "direction", "amount", "value", "text", "keys", "seconds", "raise_window", "capture_after", } class MacbookGuiError(ValueError): """Expected user-facing validation failure.""" def _config() -> dict[str, Any]: config = load_config() base = dict(config.get("macbook_remote", {}) or {}) gui = dict(base.get("gui", {}) or {}) base["gui"] = gui return base def _enabled_config(cfg: dict[str, Any] | None = None) -> bool: cfg = _config() if cfg is None else cfg gui = dict(cfg.get("gui", {}) or {}) return bool(cfg.get("enabled")) and bool(cfg.get("host")) and bool(cfg.get("user")) and bool(gui.get("enabled")) def _check_requirements() -> bool: try: return _enabled_config() except Exception: return False def _as_int(cfg: dict[str, Any], key: str, default: int, *, minimum: int, maximum: int) -> int: try: value = int(cfg.get(key, default)) except (TypeError, ValueError): value = default return max(minimum, min(maximum, value)) def _ssh_base(cfg: dict[str, Any]) -> list[str]: host = str(cfg.get("host") or "").strip() user = str(cfg.get("user") or "").strip() if not host or not user: raise MacbookGuiError("macbook_remote.host and macbook_remote.user are required") cmd = [ "ssh", "-o", "BatchMode=yes", "-o", f"ConnectTimeout={_as_int(cfg, 'connect_timeout_seconds', 8, minimum=1, maximum=60)}", ] key = str(cfg.get("ssh_key") or "").strip() if key and key.lower() != "default": cmd.extend(["-i", os.path.expanduser(key)]) port = cfg.get("port") if port: cmd.extend(["-p", str(port)]) cmd.append(f"{user}@{host}") return cmd def _gui_config(cfg: dict[str, Any]) -> dict[str, Any]: return dict(cfg.get("gui", {}) or {}) def _allowed_apps(cfg: dict[str, Any]) -> set[str]: return {str(app).strip().lower() for app in _gui_config(cfg).get("allowed_apps", []) if str(app).strip()} def _validate_app(payload: dict[str, Any], cfg: dict[str, Any]) -> None: allowed = _allowed_apps(cfg) if not allowed: return app = str(payload.get("app") or "").strip() action = str(payload.get("action") or "").strip().lower() if not app: if action == "list_apps": return raise MacbookGuiError("app is required when macbook_remote.gui.allowed_apps is configured") # allowed_apps is a pre-approved convenience list, not a hard app firewall. # Unlisted apps still go through the normal approval path and can be made # persistent with `/approve always`, which stores the app/action pattern in # command_allowlist. return def _canon_key_combo(keys: str) -> frozenset[str]: parts = [p.strip().lower() for p in re.split(r"\s*\+\s*", keys) if p.strip()] return frozenset(_KEY_ALIASES.get(p, p) for p in parts) def _validate_safety(payload: dict[str, Any], cfg: dict[str, Any]) -> None: action = payload["action"] blocked_actions = {str(v).strip().lower() for v in _gui_config(cfg).get("blocked_actions", []) if str(v).strip()} if action in blocked_actions: raise MacbookGuiError("action is blocked by macbook_remote.gui.blocked_actions") if action == "focus_app" and payload.get("raise_window") and "focus_app_raise" in blocked_actions: raise MacbookGuiError("raising Mac app windows is blocked") if action == "copy_text" and str(payload.get("app") or "").strip().lower() == "otty": raise MacbookGuiError( "copy_text is not valid for Otty screen/sidebar inspection; use capture mode='som' and read the screenshot" ) if action == "key": combo = _canon_key_combo(str(payload.get("keys") or "")) for blocked in _BLOCKED_KEY_COMBOS: if blocked.issubset(combo): raise MacbookGuiError("destructive system key combo is blocked") if action == "type": text = str(payload.get("text") or "") for pattern in _BLOCKED_TYPE_PATTERNS: if pattern.search(text): raise MacbookGuiError("dangerous shell text pattern is blocked") def _bounded_payload(action: str, args: dict[str, Any]) -> dict[str, Any]: payload = {key: args[key] for key in _PAYLOAD_KEYS if key in args} payload["action"] = action if "seconds" in payload: try: payload["seconds"] = max(0.0, min(30.0, float(payload["seconds"]))) except (TypeError, ValueError): payload["seconds"] = 1.0 if "max_elements" in payload: try: payload["max_elements"] = max(1, min(1000, int(payload["max_elements"]))) except (TypeError, ValueError): payload["max_elements"] = 100 return payload def _approval_description(action: str, payload: dict[str, Any], cfg: dict[str, Any]) -> str: app = str(payload.get("app") or "frontmost app") scope = f"app={app}" if action == "type": text = str(payload.get("text") or "") scope += f", text={text[:80]!r}" + ("..." if len(text) > 80 else "") elif action == "key": scope += f", keys={payload.get('keys', '')!r}" elif payload.get("element") is not None: scope += f", element={payload['element']}" elif payload.get("coordinate") is not None: scope += f", coordinate={payload['coordinate']}" allowed = sorted(_allowed_apps(cfg)) or ["any app"] return ( f"Run MacBook GUI action {action} ({scope}) over SSH via the approved local bridge. " f"Pre-approved apps: {', '.join(allowed)}. Unlisted apps require explicit approval. Raw shell commands are never forwarded." ) def _approval_pattern_key(action: str, payload: dict[str, Any]) -> str: app = str(payload.get("app") or ("all" if action == "list_apps" else "frontmost")).strip().lower() or "frontmost" if action in _MUTATING_ACTIONS: return f"macbook_gui:mutate:{app}:{action}" return f"macbook_gui:read:{app}:{action}" def _audit(action: str, payload: dict[str, Any], ok: bool, error: str = "") -> None: try: log_dir = get_hermes_home() / "logs" log_dir.mkdir(parents=True, exist_ok=True) record = { "ts": int(time.time()), "action": action, "app": payload.get("app"), "ok": ok, "error": error[:500], } with (log_dir / "macbook_gui_audit.jsonl").open("a", encoding="utf-8") as fh: fh.write(json.dumps(record, ensure_ascii=False) + "\n") except Exception: logger.debug("Failed to write MacBook GUI audit log", exc_info=True) def _run_remote_bridge(cfg: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]: bridge_command = str(_gui_config(cfg).get("bridge_command") or "python3 -m hermes_cli.macbook_gui_bridge").strip() if not bridge_command: raise MacbookGuiError("macbook_remote.gui.bridge_command is empty") remote = bridge_command + " " + shlex.quote(json.dumps(payload, ensure_ascii=False)) timeout = _as_int(_gui_config(cfg), "command_timeout_seconds", _as_int(cfg, "command_timeout_seconds", 20, minimum=1, maximum=120), minimum=1, maximum=120) result = subprocess.run(_ssh_base(cfg) + [remote], text=True, capture_output=True, timeout=timeout, check=False) if result.returncode != 0: return {"ok": False, "error": result.stderr.strip() or result.stdout.strip() or f"ssh exited {result.returncode}"} try: return json.loads(result.stdout.strip().splitlines()[-1]) except (json.JSONDecodeError, IndexError) as exc: return {"ok": False, "error": f"invalid bridge response: {exc}"} def _unwrap_bridge_response(response: dict[str, Any]) -> Any: # Newer bridge versions wrap computer_use returns as {"ok": true, # "result": ...}; older deployed bridge copies may return the # computer_use payload directly. Accept both so server rollout does not # have to be lock-step with the Mac-side Hermes checkout. if response.get("ok") is False: return json.dumps({"ok": False, "error": response.get("error") or "MacBook GUI bridge failed"}) if "result" in response: result = response.get("result") return result if isinstance(result, dict) else str(result) return response def macbook_gui(action: str, task_id: str | None = None, **args: Any) -> Any: """Run one approval-gated MacBook GUI action through the remote bridge.""" cfg = _config() if not _enabled_config(cfg): return json.dumps({"ok": False, "error": "macbook_remote.gui is not enabled or is missing host/user config"}) action = (action or "").strip().lower() if action not in _ALLOWED_ACTIONS: return json.dumps({"ok": False, "error": "unsupported macbook_gui action"}) payload = _bounded_payload(action, args) try: if action != "health": _validate_app(payload, cfg) _validate_safety(payload, cfg) except MacbookGuiError as exc: _audit(action, payload, False, str(exc)) return json.dumps({"ok": False, "error": str(exc)}) if action in _APPROVAL_ACTIONS: from tools.approval import request_tool_approval approval = request_tool_approval( action=f"macbook_gui:{action}", description=_approval_description(action, payload, cfg), pattern_key=_approval_pattern_key(action, payload), allow_permanent=True, allow_session=True, ) if not approval.get("approved"): _audit(action, payload, False, approval.get("message", "approval denied")) return json.dumps({"ok": False, "error": approval.get("message") or "approval denied"}) response = _run_remote_bridge(cfg, payload) _audit(action, payload, bool(response.get("ok")), str(response.get("error") or "")) return _unwrap_bridge_response(response) registry.register( name="macbook_gui", toolset="macbook_gui", schema={ "name": "macbook_gui", "description": ( "Approval-gated remote MacBook GUI control for inspecting and driving " "apps on the user's Mac. For requests like 'what is on my Mac/Otty/" "Messages screen?', call action='capture' with mode='som' and inspect " "the returned screenshot yourself; side tabs/panels are visual UI " "and will not appear in copy_text. Do not ask the user to upload a " "screenshot, copy text, expand panels, or run commands for visible " "UI. Use action='list_apps' to see open apps, and action='copy_text' " "only when exact selectable terminal buffer text is required after " "screenshot inspection. " "The Linux server sends fixed actions to a Mac-local bridge; raw shell " "commands are never accepted." ), "parameters": { "type": "object", "properties": { "action": { "type": "string", "enum": sorted(_ALLOWED_ACTIONS), "description": "Fixed Mac GUI action. For screen/state questions, use capture mode='som' and answer from the screenshot, including visible side tabs/panels. Use copy_text only for exact selectable text after screenshot inspection.", }, "mode": {"type": "string", "enum": ["som", "vision", "ax"], "description": "Capture mode. Prefer 'som' for Otty, terminals, and GUI state because it returns a screenshot the model can inspect; AX can miss terminal text."}, "app": {"type": "string", "description": "Mac app name or bundle id, e.g. Otty, Messages, Safari, Terminal, Finder, Google Chrome, or com.apple.MobileSMS. Unlisted apps require approval rather than being hard-blocked."}, "max_elements": {"type": "integer", "minimum": 1, "maximum": 1000}, "element": {"type": "integer"}, "coordinate": {"type": "array", "items": {"type": "integer"}, "minItems": 2, "maxItems": 2}, "button": {"type": "string", "enum": ["left", "right", "middle"]}, "modifiers": {"type": "array", "items": {"type": "string"}}, "from_element": {"type": "integer"}, "to_element": {"type": "integer"}, "from_coordinate": {"type": "array", "items": {"type": "integer"}, "minItems": 2, "maxItems": 2}, "to_coordinate": {"type": "array", "items": {"type": "integer"}, "minItems": 2, "maxItems": 2}, "direction": {"type": "string", "enum": ["up", "down", "left", "right"]}, "amount": {"type": "integer"}, "value": {"type": "string"}, "text": {"type": "string"}, "keys": {"type": "string"}, "seconds": {"type": "number", "maximum": 30}, "raise_window": {"type": "boolean"}, "capture_after": {"type": "boolean"}, }, "required": ["action"], }, }, handler=lambda args, **kw: macbook_gui(task_id=kw.get("task_id"), **args), check_fn=_check_requirements, description="Approval-gated remote MacBook GUI control", emoji="🖥️", max_result_size_chars=20000, )