codekingpro/portable-devtools

1import sys2from functools import lru_cache3 4import urwid5 6import mitmproxy.flow7import mitmproxy.tools.console.master8import mitmproxy_rs.syntax_highlight9from mitmproxy import contentviews10from mitmproxy import ctx11from mitmproxy import dns12from mitmproxy import http13from mitmproxy import tcp14from mitmproxy import udp15from mitmproxy.dns import DNSMessage16from mitmproxy.tools.console import common17from mitmproxy.tools.console import flowdetailview18from mitmproxy.tools.console import layoutwidget19from mitmproxy.tools.console import searchable20from mitmproxy.tools.console import tabs21from mitmproxy.utils import strutils22 23 24class FlowViewHeader(urwid.WidgetWrap):25    def __init__(26        self,27        master: "mitmproxy.tools.console.master.ConsoleMaster",28    ) -> None:29        self.master = master30        self.focus_changed()31 32    def focus_changed(self):33        cols, _ = self.master.ui.get_cols_rows()34        if self.master.view.focus.flow:35            self._w = common.format_flow(36                self.master.view.focus.flow,37                render_mode=common.RenderMode.DETAILVIEW,38                hostheader=self.master.options.showhost,39            )40        else:41            self._w = urwid.Pile([])42 43 44class FlowDetails(tabs.Tabs):45    def __init__(self, master):46        self.master = master47        super().__init__([])48        self.show()49        self.last_displayed_body = None50        self.last_displayed_websocket_messages = None51        contentviews.registry.on_change.connect(self.contentview_changed)52 53    @property54    def view(self):55        return self.master.view56 57    @property58    def flow(self) -> mitmproxy.flow.Flow:59        return self.master.view.focus.flow60 61    def contentview_changed(self, view):62        # this is called when a contentview addon is live-reloaded.63        # we clear our cache and then rerender64        self._get_content_view.cache_clear()65        if self.master.window.current_window("flowview"):66            self.show()67 68    def focus_changed(self):69        f = self.flow70        if f:71            if isinstance(f, http.HTTPFlow):72                if f.websocket:73                    self.tabs = [74                        (self.tab_http_request, self.view_request),75                        (self.tab_http_response, self.view_response),76                        (self.tab_websocket_messages, self.view_websocket_messages),77                        (self.tab_details, self.view_details),78                    ]79                else:80                    self.tabs = [81                        (self.tab_http_request, self.view_request),82                        (self.tab_http_response, self.view_response),83                        (self.tab_details, self.view_details),84                    ]85            elif isinstance(f, tcp.TCPFlow):86                self.tabs = [87                    (self.tab_tcp_stream, self.view_message_stream),88                    (self.tab_details, self.view_details),89                ]90            elif isinstance(f, udp.UDPFlow):91                self.tabs = [92                    (self.tab_udp_stream, self.view_message_stream),93                    (self.tab_details, self.view_details),94                ]95            elif isinstance(f, dns.DNSFlow):96                self.tabs = [97                    (self.tab_dns_request, self.view_dns_request),98                    (self.tab_dns_response, self.view_dns_response),99                    (self.tab_details, self.view_details),100                ]101            self.show()102        else:103            # Get the top window from the focus stack (the currently active view).104            # If it's NOT the "flowlist", it's safe to pop back to the previous view.105            if self.master.window.focus_stack().stack[-1] != "flowlist":106                self.master.window.pop()107            # If it is the "flowlist", we’re already at the main view with no flows to show.108            # Popping now would close the last window and prompt app exit, so we remain on the empty flow list screen instead.109 110    def tab_http_request(self):111        flow = self.flow112        assert isinstance(flow, http.HTTPFlow)113        if self.flow.intercepted and not flow.response:114            return "Request intercepted"115        else:116            return "Request"117 118    def tab_http_response(self):119        flow = self.flow120        assert isinstance(flow, http.HTTPFlow)121 122        # there is no good way to detect what part of the flow is intercepted,123        # so we apply some heuristics to see if it's the HTTP response.124        websocket_started = flow.websocket and len(flow.websocket.messages) != 0125        response_is_intercepted = (126            self.flow.intercepted and flow.response and not websocket_started127        )128        if response_is_intercepted:129            return "Response intercepted"130        else:131            return "Response"132 133    def tab_dns_request(self) -> str:134        flow = self.flow135        assert isinstance(flow, dns.DNSFlow)136        if self.flow.intercepted and not flow.response:137            return "Request intercepted"138        else:139            return "Request"140 141    def tab_dns_response(self) -> str:142        flow = self.flow143        assert isinstance(flow, dns.DNSFlow)144        if self.flow.intercepted and flow.response:145            return "Response intercepted"146        else:147            return "Response"148 149    def tab_tcp_stream(self):150        return "TCP Stream"151 152    def tab_udp_stream(self):153        return "UDP Stream"154 155    def tab_websocket_messages(self):156        flow = self.flow157        assert isinstance(flow, http.HTTPFlow)158        assert flow.websocket159 160        if self.flow.intercepted and len(flow.websocket.messages) != 0:161            return "WebSocket Messages intercepted"162        else:163            return "WebSocket Messages"164 165    def tab_details(self):166        return "Detail"167 168    def view_request(self):169        flow = self.flow170        assert isinstance(flow, http.HTTPFlow)171        return self.conn_text(flow.request)172 173    def view_response(self):174        flow = self.flow175        assert isinstance(flow, http.HTTPFlow)176        return self.conn_text(flow.response)177 178    def view_dns_request(self):179        flow = self.flow180        assert isinstance(flow, dns.DNSFlow)181        return self.dns_message_text("request", flow.request)182 183    def view_dns_response(self):184        flow = self.flow185        assert isinstance(flow, dns.DNSFlow)186        return self.dns_message_text("response", flow.response)187 188    def _contentview_status_bar(self, description: str, viewmode: str):189        cols = [190            urwid.Text(191                [192                    ("heading", description),193                ]194            ),195            urwid.Text(196                [197                    " ",198                    ("heading", "["),199                    ("heading_key", "m"),200                    ("heading", (":%s]" % viewmode)),201                ],202                align="right",203            ),204        ]205        contentview_status_bar = urwid.AttrMap(urwid.Columns(cols), "heading")206        return contentview_status_bar207 208    FROM_CLIENT_MARKER = ("from_client", f"{common.SYMBOL_FROM_CLIENT} ")209    TO_CLIENT_MARKER = ("to_client", f"{common.SYMBOL_TO_CLIENT} ")210 211    def view_websocket_messages(self):212        flow = self.flow213        assert isinstance(flow, http.HTTPFlow)214        assert flow.websocket is not None215 216        if not flow.websocket.messages:217            return searchable.Searchable([urwid.Text(("highlight", "No messages."))])218 219        viewmode = self.master.commands.call("console.flowview.mode")220 221        widget_lines = []222        for m in flow.websocket.messages:223            pretty = contentviews.prettify_message(m, flow, viewmode)224            chunks = mitmproxy_rs.syntax_highlight.highlight(225                pretty.text,226                language=pretty.syntax_highlight,227            )228            if m.from_client:229                marker = self.FROM_CLIENT_MARKER230            else:231                marker = self.TO_CLIENT_MARKER232            widget_lines.append(urwid.Text([marker, *chunks]))233 234        if flow.websocket.closed_by_client is not None:235            widget_lines.append(236                urwid.Text(237                    [238                        (239                            self.FROM_CLIENT_MARKER240                            if flow.websocket.closed_by_client241                            else self.TO_CLIENT_MARKER242                        ),243                        (244                            "alert"245                            if flow.websocket.close_code in (1000, 1001, 1005)246                            else "error",247                            f"Connection closed: {flow.websocket.close_code} {flow.websocket.close_reason}",248                        ),249                    ]250                )251            )252 253        if flow.intercepted:254            markup = widget_lines[-1].get_text()[0]255            widget_lines[-1].set_text(("intercept", markup))256 257        widget_lines.insert(258            0, self._contentview_status_bar(viewmode.capitalize(), viewmode)259        )260 261        if (last_view := self.last_displayed_websocket_messages) is not None:262            last_view.walker[:] = widget_lines263            view = last_view264        else:265            view = searchable.Searchable(widget_lines)266            self.last_displayed_websocket_messages = view267 268        return view269 270    def view_message_stream(self) -> urwid.Widget:271        flow = self.flow272        assert isinstance(flow, (tcp.TCPFlow, udp.UDPFlow))273 274        if not flow.messages:275            return searchable.Searchable([urwid.Text(("highlight", "No messages."))])276 277        viewmode = self.master.commands.call("console.flowview.mode")278 279        widget_lines = []280        for m in flow.messages:281            if m.from_client:282                marker = self.FROM_CLIENT_MARKER283            else:284                marker = self.TO_CLIENT_MARKER285            pretty = contentviews.prettify_message(m, flow, viewmode)286            chunks = mitmproxy_rs.syntax_highlight.highlight(287                pretty.text,288                language=pretty.syntax_highlight,289            )290 291            widget_lines.append(urwid.Text([marker, *chunks]))292 293        if flow.intercepted:294            markup = widget_lines[-1].get_text()[0]295            widget_lines[-1].set_text(("intercept", markup))296 297        widget_lines.insert(298            0, self._contentview_status_bar(viewmode.capitalize(), viewmode)299        )300 301        return searchable.Searchable(widget_lines)302 303    def view_details(self):304        return flowdetailview.flowdetails(self.view, self.flow)305 306    def content_view(307        self, viewmode: str, message: http.Message308    ) -> tuple[str, list[urwid.Text]]:309        if message.raw_content is None:310            return "", [urwid.Text([("error", "[content missing]")])]311 312        if message.raw_content == b"":313            if isinstance(message, http.Request):314                query = getattr(message, "query", "")315                if not query:316                    # No body and no query params317                    return "", [urwid.Text("No request content")]318                # else: there are query params -> fall through to render them319            else:320                return "", [urwid.Text("No content")]321 322        full = self.master.commands.execute(323            "view.settings.getval @focus fullcontents false"324        )325 326        if full == "true":327            limit = sys.maxsize328        else:329            limit = ctx.options.content_view_lines_cutoff330 331        flow_modify_cache_invalidation = hash(332            (333                message.raw_content,334                message.headers.fields,335                getattr(message, "path", None),336            )337        )338        # we need to pass the message off-band because it's not hashable339        self._get_content_view_message = message340        return self._get_content_view(viewmode, limit, flow_modify_cache_invalidation)341 342    @lru_cache(maxsize=200)343    def _get_content_view(344        self, viewmode: str, max_lines: int, _345    ) -> tuple[str, list[urwid.Text]]:346        message: http.Message = self._get_content_view_message347        self._get_content_view_message = None  # type: ignore[assignment]348 349        pretty = contentviews.prettify_message(message, self.flow, viewmode)350        cut_off = strutils.cut_after_n_lines(pretty.text, max_lines)351 352        chunks = mitmproxy_rs.syntax_highlight.highlight(353            cut_off,354            language=pretty.syntax_highlight,355        )356 357        text_objects = [urwid.Text(chunks)]358        if len(cut_off) < len(pretty.text):359            text_objects.append(360                urwid.Text(361                    [362                        (363                            "highlight",364                            "Stopped displaying data after %d lines. Press "365                            % max_lines,366                        ),367                        ("key", "f"),368                        ("highlight", " to load all data."),369                    ]370                )371            )372 373        return f"{pretty.view_name} {pretty.description}", text_objects374 375    def conn_text(self, conn):376        if conn:377            hdrs = []378            for k, v in conn.headers.fields:379                # This will always force an ascii representation of headers. For example, if the server sends a380                #381                #     X-Authors: Made with ❤ in Hamburg382                #383                # header, mitmproxy will display the following:384                #385                #     X-Authors: Made with \xe2\x9d\xa4 in Hamburg.386                #387                # The alternative would be to just use the header's UTF-8 representation and maybe388                # do `str.replace("\t", "\\t")` to exempt tabs from urwid's special characters escaping [1].389                # That would in some terminals allow rendering UTF-8 characters, but the mapping390                # wouldn't be bijective, i.e. a user couldn't distinguish "\\t" and "\t".391                # Also, from a security perspective, a mitmproxy user couldn't be fooled by homoglyphs.392                #393                # 1) https://github.com/mitmproxy/mitmproxy/issues/1833394                #    https://github.com/urwid/urwid/blob/6608ee2c9932d264abd1171468d833b7a4082e13/urwid/display_common.py#L35-L36,395 396                k = strutils.bytes_to_escaped_str(k) + ":"397                v = strutils.bytes_to_escaped_str(v)398                hdrs.append((k, v))399            txt = common.format_keyvals(hdrs, key_format="header")400            viewmode = self.master.commands.call("console.flowview.mode")401            msg, body = self.content_view(viewmode, conn)402 403            txt.append(self._contentview_status_bar(msg, viewmode))404            txt.extend(body)405        else:406            txt = [407                urwid.Text(""),408                urwid.Text(409                    [410                        ("highlight", "No response. Press "),411                        ("key", "e"),412                        ("highlight", " and edit any aspect to add one."),413                    ]414                ),415            ]416        return searchable.Searchable(txt)417 418    def dns_message_text(419        self, type: str, message: DNSMessage | None420    ) -> searchable.Searchable:421        """422        Alternative:423        if not message:424            return searchable.Searchable([urwid.Text(("highlight", f"No {typ}."))])425 426        viewmode = self.master.commands.call("console.flowview.mode")427        pretty = contentviews.prettify_message(message, flow, viewmode)428        chunks = mitmproxy_rs.syntax_highlight.highlight(429            pretty.text,430            language=pretty.syntax_highlight,431        )432 433        widget_lines = [434            self._contentview_status_bar(viewmode.capitalize(), viewmode),435            urwid.Text(chunks)436        ]437        return searchable.Searchable(widget_lines)438        """439        # Keep in sync with web/src/js/components/FlowView/DnsMessages.tsx440        if message:441 442            def rr_text(rr: dns.ResourceRecord):443                return urwid.Text(444                    f"  {rr.name} {dns.types.to_str(rr.type)} {dns.classes.to_str(rr.class_)} {rr.ttl} {rr}"445                )446 447            txt = []448            txt.append(449                urwid.Text(450                    "{recursive}Question".format(451                        recursive="Recursive " if message.recursion_desired else "",452                    )453                )454            )455            txt.extend(456                urwid.Text(457                    f"  {q.name} {dns.types.to_str(q.type)} {dns.classes.to_str(q.class_)}"458                )459                for q in message.questions460            )461            txt.append(urwid.Text(""))462            txt.append(463                urwid.Text(464                    "{authoritative}{recursive}Answer".format(465                        authoritative="Authoritative "466                        if message.authoritative_answer467                        else "",468                        recursive="Recursive " if message.recursion_available else "",469                    )470                )471            )472            txt.extend(map(rr_text, message.answers))473            txt.append(urwid.Text(""))474            txt.append(urwid.Text("Authority"))475            txt.extend(map(rr_text, message.authorities))476            txt.append(urwid.Text(""))477            txt.append(urwid.Text("Addition"))478            txt.extend(map(rr_text, message.additionals))479            return searchable.Searchable(txt)480        else:481            return searchable.Searchable([urwid.Text(("highlight", f"No {type}."))])482 483 484class FlowView(urwid.Frame, layoutwidget.LayoutWidget):485    keyctx = "flowview"486    title = "Flow Details"487 488    def __init__(self, master):489        super().__init__(490            FlowDetails(master),491            header=FlowViewHeader(master),492        )493        self.master = master494 495    def focus_changed(self, *args, **kwargs):496        self.body.focus_changed()497        self.header.focus_changed()498