import time
from mitmproxy.script import concurrent


class ConcurrentClass:

    @concurrent
    def request(self, flow):
        time.sleep(0.1)

            

Reported by Pylint.

              class ConcurrentClass:

    @concurrent
    def request(self, flow):
        time.sleep(0.1)


addons = [ConcurrentClass()]

            

Reported by Pylint.

              import time
from mitmproxy.script import concurrent


class ConcurrentClass:

    @concurrent
    def request(self, flow):
        time.sleep(0.1)

            

Reported by Pylint.

              from mitmproxy.script import concurrent


class ConcurrentClass:

    @concurrent
    def request(self, flow):
        time.sleep(0.1)


            

Reported by Pylint.

              from mitmproxy.script import concurrent


class ConcurrentClass:

    @concurrent
    def request(self, flow):
        time.sleep(0.1)


            

Reported by Pylint.

              class ConcurrentClass:

    @concurrent
    def request(self, flow):
        time.sleep(0.1)


addons = [ConcurrentClass()]

            

Reported by Pylint.

              class ConcurrentClass:

    @concurrent
    def request(self, flow):
        time.sleep(0.1)


addons = [ConcurrentClass()]

            

Reported by Pylint.

              import pytest

from mitmproxy.net import server_spec


@pytest.mark.parametrize("spec,out", [
    ("example.com", ("https", ("example.com", 443))),
    ("http://example.com", ("http", ("example.com", 80))),
    ("smtp.example.com:25", ("http", ("smtp.example.com", 25))),

            

Reported by Pylint.

              import pytest

from mitmproxy.net import server_spec


@pytest.mark.parametrize("spec,out", [
    ("example.com", ("https", ("example.com", 443))),
    ("http://example.com", ("http", ("example.com", 80))),
    ("smtp.example.com:25", ("http", ("smtp.example.com", 25))),

            

Reported by Pylint.

                  ("http://[::1]/", ("http", ("::1", 80))),
    ("https://[::1]/", ("https", ("::1", 443))),
    ("http://[::1]:8080", ("http", ("::1", 8080))),
])
def test_parse(spec, out):
    assert server_spec.parse(spec) == out


def test_parse_err():

            

Reported by Pylint.

                  ("http://[::1]:8080", ("http", ("::1", 8080))),
])
def test_parse(spec, out):
    assert server_spec.parse(spec) == out


def test_parse_err():
    with pytest.raises(ValueError, match="Invalid server specification"):
        server_spec.parse(":")

            

Reported by Bandit.

                  assert server_spec.parse(spec) == out


def test_parse_err():
    with pytest.raises(ValueError, match="Invalid server specification"):
        server_spec.parse(":")

    with pytest.raises(ValueError, match="Invalid server scheme"):
        server_spec.parse("ftp://example.com")

            

Reported by Pylint.

                      server_spec.parse("example.com:999999")


def test_parse_with_mode():
    assert server_spec.parse_with_mode("m:example.com") == ("m", ("https", ("example.com", 443)))
    with pytest.raises(ValueError):
        server_spec.parse_with_mode("moo")

            

Reported by Pylint.

              

def test_parse_with_mode():
    assert server_spec.parse_with_mode("m:example.com") == ("m", ("https", ("example.com", 443)))
    with pytest.raises(ValueError):
        server_spec.parse_with_mode("moo")

            

Reported by Bandit.

              from mitmproxy.tools.console import palettes


class TestPalette:

    def test_helptext(self):
        for i in palettes.palettes.values():
            assert i.palette(False)
        for i in palettes.palettes.values():

            

Reported by Pylint.

              from mitmproxy.tools.console import palettes


class TestPalette:

    def test_helptext(self):
        for i in palettes.palettes.values():
            assert i.palette(False)
        for i in palettes.palettes.values():

            

Reported by Pylint.

              from mitmproxy.tools.console import palettes


class TestPalette:

    def test_helptext(self):
        for i in palettes.palettes.values():
            assert i.palette(False)
        for i in palettes.palettes.values():

            

Reported by Pylint.

              
class TestPalette:

    def test_helptext(self):
        for i in palettes.palettes.values():
            assert i.palette(False)
        for i in palettes.palettes.values():
            assert i.palette(True)

            

Reported by Pylint.

              
class TestPalette:

    def test_helptext(self):
        for i in palettes.palettes.values():
            assert i.palette(False)
        for i in palettes.palettes.values():
            assert i.palette(True)

            

Reported by Pylint.

              
    def test_helptext(self):
        for i in palettes.palettes.values():
            assert i.palette(False)
        for i in palettes.palettes.values():
            assert i.palette(True)

            

Reported by Bandit.

                      for i in palettes.palettes.values():
            assert i.palette(False)
        for i in palettes.palettes.values():
            assert i.palette(True)

            

Reported by Bandit.

              from mitmproxy import http
import typing

# This scripts demonstrates how mitmproxy can switch to a second/different upstream proxy
# in upstream proxy mode.
#
# Usage: mitmdump -U http://default-upstream-proxy.local:8080/ -s change_upstream_proxy.py
#
# If you want to change the target server, you should modify flow.request.host and flow.request.port

            

Reported by Pylint.

              from mitmproxy import http
import typing

# This scripts demonstrates how mitmproxy can switch to a second/different upstream proxy
# in upstream proxy mode.
#
# Usage: mitmdump -U http://default-upstream-proxy.local:8080/ -s change_upstream_proxy.py
#
# If you want to change the target server, you should modify flow.request.host and flow.request.port

            

Reported by Pylint.

              from mitmproxy import http
import typing

# This scripts demonstrates how mitmproxy can switch to a second/different upstream proxy
# in upstream proxy mode.
#
# Usage: mitmdump -U http://default-upstream-proxy.local:8080/ -s change_upstream_proxy.py
#
# If you want to change the target server, you should modify flow.request.host and flow.request.port

            

Reported by Pylint.

              # If you want to change the target server, you should modify flow.request.host and flow.request.port


def proxy_address(flow: http.HTTPFlow) -> typing.Tuple[str, int]:
    # Poor man's loadbalancing: route every second domain through the alternative proxy.
    if hash(flow.request.host) % 2 == 1:
        return ("localhost", 8082)
    else:
        return ("localhost", 8081)

            

Reported by Pylint.

              
def proxy_address(flow: http.HTTPFlow) -> typing.Tuple[str, int]:
    # Poor man's loadbalancing: route every second domain through the alternative proxy.
    if hash(flow.request.host) % 2 == 1:
        return ("localhost", 8082)
    else:
        return ("localhost", 8081)



            

Reported by Pylint.

                      return ("localhost", 8081)


def request(flow: http.HTTPFlow) -> None:
    if flow.request.method == "CONNECT":
        # If the decision is done by domain, one could also modify the server address here.
        # We do it after CONNECT here to have the request data available as well.
        return
    address = proxy_address(flow)

            

Reported by Pylint.

              from mitmproxy import ctx


class AntiCache:
    def load(self, loader):
        loader.add_option(
            "anticache", bool, False,
            """
            Strip out request headers that might cause the server to return

            

Reported by Pylint.

              from mitmproxy import ctx


class AntiCache:
    def load(self, loader):
        loader.add_option(
            "anticache", bool, False,
            """
            Strip out request headers that might cause the server to return

            

Reported by Pylint.

              

class AntiCache:
    def load(self, loader):
        loader.add_option(
            "anticache", bool, False,
            """
            Strip out request headers that might cause the server to return
            304-not-modified.

            

Reported by Pylint.

              

class AntiCache:
    def load(self, loader):
        loader.add_option(
            "anticache", bool, False,
            """
            Strip out request headers that might cause the server to return
            304-not-modified.

            

Reported by Pylint.

                          """
        )

    def request(self, flow):
        if ctx.options.anticache:
            flow.request.anticache()

            

Reported by Pylint.

                          """
        )

    def request(self, flow):
        if ctx.options.anticache:
            flow.request.anticache()

            

Reported by Pylint.

              from mitmproxy import ctx


class AntiComp:
    def load(self, loader):
        loader.add_option(
            "anticomp", bool, False,
            "Try to convince servers to send us un-compressed data."
        )

            

Reported by Pylint.

              from mitmproxy import ctx


class AntiComp:
    def load(self, loader):
        loader.add_option(
            "anticomp", bool, False,
            "Try to convince servers to send us un-compressed data."
        )

            

Reported by Pylint.

              

class AntiComp:
    def load(self, loader):
        loader.add_option(
            "anticomp", bool, False,
            "Try to convince servers to send us un-compressed data."
        )


            

Reported by Pylint.

              

class AntiComp:
    def load(self, loader):
        loader.add_option(
            "anticomp", bool, False,
            "Try to convince servers to send us un-compressed data."
        )


            

Reported by Pylint.

                          "Try to convince servers to send us un-compressed data."
        )

    def request(self, flow):
        if ctx.options.anticomp:
            flow.request.anticomp()

            

Reported by Pylint.

                          "Try to convince servers to send us un-compressed data."
        )

    def request(self, flow):
        if ctx.options.anticomp:
            flow.request.anticomp()

            

Reported by Pylint.

              
class Comment:
    @command.command("flow.comment")
    def comment(self, flow: typing.Sequence[flow.Flow], comment: str) -> None:
        "Add a comment to a flow"

        updated = []
        for f in flow:
            f.comment = comment

            

Reported by Pylint.

              import typing
from mitmproxy import command, flow, ctx
from mitmproxy.hooks import UpdateHook


class Comment:
    @command.command("flow.comment")
    def comment(self, flow: typing.Sequence[flow.Flow], comment: str) -> None:
        "Add a comment to a flow"

            

Reported by Pylint.

              from mitmproxy.hooks import UpdateHook


class Comment:
    @command.command("flow.comment")
    def comment(self, flow: typing.Sequence[flow.Flow], comment: str) -> None:
        "Add a comment to a flow"

        updated = []

            

Reported by Pylint.

              from mitmproxy.hooks import UpdateHook


class Comment:
    @command.command("flow.comment")
    def comment(self, flow: typing.Sequence[flow.Flow], comment: str) -> None:
        "Add a comment to a flow"

        updated = []

            

Reported by Pylint.

              
class Comment:
    @command.command("flow.comment")
    def comment(self, flow: typing.Sequence[flow.Flow], comment: str) -> None:
        "Add a comment to a flow"

        updated = []
        for f in flow:
            f.comment = comment

            

Reported by Pylint.

                      "Add a comment to a flow"

        updated = []
        for f in flow:
            f.comment = comment
            updated.append(f)

        ctx.master.addons.trigger(UpdateHook(updated))

            

Reported by Pylint.

              """Add a custom command to mitmproxy's command prompt."""
from mitmproxy import command
from mitmproxy import ctx


class MyAddon:
    def __init__(self):
        self.num = 0


            

Reported by Pylint.

              """Add a custom command to mitmproxy's command prompt."""
from mitmproxy import command
from mitmproxy import ctx


class MyAddon:
    def __init__(self):
        self.num = 0


            

Reported by Pylint.

              """Add a custom command to mitmproxy's command prompt."""
from mitmproxy import command
from mitmproxy import ctx


class MyAddon:
    def __init__(self):
        self.num = 0


            

Reported by Pylint.

              from mitmproxy import ctx


class MyAddon:
    def __init__(self):
        self.num = 0

    @command.command("myaddon.inc")
    def inc(self) -> None:

            

Reported by Pylint.

              from mitmproxy import ctx


class MyAddon:
    def __init__(self):
        self.num = 0

    @command.command("myaddon.inc")
    def inc(self) -> None:

            

Reported by Pylint.

                      self.num = 0

    @command.command("myaddon.inc")
    def inc(self) -> None:
        self.num += 1
        ctx.log.info(f"num = {self.num}")


addons = [

            

Reported by Pylint.

              import typing

from mitmproxy import exceptions
from mitmproxy import flowfilter
from mitmproxy import ctx


class StickyAuth:
    def __init__(self):

            

Reported by Pylint.

              from mitmproxy import ctx


class StickyAuth:
    def __init__(self):
        self.flt = None
        self.hosts = {}

    def load(self, loader):

            

Reported by Pylint.

                      self.flt = None
        self.hosts = {}

    def load(self, loader):
        loader.add_option(
            "stickyauth", typing.Optional[str], None,
            "Set sticky auth filter. Matched against requests."
        )


            

Reported by Pylint.

                      self.flt = None
        self.hosts = {}

    def load(self, loader):
        loader.add_option(
            "stickyauth", typing.Optional[str], None,
            "Set sticky auth filter. Matched against requests."
        )


            

Reported by Pylint.

                          "Set sticky auth filter. Matched against requests."
        )

    def configure(self, updated):
        if "stickyauth" in updated:
            if ctx.options.stickyauth:
                flt = flowfilter.parse(ctx.options.stickyauth)
                if not flt:
                    raise exceptions.OptionsError(

            

Reported by Pylint.

                          else:
                self.flt = None

    def request(self, flow):
        if self.flt:
            host = flow.request.host
            if "authorization" in flow.request.headers:
                self.hosts[host] = flow.request.headers["authorization"]
            elif flowfilter.match(self.flt, flow):

            

Reported by Pylint.

                  return ts[0].lower(), ts[1].lower(), d


def assemble_content_type(type, subtype, parameters):
    if not parameters:
        return f"{type}/{subtype}"
    params = "; ".join(
        f"{k}={v}"
        for k, v in parameters.items()

            

Reported by Pylint.

              import collections
from typing import Dict, Optional, Tuple


def parse_content_type(c: str) -> Optional[Tuple[str, str, Dict[str, str]]]:
    """
        A simple parser for content-type values. Returns a (type, subtype,
        parameters) tuple, where type and subtype are strings, and parameters
        is a dict. If the string could not be parsed, return None.

            

Reported by Pylint.

              from typing import Dict, Optional, Tuple


def parse_content_type(c: str) -> Optional[Tuple[str, str, Dict[str, str]]]:
    """
        A simple parser for content-type values. Returns a (type, subtype,
        parameters) tuple, where type and subtype are strings, and parameters
        is a dict. If the string could not be parsed, return None.


            

Reported by Pylint.

                          ("text", "html", {"charset": "UTF-8"})
    """
    parts = c.split(";", 1)
    ts = parts[0].split("/", 1)
    if len(ts) != 2:
        return None
    d = collections.OrderedDict()
    if len(parts) == 2:
        for i in parts[1].split(";"):

            

Reported by Pylint.

                  ts = parts[0].split("/", 1)
    if len(ts) != 2:
        return None
    d = collections.OrderedDict()
    if len(parts) == 2:
        for i in parts[1].split(";"):
            clause = i.split("=", 1)
            if len(clause) == 2:
                d[clause[0].strip()] = clause[1].strip()

            

Reported by Pylint.

                  return ts[0].lower(), ts[1].lower(), d


def assemble_content_type(type, subtype, parameters):
    if not parameters:
        return f"{type}/{subtype}"
    params = "; ".join(
        f"{k}={v}"
        for k, v in parameters.items()

            

Reported by Pylint.