mitmproxy.proxy.mode_specs

  1"""
  2This module is responsible for parsing proxy mode specifications such as
  3`"regular"`, `"reverse:https://example.com"`, or `"socks5@1234"`. The general syntax is
  4
  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 == 1234
 14    assert isinstance(mode, RegularMode)
 15
 16    ProxyMode.parse("reverse:example.com@invalid-port")  # ValueError
 17
 18    RegularMode.parse("regular")  # ok
 19    RegularMode.parse("socks5")  # ValueError
 20
 21"""
 22
 23from __future__ import annotations
 24
 25import dataclasses
 26import sys
 27from abc import ABCMeta
 28from abc import abstractmethod
 29from dataclasses import dataclass
 30from functools import cache
 31from typing import ClassVar
 32from typing import Literal
 33
 34import mitmproxy_rs
 35
 36from mitmproxy.coretypes.serializable import Serializable
 37from mitmproxy.net import server_spec
 38
 39if sys.version_info < (3, 11):
 40    from typing_extensions import Self  # pragma: no cover
 41else:
 42    from typing import Self
 43
 44
 45@dataclass(frozen=True)  # type: ignore
 46class ProxyMode(Serializable, metaclass=ABCMeta):
 47    """
 48    Parsed representation of a proxy mode spec. Subclassed for each specific mode,
 49    which then does its own data validation.
 50    """
 51
 52    full_spec: str
 53    """The full proxy mode spec as entered by the user."""
 54    data: str
 55    """The (raw) mode data, i.e. the part after the mode name."""
 56    custom_listen_host: str | None
 57    """A custom listen host, if specified in the spec."""
 58    custom_listen_port: int | None
 59    """A custom listen port, if specified in the spec."""
 60
 61    type_name: ClassVar[
 62        str
 63    ]  # automatically derived from the class name in __init_subclass__
 64    """The unique name for this proxy mode, e.g. "regular" or "reverse"."""
 65    __types: ClassVar[dict[str, type[ProxyMode]]] = {}
 66
 67    def __init_subclass__(cls, **kwargs):
 68        cls.type_name = cls.__name__.removesuffix("Mode").lower()
 69        assert cls.type_name not in ProxyMode.__types
 70        ProxyMode.__types[cls.type_name] = cls
 71
 72    def __repr__(self):
 73        return f"ProxyMode.parse({self.full_spec!r})"
 74
 75    @abstractmethod
 76    def __post_init__(self) -> None:
 77        """Validation of data happens here."""
 78
 79    @property
 80    @abstractmethod
 81    def description(self) -> str:
 82        """The mode description that will be used in server logs and UI."""
 83
 84    @property
 85    def default_port(self) -> int:
 86        """
 87        Default listen port of servers for this mode, see `ProxyMode.listen_port()`.
 88        """
 89        return 8080
 90
 91    @property
 92    @abstractmethod
 93    def transport_protocol(self) -> Literal["tcp", "udp"] | None:
 94        """The transport protocol used by this mode's server."""
 95
 96    @classmethod
 97    @cache
 98    def parse(cls, spec: str) -> Self:
 99        """
100        Parse a proxy mode specification and return the corresponding `ProxyMode` instance.
101        """
102        head, _, listen_at = spec.rpartition("@")
103        if not head:
104            head = listen_at
105            listen_at = ""
106
107        mode, _, data = head.partition(":")
108
109        if listen_at:
110            if ":" in listen_at:
111                host, _, port_str = listen_at.rpartition(":")
112            else:
113                host = None
114                port_str = listen_at
115            try:
116                port = int(port_str)
117                if port < 0 or 65535 < port:
118                    raise ValueError
119            except ValueError:
120                raise ValueError(f"invalid port: {port_str}")
121        else:
122            host = None
123            port = None
124
125        try:
126            mode_cls = ProxyMode.__types[mode.lower()]
127        except KeyError:
128            raise ValueError(f"unknown mode")
129
130        if not issubclass(mode_cls, cls):
131            raise ValueError(f"{mode!r} is not a spec for a {cls.type_name} mode")
132
133        return mode_cls(
134            full_spec=spec, data=data, custom_listen_host=host, custom_listen_port=port
135        )
136
137    def listen_host(self, default: str | None = None) -> str:
138        """
139        Return the address a server for this mode should listen on. This can be either directly
140        specified in the spec or taken from a user-configured global default (`options.listen_host`).
141        By default, return an empty string to listen on all hosts.
142        """
143        if self.custom_listen_host is not None:
144            return self.custom_listen_host
145        elif default is not None:
146            return default
147        else:
148            return ""
149
150    def listen_port(self, default: int | None = None) -> int:
151        """
152        Return the port a server for this mode should listen on. This can be either directly
153        specified in the spec, taken from a user-configured global default (`options.listen_port`),
154        or from `ProxyMode.default_port`.
155        """
156        if self.custom_listen_port is not None:
157            return self.custom_listen_port
158        elif default is not None:
159            return default
160        else:
161            return self.default_port
162
163    @classmethod
164    def from_state(cls, state):
165        return ProxyMode.parse(state)
166
167    def get_state(self):
168        return self.full_spec
169
170    def set_state(self, state):
171        if state != self.full_spec:
172            raise dataclasses.FrozenInstanceError("Proxy modes are immutable.")
173
174
175TCP: Literal["tcp", "udp"] = "tcp"
176UDP: Literal["tcp", "udp"] = "udp"
177
178
179def _check_empty(data):
180    if data:
181        raise ValueError("mode takes no arguments")
182
183
184class RegularMode(ProxyMode):
185    """A regular HTTP(S) proxy that is interfaced with `HTTP CONNECT` calls (or absolute-form HTTP requests)."""
186
187    description = "HTTP(S) proxy"
188    transport_protocol = TCP
189
190    def __post_init__(self) -> None:
191        _check_empty(self.data)
192
193
194class TransparentMode(ProxyMode):
195    """A transparent proxy, see https://docs.mitmproxy.org/dev/howto-transparent/"""
196
197    description = "Transparent Proxy"
198    transport_protocol = TCP
199
200    def __post_init__(self) -> None:
201        _check_empty(self.data)
202
203
204class UpstreamMode(ProxyMode):
205    """A regular HTTP(S) proxy, but all connections are forwarded to a second upstream HTTP(S) proxy."""
206
207    description = "HTTP(S) proxy (upstream mode)"
208    transport_protocol = TCP
209    scheme: Literal["http", "https"]
210    address: tuple[str, int]
211
212    # noinspection PyDataclass
213    def __post_init__(self) -> None:
214        scheme, self.address = server_spec.parse(self.data, default_scheme="http")
215        if scheme != "http" and scheme != "https":
216            raise ValueError("invalid upstream proxy scheme")
217        self.scheme = scheme
218
219
220class ReverseMode(ProxyMode):
221    """A reverse proxy. This acts like a normal server, but redirects all requests to a fixed target."""
222
223    description = "reverse proxy"
224    transport_protocol = TCP
225    scheme: Literal[
226        "http", "https", "http3", "tls", "dtls", "tcp", "udp", "dns", "quic"
227    ]
228    address: tuple[str, int]
229
230    # noinspection PyDataclass
231    def __post_init__(self) -> None:
232        self.scheme, self.address = server_spec.parse(self.data, default_scheme="https")
233        if self.scheme in ("http3", "dtls", "udp", "dns", "quic"):
234            self.transport_protocol = UDP
235        self.description = f"{self.description} to {self.data}"
236
237    @property
238    def default_port(self) -> int:
239        if self.scheme == "dns":
240            return 53
241        return super().default_port
242
243
244class Socks5Mode(ProxyMode):
245    """A SOCKSv5 proxy."""
246
247    description = "SOCKS v5 proxy"
248    default_port = 1080
249    transport_protocol = TCP
250
251    def __post_init__(self) -> None:
252        _check_empty(self.data)
253
254
255class DnsMode(ProxyMode):
256    """A DNS server."""
257
258    description = "DNS server"
259    default_port = 53
260    transport_protocol = UDP
261
262    def __post_init__(self) -> None:
263        _check_empty(self.data)
264
265
266# class Http3Mode(ProxyMode):
267#     """
268#     A regular HTTP3 proxy that is interfaced with absolute-form HTTP requests.
269#     (This class will be merged into `RegularMode` once the UDP implementation is deemed stable enough.)
270#     """
271#
272#     description = "HTTP3 proxy"
273#     transport_protocol = UDP
274#
275#     def __post_init__(self) -> None:
276#         _check_empty(self.data)
277
278
279class WireGuardMode(ProxyMode):
280    """Proxy Server based on WireGuard"""
281
282    description = "WireGuard server"
283    default_port = 51820
284    transport_protocol = UDP
285
286    def __post_init__(self) -> None:
287        pass
288
289
290class LocalMode(ProxyMode):
291    """OS-level transparent proxy."""
292
293    description = "Local redirector"
294    transport_protocol = None
295
296    def __post_init__(self) -> None:
297        # should not raise
298        mitmproxy_rs.LocalRedirector.describe_spec(self.data)
299
300
301class OsProxyMode(ProxyMode):  # pragma: no cover
302    """Deprecated alias for LocalMode"""
303
304    description = "Deprecated alias for LocalMode"
305    transport_protocol = None
306
307    def __post_init__(self) -> None:
308        raise ValueError(
309            "osproxy mode has been renamed to local mode. Thanks for trying our experimental features!"
310        )