247 lines
9.2 KiB
Python
247 lines
9.2 KiB
Python
"""
|
||
test_suricata_0002.py — CWE-407 O(H×F) HTTP header field membership scan
|
||
|
||
Defect: EveHttpLogJSONHeaders in src/output-json-http.c iterates all
|
||
HTTP_FIELD_SIZE=53 entries of http_fields[] for every incoming HTTP header
|
||
to decide whether to log it. With H headers per request and F=53 fields,
|
||
this is O(H×F) per transaction.
|
||
|
||
Fix: precompute a list of only the E enabled field names at config time.
|
||
Runtime cost drops to O(H×E) where E is the count of enabled fields
|
||
(typically 1-5).
|
||
|
||
Benchmark shows the defective approach has ~F/E times more iterations than
|
||
the fixed approach. At E=1 that is a 53x ratio. Our benchmark uses
|
||
E=1 and asserts speedup > 3x.
|
||
"""
|
||
|
||
import time
|
||
import sys
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Simulated data matching Suricata's structure
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# http_fields[] — the full 53-entry table (htp field names only)
|
||
ALL_HTTP_FIELDS = [
|
||
"accept", "accept-charset", "accept-encoding", "accept-language",
|
||
"accept-datetime", "authorization", "cache-control", "cookie",
|
||
"from", "max-forwards", "origin", "pragma", "proxy-authorization",
|
||
"range", "te", "via", "x-requested-with", "dnt", "x-forwarded-proto",
|
||
"x-authenticated-user", "x-flash-version", "accept-range", "age",
|
||
"allow", "connection", "content-encoding", "content-language",
|
||
"content-length", "content-location", "content-md5", "content-range",
|
||
"content-type", "date", "etags", "expires", "last-modified", "link",
|
||
"location", "proxy-authenticate", "referer", "refresh", "retry-after",
|
||
"server", "set-cookie", "trailer", "transfer-encoding", "upgrade",
|
||
"vary", "warning", "www-authenticate", "true-client-ip", "org-src-ip",
|
||
"x-bluecoat-via",
|
||
]
|
||
|
||
HTTP_FIELD_SIZE = len(ALL_HTTP_FIELDS)
|
||
assert HTTP_FIELD_SIZE == 53, f"Expected 53 fields, got {HTTP_FIELD_SIZE}"
|
||
|
||
|
||
def make_headers(n: int):
|
||
"""Generate n realistic HTTP request headers. Most will NOT match."""
|
||
base = [
|
||
"host", "user-agent", "accept", "accept-encoding",
|
||
"accept-language", "connection", "upgrade-insecure-requests",
|
||
"sec-fetch-dest", "sec-fetch-mode", "sec-fetch-site",
|
||
"cache-control", "pragma", "x-forwarded-for",
|
||
"x-real-ip", "content-type", "content-length",
|
||
"if-modified-since", "if-none-match", "origin",
|
||
"referer",
|
||
]
|
||
headers = []
|
||
for i in range(n):
|
||
headers.append(base[i % len(base)])
|
||
return headers
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DEFECTIVE: iterate all 53 http_fields entries for every header
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def defective_should_log(header_name: str, all_fields: list, fields_bitmask: int) -> bool:
|
||
"""
|
||
Mirrors the defective inner loop in EveHttpLogJSONHeaders.
|
||
For each incoming header we iterate all HTTP_FIELD_SIZE=53 entries.
|
||
"""
|
||
for f, field_name in enumerate(all_fields):
|
||
if fields_bitmask & (1 << f):
|
||
if header_name.lower() == field_name.lower():
|
||
return True
|
||
return False
|
||
|
||
|
||
def defective_process_headers(headers: list, all_fields: list, fields_bitmask: int) -> int:
|
||
"""Process headers using the defective O(H × F) approach. Returns op count."""
|
||
ops = 0
|
||
for h in headers:
|
||
for f, field_name in enumerate(all_fields):
|
||
ops += 1
|
||
if fields_bitmask & (1 << f):
|
||
if h.lower() == field_name.lower():
|
||
break
|
||
return ops
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# FIXED: iterate only the E enabled fields (precomputed at config time)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def build_enabled_fields(all_fields: list, fields_bitmask: int) -> list:
|
||
"""Config-time: build compact list of enabled htp_field names. O(F) once."""
|
||
return [field_name for f, field_name in enumerate(all_fields)
|
||
if fields_bitmask & (1 << f)]
|
||
|
||
|
||
def fixed_should_log(header_name: str, enabled_fields: list) -> bool:
|
||
"""
|
||
Fixed inner loop: iterate only E enabled fields.
|
||
"""
|
||
h_lower = header_name.lower()
|
||
for field_name in enabled_fields:
|
||
if h_lower == field_name:
|
||
return True
|
||
return False
|
||
|
||
|
||
def fixed_process_headers(headers: list, enabled_fields: list) -> int:
|
||
"""Process headers using the fixed O(H × E) approach. Returns op count."""
|
||
ops = 0
|
||
for h in headers:
|
||
h_lower = h.lower()
|
||
for field_name in enabled_fields:
|
||
ops += 1
|
||
if h_lower == field_name:
|
||
break
|
||
return ops
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_correctness():
|
||
"""Both approaches must agree on which headers to log."""
|
||
# Enable only 'authorization' (bit 5) and 'x-forwarded-proto' (bit 18)
|
||
fields_bitmask = (1 << 5) | (1 << 18)
|
||
enabled_fields = build_enabled_fields(ALL_HTTP_FIELDS, fields_bitmask)
|
||
assert enabled_fields == ["authorization", "x-forwarded-proto"], (
|
||
f"Unexpected enabled fields: {enabled_fields}"
|
||
)
|
||
|
||
test_headers = [
|
||
"host", "authorization", "accept", "x-forwarded-proto", "content-type",
|
||
"cookie", "user-agent",
|
||
]
|
||
for h in test_headers:
|
||
defect = defective_should_log(h, ALL_HTTP_FIELDS, fields_bitmask)
|
||
fixed = fixed_should_log(h, enabled_fields)
|
||
assert defect == fixed, (
|
||
f"Mismatch for header '{h}': defective={defect}, fixed={fixed}"
|
||
)
|
||
print("PASS correctness: defective and fixed agree on all headers")
|
||
|
||
|
||
def test_op_count_ratio(n_headers: int, enabled_field_count: int):
|
||
"""
|
||
Op-count ratio must exceed 3x.
|
||
With E=1 enabled field and F=53, expected ratio is ~53x.
|
||
"""
|
||
# Enable E fields starting from index 5 (authorization onwards)
|
||
fields_bitmask = 0
|
||
for i in range(enabled_field_count):
|
||
fields_bitmask |= 1 << (5 + i)
|
||
|
||
enabled_fields = build_enabled_fields(ALL_HTTP_FIELDS, fields_bitmask)
|
||
assert len(enabled_fields) == enabled_field_count
|
||
|
||
headers = make_headers(n_headers)
|
||
|
||
defect_ops = defective_process_headers(headers, ALL_HTTP_FIELDS, fields_bitmask)
|
||
fixed_ops = fixed_process_headers(headers, enabled_fields)
|
||
|
||
ratio = defect_ops / max(fixed_ops, 1)
|
||
print(f" H={n_headers}, E={enabled_field_count}, F={HTTP_FIELD_SIZE}: "
|
||
f"defective={defect_ops} ops, fixed={fixed_ops} ops, ratio={ratio:.1f}x")
|
||
|
||
assert ratio > 3.0, (
|
||
f"Expected speedup > 3x, got {ratio:.2f}x "
|
||
f"(H={n_headers}, E={enabled_field_count})"
|
||
)
|
||
return ratio
|
||
|
||
|
||
def test_timing(n_headers: int, enabled_field_count: int, iterations: int = 5000):
|
||
"""Wall-clock timing test to confirm real-world speedup."""
|
||
fields_bitmask = 0
|
||
for i in range(enabled_field_count):
|
||
fields_bitmask |= 1 << (5 + i)
|
||
enabled_fields = build_enabled_fields(ALL_HTTP_FIELDS, fields_bitmask)
|
||
headers = make_headers(n_headers)
|
||
|
||
t0 = time.perf_counter()
|
||
for _ in range(iterations):
|
||
defective_process_headers(headers, ALL_HTTP_FIELDS, fields_bitmask)
|
||
t_defect = time.perf_counter() - t0
|
||
|
||
t0 = time.perf_counter()
|
||
for _ in range(iterations):
|
||
fixed_process_headers(headers, enabled_fields)
|
||
t_fixed = time.perf_counter() - t0
|
||
|
||
ratio = t_defect / max(t_fixed, 1e-9)
|
||
print(f" Timing H={n_headers} E={enabled_field_count} × {iterations} iters: "
|
||
f"defective={t_defect*1000:.1f}ms, fixed={t_fixed*1000:.1f}ms, "
|
||
f"ratio={ratio:.1f}x")
|
||
assert ratio > 3.0, (
|
||
f"Expected timing speedup > 3x, got {ratio:.2f}x"
|
||
)
|
||
return ratio
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Main
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def main():
|
||
print("=== suricata-0002: CWE-407 O(H×F) header field membership scan ===")
|
||
print()
|
||
|
||
print("[1] Correctness check")
|
||
test_correctness()
|
||
print()
|
||
|
||
print("[2] Op-count ratios (N=100 headers)")
|
||
r1 = test_op_count_ratio(n_headers=100, enabled_field_count=1)
|
||
r2 = test_op_count_ratio(n_headers=100, enabled_field_count=3)
|
||
r3 = test_op_count_ratio(n_headers=100, enabled_field_count=5)
|
||
print()
|
||
|
||
print("[3] Op-count ratios (N=1000 headers)")
|
||
r4 = test_op_count_ratio(n_headers=1000, enabled_field_count=1)
|
||
r5 = test_op_count_ratio(n_headers=1000, enabled_field_count=3)
|
||
print()
|
||
|
||
print("[4] Wall-clock timing")
|
||
tr1 = test_timing(n_headers=50, enabled_field_count=1, iterations=5000)
|
||
tr2 = test_timing(n_headers=50, enabled_field_count=3, iterations=5000)
|
||
print()
|
||
|
||
worst_ratio = min(r1, r2, r3, r4, r5, tr1, tr2)
|
||
print(f"Minimum speedup across all scenarios: {worst_ratio:.1f}x")
|
||
assert worst_ratio > 3.0, f"FAIL: worst-case ratio {worst_ratio:.2f}x < 3x"
|
||
|
||
print()
|
||
print("PASS all suricata-0002 tests")
|
||
print(f" Peak speedup: {max(r1, r4):.1f}x at E=1 enabled field")
|
||
print(f" Defect: EveHttpLogJSONHeaders O(H × F=53) per transaction")
|
||
print(f" Fix: precomputed enabled_htp_fields[] O(H × E) per transaction")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|