codekingpro/portable-devtools

1"""2This module is responsible for parsing proxy mode specifications such as3`"regular"`, `"reverse:https://example.com"`, or `"socks5@1234"`. The general syntax is4 5    mode [: mode_configuration] [@ [listen_addr:]listen_port]6 7For a full example, consider `reverse:https://example.com@127.0.0.1:443`.8This would spawn a reverse proxy on port 443 bound to localhost.9The mode is `reverse`, and the mode data is `https://example.com`.10Examples:11 12    mode = ProxyMode.parse("regular@1234")13    assert mode.listen_port == 123414    assert isinstance(mode, RegularMode)15 16    ProxyMode.parse("reverse:example.com@invalid-port")  # ValueError17 18    RegularMode.parse("regular")  # ok19    RegularMode.parse("socks5")  # ValueError20 21"""22 23from __future__ import annotations24 25import dataclasses26import platform27import re28import sys29from abc import ABCMeta30from abc import abstractmethod31from dataclasses import dataclass32from functools import cache33from typing import ClassVar34from typing import Literal35 36import mitmproxy_rs37from mitmproxy.coretypes.serializable import Serializable38from mitmproxy.net import server_spec39 40if sys.version_info < (3, 11):41    from typing_extensions import Self  # pragma: no cover42else:43    from typing import Self44 45 46@dataclass(frozen=True)  # type: ignore47class ProxyMode(Serializable, metaclass=ABCMeta):48    """49    Parsed representation of a proxy mode spec. Subclassed for each specific mode,50    which then does its own data validation.51    """52 53    full_spec: str54    """The full proxy mode spec as entered by the user."""55    data: str56    """The (raw) mode data, i.e. the part after the mode name."""57    custom_listen_host: str | None58    """A custom listen host, if specified in the spec."""59    custom_listen_port: int | None60    """A custom listen port, if specified in the spec."""61 62    type_name: ClassVar[63        str64    ]  # automatically derived from the class name in __init_subclass__65    """The unique name for this proxy mode, e.g. "regular" or "reverse"."""66    __types: ClassVar[dict[str, type[ProxyMode]]] = {}67 68    def __init_subclass__(cls, **kwargs):69        cls.type_name = cls.__name__.removesuffix("Mode").lower()70        assert cls.type_name not in ProxyMode.__types71        ProxyMode.__types[cls.type_name] = cls72 73    def __repr__(self):74        return f"ProxyMode.parse({self.full_spec!r})"75 76    @abstractmethod77    def __post_init__(self) -> None:78        """Validation of data happens here."""79 80    @property81    @abstractmethod82    def description(self) -> str:83        """The mode description that will be used in server logs and UI."""84 85    @property86    def default_port(self) -> int | None:87        """88        Default listen port of servers for this mode, see `ProxyMode.listen_port()`.89        """90        return 808091 92    @property93    @abstractmethod94    def transport_protocol(self) -> Literal["tcp", "udp", "both"]:95        """The transport protocol used by this mode's server."""96 97    @classmethod98    @cache99    def parse(cls, spec: str) -> Self:100        """101        Parse a proxy mode specification and return the corresponding `ProxyMode` instance.102        """103        head, _, listen_at = spec.rpartition("@")104        if not head:105            head = listen_at106            listen_at = ""107 108        mode, _, data = head.partition(":")109 110        if listen_at:111            if ":" in listen_at:112                host, _, port_str = listen_at.rpartition(":")113            else:114                host = None115                port_str = listen_at116            try:117                port = int(port_str)118                if port < 0 or 65535 < port:119                    raise ValueError120            except ValueError:121                raise ValueError(f"invalid port: {port_str}")122        else:123            host = None124            port = None125 126        try:127            mode_cls = ProxyMode.__types[mode.lower()]128        except KeyError:129            raise ValueError(f"unknown mode")130 131        if not issubclass(mode_cls, cls):132            raise ValueError(f"{mode!r} is not a spec for a {cls.type_name} mode")133 134        return mode_cls(135            full_spec=spec, data=data, custom_listen_host=host, custom_listen_port=port136        )137 138    def listen_host(self, default: str | None = None) -> str:139        """140        Return the address a server for this mode should listen on. This can be either directly141        specified in the spec or taken from a user-configured global default (`options.listen_host`).142        By default, return an empty string to listen on all hosts.143        """144        if self.custom_listen_host is not None:145            return self.custom_listen_host146        elif default is not None:147            return default148        else:149            return ""150 151    def listen_port(self, default: int | None = None) -> int | None:152        """153        Return the port a server for this mode should listen on. This can be either directly154        specified in the spec, taken from a user-configured global default (`options.listen_port`),155        or from `ProxyMode.default_port`.156        May be `None` for modes that don't bind to a specific address, e.g. local redirect mode.157        """158        if self.custom_listen_port is not None:159            return self.custom_listen_port160        elif default is not None:161            return default162        else:163            return self.default_port164 165    @classmethod166    def from_state(cls, state):167        return ProxyMode.parse(state)168 169    def get_state(self):170        return self.full_spec171 172    def set_state(self, state):173        if state != self.full_spec:174            raise dataclasses.FrozenInstanceError("Proxy modes are immutable.")175 176 177TCP: Literal["tcp", "udp", "both"] = "tcp"178UDP: Literal["tcp", "udp", "both"] = "udp"179BOTH: Literal["tcp", "udp", "both"] = "both"180 181 182def _check_empty(data):183    if data:184        raise ValueError("mode takes no arguments")185 186 187class RegularMode(ProxyMode):188    """A regular HTTP(S) proxy that is interfaced with `HTTP CONNECT` calls (or absolute-form HTTP requests)."""189 190    description = "HTTP(S) proxy"191    transport_protocol = TCP192 193    def __post_init__(self) -> None:194        _check_empty(self.data)195 196 197class TransparentMode(ProxyMode):198    """A transparent proxy, see https://docs.mitmproxy.org/dev/howto-transparent/"""199 200    description = "Transparent Proxy"201    transport_protocol = TCP202 203    def __post_init__(self) -> None:204        _check_empty(self.data)205 206 207class UpstreamMode(ProxyMode):208    """A regular HTTP(S) proxy, but all connections are forwarded to a second upstream HTTP(S) proxy."""209 210    description = "HTTP(S) proxy (upstream mode)"211    transport_protocol = TCP212    scheme: Literal["http", "https"]213    address: tuple[str, int]214 215    # noinspection PyDataclass216    def __post_init__(self) -> None:217        scheme, self.address = server_spec.parse(self.data, default_scheme="http")218        if scheme != "http" and scheme != "https":219            raise ValueError("invalid upstream proxy scheme")220        self.scheme = scheme221 222 223class ReverseMode(ProxyMode):224    """A reverse proxy. This acts like a normal server, but redirects all requests to a fixed target."""225 226    description = "reverse proxy"227    transport_protocol = TCP228    scheme: Literal[229        "http", "https", "http3", "tls", "dtls", "tcp", "udp", "dns", "quic"230    ]231    address: tuple[str, int]232 233    # noinspection PyDataclass234    def __post_init__(self) -> None:235        self.scheme, self.address = server_spec.parse(self.data, default_scheme="https")236        if self.scheme in ("http3", "dtls", "udp", "quic"):237            self.transport_protocol = UDP238        elif self.scheme in ("dns", "https"):239            self.transport_protocol = BOTH240        self.description = f"{self.description} to {self.data}"241 242    @property243    def default_port(self) -> int | None:244        if self.scheme == "dns":245            return 53246        return super().default_port247 248 249class Socks5Mode(ProxyMode):250    """A SOCKSv5 proxy."""251 252    description = "SOCKS v5 proxy"253    default_port = 1080254    transport_protocol = TCP255 256    def __post_init__(self) -> None:257        _check_empty(self.data)258 259 260class DnsMode(ProxyMode):261    """A DNS server."""262 263    description = "DNS server"264    default_port = 53265    transport_protocol = BOTH266 267    def __post_init__(self) -> None:268        _check_empty(self.data)269 270 271# class Http3Mode(ProxyMode):272#     """273#     A regular HTTP3 proxy that is interfaced with absolute-form HTTP requests.274#     (This class will be merged into `RegularMode` once the UDP implementation is deemed stable enough.)275#     """276#277#     description = "HTTP3 proxy"278#     transport_protocol = UDP279#280#     def __post_init__(self) -> None:281#         _check_empty(self.data)282 283 284class WireGuardMode(ProxyMode):285    """Proxy Server based on WireGuard"""286 287    description = "WireGuard server"288    default_port = 51820289    transport_protocol = UDP290 291    def __post_init__(self) -> None:292        pass293 294 295class LocalMode(ProxyMode):296    """OS-level transparent proxy."""297 298    description = "Local redirector"299    transport_protocol = BOTH300    default_port = None301 302    def __post_init__(self) -> None:303        # should not raise304        mitmproxy_rs.local.LocalRedirector.describe_spec(self.data)305 306 307class TunMode(ProxyMode):308    """A Tun interface."""309 310    description = "TUN interface"311    default_port = None312    transport_protocol = BOTH313 314    def __post_init__(self) -> None:315        invalid_tun_name = self.data and (316            # The Rust side is Linux only for the moment, but eventually we may need this.317            platform.system() == "Darwin" and not re.match(r"^utun\d+$", self.data)318        )319        if invalid_tun_name:  # pragma: no cover320            raise ValueError(321                f"Invalid tun name: {self.data}. "322                f"On macOS, the tun name must be the form utunx where x is a number, such as utun3."323            )324 325 326class OsProxyMode(ProxyMode):  # pragma: no cover327    """Deprecated alias for LocalMode"""328 329    description = "Deprecated alias for LocalMode"330    transport_protocol = BOTH331    default_port = None332 333    def __post_init__(self) -> None:334        raise ValueError(335            "osproxy mode has been renamed to local mode. Thanks for trying our experimental features!"336        )337