codekingpro/portable-devtools

1import errno2import logging3from typing import cast4 5import tornado.httpserver6import tornado.ioloop7 8from mitmproxy import addons9from mitmproxy import flow10from mitmproxy import log11from mitmproxy import master12from mitmproxy import options13from mitmproxy import optmanager14from mitmproxy.addons import errorcheck15from mitmproxy.addons import eventstore16from mitmproxy.addons import intercept17from mitmproxy.addons import readfile18from mitmproxy.addons import view19from mitmproxy.addons.proxyserver import Proxyserver20from mitmproxy.tools.web import app21from mitmproxy.tools.web import static_viewer22from mitmproxy.tools.web import webaddons23 24logger = logging.getLogger(__name__)25 26 27class WebMaster(master.Master):28    def __init__(self, opts: options.Options, with_termlog: bool = True):29        super().__init__(opts, with_termlog=with_termlog)30        self.view = view.View()31        self.view.sig_view_add.connect(self._sig_view_add)32        self.view.sig_view_remove.connect(self._sig_view_remove)33        self.view.sig_view_update.connect(self._sig_view_update)34        self.view.sig_view_refresh.connect(self._sig_view_refresh)35 36        self.events = eventstore.EventStore()37        self.events.sig_add.connect(self._sig_events_add)38        self.events.sig_refresh.connect(self._sig_events_refresh)39 40        self.options.changed.connect(self._sig_options_update)41 42        self.addons.add(*addons.default_addons())43        self.addons.add(44            webaddons.WebAddon(),45            webaddons.WebAuth(),46            intercept.Intercept(),47            readfile.ReadFileStdin(),48            static_viewer.StaticViewer(),49            self.view,50            self.events,51            errorcheck.ErrorCheck(),52        )53        self.app = app.Application(self, self.options.web_debug)54        self.proxyserver: Proxyserver = self.addons.get("proxyserver")55        self.proxyserver.servers.changed.connect(self._sig_servers_changed)56 57    def _sig_view_add(self, flow: flow.Flow) -> None:58        app.ClientConnection.broadcast_flow("flows/add", flow)59 60    def _sig_view_update(self, flow: flow.Flow) -> None:61        app.ClientConnection.broadcast_flow("flows/update", flow)62 63    def _sig_view_remove(self, flow: flow.Flow, index: int) -> None:64        app.ClientConnection.broadcast(65            type="flows/remove",66            payload=flow.id,67        )68 69    def _sig_view_refresh(self) -> None:70        app.ClientConnection.broadcast_flow_reset()71 72    def _sig_events_add(self, entry: log.LogEntry) -> None:73        app.ClientConnection.broadcast(74            type="events/add",75            payload=app.logentry_to_json(entry),76        )77 78    def _sig_events_refresh(self) -> None:79        app.ClientConnection.broadcast(80            type="events/reset",81        )82 83    def _sig_options_update(self, updated: set[str]) -> None:84        options_dict = optmanager.dump_dicts(self.options, updated)85        app.ClientConnection.broadcast(86            type="options/update",87            payload=options_dict,88        )89 90    def _sig_servers_changed(self) -> None:91        app.ClientConnection.broadcast(92            type="state/update",93            payload={94                "servers": {95                    s.mode.full_spec: s.to_json() for s in self.proxyserver.servers96                }97            },98        )99 100    @property101    def web_url(self) -> str:102        return cast(webaddons.WebAuth, self.addons.get("webauth")).web_url103 104    async def running(self):105        # Register tornado with the current event loop106        tornado.ioloop.IOLoop.current()107 108        # Add our web app.109        http_server = tornado.httpserver.HTTPServer(110            self.app, max_buffer_size=2**32111        )  # 4GB112        try:113            http_server.listen(self.options.web_port, self.options.web_host)114        except OSError as e:115            message = f"Web server failed to listen on {self.options.web_host or '*'}:{self.options.web_port} with {e}"116            if e.errno == errno.EADDRINUSE:117                message += f"\nTry specifying a different port by using `--set web_port={self.options.web_port + 2}`."118            raise OSError(e.errno, message, e.filename) from e119 120        logger.info(f"Web server listening at {self.web_url}")121 122        return await super().running()123