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