Howard Enos
3a1edb721b
The control methods (send-command/send-message/set-properties) and session detail are verified live on the ACG instance; the "pending unlock" help text was left over from before probing confirmed them. Skill validated against skill-creator rules (frontmatter, vault creds, gated writes, errorlog compliance, ASCII, selftest 12/12). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
271 lines
9.5 KiB
Python
271 lines
9.5 KiB
Python
#!/usr/bin/env python3
|
|
"""CLI for the screenconnect skill - ConnectWise ScreenConnect (Control) API.
|
|
|
|
Read-only subcommands run freely. State-changing subcommands (send-command,
|
|
send-message, set-properties) refuse to run without --confirm.
|
|
|
|
Usage:
|
|
python sc.py status # auth check + instance info
|
|
python sc.py sessions [--name NAME] [--json]
|
|
python sc.py session <sessionID> # full session detail
|
|
python sc.py send-command --session <id> --command "..." --confirm
|
|
python sc.py send-message --session <id> --message "..." --confirm
|
|
python sc.py set-properties --session <id> --json '[null,"Site","Tag"]' --confirm
|
|
python sc.py raw --method GetSessionsByName --body '{"sessionName":""}' [--get]
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
from sc_client import (
|
|
ScreenConnectClient,
|
|
ScreenConnectError,
|
|
SC_BASE_URL,
|
|
CUSTOM_PROPERTIES,
|
|
)
|
|
|
|
# --- errorlog (skill rule): log GENUINE failures only, never expected conditions ---
|
|
_EXPECTED_ERROR_MARKERS = (
|
|
"web method does not exist", # method not unlocked on this instance yet
|
|
"429",
|
|
"too many requests",
|
|
)
|
|
|
|
|
|
def _is_expected_error(msg: str) -> bool:
|
|
m = (msg or "").lower()
|
|
return any(marker in m for marker in _EXPECTED_ERROR_MARKERS)
|
|
|
|
|
|
def _should_log_error(command: str, msg: str) -> bool:
|
|
if os.environ.get("SC_SUPPRESS_ERRORLOG"):
|
|
return False
|
|
if command == "raw":
|
|
return False
|
|
return not _is_expected_error(msg)
|
|
|
|
|
|
def _log_skill_error(skill, msg, context=""):
|
|
"""Soft-fail append to errorlog.md via the canonical helper (never throws)."""
|
|
try:
|
|
root = os.environ.get("CLAUDETOOLS_ROOT") or os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "..", "..", "..", "..")
|
|
)
|
|
h = os.path.join(root, ".claude", "scripts", "log-skill-error.sh")
|
|
if not os.path.exists(h):
|
|
return
|
|
a = ["bash", h, skill, msg]
|
|
if context:
|
|
a += ["--context", context]
|
|
subprocess.run(a, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _emit(obj, as_json: bool) -> None:
|
|
print(json.dumps(obj, indent=2, default=str))
|
|
|
|
|
|
def _print_sessions(data) -> None:
|
|
items = data if isinstance(data, list) else [data]
|
|
print(f"Sessions: {len(items)}")
|
|
for s in items:
|
|
if not isinstance(s, dict):
|
|
print(f" {s}")
|
|
continue
|
|
cp = s.get("CustomProperties") or s.get("CustomPropertyValues") or []
|
|
cps = ",".join(str(x) for x in cp) if isinstance(cp, list) else str(cp)
|
|
print(
|
|
f" {s.get('SessionID','?')} type={s.get('SessionType','?')} "
|
|
f"name={s.get('Name','') or '-'!r} host={s.get('Host','') or '-'} "
|
|
f"guest={s.get('GuestInfoName', s.get('GuestMachineName',''))} [{cps}]"
|
|
)
|
|
|
|
|
|
# --- gating ---
|
|
def _gated(action_desc: str, confirm: bool) -> bool:
|
|
if not confirm:
|
|
print("[WARNING] Refusing state-changing action without --confirm.")
|
|
print(f"[INFO] Would: {action_desc}")
|
|
return False
|
|
return True
|
|
|
|
|
|
# --- handlers ---
|
|
def cmd_status(client, args):
|
|
# Auth check via the one verified method; report instance + custom-property map.
|
|
sessions = client.get_sessions_by_name("")
|
|
n = len(sessions) if isinstance(sessions, list) else 0
|
|
print(f"[OK] Authenticated to {SC_BASE_URL}")
|
|
print(f" extension: {client.extension_guid}")
|
|
print(f" custom properties: {CUSTOM_PROPERTIES}")
|
|
print(f" GetSessionsByName('') returned {n} session(s)")
|
|
return 0
|
|
|
|
|
|
def cmd_sessions(client, args):
|
|
_print_sessions(client.get_sessions_by_name(args.name)) if not args.json else _emit(
|
|
client.get_sessions_by_name(args.name), True
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_session(client, args):
|
|
_emit(client.get_session_details(args.session_id), True)
|
|
return 0
|
|
|
|
|
|
def cmd_build_installer(client, args):
|
|
url = client.build_installer_url(
|
|
platform=args.platform, name=args.name,
|
|
company=args.company, site=args.site, tag=args.tag,
|
|
)
|
|
if args.json:
|
|
_emit({"installer_url": url, "platform": args.platform}, True)
|
|
return 0
|
|
print(url)
|
|
return 0
|
|
|
|
|
|
def cmd_send_command(client, args):
|
|
if not _gated(f"run command on session {args.session} : {args.run_command!r}", args.confirm):
|
|
return 3
|
|
_emit(client.send_command_to_session(args.session, args.run_command), args.json)
|
|
return 0
|
|
|
|
|
|
def cmd_send_message(client, args):
|
|
if not _gated(f"message session {args.session}", args.confirm):
|
|
return 3
|
|
_emit(client.send_message_to_session(args.session, args.message), args.json)
|
|
return 0
|
|
|
|
|
|
def cmd_set_properties(client, args):
|
|
try:
|
|
props = json.loads(args.json_props)
|
|
except json.JSONDecodeError as exc:
|
|
print(f"[ERROR] --json is not valid JSON: {exc}", file=sys.stderr)
|
|
return 2
|
|
if not isinstance(props, list):
|
|
print("[ERROR] --json must be a JSON array (CP1=Company, CP2=Site, CP3=Tag).",
|
|
file=sys.stderr)
|
|
return 2
|
|
if not _gated(f"set custom properties on session {args.session}", args.confirm):
|
|
return 3
|
|
_emit(client.update_session_custom_properties(args.session, props), args.json)
|
|
return 0
|
|
|
|
|
|
DESTRUCTIVE_RAW_PATTERNS = ("sendcommand", "updatesession", "create", "delete",
|
|
"end", "remove", "transfer", "install", "uninstall")
|
|
|
|
|
|
def cmd_raw(client, args):
|
|
m = args.method.lower()
|
|
if any(p in m for p in DESTRUCTIVE_RAW_PATTERNS) and not args.confirm:
|
|
print(f"[WARNING] '{args.method}' looks state-changing; refusing without --confirm.",
|
|
file=sys.stderr)
|
|
return 3
|
|
body = None
|
|
if args.body:
|
|
try:
|
|
body = json.loads(args.body)
|
|
except json.JSONDecodeError as exc:
|
|
print(f"[ERROR] --body is not valid JSON: {exc}", file=sys.stderr)
|
|
return 2
|
|
result = client.raw(args.method, body, http_method="GET" if args.get else "POST")
|
|
print(json.dumps(result, indent=2, default=str))
|
|
return 0
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(prog="sc.py",
|
|
description="ConnectWise ScreenConnect API CLI (ACG).")
|
|
common = argparse.ArgumentParser(add_help=False)
|
|
common.add_argument("--json", action="store_true", help="Emit raw JSON.")
|
|
sub = p.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser("status", help="Auth check + instance info.", parents=[common])
|
|
|
|
sp = sub.add_parser("sessions", help="List sessions by Name (verified).", parents=[common])
|
|
sp.add_argument("--name", default="", help="Session Name filter (blank = unattended).")
|
|
|
|
sp = sub.add_parser("session", help="Session detail.", parents=[common])
|
|
sp.add_argument("session_id")
|
|
|
|
sp = sub.add_parser("build-installer",
|
|
help="Build a parameterized Access installer URL.",
|
|
parents=[common])
|
|
sp.add_argument("--platform", default="msi",
|
|
help="msi | exe | pkg | deb | rpm | sh (default msi).")
|
|
sp.add_argument("--name", help="Session name (defaults to machine name if omitted).")
|
|
sp.add_argument("--company", help="CP1 = Company.")
|
|
sp.add_argument("--site", help="CP2 = Site.")
|
|
sp.add_argument("--tag", help="CP3 = Tag.")
|
|
|
|
sp = sub.add_parser("send-command", help="Run a backstage command (gated).",
|
|
parents=[common])
|
|
sp.add_argument("--session", required=True)
|
|
sp.add_argument("--command", dest="run_command", required=True)
|
|
sp.add_argument("--confirm", action="store_true")
|
|
|
|
sp = sub.add_parser("send-message", help="Send a chat message (gated).",
|
|
parents=[common])
|
|
sp.add_argument("--session", required=True)
|
|
sp.add_argument("--message", required=True)
|
|
sp.add_argument("--confirm", action="store_true")
|
|
|
|
sp = sub.add_parser("set-properties",
|
|
help="Set CP1/CP2/CP3 (gated).", parents=[common])
|
|
sp.add_argument("--session", required=True)
|
|
sp.add_argument("--props-json", dest="json_props", required=True,
|
|
help='JSON array, e.g. ["Company","Site","Tag"]')
|
|
sp.add_argument("--confirm", action="store_true")
|
|
|
|
sp = sub.add_parser("raw", help="Call any method directly (power use / probing).",
|
|
parents=[common])
|
|
sp.add_argument("--method", required=True)
|
|
sp.add_argument("--body", help="JSON body.")
|
|
sp.add_argument("--get", action="store_true", help="Use GET instead of POST.")
|
|
sp.add_argument("--confirm", action="store_true",
|
|
help="Required for state-changing method names.")
|
|
return p
|
|
|
|
|
|
HANDLERS = {
|
|
"status": cmd_status,
|
|
"sessions": cmd_sessions,
|
|
"session": cmd_session,
|
|
"build-installer": cmd_build_installer,
|
|
"send-command": cmd_send_command,
|
|
"send-message": cmd_send_message,
|
|
"set-properties": cmd_set_properties,
|
|
"raw": cmd_raw,
|
|
}
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
handler = HANDLERS[args.command]
|
|
try:
|
|
client = ScreenConnectClient()
|
|
rc = handler(client, args)
|
|
return rc if isinstance(rc, int) else 0
|
|
except ScreenConnectError as exc:
|
|
print(f"[ERROR] {exc}", file=sys.stderr)
|
|
cmd = getattr(args, "command", "?")
|
|
if _should_log_error(cmd, str(exc)):
|
|
_log_skill_error("screenconnect", f"{exc}", context=f"cmd={cmd}")
|
|
return 1
|
|
except KeyboardInterrupt:
|
|
return 130
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|