raise exceptions.ControlException(f"Reply is {self.state}, but expected it to be taken.")
        self._state = "committed"
        try:
            self._loop.call_soon_threadsafe(lambda: self.done.set())
        except RuntimeError:  # pragma: no cover
            pass  # event loop may already be closed.

    def kill(self, force=False):  # pragma: no cover
        warnings.warn("reply.kill() is deprecated, use flow.kill() or set the error attribute instead.",

            

Reported by Pylint.

                      except RuntimeError:  # pragma: no cover
            pass  # event loop may already be closed.

    def kill(self, force=False):  # pragma: no cover
        warnings.warn("reply.kill() is deprecated, use flow.kill() or set the error attribute instead.",
                      DeprecationWarning, stacklevel=2)
        self.obj.error = flow.Error(flow.Error.KILLED_MESSAGE)

    def __del__(self):

            

Reported by Pylint.

              import asyncio
import warnings
from typing import Any

from mitmproxy import exceptions, flow


class Reply:
    """

            

Reported by Pylint.

                      For example, intercepted flows are taken out so that the connection thread does not proceed.
        """
        if self.state != "start":
            raise exceptions.ControlException(f"Reply is {self.state}, but expected it to be start.")
        self._state = "taken"

    def commit(self):
        """
        Ultimately, messages are committed. This is done either automatically by

            

Reported by Pylint.

                      called .take().
        """
        if self.state != "taken":
            raise exceptions.ControlException(f"Reply is {self.state}, but expected it to be taken.")
        self._state = "committed"
        try:
            self._loop.call_soon_threadsafe(lambda: self.done.set())
        except RuntimeError:  # pragma: no cover
            pass  # event loop may already be closed.

            

Reported by Pylint.

                      except RuntimeError:  # pragma: no cover
            pass  # event loop may already be closed.

    def kill(self, force=False):  # pragma: no cover
        warnings.warn("reply.kill() is deprecated, use flow.kill() or set the error attribute instead.",
                      DeprecationWarning, stacklevel=2)
        self.obj.error = flow.Error(flow.Error.KILLED_MESSAGE)

    def __del__(self):

            

Reported by Pylint.

                          pass  # event loop may already be closed.

    def kill(self, force=False):  # pragma: no cover
        warnings.warn("reply.kill() is deprecated, use flow.kill() or set the error attribute instead.",
                      DeprecationWarning, stacklevel=2)
        self.obj.error = flow.Error(flow.Error.KILLED_MESSAGE)

    def __del__(self):
        if self.state != "committed":

            

Reported by Pylint.

                      super().__init__(None)
        self._should_reset = False

    def mark_reset(self):
        if self.state != "committed":
            raise exceptions.ControlException(f"Uncommitted reply: {self.obj}")
        self._should_reset = True

    def reset(self):

            

Reported by Pylint.

                          raise exceptions.ControlException(f"Uncommitted reply: {self.obj}")
        self._should_reset = True

    def reset(self):
        if self._should_reset:
            self._state = "start"

    def __del__(self):
        pass

            

Reported by Pylint.

              from mitmproxy.utils import sliding_window


def test_simple():
    y = list(sliding_window.window(range(1000, 1005), 1, 2))
    assert y == [
        # prev this  next  next2
        (None, 1000, 1001, 1002),
        (1000, 1001, 1002, 1003),

            

Reported by Pylint.

              from mitmproxy.utils import sliding_window


def test_simple():
    y = list(sliding_window.window(range(1000, 1005), 1, 2))
    assert y == [
        # prev this  next  next2
        (None, 1000, 1001, 1002),
        (1000, 1001, 1002, 1003),

            

Reported by Pylint.

              

def test_simple():
    y = list(sliding_window.window(range(1000, 1005), 1, 2))
    assert y == [
        # prev this  next  next2
        (None, 1000, 1001, 1002),
        (1000, 1001, 1002, 1003),
        (1001, 1002, 1003, 1004),

            

Reported by Pylint.

              
def test_simple():
    y = list(sliding_window.window(range(1000, 1005), 1, 2))
    assert y == [
        # prev this  next  next2
        (None, 1000, 1001, 1002),
        (1000, 1001, 1002, 1003),
        (1001, 1002, 1003, 1004),
        (1002, 1003, 1004, None),

            

Reported by Bandit.

                  ]


def test_is_lazy():
    done = False

    def gen():
        nonlocal done
        done = True

            

Reported by Pylint.

                      done = True
        yield 42

    x = sliding_window.window(gen(), 1, 1)
    assert not done
    assert list(x)
    assert done

            

Reported by Pylint.

                      yield 42

    x = sliding_window.window(gen(), 1, 1)
    assert not done
    assert list(x)
    assert done

            

Reported by Bandit.

              
    x = sliding_window.window(gen(), 1, 1)
    assert not done
    assert list(x)
    assert done

            

Reported by Bandit.

                  x = sliding_window.window(gen(), 1, 1)
    assert not done
    assert list(x)
    assert done

            

Reported by Bandit.

              import os

from flask import Flask, render_template

from mitmproxy.options import CONF_BASENAME, CONF_DIR

app = Flask(__name__)
# will be overridden in the addon, setting this here so that the Flask app can be run standalone.
app.config["CONFDIR"] = CONF_DIR

            

Reported by Pylint.

              

@app.route('/')
def index():
    return render_template("index.html")


@app.route('/cert/pem')
def pem():

            

Reported by Pylint.

              

@app.route('/cert/pem')
def pem():
    return read_cert("pem", "application/x-x509-ca-cert")


@app.route('/cert/p12')
def p12():

            

Reported by Pylint.

              

@app.route('/cert/p12')
def p12():
    return read_cert("p12", "application/x-pkcs12")


@app.route('/cert/cer')
def cer():

            

Reported by Pylint.

              

@app.route('/cert/cer')
def cer():
    return read_cert("cer", "application/x-x509-ca-cert")


def read_cert(ext, content_type):
    filename = CONF_BASENAME + f"-ca-cert.{ext}"

            

Reported by Pylint.

                  return read_cert("cer", "application/x-x509-ca-cert")


def read_cert(ext, content_type):
    filename = CONF_BASENAME + f"-ca-cert.{ext}"
    p = os.path.join(app.config["CONFDIR"], filename)
    p = os.path.expanduser(p)
    with open(p, "rb") as f:
        cert = f.read()

            

Reported by Pylint.

              
def read_cert(ext, content_type):
    filename = CONF_BASENAME + f"-ca-cert.{ext}"
    p = os.path.join(app.config["CONFDIR"], filename)
    p = os.path.expanduser(p)
    with open(p, "rb") as f:
        cert = f.read()

    return cert, {

            

Reported by Pylint.

              def read_cert(ext, content_type):
    filename = CONF_BASENAME + f"-ca-cert.{ext}"
    p = os.path.join(app.config["CONFDIR"], filename)
    p = os.path.expanduser(p)
    with open(p, "rb") as f:
        cert = f.read()

    return cert, {
        "Content-Type": content_type,

            

Reported by Pylint.

                  filename = CONF_BASENAME + f"-ca-cert.{ext}"
    p = os.path.join(app.config["CONFDIR"], filename)
    p = os.path.expanduser(p)
    with open(p, "rb") as f:
        cert = f.read()

    return cert, {
        "Content-Type": content_type,
        "Content-Disposition": f"inline; filename={filename}",

            

Reported by Pylint.

              from typing import Optional

from mitmproxy.net.http import url
from . import base


class ViewURLEncoded(base.View):
    name = "URL-encoded"


            

Reported by Pylint.

                      d = url.decode(data)
        return "URLEncoded form", base.format_pairs(d)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "application/x-www-form-urlencoded")

            

Reported by Pylint.

                      d = url.decode(data)
        return "URLEncoded form", base.format_pairs(d)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "application/x-www-form-urlencoded")

            

Reported by Pylint.

              from typing import Optional

from mitmproxy.net.http import url
from . import base


class ViewURLEncoded(base.View):
    name = "URL-encoded"


            

Reported by Pylint.

              from . import base


class ViewURLEncoded(base.View):
    name = "URL-encoded"

    def __call__(self, data, **metadata):
        try:
            data = data.decode("ascii", "strict")

            

Reported by Pylint.

                          data = data.decode("ascii", "strict")
        except ValueError:
            return None
        d = url.decode(data)
        return "URLEncoded form", base.format_pairs(d)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "application/x-www-form-urlencoded")

            

Reported by Pylint.

                      d = url.decode(data)
        return "URLEncoded form", base.format_pairs(d)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "application/x-www-form-urlencoded")

            

Reported by Pylint.

                      d = url.decode(data)
        return "URLEncoded form", base.format_pairs(d)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "application/x-www-form-urlencoded")

            

Reported by Pylint.

                      d = url.decode(data)
        return "URLEncoded form", base.format_pairs(d)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "application/x-www-form-urlencoded")

            

Reported by Pylint.

              from typing import Optional

from . import base
from .. import http


class ViewQuery(base.View):
    name = "Query"


            

Reported by Pylint.

              from typing import Optional

from . import base
from .. import http


class ViewQuery(base.View):
    name = "Query"


            

Reported by Pylint.

                      else:
            return "Query", base.format_text("")

    def render_priority(self, data: bytes, *, http_message: Optional[http.Message] = None, **metadata) -> float:
        return 0.3 * float(bool(
            getattr(http_message, "query", False)
            and not data
        ))

            

Reported by Pylint.

              from typing import Optional

from . import base
from .. import http


class ViewQuery(base.View):
    name = "Query"


            

Reported by Pylint.

              from .. import http


class ViewQuery(base.View):
    name = "Query"

    def __call__(self, data: bytes, http_message: Optional[http.Message] = None, **metadata):
        query = getattr(http_message, "query", None)
        if query:

            

Reported by Pylint.

              
    def __call__(self, data: bytes, http_message: Optional[http.Message] = None, **metadata):
        query = getattr(http_message, "query", None)
        if query:
            return "Query", base.format_pairs(query.items(multi=True))
        else:
            return "Query", base.format_text("")

    def render_priority(self, data: bytes, *, http_message: Optional[http.Message] = None, **metadata) -> float:

            

Reported by Pylint.

                      else:
            return "Query", base.format_text("")

    def render_priority(self, data: bytes, *, http_message: Optional[http.Message] = None, **metadata) -> float:
        return 0.3 * float(bool(
            getattr(http_message, "query", False)
            and not data
        ))

            

Reported by Pylint.

                      else:
            return "Query", base.format_text("")

    def render_priority(self, data: bytes, *, http_message: Optional[http.Message] = None, **metadata) -> float:
        return 0.3 * float(bool(
            getattr(http_message, "query", False)
            and not data
        ))

            

Reported by Pylint.

                      else:
            return "Query", base.format_text("")

    def render_priority(self, data: bytes, *, http_message: Optional[http.Message] = None, **metadata) -> float:
        return 0.3 * float(bool(
            getattr(http_message, "query", False)
            and not data
        ))

            

Reported by Pylint.

              import contextlib
from unittest import mock

import pytest

from mitmproxy.utils import arg_check


@pytest.mark.parametrize('arg, output', [

            

Reported by Pylint.

              import io
import contextlib
from unittest import mock

import pytest

from mitmproxy.utils import arg_check



            

Reported by Pylint.

                  (["--eventlog"], "--eventlog has been removed."),
    (["--nonanonymous"], '--nonanonymous is deprecated.\n'
                         'Please use `--proxyauth SPEC` instead.\n'
                         'SPEC Format: "username:pass", "any" to accept any user/pass combination,\n'
                         '"@path" to use an Apache htpasswd file, or\n'
                         '"ldap[s]:url_server_ldap:dn_auth:password:dn_subtree" '
                         'for LDAP authentication.'),
    (["--replacements"], "--replacements is deprecated.\n"
                         "Please use `--modify-body` or `--modify-headers` instead."),

            

Reported by Pylint.

                                       'for LDAP authentication.'),
    (["--replacements"], "--replacements is deprecated.\n"
                         "Please use `--modify-body` or `--modify-headers` instead."),
    (["--underscore_option"], "--underscore_option uses underscores, please use hyphens --underscore-option")
])
def test_check_args(arg, output):
    f = io.StringIO()
    with contextlib.redirect_stdout(f):
        with mock.patch('sys.argv') as m:

            

Reported by Pylint.

                  (["--replacements"], "--replacements is deprecated.\n"
                         "Please use `--modify-body` or `--modify-headers` instead."),
    (["--underscore_option"], "--underscore_option uses underscores, please use hyphens --underscore-option")
])
def test_check_args(arg, output):
    f = io.StringIO()
    with contextlib.redirect_stdout(f):
        with mock.patch('sys.argv') as m:
            m.__getitem__.return_value = arg

            

Reported by Pylint.

                  (["--underscore_option"], "--underscore_option uses underscores, please use hyphens --underscore-option")
])
def test_check_args(arg, output):
    f = io.StringIO()
    with contextlib.redirect_stdout(f):
        with mock.patch('sys.argv') as m:
            m.__getitem__.return_value = arg
            arg_check.check()
            assert f.getvalue().strip() == output

            

Reported by Pylint.

              def test_check_args(arg, output):
    f = io.StringIO()
    with contextlib.redirect_stdout(f):
        with mock.patch('sys.argv') as m:
            m.__getitem__.return_value = arg
            arg_check.check()
            assert f.getvalue().strip() == output

            

Reported by Pylint.

                      with mock.patch('sys.argv') as m:
            m.__getitem__.return_value = arg
            arg_check.check()
            assert f.getvalue().strip() == output

            

Reported by Bandit.

                  kept reasonably current to reflect common usage.
"""

# pylint: line-too-long

# A collection of (name, shortcut, string) tuples.

UASTRINGS = [
    ("android",

            

Reported by Pylint.

                   "Mozilla/5.0 (Linux; U; Android 4.1.1; en-gb; Nexus 7 Build/JRO03D) AFL/01.04.02"),  # noqa
    ("blackberry",
     "l",
     "Mozilla/5.0 (BlackBerry; U; BlackBerry 9900; en) AppleWebKit/534.11+ (KHTML, like Gecko) Version/7.1.0.346 Mobile Safari/534.11+"),  # noqa
    ("bingbot",
     "b",
     "Mozilla/5.0 (compatible; bingbot/2.0; +http://www.bing.com/bingbot.htm)"),  # noqa
    ("chrome",
     "c",

            

Reported by Pylint.

                   "Mozilla/5.0 (compatible; bingbot/2.0; +http://www.bing.com/bingbot.htm)"),  # noqa
    ("chrome",
     "c",
     "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.1 (KHTML, like Gecko) Chrome/22.0.1207.1 Safari/537.1"),  # noqa
    ("firefox",
     "f",
     "Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:14.0) Gecko/20120405 Firefox/14.0a1"),  # noqa
    ("googlebot",
     "g",

            

Reported by Pylint.

                   "Mozilla/5.0 (Windows; U; MSIE 9.0; WIndows NT 9.0; en-US)"),  # noqa
    ("ipad",
     "p",
     "Mozilla/5.0 (iPad; CPU OS 5_1 like Mac OS X) AppleWebKit/534.46 (KHTML, like Gecko) Version/5.1 Mobile/9B176 Safari/7534.48.3"),  # noqa
    ("iphone",
     "h",
     "Mozilla/5.0 (iPhone; CPU iPhone OS 4_2_1 like Mac OS X) AppleWebKit/533.17.9 (KHTML, like Gecko) Version/5.0.2 Mobile/8C148a Safari/6533.18.5"),  # noqa
    ("safari",
     "s",

            

Reported by Pylint.

                   "Mozilla/5.0 (iPad; CPU OS 5_1 like Mac OS X) AppleWebKit/534.46 (KHTML, like Gecko) Version/5.1 Mobile/9B176 Safari/7534.48.3"),  # noqa
    ("iphone",
     "h",
     "Mozilla/5.0 (iPhone; CPU iPhone OS 4_2_1 like Mac OS X) AppleWebKit/533.17.9 (KHTML, like Gecko) Version/5.0.2 Mobile/8C148a Safari/6533.18.5"),  # noqa
    ("safari",
     "s",
     "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_3) AppleWebKit/534.55.3 (KHTML, like Gecko) Version/5.1.3 Safari/534.53.10"),  # noqa
]


            

Reported by Pylint.

                   "Mozilla/5.0 (iPhone; CPU iPhone OS 4_2_1 like Mac OS X) AppleWebKit/533.17.9 (KHTML, like Gecko) Version/5.0.2 Mobile/8C148a Safari/6533.18.5"),  # noqa
    ("safari",
     "s",
     "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_3) AppleWebKit/534.55.3 (KHTML, like Gecko) Version/5.1.3 Safari/534.53.10"),  # noqa
]


def get_by_shortcut(s):
    """

            

Reported by Pylint.

              ]


def get_by_shortcut(s):
    """
        Retrieve a user agent entry by shortcut.
    """
    for i in UASTRINGS:
        if s == i[1]:

            

Reported by Pylint.

              ]


def get_by_shortcut(s):
    """
        Retrieve a user agent entry by shortcut.
    """
    for i in UASTRINGS:
        if s == i[1]:

            

Reported by Pylint.

              """
import asyncio

from mitmproxy import ctx, http


# Simple example: Inject a message as a response to an event

def websocket_message(flow: http.HTTPFlow):

            

Reported by Pylint.

              """
Inject a WebSocket message into a running connection.

This example shows how to inject a WebSocket message into a running connection.
"""
import asyncio

from mitmproxy import ctx, http


            

Reported by Pylint.

              
# Simple example: Inject a message as a response to an event

def websocket_message(flow: http.HTTPFlow):
    assert flow.websocket is not None  # make type checker happy
    last_message = flow.websocket.messages[-1]
    if last_message.is_text and "secret" in last_message.text:
        last_message.drop()
        ctx.master.commands.call("inject.websocket", flow, last_message.from_client, "ssssssh".encode())

            

Reported by Pylint.

              # Simple example: Inject a message as a response to an event

def websocket_message(flow: http.HTTPFlow):
    assert flow.websocket is not None  # make type checker happy
    last_message = flow.websocket.messages[-1]
    if last_message.is_text and "secret" in last_message.text:
        last_message.drop()
        ctx.master.commands.call("inject.websocket", flow, last_message.from_client, "ssssssh".encode())


            

Reported by Bandit.

                  last_message = flow.websocket.messages[-1]
    if last_message.is_text and "secret" in last_message.text:
        last_message.drop()
        ctx.master.commands.call("inject.websocket", flow, last_message.from_client, "ssssssh".encode())


# Complex example: Schedule a periodic timer

async def inject_async(flow: http.HTTPFlow):

            

Reported by Pylint.

              
# Complex example: Schedule a periodic timer

async def inject_async(flow: http.HTTPFlow):
    msg = "hello from mitmproxy! "
    assert flow.websocket is not None  # make type checker happy
    while flow.websocket.timestamp_end is None:
        ctx.master.commands.call("inject.websocket", flow, True, msg.encode())
        await asyncio.sleep(1)

            

Reported by Pylint.

              
async def inject_async(flow: http.HTTPFlow):
    msg = "hello from mitmproxy! "
    assert flow.websocket is not None  # make type checker happy
    while flow.websocket.timestamp_end is None:
        ctx.master.commands.call("inject.websocket", flow, True, msg.encode())
        await asyncio.sleep(1)
        msg = msg[1:] + msg[:1]


            

Reported by Bandit.

                      msg = msg[1:] + msg[:1]


def websocket_start(flow: http.HTTPFlow):
    asyncio.create_task(inject_async(flow))

            

Reported by Pylint.

              from mitmproxy.contentviews import base
from mitmproxy.utils import strutils

"""
A custom CSS prettifier. Compared to other prettifiers, its main features are:

- Implemented in pure Python.
- Modifies whitespace only.
- Works with any input.

            

Reported by Pylint.

              CSS_SPECIAL_CHARS = "{};:"


def beautify(data: str, indent: str = "    "):
    """Beautify a string containing CSS code"""
    data = strutils.escape_special_areas(
        data.strip(),
        CSS_SPECIAL_AREAS,
        CSS_SPECIAL_CHARS,

            

Reported by Pylint.

              class ViewCSS(base.View):
    name = "CSS"

    def __call__(self, data, **metadata):
        data = data.decode("utf8", "surrogateescape")
        beautified = beautify(data)
        return "CSS", base.format_text(beautified)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:

            

Reported by Pylint.

                      beautified = beautify(data)
        return "CSS", base.format_text(beautified)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "text/css")


if __name__ == "__main__":  # pragma: no cover
    with open("../tools/web/static/vendor.css") as f:

            

Reported by Pylint.

                      beautified = beautify(data)
        return "CSS", base.format_text(beautified)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "text/css")


if __name__ == "__main__":  # pragma: no cover
    with open("../tools/web/static/vendor.css") as f:

            

Reported by Pylint.

              import re
import time
from typing import Optional

from mitmproxy.contentviews import base
from mitmproxy.utils import strutils

"""
A custom CSS prettifier. Compared to other prettifiers, its main features are:

            

Reported by Pylint.

                  return data.rstrip("\n") + "\n"


class ViewCSS(base.View):
    name = "CSS"

    def __call__(self, data, **metadata):
        data = data.decode("utf8", "surrogateescape")
        beautified = beautify(data)

            

Reported by Pylint.

                      beautified = beautify(data)
        return "CSS", base.format_text(beautified)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "text/css")


if __name__ == "__main__":  # pragma: no cover
    with open("../tools/web/static/vendor.css") as f:

            

Reported by Pylint.

              from mitmproxy.proxy import context
from mitmproxy.test import tflow, taddons


def test_context():
    with taddons.context() as tctx:
        c = context.Context(
            tflow.tclient_conn(),
            tctx.options

            

Reported by Pylint.

              from mitmproxy.test import tflow, taddons


def test_context():
    with taddons.context() as tctx:
        c = context.Context(
            tflow.tclient_conn(),
            tctx.options
        )

            

Reported by Pylint.

              
def test_context():
    with taddons.context() as tctx:
        c = context.Context(
            tflow.tclient_conn(),
            tctx.options
        )
        assert repr(c)
        c.layers.append(1)

            

Reported by Pylint.

                          tflow.tclient_conn(),
            tctx.options
        )
        assert repr(c)
        c.layers.append(1)
        assert repr(c)
        c2 = c.fork()
        c.layers.append(2)
        c2.layers.append(3)

            

Reported by Bandit.

                      )
        assert repr(c)
        c.layers.append(1)
        assert repr(c)
        c2 = c.fork()
        c.layers.append(2)
        c2.layers.append(3)
        assert c.layers == [1, 2]
        assert c2.layers == [1, 3]

            

Reported by Bandit.

                      assert repr(c)
        c.layers.append(1)
        assert repr(c)
        c2 = c.fork()
        c.layers.append(2)
        c2.layers.append(3)
        assert c.layers == [1, 2]
        assert c2.layers == [1, 3]

            

Reported by Pylint.

                      c2 = c.fork()
        c.layers.append(2)
        c2.layers.append(3)
        assert c.layers == [1, 2]
        assert c2.layers == [1, 3]

            

Reported by Bandit.

                      c.layers.append(2)
        c2.layers.append(3)
        assert c.layers == [1, 2]
        assert c2.layers == [1, 3]

            

Reported by Bandit.