make sure samples are unique

Author: bitterpanda63

aikido_zen/sources/functions/request_handler_test.py

@@ -877,8 +877,8 @@ def test_attack_wave_samples_structure(firewall_lists):
     # Get the samples that were stored for this IP
     samples = detector.get_samples_for_ip("11.11.11.11")
 
-    # Verify samples structure
-    assert len(samples) == 10  # Should keep last 10 samples
+    # Verify samples structure - should have only 1 unique sample since all requests are identical
+    assert len(samples) == 1  # Only 1 unique sample despite 15 identical requests
 
     # Verify each sample has correct structure (method and url only)
     for sample in samples:

aikido_zen/vulnerabilities/attack_wave_detection/attack_wave_detector.py

@@ -52,12 +52,23 @@ def is_attack_wave(self, context: Context) -> bool:
         self.suspicious_requests_map.set(ip, suspicious_requests)
 
         samples = self.samples_map.get(ip) or []
-        samples.append(
-            {
-                "method": context.method,
-                "url": context.url,
-            }
-        )
+
+        # There's no use in reporting a sample twice.
+        sample_exists = False
+        for existing_sample in samples:
+            if (
+                existing_sample["method"] == context.method
+                and existing_sample["url"] == context.url
+            ):
+                sample_exists = True
+                break
+        if not sample_exists:
+            samples.append(
+                {
+                    "method": context.method,
+                    "url": context.url,
+                }
+            )
 
         # Keep only the most recent samples (limit to avoid memory issues)
         if len(samples) > 10:

aikido_zen/vulnerabilities/attack_wave_detection/attack_wave_detector_test.py

@@ -136,11 +136,11 @@ def test_samples_tracking():
         for i in range(3):
             detector.is_attack_wave(context)
 
-        # Check that samples are being tracked
+        # Check that samples are being tracked (should have only 1 unique sample)
         samples = detector.get_samples_for_ip(context.remote_address)
-        assert len(samples) == 3
-        assert all(sample["method"] == "POST" for sample in samples)
-        assert all(sample["url"] == "http://localhost:8080/" for sample in samples)
+        assert len(samples) == 1  # Only 1 unique sample despite 3 identical requests
+        assert samples[0]["method"] == "POST"
+        assert samples[0]["url"] == "http://localhost:8080/"
 
         # Make more requests to exceed the sample limit
         for i in range(10):
@@ -164,9 +164,9 @@ def test_clear_samples():
         for i in range(5):
             detector.is_attack_wave(context)
 
-        # Verify samples exist
+        # Verify samples exist (should have only 1 unique sample)
         samples = detector.get_samples_for_ip(context.remote_address)
-        assert len(samples) == 5
+        assert len(samples) == 1  # Only 1 unique sample despite 5 identical requests
 
         # Clear samples
         detector.clear_samples_for_ip(context.remote_address)