mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-24 09:15:55 +00:00
378 lines
12 KiB
Python
378 lines
12 KiB
Python
from __future__ import absolute_import
|
|
import urwid
|
|
from netlib import http
|
|
from . import common, signals
|
|
|
|
|
|
def _mkhelp():
|
|
text = []
|
|
keys = [
|
|
("A", "accept all intercepted flows"),
|
|
("a", "accept this intercepted flow"),
|
|
("b", "save request/response body"),
|
|
("C", "clear flow list or eventlog"),
|
|
("d", "delete flow"),
|
|
("D", "duplicate flow"),
|
|
("e", "toggle eventlog"),
|
|
("F", "toggle follow flow list"),
|
|
("l", "set limit filter pattern"),
|
|
("L", "load saved flows"),
|
|
("m", "toggle flow mark"),
|
|
("n", "create a new request"),
|
|
("P", "copy flow to clipboard"),
|
|
("r", "replay request"),
|
|
("U", "unmark all marked flows"),
|
|
("V", "revert changes to request"),
|
|
("w", "save flows "),
|
|
("W", "stream flows to file"),
|
|
("X", "kill and delete flow, even if it's mid-intercept"),
|
|
("tab", "tab between eventlog and flow list"),
|
|
("enter", "view flow"),
|
|
("|", "run script on this flow"),
|
|
]
|
|
text.extend(common.format_keyvals(keys, key="key", val="text", indent=4))
|
|
return text
|
|
help_context = _mkhelp()
|
|
|
|
footer = [
|
|
('heading_key', "?"), ":help ",
|
|
]
|
|
|
|
|
|
class EventListBox(urwid.ListBox):
|
|
def __init__(self, master):
|
|
self.master = master
|
|
urwid.ListBox.__init__(self, master.eventlist)
|
|
|
|
def keypress(self, size, key):
|
|
key = common.shortcuts(key)
|
|
if key == "C":
|
|
self.master.clear_events()
|
|
key = None
|
|
elif key == "G":
|
|
self.set_focus(0)
|
|
elif key == "g":
|
|
self.set_focus(len(self.master.eventlist) - 1)
|
|
return urwid.ListBox.keypress(self, size, key)
|
|
|
|
|
|
class BodyPile(urwid.Pile):
|
|
def __init__(self, master):
|
|
h = urwid.Text("Event log")
|
|
h = urwid.Padding(h, align="left", width=("relative", 100))
|
|
|
|
self.inactive_header = urwid.AttrWrap(h, "heading_inactive")
|
|
self.active_header = urwid.AttrWrap(h, "heading")
|
|
|
|
urwid.Pile.__init__(
|
|
self,
|
|
[
|
|
FlowListBox(master),
|
|
urwid.Frame(
|
|
EventListBox(master),
|
|
header = self.inactive_header
|
|
)
|
|
]
|
|
)
|
|
self.master = master
|
|
|
|
def keypress(self, size, key):
|
|
if key == "tab":
|
|
self.focus_position = (
|
|
self.focus_position + 1) % len(self.widget_list)
|
|
if self.focus_position == 1:
|
|
self.widget_list[1].header = self.active_header
|
|
else:
|
|
self.widget_list[1].header = self.inactive_header
|
|
key = None
|
|
elif key == "e":
|
|
self.master.toggle_eventlog()
|
|
key = None
|
|
|
|
# This is essentially a copypasta from urwid.Pile's keypress handler.
|
|
# So much for "closed for modification, but open for extension".
|
|
item_rows = None
|
|
if len(size) == 2:
|
|
item_rows = self.get_item_rows(size, focus = True)
|
|
i = self.widget_list.index(self.focus_item)
|
|
tsize = self.get_item_size(size, i, True, item_rows)
|
|
return self.focus_item.keypress(tsize, key)
|
|
|
|
|
|
class ConnectionItem(urwid.WidgetWrap):
|
|
def __init__(self, master, state, flow, focus):
|
|
self.master, self.state, self.flow = master, state, flow
|
|
self.f = focus
|
|
w = self.get_text()
|
|
urwid.WidgetWrap.__init__(self, w)
|
|
|
|
def get_text(self):
|
|
return common.format_flow(
|
|
self.flow,
|
|
self.f,
|
|
hostheader = self.master.showhost,
|
|
marked=self.state.flow_marked(self.flow)
|
|
)
|
|
|
|
def selectable(self):
|
|
return True
|
|
|
|
def save_flows_prompt(self, k):
|
|
if k == "a":
|
|
signals.status_prompt_path.send(
|
|
prompt = "Save all flows to",
|
|
callback = self.master.save_flows
|
|
)
|
|
elif k == "m":
|
|
signals.status_prompt_path.send(
|
|
prompt = "Save marked flows to",
|
|
callback = self.master.save_marked_flows
|
|
)
|
|
else:
|
|
signals.status_prompt_path.send(
|
|
prompt = "Save this flow to",
|
|
callback = self.master.save_one_flow,
|
|
args = (self.flow,)
|
|
)
|
|
|
|
def stop_server_playback_prompt(self, a):
|
|
if a != "n":
|
|
self.master.stop_server_playback()
|
|
|
|
def server_replay_prompt(self, k):
|
|
if k == "a":
|
|
self.master.start_server_playback(
|
|
[i.copy() for i in self.master.state.view],
|
|
self.master.killextra, self.master.rheaders,
|
|
False, self.master.nopop,
|
|
self.master.options.replay_ignore_params,
|
|
self.master.options.replay_ignore_content,
|
|
self.master.options.replay_ignore_payload_params,
|
|
self.master.options.replay_ignore_host
|
|
)
|
|
elif k == "t":
|
|
self.master.start_server_playback(
|
|
[self.flow.copy()],
|
|
self.master.killextra, self.master.rheaders,
|
|
False, self.master.nopop,
|
|
self.master.options.replay_ignore_params,
|
|
self.master.options.replay_ignore_content,
|
|
self.master.options.replay_ignore_payload_params,
|
|
self.master.options.replay_ignore_host
|
|
)
|
|
else:
|
|
signals.status_prompt_path.send(
|
|
prompt = "Server replay path",
|
|
callback = self.master.server_playback_path
|
|
)
|
|
|
|
def mouse_event(self, size, event, button, col, row, focus):
|
|
if event == "mouse press" and button == 1:
|
|
if self.flow.request:
|
|
self.master.view_flow(self.flow)
|
|
return True
|
|
|
|
def keypress(self, xxx_todo_changeme, key):
|
|
(maxcol,) = xxx_todo_changeme
|
|
key = common.shortcuts(key)
|
|
if key == "a":
|
|
self.flow.accept_intercept(self.master)
|
|
signals.flowlist_change.send(self)
|
|
elif key == "d":
|
|
self.flow.kill(self.master)
|
|
self.state.delete_flow(self.flow)
|
|
signals.flowlist_change.send(self)
|
|
elif key == "D":
|
|
f = self.master.duplicate_flow(self.flow)
|
|
self.master.view_flow(f)
|
|
elif key == "m":
|
|
if self.state.flow_marked(self.flow):
|
|
self.state.set_flow_marked(self.flow, False)
|
|
else:
|
|
self.state.set_flow_marked(self.flow, True)
|
|
signals.flowlist_change.send(self)
|
|
elif key == "r":
|
|
r = self.master.replay_request(self.flow)
|
|
if r:
|
|
signals.status_message.send(message=r)
|
|
signals.flowlist_change.send(self)
|
|
elif key == "S":
|
|
if not self.master.server_playback:
|
|
signals.status_prompt_onekey.send(
|
|
prompt = "Server Replay",
|
|
keys = (
|
|
("all flows", "a"),
|
|
("this flow", "t"),
|
|
("file", "f"),
|
|
),
|
|
callback = self.server_replay_prompt,
|
|
)
|
|
else:
|
|
signals.status_prompt_onekey.send(
|
|
prompt = "Stop current server replay?",
|
|
keys = (
|
|
("yes", "y"),
|
|
("no", "n"),
|
|
),
|
|
callback = self.stop_server_playback_prompt,
|
|
)
|
|
elif key == "U":
|
|
for f in self.state.flows:
|
|
self.state.set_flow_marked(f, False)
|
|
signals.flowlist_change.send(self)
|
|
elif key == "V":
|
|
if not self.flow.modified():
|
|
signals.status_message.send(message="Flow not modified.")
|
|
return
|
|
self.state.revert(self.flow)
|
|
signals.flowlist_change.send(self)
|
|
signals.status_message.send(message="Reverted.")
|
|
elif key == "w":
|
|
signals.status_prompt_onekey.send(
|
|
self,
|
|
prompt = "Save",
|
|
keys = (
|
|
("all flows", "a"),
|
|
("this flow", "t"),
|
|
("marked flows", "m"),
|
|
),
|
|
callback = self.save_flows_prompt,
|
|
)
|
|
elif key == "X":
|
|
self.flow.kill(self.master)
|
|
elif key == "enter":
|
|
if self.flow.request:
|
|
self.master.view_flow(self.flow)
|
|
elif key == "|":
|
|
signals.status_prompt_path.send(
|
|
prompt = "Send flow to script",
|
|
callback = self.master.run_script_once,
|
|
args = (self.flow,)
|
|
)
|
|
elif key == "P":
|
|
common.ask_copy_part("a", self.flow, self.master, self.state)
|
|
elif key == "b":
|
|
common.ask_save_body(None, self.master, self.state, self.flow)
|
|
else:
|
|
return key
|
|
|
|
|
|
class FlowListWalker(urwid.ListWalker):
|
|
def __init__(self, master, state):
|
|
self.master, self.state = master, state
|
|
signals.flowlist_change.connect(self.sig_flowlist_change)
|
|
|
|
def sig_flowlist_change(self, sender):
|
|
self._modified()
|
|
|
|
def get_focus(self):
|
|
f, i = self.state.get_focus()
|
|
f = ConnectionItem(self.master, self.state, f, True) if f else None
|
|
return f, i
|
|
|
|
def set_focus(self, focus):
|
|
ret = self.state.set_focus(focus)
|
|
return ret
|
|
|
|
def get_next(self, pos):
|
|
f, i = self.state.get_next(pos)
|
|
f = ConnectionItem(self.master, self.state, f, False) if f else None
|
|
return f, i
|
|
|
|
def get_prev(self, pos):
|
|
f, i = self.state.get_prev(pos)
|
|
f = ConnectionItem(self.master, self.state, f, False) if f else None
|
|
return f, i
|
|
|
|
|
|
class FlowListBox(urwid.ListBox):
|
|
def __init__(self, master):
|
|
self.master = master
|
|
urwid.ListBox.__init__(
|
|
self,
|
|
FlowListWalker(master, master.state)
|
|
)
|
|
|
|
def get_method_raw(self, k):
|
|
if k:
|
|
self.get_url(k)
|
|
|
|
def get_method(self, k):
|
|
if k == "e":
|
|
signals.status_prompt.send(
|
|
self,
|
|
prompt = "Method",
|
|
text = "",
|
|
callback = self.get_method_raw
|
|
)
|
|
else:
|
|
method = ""
|
|
for i in common.METHOD_OPTIONS:
|
|
if i[1] == k:
|
|
method = i[0].upper()
|
|
self.get_url(method)
|
|
|
|
def get_url(self, method):
|
|
signals.status_prompt.send(
|
|
prompt = "URL",
|
|
text = "http://www.example.com/",
|
|
callback = self.new_request,
|
|
args = (method,)
|
|
)
|
|
|
|
def new_request(self, url, method):
|
|
parts = http.parse_url(str(url))
|
|
if not parts:
|
|
signals.status_message.send(message="Invalid Url")
|
|
return
|
|
scheme, host, port, path = parts
|
|
f = self.master.create_request(method, scheme, host, port, path)
|
|
self.master.view_flow(f)
|
|
|
|
def keypress(self, size, key):
|
|
key = common.shortcuts(key)
|
|
if key == "A":
|
|
self.master.accept_all()
|
|
signals.flowlist_change.send(self)
|
|
elif key == "C":
|
|
self.master.clear_flows()
|
|
elif key == "e":
|
|
self.master.toggle_eventlog()
|
|
elif key == "G":
|
|
self.master.state.set_focus(0)
|
|
signals.flowlist_change.send(self)
|
|
elif key == "g":
|
|
self.master.state.set_focus(self.master.state.flow_count())
|
|
signals.flowlist_change.send(self)
|
|
elif key == "l":
|
|
signals.status_prompt.send(
|
|
prompt = "Limit",
|
|
text = self.master.state.limit_txt,
|
|
callback = self.master.set_limit
|
|
)
|
|
elif key == "L":
|
|
signals.status_prompt_path.send(
|
|
self,
|
|
prompt = "Load flows",
|
|
callback = self.master.load_flows_callback
|
|
)
|
|
elif key == "n":
|
|
signals.status_prompt_onekey.send(
|
|
prompt = "Method",
|
|
keys = common.METHOD_OPTIONS,
|
|
callback = self.get_method
|
|
)
|
|
elif key == "F":
|
|
self.master.toggle_follow_flows()
|
|
elif key == "W":
|
|
if self.master.stream:
|
|
self.master.stop_stream()
|
|
else:
|
|
signals.status_prompt_path.send(
|
|
self,
|
|
prompt = "Stream flows to",
|
|
callback = self.master.start_stream_to_path
|
|
)
|
|
else:
|
|
return urwid.ListBox.keypress(self, size, key)
|