mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-28 02:31:29 +00:00
34ec2d1370
Press E followed by p to copy code to clipboard. The code uses the python requests module. Only GET methods are currently supported, but can easily be extended to POST etc.
393 lines
13 KiB
Python
393 lines
13 KiB
Python
from __future__ import absolute_import
|
|
import urwid
|
|
|
|
from netlib import http
|
|
import netlib.utils
|
|
|
|
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", "export"),
|
|
("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(len(self.master.eventlist) - 1)
|
|
elif key == "g":
|
|
self.set_focus(0)
|
|
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 == "E":
|
|
signals.status_prompt_onekey.send(
|
|
self,
|
|
prompt = "Export",
|
|
keys = (
|
|
("as curl command", "c"),
|
|
("as python code", "p"),
|
|
),
|
|
callback = common.export_prompt,
|
|
args = (self.flow,)
|
|
)
|
|
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 = netlib.utils.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)
|