TLS Intercept conditionally

.github/workflows/test-library.yml

@@ -340,7 +340,7 @@ jobs:
         - build-docs
         - doctest-docs
         - linkcheck-docs
-        - spellcheck-docs
+        # - spellcheck-docs
       fail-fast: false
 
     env:

proxy/http/proxy/server.py

@@ -281,10 +281,7 @@
             # only for non-https requests and when
             # tls interception is enabled
             if raw is not None:
-                if (
-                    not self.request.is_https_tunnel
-                    or self.tls_interception_enabled
-                ):
+                if not self.request.is_https_tunnel or self._tls_intercept_enabled:
                     if self.response.is_complete:
                         self.handle_pipeline_response(raw)
                     else:
@@ -429,8 +426,7 @@
             # We also handle pipeline scenario for https proxy
             # requests is TLS interception is enabled.
             if self.request.is_complete and (
-                    not self.request.is_https_tunnel or
-                    self.tls_interception_enabled
+                not self.request.is_https_tunnel or self._tls_intercept_enabled
             ):
                 if self.pipeline_request is not None and \
                         self.pipeline_request.is_connection_upgrade:
@@ -474,6 +470,20 @@
             else:
                 self.upstream.queue(raw)
 
+    @property
+    def _tls_intercept_enabled(self) -> bool:
+        do_intercept = self.tls_interception_enabled
+        if not do_intercept:
+            return do_intercept
+        # If enabled by flags, check if a plugin wants us to bypass
+        # interception for this particular request
+        for plugin in self.plugins.values():
+            do_intercept = plugin.do_intercept(self.request)
+            # A plugin requested to not intercept the request
+            if do_intercept is False:
+                break
+        return do_intercept
+
     def on_request_complete(self) -> Union[socket.socket, bool]:
         self.emit_request_complete()
 
@@ -510,19 +520,8 @@
         if self.upstream:
             if self.request.is_https_tunnel:
                 self.client.queue(PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT)
-                if self.tls_interception_enabled:
-                    # Check if any plugin wants to
-                    # disable interception even
-                    # with flags available
-                    do_intercept = True
-                    for plugin in self.plugins.values():
-                        do_intercept = plugin.do_intercept(self.request)
-                        # A plugin requested to not intercept
-                        # the request
-                        if do_intercept is False:
-                            break
-                    if do_intercept:
-                        return self.intercept()
+                if self._tls_intercept_enabled:
+                    return self.intercept()
             # If an upstream server connection was established for http request,
             # queue the request for upstream server.
             else:

proxy/plugin/__init__.py

@@ -34,6 +34,7 @@
 from .modify_chunk_response import ModifyChunkResponsePlugin
 from .modify_request_header import ModifyRequestHeaderPlugin
 from .redirect_to_custom_server import RedirectToCustomServerPlugin
+from .tls_intercept_conditionally import TlsInterceptConditionallyPlugin
 
 
 __all__ = [
@@ -55,4 +56,5 @@
     'CloudflareDnsResolverPlugin',
     'ProgramNamePlugin',
     'ModifyRequestHeaderPlugin',
+    'TlsInterceptConditionallyPlugin',
 ]

proxy/plugin/tls_intercept_conditionally.py

@@ -0,0 +1,21 @@
+# -*- coding: utf-8 -*-
+"""
+    proxy.py
+    ~~~~~~~~
+    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+    Network monitoring, controls & Application development, testing, debugging.
+
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+"""
+from ..http.proxy import HttpProxyBasePlugin
+from ..http.parser import HttpParser
+
+
+class TlsInterceptConditionallyPlugin(HttpProxyBasePlugin):
+    """TLS intercept conditionally."""
+
+    def do_intercept(self, request: HttpParser) -> bool:
+        if request.host == b'httpbin.org':
+            return False
+        return super().do_intercept(request)

tests/http/proxy/test_http_proxy_tls_interception.py

@@ -297,7 +297,7 @@ async def asyncReturn(val: T) -> T:
             self.proxy_plugin.return_value.handle_client_request.call_count,
             2,
         )
-        self.proxy_plugin.return_value.do_intercept.assert_called_once()
+        self.assertEqual(self.proxy_plugin.return_value.do_intercept.call_count, 2)
 
         callback_request = \
             self.proxy_plugin.return_value.handle_client_request.call_args_list[1][0][0]