import typing

from mitmproxy import exceptions
from mitmproxy import flow
from mitmproxy import ctx

from mitmproxy.tools.console import overlay
from mitmproxy.tools.console import signals


            

Reported by Pylint.

              from mitmproxy.tools.console import signals


class CommandExecutor:
    def __init__(self, master):
        self.master = master

    def __call__(self, cmd):
        if cmd.strip():

            

Reported by Pylint.

              from mitmproxy.tools.console import signals


class CommandExecutor:
    def __init__(self, master):
        self.master = master

    def __call__(self, cmd):
        if cmd.strip():

            

Reported by Pylint.

                      if cmd.strip():
            try:
                ret = self.master.commands.execute(cmd)
            except exceptions.CommandError as e:
                ctx.log.error(str(e))
            else:
                if ret:
                    if type(ret) == typing.Sequence[flow.Flow]:
                        signals.status_message.send(

            

Reported by Pylint.

                              ctx.log.error(str(e))
            else:
                if ret:
                    if type(ret) == typing.Sequence[flow.Flow]:
                        signals.status_message.send(
                            message="Command returned %s flows" % len(ret)
                        )
                    elif type(ret) == flow.Flow:
                        signals.status_message.send(

            

Reported by Pylint.

                                      signals.status_message.send(
                            message="Command returned %s flows" % len(ret)
                        )
                    elif type(ret) == flow.Flow:
                        signals.status_message.send(
                            message="Command returned 1 flow"
                        )
                    else:
                        self.master.overlay(

            

Reported by Pylint.

              from hypothesis import given, example
from hypothesis.strategies import binary, integers

from mitmproxy.net.tls import ClientHello
from mitmproxy.proxy.layers.tls import parse_client_hello

client_hello_with_extensions = bytes.fromhex(
    "16030300bb"  # record layer
    "010000b7"  # handshake layer

            

Reported by Pylint.

              from hypothesis import given, example
from hypothesis.strategies import binary, integers

from mitmproxy.net.tls import ClientHello
from mitmproxy.proxy.layers.tls import parse_client_hello

client_hello_with_extensions = bytes.fromhex(
    "16030300bb"  # record layer
    "010000b7"  # handshake layer

            

Reported by Pylint.

              from hypothesis import given, example
from hypothesis.strategies import binary, integers

from mitmproxy.net.tls import ClientHello
from mitmproxy.proxy.layers.tls import parse_client_hello

client_hello_with_extensions = bytes.fromhex(
    "16030300bb"  # record layer
    "010000b7"  # handshake layer

            

Reported by Pylint.

              
@given(i=integers(0, len(client_hello_with_extensions)), data=binary())
@example(i=183, data=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00')
def test_fuzz_h2_request_chunks(i, data):
    try:
        ch = parse_client_hello(client_hello_with_extensions[:i] + data)
    except ValueError:
        pass
    else:

            

Reported by Pylint.

              @example(i=183, data=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00')
def test_fuzz_h2_request_chunks(i, data):
    try:
        ch = parse_client_hello(client_hello_with_extensions[:i] + data)
    except ValueError:
        pass
    else:
        assert ch is None or isinstance(ch, ClientHello)

            

Reported by Pylint.

                  except ValueError:
        pass
    else:
        assert ch is None or isinstance(ch, ClientHello)

            

Reported by Bandit.

                          The view focus has changed. Layout objects should implement the API
            rather than directly subscribing to events.
        """
        pass

    def view_changed(self):
        """
            The view list has changed.
        """

            

Reported by Pylint.

                      """
            The view list has changed.
        """
        pass

    def layout_popping(self):
        """
            We are just about to pop a window off the stack, or exit an overlay.
        """

            

Reported by Pylint.

                      """
            We are just about to pop a window off the stack, or exit an overlay.
        """
        pass

    def layout_pushed(self, prev):
        """
            We have just pushed a window onto the stack.
        """

            

Reported by Pylint.

                      """
            We have just pushed a window onto the stack.
        """
        pass

            

Reported by Pylint.

              class LayoutWidget:
    """
        All top-level layout widgets and all widgets that may be set in an
        overlay must comply with this API.
    """
    # Title is only required for windows, not overlay components
    title = ""
    keyctx = ""


            

Reported by Pylint.

              from mitmproxy.script import concurrent


@concurrent
def load(v):
    pass

            

Reported by Pylint.

              

@concurrent
def load(v):
    pass

            

Reported by Pylint.

              from mitmproxy.script import concurrent


@concurrent
def load(v):
    pass

            

Reported by Pylint.

              

@concurrent
def load(v):
    pass

            

Reported by Pylint.

              

@concurrent
def load(v):
    pass

            

Reported by Pylint.

              from mitmproxy.test import tflow, taddons
from mitmproxy.addons.comment import Comment


def test_comment():
    c = Comment()
    f = tflow.tflow()

    with taddons.context():

            

Reported by Pylint.

              from mitmproxy.addons.comment import Comment


def test_comment():
    c = Comment()
    f = tflow.tflow()

    with taddons.context():
        c.comment([f], "foo")

            

Reported by Pylint.

              

def test_comment():
    c = Comment()
    f = tflow.tflow()

    with taddons.context():
        c.comment([f], "foo")


            

Reported by Pylint.

              
def test_comment():
    c = Comment()
    f = tflow.tflow()

    with taddons.context():
        c.comment([f], "foo")

    assert f.comment == "foo"

            

Reported by Pylint.

                  with taddons.context():
        c.comment([f], "foo")

    assert f.comment == "foo"

            

Reported by Bandit.

                  def __getattr__(self, k):
        if k in self.names:
            return self.names[k]
        raise AttributeError("No such attribute: %s", k)

    def get_name(self, n, default=None):
        return self.values.get(n, default)

            

Reported by Pylint.

              class BiDi:

    """
        A wee utility class for keeping bi-directional mappings, like field
        constants in protocols. Names are attributes on the object, dict-like
        access maps values to names:

        CONST = BiDi(a=1, b=2)
        assert CONST.a == 1

            

Reported by Pylint.

                  def __init__(self, **kwargs):
        self.names = kwargs
        self.values = {}
        for k, v in kwargs.items():
            self.values[v] = k
        if len(self.names) != len(self.values):
            raise ValueError("Duplicate values not allowed.")

    def __getattr__(self, k):

            

Reported by Pylint.

                          return self.names[k]
        raise AttributeError("No such attribute: %s", k)

    def get_name(self, n, default=None):
        return self.values.get(n, default)

            

Reported by Pylint.

                          return self.names[k]
        raise AttributeError("No such attribute: %s", k)

    def get_name(self, n, default=None):
        return self.values.get(n, default)

            

Reported by Pylint.

              import pytest

from mitmproxy import connection
from mitmproxy.addons import block
from mitmproxy.test import taddons


@pytest.mark.parametrize("block_global, block_private, should_be_killed, address", [
    # block_global: loopback

            

Reported by Pylint.

              import pytest

from mitmproxy import connection
from mitmproxy.addons import block
from mitmproxy.test import taddons


@pytest.mark.parametrize("block_global, block_private, should_be_killed, address", [
    # block_global: loopback

            

Reported by Pylint.

                  (False, True, False, (r"::ffff:216.58.207.174%scope",)),
    (False, True, False, ("2001:4860:4860::8888",)),
])
@pytest.mark.asyncio
async def test_block_global(block_global, block_private, should_be_killed, address):
    ar = block.Block()
    with taddons.context(ar) as tctx:
        tctx.configure(ar, block_global=block_global, block_private=block_private)
        client = connection.Client(address, ("127.0.0.1", 8080), 1607699500)

            

Reported by Pylint.

              ])
@pytest.mark.asyncio
async def test_block_global(block_global, block_private, should_be_killed, address):
    ar = block.Block()
    with taddons.context(ar) as tctx:
        tctx.configure(ar, block_global=block_global, block_private=block_private)
        client = connection.Client(address, ("127.0.0.1", 8080), 1607699500)
        ar.client_connected(client)
        assert bool(client.error) == should_be_killed

            

Reported by Pylint.

                      tctx.configure(ar, block_global=block_global, block_private=block_private)
        client = connection.Client(address, ("127.0.0.1", 8080), 1607699500)
        ar.client_connected(client)
        assert bool(client.error) == should_be_killed

            

Reported by Bandit.

              """
import time

from mitmproxy.script import concurrent


@concurrent  # Remove this and see what happens
def request(flow):
    # This is ugly in mitmproxy's UI, but you don't want to use mitmproxy.ctx.log from a different thread.

            

Reported by Pylint.

              """
Make events hooks non-blocking.

When event hooks are decorated with @concurrent, they will be run in their own thread, freeing the main event loop.
Please note that this generally opens the door to race conditions and decreases performance if not required.
"""
import time

from mitmproxy.script import concurrent

            

Reported by Pylint.

              Make events hooks non-blocking.

When event hooks are decorated with @concurrent, they will be run in their own thread, freeing the main event loop.
Please note that this generally opens the door to race conditions and decreases performance if not required.
"""
import time

from mitmproxy.script import concurrent


            

Reported by Pylint.

              

@concurrent  # Remove this and see what happens
def request(flow):
    # This is ugly in mitmproxy's UI, but you don't want to use mitmproxy.ctx.log from a different thread.
    print(f"handle request: {flow.request.host}{flow.request.path}")
    time.sleep(5)
    print(f"start  request: {flow.request.host}{flow.request.path}")

            

Reported by Pylint.

              
@concurrent  # Remove this and see what happens
def request(flow):
    # This is ugly in mitmproxy's UI, but you don't want to use mitmproxy.ctx.log from a different thread.
    print(f"handle request: {flow.request.host}{flow.request.path}")
    time.sleep(5)
    print(f"start  request: {flow.request.host}{flow.request.path}")

            

Reported by Pylint.

              parse_host_header = re.compile(r"^(?P<host>[^:]+|\[.+\])(?::(?P<port>\d+))?$")


class Rerouter:
    def request(self, flow):
        if flow.client_conn.tls_established:
            flow.request.scheme = "https"
            sni = flow.client_conn.connection.get_servername()
            port = 443

            

Reported by Pylint.

              parse_host_header = re.compile(r"^(?P<host>[^:]+|\[.+\])(?::(?P<port>\d+))?$")


class Rerouter:
    def request(self, flow):
        if flow.client_conn.tls_established:
            flow.request.scheme = "https"
            sni = flow.client_conn.connection.get_servername()
            port = 443

            

Reported by Pylint.

              

class Rerouter:
    def request(self, flow):
        if flow.client_conn.tls_established:
            flow.request.scheme = "https"
            sni = flow.client_conn.connection.get_servername()
            port = 443
        else:

            

Reported by Pylint.

              

class Rerouter:
    def request(self, flow):
        if flow.client_conn.tls_established:
            flow.request.scheme = "https"
            sni = flow.client_conn.connection.get_servername()
            port = 443
        else:

            

Reported by Pylint.

                          port = 80

        host_header = flow.request.host_header
        m = parse_host_header.match(host_header)
        if m:
            host_header = m.group("host").strip("[]")
            if m.group("port"):
                port = int(m.group("port"))


            

Reported by Pylint.

              import argparse


def common_options(parser, opts):
    parser.add_argument(
        '--version',
        action='store_true',
        help="show version number and exit",
        dest='version',

            

Reported by Pylint.

              import argparse


def common_options(parser, opts):
    parser.add_argument(
        '--version',
        action='store_true',
        help="show version number and exit",
        dest='version',

            

Reported by Pylint.

                  opts.make_parser(group, "modify_headers", metavar="PATTERN", short="H")


def mitmproxy(opts):
    parser = argparse.ArgumentParser(usage="%(prog)s [options]")
    common_options(parser, opts)

    opts.make_parser(parser, "console_layout")
    opts.make_parser(parser, "console_layout_headers")

            

Reported by Pylint.

                  return parser


def mitmdump(opts):
    parser = argparse.ArgumentParser(usage="%(prog)s [options] [filter]")

    common_options(parser, opts)
    opts.make_parser(parser, "flow_detail", metavar = "LEVEL")
    parser.add_argument(

            

Reported by Pylint.

                  return parser


def mitmweb(opts):
    parser = argparse.ArgumentParser(usage="%(prog)s [options]")

    group = parser.add_argument_group("Mitmweb")
    opts.make_parser(group, "web_open_browser")
    opts.make_parser(group, "web_port", metavar="PORT")

            

Reported by Pylint.