codekingpro/portable-devtools

1import enum2import math3import platform4from collections.abc import Iterable5from functools import lru_cache6 7import urwid.util8from publicsuffix2 import get_sld9from publicsuffix2 import get_tld10 11from mitmproxy import dns12from mitmproxy import flow13from mitmproxy.dns import DNSFlow14from mitmproxy.http import HTTPFlow15from mitmproxy.tcp import TCPFlow16from mitmproxy.udp import UDPFlow17from mitmproxy.utils import emoji18from mitmproxy.utils import human19 20# Detect Windows Subsystem for Linux and Windows21IS_WINDOWS_OR_WSL = (22    "Microsoft" in platform.platform() or "Windows" in platform.platform()23)24 25 26def is_keypress(k):27    """28    Is this input event a keypress?29    """30    if isinstance(k, str):31        return True32 33 34def highlight_key(text, key, textattr="text", keyattr="key"):35    lst = []36    parts = text.split(key, 1)37    if parts[0]:38        lst.append((textattr, parts[0]))39    lst.append((keyattr, key))40    if parts[1]:41        lst.append((textattr, parts[1]))42    return lst43 44 45KEY_MAX = 3046 47 48def format_keyvals(49    entries: Iterable[tuple[str, None | str | urwid.Widget]],50    key_format: str = "key",51    value_format: str = "text",52    indent: int = 0,53) -> list[urwid.Columns]:54    """55    Format a list of (key, value) tuples.56 57    Args:58        entries: The list to format. keys must be strings, values can also be None or urwid widgets.59            The latter makes it possible to use the result of format_keyvals() as a value.60        key_format: The display attribute for the key.61        value_format: The display attribute for the value.62        indent: Additional indent to apply.63    """64    max_key_len = max((len(k) for k, v in entries if k is not None), default=0)65    max_key_len = min(max_key_len, KEY_MAX)66 67    if indent > 2:68        indent -= 2  # We use dividechars=2 below, which already adds two empty spaces69 70    ret = []71    for k, v in entries:72        if v is None:73            v = urwid.Text("")74        elif not isinstance(v, urwid.Widget):75            v = urwid.Text([(value_format, v)])76        ret.append(77            urwid.Columns(78                [79                    ("fixed", indent, urwid.Text("")),80                    ("fixed", max_key_len, urwid.Text([(key_format, k)])),81                    v,82                ],83                dividechars=2,84            )85        )86    return ret87 88 89def fcol(s: str, attr: str) -> tuple[str, int, urwid.Text]:90    s = str(s)91    return ("fixed", len(s), urwid.Text([(attr, s)]))92 93 94if urwid.util.detected_encoding:95    SYMBOL_REPLAY = "\u21ba"96    SYMBOL_RETURN = "\u2190"97    SYMBOL_MARK = "\u25cf"98    SYMBOL_UP = "\u21e7"99    SYMBOL_DOWN = "\u21e9"100    SYMBOL_ELLIPSIS = "\u2026"101    SYMBOL_FROM_CLIENT = "\u21d2"102    SYMBOL_TO_CLIENT = "\u21d0"103else:104    SYMBOL_REPLAY = "[r]"105    SYMBOL_RETURN = "<-"106    SYMBOL_MARK = "#"107    SYMBOL_UP = "^"108    SYMBOL_DOWN = " "109    SYMBOL_ELLIPSIS = "~"110    SYMBOL_FROM_CLIENT = "->"111    SYMBOL_TO_CLIENT = "<-"112 113SCHEME_STYLES = {114    "http": "scheme_http",115    "https": "scheme_https",116    "ws": "scheme_ws",117    "wss": "scheme_wss",118    "tcp": "scheme_tcp",119    "udp": "scheme_udp",120    "dns": "scheme_dns",121    "quic": "scheme_quic",122}123HTTP_REQUEST_METHOD_STYLES = {124    "GET": "method_get",125    "POST": "method_post",126    "DELETE": "method_delete",127    "HEAD": "method_head",128    "PUT": "method_put",129}130HTTP_RESPONSE_CODE_STYLE = {131    2: "code_200",132    3: "code_300",133    4: "code_400",134    5: "code_500",135}136 137 138class RenderMode(enum.Enum):139    TABLE = 1140    """The flow list in table format, i.e. one row per flow."""141    LIST = 2142    """The flow list in list format, i.e. potentially multiple rows per flow."""143    DETAILVIEW = 3144    """The top lines in the detail view."""145 146 147def fixlen(s: str, maxlen: int) -> str:148    if len(s) <= maxlen:149        return s.ljust(maxlen)150    else:151        return s[0 : maxlen - len(SYMBOL_ELLIPSIS)] + SYMBOL_ELLIPSIS152 153 154def fixlen_r(s: str, maxlen: int) -> str:155    if len(s) <= maxlen:156        return s.rjust(maxlen)157    else:158        return SYMBOL_ELLIPSIS + s[len(s) - maxlen + len(SYMBOL_ELLIPSIS) :]159 160 161def render_marker(marker: str) -> str:162    rendered = emoji.emoji.get(marker, SYMBOL_MARK)163 164    # The marker can only be one glyph. Some emoji that use zero-width joiners (ZWJ)165    # will not be rendered as a single glyph and instead will show166    # multiple glyphs. Just use the first glyph as a fallback.167    # https://emojipedia.org/emoji-zwj-sequence/168    return rendered[0]169 170 171class TruncatedText(urwid.Widget):172    def __init__(self, text, attr, align="left"):173        self.text = text174        self.attr = attr175        self.align = align176        super().__init__()177 178    def pack(self, size, focus=False):179        return (len(self.text), 1)180 181    def rows(self, size, focus=False):182        return 1183 184    def render(self, size, focus=False):185        text = self.text186        attr = self.attr187        if self.align == "right":188            text = text[::-1]189            attr = attr[::-1]190 191        text_len = urwid.calc_width(text, 0, len(text))192        if size is not None and len(size) > 0:193            width = size[0]194        else:195            width = text_len196 197        if width >= text_len:198            remaining = width - text_len199            if remaining > 0:200                c_text = text + " " * remaining201                c_attr = attr + [("text", remaining)]202            else:203                c_text = text204                c_attr = attr205        else:206            trim = urwid.util.calc_trim_text(text, 0, width - 1, 0, width - 1)207            visible_text = text[0 : trim[1]]208            if trim[3] == 1:209                visible_text += " "210            c_text = visible_text + SYMBOL_ELLIPSIS211            c_attr = urwid.util.rle_subseg(attr, 0, len(visible_text.encode())) + [212                ("focus", len(SYMBOL_ELLIPSIS.encode()))213            ]214 215        if self.align == "right":216            c_text = c_text[::-1]217            c_attr = c_attr[::-1]218 219        return urwid.TextCanvas([c_text.encode()], [c_attr], maxcol=width)220 221 222def truncated_plain(text, attr, align="left"):223    return TruncatedText(text, [(attr, len(text.encode()))], align)224 225 226# Work around https://github.com/urwid/urwid/pull/330227def rle_append_beginning_modify(rle, a_r):228    """229    Append (a, r) (unpacked from *a_r*) to BEGINNING of rle.230    Merge with first run when possible231 232    MODIFIES rle parameter contents. Returns None.233    """234    a, r = a_r235    if not rle:236        rle[:] = [(a, r)]237    else:238        al, run = rle[0]239        if a == al:240            rle[0] = (a, run + r)241        else:242            rle[0:0] = [(a, r)]243 244 245def colorize_host(host: str):246    if not host:247        return []248 249    tld = get_tld(host)250    sld = get_sld(host)251 252    attr: list = []253 254    tld_size = len(tld)255    sld_size = len(sld) - tld_size256 257    for letter in reversed(range(len(host))):258        character = host[letter]259        if tld_size > 0:260            style = "url_domain"261            tld_size -= 1262        elif tld_size == 0:263            style = "text"264            tld_size -= 1265        elif sld_size > 0:266            sld_size -= 1267            style = "url_extension"268        else:269            style = "text"270        rle_append_beginning_modify(attr, (style, len(character.encode())))271    return attr272 273 274def colorize_req(s: str):275    path = s.split("?", 2)[0]276    i_query = len(path)277    i_last_slash = path.rfind("/")278    i_ext = path[i_last_slash + 1 :].rfind(".")279    i_ext = i_last_slash + i_ext if i_ext >= 0 else len(s)280    in_val = False281    attr: list = []282    for i in range(len(s)):283        c = s[i]284        if (285            (i < i_query and c == "/")286            or (i < i_query and i > i_last_slash and c == ".")287            or (i == i_query)288        ):289            a = "url_punctuation"290        elif i > i_query:291            if in_val:292                if c == "&":293                    in_val = False294                    a = "url_punctuation"295                else:296                    a = "url_query_value"297            else:298                if c == "=":299                    in_val = True300                    a = "url_punctuation"301                else:302                    a = "url_query_key"303        elif i > i_ext:304            a = "url_extension"305        elif i > i_last_slash:306            a = "url_filename"307        else:308            a = "text"309        urwid.util.rle_append_modify(attr, (a, len(c.encode())))310    return attr311 312 313def format_http_content_type(content_type: str) -> tuple[str, str]:314    content_type = content_type.split(";")[0]315    if content_type.endswith("/javascript"):316        style = "content_script"317    elif content_type.startswith("text/"):318        style = "content_text"319    elif (320        content_type.startswith("image/")321        or content_type.startswith("video/")322        or content_type.startswith("font/")323        or "/x-font-" in content_type324    ):325        style = "content_media"326    elif content_type.endswith("/json") or content_type.endswith("/xml"):327        style = "content_data"328    elif content_type.startswith("application/"):329        style = "content_raw"330    else:331        style = "content_other"332    return content_type, style333 334 335def format_duration(duration: float) -> tuple[str, str]:336    pretty_duration = human.pretty_duration(duration)337    style = "gradient_%02d" % int(338        99 - 100 * min(math.log2(max(1.0, 1000 * duration)) / 12, 0.99)339    )340    return pretty_duration, style341 342 343def format_size(num_bytes: int) -> tuple[str, str]:344    pretty_size = human.pretty_size(num_bytes)345    style = "gradient_%02d" % int(99 - 100 * min(math.log2(1 + num_bytes) / 20, 0.99))346    return pretty_size, style347 348 349def format_left_indicators(*, focused: bool, intercepted: bool, timestamp: float):350    indicators: list[str | tuple[str, str]] = []351    if focused:352        indicators.append(("focus", ">>"))353    else:354        indicators.append("  ")355    pretty_timestamp = human.format_timestamp(timestamp)[-8:]356    if intercepted:357        indicators.append(("intercept", pretty_timestamp))358    else:359        indicators.append(("text", pretty_timestamp))360    return "fixed", 10, urwid.Text(indicators)361 362 363def format_right_indicators(364    *,365    replay: bool,366    marked: str,367):368    indicators: list[str | tuple[str, str]] = []369    if replay:370        indicators.append(("replay", SYMBOL_REPLAY))371    else:372        indicators.append(" ")373    if bool(marked):374        indicators.append(("mark", render_marker(marked)))375    else:376        indicators.append("  ")377    return "fixed", 3, urwid.Text(indicators)378 379 380@lru_cache(maxsize=800)381def format_http_flow_list(382    *,383    render_mode: RenderMode,384    focused: bool,385    marked: str,386    is_replay: bool,387    request_method: str,388    request_scheme: str,389    request_host: str,390    request_path: str,391    request_url: str,392    request_http_version: str,393    request_timestamp: float,394    request_is_push_promise: bool,395    intercepted: bool,396    response_code: int | None,397    response_reason: str | None,398    response_content_length: int | None,399    response_content_type: str | None,400    duration: float | None,401    error_message: str | None,402) -> urwid.Widget:403    req = []404 405    if render_mode is RenderMode.DETAILVIEW:406        req.append(fcol(human.format_timestamp(request_timestamp), "highlight"))407    else:408        if focused:409            req.append(fcol(">>", "focus"))410        else:411            req.append(fcol("  ", "focus"))412 413    method_style = HTTP_REQUEST_METHOD_STYLES.get(request_method, "method_other")414    req.append(fcol(request_method, method_style))415 416    if request_is_push_promise:417        req.append(fcol("PUSH_PROMISE", "method_http2_push"))418 419    preamble_len = sum(x[1] for x in req) + len(req) - 1420 421    if request_http_version not in ("HTTP/1.0", "HTTP/1.1"):422        request_url += " " + request_http_version423    if intercepted and not response_code:424        url_style = "intercept"425    elif response_code or error_message:426        url_style = "text"427    else:428        url_style = "title"429 430    if render_mode is RenderMode.DETAILVIEW:431        req.append(urwid.Text([(url_style, request_url)]))432    else:433        req.append(truncated_plain(request_url, url_style))434 435    req.append(format_right_indicators(replay=is_replay, marked=marked))436 437    resp = [("fixed", preamble_len, urwid.Text(""))]438    if response_code:439        if intercepted:440            style = "intercept"441        else:442            style = ""443 444        status_style = style or HTTP_RESPONSE_CODE_STYLE.get(445            response_code // 100, "code_other"446        )447        resp.append(fcol(SYMBOL_RETURN, status_style))448        resp.append(fcol(str(response_code), status_style))449        if response_reason and render_mode is RenderMode.DETAILVIEW:450            resp.append(fcol(response_reason, status_style))451 452        if response_content_type:453            ct, ct_style = format_http_content_type(response_content_type)454            resp.append(fcol(ct, style or ct_style))455 456        if response_content_length:457            size, size_style = format_size(response_content_length)458        elif response_content_length == 0:459            size = "[no content]"460            size_style = "text"461        else:462            size = "[content missing]"463            size_style = "text"464        resp.append(fcol(size, style or size_style))465 466        if duration:467            dur, dur_style = format_duration(duration)468            resp.append(fcol(dur, style or dur_style))469    elif error_message:470        resp.append(fcol(SYMBOL_RETURN, "error"))471        resp.append(urwid.Text([("error", error_message)]))472 473    return urwid.Pile(474        [urwid.Columns(req, dividechars=1), urwid.Columns(resp, dividechars=1)]475    )476 477 478@lru_cache(maxsize=800)479def format_http_flow_table(480    *,481    render_mode: RenderMode,482    focused: bool,483    marked: str,484    is_replay: str | None,485    request_method: str,486    request_scheme: str,487    request_host: str,488    request_path: str,489    request_url: str,490    request_http_version: str,491    request_timestamp: float,492    request_is_push_promise: bool,493    intercepted: bool,494    response_code: int | None,495    response_reason: str | None,496    response_content_length: int | None,497    response_content_type: str | None,498    duration: float | None,499    error_message: str | None,500) -> urwid.Widget:501    items = [502        format_left_indicators(503            focused=focused, intercepted=intercepted, timestamp=request_timestamp504        )505    ]506 507    if intercepted and not response_code:508        request_style = "intercept"509    else:510        request_style = ""511 512    scheme_style = request_style or SCHEME_STYLES.get(request_scheme, "scheme_other")513    items.append(fcol(fixlen(request_scheme.upper(), 5), scheme_style))514 515    if request_is_push_promise:516        method_style = "method_http2_push"517    else:518        method_style = request_style or HTTP_REQUEST_METHOD_STYLES.get(519            request_method, "method_other"520        )521    items.append(fcol(fixlen(request_method, 4), method_style))522 523    items.append(524        (525            "weight",526            0.25,527            TruncatedText(request_host, colorize_host(request_host), "right"),528        )529    )530    items.append(531        ("weight", 1.0, TruncatedText(request_path, colorize_req(request_path), "left"))532    )533 534    if intercepted and response_code:535        response_style = "intercept"536    else:537        response_style = ""538 539    if response_code:540        status = str(response_code)541        status_style = response_style or HTTP_RESPONSE_CODE_STYLE.get(542            response_code // 100, "code_other"543        )544 545        if response_content_length and response_content_type:546            content, content_style = format_http_content_type(response_content_type)547            content_style = response_style or content_style548        elif response_content_length:549            content = ""550            content_style = "content_none"551        elif response_content_length == 0:552            content = "[no content]"553            content_style = "content_none"554        else:555            content = "[content missing]"556            content_style = "content_none"557 558    elif error_message:559        status = "err"560        status_style = "error"561        content = error_message562        content_style = "error"563 564    else:565        status = ""566        status_style = "text"567        content = ""568        content_style = ""569 570    items.append(fcol(fixlen(status, 3), status_style))571    items.append(("weight", 0.15, truncated_plain(content, content_style, "right")))572 573    if response_content_length:574        size, size_style = format_size(response_content_length)575        items.append(fcol(fixlen_r(size, 5), response_style or size_style))576    else:577        items.append(("fixed", 5, urwid.Text("")))578 579    if duration:580        duration_pretty, duration_style = format_duration(duration)581        items.append(582            fcol(fixlen_r(duration_pretty, 5), response_style or duration_style)583        )584    else:585        items.append(("fixed", 5, urwid.Text("")))586 587    items.append(588        format_right_indicators(589            replay=bool(is_replay),590            marked=marked,591        )592    )593    return urwid.Columns(items, dividechars=1, min_width=15)594 595 596@lru_cache(maxsize=800)597def format_message_flow(598    *,599    render_mode: RenderMode,600    focused: bool,601    timestamp_start: float,602    marked: str,603    protocol: str,604    client_address,605    server_address,606    total_size: int,607    duration: float | None,608    error_message: str | None,609):610    conn = f"{human.format_address(client_address)} <-> {human.format_address(server_address)}"611 612    items = []613 614    if render_mode in (RenderMode.TABLE, RenderMode.DETAILVIEW):615        items.append(616            format_left_indicators(617                focused=focused, intercepted=False, timestamp=timestamp_start618            )619        )620    else:621        if focused:622            items.append(fcol(">>", "focus"))623        else:624            items.append(fcol("  ", "focus"))625 626    if render_mode is RenderMode.TABLE:627        items.append(fcol(fixlen(protocol.upper(), 5), SCHEME_STYLES[protocol]))628    else:629        items.append(fcol(protocol.upper(), SCHEME_STYLES[protocol]))630 631    items.append(("weight", 1.0, truncated_plain(conn, "text", "left")))632    if error_message:633        items.append(("weight", 1.0, truncated_plain(error_message, "error", "left")))634 635    if total_size:636        size, size_style = format_size(total_size)637        items.append(fcol(fixlen_r(size, 5), size_style))638    else:639        items.append(("fixed", 5, urwid.Text("")))640 641    if duration:642        duration_pretty, duration_style = format_duration(duration)643        items.append(fcol(fixlen_r(duration_pretty, 5), duration_style))644    else:645        items.append(("fixed", 5, urwid.Text("")))646 647    items.append(format_right_indicators(replay=False, marked=marked))648 649    return urwid.Pile([urwid.Columns(items, dividechars=1, min_width=15)])650 651 652@lru_cache(maxsize=800)653def format_dns_flow(654    *,655    render_mode: RenderMode,656    focused: bool,657    intercepted: bool,658    marked: str,659    is_replay: str | None,660    op_code: str,661    request_timestamp: float,662    domain: str,663    type: str,664    response_code: str | None,665    response_code_http_equiv: int,666    answer: str | None,667    error_message: str,668    duration: float | None,669):670    items = []671 672    if render_mode in (RenderMode.TABLE, RenderMode.DETAILVIEW):673        items.append(674            format_left_indicators(675                focused=focused, intercepted=intercepted, timestamp=request_timestamp676            )677        )678    else:679        items.append(fcol(">>" if focused else "  ", "focus"))680 681    scheme_style = "intercepted" if intercepted else SCHEME_STYLES["dns"]682    t = f"DNS {op_code}"683    if render_mode is RenderMode.TABLE:684        t = fixlen(t, 10)685    items.append(fcol(t, scheme_style))686    items.append(("weight", 0.5, TruncatedText(domain, colorize_host(domain), "right")))687    items.append(fcol("(" + fixlen(type, 5)[: len(type)] + ") =", "text"))688 689    items.append(690        (691            "weight",692            1,693            (694                truncated_plain(695                    "..." if answer is None else "?" if not answer else answer, "text"696                )697                if error_message is None698                else truncated_plain(error_message, "error")699            ),700        )701    )702    status_style = (703        "intercepted"704        if intercepted705        else HTTP_RESPONSE_CODE_STYLE.get(response_code_http_equiv // 100, "code_other")706    )707    items.append(708        fcol(fixlen("" if response_code is None else response_code, 9), status_style)709    )710 711    if duration:712        duration_pretty, duration_style = format_duration(duration)713        items.append(fcol(fixlen_r(duration_pretty, 5), duration_style))714    else:715        items.append(("fixed", 5, urwid.Text("")))716 717    items.append(718        format_right_indicators(719            replay=bool(is_replay),720            marked=marked,721        )722    )723    return urwid.Pile([urwid.Columns(items, dividechars=1, min_width=15)])724 725 726def format_flow(727    f: flow.Flow,728    *,729    render_mode: RenderMode,730    hostheader: bool = False,  # pass options directly if we need more stuff from them731    focused: bool = True,732) -> urwid.Widget:733    """734    This functions calls the proper renderer depending on the flow type.735    We also want to cache the renderer output, so we extract all attributes736    relevant for display and call the render with only that. This assures that rows737    are updated if the flow is changed.738    """739    duration: float | None740    error_message: str | None741    if f.error:742        error_message = f.error.msg743    else:744        error_message = None745 746    if isinstance(f, (TCPFlow, UDPFlow)):747        total_size = 0748        for message in f.messages:749            total_size += len(message.content)750        if f.messages:751            duration = f.messages[-1].timestamp - f.client_conn.timestamp_start752        else:753            duration = None754        if f.client_conn.tls_version == "QUICv1":755            protocol = "quic"756        else:757            protocol = f.type758        return format_message_flow(759            render_mode=render_mode,760            focused=focused,761            timestamp_start=f.client_conn.timestamp_start,762            marked=f.marked,763            protocol=protocol,764            client_address=f.client_conn.peername,765            server_address=f.server_conn.address,766            total_size=total_size,767            duration=duration,768            error_message=error_message,769        )770    elif isinstance(f, DNSFlow):771        if f.request.timestamp and f.response and f.response.timestamp:772            duration = f.response.timestamp - f.request.timestamp773        else:774            duration = None775        if f.response:776            response_code_str: str | None = dns.response_codes.to_str(777                f.response.response_code778            )779            response_code_http_equiv = dns.response_codes.http_equiv_status_code(780                f.response.response_code781            )782            answer = ", ".join(str(x) for x in f.response.answers)783        else:784            response_code_str = None785            response_code_http_equiv = 0786            answer = None787        return format_dns_flow(788            render_mode=render_mode,789            focused=focused,790            intercepted=f.intercepted,791            marked=f.marked,792            is_replay=f.is_replay,793            op_code=dns.op_codes.to_str(f.request.op_code),794            request_timestamp=f.request.timestamp,795            domain=f.request.questions[0].name if f.request.questions else "",796            type=dns.types.to_str(f.request.questions[0].type)797            if f.request.questions798            else "",799            response_code=response_code_str,800            response_code_http_equiv=response_code_http_equiv,801            answer=answer,802            error_message=error_message,803            duration=duration,804        )805    elif isinstance(f, HTTPFlow):806        intercepted = f.intercepted807        response_content_length: int | None808        if f.response:809            if f.response.raw_content is not None:810                response_content_length = len(f.response.raw_content)811            else:812                response_content_length = None813            response_code: int | None = f.response.status_code814            response_reason: str | None = f.response.reason815            response_content_type = f.response.headers.get("content-type")816            if f.response.timestamp_end:817                duration = max(818                    [f.response.timestamp_end - f.request.timestamp_start, 0]819                )820            else:821                duration = None822        else:823            response_content_length = None824            response_code = None825            response_reason = None826            response_content_type = None827            duration = None828 829        scheme = f.request.scheme830        if f.websocket is not None:831            if scheme == "https":832                scheme = "wss"833            elif scheme == "http":834                scheme = "ws"835 836        if render_mode in (RenderMode.LIST, RenderMode.DETAILVIEW):837            render_func = format_http_flow_list838        else:839            render_func = format_http_flow_table840        return render_func(841            render_mode=render_mode,842            focused=focused,843            marked=f.marked,844            is_replay=f.is_replay,845            request_method=f.request.method,846            request_scheme=scheme,847            request_host=f.request.pretty_host if hostheader else f.request.host,848            request_path=f.request.path,849            request_url=f.request.pretty_url if hostheader else f.request.url,850            request_http_version=f.request.http_version,851            request_timestamp=f.request.timestamp_start,852            request_is_push_promise="h2-pushed-stream" in f.metadata,853            intercepted=intercepted,854            response_code=response_code,855            response_reason=response_reason,856            response_content_length=response_content_length,857            response_content_type=response_content_type,858            duration=duration,859            error_message=error_message,860        )861 862    else:863        raise NotImplementedError()864