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

adjust to new netlib Headers class

Browse Source
This commit is contained in:
Maximilian Hils 2015-09-05 20:45:58 +02:00
parent 00561d280c
commit 5125c669cc
22 changed files with 224 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/add_header.py
View File

@ -1,2 +1,2 @@
def response(context, flow):
flow.response.headers["newheader"] = ["foo"]
flow.response.headers["newheader"] = "foo"

10
examples/har_extractor.py
View File

@ -147,8 +147,8 @@ def response(context, flow):
response_body_size = len(flow.response.content)
response_body_decoded_size = len(flow.response.get_decoded_content())
response_body_compression = response_body_decoded_size - response_body_size
response_mime_type = flow.response.headers.get_first('Content-Type', '')
response_redirect_url = flow.response.headers.get_first('Location', '')
response_mime_type = flow.response.headers.get('Content-Type', '')
response_redirect_url = flow.response.headers.get('Location', '')
entry = HAR.entries(
{
@ -201,12 +201,12 @@ def response(context, flow):
# Lookup the referer in the page_ref of context.HARLog to point this entries
# pageref attribute to the right pages object, then set it as a new
# reference to build a reference tree.
elif context.HARLog.get_page_ref(flow.request.headers.get('Referer', (None, ))[0]) is not None:
elif context.HARLog.get_page_ref(flow.request.headers.get('Referer')) is not None:
entry['pageref'] = context.HARLog.get_page_ref(
flow.request.headers['Referer'][0]
flow.request.headers['Referer']
)
context.HARLog.set_page_ref(
flow.request.headers['Referer'][0], entry['pageref']
flow.request.headers['Referer'], entry['pageref']
)
context.HARLog.add(entry)

2
examples/modify_form.py
View File

@ -1,5 +1,5 @@
def request(context, flow):
if "application/x-www-form-urlencoded" in flow.request.headers["content-type"]:
if "application/x-www-form-urlencoded" in flow.request.headers.get("content-type", ""):
form = flow.request.get_form_urlencoded()
form["mitmproxy"] = ["rocks"]
flow.request.set_form_urlencoded(form)

5
examples/redirect_requests.py
View File

@ -2,8 +2,7 @@
This example shows two ways to redirect flows to other destinations.
"""
from libmproxy.models import HTTPResponse
from netlib.odict import ODictCaseless
from netlib.http import Headers
def request(context, flow):
# pretty_host(hostheader=True) takes the Host: header of the request into account,
@ -14,7 +13,7 @@ def request(context, flow):
if flow.request.pretty_host(hostheader=True).endswith("example.com"):
resp = HTTPResponse(
[1, 1], 200, "OK",
ODictCaseless([["Content-Type", "text/html"]]),
Headers(Content_Type="text/html"),
"helloworld")
flow.reply(resp)

10
examples/stickycookies
View File

@ -23,16 +23,16 @@ class StickyMaster(controller.Master):
def handle_request(self, flow):
hid = (flow.request.host, flow.request.port)
if flow.request.headers["cookie"]:
self.stickyhosts[hid] = flow.request.headers["cookie"]
if "cookie" in flow.request.headers:
self.stickyhosts[hid] = flow.request.headers.get_all("cookie")
elif hid in self.stickyhosts:
flow.request.headers["cookie"] = self.stickyhosts[hid]
flow.request.headers.set_all("cookie", self.stickyhosts[hid])
flow.reply()
def handle_response(self, flow):
hid = (flow.request.host, flow.request.port)
if flow.response.headers["set-cookie"]:
self.stickyhosts[hid] = flow.response.headers["set-cookie"]
if "set-cookie" in flow.response.headers:
self.stickyhosts[hid] = flow.response.headers.get_all("set-cookie")
flow.reply()

4
examples/upsidedownternet.py
View File

@ -4,7 +4,7 @@ from libmproxy.models import decoded
def response(context, flow):
if flow.response.headers.get_first("content-type", "").startswith("image"):
if flow.response.headers.get("content-type", "").startswith("image"):
with decoded(flow.response): # automatically decode gzipped responses.
try:
s = cStringIO.StringIO(flow.response.content)
@ -12,6 +12,6 @@ def response(context, flow):
s2 = cStringIO.StringIO()
img.save(s2, "png")
flow.response.content = s2.getvalue()
flow.response.headers["content-type"] = ["image/png"]
flow.response.headers["content-type"] = "image/png"
except: # Unknown image types etc.
pass

4
libmproxy/console/common.py
View File

@ -415,9 +415,9 @@ def format_flow(f, focus, extended=False, hostheader=False, padding=2,
resp_clen = contentdesc,
roundtrip = roundtrip,
))
t = f.response.headers["content-type"]
t = f.response.headers.get("content-type")
if t:
d["resp_ctype"] = t[0].split(";")[0]
d["resp_ctype"] = t.split(";")[0]
else:
d["resp_ctype"] = ""
return flowcache.get(

14
libmproxy/console/contentview.py
View File

@ -12,7 +12,7 @@ import urwid
import html2text
import netlib.utils
from netlib import odict, encoding
from netlib import encoding
from . import common, signals
from .. import utils
@ -74,7 +74,7 @@ class ViewAuto:
content_types = []
def __call__(self, hdrs, content, limit):
ctype = hdrs.get_first("content-type")
ctype = hdrs.get("content-type")
if ctype:
ct = netlib.utils.parse_content_type(ctype) if ctype else None
ct = "%s/%s" % (ct[0], ct[1])
@ -508,7 +508,7 @@ def get(name):
return i
def get_content_view(viewmode, hdrItems, content, limit, is_request):
def get_content_view(viewmode, headers, content, limit, is_request):
"""
Returns a (msg, body) tuple.
"""
@ -519,16 +519,14 @@ def get_content_view(viewmode, hdrItems, content, limit, is_request):
return "No content", ""
msg = []
hdrs = odict.ODictCaseless([list(i) for i in hdrItems])
enc = hdrs.get_first("content-encoding")
enc = headers.get("content-encoding")
if enc and enc != "identity":
decoded = encoding.decode(enc, content)
if decoded:
content = decoded
msg.append("[decoded %s]" % enc)
try:
ret = viewmode(hdrs, content, limit)
ret = viewmode(headers, content, limit)
# Third-party viewers can fail in unexpected ways...
except Exception:
s = traceback.format_exc()
@ -536,7 +534,7 @@ def get_content_view(viewmode, hdrItems, content, limit, is_request):
signals.add_event(s, "error")
ret = None
if not ret:
ret = get("Raw")(hdrs, content, limit)
ret = get("Raw")(headers, content, limit)
msg.append("Couldn't parse: falling back to Raw")
else:
msg.append(ret[0])

19
libmproxy/console/flowview.py
View File

@ -4,7 +4,7 @@ import sys
import urwid
from netlib import odict
from netlib.http.semantics import CONTENT_MISSING
from netlib.http.semantics import CONTENT_MISSING, Headers
from . import common, grideditor, contentview, signals, searchable, tabs
from . import flowdetailview
@ -182,7 +182,7 @@ class FlowView(tabs.Tabs):
description, text_objects = cache.get(
contentview.get_content_view,
viewmode,
tuple(tuple(i) for i in conn.headers.lst),
conn.headers,
conn.content,
limit,
isinstance(conn, HTTPRequest)
@ -199,7 +199,7 @@ class FlowView(tabs.Tabs):
def conn_text(self, conn):
if conn:
txt = common.format_keyvals(
[(h + ":", v) for (h, v) in conn.headers.lst],
[(h + ":", v) for (h, v) in conn.headers.fields],
key = "header",
val = "text"
)
@ -284,8 +284,8 @@ class FlowView(tabs.Tabs):
response.msg = msg
signals.flow_change.send(self, flow = self.flow)
def set_headers(self, lst, conn):
conn.headers = odict.ODictCaseless(lst)
def set_headers(self, fields, conn):
conn.headers = Headers(fields)
signals.flow_change.send(self, flow = self.flow)
def set_query(self, lst, conn):
@ -330,7 +330,7 @@ class FlowView(tabs.Tabs):
if not self.flow.response:
self.flow.response = HTTPResponse(
self.flow.request.httpversion,
200, "OK", odict.ODictCaseless(), ""
200, "OK", Headers(), ""
)
self.flow.response.reply = controller.DummyReply()
message = self.flow.response
@ -381,7 +381,7 @@ class FlowView(tabs.Tabs):
self.master.view_grideditor(
grideditor.HeaderEditor(
self.master,
message.headers.lst,
message.headers.fields,
self.set_headers,
message
)
@ -616,8 +616,7 @@ class FlowView(tabs.Tabs):
key = None
elif key == "v":
if conn.content:
t = conn.headers["content-type"] or [None]
t = t[0]
t = conn.headers.get("content-type")
if "EDITOR" in os.environ or "PAGER" in os.environ:
self.master.spawn_external_viewer(conn.content, t)
else:
@ -626,7 +625,7 @@ class FlowView(tabs.Tabs):
)
elif key == "z":
self.flow.backup()
e = conn.headers.get_first("content-encoding", "identity")
e = conn.headers.get("content-encoding", "identity")
if e != "identity":
if not conn.decode():
signals.status_message.send(

2
libmproxy/dump.py
View File

@ -174,7 +174,7 @@ class DumpMaster(flow.FlowMaster):
def _print_message(self, message):
if self.o.flow_detail >= 2:
print(self.indent(4, message.headers.format()), file=self.outfile)
print(self.indent(4, str(message.headers)), file=self.outfile)
if self.o.flow_detail >= 3:
if message.content == CONTENT_MISSING:
print(self.indent(4, "(content missing)"), file=self.outfile)

42
libmproxy/filt.py
View File

@ -77,17 +77,19 @@ class FResp(_Action):
class _Rex(_Action):
flags = 0
def __init__(self, expr):
self.expr = expr
try:
self.re = re.compile(self.expr)
self.re = re.compile(self.expr, self.flags)
except:
raise ValueError("Cannot compile expression.")
def _check_content_type(expr, o):
val = o.headers["content-type"]
if val and re.search(expr, val[0]):
val = o.headers.get("content-type")
if val and re.search(expr, val):
return True
return False
@ -145,11 +147,12 @@ class FResponseContentType(_Rex):
class FHead(_Rex):
code = "h"
help = "Header"
flags = re.MULTILINE
def __call__(self, f):
if f.request.headers.match_re(self.expr):
if f.request and self.re.search(str(f.request.headers)):
return True
elif f.response and f.response.headers.match_re(self.expr):
if f.response and self.re.search(str(f.response.headers)):
return True
return False
@ -157,18 +160,20 @@ class FHead(_Rex):
class FHeadRequest(_Rex):
code = "hq"
help = "Request header"
flags = re.MULTILINE
def __call__(self, f):
if f.request.headers.match_re(self.expr):
if f.request and self.re.search(str(f.request.headers)):
return True
class FHeadResponse(_Rex):
code = "hs"
help = "Response header"
flags = re.MULTILINE
def __call__(self, f):
if f.response and f.response.headers.match_re(self.expr):
if f.response and self.re.search(str(f.response.headers)):
return True
@ -178,10 +183,10 @@ class FBod(_Rex):
def __call__(self, f):
if f.request and f.request.content:
if re.search(self.expr, f.request.get_decoded_content()):
if self.re.search(f.request.get_decoded_content()):
return True
if f.response and f.response.content:
if re.search(self.expr, f.response.get_decoded_content()):
if self.re.search(f.response.get_decoded_content()):
return True
return False
@ -192,7 +197,7 @@ class FBodRequest(_Rex):
def __call__(self, f):
if f.request and f.request.content:
if re.search(self.expr, f.request.get_decoded_content()):
if self.re.search(f.request.get_decoded_content()):
return True
@ -202,24 +207,26 @@ class FBodResponse(_Rex):
def __call__(self, f):
if f.response and f.response.content:
if re.search(self.expr, f.response.get_decoded_content()):
if self.re.search(f.response.get_decoded_content()):
return True
class FMethod(_Rex):
code = "m"
help = "Method"
flags = re.IGNORECASE
def __call__(self, f):
return bool(re.search(self.expr, f.request.method, re.IGNORECASE))
return bool(self.re.search(f.request.method))
class FDomain(_Rex):
code = "d"
help = "Domain"
flags = re.IGNORECASE
def __call__(self, f):
return bool(re.search(self.expr, f.request.host, re.IGNORECASE))
return bool(self.re.search(f.request.host))
class FUrl(_Rex):
@ -234,21 +241,24 @@ class FUrl(_Rex):
return klass(*toks)
def __call__(self, f):
return re.search(self.expr, f.request.url)
return self.re.search(f.request.url)
class FSrc(_Rex):
code = "src"
help = "Match source address"
def __call__(self, f):
return f.client_conn.address and re.search(self.expr, repr(f.client_conn.address))
return f.client_conn.address and self.re.search(repr(f.client_conn.address))
class FDst(_Rex):
code = "dst"
help = "Match destination address"
def __call__(self, f):
return f.server_conn.address and re.search(self.expr, repr(f.server_conn.address))
return f.server_conn.address and self.re.search(repr(f.server_conn.address))
class _Int(_Action):
def __init__(self, num):

34
libmproxy/flow.py
View File

@ -11,8 +11,8 @@ import re
import urlparse
from netlib import odict, wsgi
from netlib.http.semantics import CONTENT_MISSING
from netlib import wsgi
from netlib.http.semantics import CONTENT_MISSING, Headers
import netlib.http
from . import controller, tnetstring, filt, script, version
from .onboarding import app
@ -45,7 +45,7 @@ class AppRegistry:
if (request.host, request.port) in self.apps:
return self.apps[(request.host, request.port)]
if "host" in request.headers:
host = request.headers["host"][0]
host = request.headers["host"]
return self.apps.get((host, request.port), None)
@ -144,15 +144,15 @@ class SetHeaders:
for _, header, value, cpatt in self.lst:
if cpatt(f):
if f.response:
del f.response.headers[header]
f.response.headers.pop(header, None)
else:
del f.request.headers[header]
f.request.headers.pop(header, None)
for _, header, value, cpatt in self.lst:
if cpatt(f):
if f.response:
f.response.headers.add(header, value)
f.response.headers.fields.append((header, value))
else:
f.request.headers.add(header, value)
f.request.headers.fields.append((header, value))
class StreamLargeBodies(object):
@ -278,14 +278,11 @@ class ServerPlaybackState:
key.append(p[1])
if self.headers:
hdrs = []
headers = []
for i in self.headers:
v = r.headers[i]
# Slightly subtle: we need to convert everything to strings
# to prevent a mismatch between unicode/non-unicode.
v = [str(x) for x in v]
hdrs.append((i, v))
key.append(hdrs)
v = r.headers.get(i)
headers.append((i, v))
key.append(headers)
return hashlib.sha256(repr(key)).digest()
def next_flow(self, request):
@ -329,7 +326,7 @@ class StickyCookieState:
return False
def handle_response(self, f):
for i in f.response.headers["set-cookie"]:
for i in f.response.headers.get_all("set-cookie"):
# FIXME: We now know that Cookie.py screws up some cookies with
# valid RFC 822/1123 datetime specifications for expiry. Sigh.
c = Cookie.SimpleCookie(str(i))
@ -351,7 +348,7 @@ class StickyCookieState:
l.append(self.jar[i].output(header="").strip())
if l:
f.request.stickycookie = True
f.request.headers["cookie"] = l
f.request.headers.set_all("cookie",l)
class StickyAuthState:
@ -836,7 +833,7 @@ class FlowMaster(controller.Master):
ssl_established=True
))
f = HTTPFlow(c, s)
headers = odict.ODictCaseless()
headers = Headers()
req = HTTPRequest(
"absolute",
@ -930,8 +927,7 @@ class FlowMaster(controller.Master):
f.backup()
f.request.is_replay = True
if f.request.content:
f.request.headers[
"Content-Length"] = [str(len(f.request.content))]
f.request.headers["Content-Length"] = str(len(f.request.content))
f.response = None
f.error = None
self.process_new_request(f)

62
libmproxy/models/http.py
View File

@ -5,8 +5,8 @@ from email.utils import parsedate_tz, formatdate, mktime_tz
import time
from libmproxy import utils
from netlib import odict, encoding
from netlib.http import status_codes
from netlib import encoding
from netlib.http import status_codes, Headers
from netlib.tcp import Address
from netlib.http.semantics import Request, Response, CONTENT_MISSING
from .. import version, stateobject
@ -16,7 +16,7 @@ from .flow import Flow
class MessageMixin(stateobject.StateObject):
_stateobject_attributes = dict(
httpversion=tuple,
headers=odict.ODictCaseless,
headers=Headers,
body=str,
timestamp_start=float,
timestamp_end=float
@ -40,7 +40,7 @@ class MessageMixin(stateobject.StateObject):
header.
Doesn't change the message iteself or its headers.
"""
ce = self.headers.get_first("content-encoding")
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)
@ -53,14 +53,14 @@ class MessageMixin(stateobject.StateObject):
Returns True if decoding succeeded, False otherwise.
"""
ce = self.headers.get_first("content-encoding")
ce = self.headers.get("content-encoding")
if not self.body or ce not in encoding.ENCODINGS:
return False
data = encoding.decode(ce, self.body)
if data is None:
return False
self.body = data
del self.headers["content-encoding"]
self.headers.pop("content-encoding", None)
return True
def encode(self, e):
@ -70,7 +70,7 @@ class MessageMixin(stateobject.StateObject):
"""
# FIXME: Error if there's an existing encoding header?
self.body = encoding.encode(e, self.body)
self.headers["content-encoding"] = [e]
self.headers["content-encoding"] = e
def copy(self):
c = copy.copy(self)
@ -86,11 +86,18 @@ class MessageMixin(stateobject.StateObject):
Returns the number of replacements made.
"""
with decoded(self):
self.body, c = utils.safe_subn(
self.body, count = utils.safe_subn(
pattern, repl, self.body, *args, **kwargs
)
c += self.headers.replace(pattern, repl, *args, **kwargs)
return c
fields = []
for name, value in self.headers.fields:
name, c = utils.safe_subn(pattern, repl, name, *args, **kwargs)
count += c
value, c = utils.safe_subn(pattern, repl, value, *args, **kwargs)
count += c
fields.append([name, value])
self.headers.fields = fields
return count
class HTTPRequest(MessageMixin, Request):
@ -115,7 +122,7 @@ class HTTPRequest(MessageMixin, Request):
httpversion: HTTP version tuple, e.g. (1,1)
headers: odict.ODictCaseless object
headers: Headers object
content: Content of the request, None, or CONTENT_MISSING if there
is content associated, but not present. CONTENT_MISSING evaluates
@ -266,7 +273,7 @@ class HTTPResponse(MessageMixin, Response):
msg: HTTP response message
headers: ODict Caseless object
headers: Headers object
content: Content of the request, None, or CONTENT_MISSING if there
is content associated, but not present. CONTENT_MISSING evaluates
@ -379,15 +386,15 @@ class HTTPResponse(MessageMixin, Response):
]
for i in refresh_headers:
if i in self.headers:
d = parsedate_tz(self.headers[i][0])
d = parsedate_tz(self.headers[i])
if d:
new = mktime_tz(d) + delta
self.headers[i] = [formatdate(new)]
self.headers[i] = formatdate(new)
c = []
for i in self.headers["set-cookie"]:
for i in self.headers.get_all("set-cookie"):
c.append(self._refresh_cookie(i, delta))
if c:
self.headers["set-cookie"] = c
self.headers.set_all("set-cookie", c)
class HTTPFlow(Flow):
@ -490,7 +497,7 @@ class decoded(object):
def __init__(self, o):
self.o = o
ce = o.headers.get_first("content-encoding")
ce = o.headers.get("content-encoding")
if ce in encoding.ENCODINGS:
self.ce = ce
else:
@ -517,11 +524,12 @@ def make_error_response(status_code, message, headers=None):
""".strip() % (status_code, response, message)
if not headers:
headers = odict.ODictCaseless()
headers["Server"] = [version.NAMEVERSION]
headers["Connection"] = ["close"]
headers["Content-Length"] = [len(body)]
headers["Content-Type"] = ["text/html"]
headers = Headers(
Server=version.NAMEVERSION,
Connection="close",
Content_Length=str(len(body)),
Content_Type="text/html"
)
return HTTPResponse(
(1, 1), # FIXME: Should be a string.
@ -536,15 +544,15 @@ def make_connect_request(address):
address = Address.wrap(address)
return HTTPRequest(
"authority", "CONNECT", None, address.host, address.port, None, (1, 1),
odict.ODictCaseless(), ""
Headers(), ""
)
def make_connect_response(httpversion):
headers = odict.ODictCaseless([
["Content-Length", "0"],
["Proxy-Agent", version.NAMEVERSION]
])
headers = Headers(
Content_Length="0",
Proxy_Agent=version.NAMEVERSION
)
return HTTPResponse(
httpversion,
200,

8
libmproxy/protocol/http.py
View File

@ -1,7 +1,7 @@
from __future__ import (absolute_import, print_function, division)
from netlib import tcp
from netlib.http import http1, HttpErrorConnClosed, HttpError
from netlib.http import http1, HttpErrorConnClosed, HttpError, Headers
from netlib.http.semantics import CONTENT_MISSING
from netlib import odict
from netlib.tcp import NetLibError, Address
@ -568,10 +568,6 @@ class HttpLayer(Layer):
self.send_response(make_error_response(
407,
"Proxy Authentication Required",
odict.ODictCaseless(
[
[k, v] for k, v in
self.config.authenticator.auth_challenge_headers().items()
])
Headers(**self.config.authenticator.auth_challenge_headers())
))
raise InvalidCredentials("Proxy Authentication Required")

7
libmproxy/web/app.py
View File

@ -27,8 +27,7 @@ class RequestHandler(tornado.web.RequestHandler):
@property
def json(self):
if not self.request.headers.get(
"Content-Type").startswith("application/json"):
if not self.request.headers.get("Content-Type").startswith("application/json"):
return None
return json.loads(self.request.body)
@ -186,12 +185,12 @@ class FlowContent(RequestHandler):
if not message.content:
raise APIError(400, "No content.")
content_encoding = message.headers.get_first("Content-Encoding", None)
content_encoding = message.headers.get("Content-Encoding", None)
if content_encoding:
content_encoding = re.sub(r"[^\w]", "", content_encoding)
self.set_header("Content-Encoding", content_encoding)
original_cd = message.headers.get_first("Content-Disposition", None)
original_cd = message.headers.get("Content-Disposition", None)
filename = None
if original_cd:
filename = re.search("filename=([\w\" \.\-\(\)]+)", original_cd)

56
test/test_console_contentview.py
View File

@ -1,11 +1,13 @@
import os
from nose.plugins.skip import SkipTest
from netlib.http import Headers
if os.name == "nt":
raise SkipTest("Skipped on Windows.")
import sys
import netlib.utils
from netlib import odict, encoding
from netlib import encoding
import libmproxy.console.contentview as cv
from libmproxy import utils, flow
@ -33,34 +35,28 @@ class TestContentView:
def test_view_auto(self):
v = cv.ViewAuto()
f = v(
odict.ODictCaseless(),
Headers(),
"foo",
1000
)
assert f[0] == "Raw"
f = v(
odict.ODictCaseless(
[["content-type", "text/html"]],
),
Headers(content_type="text/html"),
"<html></html>",
1000
)
assert f[0] == "HTML"
f = v(
odict.ODictCaseless(
[["content-type", "text/flibble"]],
),
Headers(content_type="text/flibble"),
"foo",
1000
)
assert f[0] == "Raw"
f = v(
odict.ODictCaseless(
[["content-type", "text/flibble"]],
),
Headers(content_type="text/flibble"),
"<xml></xml>",
1000
)
@ -168,28 +164,22 @@ Content-Disposition: form-data; name="submit-name"
Larry
--AaB03x
""".strip()
h = odict.ODictCaseless(
[("Content-Type", "multipart/form-data; boundary=AaB03x")]
)
h = Headers(content_type="multipart/form-data; boundary=AaB03x"),
assert view(h, v, 1000)
h = odict.ODictCaseless()
h = Headers(),
assert not view(h, v, 1000)
h = odict.ODictCaseless(
[("Content-Type", "multipart/form-data")]
)
h = Headers(content_type="multipart/form-data"),
assert not view(h, v, 1000)
h = odict.ODictCaseless(
[("Content-Type", "unparseable")]
)
h = Headers(content_type="unparseable"),
assert not view(h, v, 1000)
def test_get_content_view(self):
r = cv.get_content_view(
cv.get("Raw"),
[["content-type", "application/json"]],
Headers(content_type="application/json"),
"[1, 2, 3]",
1000,
False
@ -198,7 +188,7 @@ Larry
r = cv.get_content_view(
cv.get("Auto"),
[["content-type", "application/json"]],
Headers(content_type="application/json"),
"[1, 2, 3]",
1000,
False
@ -207,7 +197,7 @@ Larry
r = cv.get_content_view(
cv.get("Auto"),
[["content-type", "application/json"]],
Headers(content_type="application/json"),
"[1, 2",
1000,
False
@ -216,7 +206,7 @@ Larry
r = cv.get_content_view(
cv.get("AMF"),
[],
Headers(),
"[1, 2",
1000,
False
@ -225,10 +215,10 @@ Larry
r = cv.get_content_view(
cv.get("Auto"),
[
["content-type", "application/json"],
["content-encoding", "gzip"]
],
Headers(
content_type="application/json",
content_encoding="gzip"
),
encoding.encode('gzip', "[1, 2, 3]"),
1000,
False
@ -238,10 +228,10 @@ Larry
r = cv.get_content_view(
cv.get("XML"),
[
["content-type", "application/json"],
["content-encoding", "gzip"]
],
Headers(
content_type="application/json",
content_encoding="gzip"
),
encoding.encode('gzip', "[1, 2, 3]"),
1000,
False

2
test/test_dump.py
View File

@ -145,7 +145,7 @@ class TestDumpMaster:
o = dump.Options(setheaders=[(".*", "one", "two")])
m = dump.DumpMaster(None, o, outfile=cs)
f = self._cycle(m, "content")
assert f.request.headers["one"] == ["two"]
assert f.request.headers["one"] == "two"
def test_basic(self):
for i in (1, 2, 3):

16
test/test_filt.py
View File

@ -1,8 +1,8 @@
import cStringIO
from netlib import odict
from libmproxy import filt, flow
from libmproxy.protocol import http
from libmproxy.models import Error
from netlib.http import Headers
import tutils
@ -76,8 +76,7 @@ class TestParsing:
class TestMatching:
def req(self):
headers = odict.ODictCaseless()
headers["header"] = ["qvalue"]
headers = Headers(header="qvalue")
req = http.HTTPRequest(
"absolute",
"GET",
@ -98,8 +97,7 @@ class TestMatching:
def resp(self):
f = self.req()
headers = odict.ODictCaseless()
headers["header_response"] = ["svalue"]
headers = Headers([["header_response", "svalue"]])
f.response = http.HTTPResponse(
(1,
1),
@ -123,7 +121,7 @@ class TestMatching:
def test_asset(self):
s = self.resp()
assert not self.q("~a", s)
s.response.headers["content-type"] = ["text/javascript"]
s.response.headers["content-type"] = "text/javascript"
assert self.q("~a", s)
def test_fcontenttype(self):
@ -132,16 +130,16 @@ class TestMatching:
assert not self.q("~t content", q)
assert not self.q("~t content", s)
q.request.headers["content-type"] = ["text/json"]
q.request.headers["content-type"] = "text/json"
assert self.q("~t json", q)
assert self.q("~tq json", q)
assert not self.q("~ts json", q)
s.response.headers["content-type"] = ["text/json"]
s.response.headers["content-type"] = "text/json"
assert self.q("~t json", s)
del s.response.headers["content-type"]
s.request.headers["content-type"] = ["text/json"]
s.request.headers["content-type"] = "text/json"
assert self.q("~t json", s)
assert self.q("~tq json", s)
assert not self.q("~ts json", s)

158
test/test_flow.py
View File

@ -8,7 +8,7 @@ import mock
import netlib.utils
from netlib import odict
from netlib.http.semantics import CONTENT_MISSING, HDR_FORM_URLENCODED
from netlib.http.semantics import CONTENT_MISSING, HDR_FORM_URLENCODED, Headers
from libmproxy import filt, protocol, controller, tnetstring, flow
from libmproxy.models import Error, Flow, HTTPRequest, HTTPResponse, HTTPFlow, decoded
from libmproxy.proxy.config import HostMatcher
@ -34,7 +34,7 @@ def test_app_registry():
r.host = "domain2"
r.port = 80
assert not ar.get(r)
r.headers["host"] = ["domain"]
r.headers["host"] = "domain"
assert ar.get(r)
@ -42,7 +42,7 @@ class TestStickyCookieState:
def _response(self, cookie, host):
s = flow.StickyCookieState(filt.parse(".*"))
f = tutils.tflow(req=netlib.tutils.treq(host=host, port=80), resp=True)
f.response.headers["Set-Cookie"] = [cookie]
f.response.headers["Set-Cookie"] = cookie
s.handle_response(f)
return s, f
@ -75,13 +75,13 @@ class TestStickyAuthState:
def test_handle_response(self):
s = flow.StickyAuthState(filt.parse(".*"))
f = tutils.tflow(resp=True)
f.request.headers["authorization"] = ["foo"]
f.request.headers["authorization"] = "foo"
s.handle_request(f)
assert "address" in s.hosts
f = tutils.tflow(resp=True)
s.handle_request(f)
assert f.request.headers["authorization"] == ["foo"]
assert f.request.headers["authorization"] == "foo"
class TestClientPlaybackState:
@ -133,7 +133,7 @@ class TestServerPlaybackState:
assert s._hash(r)
assert s._hash(r) == s._hash(r2)
r.request.headers["foo"] = ["bar"]
r.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r.request.path = "voing"
assert s._hash(r) != s._hash(r2)
@ -153,12 +153,12 @@ class TestServerPlaybackState:
None,
False)
r = tutils.tflow(resp=True)
r.request.headers["foo"] = ["bar"]
r.request.headers["foo"] = "bar"
r2 = tutils.tflow(resp=True)
assert not s._hash(r) == s._hash(r2)
r2.request.headers["foo"] = ["bar"]
r2.request.headers["foo"] = "bar"
assert s._hash(r) == s._hash(r2)
r2.request.headers["oink"] = ["bar"]
r2.request.headers["oink"] = "bar"
assert s._hash(r) == s._hash(r2)
r = tutils.tflow(resp=True)
@ -167,10 +167,10 @@ class TestServerPlaybackState:
def test_load(self):
r = tutils.tflow(resp=True)
r.request.headers["key"] = ["one"]
r.request.headers["key"] = "one"
r2 = tutils.tflow(resp=True)
r2.request.headers["key"] = ["two"]
r2.request.headers["key"] = "two"
s = flow.ServerPlaybackState(
None, [
@ -179,21 +179,21 @@ class TestServerPlaybackState:
assert len(s.fmap.keys()) == 1
n = s.next_flow(r)
assert n.request.headers["key"] == ["one"]
assert n.request.headers["key"] == "one"
assert s.count() == 1
n = s.next_flow(r)
assert n.request.headers["key"] == ["two"]
assert n.request.headers["key"] == "two"
assert s.count() == 0
assert not s.next_flow(r)
def test_load_with_nopop(self):
r = tutils.tflow(resp=True)
r.request.headers["key"] = ["one"]
r.request.headers["key"] = "one"
r2 = tutils.tflow(resp=True)
r2.request.headers["key"] = ["two"]
r2.request.headers["key"] = "two"
s = flow.ServerPlaybackState(
None, [
@ -224,12 +224,10 @@ class TestServerPlaybackState:
None, [], False, False, None, False, [
"param1", "param2"], False)
r = tutils.tflow(resp=True)
r.request.headers[
"Content-Type"] = ["application/x-www-form-urlencoded"]
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = "paramx=x&param1=1"
r2 = tutils.tflow(resp=True)
r2.request.headers[
"Content-Type"] = ["application/x-www-form-urlencoded"]
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = "paramx=x&param1=1"
# same parameters
assert s._hash(r) == s._hash(r2)
@ -254,10 +252,10 @@ class TestServerPlaybackState:
None, [], False, False, None, False, [
"param1", "param2"], False)
r = tutils.tflow(resp=True)
r.request.headers["Content-Type"] = ["application/json"]
r.request.headers["Content-Type"] = "application/json"
r.request.content = '{"param1":"1"}'
r2 = tutils.tflow(resp=True)
r2.request.headers["Content-Type"] = ["application/json"]
r2.request.headers["Content-Type"] = "application/json"
r2.request.content = '{"param1":"1"}'
# same content
assert s._hash(r) == s._hash(r2)
@ -271,12 +269,10 @@ class TestServerPlaybackState:
None, [], False, False, None, True, [
"param1", "param2"], False)
r = tutils.tflow(resp=True)
r.request.headers[
"Content-Type"] = ["application/x-www-form-urlencoded"]
r.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r.request.content = "paramx=y"
r2 = tutils.tflow(resp=True)
r2.request.headers[
"Content-Type"] = ["application/x-www-form-urlencoded"]
r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded"
r2.request.content = "paramx=x"
# same parameters
assert s._hash(r) == s._hash(r2)
@ -460,17 +456,17 @@ class TestFlow:
def test_replace(self):
f = tutils.tflow(resp=True)
f.request.headers["foo"] = ["foo"]
f.request.headers["foo"] = "foo"
f.request.content = "afoob"
f.response.headers["foo"] = ["foo"]
f.response.headers["foo"] = "foo"
f.response.content = "afoob"
assert f.replace("foo", "bar") == 6
assert f.request.headers["bar"] == ["bar"]
assert f.request.headers["bar"] == "bar"
assert f.request.content == "abarb"
assert f.response.headers["bar"] == ["bar"]
assert f.response.headers["bar"] == "bar"
assert f.response.content == "abarb"
def test_replace_encoded(self):
@ -938,14 +934,14 @@ class TestFlowMaster:
fm.set_stickycookie(".*")
f = tutils.tflow(resp=True)
f.response.headers["set-cookie"] = ["foo=bar"]
f.response.headers["set-cookie"] = "foo=bar"
fm.handle_request(f)
fm.handle_response(f)
assert fm.stickycookie_state.jar
assert not "cookie" in f.request.headers
f = f.copy()
fm.handle_request(f)
assert f.request.headers["cookie"] == ["foo=bar"]
assert f.request.headers["cookie"] == "foo=bar"
def test_stickyauth(self):
s = flow.State()
@ -958,14 +954,14 @@ class TestFlowMaster:
fm.set_stickyauth(".*")
f = tutils.tflow(resp=True)
f.request.headers["authorization"] = ["foo"]
f.request.headers["authorization"] = "foo"
fm.handle_request(f)
f = tutils.tflow(resp=True)
assert fm.stickyauth_state.hosts
assert not "authorization" in f.request.headers
fm.handle_request(f)
assert f.request.headers["authorization"] == ["foo"]
assert f.request.headers["authorization"] == "foo"
def test_stream(self):
with tutils.tmpdir() as tdir:
@ -1022,7 +1018,7 @@ class TestRequest:
assert r.url == "https://address:22/path"
assert r.pretty_url(True) == "https://address:22/path"
r.headers["Host"] = ["foo.com"]
r.headers["Host"] = "foo.com"
assert r.pretty_url(False) == "https://address:22/path"
assert r.pretty_url(True) == "https://foo.com:22/path"
@ -1048,19 +1044,17 @@ class TestRequest:
def test_getset_form_urlencoded(self):
d = odict.ODict([("one", "two"), ("three", "four")])
r = HTTPRequest.wrap(netlib.tutils.treq(content=netlib.utils.urlencode(d.lst)))
r.headers["content-type"] = [HDR_FORM_URLENCODED]
r.headers["content-type"] = HDR_FORM_URLENCODED
assert r.get_form_urlencoded() == d
d = odict.ODict([("x", "y")])
r.set_form_urlencoded(d)
assert r.get_form_urlencoded() == d
r.headers["content-type"] = ["foo"]
r.headers["content-type"] = "foo"
assert not r.get_form_urlencoded()
def test_getset_query(self):
h = odict.ODictCaseless()
r = HTTPRequest.wrap(netlib.tutils.treq())
r.path = "/foo?x=y&a=b"
q = r.get_query()
@ -1083,11 +1077,10 @@ class TestRequest:
assert r.get_query() == qv
def test_anticache(self):
h = odict.ODictCaseless()
r = HTTPRequest.wrap(netlib.tutils.treq())
r.headers = h
h["if-modified-since"] = ["test"]
h["if-none-match"] = ["test"]
r.headers = Headers()
r.headers["if-modified-since"] = "test"
r.headers["if-none-match"] = "test"
r.anticache()
assert not "if-modified-since" in r.headers
assert not "if-none-match" in r.headers
@ -1095,25 +1088,29 @@ class TestRequest:
def test_replace(self):
r = HTTPRequest.wrap(netlib.tutils.treq())
r.path = "path/foo"
r.headers["Foo"] = ["fOo"]
r.headers["Foo"] = "fOo"
r.content = "afoob"
assert r.replace("foo(?i)", "boo") == 4
assert r.path == "path/boo"
assert not "foo" in r.content
assert r.headers["boo"] == ["boo"]
assert r.headers["boo"] == "boo"
def test_constrain_encoding(self):
r = HTTPRequest.wrap(netlib.tutils.treq())
r.headers["accept-encoding"] = ["gzip", "oink"]
r.headers["accept-encoding"] = "gzip, oink"
r.constrain_encoding()
assert "oink" not in r.headers["accept-encoding"]
r.headers.set_all("accept-encoding", ["gzip", "oink"])
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.headers["content-encoding"] = "identity"
r.content = "falafel"
r.decode()
assert not r.headers["content-encoding"]
assert "content-encoding" not in r.headers
assert r.content == "falafel"
r = HTTPRequest.wrap(netlib.tutils.treq())
@ -1121,26 +1118,26 @@ class TestRequest:
assert not r.decode()
r = HTTPRequest.wrap(netlib.tutils.treq())
r.headers["content-encoding"] = ["identity"]
r.headers["content-encoding"] = "identity"
r.content = "falafel"
r.encode("identity")
assert r.headers["content-encoding"] == ["identity"]
assert r.headers["content-encoding"] == "identity"
assert r.content == "falafel"
r = HTTPRequest.wrap(netlib.tutils.treq())
r.headers["content-encoding"] = ["identity"]
r.headers["content-encoding"] = "identity"
r.content = "falafel"
r.encode("gzip")
assert r.headers["content-encoding"] == ["gzip"]
assert r.headers["content-encoding"] == "gzip"
assert r.content != "falafel"
r.decode()
assert not r.headers["content-encoding"]
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
r.headers["content-encoding"] = ["identity"]
r.headers["content-encoding"] = "identity"
assert r.get_decoded_content() == None
r.content = "falafel"
@ -1148,11 +1145,9 @@ class TestRequest:
assert r.get_decoded_content() == "falafel"
def test_get_content_type(self):
h = odict.ODictCaseless()
h["Content-Type"] = ["text/plain"]
resp = HTTPResponse.wrap(netlib.tutils.tresp())
resp.headers = h
assert resp.headers.get_first("content-type") == "text/plain"
resp.headers = Headers(content_type="text/plain")
assert resp.headers["content-type"] == "text/plain"
class TestResponse:
@ -1165,19 +1160,18 @@ class TestResponse:
def test_refresh(self):
r = HTTPResponse.wrap(netlib.tutils.tresp())
n = time.time()
r.headers["date"] = [email.utils.formatdate(n)]
r.headers["date"] = email.utils.formatdate(n)
pre = r.headers["date"]
r.refresh(n)
assert pre == r.headers["date"]
r.refresh(n + 60)
d = email.utils.parsedate_tz(r.headers["date"][0])
d = email.utils.parsedate_tz(r.headers["date"])
d = email.utils.mktime_tz(d)
# Weird that this is not exact...
assert abs(60 - (d - n)) <= 1
r.headers[
"set-cookie"] = ["MOO=BAR; Expires=Tue, 08-Mar-2011 00:20:38 GMT; Path=foo.com; Secure"]
r.headers["set-cookie"] = "MOO=BAR; Expires=Tue, 08-Mar-2011 00:20:38 GMT; Path=foo.com; Secure"
r.refresh()
def test_refresh_cookie(self):
@ -1192,47 +1186,45 @@ class TestResponse:
def test_replace(self):
r = HTTPResponse.wrap(netlib.tutils.tresp())
r.headers["Foo"] = ["fOo"]
r.headers["Foo"] = "fOo"
r.content = "afoob"
assert r.replace("foo(?i)", "boo") == 3
assert not "foo" in r.content
assert r.headers["boo"] == ["boo"]
assert r.headers["boo"] == "boo"
def test_decodeencode(self):
r = HTTPResponse.wrap(netlib.tutils.tresp())
r.headers["content-encoding"] = ["identity"]
r.headers["content-encoding"] = "identity"
r.content = "falafel"
assert r.decode()
assert not r.headers["content-encoding"]
assert "content-encoding" not in r.headers
assert r.content == "falafel"
r = HTTPResponse.wrap(netlib.tutils.tresp())
r.headers["content-encoding"] = ["identity"]
r.headers["content-encoding"] = "identity"
r.content = "falafel"
r.encode("identity")
assert r.headers["content-encoding"] == ["identity"]
assert r.headers["content-encoding"] == "identity"
assert r.content == "falafel"
r = HTTPResponse.wrap(netlib.tutils.tresp())
r.headers["content-encoding"] = ["identity"]
r.headers["content-encoding"] = "identity"
r.content = "falafel"
r.encode("gzip")
assert r.headers["content-encoding"] == ["gzip"]
assert r.headers["content-encoding"] == "gzip"
assert r.content != "falafel"
assert r.decode()
assert not r.headers["content-encoding"]
assert "content-encoding" not in r.headers
assert r.content == "falafel"
r.headers["content-encoding"] = ["gzip"]
r.headers["content-encoding"] = "gzip"
assert not r.decode()
assert r.content == "falafel"
def test_get_content_type(self):
h = odict.ODictCaseless()
h["Content-Type"] = ["text/plain"]
resp = HTTPResponse.wrap(netlib.tutils.tresp())
resp.headers = h
assert resp.headers.get_first("content-type") == "text/plain"
resp.headers = Headers(content_type="text/plain")
assert resp.headers["content-type"] == "text/plain"
class TestError:
@ -1276,12 +1268,12 @@ class TestClientConnection:
def test_decoded():
r = HTTPRequest.wrap(netlib.tutils.treq())
assert r.content == "content"
assert not r.headers["content-encoding"]
assert "content-encoding" not in r.headers
r.encode("gzip")
assert r.headers["content-encoding"]
assert r.content != "content"
with decoded(r):
assert not r.headers["content-encoding"]
assert "content-encoding" not in r.headers
assert r.content == "content"
assert r.headers["content-encoding"]
assert r.content != "content"
@ -1378,18 +1370,18 @@ def test_setheaders():
h.add("~s", "one", "two")
h.add("~s", "one", "three")
f = tutils.tflow(resp=True)
f.request.headers["one"] = ["xxx"]
f.response.headers["one"] = ["xxx"]
f.request.headers["one"] = "xxx"
f.response.headers["one"] = "xxx"
h.run(f)
assert f.request.headers["one"] == ["xxx"]
assert f.response.headers["one"] == ["two", "three"]
assert f.request.headers["one"] == "xxx"
assert f.response.headers.get_all("one") == ["two", "three"]
h.clear()
h.add("~q", "one", "two")
h.add("~q", "one", "three")
f = tutils.tflow()
f.request.headers["one"] = ["xxx"]
f.request.headers["one"] = "xxx"
h.run(f)
assert f.request.headers["one"] == ["two", "three"]
assert f.request.headers.get_all("one") == ["two", "three"]
assert not h.add("~", "foo", "bar")

1
test/test_protocol_http.py
View File

@ -5,7 +5,6 @@ from mock import MagicMock
from libmproxy.protocol.http import *
import netlib.http
from netlib import odict
from netlib.http import http1
from netlib.http.semantics import CONTENT_MISSING

7
test/test_server.py
View File

@ -623,8 +623,7 @@ class MasterRedirectRequest(tservers.TestMaster):
def handle_response(self, f):
f.response.content = str(f.client_conn.address.port)
f.response.headers[
"server-conn-id"] = [str(f.server_conn.source_address.port)]
f.response.headers["server-conn-id"] = str(f.server_conn.source_address.port)
super(MasterRedirectRequest, self).handle_response(f)
@ -721,7 +720,7 @@ class TestStreamRequest(tservers.HTTPProxTest):
protocol = http.http1.HTTP1Protocol(rfile=fconn)
resp = protocol.read_response("GET", None, include_body=False)
assert resp.headers["Transfer-Encoding"][0] == 'chunked'
assert resp.headers["Transfer-Encoding"] == 'chunked'
assert resp.status_code == 200
chunks = list(protocol.read_http_body_chunked(
@ -743,7 +742,7 @@ class TestFakeResponse(tservers.HTTPProxTest):
def test_fake(self):
f = self.pathod("200")
assert "header_response" in f.headers.keys()
assert "header_response" in f.headers
class TestServerConnect(tservers.HTTPProxTest):

1
test/test_utils.py
View File

@ -1,6 +1,5 @@
import json
from libmproxy import utils
from netlib import odict
import tutils
utils.CERT_SLEEP_TIME = 0