java-topology/defects/iroh-0001/test/test_iroh_0001.py
russell@unturf.com bcdca9cb0e s3fs-fuse: 1 CWE-407 defect, MOAD 0002-0005 CLEAN
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.
2026-03-31 13:05:50 -04:00

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()