Merge pull request #2452 from mattweidner/add_intercept_toggle

Add intercept toggle feature.

mitmproxy/addons/intercept.py

@@ -11,12 +11,14 @@ class Intercept:
         if "intercept" in updated:
             if not ctx.options.intercept:
                 self.filt = None
+                ctx.options.intercept_active = False
                 return
             self.filt = flowfilter.parse(ctx.options.intercept)
             if not self.filt:
                 raise exceptions.OptionsError(
                     "Invalid interception filter: %s" % ctx.options.intercept
                 )
+            ctx.options.intercept_active = True
 
     def process_flow(self, f):
         if self.filt:
@@ -24,7 +26,7 @@ class Intercept:
                 self.filt(f),
                 not f.request.is_replay,
             ])
-            if should_intercept:
+            if should_intercept and ctx.options.intercept_active:
                 f.intercept()
 
     # Handlers

mitmproxy/options.py

@@ -385,6 +385,11 @@ class Options(optmanager.OptManager):
             """
         )
 
+        self.add_option(
+            "intercept_active", bool, False,
+            "Intercept toggle"
+        )
+
         self.add_option(
             "intercept", Optional[str], None,
             "Intercept filter expression."

mitmproxy/tools/console/consoleaddons.py

@@ -68,6 +68,13 @@ class ConsoleAddon:
         """
         return ["single", "vertical", "horizontal"]
 
+    @command.command("intercept_toggle")
+    def intercept_toggle(self) -> None:
+        """
+            Toggles interception on/off leaving intercept filters intact.
+        """
+        ctx.options.intercept_active = not ctx.options.intercept_active
+
     @command.command("console.layout.cycle")
     def layout_cycle(self) -> None:
         """

mitmproxy/tools/console/defaultkeys.py

@@ -24,6 +24,7 @@ def map(km):
     km.add("ctrl f", "console.nav.pagedown", ["global"], "Page down")
     km.add("ctrl b", "console.nav.pageup", ["global"], "Page up")
 
+    km.add("I", "console.command intercept_toggle", ["global"], "Toggle intercept")
     km.add("i", "console.command set intercept=", ["global"], "Set intercept")
     km.add("W", "console.command set save_stream_file=", ["global"], "Stream to file")
     km.add("A", "flow.resume @all", ["flowlist", "flowview"], "Resume all intercepted flows")

mitmproxy/tools/console/statusbar.py

@@ -179,6 +179,8 @@ class StatusBar(urwid.WidgetWrap):
             r.append("CP:%d]" % len(self.master.options.tcp_hosts))
         if self.master.options.intercept:
             r.append("[")
+            if not self.master.options.intercept_active:
+                r.append("X")
             r.append(("heading_key", "i"))
             r.append(":%s]" % self.master.options.intercept)
         if self.master.options.view_filter:

mitmproxy/tools/web/app.py

@@ -410,6 +410,7 @@ class Settings(RequestHandler):
         self.write(dict(
             version=version.VERSION,
             mode=str(self.master.options.mode),
+            intercept_active=self.master.options.intercept_active,
             intercept=self.master.options.intercept,
             showhost=self.master.options.showhost,
             upstream_cert=self.master.options.upstream_cert,

test/mitmproxy/addons/test_intercept.py

@@ -13,10 +13,12 @@ def test_simple():
         assert not r.filt
         tctx.configure(r, intercept="~q")
         assert r.filt
+        assert tctx.options.intercept_active
         with pytest.raises(exceptions.OptionsError):
             tctx.configure(r, intercept="~~")
         tctx.configure(r, intercept=None)
         assert not r.filt
+        assert not tctx.options.intercept_active
 
         tctx.configure(r, intercept="~s")
 
@@ -31,3 +33,13 @@ def test_simple():
         f = tflow.tflow(resp=True)
         r.response(f)
         assert f.intercepted
+
+        tctx.configure(r, intercept_active=False)
+        f = tflow.tflow(resp=True)
+        tctx.cycle(r, f)
+        assert not f.intercepted
+
+        tctx.configure(r, intercept_active=True)
+        f = tflow.tflow(resp=True)
+        tctx.cycle(r, f)
+        assert f.intercepted