diff --git a/defects/cassandra/unit/test_cassandra_cwe1333.py b/defects/cassandra/unit/test_cassandra_cwe1333.py new file mode 100644 index 000000000..32f3c89e1 --- /dev/null +++ b/defects/cassandra/unit/test_cassandra_cwe1333.py @@ -0,0 +1,222 @@ +""" +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() diff --git a/defects/hadoop/unit/test_hadoop_cwe1333.py b/defects/hadoop/unit/test_hadoop_cwe1333.py new file mode 100644 index 000000000..b7ece3dbd --- /dev/null +++ b/defects/hadoop/unit/test_hadoop_cwe1333.py @@ -0,0 +1,135 @@ +""" +CWE-1333 ReDoS tests for Apache Hadoop regex patterns. + +Tests one confirmed true positive: + 1. RPC.Server.COMPLEX_SERVER_NAME_PATTERN (hadoop-0001-rpc) + +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) pattern — Python transliteration of Java source +# --------------------------------------------------------------------------- + +# (?:[^\$]*\$)*([A-Za-z][^\$]+)(?:\$\d+)? +# In Python, $ in character class does not need escaping, but we keep it +# faithful to the Java source which escapes \$. +COMPLEX_SERVER_NAME_ORIG = re.compile( + r"(?:[^\$]*\$)*([A-Za-z][^\$]+)(?:\$\d+)?" +) + +# --------------------------------------------------------------------------- +# Patched pattern: inner * changed to + +# --------------------------------------------------------------------------- + +COMPLEX_SERVER_NAME_FIXED = re.compile( + r"(?:[^\$]+\$)*([A-Za-z][^\$]+)(?:\$\d+)?" +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def time_match(pattern, text, timeout=5.0): + """Return (match_object, elapsed_seconds).""" + start = time.monotonic() + m = pattern.search(text) + elapsed = time.monotonic() - start + return m, elapsed + + +# =========================================================================== +# Test: RPC.Server.COMPLEX_SERVER_NAME_PATTERN (hadoop-0001-rpc) +# =========================================================================== + +class TestComplexServerNameReDoS(unittest.TestCase): + """hadoop-0001-rpc: star-of-star (?:[^\\$]*\\$)* in COMPLEX_SERVER_NAME_PATTERN.""" + + # -- Functional correctness -- + + def test_simple_class_name(self): + """Simple class: 'MyServer' -> group(1) = 'MyServer'""" + m = COMPLEX_SERVER_NAME_FIXED.search("MyServer") + self.assertIsNotNone(m) + self.assertEqual(m.group(1), "MyServer") + + def test_inner_class(self): + """Inner class: 'Outer$Inner' -> group(1) = 'Inner'""" + m = COMPLEX_SERVER_NAME_FIXED.search("Outer$Inner") + self.assertIsNotNone(m) + self.assertEqual(m.group(1), "Inner") + + def test_nested_inner_class(self): + """Nested: 'A$B$CD' -> group(1) = 'CD' (capture requires 2+ chars)""" + m = COMPLEX_SERVER_NAME_FIXED.search("A$B$CD") + self.assertIsNotNone(m) + self.assertEqual(m.group(1), "CD") + + def test_anonymous_inner_class(self): + """Anonymous: 'Outer$Inner$1' -> group(1) = 'Inner', trailing $1 consumed""" + m = COMPLEX_SERVER_NAME_FIXED.search("Outer$Inner$1") + self.assertIsNotNone(m) + self.assertEqual(m.group(1), "Inner") + + def test_protobuf_generated(self): + """Protobuf: 'ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2'""" + m = COMPLEX_SERVER_NAME_FIXED.search( + "ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2" + ) + self.assertIsNotNone(m) + self.assertEqual(m.group(1), "ClientNamenodeProtocol") + + def test_deeply_nested(self): + """Deep nesting: 'A$B$C$D$RealName$3'""" + m = COMPLEX_SERVER_NAME_FIXED.search("A$B$C$D$RealName$3") + self.assertIsNotNone(m) + self.assertEqual(m.group(1), "RealName") + + def test_package_qualified(self): + """Full package path: 'org.apache.hadoop.ipc.TestRPC$TestProtocol$1'""" + m = COMPLEX_SERVER_NAME_FIXED.search( + "org.apache.hadoop.ipc.TestRPC$TestProtocol$1" + ) + self.assertIsNotNone(m) + self.assertEqual(m.group(1), "TestProtocol") + + # -- Complexity gate -- + + def test_pathological_input_fixed_fast(self): + """Patched pattern must handle N=10000 chars without $ in <1s. + + The original star-of-star tries O(2^N) empty-match partitions + when no $ delimiter exists and the trailing anchor fails. + With the fix ([^$]+$ requiring progress), the engine runs in O(N). + """ + # Input with no $ signs and a trailing char that prevents + # the ([A-Za-z][^\$]+) group from matching cleanly + evil = "a" * 10000 + "!" + _, elapsed = time_match(COMPLEX_SERVER_NAME_FIXED, evil) + self.assertLess(elapsed, 1.0, + f"Fixed pattern took {elapsed:.3f}s on N=10000 (limit 1s)") + + def test_pathological_no_match_fixed_fast(self): + """Non-alpha start forces full scan, must stay O(N).""" + evil = "1" * 10000 + _, elapsed = time_match(COMPLEX_SERVER_NAME_FIXED, evil) + self.assertLess(elapsed, 1.0, + f"Fixed pattern took {elapsed:.3f}s on numeric N=10000 (limit 1s)") + + def test_many_dollar_segments_fixed_fast(self): + """Many $ segments: 'a$' * N + '!' must stay O(N).""" + evil = "a$" * 5000 + "!" + _, elapsed = time_match(COMPLEX_SERVER_NAME_FIXED, evil) + self.assertLess(elapsed, 2.0, + f"Fixed pattern took {elapsed:.3f}s on 5000 segments (limit 2s)") + + +if __name__ == "__main__": + unittest.main()