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

View API slightly extended; codebase cleaned in some points

Browse Source
This commit is contained in:
madt1m 2018-06-13 11:56:14 +02:00
parent b6fd9b4484
commit 773c953514
10 changed files with 155 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/complex/dup_and_replay.py
View File

@ -9,6 +9,6 @@ def request(flow):
# Only interactive tools have a view. If we have one, add a duplicate entry
# for our flow.
if "view" in ctx.master.addons:
ctx.master.commands.call("view.add", [flow])
ctx.master.commands.call("view.flows.add", [flow])
flow.request.path = "/changed"
ctx.master.commands.call("replay.client", [flow])

215
mitmproxy/addons/view.py
View File

@ -166,12 +166,6 @@ class View(collections.Sequence):
def store_count(self):
return len(self._store)
def inbounds(self, index: int) -> bool:
"""
Is this 0 <= index < len(self)
"""
return 0 <= index < len(self)
def _rev(self, idx: int) -> int:
"""
Reverses an index, if needed
@ -219,7 +213,26 @@ class View(collections.Sequence):
self._base_add(i)
self.sig_view_refresh.send(self)
# API
""" View API """
# Focus
@command.command("view.focus.go")
def go(self, dst: int) -> None:
"""
Go to a specified offset. Positive offests are from the beginning of
the view, negative from the end of the view, so that 0 is the first
flow, -1 is the last flow.
"""
if len(self) == 0:
return
if dst < 0:
dst = len(self) + dst
if dst < 0:
dst = 0
if dst > len(self) - 1:
dst = len(self) - 1
self.focus.flow = self[dst]
@command.command("view.focus.next")
def focus_next(self) -> None:
"""
@ -238,6 +251,7 @@ class View(collections.Sequence):
if self.inbounds(idx):
self.focus.flow = self[idx]
# Order
@command.command("view.order.options")
def order_options(self) -> typing.Sequence[str]:
"""
@ -245,34 +259,56 @@ class View(collections.Sequence):
"""
return list(sorted(self.orders.keys()))
@command.command("view.marked.toggle")
def toggle_marked(self) -> None:
"""
Toggle whether to show marked views only.
"""
self.show_marked = not self.show_marked
self._refilter()
@command.command("view.order.reverse")
def set_reversed(self, value: bool):
self.order_reversed = value
self.sig_view_refresh.send(self)
def set_order(self, order_key: typing.Callable):
@command.command("view.order.set")
def set_order(self, order: str):
"""
Sets the current view order.
"""
if order not in self.orders:
raise exceptions.CommandError(
"Unknown flow order: %s" % order
)
order_key = self.orders[order]
self.order_key = order_key
newview = sortedcontainers.SortedListWithKey(key=order_key)
newview.update(self._view)
self._view = newview
def set_filter(self, flt: typing.Optional[flowfilter.TFilter]):
@command.command("view.order")
def get_order(self) -> str:
"""
Returns the current view order.
"""
for k in self.orders.keys():
if self.order_key == self.orders[k]:
return k
# Filter
@command.command("view.filter.set")
def filter_set(self, f: str) -> None:
"""
Sets the current view filter.
"""
filt = None
if f:
filt = flowfilter.parse(f)
if not filt:
raise exceptions.OptionsError(
"Invalid interception filter: %s" % f
)
self.set_filter(filt)
def set_filter(self, flt: typing.Optional[flowfilter.TFilter]):
self.filter = flt or matchall
self._refilter()
# View Updates
@command.command("view.clear")
def clear(self) -> None:
"""
Clears both the store and view.
@ -282,7 +318,8 @@ class View(collections.Sequence):
self.sig_view_refresh.send(self)
self.sig_store_refresh.send(self)
def clear_not_marked(self):
@command.command("view.clear_unmarked")
def clear_not_marked(self) -> None:
"""
Clears only the unmarked flows.
"""
@ -293,36 +330,15 @@ class View(collections.Sequence):
self._refilter()
self.sig_store_refresh.send(self)
@command.command("view.marked.toggle")
def add(self, flows: typing.Sequence[mitmproxy.flow.Flow]) -> None:
"""
Adds a flow to the state. If the flow already exists, it is
ignored.
"""
for f in flows:
if f.id not in self._store:
self._store[f.id] = f
if self.filter(f):
self._base_add(f)
if self.focus_follow:
self.focus.flow = f
self.sig_view_add.send(self, flow=f)
def get_by_id(self, flow_id: str) -> typing.Optional[mitmproxy.flow.Flow]:
"""
Get flow with the given id from the store.
Returns None if the flow is not found.
"""
return self._store.get(flow_id)
@command.command("view.getval")
# View Settings
@command.command("view.settings.getval")
def getvalue(self, f: mitmproxy.flow.Flow, key: str, default: str) -> str:
"""
Get a value from the settings store for the specified flow.
"""
return self.settings[f].get(key, default)
@command.command("view.setval.toggle")
@command.command("view.settings.setval.toggle")
def setvalue_toggle(
self,
flows: typing.Sequence[mitmproxy.flow.Flow],
@ -339,7 +355,7 @@ class View(collections.Sequence):
updated.append(f)
ctx.master.addons.trigger("update", updated)
@command.command("view.setval")
@command.command("view.settings.setval")
def setvalue(
self,
flows: typing.Sequence[mitmproxy.flow.Flow],
@ -354,41 +370,8 @@ class View(collections.Sequence):
updated.append(f)
ctx.master.addons.trigger("update", updated)
@command.command("view.load")
def load_file(self, path: mitmproxy.types.Path) -> None:
"""
Load flows into the view, without processing them with addons.
"""
try:
with open(path, "rb") as f:
for i in io.FlowReader(f).stream():
# Do this to get a new ID, so we can load the same file N times and
# get new flows each time. It would be more efficient to just have a
# .newid() method or something.
self.add([i.copy()])
except IOError as e:
ctx.log.error(e.strerror)
except exceptions.FlowReadException as e:
ctx.log.error(str(e))
@command.command("view.go")
def go(self, dst: int) -> None:
"""
Go to a specified offset. Positive offests are from the beginning of
the view, negative from the end of the view, so that 0 is the first
flow, -1 is the last flow.
"""
if len(self) == 0:
return
if dst < 0:
dst = len(self) + dst
if dst < 0:
dst = 0
if dst > len(self) - 1:
dst = len(self) - 1
self.focus.flow = self[dst]
@command.command("view.duplicate")
# Flows
@command.command("view.flows.duplicate")
def duplicate(self, flows: typing.Sequence[mitmproxy.flow.Flow]) -> None:
"""
Duplicates the specified flows, and sets the focus to the first
@ -400,7 +383,7 @@ class View(collections.Sequence):
self.focus.flow = dups[0]
ctx.log.alert("Duplicated %s flows" % len(dups))
@command.command("view.remove")
@command.command("view.flows.remove")
def remove(self, flows: typing.Sequence[mitmproxy.flow.Flow]) -> None:
"""
Removes the flow from the underlying store and the view.
@ -420,7 +403,7 @@ class View(collections.Sequence):
if len(flows) > 1:
ctx.log.alert("Removed %s flows" % len(flows))
@command.command("view.resolve")
@command.command("view.flows.resolve")
def resolve(self, spec: str) -> typing.Sequence[mitmproxy.flow.Flow]:
"""
Resolve a flow list specification to an actual list of flows.
@ -443,7 +426,7 @@ class View(collections.Sequence):
raise exceptions.CommandError("Invalid flow filter: %s" % spec)
return [i for i in self._store.values() if filt(i)]
@command.command("view.create")
@command.command("view.flows.create")
def create(self, method: str, url: str) -> None:
try:
req = http.HTTPRequest.make(method.upper(), url)
@ -456,6 +439,74 @@ class View(collections.Sequence):
f.request.headers["Host"] = req.host
self.add([f])
@command.command("view.flows.load")
def load_file(self, path: mitmproxy.types.Path) -> None:
"""
Load flows into the view, without processing them with addons.
"""
try:
with open(path, "rb") as f:
for i in io.FlowReader(f).stream():
# Do this to get a new ID, so we can load the same file N times and
# get new flows each time. It would be more efficient to just have a
# .newid() method or something.
self.add([i.copy()])
except IOError as e:
ctx.log.error(e.strerror)
except exceptions.FlowReadException as e:
ctx.log.error(str(e))
def add(self, flows: typing.Sequence[mitmproxy.flow.Flow]) -> None:
"""
Adds a flow to the state. If the flow already exists, it is
ignored.
"""
for f in flows:
if f.id not in self._store:
self._store[f.id] = f
if self.filter(f):
self._base_add(f)
if self.focus_follow:
self.focus.flow = f
self.sig_view_add.send(self, flow=f)
def get_by_id(self, flow_id: str) -> typing.Optional[mitmproxy.flow.Flow]:
"""
Get flow with the given id from the store.
Returns None if the flow is not found.
"""
return self._store.get(flow_id)
# View Properties
@command.command("view.properties.length")
def get_length(self) -> int:
"""
Returns view length.
"""
return len(self)
@command.command("view.properties.marked")
def get_marked(self) -> bool:
"""
Returns true if view is in marked mode.
"""
return self.show_marked
@command.command("view.properties.marked.toggle")
def toggle_marked(self) -> None:
"""
Toggle whether to show marked views only.
"""
self.show_marked = not self.show_marked
self._refilter()
@command.command("view.properties.inbounds")
def inbounds(self, index: int) -> bool:
"""
Is this 0 <= index < len(self)?
"""
return 0 <= index < len(self)
# Event handlers
def configure(self, updated):
if "view_filter" in updated:
@ -472,7 +523,7 @@ class View(collections.Sequence):
raise exceptions.OptionsError(
"Unknown flow order: %s" % ctx.options.view_order
)
self.set_order(self.orders[ctx.options.view_order])
self.set_order(ctx.options.view_order)
if "view_order_reversed" in updated:
self.set_reversed(ctx.options.view_order_reversed)
if "console_focus_follow" in updated:

4
mitmproxy/tools/console/consoleaddons.py
View File

@ -515,7 +515,7 @@ class ConsoleAddon:
try:
self.master.commands.call_strings(
"view.setval",
"view.settings.setval",
["@focus", "flowview_mode_%s" % idx, mode]
)
except exceptions.CommandError as e:
@ -538,7 +538,7 @@ class ConsoleAddon:
raise exceptions.CommandError("Not viewing a flow.")
idx = fv.body.tab_offset
return self.master.commands.call_strings(
"view.getval",
"view.settings.getval",
[
"@focus",
"flowview_mode_%s" % idx,

12
mitmproxy/tools/console/defaultkeys.py
View File

@ -36,8 +36,8 @@ def map(km):
["flowlist", "flowview"],
"Save response body to file"
)
km.add("d", "view.remove @focus", ["flowlist", "flowview"], "Delete flow from view")
km.add("D", "view.duplicate @focus", ["flowlist", "flowview"], "Duplicate flow")
km.add("d", "view.flows.remove @focus", ["flowlist", "flowview"], "Delete flow from view")
km.add("D", "view.flows.duplicate @focus", ["flowlist", "flowview"], "Duplicate flow")
km.add(
"e",
"""
@ -57,7 +57,7 @@ def map(km):
)
km.add("L", "console.command view.load ", ["flowlist"], "Load flows from file")
km.add("m", "flow.mark.toggle @focus", ["flowlist"], "Toggle mark on this flow")
km.add("M", "view.marked.toggle", ["flowlist"], "Toggle viewing marked flows")
km.add("M", "view.properties.marked.toggle", ["flowlist"], "Toggle viewing marked flows")
km.add(
"n",
"console.command view.create get https://example.com/",
@ -80,8 +80,8 @@ def map(km):
km.add("w", "console.command save.file @shown ", ["flowlist"], "Save listed flows to file")
km.add("V", "flow.revert @focus", ["flowlist", "flowview"], "Revert changes to this flow")
km.add("X", "flow.kill @focus", ["flowlist"], "Kill this flow")
km.add("z", "view.remove @all", ["flowlist"], "Clear flow list")
km.add("Z", "view.remove @hidden", ["flowlist"], "Purge all flows not showing")
km.add("z", "view.flows.remove @all", ["flowlist"], "Clear flow list")
km.add("Z", "view.flows.remove @hidden", ["flowlist"], "Purge all flows not showing")
km.add(
"|",
"console.command script.run @focus ",
@ -100,7 +100,7 @@ def map(km):
)
km.add(
"f",
"view.setval.toggle @focus fullcontents",
"view.settings.setval.toggle @focus fullcontents",
["flowview"],
"Toggle viewing full contents on this flow",
)

12
mitmproxy/tools/console/flowlist.py
View File

@ -42,7 +42,7 @@ class FlowListWalker(urwid.ListWalker):
def positions(self, reverse=False):
# The stub implementation of positions can go once this issue is resolved:
# https://github.com/urwid/urwid/issues/294
ret = range(len(self.master.view))
ret = range(self.master.commands.execute("view.properties.length"))
if reverse:
return reversed(ret)
return ret
@ -57,19 +57,19 @@ class FlowListWalker(urwid.ListWalker):
return f, self.master.view.focus.index
def set_focus(self, index):
if self.master.view.inbounds(index):
if self.master.commands.execute("view.properties.inbounds %d" % index):
self.master.view.focus.index = index
def get_next(self, pos):
pos = pos + 1
if not self.master.view.inbounds(pos):
if not self.master.commands.execute("view.properties.inbounds %d" % pos):
return None, None
f = FlowItem(self.master, self.master.view[pos])
return f, pos
def get_prev(self, pos):
pos = pos - 1
if not self.master.view.inbounds(pos):
if not self.master.commands.execute("view.properties.inbounds %d" % pos):
return None, None
f = FlowItem(self.master, self.master.view[pos])
return f, pos
@ -87,9 +87,9 @@ class FlowListBox(urwid.ListBox, layoutwidget.LayoutWidget):
def keypress(self, size, key):
if key == "m_start":
self.master.commands.execute("view.go 0")
self.master.commands.execute("view.focus.go 0")
elif key == "m_end":
self.master.commands.execute("view.go -1")
self.master.commands.execute("view.focus.go -1")
elif key == "m_select":
self.master.commands.execute("console.view.flow @focus")
return urwid.ListBox.keypress(self, size, key)

2
mitmproxy/tools/console/flowview.py
View File

@ -98,7 +98,7 @@ class FlowDetails(tabs.Tabs):
msg, body = "", [urwid.Text([("error", "[content missing]")])]
return msg, body
else:
full = self.master.commands.execute("view.getval @focus fullcontents false")
full = self.master.commands.execute("view.settings.getval @focus fullcontents false")
if full == "true":
limit = sys.maxsize
else:

4
mitmproxy/tools/console/statusbar.py
View File

@ -271,7 +271,7 @@ class StatusBar(urwid.WidgetWrap):
return r
def redraw(self):
fc = len(self.master.view)
fc = self.master.commands.execute("view.properties.length")
if self.master.view.focus.flow is None:
offset = 0
else:
@ -283,7 +283,7 @@ class StatusBar(urwid.WidgetWrap):
arrow = common.SYMBOL_DOWN
marked = ""
if self.master.view.show_marked:
if self.master.commands.execute("view.properties.marked"):
marked = "M"
t = [

4
mitmproxy/types.py
View File

@ -337,7 +337,7 @@ class _FlowType(_BaseFlowType):
def parse(self, manager: _CommandBase, t: type, s: str) -> flow.Flow:
try:
flows = manager.call_strings("view.resolve", [s])
flows = manager.call_strings("view.flows.resolve", [s])
except exceptions.CommandError as e:
raise exceptions.TypeError from e
if len(flows) != 1:
@ -356,7 +356,7 @@ class _FlowsType(_BaseFlowType):
def parse(self, manager: _CommandBase, t: type, s: str) -> typing.Sequence[flow.Flow]:
try:
return manager.call_strings("view.resolve", [s])
return manager.call_strings("view.flows.resolve", [s])
except exceptions.CommandError as e:
raise exceptions.TypeError from e

2
test/mitmproxy/test_command.py
View File

@ -313,7 +313,7 @@ def test_typename():
class DummyConsole:
@command.command("view.resolve")
@command.command("view.flows.resolve")
def resolve(self, spec: str) -> typing.Sequence[flow.Flow]:
n = int(spec)
return [tflow.tflow(resp=True)] * n

2
test/mitmproxy/test_types.py
View File

@ -146,7 +146,7 @@ def test_strseq():
class DummyConsole:
@command.command("view.resolve")
@command.command("view.flows.resolve")
def resolve(self, spec: str) -> typing.Sequence[flow.Flow]:
if spec == "err":
raise mitmproxy.exceptions.CommandError()