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://[email protected]: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 platform
 27import re
 28import sys
 29from abc import ABCMeta
 30from abc import abstractmethod
 31from dataclasses import dataclass
 32from functools import cache
 33from typing import ClassVar
 34from typing import Literal
 35
 36import mitmproxy_rs
 37from mitmproxy.coretypes.serializable import Serializable
 38from mitmproxy.net import server_spec
 39
 40if sys.version_info < (3, 11):
 41    from typing_extensions import Self  # pragma: no cover
 42else:
 43    from typing import Self
 44
 45
 46@dataclass(frozen=True)  # type: ignore
 47class 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: str
 54    """The full proxy mode spec as entered by the user."""
 55    data: str
 56    """The (raw) mode data, i.e. the part after the mode name."""
 57    custom_listen_host: str | None
 58    """A custom listen host, if specified in the spec."""
 59    custom_listen_port: int | None
 60    """A custom listen port, if specified in the spec."""
 61
 62    type_name: ClassVar[
 63        str
 64    ]  # 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.__types
 71        ProxyMode.__types[cls.type_name] = cls
 72
 73    def __repr__(self):
 74        return f"ProxyMode.parse({self.full_spec!r})"
 75
 76    @abstractmethod
 77    def __post_init__(self) -> None:
 78        """Validation of data happens here."""
 79
 80    @property
 81    @abstractmethod
 82    def description(self) -> str:
 83        """The mode description that will be used in server logs and UI."""
 84
 85    @property
 86    def default_port(self) -> int | None:
 87        """
 88        Default listen port of servers for this mode, see `ProxyMode.listen_port()`.
 89        """
 90        return 8080
 91
 92    @property
 93    @abstractmethod
 94    def transport_protocol(self) -> Literal["tcp", "udp", "both"]:
 95        """The transport protocol used by this mode's server."""
 96
 97    @classmethod
 98    @cache
 99    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_at
106            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 = None
115                port_str = listen_at
116            try:
117                port = int(port_str)
118                if port < 0 or 65535 < port:
119                    raise ValueError
120            except ValueError:
121                raise ValueError(f"invalid port: {port_str}")
122        else:
123            host = None
124            port = None
125
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=port
136        )
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 directly
141        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_host
146        elif default is not None:
147            return default
148        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 directly
154        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_port
160        elif self.default_port is None:
161            return None
162        elif default is not None:
163            return default
164        else:
165            return self.default_port
166
167    @classmethod
168    def from_state(cls, state):
169        return ProxyMode.parse(state)
170
171    def get_state(self):
172        return self.full_spec
173
174    def set_state(self, state):
175        if state != self.full_spec:
176            raise dataclasses.FrozenInstanceError("Proxy modes are immutable.")
177
178
179TCP: Literal["tcp", "udp", "both"] = "tcp"
180UDP: Literal["tcp", "udp", "both"] = "udp"
181BOTH: Literal["tcp", "udp", "both"] = "both"
182
183
184def _check_empty(data):
185    if data:
186        raise ValueError("mode takes no arguments")
187
188
189class RegularMode(ProxyMode):
190    """A regular HTTP(S) proxy that is interfaced with `HTTP CONNECT` calls (or absolute-form HTTP requests)."""
191
192    description = "HTTP(S) proxy"
193    transport_protocol = TCP
194
195    def __post_init__(self) -> None:
196        _check_empty(self.data)
197
198
199class TransparentMode(ProxyMode):
200    """A transparent proxy, see https://docs.mitmproxy.org/dev/howto-transparent/"""
201
202    description = "Transparent Proxy"
203    transport_protocol = TCP
204
205    def __post_init__(self) -> None:
206        _check_empty(self.data)
207
208
209class UpstreamMode(ProxyMode):
210    """A regular HTTP(S) proxy, but all connections are forwarded to a second upstream HTTP(S) proxy."""
211
212    description = "HTTP(S) proxy (upstream mode)"
213    transport_protocol = TCP
214    scheme: Literal["http", "https"]
215    address: tuple[str, int]
216
217    # noinspection PyDataclass
218    def __post_init__(self) -> None:
219        scheme, self.address = server_spec.parse(self.data, default_scheme="http")
220        if scheme != "http" and scheme != "https":
221            raise ValueError("invalid upstream proxy scheme")
222        self.scheme = scheme
223
224
225class ReverseMode(ProxyMode):
226    """A reverse proxy. This acts like a normal server, but redirects all requests to a fixed target."""
227
228    description = "reverse proxy"
229    transport_protocol = TCP
230    scheme: Literal[
231        "http", "https", "http3", "tls", "dtls", "tcp", "udp", "dns", "quic"
232    ]
233    address: tuple[str, int]
234
235    # noinspection PyDataclass
236    def __post_init__(self) -> None:
237        self.scheme, self.address = server_spec.parse(self.data, default_scheme="https")
238        if self.scheme in ("http3", "dtls", "udp", "quic"):
239            self.transport_protocol = UDP
240        elif self.scheme in ("dns", "https"):
241            self.transport_protocol = BOTH
242        self.description = f"{self.description} to {self.data}"
243
244    @property
245    def default_port(self) -> int | None:
246        if self.scheme == "dns":
247            return 53
248        return super().default_port
249
250
251class Socks5Mode(ProxyMode):
252    """A SOCKSv5 proxy."""
253
254    description = "SOCKS v5 proxy"
255    default_port = 1080
256    transport_protocol = TCP
257
258    def __post_init__(self) -> None:
259        _check_empty(self.data)
260
261
262class DnsMode(ProxyMode):
263    """A DNS server."""
264
265    description = "DNS server"
266    default_port = 53
267    transport_protocol = BOTH
268
269    def __post_init__(self) -> None:
270        _check_empty(self.data)
271
272
273# class Http3Mode(ProxyMode):
274#     """
275#     A regular HTTP3 proxy that is interfaced with absolute-form HTTP requests.
276#     (This class will be merged into `RegularMode` once the UDP implementation is deemed stable enough.)
277#     """
278#
279#     description = "HTTP3 proxy"
280#     transport_protocol = UDP
281#
282#     def __post_init__(self) -> None:
283#         _check_empty(self.data)
284
285
286class WireGuardMode(ProxyMode):
287    """Proxy Server based on WireGuard"""
288
289    description = "WireGuard server"
290    default_port = 51820
291    transport_protocol = UDP
292
293    def __post_init__(self) -> None:
294        pass
295
296
297class LocalMode(ProxyMode):
298    """OS-level transparent proxy."""
299
300    description = "Local redirector"
301    transport_protocol = BOTH
302    default_port = None
303
304    def __post_init__(self) -> None:
305        # should not raise
306        mitmproxy_rs.local.LocalRedirector.describe_spec(self.data)
307
308
309class TunMode(ProxyMode):
310    """A Tun interface."""
311
312    description = "TUN interface"
313    default_port = None
314    transport_protocol = BOTH
315
316    def __post_init__(self) -> None:
317        invalid_tun_name = self.data and (
318            # The Rust side is Linux only for the moment, but eventually we may need this.
319            platform.system() == "Darwin" and not re.match(r"^utun\d+$", self.data)
320        )
321        if invalid_tun_name:  # pragma: no cover
322            raise ValueError(
323                f"Invalid tun name: {self.data}. "
324                f"On macOS, the tun name must be the form utunx where x is a number, such as utun3."
325            )
326
327
328class OsProxyMode(ProxyMode):  # pragma: no cover
329    """Deprecated alias for LocalMode"""
330
331    description = "Deprecated alias for LocalMode"
332    transport_protocol = BOTH
333    default_port = None
334
335    def __post_init__(self) -> None:
336        raise ValueError(
337            "osproxy mode has been renamed to local mode. Thanks for trying our experimental features!"
338        )