from unittest import mock

import pytest

from mitmproxy import addonmanager
from mitmproxy import addons
from mitmproxy import command
from mitmproxy import exceptions
from mitmproxy import hooks

            

Reported by Pylint.

              

class TAddon:
    def __init__(self, name, addons=None):
        self.name = name
        self.response = True
        self.running_called = False
        if addons:
            self.addons = addons

            

Reported by Pylint.

                  f = tflow.tflow()
    await a.handle_lifecycle(HttpRequestHook(f))

    a._configure_all(o, o.keys())


def test_defaults():
    assert addons.default_addons()


            

Reported by Pylint.

                          l.add_option("custom_option", bool, True, "help")
            assert warn.called

            def cmd(a: str) -> str:
                return "foo"

            l.add_command("test.command", cmd)



            

Reported by Pylint.

                  m = master.Master(o)
    a = addonmanager.AddonManager(m)
    a.add(AOption())
    assert "custom_option" in m.options._options


def test_nesting():
    o = options.Options()
    m = master.Master(o)

            

Reported by Pylint.

              from unittest import mock

import pytest

from mitmproxy import addonmanager
from mitmproxy import addons
from mitmproxy import command
from mitmproxy import exceptions
from mitmproxy import hooks

            

Reported by Pylint.

              from mitmproxy.test import tflow


class TAddon:
    def __init__(self, name, addons=None):
        self.name = name
        self.response = True
        self.running_called = False
        if addons:

            

Reported by Pylint.

                          self.addons = addons

    @command.command("test.command")
    def testcommand(self) -> str:
        return "here"

    def __repr__(self):
        return "Addon(%s)" % self.name


            

Reported by Pylint.

                          self.addons = addons

    @command.command("test.command")
    def testcommand(self) -> str:
        return "here"

    def __repr__(self):
        return "Addon(%s)" % self.name


            

Reported by Pylint.

                  def __repr__(self):
        return "Addon(%s)" % self.name

    def done(self):
        pass

    def running(self):
        self.running_called = True


            

Reported by Pylint.

              from pathlib import Path
from typing import Union

import pytest

from OpenSSL import SSL
from mitmproxy import certs, connection
from mitmproxy.addons import tlsconfig
from mitmproxy.proxy import context

            

Reported by Pylint.

                  conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=b"qux"))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == b"qux"
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=b""))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == SSL.NO_OVERLAPPING_PROTOCOLS

    # Test that we try to mirror the server connection's ALPN
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=None))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == b"h2"


            

Reported by Pylint.

                  assert tlsconfig.alpn_select_callback(conn, [b"qux", b"h2", b"http/1.1"]) == b"h2"

    # Test no overlap
    assert tlsconfig.alpn_select_callback(conn, [b"qux", b"quux"]) == SSL.NO_OVERLAPPING_PROTOCOLS

    # Test that we don't select an ALPN if the server refused to select one.
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"", http2=True, client_alpn=None))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1"]) == SSL.NO_OVERLAPPING_PROTOCOLS


            

Reported by Pylint.

              
    # Test that we don't select an ALPN if the server refused to select one.
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"", http2=True, client_alpn=None))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1"]) == SSL.NO_OVERLAPPING_PROTOCOLS


here = Path(__file__).parent



            

Reported by Pylint.

              import ssl
import time
from pathlib import Path
from typing import Union

import pytest

from OpenSSL import SSL
from mitmproxy import certs, connection

            

Reported by Pylint.

              from mitmproxy.proxy import context
from mitmproxy.proxy.layers import modes, tls
from mitmproxy.test import taddons
from test.mitmproxy.proxy.layers import test_tls


def test_alpn_select_callback():
    ctx = SSL.Context(SSL.SSLv23_METHOD)
    conn = SSL.Connection(ctx)

            

Reported by Pylint.

              from test.mitmproxy.proxy.layers import test_tls


def test_alpn_select_callback():
    ctx = SSL.Context(SSL.SSLv23_METHOD)
    conn = SSL.Connection(ctx)

    # Test that we respect addons setting `client.alpn`.
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=b"qux"))

            

Reported by Pylint.

              
    # Test that we respect addons setting `client.alpn`.
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=b"qux"))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == b"qux"
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=b""))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == SSL.NO_OVERLAPPING_PROTOCOLS

    # Test that we try to mirror the server connection's ALPN
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=None))

            

Reported by Bandit.

                  conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=b"qux"))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == b"qux"
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=b""))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == SSL.NO_OVERLAPPING_PROTOCOLS

    # Test that we try to mirror the server connection's ALPN
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=None))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == b"h2"


            

Reported by Pylint.

                  conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=b"qux"))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == b"qux"
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=b""))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == SSL.NO_OVERLAPPING_PROTOCOLS

    # Test that we try to mirror the server connection's ALPN
    conn.set_app_data(tlsconfig.AppData(server_alpn=b"h2", http2=True, client_alpn=None))
    assert tlsconfig.alpn_select_callback(conn, [b"http/1.1", b"qux", b"h2"]) == b"h2"


            

Reported by Bandit.

              from unittest.mock import MagicMock

import pytest

from mitmproxy import connection
from mitmproxy.addons.next_layer import NextLayer
from mitmproxy.proxy.layers.http import HTTPMode
from mitmproxy.proxy import context, layers
from mitmproxy.test import taddons

            

Reported by Pylint.

              
    def test_configure(self):
        nl = NextLayer()
        with taddons.context(nl) as tctx:
            with pytest.raises(Exception, match="mutually exclusive"):
                tctx.configure(nl, allow_hosts=["example.org"], ignore_hosts=["example.com"])

    def test_ignore_connection(self):
        nl = NextLayer()

            

Reported by Pylint.

              
    def test_ignore_connection(self):
        nl = NextLayer()
        with taddons.context(nl) as tctx:
            assert not nl.ignore_connection(("example.com", 443), b"")

            tctx.configure(nl, ignore_hosts=["example.com"])
            assert nl.ignore_connection(("example.com", 443), b"")
            assert nl.ignore_connection(("example.com", 1234), b"")

            

Reported by Pylint.

                  def test_make_top_layer(self):
        nl = NextLayer()
        ctx = MagicMock()
        with taddons.context(nl) as tctx:
            tctx.configure(nl, mode="regular")
            assert isinstance(nl.make_top_layer(ctx), layers.modes.HttpProxy)

            tctx.configure(nl, mode="transparent")
            assert isinstance(nl.make_top_layer(ctx), layers.modes.TransparentProxy)

            

Reported by Pylint.

                      ctx = MagicMock()
        ctx.client.alpn = None
        ctx.server.address = ("example.com", 443)
        with taddons.context(nl) as tctx:
            ctx.layers = []
            assert isinstance(nl._next_layer(ctx, b"", b""), layers.modes.HttpProxy)

            assert nl._next_layer(ctx, b"", b"") is None


            

Reported by Pylint.

                      ctx.server.address = ("example.com", 443)
        with taddons.context(nl) as tctx:
            ctx.layers = []
            assert isinstance(nl._next_layer(ctx, b"", b""), layers.modes.HttpProxy)

            assert nl._next_layer(ctx, b"", b"") is None

            tctx.configure(nl, ignore_hosts=["example.com"])
            assert isinstance(nl._next_layer(ctx, b"123", b""), layers.TCPLayer)

            

Reported by Pylint.

                          ctx.layers = []
            assert isinstance(nl._next_layer(ctx, b"", b""), layers.modes.HttpProxy)

            assert nl._next_layer(ctx, b"", b"") is None

            tctx.configure(nl, ignore_hosts=["example.com"])
            assert isinstance(nl._next_layer(ctx, b"123", b""), layers.TCPLayer)
            assert nl._next_layer(ctx, client_hello_no_extensions[:10], b"") is None


            

Reported by Pylint.

                          assert nl._next_layer(ctx, b"", b"") is None

            tctx.configure(nl, ignore_hosts=["example.com"])
            assert isinstance(nl._next_layer(ctx, b"123", b""), layers.TCPLayer)
            assert nl._next_layer(ctx, client_hello_no_extensions[:10], b"") is None

            tctx.configure(nl, ignore_hosts=[])
            assert isinstance(nl._next_layer(ctx, client_hello_no_extensions, b""), layers.ServerTLSLayer)
            assert isinstance(ctx.layers[-1], layers.ClientTLSLayer)

            

Reported by Pylint.

              
            tctx.configure(nl, ignore_hosts=["example.com"])
            assert isinstance(nl._next_layer(ctx, b"123", b""), layers.TCPLayer)
            assert nl._next_layer(ctx, client_hello_no_extensions[:10], b"") is None

            tctx.configure(nl, ignore_hosts=[])
            assert isinstance(nl._next_layer(ctx, client_hello_no_extensions, b""), layers.ServerTLSLayer)
            assert isinstance(ctx.layers[-1], layers.ClientTLSLayer)


            

Reported by Pylint.

                          assert nl._next_layer(ctx, client_hello_no_extensions[:10], b"") is None

            tctx.configure(nl, ignore_hosts=[])
            assert isinstance(nl._next_layer(ctx, client_hello_no_extensions, b""), layers.ServerTLSLayer)
            assert isinstance(ctx.layers[-1], layers.ClientTLSLayer)

            ctx.layers = []
            assert isinstance(nl._next_layer(ctx, b"", b""), layers.modes.HttpProxy)
            assert isinstance(nl._next_layer(ctx, client_hello_no_extensions, b""), layers.ClientTLSLayer)

            

Reported by Pylint.

              from mitmproxy.test import taddons
from mitmproxy.test import tflow
from mitmproxy import exceptions
import pytest


def test_set():
    sa = core.Core()
    with taddons.context(loadcore=False) as tctx:

            

Reported by Pylint.

              

@mock.patch("mitmproxy.platform.original_addr")
def test_validation_modes(m):
    sa = core.Core()
    with taddons.context() as tctx:
        tctx.configure(sa, mode = "reverse:http://localhost")
        with pytest.raises(Exception, match="Invalid server specification"):
            tctx.configure(sa, mode = "reverse:")

            

Reported by Pylint.

              from unittest import mock

from mitmproxy.addons import core
from mitmproxy.test import taddons
from mitmproxy.test import tflow
from mitmproxy import exceptions
import pytest



            

Reported by Pylint.

              import pytest


def test_set():
    sa = core.Core()
    with taddons.context(loadcore=False) as tctx:
        assert tctx.master.options.server
        tctx.command(sa.set, "server", "false")
        assert not tctx.master.options.server

            

Reported by Pylint.

              

def test_set():
    sa = core.Core()
    with taddons.context(loadcore=False) as tctx:
        assert tctx.master.options.server
        tctx.command(sa.set, "server", "false")
        assert not tctx.master.options.server


            

Reported by Pylint.

              def test_set():
    sa = core.Core()
    with taddons.context(loadcore=False) as tctx:
        assert tctx.master.options.server
        tctx.command(sa.set, "server", "false")
        assert not tctx.master.options.server

        with pytest.raises(exceptions.CommandError):
            tctx.command(sa.set, "nonexistent")

            

Reported by Bandit.

                  with taddons.context(loadcore=False) as tctx:
        assert tctx.master.options.server
        tctx.command(sa.set, "server", "false")
        assert not tctx.master.options.server

        with pytest.raises(exceptions.CommandError):
            tctx.command(sa.set, "nonexistent")



            

Reported by Bandit.

                          tctx.command(sa.set, "nonexistent")


def test_resume():
    sa = core.Core()
    with taddons.context(loadcore=False):
        f = tflow.tflow()
        assert not sa.resume([f])
        f.intercept()

            

Reported by Pylint.

              

def test_resume():
    sa = core.Core()
    with taddons.context(loadcore=False):
        f = tflow.tflow()
        assert not sa.resume([f])
        f.intercept()
        sa.resume([f])

            

Reported by Pylint.

              def test_resume():
    sa = core.Core()
    with taddons.context(loadcore=False):
        f = tflow.tflow()
        assert not sa.resume([f])
        f.intercept()
        sa.resume([f])
        assert not f.reply.state == "taken"


            

Reported by Pylint.

              import io

import pytest

import mitmproxy.io
from mitmproxy import flow
from mitmproxy import flowfilter
from mitmproxy import options
from mitmproxy.exceptions import FlowReadException

            

Reported by Pylint.

              import io

import pytest

import mitmproxy.io
from mitmproxy import flow
from mitmproxy import flowfilter
from mitmproxy import options
from mitmproxy.exceptions import FlowReadException

            

Reported by Pylint.

              from mitmproxy.test import taddons, tflow


class State:
    def __init__(self):
        self.flows = []

    def request(self, f):
        if f not in self.flows:

            

Reported by Pylint.

                  def __init__(self):
        self.flows = []

    def request(self, f):
        if f not in self.flows:
            self.flows.append(f)

    def response(self, f):
        if f not in self.flows:

            

Reported by Pylint.

                  def __init__(self):
        self.flows = []

    def request(self, f):
        if f not in self.flows:
            self.flows.append(f)

    def response(self, f):
        if f not in self.flows:

            

Reported by Pylint.

                      if f not in self.flows:
            self.flows.append(f)

    def response(self, f):
        if f not in self.flows:
            self.flows.append(f)

    def websocket_start(self, f):
        if f not in self.flows:

            

Reported by Pylint.

                      if f not in self.flows:
            self.flows.append(f)

    def response(self, f):
        if f not in self.flows:
            self.flows.append(f)

    def websocket_start(self, f):
        if f not in self.flows:

            

Reported by Pylint.

                      if f not in self.flows:
            self.flows.append(f)

    def websocket_start(self, f):
        if f not in self.flows:
            self.flows.append(f)


class TestSerialize:

            

Reported by Pylint.

                      if f not in self.flows:
            self.flows.append(f)

    def websocket_start(self, f):
        if f not in self.flows:
            self.flows.append(f)


class TestSerialize:

            

Reported by Pylint.

                          self.flows.append(f)


class TestSerialize:

    def test_roundtrip(self):
        sio = io.BytesIO()
        f = tflow.tflow()
        f.marked = ":default:"

            

Reported by Pylint.

              from enum import Enum
from pkg_resources import parse_version

from kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO


if parse_version(ks_version) < parse_version('0.7'):
    raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version))


            

Reported by Pylint.

                      @property
        def has_color_table(self):
            if hasattr(self, '_m_has_color_table'):
                return self._m_has_color_table if hasattr(self, '_m_has_color_table') else None

            self._m_has_color_table = (self.flags & 128) != 0
            return self._m_has_color_table if hasattr(self, '_m_has_color_table') else None

        @property

            

Reported by Pylint.

                      @property
        def color_table_size(self):
            if hasattr(self, '_m_color_table_size'):
                return self._m_color_table_size if hasattr(self, '_m_color_table_size') else None

            self._m_color_table_size = (2 << (self.flags & 7))
            return self._m_color_table_size if hasattr(self, '_m_color_table_size') else None



            

Reported by Pylint.

                      @property
        def has_color_table(self):
            if hasattr(self, '_m_has_color_table'):
                return self._m_has_color_table if hasattr(self, '_m_has_color_table') else None

            self._m_has_color_table = (self.flags & 128) != 0
            return self._m_has_color_table if hasattr(self, '_m_has_color_table') else None

        @property

            

Reported by Pylint.

                      @property
        def has_interlace(self):
            if hasattr(self, '_m_has_interlace'):
                return self._m_has_interlace if hasattr(self, '_m_has_interlace') else None

            self._m_has_interlace = (self.flags & 64) != 0
            return self._m_has_interlace if hasattr(self, '_m_has_interlace') else None

        @property

            

Reported by Pylint.

                      @property
        def has_sorted_color_table(self):
            if hasattr(self, '_m_has_sorted_color_table'):
                return self._m_has_sorted_color_table if hasattr(self, '_m_has_sorted_color_table') else None

            self._m_has_sorted_color_table = (self.flags & 32) != 0
            return self._m_has_sorted_color_table if hasattr(self, '_m_has_sorted_color_table') else None

        @property

            

Reported by Pylint.

                      @property
        def color_table_size(self):
            if hasattr(self, '_m_color_table_size'):
                return self._m_color_table_size if hasattr(self, '_m_color_table_size') else None

            self._m_color_table_size = (2 << (self.flags & 7))
            return self._m_color_table_size if hasattr(self, '_m_color_table_size') else None



            

Reported by Pylint.

                      @property
        def transparent_color_flag(self):
            if hasattr(self, '_m_transparent_color_flag'):
                return self._m_transparent_color_flag if hasattr(self, '_m_transparent_color_flag') else None

            self._m_transparent_color_flag = (self.flags & 1) != 0
            return self._m_transparent_color_flag if hasattr(self, '_m_transparent_color_flag') else None

        @property

            

Reported by Pylint.

                      @property
        def user_input_flag(self):
            if hasattr(self, '_m_user_input_flag'):
                return self._m_user_input_flag if hasattr(self, '_m_user_input_flag') else None

            self._m_user_input_flag = (self.flags & 2) != 0
            return self._m_user_input_flag if hasattr(self, '_m_user_input_flag') else None



            

Reported by Pylint.

              # This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild

import array
import struct
import zlib
from enum import Enum
from pkg_resources import parse_version

from kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO

            

Reported by Pylint.

                      super().__init__("http", client_conn, server_conn, live)
        self.mode = mode

    _stateobject_attributes = flow.Flow._stateobject_attributes.copy()
    # mypy doesn't support update with kwargs
    _stateobject_attributes.update(dict(
        request=Request,
        response=Response,
        websocket=WebSocketData,

            

Reported by Pylint.

              import binascii
import os
import re
import time
import urllib.parse
import json
from dataclasses import dataclass
from dataclasses import fields
from email.utils import formatdate

            

Reported by Pylint.

                   - For use with the "Set-Cookie" and "Cookie" headers, either use `Response.cookies` or see `Headers.get_all`.
    """

    def __init__(self, fields: Iterable[Tuple[bytes, bytes]] = (), **headers):
        """
        *Args:*
         - *fields:* (optional) list of ``(name, value)`` header byte tuples,
           e.g. ``[(b"Host", b"example.com")]``. All names and values must be bytes.
         - *\*\*headers:* Additional headers to set. Will overwrite existing values from `fields`.

            

Reported by Pylint.

                      *Args:*
         - *fields:* (optional) list of ``(name, value)`` header byte tuples,
           e.g. ``[(b"Host", b"example.com")]``. All names and values must be bytes.
         - *\*\*headers:* Additional headers to set. Will overwrite existing values from `fields`.
           For convenience, underscores in header names will be transformed to dashes -
           this behaviour does not extend to other methods.

        If ``**headers`` contains multiple keys that have equal ``.lower()`` representations,
        the behavior is undefined.

            

Reported by Pylint.

                      *Args:*
         - *fields:* (optional) list of ``(name, value)`` header byte tuples,
           e.g. ``[(b"Host", b"example.com")]``. All names and values must be bytes.
         - *\*\*headers:* Additional headers to set. Will overwrite existing values from `fields`.
           For convenience, underscores in header names will be transformed to dashes -
           this behaviour does not extend to other methods.

        If ``**headers`` contains multiple keys that have equal ``.lower()`` representations,
        the behavior is undefined.

            

Reported by Pylint.

                      self.headers.pop("content-encoding", None)
        self.content = decoded

    def encode(self, encoding: str) -> None:
        """
        Encodes body with the given encoding, where e is "gzip", "deflate", "identity", "br", or "zstd".
        Any existing content-encodings are overwritten, the content is not decoded beforehand.

        *Raises:*

            

Reported by Pylint.

                  def make(
        cls,
        method: str,
        url: str,
        content: Union[bytes, str] = "",
        headers: Union[Headers, Dict[Union[str, bytes], Union[str, bytes]], Iterable[Tuple[bytes, bytes]]] = ()
    ) -> "Request":
        """
        Simplified API for creating request objects.

            

Reported by Pylint.

                  def _set_multipart_form(self, value):
        is_valid_content_type = self.headers.get("content-type", "").lower().startswith("multipart/form-data")
        if not is_valid_content_type:
            """
            Generate a random boundary here.

            See <https://datatracker.ietf.org/doc/html/rfc2046#section-5.1.1> for specifications
            on generating the boundary.
            """

            

Reported by Pylint.

                      Simplified API for creating response objects.
        """
        if isinstance(headers, Headers):
            headers = headers
        elif isinstance(headers, dict):
            headers = Headers(
                (always_bytes(k, "utf-8", "surrogateescape"),  # type: ignore
                 always_bytes(v, "utf-8", "surrogateescape"))
                for k, v in headers.items()

            

Reported by Pylint.

              import binascii
import os
import re
import time
import urllib.parse
import json
from dataclasses import dataclass
from dataclasses import fields
from email.utils import formatdate

            

Reported by Pylint.

              import typing
from dataclasses import dataclass

import pytest

from mitmproxy.proxy import commands, events, layer
from . import tutils



            

Reported by Pylint.

              import pytest

from mitmproxy.proxy import commands, events, layer
from . import tutils


class TEvent(events.Event):
    commands: typing.Iterable[typing.Any]


            

Reported by Pylint.

                  return tutils.Playbook(TLayer(tctx), expected=[])


def test_simple(tplaybook):
    assert (
            tplaybook
            >> TEvent()
            << TCommand()
            >> TEvent([])

            

Reported by Pylint.

                  )


def test_mismatch(tplaybook):
    with pytest.raises(AssertionError, match="Playbook mismatch"):
        assert (
                tplaybook
                >> TEvent([])
                << TCommand()

            

Reported by Pylint.

                      )


def test_partial_assert(tplaybook):
    """Developers can assert parts of a playbook and the continue later on."""
    assert (
            tplaybook
            >> TEvent()
            << TCommand()

            

Reported by Pylint.

              

@pytest.mark.parametrize("typed", [True, False])
def test_placeholder(tplaybook, typed):
    """Developers can specify placeholders for yet unknown attributes."""
    if typed:
        f = tutils.Placeholder(int)
    else:
        f = tutils.Placeholder()

            

Reported by Pylint.

                  assert f() == 42


def test_placeholder_type_mismatch(tplaybook):
    """Developers can specify placeholders for yet unknown attributes."""
    f = tutils.Placeholder(str)
    with pytest.raises(TypeError, match="Placeholder type error for TCommand.x: expected str, got int."):
        assert (
                tplaybook

            

Reported by Pylint.

                      )


def test_unfinished(tplaybook):
    """We show a warning when playbooks aren't asserted."""
    tplaybook >> TEvent()
    with pytest.raises(RuntimeError, match="Unfinished playbook"):
        tplaybook.__del__()
    tplaybook._errored = True

            

Reported by Pylint.

              
def test_unfinished(tplaybook):
    """We show a warning when playbooks aren't asserted."""
    tplaybook >> TEvent()
    with pytest.raises(RuntimeError, match="Unfinished playbook"):
        tplaybook.__del__()
    tplaybook._errored = True
    tplaybook.__del__()


            

Reported by Pylint.

                  tplaybook >> TEvent()
    with pytest.raises(RuntimeError, match="Unfinished playbook"):
        tplaybook.__del__()
    tplaybook._errored = True
    tplaybook.__del__()


def test_command_reply(tplaybook):
    """CommandReplies can use relative offsets to point to the matching command."""

            

Reported by Pylint.

              import sys
from pathlib import Path

import pytest

from mitmproxy.addons.maplocal import MapLocal, MapLocalSpec, file_candidates
from mitmproxy.utils.spec import parse_spec
from mitmproxy.test import taddons
from mitmproxy.test import tflow

            

Reported by Pylint.

                  "url,spec,expected_candidates",
    [
        # trailing slashes
        ("https://example.com/foo", ":example.com/foo:/tmp", ["/tmp/index.html"]),
        ("https://example.com/foo/", ":example.com/foo:/tmp", ["/tmp/index.html"]),
        ("https://example.com/foo", ":example.com/foo:/tmp/", ["/tmp/index.html"]),
    ] + [
        # simple prefixes
        ("http://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),

            

Reported by Bandit.

                  [
        # trailing slashes
        ("https://example.com/foo", ":example.com/foo:/tmp", ["/tmp/index.html"]),
        ("https://example.com/foo/", ":example.com/foo:/tmp", ["/tmp/index.html"]),
        ("https://example.com/foo", ":example.com/foo:/tmp/", ["/tmp/index.html"]),
    ] + [
        # simple prefixes
        ("http://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),

            

Reported by Bandit.

                      # trailing slashes
        ("https://example.com/foo", ":example.com/foo:/tmp", ["/tmp/index.html"]),
        ("https://example.com/foo/", ":example.com/foo:/tmp", ["/tmp/index.html"]),
        ("https://example.com/foo", ":example.com/foo:/tmp/", ["/tmp/index.html"]),
    ] + [
        # simple prefixes
        ("http://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg?query", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),

            

Reported by Bandit.

                      ("https://example.com/foo", ":example.com/foo:/tmp/", ["/tmp/index.html"]),
    ] + [
        # simple prefixes
        ("http://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg?query", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar/baz.jpg", ":example.com/foo:/tmp",
         ["/tmp/bar/baz.jpg", "/tmp/bar/baz.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":/foo/bar.jpg:/tmp", ["/tmp/index.html"]),

            

Reported by Bandit.

                      ("https://example.com/foo", ":example.com/foo:/tmp/", ["/tmp/index.html"]),
    ] + [
        # simple prefixes
        ("http://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg?query", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar/baz.jpg", ":example.com/foo:/tmp",
         ["/tmp/bar/baz.jpg", "/tmp/bar/baz.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":/foo/bar.jpg:/tmp", ["/tmp/index.html"]),

            

Reported by Bandit.

                  ] + [
        # simple prefixes
        ("http://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg?query", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar/baz.jpg", ":example.com/foo:/tmp",
         ["/tmp/bar/baz.jpg", "/tmp/bar/baz.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":/foo/bar.jpg:/tmp", ["/tmp/index.html"]),
    ] + [

            

Reported by Bandit.

                  ] + [
        # simple prefixes
        ("http://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg?query", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar/baz.jpg", ":example.com/foo:/tmp",
         ["/tmp/bar/baz.jpg", "/tmp/bar/baz.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":/foo/bar.jpg:/tmp", ["/tmp/index.html"]),
    ] + [

            

Reported by Bandit.

                      # simple prefixes
        ("http://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg?query", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar/baz.jpg", ":example.com/foo:/tmp",
         ["/tmp/bar/baz.jpg", "/tmp/bar/baz.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":/foo/bar.jpg:/tmp", ["/tmp/index.html"]),
    ] + [
        # URL decode and special characters

            

Reported by Bandit.

                      # simple prefixes
        ("http://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg?query", ":example.com/foo:/tmp", ["/tmp/bar.jpg", "/tmp/bar.jpg/index.html"]),
        ("https://example.com/foo/bar/baz.jpg", ":example.com/foo:/tmp",
         ["/tmp/bar/baz.jpg", "/tmp/bar/baz.jpg/index.html"]),
        ("https://example.com/foo/bar.jpg", ":/foo/bar.jpg:/tmp", ["/tmp/index.html"]),
    ] + [
        # URL decode and special characters

            

Reported by Bandit.

              import click
import collections
import collections.abc
import pydivert
import pydivert.consts

if typing.TYPE_CHECKING:
    class WindowsError(OSError):
        @property

            

Reported by Pylint.

              import collections
import collections.abc
import pydivert
import pydivert.consts

if typing.TYPE_CHECKING:
    class WindowsError(OSError):
        @property
        def winerror(self) -> int:

            

Reported by Pylint.

                      self.redirect_request = redirect_request
        super().__init__(self.handle, filter)

    def handle(self, packet):
        client = (packet.src_addr, packet.src_port)

        if client not in self.tcp_connections:
            self.tcp_connections.refresh()


            

Reported by Pylint.

                  proxy = TransparentProxy(**options)
    proxy.start()
    print(f" * Redirection active.")
    print(f"   Filter: {proxy.request_filter}")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print(" * Shutting down...")

            

Reported by Pylint.

                      self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((REDIRECT_API_HOST, REDIRECT_API_PORT))

        self.wfile = self.sock.makefile('wb')
        self.rfile = self.sock.makefile('rb')
        write(os.getpid(), self.wfile)

    def original_addr(self, csock: socket.socket):
        ip, port = csock.getpeername()[:2]

            

Reported by Pylint.

                      self.sock.connect((REDIRECT_API_HOST, REDIRECT_API_PORT))

        self.wfile = self.sock.makefile('wb')
        self.rfile = self.sock.makefile('rb')
        write(os.getpid(), self.wfile)

    def original_addr(self, csock: socket.socket):
        ip, port = csock.getpeername()[:2]
        ip = re.sub(r"^::ffff:(?=\d+.\d+.\d+.\d+$)", "", ip)

            

Reported by Pylint.

                  def __init__(
        self,
        handle: typing.Callable[[pydivert.Packet], None],
        filter: str,
        layer: pydivert.Layer = pydivert.Layer.NETWORK,
        flags: pydivert.Flag = 0
    ) -> None:
        self.handle = handle
        self.windivert = pydivert.WinDivert(filter, layer, flags=flags)

            

Reported by Pylint.

                  def __init__(
        self,
        redirect_request: typing.Callable[[pydivert.Packet], None],
        filter: str
    ) -> None:
        self.tcp_connections = TcpConnectionTable()
        self.trusted_pids = set()
        self.redirect_request = redirect_request
        super().__init__(self.handle, filter)

            

Reported by Pylint.

                      local: bool = True,
        forward: bool = True,
        proxy_port: int = 8080,
        filter: typing.Optional[str] = "tcp.DstPort == 80 or tcp.DstPort == 443",
    ) -> None:
        self.proxy_port = proxy_port
        self.filter = (
            filter
            or

            

Reported by Pylint.

              
    @classmethod
    def setup(cls):
        # TODO: Make sure that server can be killed cleanly. That's a bit difficult as we don't have access to
        # controller.should_exit when this is called.
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_unavailable = s.connect_ex((REDIRECT_API_HOST, REDIRECT_API_PORT))
        if server_unavailable:
            proxifier = TransparentProxy()

            

Reported by Pylint.