claudetools/.claude/skills/screenconnect/scripts/sc.py

#!/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 + fleet counts
  python sc.py sessions                        # whole Access fleet
  python sc.py sessions --company "Safesite"   # one client
  python sc.py sessions --like DELL            # partial name match
  python sc.py sessions --name HOST            # exact name
  python sc.py sessions --filter "SessionType = 'Access'"   # raw filter
  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 most basic verified method (GetSessionsByName), so status
    # confirms credentials even on an extension version without GetSessionsByFilter.
    client.get_sessions_by_name("")
    print(f"[OK] Authenticated to {SC_BASE_URL}")
    print(f"  extension: {client.extension_guid}")
    print(f"  custom properties: {CUSTOM_PROPERTIES}")
    # Fleet counts are a best-effort add-on: the filter method may be absent, and it
    # must never fail the auth check that status exists to perform.
    try:
        access = client.list_sessions(session_type="access")
        support = client.list_sessions(session_type="support")
        na = len(access) if isinstance(access, list) else 0
        ns = len(support) if isinstance(support, list) else 0
        print(f"  fleet: {na} Access (unattended) + {ns} Support session(s)")
    except ScreenConnectError as exc:
        print(f"  fleet: (listing unavailable via GetSessionsByFilter: {exc})")
    return 0


def _effective_type(args) -> str:
    """Resolve the SessionType filter for `sessions`. Explicit --type wins. Otherwise
    a bare listing defaults to the Access fleet, but a TARGETED search (by name/like/
    company/site/tag) spans all types - matching the old GetSessionsByName behavior,
    where a by-name lookup found Support sessions too."""
    if args.type is not None:
        return args.type
    if any([args.name, args.like, args.company, args.site, args.tag]):
        return "all"
    return "access"


def cmd_sessions(client, args):
    # Full-fleet listing is the default (no selector -> every Access session).
    # A raw --filter wins; otherwise build a filter from the structured flags.
    if args.filter:
        data = client.get_sessions_by_filter(args.filter)
    else:
        data = client.list_sessions(
            session_type=_effective_type(args),
            name=args.name or None,
            name_like=args.like,
            company=args.company,
            site=args.site,
            tag=args.tag,
        )
    # --limit only caps the human-readable print; --json is always the full set.
    if not args.json and isinstance(data, list) and args.limit and len(data) > args.limit:
        _print_sessions(data[: args.limit])
        print(f"  ... {len(data) - args.limit} more (raise --limit or use --json)")
        return 0
    _emit(data, True) if args.json else _print_sessions(data)
    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 (default = whole Access fleet).",
                        parents=[common])
    sp.add_argument("--name", default="", help="Exact session Name match.")
    sp.add_argument("--like", help="Partial Name match (wrapped in *...*).")
    sp.add_argument("--company", help="Filter by CP1 = Company.")
    sp.add_argument("--site", help="Filter by CP2 = Site.")
    sp.add_argument("--tag", help="Filter by CP3 = Tag.")
    sp.add_argument("--type", default=None,
                    help="access | support | meeting | all. Default: access for a "
                         "bare listing, all when a name/company/etc selector is given.")
    sp.add_argument("--filter", help="Raw session-filter expression (overrides the flags).")
    sp.add_argument("--limit", type=int, default=0,
                    help="Cap rows printed (0 = no cap; --json is always full).")

    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())