From d69f6f9b6ad0bf74d731b373bbdeabb462c32b26 Mon Sep 17 00:00:00 2001 From: Fedele Mantuano Date: Tue, 18 Aug 2026 10:03:54 +0200 Subject: [PATCH 1/3] fix: keep parsing when a Received date is out of range A Received header whose date carries a 12+ digit year, or a timezone offset large enough to push the timestamp past int64, aborted the parse of the whole message: receiveds_format() guarded convert_mail_date() with except (TypeError, ValueError), but calendar.timegm() raises OverflowError and datetime.fromtimestamp() reports EOVERFLOW as OSError. Neither belongs to the MailParser* hierarchy, so a caller catching only MailParserError died on the message. The date and timezone properties already wrapped the same conversion in except Exception; only the received path was left open. Co-Authored-By: Claude Opus 5 --- src/mailparser/utils.py | 7 ++++++- tests/test_mail_parser.py | 39 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/src/mailparser/utils.py b/src/mailparser/utils.py index de4c773..dc60432 100644 --- a/src/mailparser/utils.py +++ b/src/mailparser/utils.py @@ -862,7 +862,12 @@ def receiveds_format(receiveds): i["date"] = re.sub(r"^\s*(?:\([^)]*\)\s*)+", "", i["date"]) try: j["date_utc"], _ = convert_mail_date(i["date"]) - except (TypeError, ValueError): + except (TypeError, ValueError, OverflowError, OSError): + # Out-of-range dates fail in three different ways: a huge + # year overflows int64 inside calendar.timegm() + # (OverflowError), and a huge timezone offset pushes the + # timestamp into the band where datetime.fromtimestamp() + # reports EOVERFLOW (OSError). j["date_utc"] = None # Add delay diff --git a/tests/test_mail_parser.py b/tests/test_mail_parser.py index 5f8248e..818cb4a 100644 --- a/tests/test_mail_parser.py +++ b/tests/test_mail_parser.py @@ -237,6 +237,45 @@ def test_issue_received(self): self.assertIn("date_utc", i) self.assertIsNotNone(i["date_utc"]) + def test_received_date_out_of_range(self): + # A year whose epoch seconds overflow int64 makes calendar.timegm() + # raise OverflowError, which used to escape receiveds_format() and + # abort the parse of the whole message. + raw_mail = ( + "Received: from x ([1.2.3.4]) by mx.victim.com; " + "Tue, 7 Mar 292277026596 14:29:24 +0000\r\n\r\nbody\r\n" + ) + + mail = mailparser.parse_from_string(raw_mail) + + self.assertIsNone(mail.received[0]["date_utc"]) + self.assertIsInstance(mail.mail_json, str) + + def test_received_date_magnitude_sweep(self): + # Every year magnitude must fail closed, whatever the stdlib raises. + for exponent in range(4, 20): + raw_mail = ( + "Received: from x ([1.2.3.4]) by mx; " + f"Tue, 7 Mar {10**exponent} 14:29:24 +0000\r\n\r\nbody\r\n" + ) + + mail = mailparser.parse_from_string(raw_mail) + self.assertIsNone(mail.received[0]["date_utc"]) + + def test_received_date_offset_magnitude_sweep(self): + # A huge timezone offset pushes the timestamp into the band where + # datetime.fromtimestamp() reports EOVERFLOW as OSError, which is + # neither ValueError nor OverflowError. + for digits in range(4, 26): + for sign in "+-": + raw_mail = ( + "Received: from a by b; 7 Mar 1970 14:29:24 " + f"{sign}{'9' * digits}\r\n\r\nbody\r\n" + ) + + mail = mailparser.parse_from_string(raw_mail) + self.assertIsInstance(mail.mail_json, str) + def test_get_header(self): mail = mailparser.parse_from_file(mail_test_1) h1 = get_header(mail.message, "from") From 11b894a0b8b3f11055a99caff118206138d1daba Mon Sep 17 00:00:00 2001 From: Fedele Mantuano Date: Tue, 18 Aug 2026 10:07:41 +0200 Subject: [PATCH 2/3] fix: stop one bad attachment from costing the whole batch write_attachments() called _safe_attachment_filename() and write_sample() with no guard, so a single hostile attachment aborted the save and every attachment after it was silently never written: - a NUL smuggled in through RFC 2231 (filename*=us-ascii''evil%00.bin) raised ValueError out of the name check; - a basename longer than NAME_MAX raised OSError from open(), after creating a zero-byte stub; - a payload that is not valid base64 raised binascii.Error, a ValueError subclass, from the decode inside the open() block. Each is now logged and skipped. The base64 decode moved before open() so a failure leaves no stub, and it repairs padding and alphabet errors the way every client does, matching MIME::Base64 on 2008 cases: rejecting them let a sender append one character to make an attachment vanish from the extraction directory while it still reached the recipient. Containment failures must not be swallowed by that broader catch, so they now raise MailParserPathError instead of ValueError. Long names are truncated to 240 UTF-8 bytes keeping their extension. Deduplication reserves room for its own marker, because write_sample() sanitizes again and the clamp there used to cut the marker back off, collapsing distinct attachments onto one file. It resumes from the last suffix per stem rather than rescanning from 1, which was quadratic (16000 same-named attachments took 14s, now milliseconds), and folds case, because APFS, exFAT and SMB collapse Invoice.pdf and invoice.pdf. Co-Authored-By: Claude Opus 5 --- README.md | 18 ++- src/mailparser/exceptions.py | 13 ++ src/mailparser/utils.py | 212 ++++++++++++++++++++++---- tests/test_mail_parser.py | 287 ++++++++++++++++++++++++++++++++++- tests/test_utils.py | 44 +++++- 5 files changed, 540 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 7eab039..ff43899 100644 --- a/README.md +++ b/README.md @@ -203,9 +203,9 @@ The `attachments` property returns a list of dictionaries, each containing compr - `content-id` - Content identifier for referencing within HTML bodies - `filename` - Original decoded filename from the email. This is untrusted input; never use it directly to construct a filesystem path. -- `safe_filename` - Filename with directory components removed, or `None` when the original has - no usable basename. When saving attachments, prefer `write_attachments()` for full validation - and collision handling. +- `safe_filename` - Filename with directory components removed and truncated to fit the + filesystem name limit, or `None` when the original has no usable basename. When saving + attachments, prefer `write_attachments()` for full validation and collision handling. - `mail_content_type` - MIME content type - `payload` - Base64-encoded attachment data, ready for decoding or storage @@ -399,7 +399,13 @@ Attachment filenames are supplied by the email sender. The `filename` value in not pass it directly to `open()` or join it to a directory. The `safe_filename` field provides a sanitized basename when one exists, but applications saving files should prefer `write_attachments()`, which also validates containment, rejects symlink destinations, and -deduplicates names within the attachment batch. +deduplicates names within the attachment batch. Deduplication is case-insensitive, because +APFS, exFAT and SMB collapse `Invoice.pdf` and `invoice.pdf` onto a single file. + +A single unusable attachment never costs the rest of the batch: `write_attachments()` logs a +warning and moves on when a filename cannot be sanitized, a payload cannot be decoded, or the +write itself fails, so the remaining attachments are still saved. A containment failure is not +treated this way: it raises `MailParserPathError` and stops the batch. # Usage from Command Line @@ -493,7 +499,11 @@ MailParserError: Base MailParser Exception | \── MailParserOSError: Raised when there is an OS error | +\── MailParserPathError: Raised when an attachment escapes the output directory +| \── MailParserReceivedParsingError: Raised when a received header cannot be parsed +| +\── MailParserRecursionError: Raised when a message is nested too deeply to parse ``` # Docker Deployment diff --git a/src/mailparser/exceptions.py b/src/mailparser/exceptions.py index 8bf785f..d6ced5a 100644 --- a/src/mailparser/exceptions.py +++ b/src/mailparser/exceptions.py @@ -21,6 +21,7 @@ "MailParserOutlookError", "MailParserEnvironmentError", "MailParserOSError", + "MailParserPathError", "MailParserReceivedParsingError", "MailParserRecursionError", ) @@ -58,6 +59,18 @@ class MailParserOSError(MailParserError): pass +class MailParserPathError(MailParserError): + """ + Raised when an attachment would be written outside the output directory. + + This is a containment failure, not a per-attachment problem, so it keeps + its own type: ``write_attachments()`` skips individual attachments that + cannot be decoded or written, but must never swallow this one. + """ + + pass + + class MailParserReceivedParsingError(MailParserError): """ Raised when a received header cannot be parsed diff --git a/src/mailparser/utils.py b/src/mailparser/utils.py index dc60432..da6cfd7 100644 --- a/src/mailparser/utils.py +++ b/src/mailparser/utils.py @@ -50,7 +50,11 @@ JUNK_PATTERN, OTHERS_PARTS, ) -from mailparser.exceptions import MailParserOSError, MailParserReceivedParsingError +from mailparser.exceptions import ( + MailParserOSError, + MailParserPathError, + MailParserReceivedParsingError, +) log = logging.getLogger(__name__) @@ -988,11 +992,40 @@ def print_mail_fingerprints(data): # pragma: no cover print(f"sha512:\t{sha512}") +def decode_base64_payload(payload): + """ + Decode a base64 attachment payload the way a mail client would. + + ``base64.b64decode()`` rejects padding and alphabet errors that every + MUA silently repairs, so a sender can strip one padding character to + make an attachment undecodable here while it still reaches the + recipient intact. + + Args: + payload (string): base64 payload of an attachment + + Returns: + the decoded bytes + """ + data = re.sub(rb"[^A-Za-z0-9+/=]", b"", payload.encode("ascii", "ignore")) + # Padding ends the stream, as RFC 2045 requires and every client does. + # Splicing what follows onto the payload let a sender append bytes that + # only this tool sees, changing the hash it reports for the attachment. + data = data.split(b"=", 1)[0] + if len(data) % 4 == 1: + # A lone trailing character carries no complete byte. Padding it is + # impossible, so drop it as every lenient decoder does: otherwise + # adding one character is enough to make an attachment vanish from + # the extraction directory while it still reaches the recipient. + data = data[:-1] + return base64.b64decode(data + b"=" * (-len(data) % 4)) + + def print_attachments(attachments, flag_hash): # pragma: no cover if flag_hash: for i in attachments: if i.get("content_transfer_encoding") == "base64": - payload = base64.b64decode(i["payload"]) + payload = decode_base64_payload(i["payload"]) else: payload = i["payload"] @@ -1003,18 +1036,41 @@ def print_attachments(attachments, flag_hash): # pragma: no cover def write_attachments(attachments, base_path): # pragma: no cover - """Write attachments with unique filenames for this attachment batch.""" - used_filenames = set() + """ + Write attachments with unique filenames for this attachment batch. + + A single hostile attachment must not cost the rest of the batch, so an + unusable filename, an undecodable payload and a failing write are all + logged and skipped. ``MailParserPathError`` is deliberately not caught: + a containment failure is not a per-attachment problem. + + Args: + attachments (list): attachments as returned by MailParser + base_path (string): directory the attachments are written to + + Raises: + MailParserPathError: if an attachment escapes ``base_path`` + """ + used_filenames = {} for a in attachments: - filename = _safe_attachment_filename(a["filename"]) + try: + filename = _safe_attachment_filename(a["filename"]) + except ValueError: + log.warning(f"Skipped attachment with invalid filename: {a['filename']!r}") + continue + filename = _deduplicate_filename(filename, used_filenames) - write_sample( - binary=a["binary"], - payload=a["payload"], - path=base_path, - filename=filename, - ) + + try: + write_sample( + binary=a["binary"], + payload=a["payload"], + path=base_path, + filename=filename, + ) + except (OSError, ValueError): + log.warning(f"Skipped attachment {filename!r}", exc_info=True) def _safe_attachment_filename(filename): @@ -1027,7 +1083,7 @@ def _safe_attachment_filename(filename): if filename in ("", ".", ".."): raise ValueError("Invalid attachment filename") - return filename + return _truncate_filename(filename) _COMPOUND_ATTACHMENT_EXTENSIONS = ( @@ -1050,17 +1106,113 @@ def _split_attachment_extension(filename): return os.path.splitext(filename) -def _deduplicate_filename(filename, used_filenames): - """Return a unique filename within one write_attachments() operation.""" +# NAME_MAX is 255 bytes on Linux and macOS. Leave room for the "_1", "_2" +# suffixes _deduplicate_filename() appends after this truncation. +_MAX_FILENAME_BYTES = 240 + +# Room reserved inside the budget for the "_1", "_2" deduplication marker. +# Eight bytes would under-reserve from suffix 10**7 on, letting the marker +# push the name past the limit again. +_MAX_MARKER_BYTES = 11 + + +def _split_within_budget(filename): + """ + Split a basename into root and extension, both short enough to keep. + + An extension long enough to eat the whole budget is not an extension: + keeping it would leave no room for the root, and a negative budget + would slice the root from the wrong end. + + Args: + filename (str): sanitized basename + + Returns: + a (root, extension) tuple whose extension is at most half the budget + """ root, extension = _split_attachment_extension(filename) + if len(extension.encode("utf-8")) > _MAX_FILENAME_BYTES // 2: + return filename, "" + return root, extension + + +def _clamp_to_budget(text, budget): + """ + Cut a string to at most ``budget`` UTF-8 bytes. + + A partial multi-byte character left by the cut is dropped. + + Args: + text (str): string to shorten + budget (int): maximum length in UTF-8 bytes + + Returns: + the string, at most ``budget`` bytes long + """ + return text.encode("utf-8")[: max(0, budget)].decode("utf-8", "ignore") + + +def _truncate_filename(filename): + """ + Shorten a basename so it fits NAME_MAX, keeping its extension. + + The limit is measured in UTF-8 bytes, not characters, because that is + what the filesystem enforces. + + Args: + filename (str): sanitized basename + + Returns: + the basename, at most _MAX_FILENAME_BYTES bytes long + """ + if len(filename.encode("utf-8")) <= _MAX_FILENAME_BYTES: + return filename + + root, extension = _split_within_budget(filename) + budget = _MAX_FILENAME_BYTES - len(extension.encode("utf-8")) + return _clamp_to_budget(root, budget) + extension + + +def _deduplicate_filename(filename, used_filenames): + """ + Return a unique filename within one write_attachments() operation. + + Names are compared case-insensitively. ``os.path.normcase()`` is the + identity on POSIX, so two attachments differing only in case were + treated as distinct while APFS, exFAT and SMB collapse them onto one + file: the second attachment silently overwrote the first. Folding + always costs at most a ``_1`` suffix; not folding costs evidence. + + Args: + filename (str): sanitized basename + used_filenames (dict): folded name -> last suffix already handed + out for it, updated in place + + Returns: + a basename not yet used in this batch + """ + root, extension = _split_within_budget(filename) + + # Reserve room for the marker instead of appending past the limit: + # write_sample() sanitizes again, and the clamp there would cut the + # marker back off, collapsing distinct attachments onto one file. + budget = _MAX_FILENAME_BYTES - len(extension.encode("utf-8")) - _MAX_MARKER_BYTES + stem = _clamp_to_budget(root, budget) + + # Key on the stem the candidates are built from, not on the name as + # sent. Long names differing only past the clamp share one candidate + # namespace, so keying on the full name made each of them rescan the + # whole occupied range: quadratic in the number of attachments. + key = (stem + extension).casefold() + suffix = used_filenames.get(key, 0) candidate = filename - suffix = 1 - while os.path.normcase(candidate) in used_filenames: - candidate = f"{root}_{suffix}{extension}" + while candidate.casefold() in used_filenames: suffix += 1 + candidate = f"{stem}_{suffix}{extension}" - used_filenames.add(os.path.normcase(candidate)) + used_filenames[key] = suffix + used_filenames.setdefault(candidate.casefold(), 0) return candidate @@ -1073,8 +1225,20 @@ def write_sample(binary, payload, path, filename): # pragma: no cover payload: payload of sample, in base64 if it's a binary path (string): path of file filename (string): name of file - hash_ (string): file hash + + Raises: + ValueError: if a binary payload is not valid base64 + MailParserPathError: if the file would land outside ``path`` """ + # Resolve the bytes before creating the file, so a payload that cannot + # be decoded or encoded leaves no truncated stub behind. Surrogates + # produced by decoding the part are mapped back to the bytes they came + # from rather than dropped. + if binary: + content = decode_base64_payload(payload) + else: + content = payload.encode("utf-8", "surrogateescape") + filename = _safe_attachment_filename(filename) os.makedirs(path, exist_ok=True) @@ -1088,14 +1252,10 @@ def write_sample(binary, payload, path, filename): # pragma: no cover contained = False if not contained or os.path.islink(sample): - raise ValueError("Attachment path escapes the output directory") + raise MailParserPathError("Attachment path escapes the output directory") - if binary: - with open(sample, "wb") as f: - f.write(base64.b64decode(payload)) - else: - with open(sample, "w") as f: - f.write(payload) + with open(sample, "wb") as f: + f.write(content) def random_string(string_length=10): diff --git a/tests/test_mail_parser.py b/tests/test_mail_parser.py index 818cb4a..d26be8c 100644 --- a/tests/test_mail_parser.py +++ b/tests/test_mail_parser.py @@ -16,6 +16,7 @@ limitations under the License. """ +import base64 import datetime import hashlib import json @@ -32,8 +33,15 @@ import mailparser from mailparser.const import REGXIP6 -from mailparser.exceptions import MailParserOSError, MailParserRecursionError +from mailparser.exceptions import ( + MailParserOSError, + MailParserPathError, + MailParserRecursionError, +) from mailparser.utils import ( + _deduplicate_filename, + _safe_attachment_filename, + _truncate_filename, convert_mail_date, extract_msg_convert, fingerprints, @@ -171,6 +179,283 @@ def test_write_attachments_sanitizes_and_deduplicates_filenames(self): self.assertFalse(os.path.exists(os.path.join(temp_dir, "marker.txt"))) self.assertFalse(os.path.exists(os.path.join(temp_dir, "content-id.txt"))) + def test_write_attachments_skips_invalid_filename(self): + # A NUL byte smuggled in through RFC 2231 percent-encoding used to + # raise ValueError out of write_attachments(), so every attachment + # after the hostile one was silently never written. + raw_mail = ( + "Content-Type: multipart/mixed; boundary=b\r\n\r\n" + "--b\r\nContent-Type: application/octet-stream\r\n" + "Content-Disposition: attachment; " + "filename*=us-ascii''evil%00.bin\r\n\r\nxx\r\n" + "--b\r\nContent-Type: application/octet-stream\r\n" + "Content-Disposition: attachment; filename=good.bin\r\n\r\nyy\r\n" + "--b--\r\n" + ) + + mail = mailparser.parse_from_string(raw_mail) + self.assertEqual(mail.attachments[0]["filename"], "evil\x00.bin") + self.assertIsNone(mail.attachments[0]["safe_filename"]) + + with tempfile.TemporaryDirectory() as temp_dir: + output_dir = os.path.join(temp_dir, "attachments") + mail.write_attachments(output_dir) + self.assertEqual(os.listdir(output_dir), ["good.bin"]) + + def test_write_attachments_truncates_long_filename(self): + # A basename longer than NAME_MAX used to raise OSError out of + # write_sample() and abort the rest of the batch. + long_name = "a" * 300 + ".bin" + raw_mail = ( + "Content-Type: multipart/mixed; boundary=b\r\n\r\n" + "--b\r\nContent-Type: application/octet-stream\r\n" + f"Content-Disposition: attachment; filename={long_name}\r\n\r\nxx\r\n" + "--b\r\nContent-Type: application/octet-stream\r\n" + "Content-Disposition: attachment; filename=good.bin\r\n\r\nyy\r\n" + "--b--\r\n" + ) + + mail = mailparser.parse_from_string(raw_mail) + + with tempfile.TemporaryDirectory() as temp_dir: + output_dir = os.path.join(temp_dir, "attachments") + mail.write_attachments(output_dir) + written = sorted(os.listdir(output_dir)) + + self.assertEqual(len(written), 2) + self.assertIn("good.bin", written) + truncated = next(i for i in written if i != "good.bin") + self.assertTrue(truncated.endswith(".bin")) + self.assertLessEqual(len(truncated.encode("utf-8")), 240) + + def test_write_attachments_repairs_malformed_payload(self): + # base64.b64decode() raises binascii.Error, a ValueError subclass, + # so a payload with a length no padding can fix used to abort the + # batch and leave a zero-byte stub behind. Every MUA repairs it, so + # neither attachment may be lost. + raw_mail = ( + "Content-Type: multipart/mixed; boundary=b\r\n\r\n" + "--b\r\nContent-Type: application/octet-stream\r\n" + "Content-Transfer-Encoding: base64\r\n" + "Content-Disposition: attachment; filename=evil.bin\r\n\r\nAAAAA\r\n" + "--b\r\nContent-Type: application/octet-stream\r\n" + "Content-Disposition: attachment; filename=good.bin\r\n\r\nyy\r\n" + "--b--\r\n" + ) + + mail = mailparser.parse_from_string(raw_mail) + + with tempfile.TemporaryDirectory() as temp_dir: + output_dir = os.path.join(temp_dir, "attachments") + mail.write_attachments(output_dir) + self.assertEqual(sorted(os.listdir(output_dir)), ["evil.bin", "good.bin"]) + with open(os.path.join(output_dir, "evil.bin"), "rb") as attachment: + # The orphan character carries no complete byte and is + # dropped, exactly as a mail client would. + self.assertEqual(attachment.read(), b"\x00\x00\x00") + + def test_base64_attachment_with_orphan_character(self): + # Adding one character makes the length 1 more than a multiple of + # four, which no padding can repair. Rejecting it deleted the + # attachment from the extraction directory while the recipient's + # client still saved it intact. + original = b"MZ\x90\x00PAYLOAD-EXE" + poisoned = base64.b64encode(original).decode("ascii") + "A" + raw_mail = ( + "From: a@b.c\r\nContent-Type: application/octet-stream\r\n" + "Content-Transfer-Encoding: base64\r\n" + f'Content-Disposition: attachment; filename="x.bin"\r\n\r\n{poisoned}' + ) + + mail = mailparser.parse_from_string(raw_mail) + + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + self.assertEqual(os.listdir(temp_dir), ["x.bin"]) + with open(os.path.join(temp_dir, "x.bin"), "rb") as attachment: + self.assertEqual(attachment.read(), original) + + def test_base64_data_after_padding_is_ignored(self): + # RFC 2045 ends the stream at the padding and every client stops + # there. Splicing what follows onto the payload let a sender append + # bytes only this tool sees, changing the hash it reports. + original = b"MZ\x90\x00EICAR-MARKER" + poisoned = base64.b64encode(original).decode("ascii") + "QUJD" + raw_mail = ( + "From: a@b.c\r\nContent-Type: application/octet-stream\r\n" + "Content-Transfer-Encoding: base64\r\n" + f'Content-Disposition: attachment; filename="x.bin"\r\n\r\n{poisoned}' + ) + + mail = mailparser.parse_from_string(raw_mail) + + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + with open(os.path.join(temp_dir, "x.bin"), "rb") as attachment: + self.assertEqual(attachment.read(), original) + + def test_write_attachments_keeps_truncated_names_distinct(self): + # Truncation makes distinct long names collide. Appending "_1" past + # the length limit did not help: write_sample() sanitizes again and + # the clamp cut the suffix back off, so every colliding attachment + # landed on one file and only the last payload survived. + prefix = "a" * 236 + names = ( + prefix + "A" * 60 + ".bin", + prefix + "B" * 60 + ".bin", + prefix + "A" * 60 + ".bin", + ) + payloads = ("one", "two", "three") + parts = "".join( + "--b\r\nContent-Type: application/octet-stream\r\n" + f'Content-Disposition: attachment; filename="{name}"\r\n\r\n{payload}\r\n' + for name, payload in zip(names, payloads) + ) + raw_mail = f"Content-Type: multipart/mixed; boundary=b\r\n\r\n{parts}--b--\r\n" + + mail = mailparser.parse_from_string(raw_mail) + + with tempfile.TemporaryDirectory() as temp_dir: + output_dir = os.path.join(temp_dir, "attachments") + mail.write_attachments(output_dir) + written = os.listdir(output_dir) + + self.assertEqual(len(written), 3) + for name in written: + self.assertLessEqual(len(name.encode("utf-8")), 240) + + contents = [] + for name in written: + with open(os.path.join(output_dir, name)) as attachment: + contents.append(attachment.read()) + self.assertEqual(sorted(contents), ["one", "three", "two"]) + + def test_write_attachments_long_extension_collision(self): + # An extension longer than half the budget left no room for the + # dedup marker, so the reserved budget went negative and sliced the + # root from the wrong end. The name then no longer survived the + # clamp inside write_sample(), and a third attachment named like + # the re-clamped result could overwrite it. + names = ("a." + "b" * 238, "a." + "b" * 238, "_1." + "b" * 237) + payloads = ("first", "second", "third") + parts = "".join( + "--b\r\nContent-Type: application/octet-stream\r\n" + f'Content-Disposition: attachment; filename="{name}"\r\n\r\n{payload}\r\n' + for name, payload in zip(names, payloads) + ) + raw_mail = f"Content-Type: multipart/mixed; boundary=b\r\n\r\n{parts}--b--\r\n" + + mail = mailparser.parse_from_string(raw_mail) + + with tempfile.TemporaryDirectory() as temp_dir: + output_dir = os.path.join(temp_dir, "attachments") + mail.write_attachments(output_dir) + written = os.listdir(output_dir) + + self.assertEqual(len(written), 3) + contents = [] + for name in written: + self.assertLessEqual(len(name.encode("utf-8")), 240) + with open(os.path.join(output_dir, name)) as attachment: + contents.append(attachment.read()) + self.assertEqual(sorted(contents), ["first", "second", "third"]) + + def test_deduplicated_names_survive_resanitization(self): + # write_sample() sanitizes again, so every name dedup hands out must + # already be a fixed point of _truncate_filename(). + used_filenames = {} + base = _safe_attachment_filename("a." + "b" * 238) + + for _ in range(150): + candidate = _deduplicate_filename(base, used_filenames) + self.assertEqual(_truncate_filename(candidate), candidate) + + def test_write_attachments_reports_path_escape(self): + # The per-attachment guard must not swallow a containment failure. + raw_mail = ( + "Content-Type: application/octet-stream\r\n" + "Content-Disposition: attachment; filename=attachment.txt\r\n\r\nxx\r\n" + ) + mail = mailparser.parse_from_string(raw_mail) + + with tempfile.TemporaryDirectory() as temp_dir: + output_dir = os.path.join(temp_dir, "attachments") + os.makedirs(output_dir) + target = os.path.join(temp_dir, "outside.txt") + try: + os.symlink(target, os.path.join(output_dir, "attachment.txt")) + except (NotImplementedError, OSError): # pragma: no cover + self.skipTest("symlinks are not supported") + + with self.assertRaises(MailParserPathError): + mail.write_attachments(output_dir) + self.assertFalse(os.path.exists(target)) + + def test_truncate_filename(self): + self.assertEqual(_truncate_filename("short.bin"), "short.bin") + + # Extension is preserved, including compound tar extensions. + truncated = _truncate_filename("a" * 300 + ".tar.gz") + self.assertTrue(truncated.endswith(".tar.gz")) + self.assertEqual(len(truncated.encode("utf-8")), 240) + + # The limit is measured in bytes, not characters, and a multi-byte + # character cut in half by the budget is dropped, not mangled. + truncated = _truncate_filename("è" * 300 + ".bin") + self.assertTrue(truncated.endswith(".bin")) + self.assertLessEqual(len(truncated.encode("utf-8")), 240) + self.assertEqual(truncated, truncated.encode("utf-8").decode("utf-8")) + + # An extension big enough to eat the whole budget is not honoured + # as an extension, and the result still fits. + truncated = _truncate_filename("a." + "b" * 300) + self.assertEqual(len(truncated.encode("utf-8")), 240) + + def test_write_attachments_repairs_base64_padding(self): + # base64.b64decode() rejects padding errors every MUA repairs, so a + # sender could strip one character to drop the attachment from the + # extraction directory while it still reached the recipient. + original = b"MZ\x90\x00EICAR-STANDARD" + unpadded = base64.b64encode(original).decode("ascii").rstrip("=") + raw = ( + 'Content-Type: multipart/mixed; boundary="B"\r\n\r\n' + "--B\r\nContent-Type: application/octet-stream\r\n" + "Content-Transfer-Encoding: base64\r\n" + 'Content-Disposition: attachment; filename="f.bin"\r\n\r\n' + f"{unpadded}\r\n--B--\r\n" + ) + + mail = mailparser.parse_from_string(raw) + + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + with open(os.path.join(temp_dir, "f.bin"), "rb") as written: + self.assertEqual(written.read(), original) + + def test_write_attachments_case_insensitive_collision(self): + # os.path.normcase() is the identity on POSIX, so names differing + # only in case were treated as distinct while APFS, exFAT and SMB + # collapse them: the second attachment overwrote the first. + parts = "".join( + "--B\r\nContent-Type: application/octet-stream\r\n" + f'Content-Disposition: attachment; filename="{name}"\r\n\r\n{payload}\r\n' + for name, payload in (("Invoice.pdf", "benign"), ("invoice.pdf", "malware")) + ) + raw = f'Content-Type: multipart/mixed; boundary="B"\r\n\r\n{parts}--B--\r\n' + + mail = mailparser.parse_from_string(raw) + + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + written = os.listdir(temp_dir) + + self.assertEqual(len(written), 2) + contents = [] + for name in written: + with open(os.path.join(temp_dir, name), "rb") as attachment: + contents.append(attachment.read()) + self.assertEqual(sorted(contents), [b"benign", b"malware"]) + def test_attachment_with_unusable_filename_remains_parseable(self): raw_mail = """MIME-Version: 1.0 Content-Type: application/octet-stream diff --git a/tests/test_utils.py b/tests/test_utils.py index f12a30c..abf0c9d 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -26,7 +26,11 @@ import unittest from unittest.mock import Mock, patch -from mailparser.exceptions import MailParserOSError, MailParserReceivedParsingError +from mailparser.exceptions import ( + MailParserOSError, + MailParserPathError, + MailParserReceivedParsingError, +) from mailparser.utils import ( _GETADDRESSES_SUPPORTS_STRICT, decode_header_part, @@ -642,7 +646,9 @@ def test_write_sample_rejects_symlink_destination(self): except (NotImplementedError, OSError): # pragma: no cover self.skipTest("symlinks are not supported") - with self.assertRaises(ValueError): + # Containment failures keep their own type so that the + # per-attachment guard in write_attachments() cannot swallow them. + with self.assertRaises(MailParserPathError): write_sample(False, "payload", output_dir, "attachment.txt") self.assertFalse(os.path.exists(target)) @@ -650,7 +656,7 @@ def test_deduplicate_attachment_filenames(self): """Deduplication preserves extensions and handles existing suffixes.""" from mailparser.utils import _deduplicate_filename - used_filenames = set() + used_filenames = {} filenames = ( "report.tar.gz", "report.tar.gz", @@ -676,6 +682,38 @@ def test_deduplicate_attachment_filenames(self): ], ) + def test_deduplicate_filename_scales_linearly(self): + """A batch of same-named attachments must not be quadratic.""" + from mailparser.utils import _deduplicate_filename + + used_filenames = {} + start = time.monotonic() + for _ in range(16000): + _deduplicate_filename("a.bin", used_filenames) + elapsed = time.monotonic() - start + + # Restarting the suffix scan at 1 for every attachment took ~14s + # here; resuming from the last suffix takes milliseconds. + self.assertLess(elapsed, 5) + + def test_deduplicate_long_names_scale_linearly(self): + """Names sharing a clamped stem must not rescan the whole range.""" + from mailparser.utils import _deduplicate_filename, _safe_attachment_filename + + used_filenames = {} + start = time.monotonic() + for i in range(4000): + # Distinct names that all clamp onto the same stem: keying the + # resume counter on the name as sent made each one quadratic. + filename = _safe_attachment_filename( + "a" * 238 + chr(65 + i % 26) + chr(65 + (i // 26) % 26) + ) + _deduplicate_filename(filename, used_filenames) + _deduplicate_filename(filename, used_filenames) + elapsed = time.monotonic() - start + + self.assertLess(elapsed, 5) + def test_random_string(self): """Test random_string function""" from mailparser.utils import random_string From b1c85827e4a2f28da537fdb44a86e1d128f46887 Mon Sep 17 00:00:00 2001 From: Fedele Mantuano Date: Tue, 18 Aug 2026 10:08:24 +0200 Subject: [PATCH 3/3] fix: keep attachment bytes intact and contain hostile charsets An attachment is a file, not text, but every part that was not base64 was read back through ported_string(..., errors="ignore"), which drops each byte the declared charset cannot represent. A 4096-byte binary attachment reached disk as 2048 bytes, so its hash never matched the file the recipient received. Attachments are now re-encoded to base64 from their decoded bytes and report base64 as their transfer encoding. That also closes a differential the sender chose by picking an encoding: a quoted-printable application/* part was reported as a binary payload holding raw QP text, so write_attachments() base64-decoded it and saved bytes that were neither the attachment nor what was on the wire. Content-Transfer-Encoding is now stripped before comparison, as email's own get_payload() does. One trailing space made every encoding branch miss, sending the part down the lossy text path; on the body side it re-read the text through raw-unicode-escape, turning a Cyrillic phishing indicator into literal escape sequences that no content scanner matches. Three charset paths raised outside the MailParser* hierarchy on hostile input: ported_string() caught UnicodeDecodeError but the "undefined" codec raises a bare UnicodeError; get_payload(decode=False) applies the declared charset, guarded upstream only against an unknown charset name; and as_string() re-encodes an 8-bit body with a charset that may refuse it. raw_payload() and as_string_safe() fall back to the raw bytes. Co-Authored-By: Claude Opus 5 --- README.md | 6 +- src/mailparser/core.py | 73 +++++---- src/mailparser/utils.py | 72 ++++++++- tests/test_mail_parser.py | 317 +++++++++++++++++++++++++++++++------- 4 files changed, 385 insertions(+), 83 deletions(-) diff --git a/README.md b/README.md index ff43899..3aab8c8 100644 --- a/README.md +++ b/README.md @@ -207,7 +207,11 @@ The `attachments` property returns a list of dictionaries, each containing compr filesystem name limit, or `None` when the original has no usable basename. When saving attachments, prefer `write_attachments()` for full validation and collision handling. - `mail_content_type` - MIME content type -- `payload` - Base64-encoded attachment data, ready for decoding or storage +- `payload` - Base64-encoded attachment data, ready for decoding or storage. An attachment is + kept as bytes: whatever encoding it was sent with, it is re-encoded to base64 and reports + `base64` as its `content_transfer_encoding`, so the payload always matches the encoding it + declares and hashes like the file the recipient received. Only `base64` parts keep their + original wire text, which is already lossless To access custom or vendor-specific headers, replace hyphens with underscores. For example, to access the `X-MSMail-Priority` header: diff --git a/src/mailparser/core.py b/src/mailparser/core.py index 5d02b09..25b4d71 100644 --- a/src/mailparser/core.py +++ b/src/mailparser/core.py @@ -36,6 +36,7 @@ from mailparser.utils import ( _safe_attachment_filename, _safe_remove, + as_string_safe, convert_mail_date, decode_header_part, decode_headers, @@ -51,6 +52,7 @@ ported_open, ported_string, random_string, + raw_payload, receiveds_parsing, write_attachments, ) @@ -555,42 +557,52 @@ def parse(self): binary = False mail_content_type = ported_string(p.get_content_type()) log.debug(f"Mail content type {mail_content_type!r} part {i!r}") - transfer_encoding = ported_string( - p.get("content-transfer-encoding", "") - ).lower() + # Strip before comparing, exactly as email's own + # get_payload() does: a trailing space made every + # encoding branch below miss and the raw bytes fall + # through the text path, which drops the non-UTF-8 ones. + transfer_encoding = ( + ported_string(p.get("content-transfer-encoding", "")) + .strip() + .lower() + ) log.debug(f"Transfer encoding {transfer_encoding!r} part {i!r}") content_disposition = ported_string(p.get("content-disposition")) log.debug(f"content-disposition {content_disposition!r} part {i!r}") if p.is_multipart(): payload = "".join( - [m.as_string() for m in p.get_payload(decode=False)] + [as_string_safe(m) for m in p.get_payload(decode=False)] ) binary = False log.debug(f"Filename {filename!r} part {i!r} is multipart") - elif transfer_encoding == "base64" or ( - transfer_encoding == "quoted-printable" - and "application" in mail_content_type - ): - payload = p.get_payload(decode=False) + elif transfer_encoding == "base64": + payload = raw_payload(p) + if not isinstance(payload, str): + # The declared charset could not be applied, so + # the wire text is not recoverable: re-encode the + # decoded bytes to keep the payload base64. + payload = base64.b64encode(payload).decode("ascii") binary = True log.debug(f"Filename {filename!r} part {i!r} is binary") - elif "uuencode" in transfer_encoding: - # Re-encode in base64 + else: + # Every other encoding — uuencode, quoted-printable, + # 7bit, 8bit, binary — is re-encoded to base64 from + # the decoded bytes. An attachment is a file, not + # text: reading it back through a charset dropped + # every byte that charset cannot represent (half of a + # binary payload), so the extracted file was neither + # the attachment nor the bytes on the wire, and its + # hash never matched what the recipient received. payload = base64.b64encode(p.get_payload(decode=True)).decode( "ascii" ) binary = True - transfer_encoding = "base64" log.debug( - f"Filename {filename!r} part {i!r} is binary (uuencode" - " re-encoded to base64)" + f"Filename {filename!r} part {i!r} is binary" + f" ({transfer_encoding!r} re-encoded to base64)" ) - else: - payload = ported_string( - p.get_payload(decode=True), encoding=charset - ) - log.debug(f"Filename {filename!r} part {i!r} is not binary") + transfer_encoding = "base64" try: safe_filename = _safe_attachment_filename(filename) @@ -616,9 +628,15 @@ def parse(self): log.debug(f"Email part {i!r} is not an attachment") payload = p.get_payload(decode=True) - cte = p.get("Content-Transfer-Encoding") - if cte: - cte = cte.lower() + # Strip as well, for the same reason as the attachment + # branch above: a trailing space sent the part down the + # else branch, which re-reads the text through + # raw-unicode-escape and turns it into literal escapes. + cte = ( + ported_string(p.get("Content-Transfer-Encoding", "")) + .strip() + .lower() + ) if not cte or cte in ["7bit", "8bit"]: # message_from_bytes stores non-ASCII body bytes via @@ -627,16 +645,17 @@ def parse(self): # Unicode str (no surrogates). Detect which case we # have via get_payload(decode=False) and decode # accordingly so the declared charset is honoured. - raw_str = p.get_payload(decode=False) + raw_str = raw_payload(p) if isinstance(raw_str, str): try: # Raises if surrogates present (from_bytes path) raw_str.encode("utf-8") payload = raw_str except UnicodeEncodeError: - # Recover original bytes then decode with charset - orig_bytes = raw_str.encode("ascii", "surrogateescape") - payload = ported_string(orig_bytes, encoding=charset) + # These encodings are not transformed by + # get_payload(decode=True), so ``payload`` + # already holds the original bytes. + payload = ported_string(payload, encoding=charset) else: payload = ported_string(payload, encoding=charset) else: @@ -1081,7 +1100,7 @@ def message_as_string(self): """ Return the entire message flattened as a string. """ - return self.message.as_string() if self.message else "" + return as_string_safe(self.message) if self.message else "" @property def to_domains(self): diff --git a/src/mailparser/utils.py b/src/mailparser/utils.py index da6cfd7..33a8c0f 100644 --- a/src/mailparser/utils.py +++ b/src/mailparser/utils.py @@ -324,10 +324,11 @@ def ported_string(raw_data, encoding="utf-8", errors="ignore"): if isinstance(raw_data, str): return raw_data - # raw_data is bytes, decode it + # raw_data is bytes, decode it. The "undefined" codec raises a bare + # UnicodeError rather than UnicodeDecodeError, so catch the base class. try: return str(raw_data, encoding) - except (LookupError, UnicodeDecodeError): + except (LookupError, UnicodeError): return str(raw_data, "utf-8", errors) @@ -992,6 +993,73 @@ def print_mail_fingerprints(data): # pragma: no cover print(f"sha512:\t{sha512}") +def raw_payload(part): + """ + Return a part's payload without decoding its transfer encoding. + + ``Message.get_payload(decode=False)`` applies the charset the sender + declared. The email package guards that decode against an unknown + charset name, but not against a codec that refuses the operation + outright: ``charset="undefined"`` raises a bare ``UnicodeError``, which + would abort the parse of the whole message. + + Args: + part (email.message.Message): part to read + + Returns: + the payload as the email package returns it, or the undecoded bytes + when the declared charset cannot be applied + """ + try: + return part.get_payload(decode=False) + except (LookupError, UnicodeError): + log.warning("Cannot apply the declared charset, reading raw bytes") + return part.get_payload(decode=True) + + +def as_string_safe(part): + """ + Flatten a part back to its textual form. + + ``Message.as_string()`` re-encodes an 8-bit body with the charset the + sender declared, so a charset that cannot represent those bytes — + ``utf-16``, ``idna``, or a non-text codec such as ``base64`` — raises + out of the parse. Fall back to the raw bytes, which every codec + survives, rather than losing the whole message. + + Args: + part (email.message.Message): part to flatten + + Returns: + the flattened part as a string + """ + try: + return part.as_string() + except (LookupError, UnicodeError): + log.warning("Cannot flatten part with the declared charset") + + try: + return part.as_bytes().decode("utf-8", "surrogateescape") + except (LookupError, UnicodeError): + # BytesGenerator bypasses the charset only for text parts. A part + # whose type is multipart but which carries no boundary is routed + # to the generic multipart handler, which decodes with the + # declared charset just like as_string() did. + log.warning("Cannot flatten part, using its raw headers and body") + + headers = "".join(f"{k}: {v}\n" for k, v in part.items()) + body = raw_payload(part) + if isinstance(body, list): + # A conforming multipart: flatten each sub-part the same way, since + # the one that refused its charset may be nested any depth down. + body = "".join(as_string_safe(sub) for sub in body) + else: + # The charset was refused above, so raw_payload() fell back to the + # undecoded bytes. + body = body.decode("utf-8", "surrogateescape") + return f"{headers}\n{body}" + + def decode_base64_payload(payload): """ Decode a base64 attachment payload the way a mail client would. diff --git a/tests/test_mail_parser.py b/tests/test_mail_parser.py index d26be8c..b01395b 100644 --- a/tests/test_mail_parser.py +++ b/tests/test_mail_parser.py @@ -294,6 +294,72 @@ def test_base64_data_after_padding_is_ignored(self): with open(os.path.join(temp_dir, "x.bin"), "rb") as attachment: self.assertEqual(attachment.read(), original) + def test_multipart_subpart_with_unencodable_charset(self): + # BytesGenerator bypasses the charset only for text parts, so a + # part typed multipart but carrying no boundary made the as_bytes() + # fallback raise exactly like as_string() had. + for charset in (b"undefined", b"idna"): + raw_mail = ( + b'Content-Type: multipart/mixed; boundary="B1"\r\n\r\n--B1\r\n' + b'Content-Type: multipart/mixed; boundary="B2"\r\n' + b'Content-Disposition: attachment; filename="p.zip"\r\n\r\n--B2\r\n' + b"Content-Type: multipart/report; charset=" + charset + b"\r\n" + b"Content-Transfer-Encoding: 8bit\r\n\r\n\xff\xfe\x80 body\r\n" + b"--B1--\r\n" + ) + + mail = mailparser.parse_from_bytes(raw_mail) + + self.assertEqual(len(mail.attachments), 1) + self.assertIsInstance(mail.mail_json, str) + + def test_nested_multipart_with_unencodable_charset(self): + # The refusing part can be nested any depth down, so the fallback + # must walk a conforming multipart's sub-parts instead of trying to + # decode the list of them. + raw_mail = ( + b'Content-Type: multipart/mixed; boundary="B1"\r\n\r\n--B1\r\n' + b'Content-Type: multipart/mixed; boundary="B2"\r\n' + b'Content-Disposition: attachment; filename="p.zip"\r\n\r\n--B2\r\n' + b'Content-Type: multipart/mixed; boundary="B3"\r\n\r\n--B3\r\n' + b'Content-Type: multipart/report; charset="undefined"\r\n' + b"Content-Transfer-Encoding: 8bit\r\n\r\n\xff\xfe\x80 body\r\n" + b"--B3--\r\n--B2--\r\n--B1--\r\n" + ) + + mail = mailparser.parse_from_bytes(raw_mail) + + self.assertEqual(len(mail.attachments), 1) + self.assertIsInstance(mail.mail_json, str) + + def test_message_as_string_with_unencodable_charset(self): + raw_mail = ( + b'Content-Type: multipart/report; charset="undefined"\r\n' + b"Content-Transfer-Encoding: 8bit\r\n\r\n\xff\xfe\x80 body\r\n" + ) + + mail = mailparser.parse_from_bytes(raw_mail) + + self.assertIsInstance(mail.message_as_string, str) + + def test_multipart_attachment_with_unencodable_charset(self): + # as_string() re-encodes an 8-bit body with the charset the sender + # declared, and utf-16 (like idna, or a non-text codec) cannot + # represent the surrogates that carry those bytes. + for charset in (b"utf-16", b"utf-32", b"idna", b"undefined", b"base64"): + raw_mail = ( + b"From: a@b.c\r\nContent-Type: multipart/mixed; boundary=BB\r\n\r\n" + b"--BB\r\nContent-Type: message/rfc822\r\n" + b'Content-Disposition: attachment; filename="fwd.eml"\r\n\r\n' + b"Content-Type: text/plain; charset=" + charset + b"\r\n" + b"\r\n\xff\xfeA\x00\r\n--BB--\r\n" + ) + + mail = mailparser.parse_from_bytes(raw_mail) + + self.assertEqual(len(mail.attachments), 1) + self.assertIsInstance(mail.mail_json, str) + def test_write_attachments_keeps_truncated_names_distinct(self): # Truncation makes distinct long names collide. Appending "_1" past # the length limit did not help: write_sample() sanitizes again and @@ -411,51 +477,6 @@ def test_truncate_filename(self): truncated = _truncate_filename("a." + "b" * 300) self.assertEqual(len(truncated.encode("utf-8")), 240) - def test_write_attachments_repairs_base64_padding(self): - # base64.b64decode() rejects padding errors every MUA repairs, so a - # sender could strip one character to drop the attachment from the - # extraction directory while it still reached the recipient. - original = b"MZ\x90\x00EICAR-STANDARD" - unpadded = base64.b64encode(original).decode("ascii").rstrip("=") - raw = ( - 'Content-Type: multipart/mixed; boundary="B"\r\n\r\n' - "--B\r\nContent-Type: application/octet-stream\r\n" - "Content-Transfer-Encoding: base64\r\n" - 'Content-Disposition: attachment; filename="f.bin"\r\n\r\n' - f"{unpadded}\r\n--B--\r\n" - ) - - mail = mailparser.parse_from_string(raw) - - with tempfile.TemporaryDirectory() as temp_dir: - mail.write_attachments(temp_dir) - with open(os.path.join(temp_dir, "f.bin"), "rb") as written: - self.assertEqual(written.read(), original) - - def test_write_attachments_case_insensitive_collision(self): - # os.path.normcase() is the identity on POSIX, so names differing - # only in case were treated as distinct while APFS, exFAT and SMB - # collapse them: the second attachment overwrote the first. - parts = "".join( - "--B\r\nContent-Type: application/octet-stream\r\n" - f'Content-Disposition: attachment; filename="{name}"\r\n\r\n{payload}\r\n' - for name, payload in (("Invoice.pdf", "benign"), ("invoice.pdf", "malware")) - ) - raw = f'Content-Type: multipart/mixed; boundary="B"\r\n\r\n{parts}--B--\r\n' - - mail = mailparser.parse_from_string(raw) - - with tempfile.TemporaryDirectory() as temp_dir: - mail.write_attachments(temp_dir) - written = os.listdir(temp_dir) - - self.assertEqual(len(written), 2) - contents = [] - for name in written: - with open(os.path.join(temp_dir, name), "rb") as attachment: - contents.append(attachment.read()) - self.assertEqual(sorted(contents), [b"benign", b"malware"]) - def test_attachment_with_unusable_filename_remains_parseable(self): raw_mail = """MIME-Version: 1.0 Content-Type: application/octet-stream @@ -798,8 +819,12 @@ def test_defects_bug(self): self.assertEqual(1, result) def test_quoted_printable_application_attachment(self): - # A quoted-printable application/* attachment must be kept as binary - # (raw QP text), not decoded as UTF-8, which drops the non-UTF8 bytes. + # A quoted-printable application/* attachment must keep its exact + # bytes, not be decoded as UTF-8, which drops the non-UTF8 ones. + # It is re-encoded to base64 so that the reported payload matches + # the declared transfer encoding: reporting raw QP text as a binary + # payload made write_attachments() base64-decode it, saving bytes + # that are neither the attachment nor what was on the wire. import quopri original = b"\xff\xfe\x00\x01PDFdata\x80\x81\x82\xc0\xc1" @@ -815,12 +840,192 @@ def test_quoted_printable_application_attachment(self): + qp + "\r\n--B--\r\n" ) - attachment = mailparser.parse_from_string(raw).attachments[0] + mail = mailparser.parse_from_string(raw) + attachment = mail.attachments[0] self.assertTrue(attachment["binary"]) - self.assertEqual(attachment["content_transfer_encoding"], "quoted-printable") - self.assertEqual( - quopri.decodestring(attachment["payload"].encode("ascii")), original + self.assertEqual(attachment["content_transfer_encoding"], "base64") + self.assertEqual(base64.b64decode(attachment["payload"]), original) + + # The bytes written to disk are the attachment, not a re-reading of + # the wire text under a different encoding. + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + with open(os.path.join(temp_dir, "f.bin"), "rb") as written: + self.assertEqual(written.read(), original) + + def test_transfer_encoding_is_stripped(self): + # email's own get_payload() strips the header before comparing it. + # Without the same normalisation a trailing space missed every + # encoding branch, and the raw bytes fell through the text path, + # which silently drops every non-UTF-8 byte. + original = b"MZ\x90\x00\x03\xff\xfeEICAR-TEST" + encoded = base64.b64encode(original).decode("ascii") + + for transfer_encoding in ("base64", "base64 ", " base64", "base64\t"): + raw = ( + 'Content-Type: multipart/mixed; boundary="B"\r\n\r\n' + "--B\r\nContent-Type: application/octet-stream\r\n" + f"Content-Transfer-Encoding: {transfer_encoding}\r\n" + 'Content-Disposition: attachment; filename="f.bin"\r\n\r\n' + f"{encoded}\r\n--B--\r\n" + ) + + mail = mailparser.parse_from_string(raw) + self.assertTrue(mail.attachments[0]["binary"]) + + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + with open(os.path.join(temp_dir, "f.bin"), "rb") as written: + self.assertEqual(written.read(), original) + + def test_undefined_charset(self): + # The "undefined" codec raises a bare UnicodeError, which is not a + # UnicodeDecodeError and used to escape parse() entirely. + raw = ( + "From: a@b.c\r\nSubject: s\r\nMIME-Version: 1.0\r\n" + 'Content-Type: text/plain; charset="undefined"\r\n' + "Content-Transfer-Encoding: base64\r\n\r\nQUJD\r\n" + ) + + mail = mailparser.parse_from_string(raw) + + self.assertEqual(mail.text_plain, ["ABC"]) + + def test_hostile_charset_does_not_escape_parse(self): + # get_payload(decode=False) applies the declared charset, and the + # email package guards that only against an unknown charset name, + # not against a codec that refuses outright. + cases = ( + b'Content-Type: application/octet-stream; charset="undefined"\r\n' + b"Content-Transfer-Encoding: base64\r\n" + b"Content-Disposition: attachment; filename=x.bin\r\n\r\nQUJD\xff\r\n", + b'Content-Type: text/plain; charset="undefined"\r\n' + b"Content-Transfer-Encoding: 8bit\r\n\r\n\xff\xfe body\r\n", + b'Content-Type: text/plain; charset="idna"\r\n' + b"Content-Transfer-Encoding: 8bit\r\n\r\n\xff body\r\n", + b'Content-Type: text/plain; charset="unicode_escape"\r\n' + b"Content-Transfer-Encoding: 8bit\r\n\r\n\\udc80\xff\r\n", + ) + + for raw in cases: + mail = mailparser.parse_from_bytes(raw) + self.assertIsInstance(mail.mail_json, str) + + def test_body_transfer_encoding_is_stripped(self): + # A trailing space sent the body down the branch that re-reads the + # text through raw-unicode-escape, turning it into literal escapes + # and hiding every non-ASCII indicator from a content scanner. + body = "Ваш пароль истёк" + + for transfer_encoding in ("8bit", "8bit ", "8bit\t", "8BIT"): + raw = ( + "From: a@b.c\r\nMIME-Version: 1.0\r\n" + 'Content-Type: text/plain; charset="utf-8"\r\n' + f"Content-Transfer-Encoding: {transfer_encoding}\r\n\r\n{body}\r\n" + ) + + mail = mailparser.parse_from_string(raw) + self.assertIn(body, mail.text_plain[0]) + + def test_write_attachments_repairs_base64_padding(self): + # base64.b64decode() rejects padding errors every MUA repairs, so a + # sender could strip one character to drop the attachment from the + # extraction directory while it still reached the recipient. + original = b"MZ\x90\x00EICAR-STANDARD" + unpadded = base64.b64encode(original).decode("ascii").rstrip("=") + raw = ( + 'Content-Type: multipart/mixed; boundary="B"\r\n\r\n' + "--B\r\nContent-Type: application/octet-stream\r\n" + "Content-Transfer-Encoding: base64\r\n" + 'Content-Disposition: attachment; filename="f.bin"\r\n\r\n' + f"{unpadded}\r\n--B--\r\n" + ) + + mail = mailparser.parse_from_string(raw) + + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + with open(os.path.join(temp_dir, "f.bin"), "rb") as written: + self.assertEqual(written.read(), original) + + def test_write_attachments_leaves_no_empty_stub(self): + # A hostile charset used to reach the write through the text path, + # where the payload was encoded only after open() had created the + # file, leaving a zero-byte stub behind. Attachments now stay bytes, + # so the charset never touches them. + body = "AAAA" + r"\udcff" + "BBBB" + raw = ( + 'Content-Type: multipart/mixed; boundary="B"\r\n\r\n' + "--B\r\nContent-Type: application/octet-stream; " + 'charset="raw-unicode-escape"\r\n' + "Content-Transfer-Encoding: 7bit\r\n" + 'Content-Disposition: attachment; filename="f.bin"\r\n\r\n' + f"{body}\r\n--B--\r\n" + ) + + mail = mailparser.parse_from_string(raw) + + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + written = os.path.join(temp_dir, "f.bin") + self.assertNotEqual(os.path.getsize(written), 0) + with open(written, "rb") as attachment: + self.assertEqual(attachment.read(), body.encode("ascii")) + + def test_unencoded_attachment_keeps_every_byte(self): + # Attachments declaring 7bit/8bit/binary used to be read back + # through the declared charset with errors="ignore", which dropped + # every byte that charset could not represent — half of a binary + # payload — so the extracted file never hashed like the one the + # recipient received. + original = bytes(range(256)) * 16 + + for transfer_encoding in ("7bit", "8bit", "binary"): + raw = ( + b'Content-Type: multipart/mixed; boundary="B"\r\n\r\n' + b"--B\r\nContent-Type: application/octet-stream\r\n" + b"Content-Transfer-Encoding: " + + transfer_encoding.encode("ascii") + + b"\r\n" + b'Content-Disposition: attachment; filename="setup.exe"\r\n\r\n' + + original + + b"\r\n--B--\r\n" + ) + + mail = mailparser.parse_from_bytes(raw) + attachment = mail.attachments[0] + self.assertTrue(attachment["binary"]) + self.assertEqual(attachment["content_transfer_encoding"], "base64") + self.assertEqual(base64.b64decode(attachment["payload"]), original) + + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + with open(os.path.join(temp_dir, "setup.exe"), "rb") as written: + self.assertEqual(written.read(), original) + + def test_write_attachments_case_insensitive_collision(self): + # os.path.normcase() is the identity on POSIX, so names differing + # only in case were treated as distinct while APFS, exFAT and SMB + # collapse them: the second attachment overwrote the first. + parts = "".join( + "--B\r\nContent-Type: application/octet-stream\r\n" + f'Content-Disposition: attachment; filename="{name}"\r\n\r\n{payload}\r\n' + for name, payload in (("Invoice.pdf", "benign"), ("invoice.pdf", "malware")) ) + raw = f'Content-Type: multipart/mixed; boundary="B"\r\n\r\n{parts}--B--\r\n' + + mail = mailparser.parse_from_string(raw) + + with tempfile.TemporaryDirectory() as temp_dir: + mail.write_attachments(temp_dir) + written = os.listdir(temp_dir) + + self.assertEqual(len(written), 2) + contents = [] + for name in written: + with open(os.path.join(temp_dir, name), "rb") as attachment: + contents.append(attachment.read()) + self.assertEqual(sorted(contents), [b"benign", b"malware"]) def test_add_content_type(self): mail = mailparser.parse_from_file(mail_test_3) @@ -831,12 +1036,18 @@ def test_add_content_type(self): self.assertEqual(len(result["attachments"]), 1) self.assertIsInstance(result["attachments"][0]["mail_content_type"], str) - self.assertFalse(result["attachments"][0]["binary"]) + # Attachments are kept as bytes, so a quoted-printable part is + # reported base64-wrapped rather than re-read through its charset. + self.assertTrue(result["attachments"][0]["binary"]) self.assertIsInstance(result["attachments"][0]["payload"], str) self.assertEqual( - result["attachments"][0]["content_transfer_encoding"], "quoted-printable" + result["attachments"][0]["content_transfer_encoding"], "base64" ) self.assertEqual(result["attachments"][0]["charset"], "iso-8859-1") + self.assertIn( + b"The WatchGuard Firebox", + base64.b64decode(result["attachments"][0]["payload"]), + ) self.assertEqual(result["attachments"][0]["content-disposition"], "inline") mail = mailparser.parse_from_file(mail_malformed_1)