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 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 default is not None:
161            return default
162        else:
163            return self.default_port
164
165    @classmethod
166    def from_state(cls, state):
167        return ProxyMode.parse(state)
168
169    def get_state(self):
170        return self.full_spec
171
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 = TCP
192
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 = TCP
202
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 = TCP
212    scheme: Literal["http", "https"]
213    address: tuple[str, int]
214
215    # noinspection PyDataclass
216    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 = scheme
221
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 = TCP
228    scheme: Literal[
229        "http", "https", "http3", "tls", "dtls", "tcp", "udp", "dns", "quic"
230    ]
231    address: tuple[str, int]
232
233    # noinspection PyDataclass
234    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 = UDP
238        elif self.scheme in ("dns", "https"):
239            self.transport_protocol = BOTH
240        self.description = f"{self.description} to {self.data}"
241
242    @property
243    def default_port(self) -> int | None:
244        if self.scheme == "dns":
245            return 53
246        return super().default_port
247
248
249class Socks5Mode(ProxyMode):
250    """A SOCKSv5 proxy."""
251
252    description = "SOCKS v5 proxy"
253    default_port = 1080
254    transport_protocol = TCP
255
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 = 53
265    transport_protocol = BOTH
266
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 = UDP
279#
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 = 51820
289    transport_protocol = UDP
290
291    def __post_init__(self) -> None:
292        pass
293
294
295class LocalMode(ProxyMode):
296    """OS-level transparent proxy."""
297
298    description = "Local redirector"
299    transport_protocol = BOTH
300    default_port = None
301
302    def __post_init__(self) -> None:
303        # should not raise
304        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 = None
312    transport_protocol = BOTH
313
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 cover
320            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 cover
327    """Deprecated alias for LocalMode"""
328
329    description = "Deprecated alias for LocalMode"
330    transport_protocol = BOTH
331    default_port = None
332
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        )