test: add MongoDB CWE-1333 benchmark (20 tests, 2 defects)
mongo-0001: 11 tests — split-validate approach, N=5000 tags <1s mongo-0002: 9 tests — horizontal whitespace fix, N=5000 lines <1s
This commit is contained in:
parent
8a3868d583
commit
589aa340d7
1 changed files with 250 additions and 0 deletions
250
defects/mongo/unit/test_mongo_cwe1333.py
Normal file
250
defects/mongo/unit/test_mongo_cwe1333.py
Normal file
|
|
@ -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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue