Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-22 15:37:45 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #4177 from mhils/intercept-taken

Browse Source 
Intercept: don't fail on requests taken by other addons
This commit is contained in:
Maximilian Hils 2020-08-27 12:09:50 +02:00 committed by GitHub
commit 72dd567994
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 45 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
mitmproxy/addons/intercept.py
View File

@ -1,20 +1,18 @@
import typing
from mitmproxy import flowfilter
from mitmproxy import flow, flowfilter
from mitmproxy import exceptions
from mitmproxy import ctx
class Intercept:
def __init__(self):
self.filt = None
filt: typing.Optional[flowfilter.TFilter] = None
def load(self, loader):
loader.add_option(
"intercept_active", bool, False,
"Intercept toggle"
)
loader.add_option(
"intercept", typing.Optional[str], None,
"Intercept filter expression."
@ -22,25 +20,29 @@ class Intercept:
def configure(self, updated):
if "intercept" in updated:
if not ctx.options.intercept:
if ctx.options.intercept:
self.filt = flowfilter.parse(ctx.options.intercept)
if not self.filt:
raise exceptions.OptionsError(f"Invalid interception filter: {ctx.options.intercept}")
ctx.options.intercept_active = True
else:
self.filt = None
ctx.options.intercept_active = False
return
self.filt = flowfilter.parse(ctx.options.intercept)
if not self.filt:
raise exceptions.OptionsError(
"Invalid interception filter: %s" % ctx.options.intercept
)
ctx.options.intercept_active = True
def process_flow(self, f):
if self.filt:
should_intercept = all([
self.filt(f),
not f.is_replay == "request",
])
if should_intercept and ctx.options.intercept_active:
f.intercept()
def should_intercept(self, f: flow.Flow) -> bool:
return bool(
ctx.options.intercept_active
and self.filt
and self.filt(f)
and not f.is_replay
)
def process_flow(self, f: flow.Flow) -> None:
if self.should_intercept(f):
assert f.reply
if f.reply.state != "start":
return ctx.log.debug("Cannot intercept request that is already taken by another addon.")
f.intercept()
# Handlers
@ -51,5 +53,4 @@ class Intercept:
self.process_flow(f)
def tcp_message(self, f):
if self.filt and ctx.options.intercept_active:
f.intercept()
self.process_flow(f)

25
test/mitmproxy/addons/test_intercept.py
View File

@ -43,12 +43,31 @@ def test_simple():
tctx.cycle(r, f)
assert f.intercepted
def test_tcp():
r = intercept.Intercept()
with taddons.context(r) as tctx:
tctx.configure(r, intercept="~tcp")
f = tflow.ttcpflow()
tctx.cycle(r, f)
assert f.intercepted
tctx.configure(r, intercept_active=False)
f = tflow.ttcpflow()
tctx.cycle(r, f)
assert not f.intercepted
tctx.configure(r, intercept_active=True)
f = tflow.ttcpflow()
tctx.cycle(r, f)
def test_already_taken():
r = intercept.Intercept()
with taddons.context(r) as tctx:
tctx.configure(r, intercept="~q")
f = tflow.tflow()
tctx.invoke(r, "request", f)
assert f.intercepted
f = tflow.tflow()
f.reply.take()
tctx.invoke(r, "request", f)
assert not f.intercepted