s3fs-fuse-0001: StatCache::RawGetChildStats() dedup uses std::find() on std::vector<std::string> inside loop over childmap. O(N*M) on every readdir() and rename_directory() call. Fix: unordered_set for O(1) lookup. MEDIUM severity, 13.8x at N=2000/M=1000. MOAD-0002 (intertangle): globals are config-only, set at startup. Subsystems (stat cache, fd cache, curl) are properly isolated behind singleton + mutex. CLEAN. MOAD-0003 (leaked context): no thread_local usage found. CLEAN. MOAD-0004 (logged secret): all credential logging uses mask_sensitive_string(). insecure_logging is opt-in and deprecated. CLEAN. MOAD-0005 (thundering herd): stat cache uses std::mutex properly. curl handle pool uses lock_guard. No unprotected cache paths. CLEAN.
95 lines
3 KiB
Python
95 lines
3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
iroh-0001: Relay server AccessConfig Allowlist/Denylist uses Vec<EndpointId>
|
|
with .contains() for every connection, O(C * L) total work.
|
|
|
|
Fix: HashSet<EndpointId> gives O(1) lookup per connection, O(C) total.
|
|
|
|
This test demonstrates our quadratic blowup by simulating our access check
|
|
pattern with both Vec (linear scan) and HashSet (constant-time lookup).
|
|
"""
|
|
|
|
import time
|
|
import random
|
|
import string
|
|
import unittest
|
|
|
|
|
|
def make_endpoint_ids(n):
|
|
"""Generate n unique fake endpoint IDs (32-byte hex strings)."""
|
|
return [
|
|
"".join(random.choices(string.hexdigits[:16], k=64))
|
|
for _ in range(n)
|
|
]
|
|
|
|
|
|
def vec_contains_check(allow_list, endpoint_id):
|
|
"""Simulate Vec<EndpointId>.contains() -- linear scan."""
|
|
for item in allow_list:
|
|
if item == endpoint_id:
|
|
return True
|
|
return False
|
|
|
|
|
|
def hashset_contains_check(allow_set, endpoint_id):
|
|
"""Simulate HashSet<EndpointId>.contains() -- O(1) lookup."""
|
|
return endpoint_id in allow_set
|
|
|
|
|
|
class TestIroh0001(unittest.TestCase):
|
|
"""
|
|
CWE-407: Relay server allowlist/denylist Vec.contains() is O(L) per
|
|
connection. With L endpoints in our list and C connections, total work
|
|
is O(C * L). HashSet reduces to O(C).
|
|
"""
|
|
|
|
def test_vec_quadratic_vs_hashset_constant(self):
|
|
L = 5000 # endpoints in allowlist
|
|
C = 5000 # connection attempts
|
|
|
|
ids = make_endpoint_ids(L)
|
|
# Pick random IDs to check (some in list, some not)
|
|
check_ids = [random.choice(ids) for _ in range(C)]
|
|
|
|
# --- Vec path (linear scan per connection) ---
|
|
vec_list = list(ids)
|
|
start = time.perf_counter()
|
|
for eid in check_ids:
|
|
vec_contains_check(vec_list, eid)
|
|
vec_elapsed = time.perf_counter() - start
|
|
|
|
# --- HashSet path (O(1) per connection) ---
|
|
hash_set = set(ids)
|
|
start = time.perf_counter()
|
|
for eid in check_ids:
|
|
hashset_contains_check(hash_set, eid)
|
|
set_elapsed = time.perf_counter() - start
|
|
|
|
ratio = vec_elapsed / set_elapsed if set_elapsed > 0 else float("inf")
|
|
print(f"\niroh-0001: Vec elapsed={vec_elapsed:.4f}s, HashSet elapsed={set_elapsed:.6f}s, ratio={ratio:.1f}x")
|
|
|
|
# Vec path should be significantly slower
|
|
self.assertGreater(ratio, 5.0,
|
|
f"Expected Vec path to be >5x slower than HashSet, got {ratio:.1f}x")
|
|
|
|
def test_correctness(self):
|
|
"""Both paths must return identical results."""
|
|
ids = make_endpoint_ids(200)
|
|
vec_list = list(ids)
|
|
hash_set = set(ids)
|
|
|
|
# Check all known IDs
|
|
for eid in ids:
|
|
self.assertTrue(vec_contains_check(vec_list, eid))
|
|
self.assertTrue(hashset_contains_check(hash_set, eid))
|
|
|
|
# Check unknown IDs
|
|
unknown = make_endpoint_ids(50)
|
|
for eid in unknown:
|
|
vec_result = vec_contains_check(vec_list, eid)
|
|
set_result = hashset_contains_check(hash_set, eid)
|
|
self.assertEqual(vec_result, set_result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|