codekingpro/portable-devtools

1from collections import defaultdict2from dataclasses import dataclass3from dataclasses import field4 5import pylsqpack6from aioquic.buffer import Buffer7from aioquic.buffer import BufferReadError8from aioquic.h3.connection import parse_settings9from aioquic.h3.connection import Setting10 11from ..proxy.layers.http import is_h3_alpn12from mitmproxy import tcp13from mitmproxy.contentviews._api import Contentview14from mitmproxy.contentviews._api import Metadata15from mitmproxy_rs.contentviews import hex_dump16 17 18@dataclass(frozen=True)19class Frame:20    """Representation of an HTTP/3 frame."""21 22    type: int23    data: bytes24 25    def pretty(self) -> str:26        frame_name = f"0x{self.type:x} Frame"27        if self.type == 0:28            frame_name = "DATA Frame"29        elif self.type == 1:30            try:31                hdrs = pylsqpack.Decoder(4096, 16).feed_header(0, self.data)[1]32                return f"HEADERS Frame\n" + "\n".join(33                    f"{k.decode(errors='backslashreplace')}: {v.decode(errors='backslashreplace')}"34                    for k, v in hdrs35                )36            except Exception as e:37                frame_name = f"HEADERS Frame (error: {e})"38        elif self.type == 4:39            settings = []40            try:41                s = parse_settings(self.data)42            except Exception as e:43                frame_name = f"SETTINGS Frame (error: {e})"44            else:45                for k, v in s.items():46                    try:47                        key = Setting(k).name48                    except ValueError:49                        key = f"0x{k:x}"50                    settings.append(f"{key}: 0x{v:x}")51                return "SETTINGS Frame\n" + "\n".join(settings)52        return f"{frame_name}\n" + hex_dump.prettify(self.data, Metadata())53 54 55@dataclass(frozen=True)56class StreamType:57    """Representation of an HTTP/3 stream types."""58 59    type: int60 61    def pretty(self) -> str:62        stream_type = {63            0x00: "Control Stream",64            0x01: "Push Stream",65            0x02: "QPACK Encoder Stream",66            0x03: "QPACK Decoder Stream",67        }.get(self.type, f"0x{self.type:x} Stream")68        return stream_type69 70 71@dataclass72class ConnectionState:73    message_count: int = 074    frames: dict[int, list[Frame | StreamType]] = field(default_factory=dict)75    client_buf: bytearray = field(default_factory=bytearray)76    server_buf: bytearray = field(default_factory=bytearray)77 78 79class Http3Contentview(Contentview):80    def __init__(self) -> None:81        self.connections: defaultdict[tcp.TCPFlow, ConnectionState] = defaultdict(82            ConnectionState83        )84 85    @property86    def name(self) -> str:87        return "HTTP/3 Frames"88 89    def prettify(self, data: bytes, metadata: Metadata) -> str:90        flow = metadata.flow91        tcp_message = metadata.tcp_message92        assert isinstance(flow, tcp.TCPFlow)93        assert tcp_message94 95        state = self.connections[flow]96 97        for message in flow.messages[state.message_count :]:98            if message.from_client:99                buf = state.client_buf100            else:101                buf = state.server_buf102            buf += message.content103 104            if state.message_count == 0 and flow.metadata["quic_is_unidirectional"]:105                h3_buf = Buffer(data=bytes(buf[:8]))106                stream_type = h3_buf.pull_uint_var()107                consumed = h3_buf.tell()108                del buf[:consumed]109                state.frames[0] = [StreamType(stream_type)]110 111            while True:112                h3_buf = Buffer(data=bytes(buf[:16]))113                try:114                    frame_type = h3_buf.pull_uint_var()115                    frame_size = h3_buf.pull_uint_var()116                except BufferReadError:117                    break118 119                consumed = h3_buf.tell()120 121                if len(buf) < consumed + frame_size:122                    break123 124                frame_data = bytes(buf[consumed : consumed + frame_size])125 126                frame = Frame(frame_type, frame_data)127 128                state.frames.setdefault(state.message_count, []).append(frame)129 130                del buf[: consumed + frame_size]131 132            state.message_count += 1133 134        frames = state.frames.get(flow.messages.index(tcp_message), [])135        if not frames:136            return ""137        else:138            return "\n\n".join(frame.pretty() for frame in frames)139 140    def render_priority(141        self,142        data: bytes,143        metadata: Metadata,144    ) -> float:145        flow = metadata.flow146        return (147            2148            * float(bool(flow and is_h3_alpn(flow.client_conn.alpn)))149            * float(isinstance(flow, tcp.TCPFlow))150        )151 152 153http3 = Http3Contentview()154