#!/usr/bin/env python3 """ConnectWise ScreenConnect (Control) API client for the screenconnect skill. Talks to the ACG ScreenConnect instance via the RESTful API Manager extension. Standalone (no third-party hard dependency): prefers httpx, falls back to stdlib urllib. Auth (VERIFIED 2026-06-02 by Howard, re-verified 2026-06-21): HTTP header CTRLAuthHeader: (NO "Basic" prefix; Basic auth 401s) Origin: https://computerguru.screenconnect.com Endpoints live under the RESTful API Manager extension: POST /App_Extensions//Service.ashx/ body = JSON GET is used for read-only methods, POST for state-changing ones; Content-Type is application/json and the body is the method's parameters (object or array). Credentials: never hardcoded. api_secret loaded at runtime from the SOPS vault, or the SCREENCONNECT_API_SECRET env var (testing override). NOTE (instance state, 2026-06-21): the installed RESTful API Manager extension is LIMITED — only `GetSessionsByName` exists; other methods 500 "Web method does not exist". Full control (SendCommandToSession, GetSessions, UpdateSessionCustom- Properties, ...) requires updating the extension on the instance. The client is built to expose those methods as soon as the extension is unlocked; `raw()` probes arbitrary methods in the meantime. """ from __future__ import annotations import base64 import json import os import subprocess import urllib.error import urllib.request from pathlib import Path from typing import Any, Optional try: import httpx # type: ignore _HAS_HTTPX = True except ImportError: # pragma: no cover _HAS_HTTPX = False ERROR_BODY_MAX_CHARS = 500 # ACG instance config (non-secret; matches the vault entry). Env-overridable. SC_BASE_URL = os.environ.get( "SCREENCONNECT_BASE_URL", "https://computerguru.screenconnect.com" ) SC_EXTENSION_GUID = os.environ.get( "SCREENCONNECT_EXTENSION_GUID", "2d558935-686a-4bd0-9991-07539f5fe749" ) SC_TIMEOUT_SECONDS = 60.0 SC_CONNECT_TIMEOUT_SECONDS = 10.0 VAULT_ENTRY = "msp-tools/screenconnect.sops.yaml" VAULT_FIELD = "credentials.api_secret" SKILL_DIR = Path(__file__).resolve().parent.parent # Custom-property mapping on this instance (from the vault notes). CUSTOM_PROPERTIES = {"CP1": "Company", "CP2": "Site", "CP3": "Tag"} class ScreenConnectError(RuntimeError): """Raised for transport or API errors.""" def _resolve_claudetools_root() -> Path: derived_root = SKILL_DIR.parent.parent.parent # .claude/skills/screenconnect -> root env_root = os.environ.get("CLAUDETOOLS_ROOT") if env_root: return Path(env_root) identity_path = derived_root / ".claude" / "identity.json" if identity_path.exists(): try: data = json.loads(identity_path.read_text(encoding="utf-8")) root = data.get("claudetools_root") if root: return Path(root) except (json.JSONDecodeError, OSError): pass return derived_root def load_api_secret() -> str: """Load the ScreenConnect API secret: env override, then the SOPS vault.""" env_secret = os.environ.get("SCREENCONNECT_API_SECRET") if env_secret: return env_secret.strip() root = _resolve_claudetools_root() vault_script = root / ".claude" / "scripts" / "vault.sh" if not vault_script.exists(): raise ScreenConnectError( f"Cannot load API secret: vault wrapper not found at {vault_script} " "and SCREENCONNECT_API_SECRET is not set." ) try: completed = subprocess.run( ["bash", str(vault_script), "get-field", VAULT_ENTRY, VAULT_FIELD], capture_output=True, text=True, timeout=60, ) except FileNotFoundError as exc: raise ScreenConnectError( "Cannot load API secret: 'bash' not found on PATH." ) from exc except subprocess.TimeoutExpired as exc: raise ScreenConnectError("Cannot load API secret: vault call timed out.") from exc if completed.returncode != 0: raise ScreenConnectError( f"Cannot load API secret from vault (exit {completed.returncode}): " f"{completed.stderr.strip()}" ) secret = completed.stdout.strip() if not secret: raise ScreenConnectError("Vault returned an empty API secret.") return secret class ScreenConnectClient: def __init__( self, api_secret: Optional[str] = None, base_url: str = SC_BASE_URL, extension_guid: str = SC_EXTENSION_GUID, timeout: float = SC_TIMEOUT_SECONDS, connect_timeout: float = SC_CONNECT_TIMEOUT_SECONDS, ): self.base_url = base_url.rstrip("/") self.extension_guid = extension_guid self._api_secret = api_secret self.timeout = timeout self.connect_timeout = connect_timeout @property def api_secret(self) -> str: if not self._api_secret: self._api_secret = load_api_secret() return self._api_secret def _service_url(self, method: str) -> str: return ( f"{self.base_url}/App_Extensions/{self.extension_guid}" f"/Service.ashx/{method}" ) def _headers(self) -> dict: return { "CTRLAuthHeader": self.api_secret, "Origin": self.base_url, "Content-Type": "application/json", } def call(self, method: str, body: Any = None, http_method: str = "POST") -> Any: """Call a RESTful API Manager method. Returns parsed JSON (or raw text). `body` is the method's parameters (dict/list); serialized as JSON. Raises ScreenConnectError on a non-2xx response. """ url = self._service_url(method) data = json.dumps(body if body is not None else {}).encode("utf-8") status, text = self._request(url, data, http_method) if status >= 300: snippet = (text or "")[:ERROR_BODY_MAX_CHARS] raise ScreenConnectError( f"ScreenConnect API error [{method}]: HTTP {status}: {snippet}" ) if not text: return None try: return json.loads(text) except json.JSONDecodeError: return text def _request(self, url: str, data: bytes, http_method: str): headers = self._headers() if _HAS_HTTPX: try: timeout = httpx.Timeout(self.timeout, connect=self.connect_timeout) with httpx.Client(timeout=timeout) as client: resp = client.request(http_method, url, content=data, headers=headers) return resp.status_code, resp.text except httpx.HTTPError as exc: raise ScreenConnectError(f"ScreenConnect request failed: {exc}") from exc # stdlib fallback req = urllib.request.Request(url, data=data, method=http_method, headers=headers) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: return resp.status, resp.read().decode("utf-8", errors="replace") except urllib.error.HTTPError as exc: return exc.code, exc.read().decode("utf-8", errors="replace") except urllib.error.URLError as exc: raise ScreenConnectError(f"ScreenConnect request failed: {exc}") from exc # ====================================================================== # VERIFIED methods (work on the current instance) # ====================================================================== def get_sessions_by_name(self, session_name: str = "") -> Any: """List sessions whose Name matches `session_name` (RESTful API Manager GetSessionsByName). VERIFIED LIVE. Empty string returns sessions with a blank Name (the unattended access agents on this instance).""" return self.call("GetSessionsByName", {"sessionName": session_name}) # ====================================================================== # Methods pending the extension unlock (currently 500 "web method does not # exist"). Exposed here so the CLI is ready; verify each once unlocked. # Shapes are best-effort from the RESTful API Manager docs and MUST be # confirmed by live probing before relying on them. # ====================================================================== def get_session_details(self, session_id: str) -> Any: """GetSessionDetailsBySessionID — full detail for one session. PENDING UNLOCK.""" return self.call("GetSessionDetailsBySessionID", {"sessionID": session_id}) def send_command_to_session(self, session_id: str, command: str) -> Any: """SendCommandToSession — run a backstage command on a guest. PENDING UNLOCK. STATE-CHANGING (gate behind --confirm at the call site).""" return self.call( "SendCommandToSession", {"sessionID": session_id, "command": command} ) def send_message_to_session(self, session_id: str, message: str) -> Any: """SendMessageToSession — send a chat message to a guest. PENDING UNLOCK.""" return self.call( "SendMessageToSession", {"sessionID": session_id, "message": message} ) def update_session_custom_properties(self, session_id: str, properties: list) -> Any: """UpdateSessionCustomProperties (CP1=Company, CP2=Site, CP3=Tag). PENDING UNLOCK. STATE-CHANGING.""" return self.call( "UpdateSessionCustomProperties", {"sessionID": session_id, "customProperties": properties}, ) def raw(self, method: str, body: Any = None, http_method: str = "POST") -> Any: """Call any RESTful API Manager method directly (power use / probing).""" return self.call(method, body, http_method=http_method) def main() -> int: """Minimal self-check: load secret (no network call).""" try: client = ScreenConnectClient() _ = client.api_secret print("[OK] API secret loaded; transport =", "httpx" if _HAS_HTTPX else "urllib") return 0 except ScreenConnectError as exc: print(f"[ERROR] {exc}") return 1 if __name__ == "__main__": raise SystemExit(main())