chore: add nosec B324 annotations to non-cryptographic MD5 usages and update temporary database path to use system temp directory

benchmarks/compression_benchmark.py

@@ -159,7 +159,7 @@ def generate_log_data(
             entry = random.choice(log_templates).copy()
 
         entry["timestamp"] = f"2024-01-15T{10 + (i // 60):02d}:{i % 60:02d}:00Z"
-        entry["request_id"] = f"req-{hashlib.md5(str(i).encode()).hexdigest()[:8]}"
+        entry["request_id"] = f"req-{hashlib.md5(str(i).encode()).hexdigest()[:8]}"  # nosec B324
         logs.append(entry)
 
     # Questions designed to test different approaches

benchmarks/headroom_adversarial_benchmark.py

@@ -174,7 +174,7 @@ def generate_code_diff(num_files: int = 15, changes_per_file: int = 20) -> dict:
         filename = f"src/module_{file_idx}/handler.{ext}"
 
         diff_output += f"diff --git a/{filename} b/{filename}\n"
-        diff_output += f"index {hashlib.md5(f'{file_idx}a'.encode()).hexdigest()[:7]}..{hashlib.md5(f'{file_idx}b'.encode()).hexdigest()[:7]} 100644\n"
+        diff_output += f"index {hashlib.md5(f'{file_idx}a'.encode()).hexdigest()[:7]}..{hashlib.md5(f'{file_idx}b'.encode()).hexdigest()[:7]} 100644\n"  # nosec B324
         diff_output += f"--- a/{filename}\n"
         diff_output += f"+++ b/{filename}\n"
 
@@ -227,7 +227,7 @@ def generate_encrypted_data(size_kb: int = 20) -> dict:
     return {
         "tool": "encrypted_blob",
         "result": {
-            "blob_id": f"enc_{hashlib.md5(encoded[:100].encode()).hexdigest()[:16]}",
+            "blob_id": f"enc_{hashlib.md5(encoded[:100].encode()).hexdigest()[:16]}",  # nosec B324
             "encryption": "AES-256-GCM",
             "content": encoded,
             "size_bytes": len(random_bytes),
@@ -246,7 +246,7 @@ def generate_tiny_dataset(num_items: int = 5) -> dict:
                 "id": i + 1,
                 "name": f"Item {chr(65 + i)}",
                 "value": random.randint(100, 999),
-                "note": f"Unique note for item {i + 1}: {hashlib.md5(str(i).encode()).hexdigest()[:20]}",
+                "note": f"Unique note for item {i + 1}: {hashlib.md5(str(i).encode()).hexdigest()[:20]}",  # nosec B324
             }
         )
 
@@ -422,7 +422,7 @@ def run_scenario(
 
     # Add tool results
     for tool_output in scenario.tools:
-        tool_call_id = f"call_{hashlib.md5(tool_output['tool'].encode()).hexdigest()[:8]}"
+        tool_call_id = f"call_{hashlib.md5(tool_output['tool'].encode()).hexdigest()[:8]}"  # nosec B324
         messages.append(
             {
                 "role": "assistant",

benchmarks/headroom_worst_case_benchmark.py

@@ -100,14 +100,14 @@ def generate_unique_support_tickets(num_tickets: int = 50) -> dict:
             days=random.randint(2, 14),
             from_plan=random.choice(["Enterprise", "Pro"]),
             to_plan=random.choice(["Starter", "Team"]),
-            user_id=f"usr_{hashlib.md5(str(i).encode()).hexdigest()[:8]}",
+            user_id=f"usr_{hashlib.md5(str(i).encode()).hexdigest()[:8]}",  # nosec B324
         )
 
         tickets.append(
             {
                 "ticket_id": f"TKT-{20000 + i}",
                 "customer": {
-                    "id": f"cust_{hashlib.md5(f'customer{i}'.encode()).hexdigest()[:8]}",
+                    "id": f"cust_{hashlib.md5(f'customer{i}'.encode()).hexdigest()[:8]}",  # nosec B324
                     "name": f"Customer {i + 1}",
                     "company": f"Company {chr(65 + (i % 26))}{i // 26 + 1} Inc.",
                     "plan": random.choice(products),
@@ -159,7 +159,7 @@ def generate_unique_error_traces(num_traces: int = 30) -> dict:
 
         traces.append(
             {
-                "error_id": f"err_{hashlib.md5(str(i).encode()).hexdigest()[:12]}",
+                "error_id": f"err_{hashlib.md5(str(i).encode()).hexdigest()[:12]}",  # nosec B324
                 "timestamp": f"2024-01-17T{10 + (i % 12):02d}:{(i * 7) % 60:02d}:00Z",
                 "service": random.choice(["api", "worker", "scheduler", "gateway"]),
                 "environment": "production",
@@ -169,7 +169,7 @@ def generate_unique_error_traces(num_traces: int = 30) -> dict:
                 "stack_trace": trace["stack"],
                 "context": {
                     "user_id": f"user_{random.randint(10000, 99999)}",
-                    "request_id": hashlib.md5(f"req{i}".encode()).hexdigest()[:16],
+                    "request_id": hashlib.md5(f"req{i}".encode()).hexdigest()[:16],  # nosec B324
                     "endpoint": trace.get("endpoint", "/api/unknown"),
                 },
                 "occurrence_count": random.randint(1, 5),  # Low count - each is unique
@@ -630,7 +630,7 @@ def run_scenario(
 
     # Add tool results with proper format
     for tool_output in scenario.tools:
-        tool_call_id = f"call_{hashlib.md5(tool_output['tool'].encode()).hexdigest()[:8]}"
+        tool_call_id = f"call_{hashlib.md5(tool_output['tool'].encode()).hexdigest()[:8]}"  # nosec B324
         messages.append(
             {
                 "role": "assistant",

benchmarks/real_world_agent_benchmark.py

@@ -386,7 +386,7 @@ def generate_log_search(query: str, num_entries: int = 300) -> dict:
                 "level": level,
                 "service": service,
                 "message": message,
-                "trace_id": hashlib.md5(f"{i}".encode()).hexdigest()[:16],
+                "trace_id": hashlib.md5(f"{i}".encode()).hexdigest()[:16],  # nosec B324
                 "metadata": {
                     "host": f"pod-{service}-{random.randint(1, 5)}",
                     "region": random.choice(["us-east-1", "us-west-2", "eu-west-1"]),
@@ -518,7 +518,7 @@ def run_agent_scenario(
 
     # Add tool results with proper OpenAI format
     for tool_output in scenario.tools:
-        tool_call_id = f"call_{hashlib.md5(tool_output['tool'].encode()).hexdigest()[:8]}"
+        tool_call_id = f"call_{hashlib.md5(tool_output['tool'].encode()).hexdigest()[:8]}"  # nosec B324
         # Assistant message with tool_calls (required by OpenAI)
         messages.append(
             {

headroom/cache/base.py

@@ -320,7 +320,7 @@ class BaseCacheOptimizer(ABC):
         """Compute a short hash of content."""
         import hashlib
 
-        return hashlib.md5(content.encode()).hexdigest()[:12]
+        return hashlib.md5(content.encode()).hexdigest()[:12]  # nosec B324
 
     def _extract_system_content(self, messages: list[dict[str, Any]]) -> str:
         """Extract content from system messages."""

headroom/cache/compression_cache.py

@@ -135,7 +135,7 @@ class CompressionCache:
             raw = json.dumps(content, sort_keys=True, ensure_ascii=False)
         else:
             raw = content
-        return hashlib.md5(raw.encode("utf-8")).hexdigest()[:16]
+        return hashlib.md5(raw.encode("utf-8")).hexdigest()[:16]  # nosec B324
 
     def compute_frozen_count(self, messages: list[dict]) -> int:
         """Count consecutive stable messages from the start.

headroom/cache/compression_store.py

@@ -206,7 +206,7 @@ class CompressionStore:
         # collision resistance. Birthday paradox: 50% collision at sqrt(2^n) entries.
         # - 64 bits: ~4 billion entries for 50% collision
         # - 96 bits: ~280 trillion entries for 50% collision
-        hash_key = hashlib.md5(original.encode()).hexdigest()[:24]
+        hash_key = hashlib.md5(original.encode()).hexdigest()[:24]  # nosec B324
 
         entry = CompressionEntry(
             hash=hash_key,

headroom/cache/prefix_tracker.py

@@ -305,7 +305,7 @@ class SessionTrackerStore:
                 break
 
         key = f"{model}:{system_content}"
-        return hashlib.md5(key.encode()).hexdigest()[:16]
+        return hashlib.md5(key.encode()).hexdigest()[:16]  # nosec B324
 
     def _maybe_cleanup(self) -> None:
         """Remove expired trackers periodically."""

headroom/evals/memory/runner_v2.py

@@ -17,6 +17,7 @@ import json
 import logging
 import time
 import uuid
+import tempfile
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -383,7 +384,8 @@ class LoCoMoEvaluatorV2:
                 self._backend = self._config.backend_factory(self._current_user_id)
             else:
                 # Default to LocalBackend
-                db_path = self._config.db_path or f"/tmp/locomo_v2_{uuid.uuid4().hex[:8]}.db"
+                db_dir = tempfile.gettempdir()
+                db_path = self._config.db_path or f"{db_dir}/locomo_v2_{uuid.uuid4().hex[:8]}.db"
                 backend_config = LocalBackendConfig(db_path=db_path)
                 self._backend = LocalBackend(backend_config)
 

headroom/evals/memory/runner_v3.py

@@ -22,6 +22,7 @@ import json
 import logging
 import time
 import uuid
+import tempfile
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
@@ -261,7 +262,8 @@ class LoCoMoEvaluatorV3:
                 graph_status = "with graph" if self._config.mem0_enable_graph else "vector-only"
                 logger.info(f"Using Mem0 backend ({self._config.mem0_mode} mode, {graph_status})")
             else:
-                db_path = self._config.db_path or f"/tmp/locomo_v3_{uuid.uuid4().hex[:8]}.db"
+                db_dir = tempfile.gettempdir()
+                db_path = self._config.db_path or f"{db_dir}/locomo_v3_{uuid.uuid4().hex[:8]}.db"
                 self._backend = LocalBackend(LocalBackendConfig(db_path=db_path))
                 logger.info("Using LocalBackend")
 

headroom/memory/backends/direct_mem0.py

@@ -275,7 +275,7 @@ class DirectMem0Adapter:
                 "user_id": user_id,
                 "importance": importance,
                 "created_at": _utcnow().isoformat(),
-                "hash": hashlib.md5(fact.encode()).hexdigest(),
+                "hash": hashlib.md5(fact.encode()).hexdigest(),  # nosec B324
                 **(metadata or {}),
             }
 

headroom/parser.py

@@ -33,7 +33,7 @@ RAG_PATTERN = re.compile("|".join(RAG_MARKERS), re.IGNORECASE)
 
 def compute_hash(text: str) -> str:
     """Compute hash of text, truncated to 16 chars."""
-    return hashlib.md5(text.encode()).hexdigest()[:16]
+    return hashlib.md5(text.encode()).hexdigest()[:16]  # nosec B324
 
 
 def detect_waste_signals(text: str, tokenizer: Tokenizer) -> WasteSignals:

headroom/prediction/feature_extractor.py

@@ -2138,7 +2138,7 @@ class MetaExtractor(BaseFeatureExtractor):
                 features.available_output_tokens = min(features.available_output_tokens, max_tokens)
 
         # Prompt hash
-        features.prompt_hash = hashlib.md5(text.encode()).hexdigest()[:16]
+        features.prompt_hash = hashlib.md5(text.encode()).hexdigest()[:16]  # nosec B324
         features.prompt_signature = self._compute_signature(text)
 
         # Conversation features
@@ -2324,7 +2324,7 @@ class PromptFeatureExtractor:
             PromptFeatures containing all extracted features.
         """
         # Check cache
-        cache_key = hashlib.md5(f"{prompt}:{model}:{system_prompt}".encode()).hexdigest()
+        cache_key = hashlib.md5(f"{prompt}:{model}:{system_prompt}".encode()).hexdigest()  # nosec B324
 
         if use_cache and cache_key in self._cache:
             return self._cache[cache_key]

headroom/transforms/adaptive_sizer.py

@@ -200,7 +200,7 @@ def _simhash(text: str) -> int:
     # Character 4-grams
     for i in range(max(1, len(text_lower) - 3)):
         gram = text_lower[i : i + 4]
-        h = int(hashlib.md5(gram.encode(), usedforsecurity=False).hexdigest()[:16], 16)
+        h = int(hashlib.md5(gram.encode(), usedforsecurity=False).hexdigest()[:16], 16)  # nosec B324
         for j in range(64):
             if h & (1 << j):
                 v[j] += 1

headroom/transforms/anchor_selector.py

@@ -334,7 +334,7 @@ def compute_item_hash(item: dict[str, Any]) -> str:
         content = json.dumps(item, sort_keys=True, default=str)
     except (TypeError, ValueError):
         content = str(item)
-    return hashlib.md5(content.encode()).hexdigest()[:16]
+    return hashlib.md5(content.encode()).hexdigest()[:16]  # nosec B324
 
 
 class AnchorSelector:

headroom/transforms/smart_crusher.py

@@ -1762,7 +1762,7 @@ class SmartCrusher(Transform):
                 else:
                     # Non-dict items: use string representation
                     content = str(item)
-                item_hash = hashlib.md5(content.encode()).hexdigest()[:16]
+                item_hash = hashlib.md5(content.encode()).hexdigest()[:16]  # nosec B324
             except (TypeError, ValueError, RecursionError) as e:
                 # Serialization failed - keep the item (fail-safe)
                 logger.debug("Dedup hash failed for item at index %d: %s. Keeping item.", idx, e)
@@ -1832,7 +1832,7 @@ class SmartCrusher(Transform):
                         content = json.dumps(item, sort_keys=True, default=str)
                     else:
                         content = str(item)
-                    seen_hashes.add(hashlib.md5(content.encode()).hexdigest()[:16])
+                    seen_hashes.add(hashlib.md5(content.encode()).hexdigest()[:16])  # nosec B324
                 except (TypeError, ValueError, RecursionError):
                     pass  # Skip hash computation failures
 
@@ -1865,7 +1865,7 @@ class SmartCrusher(Transform):
                         content = json.dumps(item, sort_keys=True, default=str)
                     else:
                         content = str(item)
-                    item_hash = hashlib.md5(content.encode()).hexdigest()[:16]
+                    item_hash = hashlib.md5(content.encode()).hexdigest()[:16]  # nosec B324
                 except (TypeError, ValueError, RecursionError):
                     # Hash failure - use index as unique hash (fail-safe)
                     item_hash = f"__idx_{idx}__"
@@ -3336,7 +3336,7 @@ class SmartCrusher(Transform):
             clusters: dict[str, list[int]] = {}
             for i, item in enumerate(items):
                 msg = str(item.get(message_field, ""))[:50]
-                msg_hash = hashlib.md5(msg.encode()).hexdigest()[:8]
+                msg_hash = hashlib.md5(msg.encode()).hexdigest()[:8]  # nosec B324
                 if msg_hash not in clusters:
                     clusters[msg_hash] = []
                 clusters[msg_hash].append(i)

headroom/utils.py

@@ -40,7 +40,7 @@ def fast_hash(data: str | bytes, length: int = 16) -> str:
     """
     if isinstance(data, str):
         data = data.encode("utf-8")
-    return hashlib.md5(data).hexdigest()[:length]
+    return hashlib.md5(data).hexdigest()[:length]  # nosec B324
 
 
 def extract_user_query(messages: list[dict[str, Any]]) -> str:

tests/test_compression_store.py

@@ -1252,7 +1252,7 @@ class TestHashCollisionDetection:
     def test_hash_uses_md5_truncated(self, store: CompressionStore):
         """Hash is MD5 truncated to 24 characters (fast, non-crypto)."""
         content = "test content"
-        expected_hash = hashlib.md5(content.encode()).hexdigest()[:24]
+        expected_hash = hashlib.md5(content.encode()).hexdigest()[:24]  # nosec B324
 
         hash_key = store.store(original=content, compressed="[]")