from mitmproxy import ctx


def running():
    ctx.master.shutdown()
            

Reported by Pylint.

              from mitmproxy import ctx


def running():
    ctx.master.shutdown()
            

Reported by Pylint.

              from mitmproxy import ctx


def running():
    ctx.master.shutdown()
            

Reported by Pylint.

              

def running():
    ctx.master.shutdown()
            

Reported by Pylint.

              """Process individual messages from a WebSocket connection."""
import re
from mitmproxy import ctx, http


def websocket_message(flow: http.HTTPFlow):
    assert flow.websocket is not None  # make type checker happy
    # get the latest message
    message = flow.websocket.messages[-1]

            

Reported by Pylint.

              """Process individual messages from a WebSocket connection."""
import re
from mitmproxy import ctx, http


def websocket_message(flow: http.HTTPFlow):
    assert flow.websocket is not None  # make type checker happy
    # get the latest message
    message = flow.websocket.messages[-1]

            

Reported by Pylint.

              from mitmproxy import ctx, http


def websocket_message(flow: http.HTTPFlow):
    assert flow.websocket is not None  # make type checker happy
    # get the latest message
    message = flow.websocket.messages[-1]

    # was the message sent from the client or server?

            

Reported by Pylint.

              

def websocket_message(flow: http.HTTPFlow):
    assert flow.websocket is not None  # make type checker happy
    # get the latest message
    message = flow.websocket.messages[-1]

    # was the message sent from the client or server?
    if message.from_client:

            

Reported by Bandit.

              import sys
import re

DEPRECATED = """
--confdir
-Z
--body-size-limit
--stream
--palette

            

Reported by Pylint.

              }


def check():
    args = sys.argv[1:]
    print()
    if "-U" in args:
        print("-U is deprecated, please use --mode upstream:SPEC instead")


            

Reported by Pylint.

              }


def check():
    args = sys.argv[1:]
    print()
    if "-U" in args:
        print("-U is deprecated, please use --mode upstream:SPEC instead")


            

Reported by Pylint.

              
    # Check for underscores in the options. Options always follow '--'.
    for argument in args:
        underscoreParam = re.search(r'[-]{2}((.*?_)(.*?(\s|$)))+', argument)
        if underscoreParam is not None:
            print("{} uses underscores, please use hyphens {}".format(
                argument,
                argument.replace('_', '-'))
            )

            

Reported by Pylint.

              
import typing

import urwid
from mitmproxy.tools.console.grideditor import base
from mitmproxy.utils import strutils


class Column(base.Column):

            

Reported by Pylint.

              from mitmproxy.utils import strutils


class Column(base.Column):
    def Display(self, data):
        return Display(data)

    Edit = Display


            

Reported by Pylint.

                      return ""


class Display(base.Cell):
    def __init__(self, data: typing.Any) -> None:
        self.data = data
        if isinstance(data, bytes):
            data = strutils.bytes_to_escaped_str(data)
        if not isinstance(data, str):

            

Reported by Pylint.

                          data = strutils.bytes_to_escaped_str(data)
        if not isinstance(data, str):
            data = repr(data)
        w = urwid.Text(data, wrap="any")
        super().__init__(w)

    def get_data(self) -> typing.Any:
        return self.data

            

Reported by Pylint.

              from mitmproxy.net.http import status_codes


def test_simple():
    assert status_codes.IM_A_TEAPOT == 418
    assert status_codes.RESPONSES[418] == "I'm a teapot"

            

Reported by Pylint.

              from mitmproxy.net.http import status_codes


def test_simple():
    assert status_codes.IM_A_TEAPOT == 418
    assert status_codes.RESPONSES[418] == "I'm a teapot"

            

Reported by Pylint.

              

def test_simple():
    assert status_codes.IM_A_TEAPOT == 418
    assert status_codes.RESPONSES[418] == "I'm a teapot"

            

Reported by Bandit.

              
def test_simple():
    assert status_codes.IM_A_TEAPOT == 418
    assert status_codes.RESPONSES[418] == "I'm a teapot"

            

Reported by Bandit.

              from mitmproxy.net.http import user_agents


def test_get_shortcut():
    assert user_agents.get_by_shortcut("c")[0] == "chrome"
    assert not user_agents.get_by_shortcut("_")

            

Reported by Pylint.

              from mitmproxy.net.http import user_agents


def test_get_shortcut():
    assert user_agents.get_by_shortcut("c")[0] == "chrome"
    assert not user_agents.get_by_shortcut("_")

            

Reported by Pylint.

              

def test_get_shortcut():
    assert user_agents.get_by_shortcut("c")[0] == "chrome"
    assert not user_agents.get_by_shortcut("_")

            

Reported by Bandit.

              
def test_get_shortcut():
    assert user_agents.get_by_shortcut("c")[0] == "chrome"
    assert not user_agents.get_by_shortcut("_")

            

Reported by Bandit.

              from hypothesis import given
from hypothesis.strategies import binary

from mitmproxy import options
from mitmproxy.connection import Client
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import DataReceived
from mitmproxy.proxy.layers.modes import Socks5Proxy


            

Reported by Pylint.

              from hypothesis import given
from hypothesis.strategies import binary

from mitmproxy import options
from mitmproxy.connection import Client
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import DataReceived
from mitmproxy.proxy.layers.modes import Socks5Proxy


            

Reported by Pylint.

              from hypothesis import given
from hypothesis.strategies import binary

from mitmproxy import options
from mitmproxy.connection import Client
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import DataReceived
from mitmproxy.proxy.layers.modes import Socks5Proxy


            

Reported by Pylint.

              

@given(binary())
def test_socks5_fuzz(data):
    layer = Socks5Proxy(tctx)
    list(layer.handle_event(DataReceived(tctx.client, data)))

            

Reported by Pylint.

              import collections
import typing  # noqa

import blinker

from mitmproxy import command
from mitmproxy.log import LogEntry



            

Reported by Pylint.

              from mitmproxy.log import LogEntry


class EventStore:
    def __init__(self, size=10000):
        self.data: typing.Deque[LogEntry] = collections.deque(maxlen=size)
        self.sig_add = blinker.Signal()
        self.sig_refresh = blinker.Signal()


            

Reported by Pylint.

                      self.sig_refresh = blinker.Signal()

    @property
    def size(self) -> typing.Optional[int]:
        return self.data.maxlen

    def add_log(self, entry: LogEntry) -> None:
        self.data.append(entry)
        self.sig_add.send(self, entry=entry)

            

Reported by Pylint.

                  def size(self) -> typing.Optional[int]:
        return self.data.maxlen

    def add_log(self, entry: LogEntry) -> None:
        self.data.append(entry)
        self.sig_add.send(self, entry=entry)

    @command.command("eventstore.clear")
    def clear(self) -> None:

            

Reported by Pylint.

              from dataclasses import dataclass

from mitmproxy import http
from mitmproxy.proxy import commands


@dataclass
class HttpRequestHeadersHook(commands.StartHook):
    """

            

Reported by Pylint.

                  The full HTTP request has been read.

    Note: If request streaming is active, this event fires after the entire body has been streamed.
    HTTP trailers, if present, have not been transmitted to the server yet and can still be modified.
    Enabling streaming may cause unexpected event sequences: For example, `response` may now occur
    before `request` because the server replied with "413 Payload Too Large" during upload.
    """
    name = "request"
    flow: http.HTTPFlow

            

Reported by Pylint.

                  The full HTTP response has been read.

    Note: If response streaming is active, this event fires after the entire body has been streamed.
    HTTP trailers, if present, have not been transmitted to the client yet and can still be modified.
    """
    name = "response"
    flow: http.HTTPFlow



            

Reported by Pylint.

              
    This event only occurs in regular and upstream proxy modes
    when the client instructs mitmproxy to open a connection to an upstream host.
    Setting a non 2xx response on the flow will return the response to the client and abort the connection.

    CONNECT requests are HTTP proxy instructions for mitmproxy itself
    and not forwarded. They do not generate the usual HTTP handler events,
    but all requests going over the newly opened connection will.
    """

            

Reported by Pylint.

              from typing import Optional, Sequence

from mitmproxy import optmanager

CONF_DIR = "~/.mitmproxy"
CONF_BASENAME = "mitmproxy"
LISTEN_PORT = 8080
CONTENT_VIEW_LINES_CUTOFF = 512
KEY_SIZE = 2048

            

Reported by Pylint.

              KEY_SIZE = 2048


class Options(optmanager.OptManager):

    def __init__(self, **kwargs) -> None:
        super().__init__()
        self.add_option(
            "server", bool, True,

            

Reported by Pylint.

                          """
            Passphrase for decrypting the private key provided in the --cert option.

            Note that passing cert_passphrase on the command line makes your passphrase visible in your system's
            process list. Specify it in config.yaml to avoid this.
            """
        )
        self.add_option(
            "ciphers_client", Optional[str], None,

            

Reported by Pylint.