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

intercept: don't fail on requests taken by other addons

Browse Source
This commit is contained in:
Maximilian Hils 2020-08-27 11:07:04 +02:00
parent 0203202b89
commit 28f04c93fd
1 changed files with 22 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
mitmproxy/addons/intercept.py
View File

@ -1,20 +1,18 @@
import typing import typing
from mitmproxy import flowfilter from mitmproxy import flow, flowfilter
from mitmproxy import exceptions from mitmproxy import exceptions
from mitmproxy import ctx from mitmproxy import ctx
class Intercept: class Intercept:
def __init__(self): filt: typing.Optional[flowfilter.TFilter] = None
self.filt = None
def load(self, loader): def load(self, loader):
loader.add_option( loader.add_option(
"intercept_active", bool, False, "intercept_active", bool, False,
"Intercept toggle" "Intercept toggle"
) )
loader.add_option( loader.add_option(
"intercept", typing.Optional[str], None, "intercept", typing.Optional[str], None,
"Intercept filter expression." "Intercept filter expression."
@ -22,25 +20,28 @@ class Intercept:
def configure(self, updated): def configure(self, updated):
if "intercept" in 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 self.filt = None
ctx.options.intercept_active = False 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): def should_intercept(self, f: flow.Flow) -> bool:
if self.filt: return (
should_intercept = all([ ctx.options.intercept_active
self.filt(f), and self.filt
not f.is_replay == "request", and self.filt(f)
]) and not f.is_replay
if should_intercept and ctx.options.intercept_active: )
f.intercept()
def process_flow(self, f: flow.Flow) -> None:
if self.should_intercept(f):
if f.reply.state != "start":
return ctx.log.debug("Cannot intercept request that is already taken by another addon.")
f.intercept()
# Handlers # Handlers
@ -51,5 +52,4 @@ class Intercept:
self.process_flow(f) self.process_flow(f)
def tcp_message(self, f): def tcp_message(self, f):
if self.filt and ctx.options.intercept_active: self.process_flow(f)
f.intercept()