Merge pull request #1624 from cortesi/streambodies

Response body streaming to addon

mitmproxy/builtins/__init__.py

@@ -1,13 +1,14 @@
 from mitmproxy.builtins import anticache
 from mitmproxy.builtins import anticomp
+from mitmproxy.builtins import clientplayback
 from mitmproxy.builtins import filestreamer
-from mitmproxy.builtins import stickyauth
-from mitmproxy.builtins import stickycookie
-from mitmproxy.builtins import script
 from mitmproxy.builtins import replace
+from mitmproxy.builtins import script
 from mitmproxy.builtins import setheaders
 from mitmproxy.builtins import serverplayback
-from mitmproxy.builtins import clientplayback
+from mitmproxy.builtins import stickyauth
+from mitmproxy.builtins import stickycookie
+from mitmproxy.builtins import streambodies
 
 
 def default_addons():
@@ -18,6 +19,7 @@ def default_addons():
         stickycookie.StickyCookie(),
         script.ScriptLoader(),
         filestreamer.FileStreamer(),
+        streambodies.StreamBodies(),
         replace.Replace(),
         setheaders.SetHeaders(),
         serverplayback.ServerPlayback(),

mitmproxy/builtins/streambodies.py

@@ -0,0 +1,34 @@
+from netlib.http import http1
+from netlib import exceptions
+from mitmproxy import ctx
+
+
+class StreamBodies:
+    def __init__(self):
+        self.max_size = None
+
+    def configure(self, options, updated):
+        self.max_size = options.stream_large_bodies
+
+    def run(self, f, is_request):
+        if self.max_size:
+            r = f.request if is_request else f.response
+            try:
+                expected_size = http1.expected_http_body_size(
+                    f.request, f.response if not is_request else None
+                )
+            except exceptions.HTTPException:
+                f.reply.kill()
+                return
+            if expected_size and not r.raw_content and not (0 <= expected_size <= self.max_size):
+                # r.stream may already be a callable, which we want to preserve.
+                r.stream = r.stream or True
+                # FIXME: make message generic when we add rquest streaming
+                ctx.log.info("Streaming response from %s" % f.request.host)
+
+    # FIXME! Request streaming doesn't work at the moment.
+    def requestheaders(self, f):
+        self.run(f, True)
+
+    def responseheaders(self, f):
+        self.run(f, False)

mitmproxy/console/master.py

@@ -236,8 +236,6 @@ class ConsoleMaster(flow.FlowMaster):
         if options.filter:
             self.set_view_filter(options.filter)
 
-        self.set_stream_large_bodies(options.stream_large_bodies)
-
         self.palette = options.palette
         self.palette_transparent = options.palette_transparent
 

mitmproxy/console/statusbar.py

@@ -196,10 +196,10 @@ class StatusBar(urwid.WidgetWrap):
             opts.append("no-upstream-cert")
         if self.master.state.follow_focus:
             opts.append("following")
-        if self.master.stream_large_bodies:
+        if self.master.options.stream_large_bodies:
             opts.append(
                 "stream:%s" % human.pretty_size(
-                    self.master.stream_large_bodies.max_size
+                    self.master.options.stream_large_bodies
                 )
             )
 

mitmproxy/dump.py

@@ -40,7 +40,6 @@ class DumpMaster(flow.FlowMaster):
         self.addons.add(dumper.Dumper())
         # This line is just for type hinting
         self.options = self.options  # type: Options
-        self.set_stream_large_bodies(options.stream_large_bodies)
 
         if not self.options.no_server and server:
             self.add_log(

mitmproxy/flow/__init__.py

@@ -2,12 +2,12 @@ from mitmproxy.flow import export, modules
 from mitmproxy.flow.io import FlowWriter, FilteredFlowWriter, FlowReader, read_flows_from_paths
 from mitmproxy.flow.master import FlowMaster
 from mitmproxy.flow.modules import (
-    AppRegistry, StreamLargeBodies
+    AppRegistry
 )
 from mitmproxy.flow.state import State, DummyState, FlowView
 
 __all__ = [
     "export", "modules",
     "FlowWriter", "FilteredFlowWriter", "FlowReader", "read_flows_from_paths",
-    "FlowMaster", "AppRegistry", "StreamLargeBodies", "DummyState", "State", "FlowView",
+    "FlowMaster", "AppRegistry", "DummyState", "State", "FlowView",
 ]

mitmproxy/flow/master.py

@@ -3,7 +3,6 @@ import sys
 
 from typing import Optional  # noqa
 
-import netlib.exceptions
 from netlib import http
 from mitmproxy import controller
 from mitmproxy import exceptions
@@ -53,18 +52,11 @@ class FlowMaster(controller.Master):
         if server:
             self.add_server(server)
         self.state = state
-        self.stream_large_bodies = None  # type: Optional[modules.StreamLargeBodies]
         self.apps = modules.AppRegistry()
 
     def start_app(self, host, port):
         self.apps.add(app.mapp, host, port)
 
-    def set_stream_large_bodies(self, max_size):
-        if max_size is not None:
-            self.stream_large_bodies = modules.StreamLargeBodies(max_size)
-        else:
-            self.stream_large_bodies = False
-
     def duplicate_flow(self, f):
         """
             Duplicate flow, and insert it into state without triggering any of
@@ -238,12 +230,7 @@ class FlowMaster(controller.Master):
 
     @controller.handler
     def responseheaders(self, f):
-        try:
-            if self.stream_large_bodies:
-                self.stream_large_bodies.run(f, False)
-        except netlib.exceptions.HttpException:
-            f.reply.kill()
-            return
+        pass
 
     @controller.handler
     def response(self, f):

mitmproxy/flow/modules.py

@@ -1,6 +1,5 @@
 from netlib import wsgi
 from netlib import version
-from netlib.http import http1
 
 
 class AppRegistry:
@@ -28,17 +27,3 @@ class AppRegistry:
         if "host" in request.headers:
             host = request.headers["host"]
             return self.apps.get((host, request.port), None)
-
-
-class StreamLargeBodies:
-    def __init__(self, max_size):
-        self.max_size = max_size
-
-    def run(self, flow, is_request):
-        r = flow.request if is_request else flow.response
-        expected_size = http1.expected_http_body_size(
-            flow.request, flow.response if not is_request else None
-        )
-        if not r.raw_content and not (0 <= expected_size <= self.max_size):
-            # r.stream may already be a callable, which we want to preserve.
-            r.stream = r.stream or True

mitmproxy/models/http.py

@@ -53,6 +53,7 @@ class HTTPRequest(http.Request):
 
         # Is this request replayed?
         self.is_replay = is_replay
+        self.stream = None
 
     def get_state(self):
         state = super().get_state()
@@ -125,7 +126,7 @@ class HTTPResponse(http.Response):
 
         # Is this request replayed?
         self.is_replay = is_replay
-        self.stream = False
+        self.stream = None
 
     @classmethod
     def wrap(self, response):

test/mitmproxy/builtins/test_streambodies.py

@@ -0,0 +1,29 @@
+from .. import tutils, mastertest
+from mitmproxy.flow import state
+from mitmproxy.flow import master
+from mitmproxy import options
+
+from mitmproxy.builtins import streambodies
+
+
+class TestStreamBodies(mastertest.MasterTest):
+    def test_simple(self):
+        s = state.DummyState()
+        o = options.Options(stream_large_bodies = 10)
+        m = master.FlowMaster(o, None, s)
+        sa = streambodies.StreamBodies()
+        m.addons.add(sa)
+
+        f = tutils.tflow()
+        f.request.content = b""
+        f.request.headers["Content-Length"] = "1024"
+        assert not f.request.stream
+        m.requestheaders(f)
+        assert f.request.stream
+
+        f = tutils.tflow(resp=True)
+        f.response.content = b""
+        f.response.headers["Content-Length"] = "1024"
+        assert not f.response.stream
+        m.responseheaders(f)
+        assert f.response.stream

test/mitmproxy/test_server.py

@@ -288,18 +288,6 @@ class TestHTTP(tservers.HTTPProxyTest, CommonMixin, AppMixin):
             resp = p.request("get:'http://foo':h':foo'='bar'")
         assert resp.status_code == 400
 
-    def test_stream(self):
-        self.master.set_stream_large_bodies(1024 * 2)
-
-        self.pathod("200:b@1k")
-        assert not self.master.state.view[-1].response.stream
-        assert len(self.master.state.view[-1].response.content) == 1024 * 1
-
-        self.pathod("200:b@3k")
-        assert self.master.state.view[-1].response.stream
-        assert self.master.state.view[-1].response.content is None
-        self.master.set_stream_large_bodies(None)
-
     def test_stream_modify(self):
         s = script.Script(
             tutils.test_data.path("data/addonscripts/stream_modify.py")