364 lines
16 KiB
Python
364 lines
16 KiB
Python
#!/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: <raw api_secret> (NO "Basic" prefix; Basic auth 401s)
|
|
Origin: https://computerguru.screenconnect.com
|
|
Endpoints live under the RESTful API Manager extension:
|
|
POST <base>/App_Extensions/<guid>/Service.ashx/<MethodName> 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": <expr>})
|
|
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. ["<uuid>", "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.<platform>?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())
|