convert to singleton pattern

Author: maelv-filigran

pycti/api/opencti_api_client.py

@@ -8,6 +8,7 @@
 import shutil
 import signal
 import tempfile
+import threading
 from typing import Dict, Tuple, Union
 
 import magic
@@ -83,6 +84,12 @@
 from pycti.utils.opencti_stix2 import OpenCTIStix2
 from pycti.utils.opencti_stix2_utils import OpenCTIStix2Utils
 
+# Global singleton variables for proxy certificate management
+_PROXY_CERT_BUNDLE = None
+_PROXY_CERT_DIR = None
+_PROXY_CERT_LOCK = threading.Lock()
+_PROXY_SIGNAL_HANDLERS_REGISTERED = False
+
 
 def build_request_headers(token: str, custom_headers: str, app_logger):
     headers_dict = {
@@ -171,18 +178,9 @@ def __init__(
         self.app_logger = self.logger_class("api")
         self.admin_logger = self.logger_class("admin")
 
-        # Initialize temp certificate directory tracker
-        self.temp_cert_dir = None
-
         # Setup proxy certificates if provided
         self._setup_proxy_certificates()
 
-        # Register cleanup handlers for temp certificates
-        if self.temp_cert_dir:
-            atexit.register(self._cleanup_temp_certificates)
-            signal.signal(signal.SIGTERM, self._signal_handler)
-            signal.signal(signal.SIGINT, self._signal_handler)
-
         # Define API
         self.api_token = token
         self.api_url = url + "/graphql"
@@ -272,74 +270,101 @@ def _setup_proxy_certificates(self):
         Detects HTTPS_CA_CERTIFICATES environment variable and combines
         proxy certificates with system certificates for SSL verification.
         Supports both inline certificate content and file paths.
+
+        Uses a singleton pattern to ensure only one certificate bundle is created
+        across all instances, avoiding resource leaks and conflicts.
         """
+        global _PROXY_CERT_BUNDLE, _PROXY_CERT_DIR, _PROXY_SIGNAL_HANDLERS_REGISTERED
+
         https_ca_certificates = os.getenv("HTTPS_CA_CERTIFICATES")
         if not https_ca_certificates:
             return
 
-        try:
-            # Create secure temporary directory
-            cert_dir = tempfile.mkdtemp(prefix="opencti_proxy_certs_")
-            self.temp_cert_dir = cert_dir
-
-            # Determine if HTTPS_CA_CERTIFICATES contains inline content or file path
-            cert_content = self._get_certificate_content(https_ca_certificates)
-            if not cert_content:
-                self.app_logger.warning(
-                    "Invalid HTTPS_CA_CERTIFICATES: not a valid certificate or file path",
-                    {
-                        "value": (
-                            https_ca_certificates[:50] + "..."
-                            if len(https_ca_certificates) > 50
-                            else https_ca_certificates
-                        )
-                    },
+        # Thread-safe check and setup
+        with _PROXY_CERT_LOCK:
+            # If already configured, reuse existing bundle
+            if _PROXY_CERT_BUNDLE is not None:
+                self.ssl_verify = _PROXY_CERT_BUNDLE
+                self.app_logger.debug(
+                    "Reusing existing proxy certificate bundle",
+                    {"cert_bundle": _PROXY_CERT_BUNDLE},
                 )
                 return
 
-            # Write proxy certificate to temp file
-            proxy_cert_file = os.path.join(cert_dir, "proxy-ca.crt")
-            with open(proxy_cert_file, "w") as f:
-                f.write(cert_content)
-
-            # Find system certificates
-            system_cert_paths = [
-                "/etc/ssl/certs/ca-certificates.crt",  # Debian/Ubuntu
-                "/etc/pki/tls/certs/ca-bundle.crt",  # RHEL/CentOS
-                "/etc/ssl/cert.pem",  # Alpine/BSD
-            ]
-
-            # Create combined certificate bundle
-            combined_cert_file = os.path.join(cert_dir, "combined-ca-bundle.crt")
-            with open(combined_cert_file, "w") as combined:
-                # Add system certificates first
-                for system_path in system_cert_paths:
-                    if os.path.exists(system_path):
-                        with open(system_path, "r") as sys_certs:
-                            combined.write(sys_certs.read())
-                            combined.write("\n")
-                        break
-
-                # Add proxy certificate
-                combined.write(cert_content)
-
-            # Update ssl_verify to use combined certificate bundle
-            self.ssl_verify = combined_cert_file
-
-            # Set environment variables for urllib and other libraries
-            os.environ["REQUESTS_CA_BUNDLE"] = combined_cert_file
-            os.environ["SSL_CERT_FILE"] = combined_cert_file
-
-            self.app_logger.info(
-                "Proxy certificates configured",
-                {"cert_bundle": combined_cert_file},
-            )
+            # First initialization - create the certificate bundle
+            try:
+                # Create secure temporary directory
+                cert_dir = tempfile.mkdtemp(prefix="opencti_proxy_certs_")
+
+                # Determine if HTTPS_CA_CERTIFICATES contains inline content or file path
+                cert_content = self._get_certificate_content(https_ca_certificates)
+                if not cert_content:
+                    self.app_logger.warning(
+                        "Invalid HTTPS_CA_CERTIFICATES: not a valid certificate or file path",
+                        {
+                            "value": (
+                                https_ca_certificates[:50] + "..."
+                                if len(https_ca_certificates) > 50
+                                else https_ca_certificates
+                            )
+                        },
+                    )
+                    return
+
+                # Write proxy certificate to temp file
+                proxy_cert_file = os.path.join(cert_dir, "proxy-ca.crt")
+                with open(proxy_cert_file, "w") as f:
+                    f.write(cert_content)
+
+                # Find system certificates
+                system_cert_paths = [
+                    "/etc/ssl/certs/ca-certificates.crt",  # Debian/Ubuntu
+                    "/etc/pki/tls/certs/ca-bundle.crt",  # RHEL/CentOS
+                    "/etc/ssl/cert.pem",  # Alpine/BSD
+                ]
+
+                # Create combined certificate bundle
+                combined_cert_file = os.path.join(cert_dir, "combined-ca-bundle.crt")
+                with open(combined_cert_file, "w") as combined:
+                    # Add system certificates first
+                    for system_path in system_cert_paths:
+                        if os.path.exists(system_path):
+                            with open(system_path, "r") as sys_certs:
+                                combined.write(sys_certs.read())
+                                combined.write("\n")
+                            break
+
+                    # Add proxy certificate
+                    combined.write(cert_content)
+
+                # Update global singleton variables
+                _PROXY_CERT_BUNDLE = combined_cert_file
+                _PROXY_CERT_DIR = cert_dir
+                self.ssl_verify = combined_cert_file
+
+                # Set environment variables for urllib and other libraries
+                os.environ["REQUESTS_CA_BUNDLE"] = combined_cert_file
+                os.environ["SSL_CERT_FILE"] = combined_cert_file
+
+                # Register cleanup handlers only once
+                atexit.register(_cleanup_proxy_certificates)
+
+                # Register signal handlers only once
+                if not _PROXY_SIGNAL_HANDLERS_REGISTERED:
+                    signal.signal(signal.SIGTERM, _signal_handler_proxy_cleanup)
+                    signal.signal(signal.SIGINT, _signal_handler_proxy_cleanup)
+                    _PROXY_SIGNAL_HANDLERS_REGISTERED = True
+
+                self.app_logger.info(
+                    "Proxy certificates configured",
+                    {"cert_bundle": combined_cert_file},
+                )
 
-        except Exception as e:
-            self.app_logger.error(
-                "Failed to setup proxy certificates", {"error": str(e)}
-            )
-            raise
+            except Exception as e:
+                self.app_logger.error(
+                    "Failed to setup proxy certificates", {"error": str(e)}
+                )
+                raise
 
     def _get_certificate_content(self, https_ca_certificates):
         """Extract certificate content from environment variable.
@@ -353,7 +378,7 @@ def _get_certificate_content(self, https_ca_certificates):
         """
         # Strip whitespace once at the beginning
         stripped_https_ca_certificates = https_ca_certificates.strip()
-        
+
         # Check if it's inline certificate content (starts with PEM header)
         if stripped_https_ca_certificates.startswith("-----BEGIN CERTIFICATE-----"):
             self.app_logger.debug(
@@ -390,43 +415,6 @@ def _get_certificate_content(self, https_ca_certificates):
         # Neither inline content nor valid file path
         return None
 
-    def _cleanup_temp_certificates(self):
-        """Clean up temporary certificate directory.
-        
-        This method is called on normal program exit via atexit
-        or when receiving termination signals (SIGTERM/SIGINT).
-        """
-        if self.temp_cert_dir and os.path.exists(self.temp_cert_dir):
-            try:
-                shutil.rmtree(self.temp_cert_dir)
-                self.app_logger.debug(
-                    "Cleaned up temporary certificates",
-                    {"cert_dir": self.temp_cert_dir}
-                )
-            except Exception as e:
-                self.app_logger.warning(
-                    "Failed to cleanup temporary certificates",
-                    {"cert_dir": self.temp_cert_dir, "error": str(e)}
-                )
-            finally:
-                self.temp_cert_dir = None
-
-    def _signal_handler(self, signum, frame):
-        """Handle termination signals (SIGTERM/SIGINT).
-        
-        Performs cleanup and then raises SystemExit to allow
-        normal shutdown procedures to complete.
-        
-        :param signum: Signal number
-        :param frame: Current stack frame
-        """
-        self.app_logger.info(
-            "Received termination signal, cleaning up",
-            {"signal": signum}
-        )
-        self._cleanup_temp_certificates()
-        raise SystemExit(0)
-
     def set_applicant_id_header(self, applicant_id):
         self.request_headers["opencti-applicant-id"] = applicant_id
 
@@ -1062,3 +1050,33 @@ def get_attribute_in_mitre_extension(key, object) -> any:
                 "extension-definition--322b8f77-262a-4cb8-a915-1e441e00329b"
             ][key]
         return None
+
+
+# Global cleanup functions for proxy certificates singleton
+def _cleanup_proxy_certificates():
+    """Clean up temporary certificate directory for proxy certificates.
+
+    This function is called on normal program exit via atexit.
+    """
+    global _PROXY_CERT_DIR
+    if _PROXY_CERT_DIR and os.path.exists(_PROXY_CERT_DIR):
+        try:
+            shutil.rmtree(_PROXY_CERT_DIR)
+        except Exception:
+            # Silently fail cleanup - best effort
+            pass
+        finally:
+            _PROXY_CERT_DIR = None
+
+
+def _signal_handler_proxy_cleanup(signum, frame):
+    """Handle termination signals (SIGTERM/SIGINT) for proxy certificate cleanup.
+
+    Performs cleanup and then raises SystemExit to allow
+    normal shutdown procedures to complete.
+
+    :param signum: Signal number
+    :param frame: Current stack frame
+    """
+    _cleanup_proxy_certificates()
+    raise SystemExit(0)