Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-26 18:18:25 +00:00
Code Issues Projects Releases Wiki Activity

Merge branch 'http-models'

Browse Source
This commit is contained in:
Maximilian Hils 2015-09-28 13:54:53 +02:00
commit c11ab3676d
18 changed files with 136 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/Makefile
View File

@ -192,4 +192,4 @@ pseudoxml:
@echo "Build finished. The pseudo-XML files are in $(BUILDDIR)/pseudoxml."
livehtml:
sphinx-autobuild -b html -z '../libmproxy' -r '___jb_(old|bak)___$$' $(ALLSPHINXOPTS) $(BUILDDIR)/html
sphinx-autobuild -b html -z '../libmproxy' -z '../../netlib/netlib' -r '___jb_(old|bak)___$$' $(ALLSPHINXOPTS) $(BUILDDIR)/html

3
docs/conf.py
View File

@ -40,6 +40,9 @@ extensions = [
'sphinxcontrib.documentedlist'
]
# https://github.com/sphinx-doc/sphinx/pull/2053
napoleon_include_special_with_doc = False
autodoc_member_order = "bysource"
# Add any paths that contain templates here, relative to this directory.

58
docs/dev/models.rst
View File

@ -3,23 +3,57 @@
Models
======
.. warning::
The documentation for models has not been converted to rst yet and **many attributes/features
are missing**.
Please read the source code instead.
.. automodule:: netlib.http
.. automodule:: libmproxy.models
:show-inheritance:
:members:
:undoc-members:
.. autoclass:: Request
.. rubric:: Data
.. autoattribute:: first_line_format
.. autoattribute:: method
.. autoattribute:: scheme
.. autoattribute:: host
.. autoattribute:: port
.. autoattribute:: path
.. autoattribute:: http_version
.. autoattribute:: headers
.. autoattribute:: content
.. autoattribute:: timestamp_start
.. autoattribute:: timestamp_end
.. rubric:: Computed Properties and Convenience Methods
.. autoattribute:: text
.. autoattribute:: url
.. autoattribute:: pretty_host
.. autoattribute:: pretty_url
.. autoattribute:: query
.. autoattribute:: cookies
.. autoattribute:: path_components
.. automethod:: anticache
.. automethod:: anticomp
.. automethod:: constrain_encoding
.. autoattribute:: urlencoded_form
.. autoattribute:: multipart_form
.. automodule:: netlib.http.semantics
:members: Request, Response
:undoc-members:
.. autoclass:: Response
.. rubric:: Data
.. autoattribute:: http_version
.. autoattribute:: status_code
.. autoattribute:: reason
.. autoattribute:: headers
.. autoattribute:: content
.. autoattribute:: timestamp_start
.. autoattribute:: timestamp_end
.. rubric:: Computed Properties and Convenience Methods
.. autoattribute:: text
.. autoattribute:: cookies
.. autoclass:: Headers
:show-inheritance:
:members:
:special-members:
:no-undoc-members:
.. autoclass:: decoded
.. automodule:: libmproxy.models
:show-inheritance:
:members: HTTPFlow, Error, ClientConnection, ServerConnection

2
examples/dns_spoofing.py
View File

@ -39,7 +39,7 @@ def request(context, flow):
sni = None
port = 80
host_header = flow.request.pretty_host(hostheader=True)
host_header = flow.request.pretty_host
m = parse_host_header.match(host_header)
if m:
host_header = m.group("host").strip("[]")

9
examples/redirect_requests.py
View File

@ -5,12 +5,12 @@ from libmproxy.models import HTTPResponse
from netlib.http import Headers
def request(context, flow):
# pretty_host(hostheader=True) takes the Host: header of the request into account,
# pretty_host takes the "Host" header of the request into account,
# which is useful in transparent mode where we usually only have the IP
# otherwise.
# Method 1: Answer with a locally generated response
if flow.request.pretty_host(hostheader=True).endswith("example.com"):
if flow.request.pretty_host.endswith("example.com"):
resp = HTTPResponse(
[1, 1], 200, "OK",
Headers(Content_Type="text/html"),
@ -18,6 +18,5 @@ def request(context, flow):
flow.reply(resp)
# Method 2: Redirect the request to a different server
if flow.request.pretty_host(hostheader=True).endswith("example.org"):
flow.request.host = "mitmproxy.org"
flow.request.update_host_header()
if flow.request.pretty_host.endswith("example.org"):
flow.request.host = "mitmproxy.org"

2
libmproxy/console/common.py
View File

@ -390,7 +390,7 @@ def format_flow(f, focus, extended=False, hostheader=False, padding=2,
req_timestamp = f.request.timestamp_start,
req_is_replay = f.request.is_replay,
req_method = f.request.method,
req_url = f.request.pretty_url(hostheader=hostheader),
req_url = f.request.pretty_url if hostheader else f.request.url,
err_msg = f.error.msg if f.error else None,
resp_code = f.response.status_code if f.response else None,

6
libmproxy/console/flowview.py
View File

@ -168,7 +168,7 @@ class FlowView(tabs.Tabs):
self.show()
def content_view(self, viewmode, message):
if message.body == CONTENT_MISSING:
if message.content == CONTENT_MISSING:
msg, body = "", [urwid.Text([("error", "[content missing]")])]
return msg, body
else:
@ -193,13 +193,13 @@ class FlowView(tabs.Tabs):
try:
description, lines = contentviews.get_content_view(
viewmode, message.body, headers=message.headers
viewmode, message.content, headers=message.headers
)
except ContentViewException:
s = "Content viewer failed: \n" + traceback.format_exc()
signals.add_event(s, "error")
description, lines = contentviews.get_content_view(
contentviews.get("Raw"), message.body, headers=message.headers
contentviews.get("Raw"), message.content, headers=message.headers
)
description = description.replace("Raw", "Couldn't parse: falling back to Raw")

20
libmproxy/dump.py
View File

@ -174,15 +174,15 @@ class DumpMaster(flow.FlowMaster):
)
self.echo(headers, indent=4)
if self.o.flow_detail >= 3:
if message.body == CONTENT_MISSING:
if message.content == CONTENT_MISSING:
self.echo("(content missing)", indent=4)
elif message.body:
elif message.content:
self.echo("")
try:
type, lines = contentviews.get_content_view(
contentviews.get("Auto"),
message.body,
message.content,
headers=message.headers
)
except ContentViewException:
@ -190,7 +190,7 @@ class DumpMaster(flow.FlowMaster):
self.add_event(s, "debug")
type, lines = contentviews.get_content_view(
contentviews.get("Raw"),
message.body,
message.content,
headers=message.headers
)
@ -241,7 +241,11 @@ class DumpMaster(flow.FlowMaster):
DELETE="red"
).get(method.upper(), "magenta")
method = click.style(method, fg=method_color, bold=True)
url = click.style(flow.request.pretty_url(self.showhost), bold=True)
if self.showhost:
url = flow.request.pretty_url
else:
url = flow.request.url
url = click.style(url, bold=True)
line = "{stickycookie}{client} {method} {url}".format(
stickycookie=stickycookie,
@ -266,7 +270,7 @@ class DumpMaster(flow.FlowMaster):
elif 400 <= code < 600:
code_color = "red"
code = click.style(str(code), fg=code_color, bold=True, blink=(code == 418))
msg = click.style(flow.response.msg, fg=code_color, bold=True)
reason = click.style(flow.response.reason, fg=code_color, bold=True)
if flow.response.content == CONTENT_MISSING:
size = "(content missing)"
@ -276,11 +280,11 @@ class DumpMaster(flow.FlowMaster):
arrows = click.style("<<", bold=True)
line = "{replay} {arrows} {code} {msg} {size}".format(
line = "{replay} {arrows} {code} {reason} {size}".format(
replay=replay,
arrows=arrows,
code=code,
msg=msg,
reason=reason,
size=size
)
self.echo(line)

2
libmproxy/flow.py
View File

@ -252,7 +252,7 @@ class ServerPlaybackState:
]
if not self.ignore_content:
form_contents = r.get_form()
form_contents = r.urlencoded_form or r.multipart_form
if self.ignore_payload_params and form_contents:
key.extend(
p for p in form_contents

101
libmproxy/models/http.py
View File

@ -6,9 +6,8 @@ import time
from libmproxy import utils
from netlib import encoding
from netlib.http import status_codes, Headers, Request, Response, CONTENT_MISSING
from netlib.http import status_codes, Headers, Request, Response, CONTENT_MISSING, decoded
from netlib.tcp import Address
from netlib.utils import native
from .. import version, stateobject
from .flow import Flow
@ -17,7 +16,6 @@ class MessageMixin(stateobject.StateObject):
_stateobject_attributes = dict(
http_version=bytes,
headers=Headers,
body=bytes,
timestamp_start=float,
timestamp_end=float
)
@ -26,9 +24,9 @@ class MessageMixin(stateobject.StateObject):
def get_state(self, short=False):
ret = super(MessageMixin, self).get_state(short)
if short:
if self.body:
ret["contentLength"] = len(self.body)
elif self.body == CONTENT_MISSING:
if self.content:
ret["contentLength"] = len(self.content)
elif self.content == CONTENT_MISSING:
ret["contentLength"] = None
else:
ret["contentLength"] = 0
@ -41,9 +39,9 @@ class MessageMixin(stateobject.StateObject):
Doesn't change the message iteself or its headers.
"""
ce = self.headers.get("content-encoding")
if not self.body or ce not in encoding.ENCODINGS:
return self.body
return encoding.decode(ce, self.body)
if not self.content or ce not in encoding.ENCODINGS:
return self.content
return encoding.decode(ce, self.content)
def decode(self):
"""
@ -54,12 +52,12 @@ class MessageMixin(stateobject.StateObject):
Returns True if decoding succeeded, False otherwise.
"""
ce = self.headers.get("content-encoding")
if not self.body or ce not in encoding.ENCODINGS:
if not self.content or ce not in encoding.ENCODINGS:
return False
data = encoding.decode(ce, self.body)
data = encoding.decode(ce, self.content)
if data is None:
return False
self.body = data
self.content = data
self.headers.pop("content-encoding", None)
return True
@ -69,11 +67,14 @@ class MessageMixin(stateobject.StateObject):
or "identity".
"""
# FIXME: Error if there's an existing encoding header?
self.body = encoding.encode(e, self.body)
self.content = encoding.encode(e, self.content)
self.headers["content-encoding"] = e
def copy(self):
c = copy.copy(self)
if hasattr(self, "data"): # FIXME remove condition
c.data = copy.copy(self.data)
c.headers = self.headers.copy()
return c
@ -86,8 +87,8 @@ class MessageMixin(stateobject.StateObject):
Returns the number of replacements made.
"""
with decoded(self):
self.body, count = utils.safe_subn(
pattern, repl, self.body, *args, **kwargs
self.content, count = utils.safe_subn(
pattern, repl, self.content, *args, **kwargs
)
fields = []
for name, value in self.headers.fields:
@ -147,7 +148,7 @@ class HTTPRequest(MessageMixin, Request):
def __init__(
self,
form_in,
first_line_format,
method,
scheme,
host,
@ -155,14 +156,14 @@ class HTTPRequest(MessageMixin, Request):
path,
http_version,
headers,
body,
content,
timestamp_start=None,
timestamp_end=None,
form_out=None,
):
Request.__init__(
self,
form_in,
first_line_format,
method,
scheme,
host,
@ -170,11 +171,11 @@ class HTTPRequest(MessageMixin, Request):
path,
http_version,
headers,
body,
content,
timestamp_start,
timestamp_end,
)
self.form_out = form_out or form_in
self.form_out = form_out or first_line_format # FIXME remove
# Have this request's cookies been modified by sticky cookies or auth?
self.stickycookie = False
@ -185,7 +186,8 @@ class HTTPRequest(MessageMixin, Request):
_stateobject_attributes = MessageMixin._stateobject_attributes.copy()
_stateobject_attributes.update(
form_in=str,
content=bytes,
first_line_format=str,
method=bytes,
scheme=bytes,
host=bytes,
@ -225,7 +227,7 @@ class HTTPRequest(MessageMixin, Request):
@classmethod
def wrap(self, request):
req = HTTPRequest(
form_in=request.form_in,
first_line_format=request.form_in,
method=request.method,
scheme=request.scheme,
host=request.host,
@ -233,7 +235,7 @@ class HTTPRequest(MessageMixin, Request):
path=request.path,
http_version=request.http_version,
headers=request.headers,
body=request.body,
content=request.content,
timestamp_start=request.timestamp_start,
timestamp_end=request.timestamp_end,
form_out=(request.form_out if hasattr(request, 'form_out') else None),
@ -288,9 +290,9 @@ class HTTPResponse(MessageMixin, Response):
self,
http_version,
status_code,
msg,
reason,
headers,
body,
content,
timestamp_start=None,
timestamp_end=None,
):
@ -298,9 +300,9 @@ class HTTPResponse(MessageMixin, Response):
self,
http_version,
status_code,
msg,
reason,
headers,
body,
content,
timestamp_start=timestamp_start,
timestamp_end=timestamp_end,
)
@ -311,6 +313,7 @@ class HTTPResponse(MessageMixin, Response):
_stateobject_attributes = MessageMixin._stateobject_attributes.copy()
_stateobject_attributes.update(
body=bytes,
status_code=int,
msg=bytes
)
@ -336,9 +339,9 @@ class HTTPResponse(MessageMixin, Response):
resp = HTTPResponse(
http_version=response.http_version,
status_code=response.status_code,
msg=response.msg,
reason=response.reason,
headers=response.headers,
body=response.body,
content=response.content,
timestamp_start=response.timestamp_start,
timestamp_end=response.timestamp_end,
)
@ -400,22 +403,20 @@ class HTTPResponse(MessageMixin, Response):
class HTTPFlow(Flow):
"""
A HTTPFlow is a collection of objects representing a single HTTP
transaction. The main attributes are:
transaction.
Attributes:
request: HTTPRequest object
response: HTTPResponse object
error: Error object
server_conn: ServerConnection object
client_conn: ClientConnection object
intercepted: Is this flow currently being intercepted?
live: Does this flow have a live client connection?
Note that it's possible for a Flow to have both a response and an error
object. This might happen, for instance, when a response was received
from the server, but there was an error sending it back to the client.
The following additional attributes are exposed:
intercepted: Is this flow currently being intercepted?
live: Does this flow have a live client connection?
"""
def __init__(self, client_conn, server_conn, live=None):
@ -485,36 +486,6 @@ class HTTPFlow(Flow):
return c
class decoded(object):
"""
A context manager that decodes a request or response, and then
re-encodes it with the same encoding after execution of the block.
Example:
with decoded(request):
request.content = request.content.replace("foo", "bar")
"""
def __init__(self, o):
self.o = o
ce = o.headers.get("content-encoding")
if ce:
ce = native(ce, "ascii", "ignore")
if ce in encoding.ENCODINGS:
self.ce = ce
else:
self.ce = None
def __enter__(self):
if self.ce:
if not self.o.decode():
self.ce = None
def __exit__(self, type, value, tb):
if self.ce:
self.o.encode(self.ce)
def make_error_response(status_code, message, headers=None):
response = status_codes.RESPONSES.get(status_code, "Unknown")
body = """

6
libmproxy/protocol/http.py
View File

@ -54,7 +54,7 @@ class _StreamingHttpLayer(_HttpLayer):
def read_response(self, request):
response = self.read_response_headers()
response.body = b"".join(
response.content = b"".join(
self.read_response_body(request, response)
)
return response
@ -66,10 +66,10 @@ class _StreamingHttpLayer(_HttpLayer):
raise NotImplementedError()
def send_response(self, response):
if response.body == CONTENT_MISSING:
if response.content == CONTENT_MISSING:
raise HttpException("Cannot assemble flow with CONTENT_MISSING")
self.send_response_headers(response)
self.send_response_body(response, [response.body])
self.send_response_body(response, [response.content])
class Http1Layer(_StreamingHttpLayer):

5
libmproxy/protocol/tls.py
View File

@ -7,7 +7,6 @@ from construct import ConstructError
import six
from netlib.exceptions import InvalidCertificateException, TcpException, TlsException
from netlib.http import ALPN_PROTO_HTTP1
from ..contrib.tls._constructs import ClientHello
from ..exceptions import ProtocolException, TlsProtocolException, ClientHandshakeException
from .base import Layer
@ -367,8 +366,8 @@ class TlsLayer(Layer):
"""
# This gets triggered if we haven't established an upstream connection yet.
default_alpn = ALPN_PROTO_HTTP1
# alpn_preference = ALPN_PROTO_H2
default_alpn = b'http/1.1'
# alpn_preference = b'h2'
if self.alpn_for_client_connection in options:
choice = bytes(self.alpn_for_client_connection)

5
libmproxy/proxy/root_context.py
View File

@ -6,7 +6,6 @@ import six
from libmproxy.exceptions import ProtocolException
from netlib.exceptions import TcpException
from netlib.http import ALPN_PROTO_H2, ALPN_PROTO_HTTP1
from ..protocol import (
RawTCPLayer, TlsLayer, Http1Layer, Http2Layer, is_tls_record_magic, ServerConnectionMixin,
UpstreamConnectLayer
@ -85,9 +84,9 @@ class RootContext(object):
# 5. Check for TLS ALPN (HTTP1/HTTP2)
if isinstance(top_layer, TlsLayer):
alpn = top_layer.client_conn.get_alpn_proto_negotiated()
if alpn == ALPN_PROTO_H2:
if alpn == b'h2':
return Http2Layer(top_layer, 'transparent')
if alpn == ALPN_PROTO_HTTP1:
if alpn == b'http/1.1':
return Http1Layer(top_layer, 'transparent')
# 6. Check for raw tcp mode

2
libmproxy/web/app.py
View File

@ -29,7 +29,7 @@ class RequestHandler(tornado.web.RequestHandler):
def json(self):
if not self.request.headers.get("Content-Type").startswith("application/json"):
return None
return json.loads(self.request.body)
return json.loads(self.request.content)
@property
def state(self):

8
test/test_dump.py
View File

@ -38,13 +38,13 @@ def test_strfuncs():
flow.request.stickycookie = True
flow.client_conn = mock.MagicMock()
flow.client_conn.address.host = "foo"
flow.response = netlib.tutils.tresp(body=CONTENT_MISSING)
flow.response = netlib.tutils.tresp(content=CONTENT_MISSING)
flow.response.is_replay = True
flow.response.status_code = 300
m.echo_flow(flow)
flow = tutils.tflow(resp=netlib.tutils.tresp(body="{"))
flow = tutils.tflow(resp=netlib.tutils.tresp(content="{"))
flow.response.headers["content-type"] = "application/json"
flow.response.status_code = 400
m.echo_flow(flow)
@ -62,14 +62,14 @@ def test_contentview(get_content_view):
class TestDumpMaster:
def _cycle(self, m, content):
f = tutils.tflow(req=netlib.tutils.treq(body=content))
f = tutils.tflow(req=netlib.tutils.treq(content=content))
l = Log("connect")
l.reply = mock.MagicMock()
m.handle_log(l)
m.handle_clientconnect(f.client_conn)
m.handle_serverconnect(f.server_conn)
m.handle_request(f)
f.response = HTTPResponse.wrap(netlib.tutils.tresp(body=content))
f.response = HTTPResponse.wrap(netlib.tutils.tresp(content=content))
f = m.handle_response(f)
m.handle_clientdisconnect(f.client_conn)
return f

99
test/test_flow.py
View File

@ -8,8 +8,8 @@ import mock
import netlib.utils
from netlib import odict
from netlib.http import CONTENT_MISSING, HDR_FORM_URLENCODED, Headers
from libmproxy import filt, protocol, controller, tnetstring, flow
from netlib.http import CONTENT_MISSING, Headers
from libmproxy import filt, controller, tnetstring, flow
from libmproxy.models import Error, Flow, HTTPRequest, HTTPResponse, HTTPFlow, decoded
from libmproxy.proxy.config import HostMatcher
from libmproxy.proxy import ProxyConfig
@ -338,7 +338,7 @@ class TestServerPlaybackState:
assert s._hash(r) == s._hash(r2)
class TestFlow:
class TestFlow(object):
def test_copy(self):
f = tutils.tflow(resp=True)
a0 = f.get_state()
@ -849,7 +849,7 @@ class TestFlowMaster:
s = flow.State()
f = tutils.tflow()
f.response = HTTPResponse.wrap(netlib.tutils.tresp(body=f.request))
f.response = HTTPResponse.wrap(netlib.tutils.tresp(content=f.request))
pb = [f]
fm = flow.FlowMaster(None, s)
@ -903,7 +903,7 @@ class TestFlowMaster:
def test_server_playback_kill(self):
s = flow.State()
f = tutils.tflow()
f.response = HTTPResponse.wrap(netlib.tutils.tresp(body=f.request))
f.response = HTTPResponse.wrap(netlib.tutils.tresp(content=f.request))
pb = [f]
fm = flow.FlowMaster(None, s)
fm.refresh_server_playback = True
@ -1017,10 +1017,10 @@ class TestRequest:
r.port = 22
assert r.url == "https://address:22/path"
assert r.pretty_url(True) == "https://address:22/path"
assert r.pretty_url == "https://address:22/path"
r.headers["Host"] = "foo.com"
assert r.pretty_url(False) == "https://address:22/path"
assert r.pretty_url(True) == "https://foo.com:22/path"
assert r.url == "https://address:22/path"
assert r.pretty_url == "https://foo.com:22/path"
def test_path_components(self):
r = HTTPRequest.wrap(netlib.tutils.treq())
@ -1043,8 +1043,8 @@ class TestRequest:
def test_getset_form_urlencoded(self):
d = odict.ODict([("one", "two"), ("three", "four")])
r = HTTPRequest.wrap(netlib.tutils.treq(body=netlib.utils.urlencode(d.lst)))
r.headers["content-type"] = HDR_FORM_URLENCODED
r = HTTPRequest.wrap(netlib.tutils.treq(content=netlib.utils.urlencode(d.lst)))
r.headers["content-type"] = "application/x-www-form-urlencoded"
assert r.get_form_urlencoded() == d
d = odict.ODict([("x", "y")])
@ -1105,35 +1105,6 @@ class TestRequest:
r.constrain_encoding()
assert "oink" not in r.headers["accept-encoding"]
def test_decodeencode(self):
r = HTTPRequest.wrap(netlib.tutils.treq())
r.headers["content-encoding"] = "identity"
r.content = "falafel"
r.decode()
assert "content-encoding" not in r.headers
assert r.content == "falafel"
r = HTTPRequest.wrap(netlib.tutils.treq())
r.content = "falafel"
assert not r.decode()
r = HTTPRequest.wrap(netlib.tutils.treq())
r.headers["content-encoding"] = "identity"
r.content = "falafel"
r.encode("identity")
assert r.headers["content-encoding"] == "identity"
assert r.content == "falafel"
r = HTTPRequest.wrap(netlib.tutils.treq())
r.headers["content-encoding"] = "identity"
r.content = "falafel"
r.encode("gzip")
assert r.headers["content-encoding"] == "gzip"
assert r.content != "falafel"
r.decode()
assert "content-encoding" not in r.headers
assert r.content == "falafel"
def test_get_decoded_content(self):
r = HTTPRequest.wrap(netlib.tutils.treq())
r.content = None
@ -1192,35 +1163,6 @@ class TestResponse:
assert not "foo" in r.content
assert r.headers["boo"] == "boo"
def test_decodeencode(self):
r = HTTPResponse.wrap(netlib.tutils.tresp())
r.headers["content-encoding"] = "identity"
r.content = "falafel"
assert r.decode()
assert "content-encoding" not in r.headers
assert r.content == "falafel"
r = HTTPResponse.wrap(netlib.tutils.tresp())
r.headers["content-encoding"] = "identity"
r.content = "falafel"
r.encode("identity")
assert r.headers["content-encoding"] == "identity"
assert r.content == "falafel"
r = HTTPResponse.wrap(netlib.tutils.tresp())
r.headers["content-encoding"] = "identity"
r.content = "falafel"
r.encode("gzip")
assert r.headers["content-encoding"] == "gzip"
assert r.content != "falafel"
assert r.decode()
assert "content-encoding" not in r.headers
assert r.content == "falafel"
r.headers["content-encoding"] = "gzip"
assert not r.decode()
assert r.content == "falafel"
def test_get_content_type(self):
resp = HTTPResponse.wrap(netlib.tutils.tresp())
resp.headers = Headers(content_type="text/plain")
@ -1265,27 +1207,6 @@ class TestClientConnection:
assert str(c)
def test_decoded():
r = HTTPRequest.wrap(netlib.tutils.treq())
assert r.content == "content"
assert "content-encoding" not in r.headers
r.encode("gzip")
assert r.headers["content-encoding"]
assert r.content != "content"
with decoded(r):
assert "content-encoding" not in r.headers
assert r.content == "content"
assert r.headers["content-encoding"]
assert r.content != "content"
with decoded(r):
r.content = "foo"
assert r.content != "foo"
r.decode()
assert r.content == "foo"
def test_replacehooks():
h = flow.ReplaceHooks()
h.add("~q", "foo", "bar")

6
test/test_protocol_http.py
View File

@ -48,14 +48,14 @@ class TestInvalidRequests(tservers.HTTPProxTest):
p = self.pathoc()
r = p.request("connect:'%s:%s'" % ("127.0.0.1", self.server2.port))
assert r.status_code == 400
assert "Invalid HTTP request form" in r.body
assert "Invalid HTTP request form" in r.content
def test_relative_request(self):
p = self.pathoc_raw()
p.connect()
r = p.request("get:/p/200")
assert r.status_code == 400
assert "Invalid HTTP request form" in r.body
assert "Invalid HTTP request form" in r.content
class TestExpectHeader(tservers.HTTPProxTest):
@ -81,4 +81,4 @@ class TestExpectHeader(tservers.HTTPProxTest):
resp = http1.read_response(client.rfile, treq())
assert resp.status_code == 200
client.finish()
client.finish()

6
test/test_server.py
View File

@ -235,7 +235,8 @@ class TestHTTP(tservers.HTTPProxTest, CommonMixin, AppMixin):
# There's a race here, which means we can get any of a number of errors.
# Rather than introduce yet another sleep into the test suite, we just
# relax the Exception specification.
tutils.raises(Exception, p.request, "get:'%s'" % response)
with raises(Exception):
p.request("get:'%s'" % response)
def test_reconnect(self):
req = "get:'%s/p/200:b@1:da'" % self.server.urlbase
@ -573,7 +574,6 @@ class TestProxy(tservers.HTTPProxTest):
recvd += len(connection.recv(5000))
connection.close()
print(self.master.state.view._list)
first_flow = self.master.state.view[0]
second_flow = self.master.state.view[1]
assert first_flow.server_conn.timestamp_tcp_setup
@ -616,7 +616,7 @@ class MasterRedirectRequest(tservers.TestMaster):
super(MasterRedirectRequest, self).handle_request(f)
def handle_response(self, f):
f.response.body = str(f.client_conn.address.port)
f.response.content = str(f.client_conn.address.port)
f.response.headers["server-conn-id"] = str(f.server_conn.source_address.port)
super(MasterRedirectRequest, self).handle_response(f)