Create a helper function is_blocking_enabled

aikido_firewall/helpers/blocking_enabled.py

@@ -0,0 +1,13 @@
+"""Helper function file, see function docstring"""
+
+from aikido_firewall.background_process import get_comms
+
+
+def is_blocking_enabled():
+    """
+    Checks with the background process if blocking is enabled
+    """
+    should_block_res = get_comms().send_data_to_bg_process(
+        action="READ_PROPERTY", obj="block", receive=True
+    )
+    return should_block_res["success"] and should_block_res["data"]

aikido_firewall/sinks/builtins.py

@@ -11,6 +11,7 @@
 from aikido_firewall.context import get_current_context
 from aikido_firewall.background_process import get_comms
 from aikido_firewall.errors import AikidoPathTraversal
+from aikido_firewall.helpers.blocking_enabled import is_blocking_enabled
 
 
 @importhook.on_import("builtins")
@@ -34,10 +35,7 @@
         )
         if len(result) != 0:
             get_comms().send_data_to_bg_process("ATTACK", (result, context))
-            should_block_res = get_comms().send_data_to_bg_process(
-                action="READ_PROPERTY", obj="block", receive=True
-            )
-            if should_block_res["success"] and should_block_res["data"]:
+            if is_blocking_enabled():
                 raise AikidoPathTraversal()
         return former_open(*args, **kwargs)
 

aikido_firewall/sinks/mysqlclient.py

@@ -14,6 +14,7 @@
 from aikido_firewall.helpers.logging import logger
 from aikido_firewall.background_process import get_comms
 from aikido_firewall.errors import AikidoSQLInjection
+from aikido_firewall.helpers.blocking_enabled import is_blocking_enabled
 
 
 @importhook.on_import("MySQLdb.connections")
@@ -39,10 +40,7 @@
         logger.debug("sql_injection results : %s", json.dumps(contains_injection))
         if contains_injection:
             get_comms().send_data_to_bg_process("ATTACK", (contains_injection, context))
-            should_block_res = get_comms().send_data_to_bg_process(
-                action="READ_PROPERTY", obj="block", receive=True
-            )
-            if should_block_res["success"] and should_block_res["data"]:
+            if is_blocking_enabled():
                 raise AikidoSQLInjection("SQL Injection [aikido_firewall]")
 
         return prev_query_function(_self, sql)

aikido_firewall/sinks/os.py

@@ -11,6 +11,7 @@
 from aikido_firewall.context import get_current_context
 from aikido_firewall.background_process import get_comms
 from aikido_firewall.errors import AikidoPathTraversal
+from aikido_firewall.helpers.blocking_enabled import is_blocking_enabled
 
 # File functions :
 OS_FILE_FUNCTIONS = [
@@ -59,10 +60,7 @@
         )
         if len(result) != 0:
             get_comms().send_data_to_bg_process("ATTACK", (result, context))
-            should_block_res = get_comms().send_data_to_bg_process(
-                action="READ_PROPERTY", obj="block", receive=True
-            )
-            if should_block_res["success"] and should_block_res["data"]:
+            if is_blocking_enabled():
                 raise AikidoPathTraversal()
         return former_func(*args, **kwargs)
 

aikido_firewall/sinks/os_system.py

@@ -12,6 +12,7 @@
 from aikido_firewall.helpers.logging import logger
 from aikido_firewall.background_process import get_comms
 from aikido_firewall.errors import AikidoShellInjection
+from aikido_firewall.helpers.blocking_enabled import is_blocking_enabled
 
 
 @importhook.on_import("os")
@@ -37,10 +38,7 @@
         logger.debug("Shell injection results : %s", json.dumps(contains_injection))
         if contains_injection:
             get_comms().send_data_to_bg_process("ATTACK", (contains_injection, context))
-            should_block_res = get_comms().send_data_to_bg_process(
-                action="READ_PROPERTY", obj="block", receive=True
-            )
-            if should_block_res["success"] and should_block_res["data"]:
+            if is_blocking_enabled():
                 raise AikidoShellInjection()
 
         return former_system_func(*args, **kwargs)

aikido_firewall/sinks/psycopg2.py

@@ -13,6 +13,7 @@
 from aikido_firewall.vulnerabilities.sql_injection.dialects import Postgres
 from aikido_firewall.background_process import get_comms
 from aikido_firewall.errors import AikidoSQLInjection
+from aikido_firewall.helpers.blocking_enabled import is_blocking_enabled
 
 
 class MutableAikidoConnection:
@@ -46,10 +47,7 @@
     logger.info("sql_injection results : %s", json.dumps(contains_injection))
     if contains_injection:
         get_comms().send_data_to_bg_process("ATTACK", (contains_injection, context))
-        should_block_res = get_comms().send_data_to_bg_process(
-            action="READ_PROPERTY", obj="block", receive=True
-        )
-        if should_block_res["success"] and should_block_res["data"]:
+        if is_blocking_enabled():
             raise AikidoSQLInjection("SQL Injection [aikido_firewall]")
 
 

aikido_firewall/sinks/pymongo.py

@@ -10,6 +10,7 @@
 from aikido_firewall.context import get_current_context
 from aikido_firewall.background_process import get_comms
 from aikido_firewall.errors import AikidoNoSQLInjection
+from aikido_firewall.helpers.blocking_enabled import is_blocking_enabled
 
 OPERATIONS_WITH_FILTER = [
     "replace_one",  # L1087
@@ -55,7 +56,8 @@
                 get_comms().send_data_to_bg_process(
                     "ATTACK", (injection_results, context)
                 )
-                raise AikidoNoSQLInjection("NOSQL Injection [aikido_firewall]")
+                if is_blocking_enabled():
+                    raise AikidoNoSQLInjection("NOSQL Injection [aikido_firewall]")
             return prev_func(_self, _filter, *args, **kwargs)
 
         setattr(modified_pymongo.Collection, operation, wrapped_operation_function)

aikido_firewall/sinks/pymysql.py

@@ -14,6 +14,7 @@
 from aikido_firewall.vulnerabilities.sql_injection.dialects import MySQL
 from aikido_firewall.background_process import get_comms
 from aikido_firewall.errors import AikidoSQLInjection
+from aikido_firewall.helpers.blocking_enabled import is_blocking_enabled
 
 logger = logging.getLogger("aikido_firewall")
 
@@ -41,10 +42,7 @@
         logger.info("sql_injection results : %s", json.dumps(contains_injection))
         if contains_injection:
             get_comms().send_data_to_bg_process("ATTACK", (contains_injection, context))
-            should_block_res = get_comms().send_data_to_bg_process(
-                action="READ_PROPERTY", obj="block", receive=True
-            )
-            if should_block_res["success"] and should_block_res["data"]:
+            if is_blocking_enabled():
                 raise AikidoSQLInjection("SQL Injection [aikido_firewall]")
 
         return prev_query_function(_self, sql, unbuffered=False)

aikido_firewall/sinks/subprocess.py

@@ -12,6 +12,7 @@
 from aikido_firewall.helpers.logging import logger
 from aikido_firewall.background_process import get_comms
 from aikido_firewall.errors import AikidoShellInjection
+from aikido_firewall.helpers.blocking_enabled import is_blocking_enabled
 
 SUBPROCESS_OPERATIONS = ["call", "run", "check_call", "Popen", "check_output"]
 
@@ -39,10 +40,7 @@
         logger.debug("Shell injection results : %s", json.dumps(contains_injection))
         if contains_injection:
             get_comms().send_data_to_bg_process("ATTACK", (contains_injection, context))
-            should_block_res = get_comms().send_data_to_bg_process(
-                action="READ_PROPERTY", obj="block", receive=True
-            )
-            if should_block_res["success"] and should_block_res["data"]:
+            if is_blocking_enabled():
                 raise AikidoShellInjection()
 
         return former_func(*args, **kwargs)

aikido_firewall/vulnerabilities/ssrf/inspect_getaddrinfo_result.py

@@ -8,6 +8,7 @@
 from aikido_firewall.helpers.logging import logger
 from aikido_firewall.background_process import get_comms
 from aikido_firewall.errors import AikidoSSRF
+from aikido_firewall.helpers.blocking_enabled import is_blocking_enabled
 from .imds import is_trusted_hostname, is_imds_ip_address
 from .is_private_ip import is_private_ip
 from .find_hostname_in_context import find_hostname_in_context
@@ -23,17 +24,12 @@
 
     context = get_current_context()
 
-    should_block_res = get_comms().send_data_to_bg_process(
-        action="READ_PROPERTY", obj="block", receive=True
-    )
-    should_block = should_block_res["success"] and should_block_res["data"]
-
     ip_addresses = extract_ip_array_from_results(dns_results)
     if resolves_to_imds_ip(ip_addresses, hostname):
         #  Block stored SSRF attack that target IMDS IP addresses
         #  An attacker could have stored a hostname in a database that points to an IMDS IP address
         #  We don't check if the user input contains the hostname because there's no context
-        if should_block:
+        if is_blocking_enabled():
             raise AikidoSSRF()
 
     if not context:
@@ -47,6 +43,7 @@
     if not found:
         return
 
+    should_block = is_blocking_enabled()
     stack = " ".join(traceback.format_stack())
     attack = {
         "module": "socket",