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

add very simply tcp detailview

Browse Source
This commit is contained in:
Maximilian Hils 2020-04-12 00:55:04 +02:00
parent d33857588c
commit d1cb0dbec5
5 changed files with 102 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
mitmproxy/addons/clientplayback.py
View File

@ -127,15 +127,18 @@ class ClientPlayback:
self.q = queue.Queue()
self.thread: RequestReplayThread = None
def check(self, f: http.HTTPFlow):
def check(self, f: flow.Flow):
if f.live:
return "Can't replay live flow."
if f.intercepted:
return "Can't replay intercepted flow."
if not f.request:
return "Can't replay flow with missing request."
if f.request.raw_content is None:
return "Can't replay flow with missing content."
if isinstance(f, http.HTTPFlow):
if not f.request:
return "Can't replay flow with missing request."
if f.request.raw_content is None:
return "Can't replay flow with missing content."
else:
return "Can only replay HTTP flows."
def load(self, loader):
loader.add_option(

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

@ -9,6 +9,7 @@ from mitmproxy import exceptions
from mitmproxy import flow
from mitmproxy import http
from mitmproxy import log
from mitmproxy import tcp
from mitmproxy.tools.console import keymap
from mitmproxy.tools.console import overlay
from mitmproxy.tools.console import signals
@ -334,9 +335,10 @@ class ConsoleAddon:
@command.command("console.view.flow")
def view_flow(self, flow: flow.Flow) -> None:
"""View a flow."""
if hasattr(flow, "request"):
# FIME: Also set focus?
if isinstance(flow, (http.HTTPFlow, tcp.TCPFlow)):
self.master.switch_view("flowview")
else:
ctx.log.warn(f"No detail view for {type(flow).__name__}.")
@command.command("console.exit")
def exit(self) -> None:

23
mitmproxy/tools/console/flowdetailview.py
View File

@ -1,5 +1,6 @@
import urwid
import mitmproxy.flow
from mitmproxy import http
from mitmproxy.tools.console import common, searchable
from mitmproxy.utils import human
@ -13,13 +14,17 @@ def maybe_timestamp(base, attr):
return "active"
def flowdetails(state, flow: http.HTTPFlow):
def flowdetails(state, flow: mitmproxy.flow.Flow):
text = []
sc = flow.server_conn
cc = flow.client_conn
req = flow.request
resp = flow.response
if isinstance(flow, http.HTTPFlow):
req = flow.request
resp = flow.response
else:
req = None
resp = None
metadata = flow.metadata
if metadata is not None and len(metadata) > 0:
@ -126,6 +131,12 @@ def flowdetails(state, flow: http.HTTPFlow):
maybe_timestamp(cc, "timestamp_tls_setup")
)
)
parts.append(
(
"Client conn. closed",
maybe_timestamp(cc, "timestamp_end")
)
)
if sc is not None and sc.timestamp_start:
parts.append(
@ -147,6 +158,12 @@ def flowdetails(state, flow: http.HTTPFlow):
maybe_timestamp(sc, "timestamp_tls_setup")
)
)
parts.append(
(
"Server conn. closed",
maybe_timestamp(sc, "timestamp_end")
)
)
if req is not None and req.timestamp_start:
parts.append(

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

@ -32,9 +32,8 @@ class FlowItem(urwid.WidgetWrap):
def mouse_event(self, size, event, button, col, row, focus):
if event == "mouse press" and button == 1:
if self.flow.request:
self.master.commands.execute("console.view.flow @focus")
return True
self.master.commands.execute("console.view.flow @focus")
return True
def keypress(self, size, key):
return key

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

@ -5,9 +5,11 @@ from typing import Optional, Union # noqa
import urwid
import mitmproxy.flow
from mitmproxy import contentviews
from mitmproxy import ctx
from mitmproxy import http
from mitmproxy import tcp
from mitmproxy.tools.console import common
from mitmproxy.tools.console import layoutwidget
from mitmproxy.tools.console import flowdetailview
@ -24,8 +26,8 @@ class SearchError(Exception):
class FlowViewHeader(urwid.WidgetWrap):
def __init__(
self,
master: "mitmproxy.tools.console.master.ConsoleMaster",
self,
master: "mitmproxy.tools.console.master.ConsoleMaster",
) -> None:
self.master = master
self.focus_changed()
@ -49,45 +51,90 @@ class FlowDetails(tabs.Tabs):
self.show()
self.last_displayed_body = None
def focus_changed(self):
if self.master.view.focus.flow:
self.tabs = [
(self.tab_request, self.view_request),
(self.tab_response, self.view_response),
(self.tab_details, self.view_details),
]
self.show()
else:
self.master.window.pop()
@property
def view(self):
return self.master.view
@property
def flow(self):
def flow(self) -> mitmproxy.flow.Flow:
return self.master.view.focus.flow
def tab_request(self):
if self.flow.intercepted and not self.flow.response:
def focus_changed(self):
if self.flow:
if isinstance(self.flow, http.HTTPFlow):
self.tabs = [
(self.tab_http_request, self.view_request),
(self.tab_http_response, self.view_response),
(self.tab_details, self.view_details),
]
elif isinstance(self.flow, tcp.TCPFlow):
self.tabs = [
(self.tab_tcp_stream, self.view_tcp_stream),
(self.tab_details, self.view_details),
]
self.show()
else:
self.master.window.pop()
def tab_http_request(self):
flow = self.flow
assert isinstance(flow, http.HTTPFlow)
if self.flow.intercepted and not flow.response:
return "Request intercepted"
else:
return "Request"
def tab_response(self):
if self.flow.intercepted and self.flow.response:
def tab_http_response(self):
flow = self.flow
assert isinstance(flow, http.HTTPFlow)
if self.flow.intercepted and flow.response:
return "Response intercepted"
else:
return "Response"
def tab_tcp_stream(self):
return "TCP Stream"
def tab_details(self):
return "Detail"
def view_request(self):
return self.conn_text(self.flow.request)
flow = self.flow
assert isinstance(flow, http.HTTPFlow)
return self.conn_text(flow.request)
def view_response(self):
return self.conn_text(self.flow.response)
flow = self.flow
assert isinstance(flow, http.HTTPFlow)
return self.conn_text(flow.response)
def view_tcp_stream(self) -> urwid.Widget:
flow = self.flow
assert isinstance(flow, tcp.TCPFlow)
if not flow.messages:
return searchable.Searchable([urwid.Text(("highlight", "No messages."))])
from_client = None
messages = []
for message in flow.messages:
if message.from_client is not from_client:
messages.append(message.content)
from_client = message.from_client
else:
messages[-1] += message.content
from_client = flow.messages[0].from_client
parts = []
for message in messages:
parts.append(
(
"head" if from_client else "key",
message
)
)
from_client = not from_client
return searchable.Searchable([urwid.Text(parts)])
def view_details(self):
return flowdetailview.flowdetails(self.view, self.flow)
@ -226,7 +273,7 @@ class FlowView(urwid.Frame, layoutwidget.LayoutWidget):
def __init__(self, master):
super().__init__(
FlowDetails(master),
header = FlowViewHeader(master),
header=FlowViewHeader(master),
)
self.master = master