codekingpro/portable-devtools

1import logging2import os3from collections.abc import Sequence4 5import mitmproxy.types6from mitmproxy import command7from mitmproxy import ctx8from mitmproxy import exceptions9from mitmproxy import flow10from mitmproxy import hooks11from mitmproxy import optmanager12from mitmproxy.log import ALERT13from mitmproxy.net.http import status_codes14from mitmproxy.utils import emoji15 16logger = logging.getLogger(__name__)17 18CONF_DIR = "~/.mitmproxy"19LISTEN_PORT = 808020 21 22class Core:23    def configure(self, updated):24        opts = ctx.options25        if opts.add_upstream_certs_to_client_chain and not opts.upstream_cert:26            raise exceptions.OptionsError(27                "add_upstream_certs_to_client_chain requires the upstream_cert option to be enabled."28            )29        if "client_certs" in updated:30            if opts.client_certs:31                client_certs = os.path.expanduser(opts.client_certs)32                if not os.path.exists(client_certs):33                    raise exceptions.OptionsError(34                        f"Client certificate path does not exist: {opts.client_certs}"35                    )36 37    @command.command("set")38    def set(self, option: str, *value: str) -> None:39        """40        Set an option. When the value is omitted, booleans are set to true,41        strings and integers are set to None (if permitted), and sequences42        are emptied. Boolean values can be true, false or toggle.43        Multiple values are concatenated with a single space.44        """45        if value:46            specs = [f"{option}={v}" for v in value]47        else:48            specs = [option]49        try:50            ctx.options.set(*specs)51        except exceptions.OptionsError as e:52            raise exceptions.CommandError(e) from e53 54    @command.command("flow.resume")55    def resume(self, flows: Sequence[flow.Flow]) -> None:56        """57        Resume flows if they are intercepted.58        """59        intercepted = [i for i in flows if i.intercepted]60        for f in intercepted:61            f.resume()62        ctx.master.addons.trigger(hooks.UpdateHook(intercepted))63 64    # FIXME: this will become view.mark later65    @command.command("flow.mark")66    def mark(self, flows: Sequence[flow.Flow], marker: mitmproxy.types.Marker) -> None:67        """68        Mark flows.69        """70        updated = []71        if not (marker == "" or marker in emoji.emoji):72            raise exceptions.CommandError(f"invalid marker value")73 74        for i in flows:75            i.marked = marker76            updated.append(i)77        ctx.master.addons.trigger(hooks.UpdateHook(updated))78 79    # FIXME: this will become view.mark.toggle later80    @command.command("flow.mark.toggle")81    def mark_toggle(self, flows: Sequence[flow.Flow]) -> None:82        """83        Toggle mark for flows.84        """85        for i in flows:86            if i.marked:87                i.marked = ""88            else:89                i.marked = ":default:"90        ctx.master.addons.trigger(hooks.UpdateHook(flows))91 92    @command.command("flow.kill")93    def kill(self, flows: Sequence[flow.Flow]) -> None:94        """95        Kill running flows.96        """97        updated = []98        for f in flows:99            if f.killable:100                f.kill()101                updated.append(f)102        logger.log(ALERT, "Killed %s flows." % len(updated))103        ctx.master.addons.trigger(hooks.UpdateHook(updated))104 105    # FIXME: this will become view.revert later106    @command.command("flow.revert")107    def revert(self, flows: Sequence[flow.Flow]) -> None:108        """109        Revert flow changes.110        """111        updated = []112        for f in flows:113            if f.modified():114                f.revert()115                updated.append(f)116        logger.log(ALERT, "Reverted %s flows." % len(updated))117        ctx.master.addons.trigger(hooks.UpdateHook(updated))118 119    @command.command("flow.set.options")120    def flow_set_options(self) -> Sequence[str]:121        return [122            "host",123            "status_code",124            "method",125            "path",126            "url",127            "reason",128        ]129 130    @command.command("flow.set")131    @command.argument("attr", type=mitmproxy.types.Choice("flow.set.options"))132    def flow_set(self, flows: Sequence[flow.Flow], attr: str, value: str) -> None:133        """134        Quickly set a number of common values on flows.135        """136        val: int | str = value137        if attr == "status_code":138            try:139                val = int(val)  # type: ignore140            except ValueError as v:141                raise exceptions.CommandError(142                    "Status code is not an integer: %s" % val143                ) from v144 145        updated = []146        for f in flows:147            req = getattr(f, "request", None)148            rupdate = True149            if req:150                if attr == "method":151                    req.method = val152                elif attr == "host":153                    req.host = val154                elif attr == "path":155                    req.path = val156                elif attr == "url":157                    try:158                        req.url = val159                    except ValueError as e:160                        raise exceptions.CommandError(161                            f"URL {val!r} is invalid: {e}"162                        ) from e163                else:164                    self.rupdate = False165 166            resp = getattr(f, "response", None)167            supdate = True168            if resp:169                if attr == "status_code":170                    resp.status_code = val171                    if val in status_codes.RESPONSES:172                        resp.reason = status_codes.RESPONSES[val]  # type: ignore173                elif attr == "reason":174                    resp.reason = val175                else:176                    supdate = False177 178            if rupdate or supdate:179                updated.append(f)180 181        ctx.master.addons.trigger(hooks.UpdateHook(updated))182        logger.log(ALERT, f"Set {attr} on  {len(updated)} flows.")183 184    @command.command("flow.decode")185    def decode(self, flows: Sequence[flow.Flow], part: str) -> None:186        """187        Decode flows.188        """189        updated = []190        for f in flows:191            p = getattr(f, part, None)192            if p:193                f.backup()194                p.decode()195                updated.append(f)196        ctx.master.addons.trigger(hooks.UpdateHook(updated))197        logger.log(ALERT, "Decoded %s flows." % len(updated))198 199    @command.command("flow.encode.toggle")200    def encode_toggle(self, flows: Sequence[flow.Flow], part: str) -> None:201        """202        Toggle flow encoding on and off, using deflate for encoding.203        """204        updated = []205        for f in flows:206            p = getattr(f, part, None)207            if p:208                f.backup()209                current_enc = p.headers.get("content-encoding", "identity")210                if current_enc == "identity":211                    p.encode("deflate")212                else:213                    p.decode()214                updated.append(f)215        ctx.master.addons.trigger(hooks.UpdateHook(updated))216        logger.log(ALERT, "Toggled encoding on %s flows." % len(updated))217 218    @command.command("flow.encode")219    @command.argument("encoding", type=mitmproxy.types.Choice("flow.encode.options"))220    def encode(221        self,222        flows: Sequence[flow.Flow],223        part: str,224        encoding: str,225    ) -> None:226        """227        Encode flows with a specified encoding.228        """229        updated = []230        for f in flows:231            p = getattr(f, part, None)232            if p:233                current_enc = p.headers.get("content-encoding", "identity")234                if current_enc == "identity":235                    f.backup()236                    p.encode(encoding)237                    updated.append(f)238        ctx.master.addons.trigger(hooks.UpdateHook(updated))239        logger.log(ALERT, "Encoded %s flows." % len(updated))240 241    @command.command("flow.encode.options")242    def encode_options(self) -> Sequence[str]:243        """244        The possible values for an encoding specification.245        """246        return ["gzip", "deflate", "br", "zstd"]247 248    @command.command("options.load")249    def options_load(self, path: mitmproxy.types.Path) -> None:250        """251        Load options from a file.252        """253        try:254            optmanager.load_paths(ctx.options, path)255        except (OSError, exceptions.OptionsError) as e:256            raise exceptions.CommandError("Could not load options - %s" % e) from e257 258    @command.command("options.save")259    def options_save(self, path: mitmproxy.types.Path) -> None:260        """261        Save options to a file.262        """263        try:264            optmanager.save(ctx.options, path)265        except OSError as e:266            raise exceptions.CommandError("Could not save options - %s" % e) from e267 268    @command.command("options.reset")269    def options_reset(self) -> None:270        """271        Reset all options to defaults.272        """273        ctx.options.reset()274 275    @command.command("options.reset.one")276    def options_reset_one(self, name: str) -> None:277        """278        Reset one option to its default value.279        """280        if name not in ctx.options:281            raise exceptions.CommandError("No such option: %s" % name)282        setattr(283            ctx.options,284            name,285            ctx.options.default(name),286        )287