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.
135 lines
5.1 KiB
Python
135 lines
5.1 KiB
Python
"""
|
|
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()
|