kylefang/mitmproxy
1
0
Fork 0
mirror of https://github.com/zhigang1992/mitmproxy.git synced 2026-04-23 20:00:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor filter matching.

Browse Source
This commit is contained in:
Aldo Cortesi
2012-02-23 17:03:58 +13:00
parent ee8058a2d9
commit 2df9c52c09
3 changed files with 71 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
libmproxy/filt.py
View File

@@ -60,22 +60,22 @@ class _Action(_Token):
class FErr(_Action):
code = "e"
help = "Match error"
def __call__(self, conn):
return isinstance(conn, flow.Error)
def __call__(self, f):
return True if f.error else False
class FReq(_Action):
code = "q"
help = "Match request"
def __call__(self, conn):
return isinstance(conn, flow.Request)
def __call__(self, f):
return True if f.request else False
class FResp(_Action):
code = "s"
help = "Match response"
def __call__(self, conn):
return isinstance(conn, flow.Response)
def __call__(self, f):
return True if f.response else False
class _Rex(_Action):
@@ -96,72 +96,64 @@ def _check_content_type(expr, o):
class FContentType(_Rex):
code = "t"
help = "Content-type header"
def __call__(self, o):
if _check_content_type(self.expr, o):
def __call__(self, f):
if _check_content_type(self.expr, f.request):
return True
elif isinstance(o, flow.Response) and _check_content_type(self.expr, o.request):
elif f.response and _check_content_type(self.expr, f.response):
return True
else:
return False
return False
class FRequestContentType(_Rex):
code = "tq"
help = "Request Content-Type header"
def __call__(self, o):
if isinstance(o, flow.Response):
return _check_content_type(self.expr, o.request)
else:
return _check_content_type(self.expr, o)
def __call__(self, f):
return _check_content_type(self.expr, f.request)
class FResponseContentType(_Rex):
code = "ts"
help = "Request Content-Type header"
def __call__(self, o):
if isinstance(o, flow.Response):
return _check_content_type(self.expr, o)
else:
return False
def __call__(self, f):
if f.response:
return _check_content_type(self.expr, f.response)
return False
class FHead(_Rex):
code = "h"
help = "Header"
def __call__(self, o):
val = o.headers.match_re(self.expr)
if not val and isinstance(o, flow.Response):
val = o.request.headers.match_re(self.expr)
return val
def __call__(self, f):
if f.request.headers.match_re(self.expr):
return True
elif f.response and f.response.headers.match_re(self.expr):
return True
return False
class FHeadRequest(_Rex):
code = "hq"
help = "Request header"
def __call__(self, o):
if isinstance(o, flow.Response):
h = o.request.headers
else:
h = o.headers
return h.match_re(self.expr)
def __call__(self, f):
if f.request.headers.match_re(self.expr):
return True
class FHeadResponse(_Rex):
code = "hs"
help = "Response header"
def __call__(self, o):
if not isinstance(o, flow.Response):
return False
return o.headers.match_re(self.expr)
def __call__(self, f):
if f.response and f.response.headers.match_re(self.expr):
return True
class FBod(_Rex):
code = "b"
help = "Body"
def __call__(self, o):
if o.content and re.search(self.expr, o.content):
def __call__(self, f):
if f.request.content and re.search(self.expr, f.request.content):
return True
elif isinstance(o, flow.Response) and o.request.content and re.search(self.expr, o.request.content):
elif f.response and f.response.content and re.search(self.expr, f.response.content):
return True
return False
@@ -169,32 +161,24 @@ class FBod(_Rex):
class FBodRequest(_Rex):
code = "bq"
help = "Request body"
def __call__(self, o):
if isinstance(o, flow.Response) and o.request.content and re.search(self.expr, o.request.content):
def __call__(self, f):
if f.request.content and re.search(self.expr, f.request.content):
return True
elif not isinstance(o, flow.Response) and o.content and re.search(self.expr, o.content):
return True
return False
class FBodResponse(_Rex):
code = "bs"
help = "Response body"
def __call__(self, o):
if not isinstance(o, flow.Response):
return False
elif o.content and re.search(self.expr, o.content):
def __call__(self, f):
if f.response and f.response.content and re.search(self.expr, f.response.content):
return True
return False
class FMethod(_Rex):
code = "m"
help = "Method"
def __call__(self, o):
if isinstance(o, flow.Request):
return re.search(self.expr, o.method, re.IGNORECASE)
return False
def __call__(self, f):
return bool(re.search(self.expr, f.request.method, re.IGNORECASE))
class FUrl(_Rex):
@@ -207,14 +191,8 @@ class FUrl(_Rex):
toks = toks[1:]
return klass(*toks)
def __call__(self, o):
if isinstance(o, flow.Response):
c = o.request
elif isinstance(o, flow.Request):
c = o
else:
return False
return re.search(self.expr, c.get_url())
def __call__(self, f):
return re.search(self.expr, f.request.get_url())
class _Int(_Action):
@@ -225,10 +203,9 @@ class _Int(_Action):
class FCode(_Int):
code = "c"
help = "HTTP response code"
def __call__(self, o):
if isinstance(o, flow.Response):
return o.code == self.num
return False
def __call__(self, f):
if f.response and f.response.code == self.num:
return True
class FAnd(_Token):
@@ -240,8 +217,8 @@ class FAnd(_Token):
for i in self.lst:
i.dump(indent+1, fp)
def __call__(self, o):
return all(i(o) for i in self.lst)
def __call__(self, f):
return all(i(f) for i in self.lst)
class FOr(_Token):
@@ -253,8 +230,8 @@ class FOr(_Token):
for i in self.lst:
i.dump(indent+1, fp)
def __call__(self, o):
return any(i(o) for i in self.lst)
def __call__(self, f):
return any(i(f) for i in self.lst)
class FNot(_Token):
@@ -265,8 +242,9 @@ class FNot(_Token):
print >> fp, "\t"*indent, self.__class__.__name__
self.itm.dump(indent + 1, fp)
def __call__(self, o):
return not self.itm(o)
def __call__(self, f):
return not self.itm(f)
filt_unary = [
FReq,

12
libmproxy/flow.py
View File

@@ -979,16 +979,8 @@ class Flow:
if matched, False if not.
"""
if f:
if self.response and f(self.response):
return True
elif self.request and f(self.request):
return True
elif self.error and f(self.error):
return True
else:
return False
else:
return True
return f(self)
return True
def kill(self, master):
"""

34
test/test_filt.py
View File

@@ -76,7 +76,7 @@ class uMatching(libpry.AutoTree):
conn = flow.ClientConnect(("one", 2222))
headers = flow.ODictCaseless()
headers["header"] = ["qvalue"]
return flow.Request(
req = flow.Request(
conn,
"host",
80,
@@ -86,22 +86,26 @@ class uMatching(libpry.AutoTree):
headers,
"content_request"
)
return flow.Flow(req)
def resp(self):
q = self.req()
f = self.req()
headers = flow.ODictCaseless()
headers["header_response"] = ["svalue"]
return flow.Response(
q,
f.response = flow.Response(
f.request,
200,
"message",
headers,
"content_response"
)
return f
def err(self):
q = self.req()
return flow.Error(q, "msg")
f = self.req()
f.error = flow.Error(f.request, "msg")
return f
def q(self, q, o):
return filt.parse(q)(o)
@@ -112,15 +116,15 @@ class uMatching(libpry.AutoTree):
assert not self.q("~t content", q)
assert not self.q("~t content", s)
q.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.headers["content-type"] = ["text/json"]
s.response.headers["content-type"] = ["text/json"]
assert self.q("~t json", s)
del s.headers["content-type"]
del s.response.headers["content-type"]
s.request.headers["content-type"] = ["text/json"]
assert self.q("~t json", s)
assert self.q("~tq json", s)
@@ -131,7 +135,8 @@ class uMatching(libpry.AutoTree):
s = self.resp()
assert self.q("~q", q)
assert not self.q("~q", s)
# FIXME
assert self.q("~q", s)
assert not self.q("~s", q)
assert self.q("~s", s)
@@ -185,8 +190,8 @@ class uMatching(libpry.AutoTree):
s = self.resp()
assert self.q("~m get", q)
assert not self.q("~m post", q)
assert not self.q("~m get", s)
q.method = ""
q.request.method = "oink"
assert not self.q("~m get", q)
def test_url(self):
@@ -210,6 +215,7 @@ class uMatching(libpry.AutoTree):
def test_and(self):
s = self.resp()
assert self.q("~c 200 & ~h head", s)
assert self.q("~c 200 & ~h head", s)
assert not self.q("~c 200 & ~h nohead", s)
assert self.q("(~c 200 & ~h head) & ~b content", s)
assert not self.q("(~c 200 & ~h head) & ~b nonexistent", s)
@@ -220,8 +226,8 @@ class uMatching(libpry.AutoTree):
assert self.q("~c 200 | ~h nohead", s)
assert self.q("~c 201 | ~h head", s)
assert not self.q("~c 201 | ~h nohead", s)
assert self.q("(~c 201 | ~h nohead) | ~s", s)
assert not self.q("(~c 201 | ~h nohead) | ~q", s)
# FIXME
#assert self.q("(~c 201 | ~h nohead) | ~s", s)
def test_not(self):
s = self.resp()