codekingpro/portable-devtools

1import urwid2 3import mitmproxy.flow4from mitmproxy import http5from mitmproxy.tools.console import common6from mitmproxy.tools.console import searchable7from mitmproxy.utils import human8from mitmproxy.utils import strutils9 10 11def maybe_timestamp(base, attr):12    if base is not None and getattr(base, attr):13        return human.format_timestamp_with_milli(getattr(base, attr))14    else:15        # in mitmdump we serialize before a connection is closed.16        # loading those flows at a later point shouldn't display "active".17        # We also use a ndash (and not a regular dash) so that it is sorted18        # after other timestamps. We may need to revisit that in the future if it turns out19        # to render ugly in consoles.20        return "–"21 22 23def flowdetails(state, flow: mitmproxy.flow.Flow):24    text = []25 26    sc = flow.server_conn27    cc = flow.client_conn28    req: http.Request | None29    resp: http.Response | None30    if isinstance(flow, http.HTTPFlow):31        req = flow.request32        resp = flow.response33    else:34        req = None35        resp = None36    metadata = flow.metadata37    comment = flow.comment38 39    if comment:40        text.append(urwid.Text([("head", "Comment: "), ("text", comment)]))41 42    if metadata is not None and len(metadata) > 0:43        parts = [(str(k), repr(v)) for k, v in metadata.items()]44        text.append(urwid.Text([("head", "Metadata:")]))45        text.extend(common.format_keyvals(parts, indent=4))46 47    if sc is not None and sc.peername:48        text.append(urwid.Text([("head", "Server Connection:")]))49        parts = [50            ("Address", human.format_address(sc.address)),51        ]52        if sc.peername:53            parts.append(("Resolved Address", human.format_address(sc.peername)))54        if resp:55            parts.append(("HTTP Version", resp.http_version))56        if sc.alpn:57            parts.append(("ALPN", strutils.bytes_to_escaped_str(sc.alpn)))58 59        text.extend(common.format_keyvals(parts, indent=4))60 61        if sc.certificate_list:62            c = sc.certificate_list[0]63            text.append(urwid.Text([("head", "Server Certificate:")]))64            parts = [65                ("Type", "%s, %s bits" % c.keyinfo),66                ("SHA256 digest", c.fingerprint().hex(" ")),67                ("Valid from", str(c.notbefore)),68                ("Valid to", str(c.notafter)),69                ("Serial", str(c.serial)),70                (71                    "Subject",72                    urwid.Pile(73                        common.format_keyvals(c.subject, key_format="highlight")74                    ),75                ),76                (77                    "Issuer",78                    urwid.Pile(common.format_keyvals(c.issuer, key_format="highlight")),79                ),80            ]81 82            if c.altnames:83                parts.append(("Alt names", ", ".join(str(x.value) for x in c.altnames)))84            text.extend(common.format_keyvals(parts, indent=4))85 86    if cc is not None:87        text.append(urwid.Text([("head", "Client Connection:")]))88 89        parts = [90            ("Address", human.format_address(cc.peername)),91        ]92        if req:93            parts.append(("HTTP Version", req.http_version))94        if cc.tls_version:95            parts.append(("TLS Version", cc.tls_version))96        if cc.sni:97            parts.append(("Server Name Indication", cc.sni))98        if cc.cipher:99            parts.append(("Cipher Name", cc.cipher))100        if cc.alpn:101            parts.append(("ALPN", strutils.bytes_to_escaped_str(cc.alpn)))102 103        text.extend(common.format_keyvals(parts, indent=4))104 105    parts = []106 107    if cc is not None and cc.timestamp_start:108        parts.append(109            ("Client conn. established", maybe_timestamp(cc, "timestamp_start"))110        )111        if cc.tls_established:112            parts.append(113                (114                    "Client conn. TLS handshake",115                    maybe_timestamp(cc, "timestamp_tls_setup"),116                )117            )118        parts.append(("Client conn. closed", maybe_timestamp(cc, "timestamp_end")))119 120    if sc is not None and sc.timestamp_start:121        parts.append(("Server conn. initiated", maybe_timestamp(sc, "timestamp_start")))122        parts.append(123            ("Server conn. TCP handshake", maybe_timestamp(sc, "timestamp_tcp_setup"))124        )125        if sc.tls_established:126            parts.append(127                (128                    "Server conn. TLS handshake",129                    maybe_timestamp(sc, "timestamp_tls_setup"),130                )131            )132        parts.append(("Server conn. closed", maybe_timestamp(sc, "timestamp_end")))133 134    if req is not None and req.timestamp_start:135        parts.append(("First request byte", maybe_timestamp(req, "timestamp_start")))136        parts.append(("Request complete", maybe_timestamp(req, "timestamp_end")))137 138    if resp is not None and resp.timestamp_start:139        parts.append(("First response byte", maybe_timestamp(resp, "timestamp_start")))140        parts.append(("Response complete", maybe_timestamp(resp, "timestamp_end")))141 142    if parts:143        # sort operations by timestamp144        parts = sorted(parts, key=lambda p: p[1])145 146        text.append(urwid.Text([("head", "Timing:")]))147        text.extend(common.format_keyvals(parts, indent=4))148 149    return searchable.Searchable(text)150