[refac]refactor http worker to support different session (#467)

Tested with 
```
address = "https://127.0.0.1:8443"
        proxyConfig = ProxyConfig(NetworkProtocol.HTTP,
                                                               address,
                                                               authentication_mode="tls",
                                                               tls_ca_cert_path="x509_test_certs/root/certs/root_ca.crt", tls_client_cert_path="x509_test_certs/intermediate/certs/client.crt", tls_client_key_path="x509_test_certs/intermediate/private/client.key")
        statsig.initialize(secret_key, StatsigOptions(proxy_configs={NetworkEndpoint.DOWNLOAD_CONFIG_SPECS: proxyConfig}))
        user = StatsigUser('lrs_grpc_user')
        i = 0
        while i<2 :
            i = i+1
            config_names = await get_config_names()
            print("Config names to check")
            print(config_names)
            for gate in config_names["feature_gates"]:
                statsig.check_gate(user, gate)
                await asyncio.sleep(IDLE_INTERVAL / 1000)

```

against sfp enforce tls 

And verified it can work

Author: xinlili-statsig

statsig/http_worker.py

@@ -21,7 +21,7 @@
 from .sdk_configs import _SDK_Configs
 from .statsig_context import InitContext
 from .statsig_error_boundary import _StatsigErrorBoundary
-from .statsig_options import StatsigOptions, STATSIG_API, STATSIG_CDN, AuthenticationMode
+from .statsig_options import ProxyConfig, StatsigOptions, STATSIG_API, STATSIG_CDN, AuthenticationMode
 from .grpc_websocket_worker import load_credential_from_file
 
 REQUEST_TIMEOUT = 20
@@ -51,50 +51,8 @@ def __init__(
         self.__diagnostics = diagnostics
         self.__request_count = 0
         self.__temp_cert_files: List[str] = []
-        self.__request_session = self.__init_session(options)
-
-    def __init_session(self, options: StatsigOptions) -> requests.Session:
-        session = requests.Session()
-        http_proxy_config = None
-        for _, config in options.proxy_configs.items():
-            if config.protocol == NetworkProtocol.HTTP:
-                if config.authentication_mode in [AuthenticationMode.TLS, AuthenticationMode.MTLS]:
-                    http_proxy_config = config
-                    break
-        if http_proxy_config is None:
-            return session
-        try:
-            if http_proxy_config.authentication_mode == AuthenticationMode.TLS:
-                ca_cert = load_credential_from_file(http_proxy_config.tls_ca_cert_path, "TLS CA certificate")
-                if ca_cert:
-                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.pem') as ca_file:
-                        ca_file.write(ca_cert)
-                        session.verify = ca_file.name
-                        self.__temp_cert_files.append(ca_file.name)
-                    globals.logger.log_process("HTTP Worker", "Connecting using an TLS secure channel for HTTP")
-            elif http_proxy_config.authentication_mode == AuthenticationMode.MTLS:
-                client_cert = load_credential_from_file(http_proxy_config.tls_client_cert_path, "TLS client certificate")
-                client_key = load_credential_from_file(http_proxy_config.tls_client_key_path, "TLS client key")
-                ca_cert = load_credential_from_file(http_proxy_config.tls_ca_cert_path, "TLS CA certificate")
-                if client_cert and client_key and ca_cert:
-                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.pem') as cert_file:
-                        cert_file.write(client_cert)
-                        cert_path = cert_file.name
-                        self.__temp_cert_files.append(cert_path)
-                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.key') as key_file:
-                        key_file.write(client_key)
-                        key_path = key_file.name
-                        self.__temp_cert_files.append(key_path)
-                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.pem') as ca_file:
-                        ca_file.write(ca_cert)
-                        ca_path = ca_file.name
-                        self.__temp_cert_files.append(ca_path)
-                    session.cert = (cert_path, key_path)
-                    session.verify = ca_path
-                    globals.logger.log_process("HTTP Worker", "Connecting using an mTLS secure channel for HTTP")
-        except Exception as e:
-            self.__error_boundary.log_exception("http_worker:init_session", e)
-        return session
+        self.__statsig_request_session = requests.Session()
+        self.__request_session = requests.Session()
 
     def is_pull_worker(self) -> bool:
         return True
@@ -138,6 +96,7 @@ def get_dcs_fallback(
             init_timeout=init_timeout,
             log_on_exception=log_on_exception,
             tag="download_config_specs",
+            useStatsigClient = True,
         )
         self._context.source_api = STATSIG_CDN
         if response is not None and self._is_success_code(response.status_code):
@@ -175,6 +134,7 @@ def get_id_lists_fallback(
             log_on_exception=log_on_exception,
             init_timeout=init_timeout,
             tag="get_id_lists",
+            useStatsigClient = True,
         )
         if response is not None and self._is_success_code(response.status_code):
             return on_complete(response.data, None)
@@ -220,6 +180,39 @@ def shutdown(self) -> None:
                 pass
         self.__temp_cert_files.clear()
 
+    def authenticate_request_session(self, http_proxy_config: ProxyConfig):
+        try:
+            if http_proxy_config.authentication_mode == AuthenticationMode.TLS:
+                ca_cert = load_credential_from_file(http_proxy_config.tls_ca_cert_path, "TLS CA certificate")
+                if ca_cert:
+                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.pem') as ca_file:
+                        ca_file.write(ca_cert)
+                        self.__request_session.verify = ca_file.name
+                        self.__temp_cert_files.append(ca_file.name)
+                    globals.logger.log_process("HTTP Worker", "Connecting using an TLS secure channel for HTTP")
+            elif http_proxy_config.authentication_mode == AuthenticationMode.MTLS:
+                client_cert = load_credential_from_file(http_proxy_config.tls_client_cert_path, "TLS client certificate")
+                client_key = load_credential_from_file(http_proxy_config.tls_client_key_path, "TLS client key")
+                ca_cert = load_credential_from_file(http_proxy_config.tls_ca_cert_path, "TLS CA certificate")
+                if client_cert and client_key and ca_cert:
+                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.pem') as cert_file:
+                        cert_file.write(client_cert)
+                        cert_path = cert_file.name
+                        self.__temp_cert_files.append(cert_path)
+                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.key') as key_file:
+                        key_file.write(client_key)
+                        key_path = key_file.name
+                        self.__temp_cert_files.append(key_path)
+                    with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.pem') as ca_file:
+                        ca_file.write(ca_cert)
+                        ca_path = ca_file.name
+                        self.__temp_cert_files.append(ca_path)
+                    self.__request_session.cert = (cert_path, key_path)
+                    self.__request_session.verify = ca_path
+                    globals.logger.log_process("HTTP Worker", "Connecting using an mTLS secure channel for HTTP")
+        except Exception as e:
+            self.__error_boundary.log_exception("http_worker:init_session", e)
+
     def _run_task_for_initialize(
         self, task, timeout
     ) -> Tuple[Optional[Any], Optional[Exception]]:
@@ -239,9 +232,10 @@ def _post_request(
         init_timeout=None,
         zipped=None,
         tag=None,
+        useStatsigClient=False,
     ):
         return self._request(
-            "POST", url, headers, payload, log_on_exception, init_timeout, zipped, tag
+            "POST", url, headers, payload, log_on_exception, init_timeout, zipped, tag, useStatsigClient
         )
 
     def _get_request(
@@ -253,6 +247,7 @@ def _get_request(
         zipped=None,
         tag=None,
         get_text_value_only=False,
+        useStatsigClient=False,
     ):
         return self._request(
             "GET",
@@ -264,6 +259,7 @@ def _get_request(
             zipped,
             tag,
             get_text_value_only,
+            useStatsigClient
         )
 
     def _request(
@@ -277,6 +273,7 @@ def _request(
         zipped=False,
         tag=None,
         get_text_value_only=False,
+        useStatsigClient = False,
     ) -> RequestResult:
         if self.__local_mode:
             globals.logger.debug("Using local mode. Dropping network request")
@@ -312,6 +309,7 @@ def _request(
             timeout,
             init_timeout is not None,
             get_text_value_only,
+            useStatsigClient
         )
 
         if create_marker is not None:
@@ -333,10 +331,12 @@ def _run_request_with_strict_timeout(
         timeout,
         for_initialize=False,
         get_text_value_only=False,
+        useStatsigClient=False
     ) -> RequestResult:
         def request_task():
             try:
-                with self.__request_session.request(
+                request_session = self.__statsig_request_session if useStatsigClient else self.__request_session
+                with request_session.request(
                     method,
                     url,
                     data=payload,

statsig/statsig_network.py

@@ -1,6 +1,6 @@
 import importlib
 import threading
-from typing import Any, Callable, Optional
+from typing import Any, Callable, Optional, cast
 
 from . import globals
 from .diagnostics import Diagnostics
@@ -17,6 +17,7 @@
 from .statsig_error_boundary import _StatsigErrorBoundary
 from .statsig_options import (
     DEFAULT_RULESET_SYNC_INTERVAL,
+    AuthenticationMode,
     StatsigOptions,
     STATSIG_CDN,
     ProxyConfig,
@@ -100,11 +101,33 @@ def __init__(
                 self.load_grpc_worker(endpoint, config)
             elif protocol == NetworkProtocol.GRPC_WEBSOCKET:
                 self.load_grpc_websocket_worker(endpoint, config)
+            elif protocol == NetworkProtocol.HTTP:
+                if config.authentication_mode in [AuthenticationMode.TLS, AuthenticationMode.MTLS]:
+                    self.load_authenticated_http_worker(endpoint, config)
 
         self._background_download_configs_from_statsig = None
         self._background_download_id_lists_from_statsig = None
         self._streaming_fallback: Optional[StreamingFallback] = None
 
+    def load_authenticated_http_worker(self, endpoint: NetworkEndpoint, config: ProxyConfig):
+        if endpoint == NetworkEndpoint.ALL:
+            http_worker = cast(HttpWorker, self.http_worker)
+            http_worker.authenticate_request_session(config)
+            self.log_event_worker = http_worker
+            self.id_list_worker = http_worker
+            self.dcs_worker = http_worker
+            return
+
+        worker = HttpWorker(self.sdk_key, self.options, self.statsig_metadata, self.error_boundary, self.diagnostics, self.context)
+        worker.authenticate_request_session(config)
+        if endpoint == NetworkEndpoint.DOWNLOAD_CONFIG_SPECS:
+            self.dcs_worker = worker
+        elif endpoint == NetworkEndpoint.GET_ID_LISTS:
+            self.id_list_worker = worker
+        elif endpoint == NetworkEndpoint.LOG_EVENT:
+            self.log_event_worker = worker
+
+
     def load_grpc_websocket_worker(self, endpoint: NetworkEndpoint, config: ProxyConfig):
         grpc_worker_module = importlib.import_module("statsig.grpc_websocket_worker")
         grpc_webhook_worker_class = getattr(grpc_worker_module, 'GRPCWebsocketWorker')