Support unary flow command arguments

Use this to add a console.view.flow command and bind "enter" in flowlist.

mitmproxy/command.py

@@ -15,8 +15,10 @@ def typename(t: type, ret: bool) -> str:
     """
     if t in (str, int, bool):
         return t.__name__
-    if t == typing.Sequence[flow.Flow]:
+    elif t == typing.Sequence[flow.Flow]:
         return "[flow]" if ret else "flowspec"
+    elif t == flow.Flow:
+        return "flow"
     else:  # pragma: no cover
         raise NotImplementedError(t)
 
@@ -101,5 +103,12 @@ def parsearg(manager: CommandManager, spec: str, argtype: type) -> typing.Any:
         return spec
     elif argtype == typing.Sequence[flow.Flow]:
         return manager.call_args("console.resolve", [spec])
+    elif argtype == flow.Flow:
+        flows = manager.call_args("console.resolve", [spec])
+        if len(flows) != 1:
+            raise exceptions.CommandError(
+                "Command requires one flow, specification matched %s." % len(flows)
+            )
+        return flows[0]
     else:
         raise exceptions.CommandError("Unsupported argument type: %s" % argtype)

mitmproxy/tools/console/flowlist.py

@@ -133,7 +133,6 @@ class FlowItem(urwid.WidgetWrap):
     def selectable(self):
         return True
 
-
     def server_replay_prompt(self, k):
         a = self.master.addons.get("serverplayback")
         if k == "a":
@@ -209,9 +208,6 @@ class FlowItem(urwid.WidgetWrap):
             if self.flow.killable:
                 self.flow.kill()
                 self.master.view.update(self.flow)
-        elif key == "enter":
-            if self.flow.request:
-                self.master.view_flow(self.flow)
         elif key == "|":
             signals.status_prompt_path.send(
                 prompt = "Send flow to script",

mitmproxy/tools/console/master.py

@@ -17,6 +17,7 @@ from mitmproxy import exceptions
 from mitmproxy import master
 from mitmproxy import io
 from mitmproxy import log
+from mitmproxy import flow
 from mitmproxy.addons import intercept
 from mitmproxy.addons import readfile
 from mitmproxy.addons import view
@@ -102,6 +103,12 @@ class ConsoleCommands:
         """View help."""
         self.master.view_help()
 
+    def view_flow(self, flow: flow.Flow) -> None:
+        """View a flow."""
+        if hasattr(flow, "request"):
+            # FIME: Also set focus?
+            self.master.view_flow(flow)
+
     def exit(self) -> None:
         """Exit mitmproxy."""
         raise urwid.ExitMainLoop
@@ -120,6 +127,7 @@ class ConsoleCommands:
         l.add_command("console.view.help", self.view_help)
         l.add_command("console.view.options", self.view_options)
         l.add_command("console.view.pop", self.view_pop)
+        l.add_command("console.view.flow", self.view_flow)
 
     def running(self):
         self.started = True
@@ -145,6 +153,7 @@ def default_keymap(km):
     km.add("f", "console.command 'set view_filter='", context="flowlist")
     km.add("e", "set console_eventlog=toggle", context="flowlist")
     km.add("w", "console.command 'save.file @shown '", context="flowlist")
+    km.add("enter", "console.view.flow @focus", context="flowlist")
 
 
 class ConsoleMaster(master.Master):

test/mitmproxy/test_command.py

@@ -61,6 +61,7 @@ def test_typename():
     assert command.typename(str, True) == "str"
     assert command.typename(typing.Sequence[flow.Flow], True) == "[flow]"
     assert command.typename(typing.Sequence[flow.Flow], False) == "flowspec"
+    assert command.typename(flow.Flow, False) == "flow"
 
 
 class DummyConsole:
@@ -68,7 +69,8 @@ class DummyConsole:
         l.add_command("console.resolve", self.resolve)
 
     def resolve(self, spec: str) -> typing.Sequence[flow.Flow]:
-        return [tflow.tflow(resp=True)]
+        n = int(spec)
+        return [tflow.tflow(resp=True)] * n
 
 
 def test_parsearg():
@@ -76,7 +78,12 @@ def test_parsearg():
         tctx.master.addons.add(DummyConsole())
         assert command.parsearg(tctx.master.commands, "foo", str) == "foo"
         assert len(command.parsearg(
-            tctx.master.commands, "~b", typing.Sequence[flow.Flow]
-        )) == 1
+            tctx.master.commands, "2", typing.Sequence[flow.Flow]
+        )) == 2
+        assert command.parsearg(tctx.master.commands, "1", flow.Flow)
+        with pytest.raises(exceptions.CommandError):
+            command.parsearg(tctx.master.commands, "2", flow.Flow)
+        with pytest.raises(exceptions.CommandError):
+            command.parsearg(tctx.master.commands, "0", flow.Flow)
         with pytest.raises(exceptions.CommandError):
             command.parsearg(tctx.master.commands, "foo", Exception)