Avoid running out of FDs in tests on MacOS

Author: SaladDais

hippolyzer/apps/proxy.py

@@ -128,6 +128,7 @@ def start_proxy(session_manager: SessionManager, extra_addons: Optional[list] =
     if proxy_host is None:
         proxy_host = session_manager.settings.PROXY_BIND_ADDR
 
+    session_manager.flow_context = HTTPFlowContext()
     flow_context = session_manager.flow_context
     session_manager.name_cache.load_viewer_caches()
 

hippolyzer/lib/proxy/http_flow.py

@@ -140,7 +140,7 @@ def from_state(cls, flow_state: Dict, session_manager: Optional[SessionManager])
         assert flow is not None
         cap_data_ser = flow.metadata.get("cap_data_ser")
         callback_queue = None
-        if session_manager:
+        if session_manager and session_manager.flow_context:
             callback_queue = session_manager.flow_context.to_proxy_queue
         if cap_data_ser is not None:
             flow.metadata["cap_data"] = CapData.deserialize(cap_data_ser, session_manager)

hippolyzer/lib/proxy/http_proxy.py

@@ -102,6 +102,11 @@ def __init__(self):
         self.shutdown_signal = multiprocessing.Event()
         self.mitmproxy_ready = multiprocessing.Event()
 
+    def close(self):
+        self.shutdown_signal.set()
+        self.from_proxy_queue.close()
+        self.to_proxy_queue.close()
+
 
 class IPCInterceptionAddon:
     """

hippolyzer/lib/proxy/sessions.py

@@ -5,7 +5,6 @@
 import datetime
 import functools
 import logging
-import multiprocessing
 from typing import *
 from weakref import ref
 
@@ -121,14 +120,17 @@ def __init__(self, settings: ProxySettings):
         BaseClientSessionManager.__init__(self)
         self.settings: ProxySettings = settings
         self.sessions: List[Session] = []
-        self.shutdown_signal = multiprocessing.Event()
-        self.flow_context = HTTPFlowContext()
+        self.flow_context: HTTPFlowContext | None = None
         self.asset_repo = HTTPAssetRepo()
         self.message_logger: Optional[BaseMessageLogger] = None
         self.addon_ctx: Dict[str, Dict[str, Any]] = collections.defaultdict(dict)
         self.name_cache = ProxyNameCache()
         self.pending_leap_clients: List[LEAPClient] = []
 
+    def close(self):
+        if self.flow_context is not None:
+            self.flow_context.close()
+
     def create_session(self, login_data) -> Session:
         session = Session.from_login_data(login_data, self)
         self.name_cache.create_subscriptions(

hippolyzer/lib/proxy/test_utils.py

@@ -39,6 +39,7 @@ async def asyncSetUp(self) -> None:
 
     def tearDown(self) -> None:
         self.protocol.close()
+        self.session_manager.close()
 
     async def _wait_drained(self):
         await asyncio.sleep(0.001)

tests/proxy/integration/test_http.py

@@ -16,6 +16,7 @@
 from hippolyzer.lib.proxy.addon_utils import BaseAddon
 from hippolyzer.lib.proxy.addons import AddonManager
 from hippolyzer.lib.proxy.http_event_manager import MITMProxyEventManager
+from hippolyzer.lib.proxy.http_proxy import HTTPFlowContext
 from hippolyzer.lib.proxy.http_flow import HippoHTTPFlow
 from hippolyzer.lib.proxy.caps import SerializedCapData
 from hippolyzer.lib.proxy.sessions import SessionManager
@@ -35,6 +36,7 @@ async def asyncSetUp(self) -> None:
         await super().asyncSetUp()
         self.addon = MockAddon()
         AddonManager.init([], self.session_manager, [self.addon])
+        self.session_manager.flow_context = HTTPFlowContext()
         self.flow_context = self.session_manager.flow_context
         self.http_event_manager = MITMProxyEventManager(self.session_manager, self.flow_context)
         self._setup_default_circuit()
@@ -148,6 +150,7 @@ async def asyncSetUp(self) -> None:
         self.caps_client = self.session.main_region.caps_client
         proxy_port = 9905
         self.session_manager.settings.HTTP_PROXY_PORT = proxy_port
+        self.session_manager.flow_context = HTTPFlowContext()
 
         self.http_proc = multiprocessing.Process(
             target=run_http_proxy_process,
@@ -162,6 +165,15 @@ async def asyncSetUp(self) -> None:
             self.session_manager.flow_context
         )
 
+    def tearDown(self) -> None:
+        self.session_manager.flow_context.shutdown_signal.set()
+        self.http_proc.join(timeout=2.0)
+        if self.http_proc.is_alive():
+            self.http_proc.terminate()
+            self.http_proc.join(timeout=1.0)
+        self.http_proc.close()
+        super().tearDown()
+
     def test_mitmproxy_works(self):
         async def _request_example_com():
             # Pump callbacks from mitmproxy