"Port to serve the onboarding app from."
        )

    def configure(self, updated):
        self.host = ctx.options.onboarding_host
        self.port = ctx.options.onboarding_port
        app.config["CONFDIR"] = ctx.options.confdir

    def request(self, f):

            

Reported by Pylint.

                      self.port = ctx.options.onboarding_port
        app.config["CONFDIR"] = ctx.options.confdir

    def request(self, f):
        if ctx.options.onboarding:
            super().request(f)

            

Reported by Pylint.

              from mitmproxy.addons import asgiapp
from mitmproxy.addons.onboardingapp import app
from mitmproxy import ctx

APP_HOST = "mitm.it"
APP_PORT = 80


class Onboarding(asgiapp.WSGIApp):

            

Reported by Pylint.

              APP_PORT = 80


class Onboarding(asgiapp.WSGIApp):
    name = "onboarding"

    def __init__(self):
        super().__init__(app, APP_HOST, APP_PORT)


            

Reported by Pylint.

                  def __init__(self):
        super().__init__(app, APP_HOST, APP_PORT)

    def load(self, loader):
        loader.add_option(
            "onboarding", bool, True,
            "Toggle the mitmproxy onboarding app."
        )
        loader.add_option(

            

Reported by Pylint.

                  def __init__(self):
        super().__init__(app, APP_HOST, APP_PORT)

    def load(self, loader):
        loader.add_option(
            "onboarding", bool, True,
            "Toggle the mitmproxy onboarding app."
        )
        loader.add_option(

            

Reported by Pylint.

                          "Port to serve the onboarding app from."
        )

    def configure(self, updated):
        self.host = ctx.options.onboarding_host
        self.port = ctx.options.onboarding_port
        app.config["CONFDIR"] = ctx.options.confdir

    def request(self, f):

            

Reported by Pylint.

              from mitmproxy.utils import strutils
from . import base


class ViewHex(base.View):
    name = "Hex"

    @staticmethod
    def _format(data):

            

Reported by Pylint.

                  def __call__(self, data, **metadata):
        return "Hex", self._format(data)

    def render_priority(self, data: bytes, **metadata) -> float:
        return 0.2 * strutils.is_mostly_bin(data)

            

Reported by Pylint.

              from mitmproxy.utils import strutils
from . import base


class ViewHex(base.View):
    name = "Hex"

    @staticmethod
    def _format(data):

            

Reported by Pylint.

              from . import base


class ViewHex(base.View):
    name = "Hex"

    @staticmethod
    def _format(data):
        for offset, hexa, s in strutils.hexdump(data):

            

Reported by Pylint.

              
    @staticmethod
    def _format(data):
        for offset, hexa, s in strutils.hexdump(data):
            yield [
                ("offset", offset + " "),
                ("text", hexa + "   "),
                ("text", s)
            ]

            

Reported by Pylint.

                  def __call__(self, data, **metadata):
        return "Hex", self._format(data)

    def render_priority(self, data: bytes, **metadata) -> float:
        return 0.2 * strutils.is_mostly_bin(data)

            

Reported by Pylint.

                  def __call__(self, data, **metadata):
        return "Hex", self._format(data)

    def render_priority(self, data: bytes, **metadata) -> float:
        return 0.2 * strutils.is_mostly_bin(data)

            

Reported by Pylint.

              from typing import List  # noqa

from mitmproxy.utils import strutils
from . import base


class ViewRaw(base.View):
    name = "Raw"


            

Reported by Pylint.

              from typing import List  # noqa

from mitmproxy.utils import strutils
from . import base


class ViewRaw(base.View):
    name = "Raw"


            

Reported by Pylint.

                  def __call__(self, data, **metadata):
        return "Raw", base.format_text(strutils.bytes_to_escaped_str(data, True))

    def render_priority(self, data: bytes, **metadata) -> float:
        return 0.1 * float(bool(data))

            

Reported by Pylint.

              from typing import List  # noqa

from mitmproxy.utils import strutils
from . import base


class ViewRaw(base.View):
    name = "Raw"


            

Reported by Pylint.

              from . import base


class ViewRaw(base.View):
    name = "Raw"

    def __call__(self, data, **metadata):
        return "Raw", base.format_text(strutils.bytes_to_escaped_str(data, True))


            

Reported by Pylint.

                  def __call__(self, data, **metadata):
        return "Raw", base.format_text(strutils.bytes_to_escaped_str(data, True))

    def render_priority(self, data: bytes, **metadata) -> float:
        return 0.1 * float(bool(data))

            

Reported by Pylint.

                  def __call__(self, data, **metadata):
        return "Raw", base.format_text(strutils.bytes_to_escaped_str(data, True))

    def render_priority(self, data: bytes, **metadata) -> float:
        return 0.1 * float(bool(data))

            

Reported by Pylint.

              from typing import Any, Callable, Dict, Iterator, Type

from mitmproxy import controller
from mitmproxy import flow
from mitmproxy import hooks
from mitmproxy import http
from mitmproxy import tcp
from mitmproxy.proxy import layers


            

Reported by Pylint.

              TEventGenerator = Iterator[hooks.Hook]


def _iterate_http(f: http.HTTPFlow) -> TEventGenerator:
    if f.request:
        yield layers.http.HttpRequestHeadersHook(f)
        yield layers.http.HttpRequestHook(f)
    if f.response:
        yield layers.http.HttpResponseHeadersHook(f)

            

Reported by Pylint.

                      message_queue = f.websocket.messages
        f.websocket.messages = []
        yield layers.websocket.WebsocketStartHook(f)
        for m in message_queue:
            f.websocket.messages.append(m)
            yield layers.websocket.WebsocketMessageHook(f)
        yield layers.websocket.WebsocketEndHook(f)
    elif f.error:
        yield layers.http.HttpErrorHook(f)

            

Reported by Pylint.

                      yield layers.http.HttpErrorHook(f)


def _iterate_tcp(f: tcp.TCPFlow) -> TEventGenerator:
    messages = f.messages
    f.messages = []
    f.reply = controller.DummyReply()
    yield layers.tcp.TcpStartHook(f)
    while messages:

            

Reported by Pylint.

              }


def iterate(f: flow.Flow) -> TEventGenerator:
    try:
        e = _iterate_map[type(f)]
    except KeyError as err:
        raise TypeError(f"Unknown flow type: {f.__class__.__name__}") from err
    else:

            

Reported by Pylint.

              }


def iterate(f: flow.Flow) -> TEventGenerator:
    try:
        e = _iterate_map[type(f)]
    except KeyError as err:
        raise TypeError(f"Unknown flow type: {f.__class__.__name__}") from err
    else:

            

Reported by Pylint.

              
def iterate(f: flow.Flow) -> TEventGenerator:
    try:
        e = _iterate_map[type(f)]
    except KeyError as err:
        raise TypeError(f"Unknown flow type: {f.__class__.__name__}") from err
    else:
        yield from e(f)

            

Reported by Pylint.

                          args.append(getattr(self, field.name))
        return args

    def __new__(cls, *args, **kwargs):
        if cls is Hook:
            raise TypeError("Hook may not be instantiated directly.")
        if not is_dataclass(cls):
            raise TypeError("Subclass is not a dataclass.")
        return super().__new__(cls)

            

Reported by Pylint.

                          args.append(getattr(self, field.name))
        return args

    def __new__(cls, *args, **kwargs):
        if cls is Hook:
            raise TypeError("Hook may not be instantiated directly.")
        if not is_dataclass(cls):
            raise TypeError("Subclass is not a dataclass.")
        return super().__new__(cls)

            

Reported by Pylint.

                          raise TypeError("Subclass is not a dataclass.")
        return super().__new__(cls)

    def __init_subclass__(cls, **kwargs):
        # initialize .name attribute. HttpRequestHook -> http_request
        if cls.__dict__.get("name", None) is None:
            name = cls.__name__.replace("Hook", "")
            cls.name = re.sub('(?!^)([A-Z]+)', r'_\1', name).lower()
        if cls.name in all_hooks:

            

Reported by Pylint.

              import re
import warnings
from dataclasses import dataclass, is_dataclass, fields
from typing import ClassVar, Any, Dict, Type, Set, List, TYPE_CHECKING, Sequence

import mitmproxy.flow

if TYPE_CHECKING:
    import mitmproxy.addonmanager

            

Reported by Pylint.

                  import mitmproxy.log


class Hook:
    name: ClassVar[str]

    def args(self) -> List[Any]:
        args = []
        for field in fields(self):

            

Reported by Pylint.

              class Hook:
    name: ClassVar[str]

    def args(self) -> List[Any]:
        args = []
        for field in fields(self):
            args.append(getattr(self, field.name))
        return args


            

Reported by Pylint.

                          cls.name = re.sub('(?!^)([A-Z]+)', r'_\1', name).lower()
        if cls.name in all_hooks:
            other = all_hooks[cls.name]
            warnings.warn(f"Two conflicting event classes for {cls.name}: {cls} and {other}", RuntimeWarning)
        if cls.name == "":
            return  # don't register Hook class.
        all_hooks[cls.name] = cls

        # define a custom hash and __eq__ function so that events are hashable and not comparable.

            

Reported by Pylint.

              import mimetypes
import re
from typing import Tuple, List, Optional
from urllib.parse import quote

from mitmproxy.net.http import headers


def encode(head, l):

            

Reported by Pylint.

              from mitmproxy.net.http import headers


def encode(head, l):
    k = head.get("content-type")
    if k:
        k = headers.parse_content_type(k)
        if k is not None:
            try:

            

Reported by Pylint.

              from mitmproxy.net.http import headers


def encode(head, l):
    k = head.get("content-type")
    if k:
        k = headers.parse_content_type(k)
        if k is not None:
            try:

            

Reported by Pylint.

              from mitmproxy.net.http import headers


def encode(head, l):
    k = head.get("content-type")
    if k:
        k = headers.parse_content_type(k)
        if k is not None:
            try:

            

Reported by Pylint.

                      Takes a multipart boundary encoded string and returns list of (key, value) tuples.
    """
    if content_type:
        ct = headers.parse_content_type(content_type)
        if not ct:
            return []
        try:
            boundary = ct[2]["boundary"].encode("ascii")
        except (KeyError, UnicodeError):

            

Reported by Pylint.

                      except (KeyError, UnicodeError):
            return []

        rx = re.compile(br'\bname="([^"]+)"')
        r = []
        if content is not None:
            for i in content.split(b"--" + boundary):
                parts = i.splitlines()
                if len(parts) > 1 and parts[0][0:2] != b"--":

            

Reported by Pylint.

                          return []

        rx = re.compile(br'\bname="([^"]+)"')
        r = []
        if content is not None:
            for i in content.split(b"--" + boundary):
                parts = i.splitlines()
                if len(parts) > 1 and parts[0][0:2] != b"--":
                    match = rx.search(parts[1])

            

Reported by Pylint.

              
    - Layers: represent protocol layers, e.g. one for TCP, TLS, and so on. Layers are nested, so
      a typical configuration might be ReverseProxy/TLS/TCP.
      Most importantly, layers are implemented using the sans-io pattern (https://sans-io.readthedocs.io/).
      This means that calls return immediately, their is no blocking sync or async code.
    - Server: the proxy server handles all I/O. This is implemented using asyncio, but could be done any other way.
      The ConnectionHandler is subclassed in the Proxyserver addon, which handles the communication with the
      rest of mitmproxy.
    - Events: When I/O actions occur at the proxy server, they are passed to the outermost layer as events,

            

Reported by Pylint.

                    a typical configuration might be ReverseProxy/TLS/TCP.
      Most importantly, layers are implemented using the sans-io pattern (https://sans-io.readthedocs.io/).
      This means that calls return immediately, their is no blocking sync or async code.
    - Server: the proxy server handles all I/O. This is implemented using asyncio, but could be done any other way.
      The ConnectionHandler is subclassed in the Proxyserver addon, which handles the communication with the
      rest of mitmproxy.
    - Events: When I/O actions occur at the proxy server, they are passed to the outermost layer as events,
      e.g. "DataReceived" or "ConnectionClosed".
    - Commands: In the other direction, layers can emit commands to higher layers or the proxy server.

            

Reported by Pylint.

                    Most importantly, layers are implemented using the sans-io pattern (https://sans-io.readthedocs.io/).
      This means that calls return immediately, their is no blocking sync or async code.
    - Server: the proxy server handles all I/O. This is implemented using asyncio, but could be done any other way.
      The ConnectionHandler is subclassed in the Proxyserver addon, which handles the communication with the
      rest of mitmproxy.
    - Events: When I/O actions occur at the proxy server, they are passed to the outermost layer as events,
      e.g. "DataReceived" or "ConnectionClosed".
    - Commands: In the other direction, layers can emit commands to higher layers or the proxy server.
      This is used to e.g. send data, request for new connections to be opened, or to call mitmproxy's

            

Reported by Pylint.

                  - Server: the proxy server handles all I/O. This is implemented using asyncio, but could be done any other way.
      The ConnectionHandler is subclassed in the Proxyserver addon, which handles the communication with the
      rest of mitmproxy.
    - Events: When I/O actions occur at the proxy server, they are passed to the outermost layer as events,
      e.g. "DataReceived" or "ConnectionClosed".
    - Commands: In the other direction, layers can emit commands to higher layers or the proxy server.
      This is used to e.g. send data, request for new connections to be opened, or to call mitmproxy's
      event hooks.
    - Context: The context is the connection context each layer is provided with, which is always a client connection

            

Reported by Pylint.

                    rest of mitmproxy.
    - Events: When I/O actions occur at the proxy server, they are passed to the outermost layer as events,
      e.g. "DataReceived" or "ConnectionClosed".
    - Commands: In the other direction, layers can emit commands to higher layers or the proxy server.
      This is used to e.g. send data, request for new connections to be opened, or to call mitmproxy's
      event hooks.
    - Context: The context is the connection context each layer is provided with, which is always a client connection
      and sometimes also a server connection.
"""

            

Reported by Pylint.

                  - Events: When I/O actions occur at the proxy server, they are passed to the outermost layer as events,
      e.g. "DataReceived" or "ConnectionClosed".
    - Commands: In the other direction, layers can emit commands to higher layers or the proxy server.
      This is used to e.g. send data, request for new connections to be opened, or to call mitmproxy's
      event hooks.
    - Context: The context is the connection context each layer is provided with, which is always a client connection
      and sometimes also a server connection.
"""

            

Reported by Pylint.

                  - Commands: In the other direction, layers can emit commands to higher layers or the proxy server.
      This is used to e.g. send data, request for new connections to be opened, or to call mitmproxy's
      event hooks.
    - Context: The context is the connection context each layer is provided with, which is always a client connection
      and sometimes also a server connection.
"""

            

Reported by Pylint.

              def map(km):
    km.add(":", "console.command ", ["commonkey", "global"], "Command prompt")
    km.add(";", "console.command flow.comment @focus ''", ["flowlist", "flowview"], "Add comment to flow")
    km.add("?", "console.view.help", ["global"], "View help")
    km.add("B", "browser.start", ["global"], "Start an attached browser")
    km.add("C", "console.view.commands", ["global"], "View commands")
    km.add("K", "console.view.keybindings", ["global"], "View key bindings")
    km.add("O", "console.view.options", ["commonkey", "global"], "View options")
    km.add("E", "console.view.eventlog", ["commonkey", "global"], "View event log")

            

Reported by Pylint.

              def map(km):
    km.add(":", "console.command ", ["commonkey", "global"], "Command prompt")
    km.add(";", "console.command flow.comment @focus ''", ["flowlist", "flowview"], "Add comment to flow")
    km.add("?", "console.view.help", ["global"], "View help")
    km.add("B", "browser.start", ["global"], "Start an attached browser")
    km.add("C", "console.view.commands", ["global"], "View commands")
    km.add("K", "console.view.keybindings", ["global"], "View key bindings")
    km.add("O", "console.view.options", ["commonkey", "global"], "View options")
    km.add("E", "console.view.eventlog", ["commonkey", "global"], "View event log")

            

Reported by Pylint.

              def map(km):
    km.add(":", "console.command ", ["commonkey", "global"], "Command prompt")
    km.add(";", "console.command flow.comment @focus ''", ["flowlist", "flowview"], "Add comment to flow")
    km.add("?", "console.view.help", ["global"], "View help")
    km.add("B", "browser.start", ["global"], "Start an attached browser")
    km.add("C", "console.view.commands", ["global"], "View commands")
    km.add("K", "console.view.keybindings", ["global"], "View key bindings")
    km.add("O", "console.view.options", ["commonkey", "global"], "View options")
    km.add("E", "console.view.eventlog", ["commonkey", "global"], "View event log")

            

Reported by Pylint.

              def map(km):
    km.add(":", "console.command ", ["commonkey", "global"], "Command prompt")
    km.add(";", "console.command flow.comment @focus ''", ["flowlist", "flowview"], "Add comment to flow")
    km.add("?", "console.view.help", ["global"], "View help")
    km.add("B", "browser.start", ["global"], "Start an attached browser")
    km.add("C", "console.view.commands", ["global"], "View commands")
    km.add("K", "console.view.keybindings", ["global"], "View key bindings")
    km.add("O", "console.view.options", ["commonkey", "global"], "View options")
    km.add("E", "console.view.eventlog", ["commonkey", "global"], "View event log")

            

Reported by Pylint.

              def map(km):
    km.add(":", "console.command ", ["commonkey", "global"], "Command prompt")
    km.add(";", "console.command flow.comment @focus ''", ["flowlist", "flowview"], "Add comment to flow")
    km.add("?", "console.view.help", ["global"], "View help")
    km.add("B", "browser.start", ["global"], "Start an attached browser")
    km.add("C", "console.view.commands", ["global"], "View commands")
    km.add("K", "console.view.keybindings", ["global"], "View key bindings")
    km.add("O", "console.view.options", ["commonkey", "global"], "View options")
    km.add("E", "console.view.eventlog", ["commonkey", "global"], "View event log")

            

Reported by Pylint.

              def map(km):
    km.add(":", "console.command ", ["commonkey", "global"], "Command prompt")
    km.add(";", "console.command flow.comment @focus ''", ["flowlist", "flowview"], "Add comment to flow")
    km.add("?", "console.view.help", ["global"], "View help")
    km.add("B", "browser.start", ["global"], "Start an attached browser")
    km.add("C", "console.view.commands", ["global"], "View commands")
    km.add("K", "console.view.keybindings", ["global"], "View key bindings")
    km.add("O", "console.view.options", ["commonkey", "global"], "View options")
    km.add("E", "console.view.eventlog", ["commonkey", "global"], "View event log")

            

Reported by Pylint.

                  km.add("ctrl f", "console.nav.pagedown", ["global"], "Page down")
    km.add("ctrl b", "console.nav.pageup", ["global"], "Page up")

    km.add("I", "set intercept_active toggle", ["global"], "Toggle whether the filtering via the intercept option is enabled")
    km.add("i", "console.command.set intercept", ["global"], "Set intercept")
    km.add("W", "console.command.set save_stream_file", ["global"], "Stream to file")
    km.add("A", "flow.resume @all", ["flowlist", "flowview"], "Resume all intercepted flows")
    km.add("a", "flow.resume @focus", ["flowlist", "flowview"], "Resume this intercepted flow")
    km.add(

            

Reported by Pylint.

              
    def __init__(
        self,
        options: options.Options,
        with_termlog=True,
        with_dumper=True,
    ) -> None:
        super().__init__(options)
        self.errorcheck = ErrorCheck()

            

Reported by Pylint.

              from mitmproxy import addons
from mitmproxy import options
from mitmproxy import master
from mitmproxy.addons import dumper, termlog, keepserving, readfile


class ErrorCheck:
    def __init__(self):
        self.has_errored = False

            

Reported by Pylint.

              from mitmproxy.addons import dumper, termlog, keepserving, readfile


class ErrorCheck:
    def __init__(self):
        self.has_errored = False

    def add_log(self, e):
        if e.level == "error":

            

Reported by Pylint.

              from mitmproxy.addons import dumper, termlog, keepserving, readfile


class ErrorCheck:
    def __init__(self):
        self.has_errored = False

    def add_log(self, e):
        if e.level == "error":

            

Reported by Pylint.

                  def __init__(self):
        self.has_errored = False

    def add_log(self, e):
        if e.level == "error":
            self.has_errored = True


class DumpMaster(master.Master):

            

Reported by Pylint.

                  def __init__(self):
        self.has_errored = False

    def add_log(self, e):
        if e.level == "error":
            self.has_errored = True


class DumpMaster(master.Master):

            

Reported by Pylint.

                          self.has_errored = True


class DumpMaster(master.Master):

    def __init__(
        self,
        options: options.Options,
        with_termlog=True,

            

Reported by Pylint.

              
import click

from mitmproxy.io import tnetstring


def read_tnetstring(input):
    # tnetstring throw a ValueError on EOF, which is hard to catch
    # because they raise ValueErrors for a couple of other reasons.

            

Reported by Pylint.

              

if __name__ == "__main__":
    inspect()

            

Reported by Pylint.

              from mitmproxy.io import tnetstring


def read_tnetstring(input):
    # tnetstring throw a ValueError on EOF, which is hard to catch
    # because they raise ValueErrors for a couple of other reasons.
    # Check for EOF to avoid this.
    if not input.read(1):
        return None

            

Reported by Pylint.

              
@click.command()
@click.argument("input", type=click.File('rb'))
def inspect(input):
    """
    pretty-print a dumpfile
    """
    while True:
        data = read_tnetstring(input)

            

Reported by Pylint.

              #!/usr/bin/env python3
from pprint import pprint

import click

from mitmproxy.io import tnetstring


def read_tnetstring(input):

            

Reported by Pylint.

              from mitmproxy.io import tnetstring


def read_tnetstring(input):
    # tnetstring throw a ValueError on EOF, which is hard to catch
    # because they raise ValueErrors for a couple of other reasons.
    # Check for EOF to avoid this.
    if not input.read(1):
        return None

            

Reported by Pylint.

                  # tnetstring throw a ValueError on EOF, which is hard to catch
    # because they raise ValueErrors for a couple of other reasons.
    # Check for EOF to avoid this.
    if not input.read(1):
        return None
    else:
        input.seek(-1, 1)
    return tnetstring.load(input)


            

Reported by Pylint.