codekingpro/portable-devtools

1from abc import abstractmethod2from enum import Enum3 4from mitmproxy.contentviews._api import Contentview5from mitmproxy.contentviews._api import Metadata6from mitmproxy.http import HTTPFlow7from mitmproxy.utils import strutils8 9 10class PacketType(Enum):11    @property12    @abstractmethod13    def visible(self) -> bool:14        raise RuntimeError  # pragma: no cover15 16    def __str__(self):17        return f"{type(self).__name__}.{self.name}"18 19 20class EngineIO(PacketType):21    # https://github.com/socketio/engine.io-protocol?tab=readme-ov-file#protocol22    OPEN = ord("0")23    CLOSE = ord("1")24    PING = ord("2")25    PONG = ord("3")26    MESSAGE = ord("4")27    UPGRADE = ord("5")28    NOOP = ord("6")29 30    @property31    def visible(self):32        return self not in (33            self.PING,34            self.PONG,35        )36 37 38class SocketIO(PacketType):39    # https://github.com/socketio/socket.io-protocol?tab=readme-ov-file#exchange-protocol40    CONNECT = ord("0")41    DISCONNECT = ord("1")42    EVENT = ord("2")43    ACK = ord("3")44    CONNECT_ERROR = ord("4")45    BINARY_EVENT = ord("5")46    BINARY_ACK = ord("6")47 48    @property49    def visible(self):50        return self not in (51            self.ACK,52            self.BINARY_ACK,53        )54 55 56def parse_packet(data: bytes) -> tuple[PacketType, bytes]:57    # throws IndexError/ValueError if invalid packet58    engineio_type = EngineIO(data[0])59    data = data[1:]60 61    if engineio_type is not EngineIO.MESSAGE:62        return engineio_type, data63 64    socketio_type = SocketIO(data[0])65    data = data[1:]66 67    return socketio_type, data68 69 70class SocketIOContentview(Contentview):71    name = "Socket.IO"72 73    def prettify(74        self,75        data: bytes,76        metadata: Metadata,77    ) -> str:78        packet_type, msg = parse_packet(data)79        if not packet_type.visible:80            return ""81        return f"{packet_type} {strutils.bytes_to_escaped_str(msg)}"82 83    def render_priority(84        self,85        data: bytes,86        metadata: Metadata,87    ) -> float:88        return float(89            bool(90                data91                and isinstance(metadata.flow, HTTPFlow)92                and metadata.flow.websocket is not None93                and "/socket.io/?" in metadata.flow.request.path94            )95        )96 97 98socket_io = SocketIOContentview()99