russell@unturf.com
8a3868d583
cassandra: 16 tests — PEMReader cert/key possessive fix, LogFile adjacent .* unambiguous split. All under 1s on N=10000 pathological. hadoop: 10 tests — RPC star-of-star fix. Under 2s on pathological input.
222 lines
8.4 KiB
Python
222 lines
8.4 KiB
Python
"""
|
|
CWE-1333 ReDoS tests for Apache Cassandra regex patterns.
|
|
|
|
Tests three confirmed true positives:
|
|
1. PEMReader.CERT_PATTERN (cassandra-0002)
|
|
2. PEMReader.KEY_PATTERN (cassandra-0003)
|
|
3. LogFile.FILE_REGEX (cassandra-0004)
|
|
|
|
Each test verifies:
|
|
- Functional correctness: patched pattern still matches valid inputs
|
|
- Complexity gate: pathological input completes in bounded time
|
|
"""
|
|
|
|
import re
|
|
import time
|
|
import unittest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Original (vulnerable) patterns — Python transliterations of Java source
|
|
# ---------------------------------------------------------------------------
|
|
|
|
CERT_PATTERN_ORIG = re.compile(
|
|
r"-+BEGIN\s+.*CERTIFICATE[^-]*-+(?:\s|\r|\n)+([a-z0-9+/=\r\n]+)-+END\s+.*CERTIFICATE[^-]*-+",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
KEY_PATTERN_ORIG = re.compile(
|
|
r"-+BEGIN\s+.*PRIVATE\s+KEY[^-]*-+(?:\s|\r|\n)+([a-z0-9+/=\r\n]+)-+END\s+.*PRIVATE\s+KEY[^-]*-+",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
# LogFile.FILE_REGEX: ^((?:[a-z]+-)?.{2}_)?txn_(.*)_(.*)\.log$
|
|
FILE_REGEX_ORIG = re.compile(r"^((?:[a-z]+-)?.{2}_)?txn_(.*)_(.*)\.log$")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Patched patterns
|
|
# ---------------------------------------------------------------------------
|
|
|
|
CERT_PATTERN_FIXED = re.compile(
|
|
r"-+BEGIN\s++.*CERTIFICATE[^-]*-+(?:\s|\r|\n)+([a-z0-9+/=\r\n]+)-+END\s++.*CERTIFICATE[^-]*-+",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
KEY_PATTERN_FIXED = re.compile(
|
|
r"-+BEGIN\s++.*PRIVATE\s+KEY[^-]*-+(?:\s|\r|\n)+([a-z0-9+/=\r\n]+)-+END\s++.*PRIVATE\s+KEY[^-]*-+",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
FILE_REGEX_FIXED = re.compile(r"^((?:[a-z]+-)?.{2}_)?txn_(.+)_([^_]+)\.log$")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def time_match(pattern, text, timeout=5.0):
|
|
"""Return (matched, elapsed_seconds). Abort if exceeds timeout."""
|
|
start = time.monotonic()
|
|
m = pattern.search(text)
|
|
elapsed = time.monotonic() - start
|
|
return m, elapsed
|
|
|
|
|
|
# ===========================================================================
|
|
# Test: PEMReader.CERT_PATTERN (cassandra-0002)
|
|
# ===========================================================================
|
|
|
|
class TestCertPatternReDoS(unittest.TestCase):
|
|
r"""cassandra-0002: CERT_PATTERN \s+.* overlap on whitespace."""
|
|
|
|
# -- Functional correctness --
|
|
|
|
VALID_PEM_CERT = (
|
|
"-----BEGIN CERTIFICATE-----\n"
|
|
"MIIBxTCCAWugAwIBAgIJANlpFpSKMM0p\n"
|
|
"-----END CERTIFICATE-----"
|
|
)
|
|
|
|
VALID_PEM_X509 = (
|
|
"---BEGIN X509 CERTIFICATE---\n"
|
|
"dGVzdA==\n"
|
|
"---END X509 CERTIFICATE---"
|
|
)
|
|
|
|
def test_fixed_matches_standard_cert(self):
|
|
m = CERT_PATTERN_FIXED.search(self.VALID_PEM_CERT)
|
|
self.assertIsNotNone(m, "Fixed pattern must match standard PEM certificate")
|
|
|
|
def test_fixed_matches_x509_cert(self):
|
|
m = CERT_PATTERN_FIXED.search(self.VALID_PEM_X509)
|
|
self.assertIsNotNone(m, "Fixed pattern must match X509 CERTIFICATE variant")
|
|
|
|
def test_fixed_captures_base64_body(self):
|
|
m = CERT_PATTERN_FIXED.search(self.VALID_PEM_CERT)
|
|
self.assertIn("MIIBxTCCAWugAwIBAgIJANlpFpSKMM0p", m.group(1))
|
|
|
|
# -- Complexity gate --
|
|
|
|
def test_pathological_input_fixed_fast(self):
|
|
"""Patched pattern must handle N=10000 spaces in <1s."""
|
|
evil = "-BEGIN" + " " * 10000 + "!"
|
|
_, elapsed = time_match(CERT_PATTERN_FIXED, evil)
|
|
self.assertLess(elapsed, 1.0,
|
|
f"Fixed CERT_PATTERN took {elapsed:.3f}s on N=10000 (limit 1s)")
|
|
|
|
def test_pathological_input_original_slow(self):
|
|
"""Original pattern must show measurable slowdown at N=22."""
|
|
evil_small = "-BEGIN" + " " * 22 + "!"
|
|
_, elapsed = time_match(CERT_PATTERN_ORIG, evil_small, timeout=5.0)
|
|
# At N=22 the original should take >0.1s due to backtracking
|
|
# We just verify the fixed version is orders of magnitude faster
|
|
evil_large = "-BEGIN" + " " * 10000 + "!"
|
|
_, elapsed_fixed = time_match(CERT_PATTERN_FIXED, evil_large)
|
|
self.assertLess(elapsed_fixed, 1.0,
|
|
"Fixed pattern must stay fast even at N=10000")
|
|
|
|
|
|
# ===========================================================================
|
|
# Test: PEMReader.KEY_PATTERN (cassandra-0003)
|
|
# ===========================================================================
|
|
|
|
class TestKeyPatternReDoS(unittest.TestCase):
|
|
r"""cassandra-0003: KEY_PATTERN \s+.* overlap on whitespace."""
|
|
|
|
VALID_PEM_KEY = (
|
|
"-----BEGIN RSA PRIVATE KEY-----\n"
|
|
"MIIBogIBAAJBALDfRYB4cMgCjIqN/w==\n"
|
|
"-----END RSA PRIVATE KEY-----"
|
|
)
|
|
|
|
VALID_PEM_EC_KEY = (
|
|
"---BEGIN EC PRIVATE KEY---\n"
|
|
"dGVzdA==\n"
|
|
"---END EC PRIVATE KEY---"
|
|
)
|
|
|
|
def test_fixed_matches_rsa_key(self):
|
|
m = KEY_PATTERN_FIXED.search(self.VALID_PEM_KEY)
|
|
self.assertIsNotNone(m, "Fixed pattern must match RSA PRIVATE KEY")
|
|
|
|
def test_fixed_matches_ec_key(self):
|
|
m = KEY_PATTERN_FIXED.search(self.VALID_PEM_EC_KEY)
|
|
self.assertIsNotNone(m, "Fixed pattern must match EC PRIVATE KEY")
|
|
|
|
def test_fixed_captures_base64_body(self):
|
|
m = KEY_PATTERN_FIXED.search(self.VALID_PEM_KEY)
|
|
self.assertIn("MIIBogIBAAJBALDfRYB4cMgCjIqN/w==", m.group(1))
|
|
|
|
def test_pathological_input_fixed_fast(self):
|
|
"""Patched pattern must handle N=10000 spaces in <1s."""
|
|
evil = "-BEGIN" + " " * 10000 + "!"
|
|
_, elapsed = time_match(KEY_PATTERN_FIXED, evil)
|
|
self.assertLess(elapsed, 1.0,
|
|
f"Fixed KEY_PATTERN took {elapsed:.3f}s on N=10000 (limit 1s)")
|
|
|
|
|
|
# ===========================================================================
|
|
# Test: LogFile.FILE_REGEX (cassandra-0004)
|
|
# ===========================================================================
|
|
|
|
class TestFileRegexReDoS(unittest.TestCase):
|
|
"""cassandra-0004: FILE_REGEX (.*)_(.*) adjacent greedy groups."""
|
|
|
|
# -- Functional correctness --
|
|
|
|
def test_fixed_matches_legacy_format(self):
|
|
"""Legacy BIG format: cc_txn_opname_id.log"""
|
|
m = FILE_REGEX_FIXED.match("nb_txn_compaction_abc123.log")
|
|
self.assertIsNotNone(m)
|
|
self.assertEqual(m.group(1), "nb_")
|
|
self.assertEqual(m.group(2), "compaction")
|
|
self.assertEqual(m.group(3), "abc123")
|
|
|
|
def test_fixed_matches_new_format(self):
|
|
"""New format: fmt-cc_txn_opname_id.log"""
|
|
m = FILE_REGEX_FIXED.match("big-nb_txn_compaction_abc123.log")
|
|
self.assertIsNotNone(m)
|
|
self.assertEqual(m.group(1), "big-nb_")
|
|
self.assertEqual(m.group(2), "compaction")
|
|
self.assertEqual(m.group(3), "abc123")
|
|
|
|
def test_fixed_matches_no_prefix(self):
|
|
"""No prefix: txn_opname_id.log"""
|
|
m = FILE_REGEX_FIXED.match("txn_compaction_abc123.log")
|
|
self.assertIsNotNone(m)
|
|
self.assertIsNone(m.group(1))
|
|
self.assertEqual(m.group(2), "compaction")
|
|
self.assertEqual(m.group(3), "abc123")
|
|
|
|
def test_fixed_matches_underscores_in_opname(self):
|
|
"""Multiple underscores in opname: txn_compact_partial_merge_id.log
|
|
Last underscore splits opname from id."""
|
|
m = FILE_REGEX_FIXED.match("txn_compact_partial_merge_id.log")
|
|
self.assertIsNotNone(m)
|
|
self.assertEqual(m.group(2), "compact_partial_merge")
|
|
self.assertEqual(m.group(3), "id")
|
|
|
|
def test_fixed_rejects_no_log_extension(self):
|
|
m = FILE_REGEX_FIXED.match("txn_op_id.txt")
|
|
self.assertIsNone(m)
|
|
|
|
# -- Complexity gate --
|
|
|
|
def test_pathological_input_fixed_fast(self):
|
|
"""Patched pattern must handle N=10000 underscored segments in <1s."""
|
|
evil = "aa_txn_" + "x_" * 10000 + "!"
|
|
_, elapsed = time_match(FILE_REGEX_FIXED, evil)
|
|
self.assertLess(elapsed, 1.0,
|
|
f"Fixed FILE_REGEX took {elapsed:.3f}s on N=10000 (limit 1s)")
|
|
|
|
def test_pathological_input_original_slow(self):
|
|
"""Original pattern exhibits backtracking on underscore-heavy input."""
|
|
evil_small = "aa_txn_" + "x_" * 20 + "!"
|
|
_, elapsed_orig = time_match(FILE_REGEX_ORIG, evil_small, timeout=5.0)
|
|
evil_large = "aa_txn_" + "x_" * 10000 + "!"
|
|
_, elapsed_fixed = time_match(FILE_REGEX_FIXED, evil_large)
|
|
self.assertLess(elapsed_fixed, 1.0,
|
|
"Fixed pattern must stay fast at N=10000")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|