#!/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). INSTANCE METHOD SURFACE (probed live 2026-06-22, extended 2026-07-06): control IS available - GetSessionsByFilter, GetSessionsByName, GetSessionDetailsBySessionID, GetSessionBySessionID (reads, JSON-object params); SendCommandToSession, SendMessageToSession, UpdateSessionCustomProperties, CreateSession (writes, POSITIONAL-ARRAY params). FULL-FLEET LISTING (fixed 2026-07-06): GetSessionsByFilter({"sessionFilter": }) IS exposed and is the enumeration path. It takes a ScreenConnect session-filter expression, e.g. SessionType = 'Access' (all 958 unattended agents), CustomProperty1 = 'Safesite' (per-client), Name LIKE '*DELL*' (partial name). SessionType 'Access' (=2) is the agent fleet; 'Support' (=0) the on-demand sessions. The bare aliases GetSessions/GetAllSessions/GetSessionGroups remain absent, but GetSessionsByFilter covers the inventory use case fully - no extension update needed. `raw()` probes arbitrary methods. """ 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 from urllib.parse import quote 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"} # SessionType filter literals (verified live 2026-07-06): 'Access' (=2) is the # unattended agent fleet; 'Support' (=0) the on-demand sessions. SESSION_TYPE_LITERALS = {"access": "Access", "support": "Support", "meeting": "Meeting"} 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 EXACTLY equals `session_name` (RESTful API Manager GetSessionsByName). VERIFIED LIVE. This is an exact match, not a contains match ("" returns only the blank-Name sessions, not the fleet). For full-fleet or partial-name listing use list_sessions()/get_sessions_by_filter().""" return self.call("GetSessionsByName", {"sessionName": session_name}) def get_sessions_by_filter(self, session_filter: str) -> Any: """Enumerate sessions matching a ScreenConnect session-filter expression (GetSessionsByFilter, param `sessionFilter`). VERIFIED LIVE 2026-07-06 - this IS the full-fleet listing path. Examples: SessionType = 'Access' -> every unattended agent (the fleet) CustomProperty1 = 'Safesite' -> one client's machines Name LIKE '*DELL*' -> partial name (case-insensitive) Combine with AND/OR. String literals are single-quoted.""" return self.call("GetSessionsByFilter", {"sessionFilter": session_filter}) @staticmethod def _filter_literal(value: str) -> str: """Single-quote a value for the session-filter language, escaping any embedded single quote by doubling it. VERIFIED LIVE 2026-07-06: `CustomProperty1 LIKE '*Andy''s*'` matched the "Andy's Mobile Fuel" session, so doubling is the correct escape for this filter grammar.""" return "'" + str(value).replace("'", "''") + "'" def build_session_filter( self, session_type: Optional[str] = "access", name: Optional[str] = None, name_like: Optional[str] = None, company: Optional[str] = None, site: Optional[str] = None, tag: Optional[str] = None, extra: Optional[str] = None, ) -> str: """Assemble a session-filter expression from structured parts (AND-joined). `session_type`: access | support | meeting | all (None/all -> no type clause). `name` exact, `name_like` partial (wrapped in *...*), company/site/tag map to CP1/CP2/CP3, `extra` is a raw clause appended verbatim. Defaults to the whole Access fleet when nothing else is specified.""" clauses: list[str] = [] if session_type and session_type.lower() == "all": # 'all' = every session type. Only Access + Support exist on this # instance; an OR clause is the verified match-all (a bare '*' matches # nothing). VERIFIED LIVE 2026-07-06 -> 982 = 958 Access + 24 Support. clauses.append("(SessionType = 'Access' OR SessionType = 'Support')") elif session_type: literal = SESSION_TYPE_LITERALS.get(session_type.lower(), session_type) clauses.append(f"SessionType = {self._filter_literal(literal)}") if name: clauses.append(f"Name = {self._filter_literal(name)}") if name_like: clauses.append(f"Name LIKE {self._filter_literal('*' + name_like + '*')}") if company: clauses.append(f"CustomProperty1 = {self._filter_literal(company)}") if site: clauses.append(f"CustomProperty2 = {self._filter_literal(site)}") if tag: clauses.append(f"CustomProperty3 = {self._filter_literal(tag)}") if extra: clauses.append(extra) return " AND ".join(clauses) if clauses else "SessionType = 'Access'" def list_sessions(self, **kwargs) -> Any: """Convenience: build a filter from keyword parts and enumerate. See build_session_filter for the accepted keywords.""" return self.get_sessions_by_filter(self.build_session_filter(**kwargs)) # ====================================================================== # Control + detail methods (all VERIFIED LIVE 2026-06-22 on the ACG instance). # Reads take a JSON object {sessionID:...}; the POST/write methods take a # POSITIONAL ARRAY. Full-fleet inventory is served by GetSessionsByFilter # (see get_sessions_by_filter / list_sessions above) - the bare GetSessions # alias is absent but not needed. # ====================================================================== def get_session_details(self, session_id: str) -> Any: """GetSessionDetailsBySessionID - full detail for one session. VERIFIED.""" 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. EXISTS on this instance. POST body is a POSITIONAL ARRAY [sessionID, command] (e.g. ["", "ipconfig"]). STATE-CHANGING (gate behind --confirm).""" return self.call("SendCommandToSession", [session_id, command]) def send_message_to_session(self, session_id: str, message: str) -> Any: """SendMessageToSession - send a chat message to a guest. EXISTS. POST body positional array [sessionID, message]. STATE-CHANGING.""" return self.call("SendMessageToSession", [session_id, message]) def update_session_custom_properties(self, session_id: str, properties: list) -> Any: """UpdateSessionCustomProperties (CP1=Company, CP2=Site, CP3=Tag). EXISTS. POST body positional array [sessionID, [cp1, cp2, cp3, ...]]. STATE-CHANGING. (Confirm exact arg order against a safe test session before relying on it.)""" return self.call("UpdateSessionCustomProperties", [session_id, 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) # ====================================================================== # INSTALLER BUILDER (no API call - constructs the parameterized access # installer URL so a device self-tags into the right Company/Site/Tag). # ====================================================================== def build_installer_url( self, platform: str = "msi", name: Optional[str] = None, company: Optional[str] = None, site: Optional[str] = None, tag: Optional[str] = None, extra_props: Optional[list] = None, ) -> str: """Build a parameterized ScreenConnect ACCESS installer URL. VERIFIED LIVE 2026-06-22: the cloud instance serves a pre-keyed installer at /Bin/ScreenConnect.ClientSetup.?e=Access&y=Guest ; append `t=` (session name) and repeated `c=` for the custom properties (order = CP1=Company, CP2=Site, CP3=Tag, ... up to 8). The installed agent self-tags with these, so it lands under the matching session-group filters. platform: msi | exe | pkg | deb | rpm | sh """ props = [company or "", site or "", tag or ""] if extra_props: props.extend(extra_props) params = ["e=Access", "y=Guest"] if name: params.append("t=" + quote(name, safe="")) for p in props: params.append("c=" + quote(p, safe="")) return ( f"{self.base_url}/Bin/ScreenConnect.ClientSetup.{platform}?" + "&".join(params) ) 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())