codekingpro/portable-devtools

1from typing import Optional2 3from mitmproxy import ctx4from mitmproxy import exceptions5from mitmproxy import flow6from mitmproxy import flowfilter7 8 9class Intercept:10    filt: flowfilter.TFilter | None = None11 12    def load(self, loader):13        loader.add_option("intercept_active", bool, False, "Intercept toggle")14        loader.add_option(15            "intercept", Optional[str], None, "Intercept filter expression."16        )17 18    def configure(self, updated):19        if "intercept" in updated:20            if ctx.options.intercept:21                try:22                    self.filt = flowfilter.parse(ctx.options.intercept)23                except ValueError as e:24                    raise exceptions.OptionsError(str(e)) from e25                ctx.options.intercept_active = True26            else:27                self.filt = None28                ctx.options.intercept_active = False29 30    def should_intercept(self, f: flow.Flow) -> bool:31        return bool(32            ctx.options.intercept_active33            and self.filt34            and self.filt(f)35            and not f.is_replay36        )37 38    def process_flow(self, f: flow.Flow) -> None:39        if self.should_intercept(f):40            f.intercept()41 42    # Handlers43 44    def request(self, f):45        self.process_flow(f)46 47    def response(self, f):48        self.process_flow(f)49 50    def tcp_message(self, f):51        self.process_flow(f)52 53    def udp_message(self, f):54        self.process_flow(f)55 56    def dns_request(self, f):57        self.process_flow(f)58 59    def dns_response(self, f):60        self.process_flow(f)61 62    def websocket_message(self, f):63        self.process_flow(f)64