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
 25from abc import ABCMeta, abstractmethod
 26from dataclasses import dataclass
 27from functools import cache
 28from typing import ClassVar, Literal, Type, TypeVar
 29
 30from mitmproxy.coretypes.serializable import Serializable
 31from mitmproxy.net import server_spec
 32
 33# Python 3.11: Use typing.Self
 34Self = TypeVar("Self", bound="ProxyMode")
 35
 36
 37@dataclass(frozen=True)  # type: ignore
 38class ProxyMode(Serializable, metaclass=ABCMeta):
 39    """
 40    Parsed representation of a proxy mode spec. Subclassed for each specific mode,
 41    which then does its own data validation.
 42    """
 43    full_spec: str
 44    """The full proxy mode spec as entered by the user."""
 45    data: str
 46    """The (raw) mode data, i.e. the part after the mode name."""
 47    custom_listen_host: str | None
 48    """A custom listen host, if specified in the spec."""
 49    custom_listen_port: int | None
 50    """A custom listen port, if specified in the spec."""
 51
 52    type_name: ClassVar[str]  # automatically derived from the class name in __init_subclass__
 53    """The unique name for this proxy mode, e.g. "regular" or "reverse"."""
 54    __types: ClassVar[dict[str, Type[ProxyMode]]] = {}
 55
 56    def __init_subclass__(cls, **kwargs):
 57        cls.type_name = cls.__name__.removesuffix("Mode").lower()
 58        assert cls.type_name not in ProxyMode.__types
 59        ProxyMode.__types[cls.type_name] = cls
 60
 61    def __repr__(self):
 62        return f"ProxyMode.parse({self.full_spec!r})"
 63
 64    @abstractmethod
 65    def __post_init__(self) -> None:
 66        """Validation of data happens here."""
 67
 68    @property
 69    @abstractmethod
 70    def description(self) -> str:
 71        """The mode description that will be used in server logs and UI."""
 72
 73    @property
 74    def default_port(self) -> int:
 75        """
 76        Default listen port of servers for this mode, see `ProxyMode.listen_port()`.
 77        """
 78        return 8080
 79
 80    @property
 81    @abstractmethod
 82    def transport_protocol(self) -> Literal["tcp", "udp"]:
 83        """The transport protocol used by this mode's server."""
 84
 85    @classmethod
 86    @cache
 87    def parse(cls: Type[Self], spec: str) -> Self:
 88        """
 89        Parse a proxy mode specification and return the corresponding `ProxyMode` instance.
 90        """
 91        head, _, listen_at = spec.rpartition("@")
 92        if not head:
 93            head = listen_at
 94            listen_at = ""
 95
 96        mode, _, data = head.partition(":")
 97
 98        if listen_at:
 99            if ":" in listen_at:
100                host, _, port_str = listen_at.rpartition(":")
101            else:
102                host = None
103                port_str = listen_at
104            try:
105                port = int(port_str)
106                if port < 0 or 65535 < port:
107                    raise ValueError
108            except ValueError:
109                raise ValueError(f"invalid port: {port_str}")
110        else:
111            host = None
112            port = None
113
114        try:
115            mode_cls = ProxyMode.__types[mode.lower()]
116        except KeyError:
117            raise ValueError(f"unknown mode")
118
119        if not issubclass(mode_cls, cls):
120            raise ValueError(f"{mode!r} is not a spec for a {cls.type_name} mode")
121
122        return mode_cls(
123            full_spec=spec,
124            data=data,
125            custom_listen_host=host,
126            custom_listen_port=port
127        )
128
129    def listen_host(self, default: str | None = None) -> str:
130        """
131        Return the address a server for this mode should listen on. This can be either directly
132        specified in the spec or taken from a user-configured global default (`options.listen_host`).
133        By default, return an empty string to listen on all hosts.
134        """
135        if self.custom_listen_host is not None:
136            return self.custom_listen_host
137        elif default is not None:
138            return default
139        else:
140            return ""
141
142    def listen_port(self, default: int | None = None) -> int:
143        """
144        Return the port a server for this mode should listen on. This can be either directly
145        specified in the spec, taken from a user-configured global default (`options.listen_port`),
146        or from `ProxyMode.default_port`.
147        """
148        if self.custom_listen_port is not None:
149            return self.custom_listen_port
150        elif default is not None:
151            return default
152        else:
153            return self.default_port
154
155    @classmethod
156    def from_state(cls, state):
157        return ProxyMode.parse(state)
158
159    def get_state(self):
160        return self.full_spec
161
162    def set_state(self, state):
163        if state != self.full_spec:
164            raise RuntimeError("Proxy modes are frozen.")
165
166
167TCP: Literal['tcp', 'udp'] = "tcp"
168UDP: Literal['tcp', 'udp'] = "udp"
169
170
171def _check_empty(data):
172    if data:
173        raise ValueError("mode takes no arguments")
174
175
176class RegularMode(ProxyMode):
177    """A regular HTTP(S) proxy that is interfaced with `HTTP CONNECT` calls (or absolute-form HTTP requests)."""
178    description = "HTTP(S) proxy"
179    transport_protocol = TCP
180
181    def __post_init__(self) -> None:
182        _check_empty(self.data)
183
184
185class TransparentMode(ProxyMode):
186    """A transparent proxy, see https://docs.mitmproxy.org/dev/howto-transparent/"""
187    description = "transparent proxy"
188    transport_protocol = TCP
189
190    def __post_init__(self) -> None:
191        _check_empty(self.data)
192
193
194class UpstreamMode(ProxyMode):
195    """A regular HTTP(S) proxy, but all connections are forwarded to a second upstream HTTP(S) proxy."""
196    description = "HTTP(S) proxy (upstream mode)"
197    transport_protocol = TCP
198    scheme: Literal["http", "https"]
199    address: tuple[str, int]
200
201    # noinspection PyDataclass
202    def __post_init__(self) -> None:
203        scheme, self.address = server_spec.parse(self.data, default_scheme="http")
204        if scheme != "http" and scheme != "https":
205            raise ValueError("invalid upstream proxy scheme")
206        self.scheme = scheme
207
208
209class ReverseMode(ProxyMode):
210    """A reverse proxy. This acts like a normal server, but redirects all requests to a fixed target."""
211    description = "reverse proxy"
212    transport_protocol = TCP
213    scheme: Literal["http", "https", "tls", "dtls", "tcp", "udp", "dns"]
214    address: tuple[str, int]
215
216    # noinspection PyDataclass
217    def __post_init__(self) -> None:
218        self.scheme, self.address = server_spec.parse(self.data, default_scheme="https")
219        if self.scheme in ("dns", "dtls", "udp"):
220            self.transport_protocol = UDP
221        self.description = f"{self.description} to {self.data}"
222
223    @property
224    def default_port(self) -> int:
225        if self.scheme == "dns":
226            return 53
227        return super().default_port
228
229
230class Socks5Mode(ProxyMode):
231    """A SOCKSv5 proxy."""
232    description = "SOCKS v5 proxy"
233    default_port = 1080
234    transport_protocol = TCP
235
236    def __post_init__(self) -> None:
237        _check_empty(self.data)
238
239
240class DnsMode(ProxyMode):
241    """A DNS server."""
242    description = "DNS server"
243    default_port = 53
244    transport_protocol = UDP
245
246    def __post_init__(self) -> None:
247        _check_empty(self.data)
248
249
250class WireGuardMode(ProxyMode):
251    """Proxy Server based on WireGuard"""
252    description = "WireGuard server"
253    default_port = 51820
254    transport_protocol = UDP
255
256    def __post_init__(self) -> None:
257        pass