perf: do not search and dlopen on every sql cmd

This extracts a handle of zen-internal FFI to a thread safe singleton.
Ensuring the DLL is searched for and loaded only once. Previously, we
would run multiple syscalls to get the location of the dll/so and then
built the python datastructures (CDLL object) that has to be later GCed.

I haven't microbenchmarked the improvements rigorously, as I feel GC
savings may contribute considerably in real-world scenario.

Although,

    poetry run pytest -rP  ./aikido_zen/vulnerabilities/sql_injection/

improves by over 4.3% on my set-up.

Author: isimluk

aikido_zen/vulnerabilities/sql_injection/__init__.py

@@ -3,11 +3,8 @@
 """
 
 import re
-import ctypes
 from aikido_zen.helpers.logging import logger
-from .map_dialect_to_rust_int import map_dialect_to_rust_int
-from .get_lib_path import get_binary_path
-from ...helpers.encode_safely import encode_safely
+from .zen_internal_ffi import ZenInternal
 
 
 def detect_sql_injection(query, user_input, dialect):
@@ -20,32 +17,7 @@ def detect_sql_injection(query, user_input, dialect):
         if should_return_early(query_l, userinput_l):
             return False
 
-        internals_lib = ctypes.CDLL(get_binary_path())
-        internals_lib.detect_sql_injection.argtypes = [
-            ctypes.POINTER(ctypes.c_uint8),
-            ctypes.c_size_t,
-            ctypes.POINTER(ctypes.c_uint8),
-            ctypes.c_size_t,
-            ctypes.c_int,
-        ]
-        internals_lib.detect_sql_injection.restype = ctypes.c_int
-
-        # Parse input variables for rust function
-        query_bytes = encode_safely(query_l)
-        userinput_bytes = encode_safely(userinput_l)
-        query_buffer = (ctypes.c_uint8 * len(query_bytes)).from_buffer_copy(query_bytes)
-        userinput_buffer = (ctypes.c_uint8 * len(userinput_bytes)).from_buffer_copy(
-            userinput_bytes
-        )
-        dialect_int = map_dialect_to_rust_int(dialect)
-
-        c_int_res = internals_lib.detect_sql_injection(
-            query_buffer,
-            len(query_bytes),
-            userinput_buffer,
-            len(userinput_bytes),
-            dialect_int,
-        )
+        c_int_res = ZenInternal().detect_sql_injection(query_l, userinput_l, dialect)
 
         # This means that an error occurred in the library
         if c_int_res == 2:

aikido_zen/vulnerabilities/sql_injection/zen_internal_ffi.py

@@ -0,0 +1,49 @@
+"""
+Interface for calling zen-internal shared library
+"""
+
+import ctypes
+import threading
+from .get_lib_path import get_binary_path
+from .map_dialect_to_rust_int import map_dialect_to_rust_int
+from ...helpers.encode_safely import encode_safely
+
+
+class __Singleton(type):
+    _instances = {}
+    _lock = threading.Lock()  # Ensures thread safety
+
+    def __call__(cls, *args, **kwargs):
+        with cls._lock:  # Lock to make the check-and-create operation atomic
+            if cls not in cls._instances:
+                cls._instances[cls] = super().__call__(*args, **kwargs)
+        return cls._instances[cls]
+
+
+class ZenInternal(metaclass=__Singleton):
+    def __init__(self):
+        self._lib = ctypes.CDLL(get_binary_path())
+        self._lib.detect_sql_injection.argtypes = [
+            ctypes.POINTER(ctypes.c_uint8),
+            ctypes.c_size_t,
+            ctypes.POINTER(ctypes.c_uint8),
+            ctypes.c_size_t,
+            ctypes.c_int,
+        ]
+        self._lib.detect_sql_injection.restype = ctypes.c_int
+
+    def detect_sql_injection(self, query, user_input, dialect):
+        query_bytes = encode_safely(query)
+        userinput_bytes = encode_safely(user_input)
+        query_buffer = (ctypes.c_uint8 * len(query_bytes)).from_buffer_copy(query_bytes)
+        userinput_buffer = (ctypes.c_uint8 * len(userinput_bytes)).from_buffer_copy(
+            userinput_bytes
+        )
+        dialect_int = map_dialect_to_rust_int(dialect)
+        return self._lib.detect_sql_injection(
+            query_buffer,
+            len(query_bytes),
+            userinput_buffer,
+            len(userinput_bytes),
+            dialect_int,
+        )