From c47ddaa3a025597f8706c437f792c1ad12c388ab Mon Sep 17 00:00:00 2001 From: Brad Peabody Date: Thu, 17 Jul 2014 22:43:26 -0700 Subject: basic attempt to implement streaming response, needs testing --- libmproxy/protocol/http.py | 61 ++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 56 insertions(+), 5 deletions(-) (limited to 'libmproxy/protocol/http.py') diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py index b7ff5b4b..f3d5f666 100644 --- a/libmproxy/protocol/http.py +++ b/libmproxy/protocol/http.py @@ -865,15 +865,16 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): pass self.c.close = True - def get_response_from_server(self, request): + def get_response_from_server(self, request, stream=False): self.c.establish_server_connection() request_raw = request._assemble() for i in range(2): try: self.c.server_conn.send(request_raw) - return HTTPResponse.from_stream(self.c.server_conn.rfile, request.method, - body_size_limit=self.c.config.body_size_limit) + res = HTTPResponse.from_stream(self.c.server_conn.rfile, request.method, + body_size_limit=self.c.config.body_size_limit, include_content=(not stream)) + return res except (tcp.NetLibDisconnect, http.HttpErrorConnClosed), v: self.c.log("error in server communication: %s" % str(v), level="debug") if i < 1: @@ -892,6 +893,8 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): def handle_flow(self): flow = HTTPFlow(self.c.client_conn, self.c.server_conn, self.change_server) + flow.stream_expecting_body = False + flow.stream = False try: req = HTTPRequest.from_stream(self.c.client_conn.rfile, body_size_limit=self.c.config.body_size_limit) @@ -915,7 +918,23 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): if isinstance(request_reply, HTTPResponse): flow.response = request_reply else: - flow.response = self.get_response_from_server(flow.request) + + # read initially in "stream" mode, so we can get the headers separately + flow.response = self.get_response_from_server(flow.request,stream=True) + + if flow.response.content == None: + flow.stream_expecting_body = True + flow.response.content = "" # set this to empty string or other things get really confused, + # flow.stream_expecting_body now contains the state info of whether or not + # body still remains to be read + + # call the appropriate script hook - this is an opportunity for + responseheaders_reply = self.c.channel.ask("responseheaders", flow.response) + # hm - do we need to do something with responseheaders_reply?? + + # now get the rest of the request body, if body still needs to be read but not streaming this response + if flow.stream_expecting_body and not flow.stream: + flow.response.content = http.read_http_body(self.c.server_conn.rfile, flow.response.headers, self.c.config.body_size_limit, False) flow.server_conn = self.c.server_conn # no further manipulation of self.c.server_conn beyond this point # we can safely set it as the final attribute value here. @@ -925,7 +944,39 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): if response_reply is None or response_reply == KILL: return False - self.c.client_conn.send(flow.response._assemble()) + if not flow.stream or not flow.stream_expecting_body: + # if not streaming or there is no body to be read, we'll already have the body, just send it + self.c.client_conn.send(flow.response._assemble()) + else: + + # if streaming, we still need to read the body and stream its bits back to the client + + # start with head + h = flow.response._assemble_head() + self.c.client_conn.send(h) + + # if chunked then we send back each chunk + if http.has_chunked_encoding(flow.response.headers): + while 1: + content = http.read_next_chunk(self.c.server_conn.rfile, flow.response.headers, False) + if not http.write_chunk(self.c.client_conn.rfile, content): + break + + else: # not chunked, we send back 4k at a time + clen = http.expected_http_body_size(flow.response.headers, False) + clen = clen if clen >= 0 else (64 * 1024 * 1024 * 1024) # arbitrary max of 64G if no length set + rcount = 0 + blocksize = 4096 + while 1: + bytes_to_read = min(blocksize, clen - rcount) + content = self.c.server_conn.rfile.read(bytes_to_read) + if content == "": # check for EOF + break + rcount += len(content) + self.c.client_conn.rfile.write(content) + if rcount >= clen: # check for having read up to clen + break + flow.timestamp_end = utils.timestamp() if (http.connection_close(flow.request.httpversion, flow.request.headers) or -- cgit v1.2.3 From 560e23af092ab566e75060346ebde739ac07f179 Mon Sep 17 00:00:00 2001 From: Brad Peabody Date: Sat, 19 Jul 2014 19:10:14 -0700 Subject: fixed handling of Transfer-Encoding header during streaming; wrote tests for streaming support --- libmproxy/protocol/http.py | 38 ++++++++++++++++++++++---------------- 1 file changed, 22 insertions(+), 16 deletions(-) (limited to 'libmproxy/protocol/http.py') diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py index f3d5f666..2c3c8f97 100644 --- a/libmproxy/protocol/http.py +++ b/libmproxy/protocol/http.py @@ -632,24 +632,21 @@ class HTTPResponse(HTTPMessage): return 'HTTP/%s.%s %s %s' % \ (self.httpversion[0], self.httpversion[1], self.code, self.msg) - def _assemble_headers(self): + def _assemble_headers(self, preserve_transfer_encoding=False): headers = self.headers.copy() - utils.del_all( - headers, - [ - 'Proxy-Connection', - 'Transfer-Encoding' - ] - ) + utils.del_all(headers,['Proxy-Connection']) + if not preserve_transfer_encoding: + utils.del_all(headers,['Transfer-Encoding']) + if self.content: headers["Content-Length"] = [str(len(self.content))] - elif 'Transfer-Encoding' in self.headers: # add content-length for chuncked transfer-encoding with no content + elif not preserve_transfer_encoding and 'Transfer-Encoding' in self.headers: # add content-length for chuncked transfer-encoding with no content headers["Content-Length"] = ["0"] return str(headers) - def _assemble_head(self): - return '%s\r\n%s\r\n' % (self._assemble_first_line(), self._assemble_headers()) + def _assemble_head(self, preserve_transfer_encoding=False): + return '%s\r\n%s\r\n' % (self._assemble_first_line(), self._assemble_headers(preserve_transfer_encoding=preserve_transfer_encoding)) def _assemble(self): """ @@ -928,7 +925,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): # flow.stream_expecting_body now contains the state info of whether or not # body still remains to be read - # call the appropriate script hook - this is an opportunity for + # call the appropriate script hook - this is an opportunity for an inline script to set flow.stream = True responseheaders_reply = self.c.channel.ask("responseheaders", flow.response) # hm - do we need to do something with responseheaders_reply?? @@ -944,6 +941,8 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): if response_reply is None or response_reply == KILL: return False + disconnected_while_streaming = False + if not flow.stream or not flow.stream_expecting_body: # if not streaming or there is no body to be read, we'll already have the body, just send it self.c.client_conn.send(flow.response._assemble()) @@ -952,15 +951,17 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): # if streaming, we still need to read the body and stream its bits back to the client # start with head - h = flow.response._assemble_head() + h = flow.response._assemble_head(preserve_transfer_encoding=True) self.c.client_conn.send(h) # if chunked then we send back each chunk if http.has_chunked_encoding(flow.response.headers): while 1: content = http.read_next_chunk(self.c.server_conn.rfile, flow.response.headers, False) - if not http.write_chunk(self.c.client_conn.rfile, content): + if not http.write_chunk(self.c.client_conn.wfile, content): break + self.c.client_conn.wfile.flush() + self.c.client_conn.wfile.flush() else: # not chunked, we send back 4k at a time clen = http.expected_http_body_size(flow.response.headers, False) @@ -969,17 +970,21 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): blocksize = 4096 while 1: bytes_to_read = min(blocksize, clen - rcount) + if bytes_to_read == 0: + break content = self.c.server_conn.rfile.read(bytes_to_read) if content == "": # check for EOF + disconnected_while_streaming = True break rcount += len(content) - self.c.client_conn.rfile.write(content) + self.c.client_conn.wfile.write(content) + self.c.client_conn.wfile.flush() if rcount >= clen: # check for having read up to clen break flow.timestamp_end = utils.timestamp() - if (http.connection_close(flow.request.httpversion, flow.request.headers) or + if (disconnected_while_streaming or http.connection_close(flow.request.httpversion, flow.request.headers) or http.connection_close(flow.response.httpversion, flow.response.headers)): return False @@ -989,6 +994,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): # If the user has changed the target server on this connection, # restore the original target server self.restore_server() + return True except (HttpAuthenticationError, http.HttpError, proxy.ProxyError, tcp.NetLibError), e: self.handle_error(e, flow) -- cgit v1.2.3 From 7398db80db004546070139c0c7e79bba4f92b318 Mon Sep 17 00:00:00 2001 From: Maximilian Hils Date: Sun, 20 Jul 2014 11:17:53 +0200 Subject: simplify responseheader scripthook --- libmproxy/protocol/http.py | 20 ++++++-------------- 1 file changed, 6 insertions(+), 14 deletions(-) (limited to 'libmproxy/protocol/http.py') diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py index 242443ec..711cb06c 100644 --- a/libmproxy/protocol/http.py +++ b/libmproxy/protocol/http.py @@ -870,7 +870,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): try: self.c.server_conn.send(request_raw) res = HTTPResponse.from_stream(self.c.server_conn.rfile, request.method, - body_size_limit=self.c.config.body_size_limit, include_content=(not stream)) + body_size_limit=self.c.config.body_size_limit, include_body=(not stream)) return res except (tcp.NetLibDisconnect, http.HttpErrorConnClosed), v: self.c.log("error in server communication: %s" % str(v), level="debug") @@ -890,8 +890,6 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): def handle_flow(self): flow = HTTPFlow(self.c.client_conn, self.c.server_conn, self.change_server) - flow.stream_expecting_body = False - flow.stream = False try: req = HTTPRequest.from_stream(self.c.client_conn.rfile, body_size_limit=self.c.config.body_size_limit) @@ -917,20 +915,14 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): else: # read initially in "stream" mode, so we can get the headers separately - flow.response = self.get_response_from_server(flow.request,stream=True) - - if flow.response.content == None: - flow.stream_expecting_body = True - flow.response.content = "" # set this to empty string or other things get really confused, - # flow.stream_expecting_body now contains the state info of whether or not - # body still remains to be read + flow.response = self.get_response_from_server(flow.request, stream=True) + flow.response.stream = False # call the appropriate script hook - this is an opportunity for an inline script to set flow.stream = True - responseheaders_reply = self.c.channel.ask("responseheaders", flow.response) - # hm - do we need to do something with responseheaders_reply?? + self.c.channel.ask("responseheaders", flow) # now get the rest of the request body, if body still needs to be read but not streaming this response - if flow.stream_expecting_body and not flow.stream: + if not flow.response.stream and flow.response.content is None: flow.response.content = http.read_http_body(self.c.server_conn.rfile, flow.response.headers, self.c.config.body_size_limit, False) flow.server_conn = self.c.server_conn # no further manipulation of self.c.server_conn beyond this point @@ -943,7 +935,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): disconnected_while_streaming = False - if not flow.stream or not flow.stream_expecting_body: + if flow.response.content is not None: # if not streaming or there is no body to be read, we'll already have the body, just send it self.c.client_conn.send(flow.response._assemble()) else: -- cgit v1.2.3 From 562ac9e721c33b05e8889d4932dede794a9746a8 Mon Sep 17 00:00:00 2001 From: Maximilian Hils Date: Mon, 21 Jul 2014 14:09:24 +0200 Subject: unify stream handling --- libmproxy/protocol/http.py | 65 +++++++++++++++++----------------------------- 1 file changed, 24 insertions(+), 41 deletions(-) (limited to 'libmproxy/protocol/http.py') diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py index 711cb06c..31dd39f5 100644 --- a/libmproxy/protocol/http.py +++ b/libmproxy/protocol/http.py @@ -293,7 +293,8 @@ class HTTPRequest(HTTPMessage): raise http.HttpError(400, "Invalid headers") if include_body: - content = http.read_http_body(rfile, headers, body_size_limit, True) + content = http.read_http_body(rfile, headers, body_size_limit, + method, None, True) timestamp_end = utils.timestamp() return HTTPRequest(form_in, method, scheme, host, port, path, httpversion, headers, @@ -305,7 +306,7 @@ class HTTPRequest(HTTPMessage): if form == "relative": path = self.path if self.method != "OPTIONS" else "*" request_line = '%s %s HTTP/%s.%s' % \ - (self.method, path, self.httpversion[0], self.httpversion[1]) + (self.method, path, self.httpversion[0], self.httpversion[1]) elif form == "authority": request_line = '%s %s:%s HTTP/%s.%s' % (self.method, self.host, self.port, self.httpversion[0], self.httpversion[1]) @@ -634,9 +635,9 @@ class HTTPResponse(HTTPMessage): def _assemble_headers(self, preserve_transfer_encoding=False): headers = self.headers.copy() - utils.del_all(headers,['Proxy-Connection']) + utils.del_all(headers, ['Proxy-Connection']) if not preserve_transfer_encoding: - utils.del_all(headers,['Transfer-Encoding']) + utils.del_all(headers, ['Transfer-Encoding']) if self.content: headers["Content-Length"] = [str(len(self.content))] @@ -646,7 +647,8 @@ class HTTPResponse(HTTPMessage): return str(headers) def _assemble_head(self, preserve_transfer_encoding=False): - return '%s\r\n%s\r\n' % (self._assemble_first_line(), self._assemble_headers(preserve_transfer_encoding=preserve_transfer_encoding)) + return '%s\r\n%s\r\n' % ( + self._assemble_first_line(), self._assemble_headers(preserve_transfer_encoding=preserve_transfer_encoding)) def _assemble(self): """ @@ -862,7 +864,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): pass self.c.close = True - def get_response_from_server(self, request, stream=False): + def get_response_from_server(self, request, include_body=True): self.c.establish_server_connection() request_raw = request._assemble() @@ -870,7 +872,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): try: self.c.server_conn.send(request_raw) res = HTTPResponse.from_stream(self.c.server_conn.rfile, request.method, - body_size_limit=self.c.config.body_size_limit, include_body=(not stream)) + body_size_limit=self.c.config.body_size_limit, include_body=include_body) return res except (tcp.NetLibDisconnect, http.HttpErrorConnClosed), v: self.c.log("error in server communication: %s" % str(v), level="debug") @@ -915,7 +917,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): else: # read initially in "stream" mode, so we can get the headers separately - flow.response = self.get_response_from_server(flow.request, stream=True) + flow.response = self.get_response_from_server(flow.request, include_body=False) flow.response.stream = False # call the appropriate script hook - this is an opportunity for an inline script to set flow.stream = True @@ -923,7 +925,9 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): # now get the rest of the request body, if body still needs to be read but not streaming this response if not flow.response.stream and flow.response.content is None: - flow.response.content = http.read_http_body(self.c.server_conn.rfile, flow.response.headers, self.c.config.body_size_limit, False) + flow.response.content = http.read_http_body(self.c.server_conn.rfile, flow.response.headers, + self.c.config.body_size_limit, + flow.request.method, flow.response.code, False) flow.server_conn = self.c.server_conn # no further manipulation of self.c.server_conn beyond this point # we can safely set it as the final attribute value here. @@ -933,8 +937,6 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): if response_reply is None or response_reply == KILL: return False - disconnected_while_streaming = False - if flow.response.content is not None: # if not streaming or there is no body to be read, we'll already have the body, just send it self.c.client_conn.send(flow.response._assemble()) @@ -946,38 +948,19 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): h = flow.response._assemble_head(preserve_transfer_encoding=True) self.c.client_conn.send(h) - # if chunked then we send back each chunk - if http.has_chunked_encoding(flow.response.headers): - while 1: - content = http.read_next_chunk(self.c.server_conn.rfile, flow.response.headers, False) - if not http.write_chunk(self.c.client_conn.wfile, content): - break - self.c.client_conn.wfile.flush() + for chunk in http.read_http_body_chunked(self.c.server_conn.rfile, + flow.response.headers, + self.c.config.body_size_limit, "GET", 200, False, 4096): + for part in chunk: + self.c.client_conn.wfile.write(part) self.c.client_conn.wfile.flush() - else: # not chunked, we send back 4k at a time - clen = http.expected_http_body_size(flow.response.headers, False) - clen = clen if clen >= 0 else (64 * 1024 * 1024 * 1024) # arbitrary max of 64G if no length set - rcount = 0 - blocksize = 4096 - while 1: - bytes_to_read = min(blocksize, clen - rcount) - if bytes_to_read == 0: - break - content = self.c.server_conn.rfile.read(bytes_to_read) - if content == "": # check for EOF - disconnected_while_streaming = True - break - rcount += len(content) - self.c.client_conn.wfile.write(content) - self.c.client_conn.wfile.flush() - if rcount >= clen: # check for having read up to clen - break - flow.timestamp_end = utils.timestamp() - if (disconnected_while_streaming or http.connection_close(flow.request.httpversion, flow.request.headers) or - http.connection_close(flow.response.httpversion, flow.response.headers)): + if (http.connection_close(flow.request.httpversion, flow.request.headers) or + http.connection_close(flow.response.httpversion, flow.response.headers) or + http.expected_http_body_size(flow.response.headers, False, flow.request.method, + flow.response.code) == -1): return False if flow.request.form_in == "authority": @@ -1009,7 +992,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): if flow.request and not flow.response: self.c.channel.ask("error", flow.error) else: - pass # FIXME: Do we want to persist errors without flows? + pass # FIXME: Do we want to persist errors without flows? try: self.send_error(code, message, headers) @@ -1109,7 +1092,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): return True raise http.HttpError(400, "Invalid HTTP request form (expected: %s, got: %s)" % - (self.expected_form_in, request.form_in)) + (self.expected_form_in, request.form_in)) def authenticate(self, request): if self.c.config.authenticator: -- cgit v1.2.3 From 4b4a18a2e4d7cf3e8862192b68f5a2295da9acbe Mon Sep 17 00:00:00 2001 From: Maximilian Hils Date: Mon, 21 Jul 2014 21:06:55 +0200 Subject: add --stream options, various fixes --- libmproxy/protocol/http.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) (limited to 'libmproxy/protocol/http.py') diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py index 31dd39f5..4648c7cf 100644 --- a/libmproxy/protocol/http.py +++ b/libmproxy/protocol/http.py @@ -924,7 +924,9 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): self.c.channel.ask("responseheaders", flow) # now get the rest of the request body, if body still needs to be read but not streaming this response - if not flow.response.stream and flow.response.content is None: + if flow.response.stream: + flow.response.content = CONTENT_MISSING + else: flow.response.content = http.read_http_body(self.c.server_conn.rfile, flow.response.headers, self.c.config.body_size_limit, flow.request.method, flow.response.code, False) @@ -937,20 +939,19 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): if response_reply is None or response_reply == KILL: return False - if flow.response.content is not None: - # if not streaming or there is no body to be read, we'll already have the body, just send it + if not flow.response.stream: + # no streaming: + # we already received the full response from the server and can send it to the client straight away. self.c.client_conn.send(flow.response._assemble()) else: - - # if streaming, we still need to read the body and stream its bits back to the client - - # start with head + # streaming: + # First send the body and then transfer the response incrementally: h = flow.response._assemble_head(preserve_transfer_encoding=True) self.c.client_conn.send(h) - for chunk in http.read_http_body_chunked(self.c.server_conn.rfile, flow.response.headers, - self.c.config.body_size_limit, "GET", 200, False, 4096): + self.c.config.body_size_limit, flow.request.method, + flow.response.code, False, 4096): for part in chunk: self.c.client_conn.wfile.write(part) self.c.client_conn.wfile.flush() -- cgit v1.2.3 From d9ac029ec7d18e5c1a483c7141ba86ad185874b0 Mon Sep 17 00:00:00 2001 From: Maximilian Hils Date: Wed, 23 Jul 2014 00:21:33 +0200 Subject: always initialize HTTPResponse.stream attribute --- libmproxy/protocol/http.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'libmproxy/protocol/http.py') diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py index 4648c7cf..8b9cb448 100644 --- a/libmproxy/protocol/http.py +++ b/libmproxy/protocol/http.py @@ -592,6 +592,7 @@ class HTTPResponse(HTTPMessage): # Is this request replayed? self.is_replay = False + self.stream = False _stateobject_attributes = HTTPMessage._stateobject_attributes.copy() _stateobject_attributes.update( @@ -918,7 +919,6 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): # read initially in "stream" mode, so we can get the headers separately flow.response = self.get_response_from_server(flow.request, include_body=False) - flow.response.stream = False # call the appropriate script hook - this is an opportunity for an inline script to set flow.stream = True self.c.channel.ask("responseheaders", flow) -- cgit v1.2.3 From 4382829b7d9fde4358e12f7c9f195c7d7c854ff1 Mon Sep 17 00:00:00 2001 From: Maximilian Hils Date: Fri, 25 Jul 2014 18:47:48 +0200 Subject: workaround: always make sure that flow.response.reply exists --- libmproxy/protocol/http.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) (limited to 'libmproxy/protocol/http.py') diff --git a/libmproxy/protocol/http.py b/libmproxy/protocol/http.py index 8b9cb448..cc6533b2 100644 --- a/libmproxy/protocol/http.py +++ b/libmproxy/protocol/http.py @@ -921,7 +921,7 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): flow.response = self.get_response_from_server(flow.request, include_body=False) # call the appropriate script hook - this is an opportunity for an inline script to set flow.stream = True - self.c.channel.ask("responseheaders", flow) + self.c.channel.ask("responseheaders", flow.response) # now get the rest of the request body, if body still needs to be read but not streaming this response if flow.response.stream: @@ -931,8 +931,9 @@ class HTTPHandler(ProtocolHandler, TemporaryServerChangeMixin): self.c.config.body_size_limit, flow.request.method, flow.response.code, False) - flow.server_conn = self.c.server_conn # no further manipulation of self.c.server_conn beyond this point + # no further manipulation of self.c.server_conn beyond this point # we can safely set it as the final attribute value here. + flow.server_conn = self.c.server_conn self.c.log("response", "debug", [flow.response._assemble_first_line()]) response_reply = self.c.channel.ask("response", flow.response) -- cgit v1.2.3