create a unified should_skip_attack_scan

Author: bitterpanda63

…lpers/is_protection_forced_off_cached.py …o_zen/helpers/should_skip_attack_scan.pyaikido_zen/helpers/is_protection_forced_off_cached.py renamed to aikido_zen/helpers/should_skip_attack_scan.py aikido_zen/helpers/is_protection_forced_off_cached.py renamed to aikido_zen/helpers/should_skip_attack_scan.py

@@ -1,11 +1,12 @@
 from aikido_zen.thread.thread_cache import get_cache
 from aikido_zen.helpers.protection_forced_off import protection_forced_off
+from aikido_zen.helpers.logging import logger
 from aikido_zen.context import Context
 
 
-def is_protection_forced_off_cached(context: Context) -> bool:
+def should_skip_attack_scan(context: Context) -> bool:
     """
-    Check if protection is forced off using cached endpoints.
+    Check if protection is forced off or IP bypassed using cache stored in the context.
     This function assumes that the thread cache has already been retrieved
     and uses it to determine if protection is forced off for the given context.
     """
@@ -21,9 +22,16 @@ def is_protection_forced_off_cached(context: Context) -> bool:
     if not thread_cache:
         return False
 
-    is_forced_off = protection_forced_off(
+    is_forced_off = False
+    # We check for a boolean protectionForcedOff on the matching endpoints, allows users to disable scans on certain routes.
+    if protection_forced_off(
         context.get_route_metadata(), thread_cache.get_endpoints()
-    )
+    ):
+        is_forced_off = True
+    # We check for Bypassed IPs : Allows users to let their DAST not be blocked by Zen
+    if thread_cache.is_bypassed_ip(context.remote_address):
+        is_forced_off = True
+
     context.set_force_protection_off(is_forced_off)
     context.set_as_current_context()
 

aikido_zen/vulnerabilities/__init__.py

@@ -16,8 +16,8 @@
 from aikido_zen.helpers.logging import logger
 from aikido_zen.helpers.get_clean_stacktrace import get_clean_stacktrace
 from aikido_zen.helpers.blocking_enabled import is_blocking_enabled
-from aikido_zen.helpers.is_protection_forced_off_cached import (
-    is_protection_forced_off_cached,
+from aikido_zen.helpers.should_skip_attack_scan import (
+    should_skip_attack_scan,
 )
 from aikido_zen.thread.thread_cache import get_cache
 from .sql_injection.context_contains_sql_injection import context_contains_sql_injection
@@ -37,25 +37,12 @@ def run_vulnerability_scan(kind, op, args):
     raises error if blocking is enabled, communicates it with connection_manager
     """
     context = get_current_context()
-
-    if is_protection_forced_off_cached(context):
+    if should_skip_attack_scan(context) and kind != "ssrf":
+        # Make a special exception for SSRF:
+        # For stored ssrf we don't want to check bypassed IPs or protection forced off.
         return
 
     comms = comm.get_comms()
-    thread_cache = get_cache()
-    if not context and kind != "ssrf":
-        # Make a special exception for SSRF, which checks itself if context is set.
-        # This is because some scans/tests for SSRF do not require a context to be set.
-        return
-
-    if not thread_cache and kind != "ssrf":
-        # Make a special exception for SSRF, which checks itself if thread cache is set.
-        # This is because some scans/tests for SSRF do not require a thread cache to be set.
-        return
-    if thread_cache and context:
-        if thread_cache.is_bypassed_ip(context.remote_address):
-            #  This IP is on the bypass list, not scanning
-            return
 
     error_type = AikidoException  # Default error
     error_args = tuple()
@@ -87,6 +74,7 @@ def run_vulnerability_scan(kind, op, args):
             injection_results = inspect_getaddrinfo_result(dns_results, hostname, port)
             error_type = AikidoSSRF
 
+            thread_cache = get_cache()
             if thread_cache and port > 0:
                 thread_cache.hostnames.add(hostname, port)
         else:
@@ -101,7 +89,10 @@ def run_vulnerability_scan(kind, op, args):
 
         blocked = is_blocking_enabled()
         operation = injection_results["operation"]
-        thread_cache.stats.on_detected_attack(blocked, operation)
+
+        thread_cache = get_cache()
+        if thread_cache:
+            thread_cache.stats.on_detected_attack(blocked, operation)
 
         stack = get_clean_stacktrace()
 

aikido_zen/vulnerabilities/ssrf/inspect_getaddrinfo_result.py

@@ -13,6 +13,7 @@
 from .find_hostname_in_context import find_hostname_in_context
 from .extract_ip_array_from_results import extract_ip_array_from_results
 from .is_redirect_to_private_ip import is_redirect_to_private_ip
+from aikido_zen.helpers.should_skip_attack_scan import should_skip_attack_scan
 
 
 #  gets called when the result of the DNS resolution has come in
@@ -27,11 +28,7 @@ def inspect_getaddrinfo_result(dns_results, hostname, port):
         return
 
     context = get_current_context()
-    if not context:
-        return  # Context should be set to check user input.
-    if get_cache() and get_cache().is_bypassed_ip(context.remote_address):
-        # We check for bypassed ip's here since it is not checked for us
-        # in run_vulnerability_scan due to the exception for SSRF (see above code)
+    if should_skip_attack_scan(context):
         return
 
     # attack_findings is an object containing source, pathToPayload and payload.