diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ec3b9bf..613361cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## 11.0.1 + +### Security + +- **Fixed a path traversal in the failure report sample filenames written by `--output`** ([GHSA-c284-w5m6-jhjm](https://github.com/domainaware/parsedmarc/security/advisories/GHSA-c284-w5m6-jhjm), affects 9.0.6 through 11.0.0). `save_output()` named each failure report's message sample after the sample's `Subject` header, falling back to the raw, unsanitized subject whenever sanitizing it produced an empty string. A subject consisting only of path separators and dots — `../../../` or `/` — sanitizes to nothing, so the raw value reached `os.path.join()` and the `.eml` file was written outside the `samples` directory, or at an absolute path. The subject comes from a message that failed authentication, so anyone who can send mail to a monitored mailbox controls it. The filename is now sanitized at write time and falls back to `sample`, and the `filename_safe_subject` key that a library caller may supply alongside the subject is no longer trusted. +- **Fixed unbounded decompression of report attachments** ([GHSA-43qf-f35w-2x4r](https://github.com/domainaware/parsedmarc/security/advisories/GHSA-43qf-f35w-2x4r), affects all versions through 11.0.0). `extract_report()` inflated gzip and zip attachments with no limit on the output size. The attachment's content is chosen by whoever sent it, and deflate reaches compression ratios of about 1000:1 on degenerate input, so a 100 KB attachment from any sender a monitored mailbox accepts inflated to 100 MB, with a peak of about twice that. Extraction now stops at 100 MiB of decompressed data and raises `ParserError`. The limit is deliberately not configurable: real DMARC aggregate, failure, and SMTP TLS reports are orders of magnitude smaller. + +### Bug fixes + +- **CSV output (`failure.csv`, and the other CSV renderers) no longer raises `_csv.Error` on Python 3.10 when a report field contains a NUL character.** Python 3.10's `csv` writer rejects any field containing NUL unless an escapechar is set ([CPython issue 97503](https://github.com/python/cpython/issues/97503), a 3.10 regression fixed in 3.11+). Failure report text fields (subject, user agent, authentication results, addresses, and more) come from untrusted mail, so a NUL byte anywhere in one made `parsed_failure_reports_to_csv()` — and therefore `save_output()` — raise on 3.10, uncaught by the CLI's `except (OSError, ValueError)` around `save_output`. The character is now stripped from CSV fields on every Python version, matching the sanitizing already applied to sample filenames. + ## 11.0.0 ### Changes diff --git a/parsedmarc/__init__.py b/parsedmarc/__init__.py index 01b4e953..a1317c15 100644 --- a/parsedmarc/__init__.py +++ b/parsedmarc/__init__.py @@ -72,6 +72,7 @@ from parsedmarc.types import ( from parsedmarc.utils import ( convert_outlook_msg, get_base_domain, + get_filename_safe_string, get_ip_address_info, human_timestamp_to_datetime, is_outlook_msg, @@ -123,6 +124,17 @@ MAGIC_XML_TAG = b"\x3c" # '<' - XML starting with an element tag (no declaratio # or "{" check, which masked it. MAGIC_JSON = b"\x7b" +# Maximum size, in bytes, of the data extracted from one compressed report +# attachment. Real reports compress at roughly 10:1, but the attachment's +# content is chosen by whoever sent it, and deflate (used by both gzip and +# zip) reaches about 1000:1 on degenerate input, so a 100 KB attachment from +# any sender a monitored mailbox accepts could inflate to 100 MB -- and the +# decoded str coexists with the decompressed bytes, so the peak is roughly +# twice that. Deliberately not configurable: real DMARC aggregate, failure, +# and SMTP TLS reports are orders of magnitude smaller than 100 MB, so no +# legitimate deployment needs to change it. +MAX_DECOMPRESSED_REPORT_SIZE = 100 * 1024 * 1024 + # Per-message count of consecutive failed saves, keyed on # ``(reports_folder, str(message_uid))``. Populated only when # ``get_dmarc_reports_from_mailbox()`` is given a ``save_callback`` that @@ -764,6 +776,20 @@ def parse_smtp_tls_report_json(report: str | bytes) -> SMTPTLSReport: raise InvalidSMTPTLSReport(str(e) + _exc_origin(e)) from e +def _csv_safe(value: Any) -> Any: + """Strips NUL characters from a string CSV field. + + Python 3.10's ``csv`` writer raises ``_csv.Error: need to escape, but + no escapechar set`` on any field containing NUL (CPython issue 97503, + fixed in 3.11). Report text fields come from untrusted mail, so the + character is dropped here, matching ``get_filename_safe_string``. + Non-string values are returned unchanged. + """ + if isinstance(value, str): + return value.replace("\x00", "") + return value + + def parsed_smtp_tls_reports_to_csv_rows( reports: SMTPTLSReport | list[SMTPTLSReport], ) -> list[dict[str, Any]]: @@ -845,7 +871,7 @@ def parsed_smtp_tls_reports_to_csv( rows = parsed_smtp_tls_reports_to_csv_rows(reports) for row in rows: - writer.writerow(row) + writer.writerow({key: _csv_safe(value) for key, value in row.items()}) csv_file_object.flush() return csv_file_object.getvalue() @@ -1152,6 +1178,46 @@ def parse_aggregate_report_xml( ) from error +def _decompress_gzip_bounded(data: bytes) -> bytes: + """ + Decompresses a gzip stream, refusing to produce more than + ``MAX_DECOMPRESSED_REPORT_SIZE`` bytes. + + The limit is read from the module global on every call so that it stays + a single source of truth (and can be patched in tests). + + Args: + data: The gzip stream + + Returns: + bytes: The decompressed data + + Raises: + ParserError: The decompressed data exceeds the limit, or the gzip + stream ends before the decompressor reaches its end marker. + zlib.error: The stream is corrupt (bad header, deflate data, or + CRC); ``extract_report()`` wraps it in ``ParserError``. + """ + limit = MAX_DECOMPRESSED_REPORT_SIZE + decompressor = zlib.decompressobj(zlib.MAX_WBITS | 16) + decompressed = decompressor.decompress(data, max_length=limit + 1) + if len(decompressed) > limit: + raise ParserError(f"Decompressed report exceeds the {limit} byte limit") + # ``flush()`` ignores ``max_length``, so it must not run until the + # bounded read above has been accepted. It is bounded here: a bounded + # ``decompress()`` that returns fewer than ``max_length`` bytes has + # either consumed all of its input or hit the end of the stream (after + # which any trailing bytes are ``unused_data``, not input), so nothing + # unbounded is left for ``flush()`` to produce. + decompressed += decompressor.flush() + if not decompressor.eof: + # A one-shot zlib.decompress() raises on a stream that ends early; + # a decompressobj just stops, so the truncation is detected here. + raise ParserError("Incomplete or truncated gzip stream") + + return decompressed + + def extract_report(content: bytes | str | BinaryIO) -> str: """ Extracts report text from zip- or gzip-compressed content, and returns @@ -1210,9 +1276,13 @@ def extract_report(content: bytes | str | BinaryIO) -> str: if header[: len(MAGIC_ZIP)] == MAGIC_ZIP: _zip = zipfile.ZipFile(file_object) - report = _zip.open(_zip.namelist()[0]).read().decode(errors="ignore") + limit = MAX_DECOMPRESSED_REPORT_SIZE + member = _zip.open(_zip.namelist()[0]).read(limit + 1) + if len(member) > limit: + raise ParserError(f"Decompressed report exceeds the {limit} byte limit") + report = member.decode(errors="ignore") elif header[: len(MAGIC_GZIP)] == MAGIC_GZIP: - report = zlib.decompress(file_object.read(), zlib.MAX_WBITS | 16).decode( + report = _decompress_gzip_bounded(file_object.read()).decode( errors="ignore" ) elif ( @@ -1518,7 +1588,7 @@ def parsed_aggregate_reports_to_csv( rows = parsed_aggregate_reports_to_csv_rows(reports) for row in rows: - writer.writerow(row) + writer.writerow({key: _csv_safe(value) for key, value in row.items()}) csv_file_object.flush() return csv_file_object.getvalue() @@ -1837,7 +1907,7 @@ def parsed_failure_reports_to_csv( for row in rows: new_row: dict[str, Any] = {} for key in fields: - new_row[key] = row.get(key) + new_row[key] = _csv_safe(row.get(key)) csv_writer.writerow(new_row) return csv_file.getvalue() @@ -3382,6 +3452,15 @@ def save_output( """ Save report data in the given directory + The message sample of each failure report is written to a ``samples`` + subdirectory, named after the sample's own ``subject`` header run + through :func:`parsedmarc.utils.get_filename_safe_string`, falling back + to ``sample`` when sanitizing leaves nothing. Since a subject arrives + from an untrusted sender, sanitizing happens here, at write time; the + ``filename_safe_subject`` key a caller may supply alongside it is not + trusted and not used. Names that collide get a ``(1)``, ``(2)``, … + suffix. + Args: results: Parsing results output_directory (str): The path to the directory to save in @@ -3438,11 +3517,7 @@ def save_output( sample = failure_report["sample"] message_count = 0 parsed_sample = failure_report["parsed_sample"] - subject = ( - parsed_sample.get("filename_safe_subject") - or parsed_sample.get("subject") - or "sample" - ) + subject = get_filename_safe_string(parsed_sample.get("subject")) or "sample" filename = subject while filename in sample_filenames: diff --git a/parsedmarc/constants.py b/parsedmarc/constants.py index ed290fa5..4d181923 100644 --- a/parsedmarc/constants.py +++ b/parsedmarc/constants.py @@ -1,4 +1,4 @@ -__version__ = "11.0.0" +__version__ = "11.0.1" USER_AGENT = f"parsedmarc/{__version__}" diff --git a/parsedmarc/utils.py b/parsedmarc/utils.py index b4b3f06b..40e494d8 100644 --- a/parsedmarc/utils.py +++ b/parsedmarc/utils.py @@ -1241,24 +1241,41 @@ def parse_email_address(original_address: str) -> dict[str, str | None]: } -def get_filename_safe_string(string: str) -> str: +def get_filename_safe_string(string: str | None) -> str: """ - Converts a string to a string that is safe for a filename + Converts a string to a string that is safe to use as a filename + + The returned string is a single path component, never a path: it + contains no path separator (``/`` or ``\\``), no drive separator + (``:``), and no NUL byte, so it cannot be absolute, drive-relative, or + escape the directory it is joined to. It never ends in ``.`` or a + space (Windows drops both when creating a file, which would make two + distinct subjects collide on one name), and it is never ``.`` or ``..`` + (both consist only of stripped characters and collapse to ``""``). It + is at most 100 characters long. It can be empty -- when the input is + empty, consists only of stripped characters, or is truncated to a run + of dots and spaces -- so callers that need a non-empty name must supply + their own fallback (e.g. ``get_filename_safe_string(subject) or + "sample"``). Windows reserved device names such as ``CON`` are not + rewritten. Args: - string (str): A string to make safe for a filename + string (str | None): A string to make safe for a filename. + ``None`` is treated as the literal string ``"None"``. Returns: str: A string safe for a filename """ - invalid_filename_chars = ["\\", "/", ":", '"', "*", "?", "|", "\n", "\r"] + invalid_filename_chars = ["\\", "/", ":", '"', "*", "?", "|", "\n", "\r", "\x00"] if string is None: string = "None" for char in invalid_filename_chars: string = string.replace(char, "") - string = string.rstrip(".") + # Truncate before stripping trailing dots and spaces, so that a name cut + # off just after one cannot end in it. string = (string[:100]) if len(string) > 100 else string + string = string.rstrip(". ") return string diff --git a/tests/test_init.py b/tests/test_init.py index 2fd085cb..bf60e8af 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -7,6 +7,7 @@ extract_report, get_dmarc_reports_from_mbox, and the CSV / JSON renderers. import base64 import binascii +import csv import email import gzip import inspect @@ -22,7 +23,7 @@ from email.mime.multipart import MIMEMultipart from email.mime.nonmultipart import MIMENonMultipart from email.mime.text import MIMEText from glob import glob -from io import BytesIO +from io import BytesIO, StringIO from pathlib import Path from shutil import rmtree from tempfile import NamedTemporaryFile, mkdtemp @@ -962,6 +963,29 @@ class Test(unittest.TestCase): ) print("Passed!") + def testSmtpTlsCsvStripsNulFromFields(self): + """A NUL character in an SMTP TLS report text field is stripped + from CSV output instead of reaching the ``csv`` writer. + + Python 3.10's ``csv`` module raises ``_csv.Error: need to escape, + but no escapechar set`` on any field containing NUL + (https://github.com/python/cpython/issues/97503, a 3.10 regression + fixed in 3.11). This test fails on unfixed code on 3.10 with that + error, and on unfixed code on 3.11+ because the NUL passes through + to the CSV text uncleaned.""" + result = parsedmarc.parse_report_file( + "samples/smtp_tls/smtp_tls.json", offline=True + ) + report = cast(SMTPTLSReport, result["report"]) + report["organization_name"] = "Example\x00Inc." + + csv_text = parsedmarc.parsed_smtp_tls_reports_to_csv(report) + + self.assertNotIn("\x00", csv_text) + reader = csv.DictReader(StringIO(csv_text)) + row = next(reader) + self.assertEqual(row["organization_name"], "ExampleInc.") + def testAggregateCsvExposesASNColumns(self): """The aggregate CSV output should include source_asn, source_as_name, and source_as_domain columns.""" @@ -2106,6 +2130,33 @@ class Test(unittest.TestCase): self.assertTrue(len(rows) > 0) print("Passed!") + def testFailureReportCsvStripsNulFromFields(self): + """A NUL character in a failure report text field is stripped from + CSV output instead of reaching the ``csv`` writer. + + Python 3.10's ``csv`` module raises ``_csv.Error: need to escape, + but no escapechar set`` on any field containing NUL + (https://github.com/python/cpython/issues/97503, a 3.10 regression + fixed in 3.11). This test fails on unfixed code on 3.10 with that + error, and on unfixed code on 3.11+ because the NUL passes through + to the CSV text uncleaned.""" + parsed_report = cast( + FailureReport, + parsedmarc.parse_report_file( + "samples/failure/dmarc_ruf_report_linkedin.eml", offline=True + )["report"], + ) + parsed_report["parsed_sample"]["subject"] = "re\x00port" + parsed_report["user_agent"] = "Agent\x00X" + + csv_text = parsedmarc.parsed_failure_reports_to_csv(parsed_report) + + self.assertNotIn("\x00", csv_text) + reader = csv.DictReader(StringIO(csv_text)) + row = next(reader) + self.assertEqual(row["subject"], "report") + self.assertEqual(row["user_agent"], "AgentX") + class TestExtractReport(unittest.TestCase): """Tests for parsedmarc.extract_report()""" @@ -2216,6 +2267,93 @@ class TestExtractReport(unittest.TestCase): with self.assertRaises(parsedmarc.ParserError): parsedmarc.extract_report(cast(BinaryIO, TextStream())) + def testExtractReportGzipOverLimitRejected(self): + """A gzip attachment that inflates past + MAX_DECOMPRESSED_REPORT_SIZE raises ParserError instead of being + decompressed. + + Regression test for GHSA-43qf-f35w-2x4r: extract_report inflated + gzip with a single unbounded zlib.decompress(), so a small + attachment from any sender a monitored mailbox accepts could + allocate hundreds of MB. The limit is patched down here so the test + stays fast; the real 100 MiB cap is the same code path.""" + limit = 4096 + compressed = gzip.compress(b"a" * (limit + 1)) + # The attack is amplification: the wire form is far below the cap. + self.assertLess(len(compressed), limit) + with patch("parsedmarc.MAX_DECOMPRESSED_REPORT_SIZE", limit): + with self.assertRaises(parsedmarc.ParserError) as ctx: + parsedmarc.extract_report(compressed) + self.assertIn( + f"Decompressed report exceeds the {limit} byte limit", str(ctx.exception) + ) + + def testExtractReportGzipAtLimitIsExtracted(self): + """A gzip attachment inflating to exactly the limit is still + extracted in full, so the cap rejects only what is over it""" + limit = 4096 + with patch("parsedmarc.MAX_DECOMPRESSED_REPORT_SIZE", limit): + result = parsedmarc.extract_report(gzip.compress(b"a" * limit)) + self.assertEqual(result, "a" * limit) + + def testExtractReportZipOverLimitRejected(self): + """A zip attachment whose member inflates past + MAX_DECOMPRESSED_REPORT_SIZE raises ParserError. + + Regression test for GHSA-43qf-f35w-2x4r, zip half: the member was + read with an unbounded .read().""" + import zipfile + + limit = 4096 + buf = BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: + zf.writestr("report.xml", b"a" * (limit + 1)) + compressed = buf.getvalue() + self.assertLess(len(compressed), limit) + with patch("parsedmarc.MAX_DECOMPRESSED_REPORT_SIZE", limit): + with self.assertRaises(parsedmarc.ParserError) as ctx: + parsedmarc.extract_report(compressed) + self.assertIn( + f"Decompressed report exceeds the {limit} byte limit", str(ctx.exception) + ) + + def testExtractReportZipAtLimitIsExtracted(self): + """A zip member inflating to exactly the limit is still extracted in + full""" + import zipfile + + limit = 4096 + buf = BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: + zf.writestr("report.xml", b"a" * limit) + with patch("parsedmarc.MAX_DECOMPRESSED_REPORT_SIZE", limit): + result = parsedmarc.extract_report(buf.getvalue()) + self.assertEqual(result, "a" * limit) + + def testExtractReportTruncatedGzipRaises(self): + """A gzip stream cut short still raises ParserError. + + Behavior parity check for the bounded-decompression change: the old + one-shot zlib.decompress() raised "Error -5 ... incomplete or + truncated stream" here, while a zlib.decompressobj() stops silently, + so the replacement has to detect the missing end-of-stream marker + itself.""" + xml = b'' + with self.assertRaises(parsedmarc.ParserError) as ctx: + parsedmarc.extract_report(gzip.compress(xml)[:-8]) + self.assertIn("truncated", str(ctx.exception)) + + def testExtractReportGzipWithTrailingBytesIsExtracted(self): + """Bytes after the end of the gzip member are ignored, exactly as + the old one-shot zlib.decompress() ignored them. + + Behavior parity check for the bounded-decompression change: verified + against the unmodified code, which returned the report text and did + not raise.""" + xml = b'' + result = parsedmarc.extract_report(gzip.compress(xml) + b"trailing garbage") + self.assertEqual(result, xml.decode()) + class TestMalformedXmlRecovery(unittest.TestCase): """Tests for XML recovery in parse_aggregate_report_xml""" @@ -4916,5 +5054,163 @@ class TestExtractReportStreams(unittest.TestCase): self.assertIn("", result) +class TestSaveOutput(unittest.TestCase): + """save_output() writes each failure report's message sample to a file + named after the sample's Subject header. + + That header comes from the message that failed authentication, i.e. + from an untrusted sender, so every test here asserts on the on-disk + layout of the whole temporary root -- not just of the samples + directory -- because the defect being guarded against + (GHSA-c284-w5m6-jhjm) wrote files above it.""" + + def setUp(self): + self.temp_root = mkdtemp() + self.addCleanup(rmtree, self.temp_root, True) + # Nested, so a traversal out of samples/ has somewhere to land + # inside the tree the assertions walk. + self.output_directory = os.path.join(self.temp_root, "parsedmarc", "output") + + def _failure_report(self, subject, **parsed_sample_extra) -> FailureReport: + """A failure report carrying only the keys save_output() reads""" + parsed_sample = {"subject": subject} + parsed_sample.update(parsed_sample_extra) + return cast( + FailureReport, + { + "sample": f"Subject: {subject!r}\r\n\r\nThe message body.\r\n", + "parsed_sample": parsed_sample, + # Read only by the CSV renderer save_output() also calls. + "auth_failure": ["dmarc"], + "authentication_mechanisms": [], + "source": { + "ip_address": "192.0.2.1", + "reverse_dns": None, + "base_domain": None, + "name": None, + "type": None, + "asn": None, + "as_name": None, + "as_domain": None, + "country": None, + }, + }, + ) + + def _save(self, *failure_reports: FailureReport) -> None: + results = cast( + ParsingResults, + { + "aggregate_reports": [], + "failure_reports": list(failure_reports), + "smtp_tls_reports": [], + }, + ) + parsedmarc.save_output(results, output_directory=self.output_directory) + + def _written_files(self) -> list[str]: + """Every file under the temp root, as a path relative to it""" + written = [] + for root, _dirs, files in os.walk(self.temp_root): + for name in files: + written.append( + os.path.relpath(os.path.join(root, name), self.temp_root) + ) + return sorted(written) + + def testTraversalSubjectsAreConfinedToTheSamplesDirectory(self): + """Subjects made only of path separators and dots all collapse to + the "sample" fallback inside samples/, instead of escaping it. + + Regression test for GHSA-c284-w5m6-jhjm: save_output() fell back to + the raw subject whenever sanitizing it produced an empty string, so + os.path.join(samples_directory, "../../../.eml") escaped three + levels up and os.path.join(samples_directory, "/.eml") became the + absolute path /.eml.""" + self._save( + self._failure_report("../../../"), + self._failure_report("/"), + self._failure_report(".."), + ) + + samples = os.path.join("parsedmarc", "output", "samples") + eml_files = [f for f in self._written_files() if f.endswith(".eml")] + self.assertEqual( + eml_files, + [ + os.path.join(samples, "sample (1).eml"), + os.path.join(samples, "sample (2).eml"), + os.path.join(samples, "sample.eml"), + ], + ) + # Nothing at all was written above the output directory. + self.assertTrue( + all( + f.startswith(os.path.join("parsedmarc", "output") + os.sep) + for f in self._written_files() + ), + self._written_files(), + ) + + def testCraftedFilenameSafeSubjectIsIgnored(self): + """The filename is derived from the subject at write time, so a + caller-supplied filename_safe_subject cannot name the file. + + parse_email() derives that key with the same sanitizer, but + save_output() is public API: a library caller assembling its own + ParsingResults can put anything there, and the value used to win + over the subject.""" + self._save( + self._failure_report("monthly digest", filename_safe_subject="../evil") + ) + + written = self._written_files() + samples = os.path.join("parsedmarc", "output", "samples") + self.assertIn(os.path.join(samples, "monthly digest.eml"), written) + self.assertFalse([f for f in written if "evil" in f], written) + + def testNullByteSubjectIsStrippedAndWritten(self): + """A NUL in the subject is stripped rather than reaching open(), + which would raise ValueError: embedded null byte and, in the CLI, + hold back the whole mailbox batch""" + self._save(self._failure_report("re\x00port")) + + samples = os.path.join("parsedmarc", "output", "samples") + self.assertEqual( + [f for f in self._written_files() if f.endswith(".eml")], + [os.path.join(samples, "report.eml")], + ) + + def testOrdinarySubjectNamesTheFileAndSampleIsWrittenVerbatim(self): + """An ordinary subject is unchanged by sanitizing, and the file + holds the message sample""" + report = self._failure_report("DMARC failure report") + self._save(report) + + path = os.path.join( + self.output_directory, "samples", "DMARC failure report.eml" + ) + # newline="" so the sample's CRLFs are not translated on the way in. + with open(path, newline="", encoding="utf-8") as sample_file: + self.assertEqual(sample_file.read(), report["sample"]) + + def testParsedSampleReportIsNamedFromItsSubject(self): + """End to end: a real failure report parsed from the sample corpus + is written under its own subject line""" + report = cast( + FailureReport, + parsedmarc.parse_report_file( + "samples/failure/dmarc_ruf_report_linkedin.eml", offline=True + )["report"], + ) + self._save(report) + + samples = os.path.join("parsedmarc", "output", "samples") + self.assertEqual( + [f for f in self._written_files() if f.endswith(".eml")], + [os.path.join(samples, "Subject line, could be UTF8 encoded.eml")], + ) + + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/tests/test_utils.py b/tests/test_utils.py index 818ff894..13be833e 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -348,9 +348,57 @@ class Test(unittest.TestCase): def testGetFilenameSafeStringNone(self): """get_filename_safe_string with None returns 'None'""" - result = parsedmarc.utils.get_filename_safe_string(None) # type: ignore[arg-type] + result = parsedmarc.utils.get_filename_safe_string(None) self.assertEqual(result, "None") + def testGetFilenameSafeStringTraversalSequencesCollapseToEmpty(self): + """A subject made only of path separators and dots sanitizes to the + empty string, so a caller's `or "sample"` fallback takes over rather + than a relative path reaching os.path.join() + + Regression test for GHSA-c284-w5m6-jhjm: `../../../` and `/` are the + reporter's proof-of-concept subjects.""" + for traversal in ("../../../", "/", "..", ".", "..\\..\\", "./../"): + with self.subTest(traversal=traversal): + self.assertEqual( + parsedmarc.utils.get_filename_safe_string(traversal), "" + ) + + def testGetFilenameSafeStringStripsNullByte(self): + """A NUL byte is stripped, so the result cannot make open() raise + ValueError: embedded null byte""" + result = parsedmarc.utils.get_filename_safe_string("re\x00port") + self.assertEqual(result, "report") + + def testGetFilenameSafeStringHasNoPathSeparators(self): + """No os.sep or os.altsep survives a mixed-separator subject""" + result = parsedmarc.utils.get_filename_safe_string("a/b\\c:d..\\..\\e/../f") + self.assertEqual(result, "abcd....e..f") + self.assertNotIn(os.sep, result) + if os.altsep is not None: + self.assertNotIn(os.altsep, result) + + def testGetFilenameSafeStringTruncatedNameHasNoTrailingDot(self): + """Truncation happens before trailing dots are stripped, so a name + cut off just after a dot does not end in one + + A trailing dot is silently dropped by Windows when the file is + created, which would make two distinct subjects collide on one + filename.""" + subject = f"{'a' * 99}.{'b' * 50}" + result = parsedmarc.utils.get_filename_safe_string(subject) + self.assertEqual(result, "a" * 99) + self.assertFalse(result.endswith(".")) + + def testGetFilenameSafeStringStripsTrailingSpaces(self): + """Trailing spaces are stripped like trailing dots, and mixed runs of + both, because Windows drops them when creating the file""" + for subject, expected in (("report ", "report"), ("report . .", "report")): + with self.subTest(subject=subject): + self.assertEqual( + parsedmarc.utils.get_filename_safe_string(subject), expected + ) + def testGetFilenameSafeStringLong(self): """get_filename_safe_string truncates to 100 chars""" result = parsedmarc.utils.get_filename_safe_string("a" * 200)