"""Authenticated loopback HTTP companion for Hermes computer-use activity.""" from __future__ import annotations import argparse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import hashlib import hmac import json import signal import ssl import string import threading from typing import Any, Callable, Mapping from urllib.parse import parse_qs, urlparse from .config import CompanionConfig, TOKEN_ENV from .discovery import MdnsAdvertiser from .drivers import ComputerDriver, build_driver from .hermes_bridge import BridgeError, HermesToolBridge from .hermes_proxy import HermesProxy from .kill_switch import GlobalEscapeSwitch, GlobalKillSwitch from .node import ACTION_POLICY, AuditLog, CapabilityManifest, OperationRegistry, make_operation from .overlay import Overlay, create_overlay from .pairing import PairingError, PairingManager from .security import GrantAuthority, GrantError, PeerIdentity, RevocationRegistry, TrustPolicy from .state import HermesEventConsumer, OverlayStateStore from .window_control import NellozWindowState, minimize_nelloz_window, restore_nelloz_window class _LocalHTTPServer(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True class DesktopCompanion: """Owns the state projection, overlay, hotkey and local API lifecycle.""" def __init__( self, config: CompanionConfig, *, driver: ComputerDriver | None = None, overlay: Overlay | None = None, kill_switch: GlobalKillSwitch | None = None, overtake_switch: GlobalEscapeSwitch | None = None, shutdown_callback: Callable[[], None] | None = None, ) -> None: self.config = config self.state = OverlayStateStore( activity_ttl_seconds=config.activity_ttl_seconds, error_ttl_seconds=config.error_ttl_seconds, ) self.events = HermesEventConsumer(self.state) self.shutdown_callback = shutdown_callback self.driver = driver or build_driver( config.driver, artifact_dir=config.artifact_dir, max_file_read_bytes=config.max_file_read_bytes, ) self.revocations = RevocationRegistry() self.grants = ( GrantAuthority(config.grant_hmac_key, revocations=self.revocations) if config.grant_hmac_key else None ) self.trust = ( TrustPolicy.from_file( config.trust_store_path, allow_empty=config.pairing_enabled, ) if config.remote_listener_enabled and config.trust_store_path is not None else TrustPolicy() ) self.pairing = PairingManager( enabled=config.pairing_enabled, one_time_code=config.pairing_code, sas_confirmed=config.pairing_sas_confirmed, window_seconds=config.pairing_window_seconds, max_attempts=config.pairing_max_attempts, node_id=config.node_id, node_name=config.node_name, server_cert_path=config.mtls_cert_path, allowed_permissions=frozenset(config.pairing_allowed_permissions), expected_client_fingerprint_sha256=( config.pairing_expected_client_fingerprint_sha256 ), ) self.discovery = MdnsAdvertiser( enabled=config.mdns_enabled, host=config.host, port=config.port, node_id=config.node_id, ) self.hermes_proxy = HermesProxy( enabled=config.hermes_proxy_enabled, base_url=config.hermes_url, api_key=config.hermes_api_key, timeout_seconds=900.0, ) self.hermes_bridge = HermesToolBridge( enabled=config.hermes_bridge_enabled, token=config.hermes_bridge_token, port=config.hermes_bridge_port, node_id=config.node_id, node_name=config.node_name, grant_authority=self.grants, action_handler=lambda body, peer: self.start_node_action(body, peer=peer), peer_is_trusted=lambda fingerprint: ( self.trust.identity_for_fingerprint(fingerprint) is not None ), ) self.operations = OperationRegistry() self.audit = AuditLog(config.audit_path) self.manifest = CapabilityManifest(node_id=config.node_id, node_name=config.node_name) self.overlay = overlay or create_overlay( self.state, enabled=config.overlay_enabled, cursor_scale=config.cursor_scale, ) self.kill_switch = kill_switch or GlobalKillSwitch(self.emergency_kill) self.overtake_switch = overtake_switch or GlobalEscapeSwitch(self.abort_overtake) self._overtake_window: NellozWindowState | None = None self._server: _LocalHTTPServer | None = None self._thread: threading.Thread | None = None self._started = threading.Event() self._remote_streams = 0 self._remote_streams_lock = threading.RLock() self._safety_stop = threading.Event() self._safety_thread: threading.Thread | None = None @property def port(self) -> int: if self._server is None: return self.config.port return int(self._server.server_address[1]) @property def running(self) -> bool: return bool(self._thread and self._thread.is_alive()) def start(self) -> None: if self.running: return handler = self._handler_class() self._server = _LocalHTTPServer((self.config.host, self.config.port), handler) try: if self.config.remote_listener_enabled: context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) context.minimum_version = ssl.TLSVersion.TLSv1_3 context.verify_mode = ssl.CERT_REQUIRED assert self.config.mtls_ca_path is not None assert self.config.mtls_cert_path is not None assert self.config.mtls_key_path is not None context.load_verify_locations(cafile=str(self.config.mtls_ca_path)) context.load_cert_chain( certfile=str(self.config.mtls_cert_path), keyfile=str(self.config.mtls_key_path), ) self._server.socket = context.wrap_socket(self._server.socket, server_side=True) except Exception: self._server.server_close() self._server = None raise self.overlay.start() self.kill_switch.start() self.overtake_switch.start() self.pairing.arm() self._started.clear() self._thread = threading.Thread(target=self._serve, name="omni-desktop-http", daemon=True) self._thread.start() if not self._started.wait(timeout=2.0): self.stop() raise RuntimeError("desktop companion did not start") try: self.hermes_bridge.start() except Exception: self.stop() raise self._safety_stop.clear() self._safety_thread = threading.Thread( target=self._safety_monitor, name="omni-desktop-safety", daemon=True, ) self._safety_thread.start() # Never advertise an endpoint until its listener is ready. Existing # trusted peers make a non-pairing boot "paired" even though the # one-time claim window itself is closed. pairing_state = self.pairing.state if pairing_state == "closed" and self.trust.manifest()["trusted_peers"]: pairing_state = "paired" self.discovery.port = self.port self.discovery.start(pairing_state=pairing_state) if self.config.pairing_enabled: self.audit.record( "pairing_started", node_id=self.config.node_id, payload={"window_seconds": self.config.pairing_window_seconds}, ) def _serve(self) -> None: self._started.set() server = self._server if server is not None: server.serve_forever(poll_interval=0.1) def stop(self) -> None: # Remove discovery metadata first so a shutting-down endpoint is not # returned to new clients. self.sync_overtake_window(False) self.discovery.stop() self.hermes_bridge.stop() self._safety_stop.set() safety_thread, self._safety_thread = self._safety_thread, None if ( safety_thread and safety_thread.is_alive() and safety_thread is not threading.current_thread() ): safety_thread.join(timeout=1.0) server, self._server = self._server, None if server is not None: server.shutdown() server.server_close() thread, self._thread = self._thread, None if thread and thread.is_alive() and thread is not threading.current_thread(): thread.join(timeout=2.0) self.kill_switch.stop() self.overtake_switch.stop() self.overlay.stop() def _safety_monitor(self) -> None: while not self._safety_stop.wait(0.1): if ( self.driver.kind == "windows" and self.operations.active_webcams() and not self.overlay.status.running ): self._fail_closed_webcams("camera_indicator_unavailable") def emergency_kill(self, source: str = "Emergency stop") -> dict[str, object]: was_killed = self.state.snapshot().killed active_cameras = self.operations.active_webcams() stopped = self.operations.stop_matching(reason="emergency_stop") self.revocations.revoke_all() self.hermes_bridge.revoke_all() snapshot = self.state.kill(source) self.sync_overtake_window(False) self.driver.emergency_stop() if not was_killed: for camera in active_cameras: self.audit.record( "camera_stopped", node_id=self.config.node_id, session_id=camera.session_id, operation_id=camera.operation_id, trace_id=camera.trace_id, payload={"action_id": camera.action_id, "reason": "emergency_stop"}, ) self.audit.record( "emergency_stop", node_id=self.config.node_id, payload={"source": source[:80], "stopped_operations": len(stopped)}, ) return snapshot.as_dict() def abort_overtake(self, source: str = "Esc") -> dict[str, object]: """Stop only active computer-control actions and clear the Overtake UI.""" stopped: list[str] = [] for operation in self.operations.active(): if operation.action_type != "desktop.control": continue self.driver.stop(operation.action_id) self.operations.stop(operation.action_id, "overtake_escape") stopped.append(operation.action_id) self.hermes_bridge.revoke_all() snapshot = self.state.set_overtake(False) self.sync_overtake_window(False) self.audit.record( "overtake_stopped", node_id=self.config.node_id, payload={"source": source[:80], "stopped_actions": len(stopped)}, ) return { "aborted": True, "stopped_action_ids": stopped, "state": snapshot.as_dict(), } def sync_overtake_window(self, active: bool) -> None: """Minimize Nelloz while Hermes owns the desktop, then restore it on exit.""" if active: if self._overtake_window is None: self._overtake_window = minimize_nelloz_window() return if self._overtake_window is not None: restore_nelloz_window(self._overtake_window) self._overtake_window = None def health(self) -> dict[str, object]: snapshot = self.state.snapshot() hermes_health = self.hermes_proxy.health() status = "killed" if snapshot.killed else "ready" if self.config.hermes_proxy_enabled and hermes_health.get("status") != "ready": status = "unavailable" if self.config.hermes_proxy_enabled and not self.hermes_bridge.running: status = "unavailable" return { "status": status, "service": "omni-desktop-companion", "bind": f"{self.config.host}:{self.port}", "driver": self.driver.kind, "overlay": self.overlay.status.as_dict(), "kill_switch": { "hotkey": self.kill_switch.HOTKEY, "registered": self.kill_switch.registered, "error": self.kill_switch.error, }, "overtake": { "active": snapshot.overtake_active, "cancel_key": "Esc", "escape_registered": self.overtake_switch.registered, "escape_error": self.overtake_switch.error, }, "revision": snapshot.revision, "node_id": self.config.node_id, "remote_listener": self.config.remote_listener_enabled, "active_operations": len(self.operations.active()), "active_cameras": len(self.operations.active_webcams()), "revocations": self.revocations.as_dict(), "trust": self.trust.manifest(), "pairing_state": self.pairing.state, "discovery": self.discovery.status.as_dict(), "hermes": hermes_health, "hermes_tool_bridge": self.hermes_bridge.status(), } def paired_capabilities(self) -> list[str]: capabilities = set(ACTION_POLICY) if self.config.hermes_proxy_enabled: capabilities.update( { "hermes.run", "qwen.inference", "computer.use", "computer.desktop", "code", "files", "tools", "webcam", "internet.search", "context.vision", } ) return sorted(capabilities) def claim_pairing( self, body: Mapping[str, Any], *, client_certificate_der: bytes, ) -> dict[str, object]: endpoint = f"https://{self.config.host}:{self.port}" receipt = self.pairing.claim( body, client_certificate_der=client_certificate_der, endpoint=endpoint, trust=self.trust, capabilities=self.paired_capabilities(), on_paired=lambda: self.discovery.update_pairing_state("paired"), ) self.audit.record( "pairing_completed", node_id=self.config.node_id, payload={ "peer_fingerprint_suffix": hashlib.sha256(client_certificate_der).hexdigest()[-12:], "granted_permissions": receipt["granted_permissions"], }, ) return receipt def capability_manifest(self) -> dict[str, object]: runtime_capabilities = dict(self.driver.runtime_capabilities()) if self.driver.kind == "windows" and not self.overlay.status.running: runtime_capabilities["webcam"] = False return self.manifest.build( driver_kind=self.driver.kind, remote_listener=self.config.remote_listener_enabled, healthy=not self.state.snapshot().killed, active_operations=len(self.operations.active()), signing_key=self.config.grant_hmac_key or None, runtime_capabilities=runtime_capabilities, provider_capabilities=set(self.paired_capabilities()), hermes_proxy_enabled=self.config.hermes_proxy_enabled, ) def begin_remote_run_stream(self, peer: PeerIdentity | None) -> None: with self._remote_streams_lock: self._remote_streams += 1 self.state.set_remote_activity( True, node_name=peer.name if peer is not None else "Remote conductor", ) def end_remote_run_stream(self) -> None: with self._remote_streams_lock: self._remote_streams = max(0, self._remote_streams - 1) active = self._remote_streams > 0 if not active: self._refresh_indicators() def start_node_action( self, body: Mapping[str, Any], *, peer: PeerIdentity | None, ) -> tuple[int, dict[str, object]]: target_node_id = _identifier(body.get("node_id"), "node_id") session_id = _identifier(body.get("session_id"), "session_id") operation_id = _identifier(body.get("operation_id"), "operation_id") trace_id = _identifier(body.get("trace_id"), "trace_id") if target_node_id != self.config.node_id: self._audit_denied(session_id, operation_id, trace_id, "target_node_mismatch") raise _NodeRequestError(409, "target_node_mismatch", "request targets another node") action = body.get("action") if not isinstance(action, Mapping): raise _NodeRequestError(400, "invalid_action", "action must be an object") action_id = _identifier(action.get("action_id") or action.get("id"), "action.action_id") action_type = str(action.get("type") or "") policy = ACTION_POLICY.get(action_type) if policy is None: raise _NodeRequestError(400, "unsupported_action", "unknown node action type") capability, expected_permission = policy permission = str(action.get("permission") or "") if permission != expected_permission: self._audit_denied(session_id, operation_id, trace_id, "permission_mismatch") raise _NodeRequestError(403, "permission_mismatch", "action permission does not match its type") arguments = action.get("arguments", {}) if not isinstance(arguments, Mapping): raise _NodeRequestError(400, "invalid_arguments", "action arguments must be an object") _validate_action_arguments(action_type, arguments) source_node_id = peer.node_id if peer else _optional_identifier(body.get("source_node_id"), "local-hermes") source_node_name = ( peer.name if peer else " ".join(str(body.get("source_node_name") or "Local Hermes").split())[:80] ) source_fingerprint = peer.certificate_fingerprint_sha256 if peer else None if peer is not None: allowed_roles = { capability, permission, "desktop.node.client", "full_qwen_and_hermes_feature_parity", } if not allowed_roles.intersection(peer.roles): self._audit_denied(session_id, operation_id, trace_id, "certificate_role_denied") raise _NodeRequestError(403, "certificate_role_denied", "client certificate role is insufficient") existing = self.operations.get(action_id) if existing is not None: if ( existing.operation_id == operation_id and existing.session_id == session_id and existing.action_type == action_type ): return 200, {"idempotent_replay": True, "operation": existing.as_dict()} raise _NodeRequestError(409, "action_id_conflict", "action_id is already in use") if action_type == "webcam.stop": device_id = str(arguments.get("device_id") or "") stopped = [] for camera in self.operations.active_webcams(): if camera.session_id != session_id: continue if device_id and camera.resource_id != device_id: continue stopped.append(self.stop_node_action(camera.action_id, reason="camera_stop", peer=peer)) self.audit.record( "camera_stopped", node_id=self.config.node_id, session_id=session_id, operation_id=operation_id, trace_id=trace_id, payload={"requested_action_id": action_id, "stopped_count": len(stopped)}, ) return 200, {"stopped": stopped, "idempotent": True} if action_type == "webcam.frame": device_id = str(arguments.get("device_id") or "") matching_camera = any( item.session_id == session_id and item.resource_id == device_id for item in self.operations.active_webcams() ) if not matching_camera: raise _NodeRequestError(409, "camera_not_active", "camera must be started before frame status") if self.grants is None: self._audit_denied(session_id, operation_id, trace_id, "grant_verification_unavailable") raise _NodeRequestError(503, "grant_verification_unavailable", "signed grant verification is not configured") grant = body.get("grant") if not isinstance(grant, Mapping): self._audit_denied(session_id, operation_id, trace_id, "grant_required") raise _NodeRequestError(403, "grant_required", "a signed scoped grant is required") try: verified = self.grants.verify( grant, node_id=self.config.node_id, session_id=session_id, capability=capability, arguments=arguments, ) except GrantError as exc: self._audit_denied(session_id, operation_id, trace_id, exc.code) raise _NodeRequestError(403, exc.code, exc.detail) from None if ( action_type in {"webcam.start", "webcam.frame"} and self.driver.kind == "windows" and not self.overlay.status.running ): self._audit_denied( session_id, operation_id, trace_id, "camera_indicator_unavailable", ) self._fail_closed_webcams("camera_indicator_unavailable") raise _NodeRequestError( 503, "camera_indicator_unavailable", "real webcam access requires a running visible local indicator", ) resource_id = None if action_type.startswith("webcam."): resource_id = str(arguments.get("device_id") or "")[:160] or None elif action_type == "desktop.screenshot": resource_id = str(arguments.get("screen_id") or "primary")[:160] self.state.set_remote_activity(True, node_name=source_node_name) if action_type in {"desktop.screenshot", "desktop.control"}: self.events.consume( { "type": "hermes.computer-use.action", "tool": "computer", "action": { "type": ( str(arguments.get("action") or arguments.get("operation") or "observe") if action_type == "desktop.control" else "observe" ), **dict(arguments), }, } ) self.sync_overtake_window(self.state.snapshot().overtake_active) if action_type == "webcam.start": self.state.set_camera_activity( True, device_name=str(arguments.get("device_name") or resource_id or "Camera"), session_id=session_id, operation_id=operation_id, ) driver_result = self.driver.execute( {"id": action_id, "type": action_type, "arguments": dict(arguments)} ) if not driver_result.accepted: self._refresh_indicators() raise _NodeRequestError(409, "driver_rejected", driver_result.detail or "driver rejected action") operation = make_operation( action_id=action_id, action_type=action_type, permission=permission, capability=capability, node_id=self.config.node_id, source_node_id=source_node_id, source_node_name=source_node_name, source_fingerprint=source_fingerprint, session_id=session_id, operation_id=operation_id, trace_id=trace_id, grant_id=verified.grant_id, resource_id=resource_id, performed=driver_result.performed, simulated=not driver_result.performed, initial_status=( "running" if driver_result.status in {"running", "mocked", "noop"} else "completed" ), ) operation, created = self.operations.start(operation) if created: self.audit.record( "tool_started", node_id=self.config.node_id, session_id=session_id, operation_id=operation_id, trace_id=trace_id, payload={ "action_id": action_id, "action_type": action_type, "capability": capability, "grant_id": verified.grant_id, "source_node_id": source_node_id, }, ) if action_type == "webcam.start": self.audit.record( "camera_started", node_id=self.config.node_id, session_id=session_id, operation_id=operation_id, trace_id=trace_id, payload={"action_id": action_id, "device_id": resource_id}, ) if operation.status == "completed": self.audit.record( "tool_completed", node_id=self.config.node_id, session_id=session_id, operation_id=operation_id, trace_id=trace_id, payload={"action_id": action_id, "status": "completed"}, ) self._refresh_indicators() return 202, { "operation": operation.as_dict(), "driver": driver_result.as_dict(), "raw_media": None, } def node_action(self, action_id: str, *, peer: PeerIdentity | None) -> dict[str, object]: operation = self.operations.get(action_id) if operation is None: raise _NodeRequestError(404, "action_not_found", "unknown action_id") if peer is not None and operation.source_fingerprint != peer.certificate_fingerprint_sha256: raise _NodeRequestError(403, "action_owner_mismatch", "action belongs to another peer") return operation.as_dict() def stop_node_action( self, action_id: str, *, reason: str, peer: PeerIdentity | None, ) -> dict[str, object]: operation = self.operations.get(action_id) if operation is None: raise _NodeRequestError(404, "action_not_found", "unknown action_id") if peer is not None and operation.source_fingerprint != peer.certificate_fingerprint_sha256: raise _NodeRequestError(403, "action_owner_mismatch", "action belongs to another peer") was_running = operation.status == "running" if was_running: self.driver.stop(action_id) stopped = self.operations.stop(action_id, reason) assert stopped is not None if was_running: if operation.action_type == "webcam.start": self.audit.record( "camera_stopped", node_id=self.config.node_id, session_id=operation.session_id, operation_id=operation.operation_id, trace_id=operation.trace_id, payload={"action_id": action_id, "reason": reason[:80]}, ) self.audit.record( "tool_completed", node_id=self.config.node_id, session_id=operation.session_id, operation_id=operation.operation_id, trace_id=operation.trace_id, payload={"action_id": action_id, "status": stopped.status}, ) self._refresh_indicators() return stopped.as_dict() def revoke(self, body: Mapping[str, Any], *, peer: PeerIdentity | None) -> dict[str, object]: grant_id = str(body.get("grant_id") or "")[:160] or None session_id = str(body.get("session_id") or "")[:160] or None fingerprint = str(body.get("peer_fingerprint_sha256") or "").lower() or None revoke_all = body.get("all") is True reason = str(body.get("reason") or "operator_request")[:80] if not any((grant_id, session_id, fingerprint, revoke_all)): raise _NodeRequestError(400, "revoke_target_required", "specify grant, session, peer or all") if peer is not None and fingerprint is not None and fingerprint != peer.certificate_fingerprint_sha256: raise _NodeRequestError(403, "peer_revoke_denied", "remote peer may only revoke itself") stopped = [] if grant_id: self.revocations.revoke_grant(grant_id) stopped.extend(self.operations.stop_matching(reason="revoked_grant", grant_id=grant_id)) self.audit.record( "capability_revoked", node_id=self.config.node_id, payload={"grant_id": grant_id, "reason": reason}, ) if session_id: self.revocations.revoke_session(session_id) stopped.extend(self.operations.stop_matching(reason="revoked_session", session_id=session_id)) self.audit.record( "capability_revoked", node_id=self.config.node_id, session_id=session_id, payload={"scope": "session", "reason": reason}, ) if fingerprint: self.trust.revoke(fingerprint) self.hermes_bridge.revoke_peer(fingerprint) stopped.extend( self.operations.stop_matching( reason="revoked_pairing", source_fingerprint=fingerprint, ) ) self.audit.record( "pairing_revoked", node_id=self.config.node_id, payload={"fingerprint_suffix": fingerprint[-12:], "reason": reason}, ) if revoke_all: self.revocations.revoke_all() stopped.extend(self.operations.stop_matching(reason="revoked_all")) self.audit.record( "capability_revoked", node_id=self.config.node_id, payload={"scope": "all", "reason": reason}, ) for operation in {item.action_id: item for item in stopped}.values(): self.driver.stop(operation.action_id) if operation.action_type == "webcam.start": self.audit.record( "camera_stopped", node_id=self.config.node_id, session_id=operation.session_id, operation_id=operation.operation_id, trace_id=operation.trace_id, payload={"action_id": operation.action_id, "reason": "revoked"}, ) self._refresh_indicators() return { "revoked": True, "stopped_action_ids": sorted({item.action_id for item in stopped}), "state": self.revocations.as_dict(), "controller_follow_up": ( "remove_peer_from_persistent_allowlist_and_revoke_certificate" if fingerprint else None ), } def _refresh_indicators(self) -> None: active = self.operations.active() if active: current = active[-1] self.state.set_remote_activity(True, node_name=current.source_node_name) else: self.state.set_remote_activity(False) cameras = self.operations.active_webcams() if cameras: camera = cameras[-1] self.state.set_camera_activity( True, device_name=camera.resource_id or "Camera", session_id=camera.session_id, operation_id=camera.operation_id, ) else: self.state.set_camera_activity(False) def _fail_closed_webcams(self, reason: str) -> None: cameras = self.operations.active_webcams() for camera in cameras: self.driver.stop(camera.action_id) self.operations.stop(camera.action_id, reason) self.audit.record( "camera_stopped", node_id=self.config.node_id, session_id=camera.session_id, operation_id=camera.operation_id, trace_id=camera.trace_id, payload={"action_id": camera.action_id, "reason": reason[:80]}, ) if cameras: self.audit.record( "capability_denied", node_id=self.config.node_id, payload={"reason": reason, "stopped_cameras": len(cameras)}, ) self._refresh_indicators() def _audit_denied(self, session_id: str, operation_id: str, trace_id: str, reason: str) -> None: self.audit.record( "capability_denied", node_id=self.config.node_id, session_id=session_id, operation_id=operation_id, trace_id=trace_id, payload={"reason": reason}, ) def _handler_class(self) -> type[BaseHTTPRequestHandler]: companion = self class Handler(BaseHTTPRequestHandler): server_version = "OMNIDesktop/1" sys_version = "" def log_message(self, _format: str, *_args: object) -> None: # Avoid leaking request metadata or bearer values to stdout. return None def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API parsed = urlparse(self.path) path = parsed.path proxy_request = path.startswith("/v1/runs") mtls_only = companion.config.remote_listener_enabled and ( proxy_request or path in {"/health", "/healthz", "/v1/capabilities"} ) if not self._authorized(allow_mtls_only=mtls_only): return if proxy_request: run_event_stream = path.rstrip("/").endswith("/events") if run_event_stream: companion.begin_remote_run_stream(self.peer_identity) try: result = companion.hermes_proxy.forward( self, method="GET", path=self.path, body=None, killed=lambda: companion.state.snapshot().killed, ) if run_event_stream and result is not None: parts = [part for part in path.split("/") if part] if len(parts) >= 3: companion.hermes_bridge.revoke_run(parts[2]) finally: if run_event_stream: companion.end_remote_run_stream() return if path in {"/health", "/healthz"}: self._json(200, companion.health()) elif path == "/v1/state": self._json(200, companion.state.snapshot().as_dict()) elif path == "/v1/capabilities": self._json(200, companion.capability_manifest()) elif path == "/v1/camera/status": camera = companion.state.snapshot().as_dict()["camera"] self._json(200, {"camera": camera, "captures": [item.as_dict() for item in companion.operations.active_webcams()]}) elif path == "/v1/audit": if self.peer_identity is not None and not { "audit.read", "node.admin", }.intersection(self.peer_identity.roles): self._json(403, {"error": "audit_role_required"}) return query = parse_qs(parsed.query) try: after_seq = max(0, int(query.get("after_seq", ["0"])[0])) limit = min(1000, max(0, int(query.get("limit", ["1000"])[0]))) except ValueError: self._json(400, {"error": "invalid_audit_cursor"}) return events = companion.audit.replay(after_seq=after_seq, limit=limit) self._json( 200, { "protocol_version": "1.0", "events": events, "last_seq": events[-1]["seq"] if events else after_seq, }, ) elif path.startswith("/v1/node/actions/") and path.count("/") == 4: action_id = path.rsplit("/", 1)[-1] try: self._json(200, companion.node_action(action_id, peer=self.peer_identity)) except _NodeRequestError as exc: self._json(exc.status, {"error": exc.code, "detail": exc.detail}) else: self._json(404, {"error": "not_found"}) def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API path = self.path.partition("?")[0] if path == "/v1/pairing/claim": certificate = self._pairing_certificate() if certificate is None: return try: body = self._read_object() except _HTTPInputError as exc: self._json(exc.status, {"error": exc.code, "detail": exc.detail}) return try: receipt = companion.claim_pairing( body, client_certificate_der=certificate, ) except PairingError as exc: companion.audit.record( "pairing_denied", node_id=companion.config.node_id, payload={"reason": exc.code}, ) self._json(exc.status, {"error": exc.code, "detail": exc.detail}) else: self._json(201, receipt) return proxy_request = path.startswith("/v1/runs") if not self._authorized( allow_mtls_only=companion.config.remote_listener_enabled and proxy_request ): return if proxy_request: try: body = self._read_optional_object() except _HTTPInputError as exc: self._json(exc.status, {"error": exc.code, "detail": exc.detail}) return run_ticket: str | None = None if path.rstrip("/") == "/v1/runs": if body is None or self.peer_identity is None: self._json(400, {"error": "run_body_required"}) return try: body, run_ticket = companion.hermes_bridge.prepare_run( body, peer=self.peer_identity, ) except BridgeError as exc: self._json(exc.status, {"error": exc.code, "detail": exc.detail}) return if path.rstrip("/") == "/v1/runs" and body is not None: companion.audit.record( "operation_routed", node_id=companion.config.node_id, session_id=str(body.get("session_id") or "unknown"), operation_id=str(body.get("operation_id") or "hermes_run"), trace_id=str(body.get("trace_id") or "unknown"), payload={"provider": "local_hermes", "remote_peer": self.peer_identity.node_id if self.peer_identity else "local"}, ) result = companion.hermes_proxy.forward( self, method="POST", path=self.path, body=body, killed=lambda: companion.state.snapshot().killed, ) if run_ticket is not None: if result is not None and 200 <= result.status < 300: companion.hermes_bridge.bind_run(run_ticket, result.payload) else: companion.hermes_bridge.revoke_ticket(run_ticket) if path.rstrip("/").endswith("/stop"): parts = [part for part in path.split("/") if part] if len(parts) >= 3: companion.hermes_bridge.revoke_run(parts[2]) return try: body = self._read_object() except _HTTPInputError as exc: self._json(exc.status, {"error": exc.code, "detail": exc.detail}) return if path == "/v1/kill": reason = str(body.get("reason") or "Remote emergency stop")[:240] self._json(200, {"state": companion.emergency_kill(reason)}) return if path == "/v1/overtake/stop": reason = str(body.get("reason") or "Esc")[:80] self._json(200, companion.abort_overtake(reason)) return if path == "/v1/shutdown": self._json(202, {"ok": True, "stopping": True}) callback = companion.shutdown_callback if callback is not None: threading.Thread(target=callback, name="nelloz-companion-stop", daemon=True).start() return if path == "/v1/revoke": try: result = companion.revoke(body, peer=self.peer_identity) except _NodeRequestError as exc: self._json(exc.status, {"error": exc.code, "detail": exc.detail}) else: self._json(200, result) return if path.startswith("/v1/node/actions/") and path.endswith("/stop"): action_id = path.removesuffix("/stop").rsplit("/", 1)[-1] try: result = companion.stop_node_action( action_id, reason=str(body.get("reason") or "stop_requested")[:80], peer=self.peer_identity, ) except _NodeRequestError as exc: self._json(exc.status, {"error": exc.code, "detail": exc.detail}) else: self._json(200, {"operation": result, "idempotent": True}) return if companion.state.snapshot().killed: self._json(423, {"error": "kill_switch_active"}) return if path == "/v1/node/actions": try: status, result = companion.start_node_action(body, peer=self.peer_identity) except _NodeRequestError as exc: self._json(exc.status, {"error": exc.code, "detail": exc.detail}) else: self._json(status, result) return if path == "/v1/events": event = body.get("event", body) if not isinstance(event, Mapping): self._json(400, {"error": "event_must_be_an_object"}) return consumed = companion.events.consume(event) companion.sync_overtake_window(companion.state.snapshot().overtake_active) self._json( 202, { "consumed": consumed, "state": companion.state.snapshot().as_dict(), }, ) return if path == "/v1/actions": if companion.config.remote_listener_enabled: self._json(403, {"error": "targeted_action_contract_required"}) return action = body.get("action", body) if not isinstance(action, Mapping): self._json(400, {"error": "action_must_be_an_object"}) return event = { "type": "hermes.computer-use.action", "tool": "computer", "action": dict(action), } companion.events.consume(event) companion.sync_overtake_window(companion.state.snapshot().overtake_active) result = companion.driver.execute(action) self._json( 202 if result.accepted else 409, { "result": result.as_dict(), "state": companion.state.snapshot().as_dict(), }, ) return self._json(404, {"error": "not_found"}) def _authorized(self, *, allow_mtls_only: bool = False) -> bool: self.peer_identity: PeerIdentity | None = None if companion.config.remote_listener_enabled: try: certificate = self.connection.getpeercert(binary_form=True) except (AttributeError, ssl.SSLError): certificate = None identity = ( companion.trust.identity_for_certificate(certificate) if isinstance(certificate, bytes) and certificate else None ) if identity is None: self._json(403, {"error": "untrusted_client_certificate"}) return False self.peer_identity = identity if allow_mtls_only: return True header = self.headers.get("Authorization", "") scheme, separator, credential = header.partition(" ") authorized = ( bool(separator) and scheme.lower() == "bearer" and hmac.compare_digest( credential.strip().encode("utf-8"), companion.config.token.encode("utf-8"), ) ) if not authorized: self.send_response(401) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("WWW-Authenticate", 'Bearer realm="omni-desktop"') payload = b'{"error":"unauthorized"}' self.send_header("Content-Length", str(len(payload))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(payload) return False return True def _pairing_certificate(self) -> bytes | None: if not companion.config.remote_listener_enabled or not companion.config.pairing_enabled: self._json(404, {"error": "pairing_not_enabled"}) return None try: certificate = self.connection.getpeercert(binary_form=True) except (AttributeError, ssl.SSLError): certificate = None if not isinstance(certificate, bytes) or not certificate: self._json(403, {"error": "client_certificate_required"}) return None return certificate def _read_object(self) -> dict[str, Any]: content_type = self.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() if content_type != "application/json": raise _HTTPInputError(415, "unsupported_media_type", "use application/json") raw_length = self.headers.get("Content-Length") try: length = int(raw_length or "") except ValueError as exc: raise _HTTPInputError(411, "content_length_required", "valid Content-Length required") from exc if length < 0: raise _HTTPInputError(400, "invalid_content_length", "negative body length") if length > companion.config.max_body_bytes: raise _HTTPInputError(413, "payload_too_large", "request body exceeds limit") try: value = json.loads(self.rfile.read(length).decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise _HTTPInputError(400, "invalid_json", "malformed JSON object") from exc if not isinstance(value, dict): raise _HTTPInputError(400, "body_must_be_an_object", "JSON root must be an object") return value def _read_optional_object(self) -> dict[str, Any] | None: raw_length = self.headers.get("Content-Length") if raw_length in (None, "", "0"): return None return self._read_object() def _json(self, status: int, value: Mapping[str, object]) -> None: payload = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") self.end_headers() self.wfile.write(payload) return Handler class _HTTPInputError(ValueError): def __init__(self, status: int, code: str, detail: str) -> None: super().__init__(detail) self.status = status self.code = code self.detail = detail class _NodeRequestError(ValueError): def __init__(self, status: int, code: str, detail: str) -> None: super().__init__(detail) self.status = status self.code = code self.detail = detail def _identifier(value: object, field_name: str) -> str: text = str(value or "") valid = ( 1 <= len(text) <= 160 and text[0] in string.ascii_letters + string.digits and all(character in string.ascii_letters + string.digits + "_.:-" for character in text) ) if not valid: raise _NodeRequestError(400, "invalid_identifier", f"{field_name} is invalid") return text def _optional_identifier(value: object, default: str) -> str: try: return _identifier(value, "source_node_id") if value else default except _NodeRequestError: return default def _validate_action_arguments(action_type: str, arguments: Mapping[str, Any]) -> None: if action_type == "desktop.control": operation = str(arguments.get("action") or arguments.get("operation") or "") allowed = { "move", "click", "double_click", "right_click", "drag", "scroll", "type", "hotkey", "clipboard.read", "clipboard.write", } if operation not in allowed: raise _NodeRequestError(400, "invalid_computer_action", "desktop control action is not allowed") if operation == "type": text = arguments.get("text", "") if not isinstance(text, str) or len(text) > 8192: raise _NodeRequestError(400, "invalid_text", "typed text must be at most 8192 characters") if operation == "clipboard.write": text = arguments.get("text", "") if not isinstance(text, str) or len(text) > 1_000_000: raise _NodeRequestError(400, "invalid_text", "clipboard text is too large") if operation == "hotkey": keys = arguments.get("keys") if isinstance(keys, str): keys = [part for part in keys.split("+") if part] if not isinstance(keys, list) or not 1 <= len(keys) <= 8: raise _NodeRequestError(400, "invalid_hotkey", "hotkey requires one to eight keys") if any(not isinstance(key, str) or not 1 <= len(key) <= 24 for key in keys): raise _NodeRequestError(400, "invalid_hotkey", "hotkey names are invalid") elif action_type in {"files.read", "files.write"}: path = arguments.get("path") or arguments.get("source") if not isinstance(path, str) or not path or len(path) > 32_767: raise _NodeRequestError(400, "file_path_required", "file action requires a bounded path") elif action_type == "desktop.screenshot": screen_id = arguments.get("screen_id") if screen_id is not None and (not isinstance(screen_id, str) or len(screen_id) > 160): raise _NodeRequestError(400, "invalid_screen_id", "screen_id is invalid") all_screens = arguments.get("all_screens") if all_screens is not None and not isinstance(all_screens, bool): raise _NodeRequestError(400, "invalid_all_screens", "all_screens must be boolean") if all_screens is True and screen_id is not None: raise _NodeRequestError(400, "ambiguous_screen_scope", "choose screen_id or all_screens") elif action_type in {"webcam.start", "webcam.frame"}: device_id = arguments.get("device_id") if not isinstance(device_id, str) or not device_id or len(device_id) > 160: raise _NodeRequestError(400, "camera_device_required", "webcam action requires device_id") def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="OMNI authenticated Windows desktop companion") parser.add_argument("--host", default="127.0.0.1", help="literal loopback address only") parser.add_argument("--port", type=int, default=8765) parser.add_argument("--token", help=f"Bearer credential; defaults to ${TOKEN_ENV}") parser.add_argument("--driver", choices=("noop", "mock", "windows"), default="noop") parser.add_argument("--no-overlay", action="store_true") parser.add_argument( "--remote-listener", action="store_true", default=None, help="require mTLS/trust environment and allow the explicit LAN host", ) parser.add_argument("--node-id", default=None) parser.add_argument("--node-name", default=None) parser.add_argument( "--enable-real-control", action="store_true", default=None, help="explicitly unlock the grant-gated real Windows driver", ) return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) try: overrides: dict[str, object] = { "host": args.host, "port": args.port, "token": args.token if args.token is not None else __import__("os").environ.get(TOKEN_ENV, ""), "driver": args.driver, "overlay_enabled": not args.no_overlay, } if args.remote_listener is not None: overrides["remote_listener_enabled"] = args.remote_listener if args.node_id is not None: overrides["node_id"] = args.node_id if args.node_name is not None: overrides["node_name"] = args.node_name if args.enable_real_control is not None: overrides["enable_real_control"] = args.enable_real_control config = CompanionConfig.from_env( **overrides, ) except ValueError as exc: _parser().error(str(exc)) return 2 finished = threading.Event() companion = DesktopCompanion(config, shutdown_callback=finished.set) def request_stop(_signum: int, _frame: object) -> None: finished.set() signal.signal(signal.SIGINT, request_stop) if hasattr(signal, "SIGTERM"): signal.signal(signal.SIGTERM, request_stop) companion.start() try: finished.wait() finally: companion.stop() return 0 if __name__ == "__main__": raise SystemExit(main())