Skip wave-attack detection for bypassed requests

Adds a defensive bypass-flag check inside is_attack_wave so that wave
detection consistently honours the BypassedContextStore signal even if a
caller forgot to skip post_response for a bypassed request.

Also adds two regression tests:
- 16 requests with distinct query strings produce 15 unique samples
  (previously collapsed to 1 sample because URL stripped the query string).
- Bypassed flag set: is_attack_wave returns False and never collects samples.

Re-enables test_wave_attack in the QA suite. The QA test relies on:
- The BypassedContextStore wiring from "Skip context creation for bypassed IPs".
- The query-string preservation from "Include query string in URL for ASGI/WSGI requests".

Author: bitterpanda63

.github/workflows/qa-tests.yml

@@ -52,4 +52,4 @@ jobs:
           app_port: 8080
           sleep_before_test: 30
           config_update_delay: 100
-          skip_tests: test_bypassed_ip_for_geo_blocking,test_demo_apps_generic_tests,test_outbound_domain_blocking,test_bypassed_ip,test_wave_attack,test_block_traffic_by_countries,test_user_rate_limiting_1_minute
+          skip_tests: test_bypassed_ip_for_geo_blocking,test_demo_apps_generic_tests,test_outbound_domain_blocking,test_bypassed_ip,test_block_traffic_by_countries,test_user_rate_limiting_1_minute

aikido_zen/storage/attack_wave_detector_store_test.py

@@ -10,9 +10,17 @@
     AttackWaveDetectorStore,
     attack_wave_detector_store,
 )
+from . import bypassed_context_store
 import aikido_zen.test_utils as test_utils
 
 
+@pytest.fixture(autouse=True)
+def _clear_bypass_flag():
+    bypassed_context_store.clear()
+    yield
+    bypassed_context_store.clear()
+
+
 def test_attack_wave_detector_store_initialization():
     """Test that the store initializes correctly"""
     store = AttackWaveDetectorStore()
@@ -436,6 +444,49 @@ def create_context_with_url(ip, url, method="GET"):
                 assert set(sample.keys()) == {"method", "url"}
 
 
+def test_samples_collect_one_per_unique_url():
+    """16 requests with distinct query strings should produce 16 unique samples."""
+    store = AttackWaveDetectorStore()
+    ip = "2.16.53.8"
+
+    with patch(
+        "aikido_zen.vulnerabilities.attack_wave_detection.attack_wave_detector.is_web_scanner",
+        return_value=True,
+    ):
+        for i in range(16):
+            ctx = test_utils.generate_context(
+                ip=ip,
+                method="GET",
+                url=f"http://localhost:3018/api/pets/?path=q{i}",
+            )
+            store.is_attack_wave(ctx)
+
+    samples = store.get_samples_for_ip(ip)
+    assert len(samples) == 15, f"expected 15 samples, got {len(samples)}"
+
+
+def test_bypassed_request_skips_wave_detection():
+    """When the bypass flag is set, is_attack_wave never reports a wave."""
+    store = AttackWaveDetectorStore()
+    ip = "2.16.53.10"
+
+    with patch(
+        "aikido_zen.vulnerabilities.attack_wave_detection.attack_wave_detector.is_web_scanner",
+        return_value=True,
+    ):
+        bypassed_context_store.set_bypassed(True)
+        for i in range(16):
+            ctx = test_utils.generate_context(
+                ip=ip,
+                method="GET",
+                url=f"http://localhost:3018/api/execute/.env{i}",
+            )
+            assert store.is_attack_wave(ctx) is False
+
+    # Bypassed requests should not collect samples either.
+    assert store.get_samples_for_ip(ip) == []
+
+
 @patch("aikido_zen.storage.attack_wave_detector_store.AttackWaveDetector")
 def test_mock_detector_integration(mock_detector_class):
     """Test integration with mocked AttackWaveDetector"""

aikido_zen/storage/bypassed_context_store.py

@@ -0,0 +1,27 @@
+"""
+Records whether the current request's remote IP is in the bypass list.
+
+Bypassed requests intentionally do not set a Context (so all per-request
+protection short-circuits), but checks that run without a Context — for
+example outbound DNS reporting — still need a way to detect "this work
+was triggered by a bypassed request".
+"""
+
+import contextvars
+
+_bypassed = contextvars.ContextVar("aikido_bypassed_ip", default=False)
+
+
+def set_bypassed(value: bool) -> None:
+    _bypassed.set(bool(value))
+
+
+def is_bypassed() -> bool:
+    try:
+        return _bypassed.get()
+    except Exception:
+        return False
+
+
+def clear() -> None:
+    _bypassed.set(False)

aikido_zen/storage/bypassed_context_store_test.py

@@ -0,0 +1,37 @@
+import pytest
+
+from aikido_zen.storage import bypassed_context_store
+
+
+@pytest.fixture(autouse=True)
+def _reset_after_each_test():
+    yield
+    bypassed_context_store.clear()
+
+
+def test_default_is_false():
+    assert bypassed_context_store.is_bypassed() is False
+
+
+def test_set_bypassed_true_then_false():
+    bypassed_context_store.set_bypassed(True)
+    assert bypassed_context_store.is_bypassed() is True
+
+    bypassed_context_store.set_bypassed(False)
+    assert bypassed_context_store.is_bypassed() is False
+
+
+def test_clear_resets_to_false():
+    bypassed_context_store.set_bypassed(True)
+    assert bypassed_context_store.is_bypassed() is True
+
+    bypassed_context_store.clear()
+    assert bypassed_context_store.is_bypassed() is False
+
+
+def test_truthy_values_coerced_to_bool():
+    bypassed_context_store.set_bypassed("yes")
+    assert bypassed_context_store.is_bypassed() is True
+
+    bypassed_context_store.set_bypassed(0)
+    assert bypassed_context_store.is_bypassed() is False

aikido_zen/vulnerabilities/attack_wave_detection/attack_wave_detector.py

@@ -1,6 +1,7 @@
 import aikido_zen.helpers.get_current_unixtime_ms as internal_time
 from aikido_zen.ratelimiting.lru_cache import LRUCache
 from aikido_zen.context import Context
+from aikido_zen.storage import bypassed_context_store
 from aikido_zen.vulnerabilities.attack_wave_detection.is_web_scanner import (
     is_web_scanner,
 )
@@ -40,6 +41,11 @@ def is_attack_wave(self, context: Context) -> bool:
         """
         if not context or not context.remote_address:
             return False
+        if bypassed_context_store.is_bypassed():
+            # Defensive: callers should normally never reach here for bypassed
+            # IPs (the framework entry point clears the context), but keep wave
+            # detection consistent with all other per-request blocking sites.
+            return False
         ip = context.remote_address
 
         if self.sent_events_map.get(ip) is not None: