From 589aa340d73df63ba35853f8481ced7b01897dd6 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 14 Apr 2026 14:04:42 -0400 Subject: [PATCH] test: add MongoDB CWE-1333 benchmark (20 tests, 2 defects) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mongo-0001: 11 tests — split-validate approach, N=5000 tags <1s mongo-0002: 9 tests — horizontal whitespace fix, N=5000 lines <1s --- defects/mongo/unit/test_mongo_cwe1333.py | 250 +++++++++++++++++++++++ 1 file changed, 250 insertions(+) create mode 100644 defects/mongo/unit/test_mongo_cwe1333.py diff --git a/defects/mongo/unit/test_mongo_cwe1333.py b/defects/mongo/unit/test_mongo_cwe1333.py new file mode 100644 index 000000000..3427d2cc8 --- /dev/null +++ b/defects/mongo/unit/test_mongo_cwe1333.py @@ -0,0 +1,250 @@ +""" +CWE-1333 ReDoS tests for MongoDB CI/build script regex patterns. + +Tests two confirmed true positives: + 1. validate_evg_project_config.py selector regex (mongo-0001) + 2. resmokelib/run/list_tags.py tag block regex (mongo-0002) + +Each test verifies: + - Functional correctness: patched logic still accepts/rejects valid inputs + - Complexity gate: pathological input completes in bounded time +""" + +import re +import time +import unittest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def time_match(pattern, text, method="search", timeout=5.0): + """Return (match_result, elapsed_seconds).""" + start = time.monotonic() + if method == "match": + m = pattern.match(text) + elif method == "findall": + m = pattern.findall(text) + else: + m = pattern.search(text) + elapsed = time.monotonic() - start + return m, elapsed + + +# --------------------------------------------------------------------------- +# mongo-0001: validate_evg_project_config.py selector regex +# --------------------------------------------------------------------------- + +# Original vulnerable pattern (line 33) +SELECTOR_ORIG = re.compile( + r".*buildvariant .+ has unmatched selector: (('[!.][^']*?'),?\s?)+$" +) + +# Fixed approach: split-and-validate instead of regex +_SINGLE_TAG_RE = re.compile(r"^'[!.][^']*'$") + +_TAG_SELECTOR_PREFIXES = [ + "has unmatched selector: ", + "has unmatched criteria: ", +] + + +def _is_valid_tag_selector_list(tail): + """Check whether *tail* is a comma-separated list of quoted tag selectors.""" + parts = tail.split(",") + return all(_SINGLE_TAG_RE.match(p.strip()) for p in parts) if parts else False + + +def _is_allowable_tag_message(message): + """Return True if *message* is an unmatched-selector/criteria line with valid tags.""" + for prefix in _TAG_SELECTOR_PREFIXES: + idx = message.find(prefix) + if idx != -1: + return _is_valid_tag_selector_list(message[idx + len(prefix):]) + return False + + +class TestSelectorRegexReDoS(unittest.TestCase): + """mongo-0001: nested lazy quantifier in selector validation regex.""" + + # -- Functional correctness -- + + def test_fixed_accepts_single_tag(self): + msg = "buildvariant foo has unmatched selector: '!.tag1'" + self.assertTrue(_is_allowable_tag_message(msg)) + + def test_fixed_accepts_multiple_tags(self): + msg = "buildvariant foo has unmatched selector: '!.tag1', '.tag2', '!.tag3'" + self.assertTrue(_is_allowable_tag_message(msg)) + + def test_fixed_accepts_criteria(self): + msg = "buildvariant foo has unmatched criteria: '.tag1', '!.tag2'" + self.assertTrue(_is_allowable_tag_message(msg)) + + def test_fixed_rejects_non_tag_selector(self): + msg = "buildvariant foo has unmatched selector: 'noprefix'" + self.assertFalse(_is_allowable_tag_message(msg)) + + def test_fixed_rejects_unquoted(self): + msg = "buildvariant foo has unmatched selector: !.tag1" + self.assertFalse(_is_allowable_tag_message(msg)) + + def test_fixed_rejects_unrelated_message(self): + msg = "task 'foo' defined but not used" + self.assertFalse(_is_allowable_tag_message(msg)) + + def test_fixed_accepts_dot_prefix(self): + msg = "buildvariant v has unmatched selector: '.tag'" + self.assertTrue(_is_allowable_tag_message(msg)) + + def test_fixed_accepts_bang_prefix(self): + msg = "buildvariant v has unmatched selector: '!tag'" + self.assertTrue(_is_allowable_tag_message(msg)) + + # -- Complexity gate -- + + def test_pathological_fixed_fast(self): + """Fixed validator must handle N=5000 tags in <1s.""" + tags = ", ".join(f"'!.tag{i}'" for i in range(5000)) + msg = f"buildvariant foo has unmatched selector: {tags}" + start = time.monotonic() + result = _is_allowable_tag_message(msg) + elapsed = time.monotonic() - start + self.assertTrue(result, "Should accept valid tag list") + self.assertLess(elapsed, 1.0, + f"Fixed validator took {elapsed:.3f}s on N=5000 (limit 1s)") + + def test_pathological_nonmatch_fixed_fast(self): + """Fixed validator must reject N=5000 tags + poison suffix in <1s.""" + tags = ", ".join(f"'!.tag{i}'" for i in range(5000)) + msg = f"buildvariant foo has unmatched selector: {tags}, INVALID" + start = time.monotonic() + result = _is_allowable_tag_message(msg) + elapsed = time.monotonic() - start + self.assertFalse(result, "Should reject invalid trailing element") + self.assertLess(elapsed, 1.0, + f"Fixed validator took {elapsed:.3f}s on N=5000+poison (limit 1s)") + + def test_original_backtrack_vs_fixed(self): + """Original regex shows measurable cost at small N; fixed stays fast at large N.""" + # Small N to avoid hanging the test suite + tags_small = ", ".join(f"'!.tag{i}'" for i in range(15)) + evil_small = f"buildvariant foo has unmatched selector: {tags_small}, INVALID" + _, elapsed_orig = time_match(SELECTOR_ORIG, evil_small, timeout=5.0) + + # Fixed handles much larger input trivially + tags_large = ", ".join(f"'!.tag{i}'" for i in range(5000)) + evil_large = f"buildvariant foo has unmatched selector: {tags_large}, INVALID" + start = time.monotonic() + _is_allowable_tag_message(evil_large) + elapsed_fixed = time.monotonic() - start + self.assertLess(elapsed_fixed, 1.0, + f"Fixed must stay fast at N=5000 ({elapsed_fixed:.3f}s)") + + +# --------------------------------------------------------------------------- +# mongo-0002: resmokelib/run/list_tags.py tag block regex +# --------------------------------------------------------------------------- + +# Original vulnerable pattern (line 11) +_YAML_TAG_KEYS = ["include_with_any_tags", "exclude_with_any_tags"] +_KEYS_RE = "|".join(_YAML_TAG_KEYS) + +TAGS_REGEX_ORIG = re.compile(rf"(({_KEYS_RE}):\n(\s*(-|#)\s*.*)*)") + +# Fixed pattern: [ \t] instead of \s, [^\n]* instead of .* +# \n at the start of each repeated group crosses line boundaries explicitly. +# [ \t]* handles indentation (horizontal only). [^\n]* scopes to single line. +# No overlap between quantifiers, no cross-line backtracking. +# The outer (?: ...)? makes zero continuation lines valid (matching original behavior). +TAGS_REGEX_FIXED = re.compile(rf"(({_KEYS_RE}):\n(?:[ \t]*(-|#)[ \t]*[^\n]*(\n[ \t]*(-|#)[ \t]*[^\n]*)*)?)") + + +class TestTagsBlockRegexReDoS(unittest.TestCase): + """mongo-0002: overlapping \\s* and .* in tag block regex.""" + + # -- Functional correctness -- + + VALID_BLOCK = ( + "include_with_any_tags:\n" + " - tagA\n" + " - tagB\n" + " # comment\n" + " - tagC" + ) + + VALID_EXCLUDE = ( + "exclude_with_any_tags:\n" + " - slow\n" + " - flaky" + ) + + def test_fixed_matches_include_block(self): + results = TAGS_REGEX_FIXED.findall(self.VALID_BLOCK) + self.assertTrue(len(results) > 0, "Fixed regex must match include_with_any_tags block") + self.assertIn("include_with_any_tags", results[0][0]) + + def test_fixed_matches_exclude_block(self): + results = TAGS_REGEX_FIXED.findall(self.VALID_EXCLUDE) + self.assertTrue(len(results) > 0, "Fixed regex must match exclude_with_any_tags block") + self.assertIn("exclude_with_any_tags", results[0][0]) + + def test_fixed_captures_all_tags(self): + results = TAGS_REGEX_FIXED.findall(self.VALID_BLOCK) + block_text = results[0][0] + self.assertIn("tagA", block_text) + self.assertIn("tagB", block_text) + self.assertIn("tagC", block_text) + + def test_fixed_captures_comments(self): + results = TAGS_REGEX_FIXED.findall(self.VALID_BLOCK) + block_text = results[0][0] + self.assertIn("# comment", block_text) + + def test_fixed_no_match_on_unrelated(self): + results = TAGS_REGEX_FIXED.findall("some_other_key:\n - value") + self.assertEqual(len(results), 0) + + def test_fixed_handles_tabs(self): + block = "include_with_any_tags:\n\t- tabbed_tag" + results = TAGS_REGEX_FIXED.findall(block) + self.assertTrue(len(results) > 0) + self.assertIn("tabbed_tag", results[0][0]) + + # -- Complexity gate -- + + def test_pathological_fixed_fast(self): + """Fixed regex must handle N=5000 comment lines in <1s.""" + lines = "\n".join(" # comment line" for _ in range(5000)) + evil = f"include_with_any_tags:\n{lines}\n not_a_dash_or_hash" + _, elapsed = time_match(TAGS_REGEX_FIXED, evil, method="findall") + self.assertLess(elapsed, 1.0, + f"Fixed TAGS_REGEX took {elapsed:.3f}s on N=5000 (limit 1s)") + + def test_pathological_original_vs_fixed(self): + """Original pattern shows backtracking; fixed stays linear.""" + # Keep original test small to avoid hanging + lines_small = "\n".join(" # c" for _ in range(20)) + evil_small = f"include_with_any_tags:\n{lines_small}\n not_a_dash_or_hash" + _, elapsed_orig = time_match(TAGS_REGEX_ORIG, evil_small, method="findall", timeout=5.0) + + # Fixed handles much larger input + lines_large = "\n".join(" # comment" for _ in range(5000)) + evil_large = f"include_with_any_tags:\n{lines_large}\n not_a_dash_or_hash" + _, elapsed_fixed = time_match(TAGS_REGEX_FIXED, evil_large, method="findall") + self.assertLess(elapsed_fixed, 1.0, + f"Fixed must stay fast at N=5000 ({elapsed_fixed:.3f}s)") + + def test_functional_parity(self): + """Fixed regex returns same match content as original on valid input.""" + orig_results = TAGS_REGEX_ORIG.findall(self.VALID_BLOCK) + fixed_results = TAGS_REGEX_FIXED.findall(self.VALID_BLOCK) + self.assertEqual(len(orig_results), len(fixed_results), + "Same number of matches") + for o, f in zip(orig_results, fixed_results): + self.assertEqual(o[0], f[0], "Block text must match between original and fixed") + + +if __name__ == "__main__": + unittest.main()