from mitmproxy import ctx
event_log = []


class Addon:
    @property
    def event_log(self):
        return event_log


            

Reported by Pylint.

                  def event_log(self):
        return event_log

    def load(self, opts):
        ctx.log.info("addon running")
        event_log.append("addonload")

    def configure(self, updated):
        event_log.append("addonconfigure")

            

Reported by Pylint.

                      ctx.log.info("addon running")
        event_log.append("addonload")

    def configure(self, updated):
        event_log.append("addonconfigure")


def configure(updated):
    event_log.append("scriptconfigure")

            

Reported by Pylint.

                      event_log.append("addonconfigure")


def configure(updated):
    event_log.append("scriptconfigure")


def load(l):
    event_log.append("scriptload")

            

Reported by Pylint.

                  event_log.append("scriptconfigure")


def load(l):
    event_log.append("scriptload")


addons = [Addon()]

            

Reported by Pylint.

              from mitmproxy import ctx
event_log = []


class Addon:
    @property
    def event_log(self):
        return event_log


            

Reported by Pylint.

              event_log = []


class Addon:
    @property
    def event_log(self):
        return event_log

    def load(self, opts):

            

Reported by Pylint.

              
class Addon:
    @property
    def event_log(self):
        return event_log

    def load(self, opts):
        ctx.log.info("addon running")
        event_log.append("addonload")

            

Reported by Pylint.

                  def event_log(self):
        return event_log

    def load(self, opts):
        ctx.log.info("addon running")
        event_log.append("addonload")

    def configure(self, updated):
        event_log.append("addonconfigure")

            

Reported by Pylint.

                  def event_log(self):
        return event_log

    def load(self, opts):
        ctx.log.info("addon running")
        event_log.append("addonload")

    def configure(self, updated):
        event_log.append("addonconfigure")

            

Reported by Pylint.

              import typing

Type = typing.Union[
    typing.Any  # anything more elaborate really fails with mypy at the moment.
]


def sequence_type(typeinfo: typing.Type[typing.List]) -> Type:
    """Return the type of a sequence, e.g. typing.List"""

            

Reported by Pylint.

                  return typeinfo.__args__  # type: ignore


def check_option_type(name: str, value: typing.Any, typeinfo: Type) -> None:
    """
    Check if the provided value is an instance of typeinfo and raises a
    TypeError otherwise. This function supports only those types required for
    options.
    """

            

Reported by Pylint.

                  TypeError otherwise. This function supports only those types required for
    options.
    """
    e = TypeError("Expected {} for {}, but got {}.".format(
        typeinfo,
        name,
        type(value)
    ))


            

Reported by Pylint.

              
    typename = str(typeinfo)

    if typename.startswith("typing.Union") or typename.startswith("typing.Optional"):
        for T in union_types(typeinfo):
            try:
                check_option_type(name, value, T)
            except TypeError:
                pass

            

Reported by Pylint.

                  typename = str(typeinfo)

    if typename.startswith("typing.Union") or typename.startswith("typing.Optional"):
        for T in union_types(typeinfo):
            try:
                check_option_type(name, value, T)
            except TypeError:
                pass
            else:

            

Reported by Pylint.

                          raise e
        if len(types) != len(value):
            raise e
        for i, (x, T) in enumerate(zip(value, types)):
            check_option_type(f"{name}[{i}]", x, T)
        return
    elif typename.startswith("typing.Sequence"):
        T = sequence_type(typeinfo)
        if not isinstance(value, (tuple, list)):

            

Reported by Pylint.

                          raise e
        if len(types) != len(value):
            raise e
        for i, (x, T) in enumerate(zip(value, types)):
            check_option_type(f"{name}[{i}]", x, T)
        return
    elif typename.startswith("typing.Sequence"):
        T = sequence_type(typeinfo)
        if not isinstance(value, (tuple, list)):

            

Reported by Pylint.

                          check_option_type(f"{name}[{i}]", x, T)
        return
    elif typename.startswith("typing.Sequence"):
        T = sequence_type(typeinfo)
        if not isinstance(value, (tuple, list)):
            raise e
        for v in value:
            check_option_type(name, v, T)
    elif typename.startswith("typing.IO"):

            

Reported by Pylint.

                      T = sequence_type(typeinfo)
        if not isinstance(value, (tuple, list)):
            raise e
        for v in value:
            check_option_type(name, v, T)
    elif typename.startswith("typing.IO"):
        if hasattr(value, "read"):
            return
        else:

            

Reported by Pylint.

                      for v in value:
            check_option_type(name, v, T)
    elif typename.startswith("typing.IO"):
        if hasattr(value, "read"):
            return
        else:
            raise e
    elif typename.startswith("typing.Any"):
        return

            

Reported by Pylint.

                  subprocess.check_call([
        "aws", "s3", "cp",
        "--acl", "public-read",
        f"./release/dist/",
        f"s3://snapshots.mitmproxy.org/{upload_dir}/",
        "--recursive",
    ])

    # Upload releases to PyPI

            

Reported by Pylint.

                      subprocess.check_call(["twine", "upload", whl])

    # Upload dev docs
    if branch == "main" or branch == "actions-hardening":  # FIXME remove
        subprocess.check_call([
            "aws", "configure",
            "set", "preview.cloudfront", "true"
        ])
        subprocess.check_call([

            

Reported by Pylint.

              #!/usr/bin/env python3
import os
import re
import subprocess
from pathlib import Path
from typing import Optional
# Security: No third-party dependencies here!

if __name__ == "__main__":

            

Reported by Pylint.

              #!/usr/bin/env python3
import os
import re
import subprocess
from pathlib import Path
from typing import Optional
# Security: No third-party dependencies here!

if __name__ == "__main__":

            

Reported by Bandit.

                      upload_dir = re.sub(r"^v([\d.]+)$", r"\1", tag)
    else:
        upload_dir = f"branches/{branch}"
    subprocess.check_call([
        "aws", "s3", "cp",
        "--acl", "public-read",
        f"./release/dist/",
        f"s3://snapshots.mitmproxy.org/{upload_dir}/",
        "--recursive",

            

Reported by Bandit.

                      upload_dir = re.sub(r"^v([\d.]+)$", r"\1", tag)
    else:
        upload_dir = f"branches/{branch}"
    subprocess.check_call([
        "aws", "s3", "cp",
        "--acl", "public-read",
        f"./release/dist/",
        f"s3://snapshots.mitmproxy.org/{upload_dir}/",
        "--recursive",

            

Reported by Bandit.

                  # Upload releases to PyPI
    if tag:
        whl, = Path("release/dist/").glob('mitmproxy-*-py3-none-any.whl')
        subprocess.check_call(["twine", "upload", whl])

    # Upload dev docs
    if branch == "main" or branch == "actions-hardening":  # FIXME remove
        subprocess.check_call([
            "aws", "configure",

            

Reported by Bandit.

                  # Upload releases to PyPI
    if tag:
        whl, = Path("release/dist/").glob('mitmproxy-*-py3-none-any.whl')
        subprocess.check_call(["twine", "upload", whl])

    # Upload dev docs
    if branch == "main" or branch == "actions-hardening":  # FIXME remove
        subprocess.check_call([
            "aws", "configure",

            

Reported by Bandit.

                      subprocess.check_call(["twine", "upload", whl])

    # Upload dev docs
    if branch == "main" or branch == "actions-hardening":  # FIXME remove
        subprocess.check_call([
            "aws", "configure",
            "set", "preview.cloudfront", "true"
        ])
        subprocess.check_call([

            

Reported by Pylint.

              
    # Upload dev docs
    if branch == "main" or branch == "actions-hardening":  # FIXME remove
        subprocess.check_call([
            "aws", "configure",
            "set", "preview.cloudfront", "true"
        ])
        subprocess.check_call([
            "aws", "s3",

            

Reported by Bandit.

              #!/usr/bin/env python3
import click

from mitmproxy.addons import dumper
from mitmproxy.test import tflow
from mitmproxy.test import taddons


def show(flow_detail, flows):

            

Reported by Pylint.

              import click

from mitmproxy.addons import dumper
from mitmproxy.test import tflow
from mitmproxy.test import taddons


def show(flow_detail, flows):
    d = dumper.Dumper()

            

Reported by Pylint.

              
from mitmproxy.addons import dumper
from mitmproxy.test import tflow
from mitmproxy.test import taddons


def show(flow_detail, flows):
    d = dumper.Dumper()
    with taddons.context() as ctx:

            

Reported by Pylint.

              #!/usr/bin/env python3
import click

from mitmproxy.addons import dumper
from mitmproxy.test import tflow
from mitmproxy.test import taddons


def show(flow_detail, flows):

            

Reported by Pylint.

              from mitmproxy.test import taddons


def show(flow_detail, flows):
    d = dumper.Dumper()
    with taddons.context() as ctx:
        ctx.configure(d, flow_detail=flow_detail)
        for f in flows:
            ctx.cycle(d, f)

            

Reported by Pylint.

              

def show(flow_detail, flows):
    d = dumper.Dumper()
    with taddons.context() as ctx:
        ctx.configure(d, flow_detail=flow_detail)
        for f in flows:
            ctx.cycle(d, f)


            

Reported by Pylint.

                  d = dumper.Dumper()
    with taddons.context() as ctx:
        ctx.configure(d, flow_detail=flow_detail)
        for f in flows:
            ctx.cycle(d, f)


@click.group()
def cli():

            

Reported by Pylint.

              

@click.group()
def cli():
    pass


@cli.command()
@click.option('--level', default=1, help='Detail level')

            

Reported by Pylint.

              
@cli.command()
@click.option('--level', default=1, help='Detail level')
def tcp(level):
    f1 = tflow.ttcpflow(client_conn=True, server_conn=True)
    show(level, [f1])


@cli.command()

            

Reported by Pylint.

              @cli.command()
@click.option('--level', default=1, help='Detail level')
def tcp(level):
    f1 = tflow.ttcpflow(client_conn=True, server_conn=True)
    show(level, [f1])


@cli.command()
@click.option('--level', default=1, help='Detail level')

            

Reported by Pylint.

              from mitmproxy import flow
from mitmproxy.addons import disable_h2c
from mitmproxy.test import taddons, tutils
from mitmproxy.test import tflow


class TestDisableH2CleartextUpgrade:
    def test_upgrade(self):
        with taddons.context() as tctx:

            

Reported by Pylint.

              from mitmproxy.test import tflow


class TestDisableH2CleartextUpgrade:
    def test_upgrade(self):
        with taddons.context() as tctx:
            a = disable_h2c.DisableH2C()
            tctx.configure(a)


            

Reported by Pylint.

              

class TestDisableH2CleartextUpgrade:
    def test_upgrade(self):
        with taddons.context() as tctx:
            a = disable_h2c.DisableH2C()
            tctx.configure(a)

            f = tflow.tflow()

            

Reported by Pylint.

              

class TestDisableH2CleartextUpgrade:
    def test_upgrade(self):
        with taddons.context() as tctx:
            a = disable_h2c.DisableH2C()
            tctx.configure(a)

            f = tflow.tflow()

            

Reported by Pylint.

              class TestDisableH2CleartextUpgrade:
    def test_upgrade(self):
        with taddons.context() as tctx:
            a = disable_h2c.DisableH2C()
            tctx.configure(a)

            f = tflow.tflow()
            f.request.headers['upgrade'] = 'h2c'
            f.request.headers['connection'] = 'foo'

            

Reported by Pylint.

                          a = disable_h2c.DisableH2C()
            tctx.configure(a)

            f = tflow.tflow()
            f.request.headers['upgrade'] = 'h2c'
            f.request.headers['connection'] = 'foo'
            f.request.headers['http2-settings'] = 'bar'

            a.request(f)

            

Reported by Pylint.

                          f.request.headers['http2-settings'] = 'bar'

            a.request(f)
            assert 'upgrade' not in f.request.headers
            assert 'connection' not in f.request.headers
            assert 'http2-settings' not in f.request.headers

    def test_prior_knowledge(self):
        with taddons.context() as tctx:

            

Reported by Bandit.

              
            a.request(f)
            assert 'upgrade' not in f.request.headers
            assert 'connection' not in f.request.headers
            assert 'http2-settings' not in f.request.headers

    def test_prior_knowledge(self):
        with taddons.context() as tctx:
            a = disable_h2c.DisableH2C()

            

Reported by Bandit.

                          a.request(f)
            assert 'upgrade' not in f.request.headers
            assert 'connection' not in f.request.headers
            assert 'http2-settings' not in f.request.headers

    def test_prior_knowledge(self):
        with taddons.context() as tctx:
            a = disable_h2c.DisableH2C()
            tctx.configure(a)

            

Reported by Bandit.

                          assert 'connection' not in f.request.headers
            assert 'http2-settings' not in f.request.headers

    def test_prior_knowledge(self):
        with taddons.context() as tctx:
            a = disable_h2c.DisableH2C()
            tctx.configure(a)

            f = tflow.tflow()

            

Reported by Pylint.

              import asyncio
import pytest

from mitmproxy.addons import keepserving
from mitmproxy.test import taddons
from mitmproxy import command


class Dummy:

            

Reported by Pylint.

                  _is_shutdown = False

    def shutdown(self):
        self.is_shutdown = True


@pytest.mark.asyncio
async def test_keepserving():
    ks = TKS()

            

Reported by Pylint.

              import asyncio
import pytest

from mitmproxy.addons import keepserving
from mitmproxy.test import taddons
from mitmproxy import command


class Dummy:

            

Reported by Pylint.

              from mitmproxy import command


class Dummy:
    def __init__(self, val: bool):
        self.val = val

    def load(self, loader):
        loader.add_option("client_replay", bool, self.val, "test")

            

Reported by Pylint.

                  def __init__(self, val: bool):
        self.val = val

    def load(self, loader):
        loader.add_option("client_replay", bool, self.val, "test")
        loader.add_option("server_replay", bool, self.val, "test")
        loader.add_option("rfile", bool, self.val, "test")

    @command.command("readfile.reading")

            

Reported by Pylint.

                      loader.add_option("rfile", bool, self.val, "test")

    @command.command("readfile.reading")
    def readfile(self) -> bool:
        return self.val

    @command.command("replay.client.count")
    def creplay(self) -> int:
        return 1 if self.val else 0

            

Reported by Pylint.

                      return self.val

    @command.command("replay.client.count")
    def creplay(self) -> int:
        return 1 if self.val else 0

    @command.command("replay.server.count")
    def sreplay(self) -> int:
        return 1 if self.val else 0

            

Reported by Pylint.

                      return 1 if self.val else 0

    @command.command("replay.server.count")
    def sreplay(self) -> int:
        return 1 if self.val else 0


class TKS(keepserving.KeepServing):
    _is_shutdown = False

            

Reported by Pylint.

                      return 1 if self.val else 0


class TKS(keepserving.KeepServing):
    _is_shutdown = False

    def shutdown(self):
        self.is_shutdown = True


            

Reported by Pylint.

              

@pytest.mark.asyncio
async def test_keepserving():
    ks = TKS()
    d = Dummy(True)
    with taddons.context(ks) as tctx:
        tctx.master.addons.add(d)
        ks.running()

            

Reported by Pylint.

              import pytest
from mitmproxy.contentviews import base


def test_format_dict():
    d = {"one": "two", "three": "four"}
    f_d = base.format_dict(d)
    assert next(f_d)


            

Reported by Pylint.

              import pytest
from mitmproxy.contentviews import base


def test_format_dict():
    d = {"one": "two", "three": "four"}
    f_d = base.format_dict(d)
    assert next(f_d)


            

Reported by Pylint.

              from mitmproxy.contentviews import base


def test_format_dict():
    d = {"one": "two", "three": "four"}
    f_d = base.format_dict(d)
    assert next(f_d)

    d = {"adsfa": ""}

            

Reported by Pylint.

              

def test_format_dict():
    d = {"one": "two", "three": "four"}
    f_d = base.format_dict(d)
    assert next(f_d)

    d = {"adsfa": ""}
    f_d = base.format_dict(d)

            

Reported by Pylint.

              def test_format_dict():
    d = {"one": "two", "three": "four"}
    f_d = base.format_dict(d)
    assert next(f_d)

    d = {"adsfa": ""}
    f_d = base.format_dict(d)
    assert next(f_d)


            

Reported by Bandit.

                  f_d = base.format_dict(d)
    assert next(f_d)

    d = {"adsfa": ""}
    f_d = base.format_dict(d)
    assert next(f_d)

    d = {}
    f_d = base.format_dict(d)

            

Reported by Pylint.

              
    d = {"adsfa": ""}
    f_d = base.format_dict(d)
    assert next(f_d)

    d = {}
    f_d = base.format_dict(d)
    with pytest.raises(StopIteration):
        next(f_d)

            

Reported by Bandit.

                  f_d = base.format_dict(d)
    assert next(f_d)

    d = {}
    f_d = base.format_dict(d)
    with pytest.raises(StopIteration):
        next(f_d)



            

Reported by Pylint.

                      next(f_d)


def test_format_pairs():
    d = [("a", "c"), ("b", "d")]
    f_d = base.format_pairs(d)
    assert next(f_d)

    d = [("abc", "")]

            

Reported by Pylint.

              

def test_format_pairs():
    d = [("a", "c"), ("b", "d")]
    f_d = base.format_pairs(d)
    assert next(f_d)

    d = [("abc", "")]
    f_d = base.format_pairs(d)

            

Reported by Pylint.

              import collections

import urwid
from mitmproxy.tools.console import layoutwidget
from mitmproxy import log


class LogBufferWalker(urwid.SimpleListWalker):
    pass

            

Reported by Pylint.

                          self.set_focus(0)
        return super().keypress(size, key)

    def add_event(self, event_store, entry: log.LogEntry):
        if log.log_tier(self.master.options.console_eventlog_verbosity) < log.log_tier(entry.level):
            return
        txt = "{}: {}".format(entry.level, str(entry.msg))
        if entry.level in ("error", "warn", "alert"):
            e = urwid.Text((entry.level, txt))

            

Reported by Pylint.

              import collections

import urwid
from mitmproxy.tools.console import layoutwidget
from mitmproxy import log


class LogBufferWalker(urwid.SimpleListWalker):
    pass

            

Reported by Pylint.

              from mitmproxy import log


class LogBufferWalker(urwid.SimpleListWalker):
    pass


class EventLog(urwid.ListBox, layoutwidget.LayoutWidget):
    keyctx = "eventlog"

            

Reported by Pylint.

              from mitmproxy import log


class LogBufferWalker(urwid.SimpleListWalker):
    pass


class EventLog(urwid.ListBox, layoutwidget.LayoutWidget):
    keyctx = "eventlog"

            

Reported by Pylint.

                  pass


class EventLog(urwid.ListBox, layoutwidget.LayoutWidget):
    keyctx = "eventlog"
    title = "Events"

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

            

Reported by Pylint.

              
        super().__init__(self.walker)

    def load(self, loader):
        loader.add_option(
            "console_focus_follow", bool, False,
            "Focus follows new flows."
        )


            

Reported by Pylint.

              
        super().__init__(self.walker)

    def load(self, loader):
        loader.add_option(
            "console_focus_follow", bool, False,
            "Focus follows new flows."
        )


            

Reported by Pylint.

                          "Focus follows new flows."
        )

    def set_focus(self, index):
        if 0 <= index < len(self.walker):
            super().set_focus(index)

    def keypress(self, size, key):
        if key == "m_end":

            

Reported by Pylint.

                      if 0 <= index < len(self.walker):
            super().set_focus(index)

    def keypress(self, size, key):
        if key == "m_end":
            self.set_focus(len(self.walker) - 1)
        elif key == "m_start":
            self.set_focus(0)
        return super().keypress(size, key)

            

Reported by Pylint.

                      return f

    async def dummy_cycle(self, master, n, content):
        for i in range(n):
            await self.cycle(master, content)
        await master._shutdown()

    def flowfile(self, path):
        with open(path, "wb") as f:

            

Reported by Pylint.

                  async def dummy_cycle(self, master, n, content):
        for i in range(n):
            await self.cycle(master, content)
        await master._shutdown()

    def flowfile(self, path):
        with open(path, "wb") as f:
            fw = io.FlowWriter(f)
            t = tflow.tflow(resp=True)

            

Reported by Pylint.

              from unittest import mock

from mitmproxy import controller
from mitmproxy import eventsequence
from mitmproxy import io
from mitmproxy.proxy import server_hooks
from mitmproxy.test import tflow
from mitmproxy.test import tutils


            

Reported by Pylint.

              from mitmproxy.test import tutils


class MasterTest:

    async def cycle(self, master, content):
        f = tflow.tflow(req=tutils.treq(content=content))
        layer = mock.Mock("mitmproxy.proxy.protocol.base.Layer")
        layer.client_conn = f.client_conn

            

Reported by Pylint.

              
class MasterTest:

    async def cycle(self, master, content):
        f = tflow.tflow(req=tutils.treq(content=content))
        layer = mock.Mock("mitmproxy.proxy.protocol.base.Layer")
        layer.client_conn = f.client_conn
        layer.reply = controller.DummyReply()
        await master.addons.handle_lifecycle(server_hooks.ClientConnectedHook(layer))

            

Reported by Pylint.

              class MasterTest:

    async def cycle(self, master, content):
        f = tflow.tflow(req=tutils.treq(content=content))
        layer = mock.Mock("mitmproxy.proxy.protocol.base.Layer")
        layer.client_conn = f.client_conn
        layer.reply = controller.DummyReply()
        await master.addons.handle_lifecycle(server_hooks.ClientConnectedHook(layer))
        for e in eventsequence.iterate(f):

            

Reported by Pylint.

                      layer.client_conn = f.client_conn
        layer.reply = controller.DummyReply()
        await master.addons.handle_lifecycle(server_hooks.ClientConnectedHook(layer))
        for e in eventsequence.iterate(f):
            await master.addons.handle_lifecycle(e)
        await master.addons.handle_lifecycle(server_hooks.ClientDisconnectedHook(layer))
        return f

    async def dummy_cycle(self, master, n, content):

            

Reported by Pylint.

                      await master.addons.handle_lifecycle(server_hooks.ClientDisconnectedHook(layer))
        return f

    async def dummy_cycle(self, master, n, content):
        for i in range(n):
            await self.cycle(master, content)
        await master._shutdown()

    def flowfile(self, path):

            

Reported by Pylint.

                      await master.addons.handle_lifecycle(server_hooks.ClientDisconnectedHook(layer))
        return f

    async def dummy_cycle(self, master, n, content):
        for i in range(n):
            await self.cycle(master, content)
        await master._shutdown()

    def flowfile(self, path):

            

Reported by Pylint.

                          await self.cycle(master, content)
        await master._shutdown()

    def flowfile(self, path):
        with open(path, "wb") as f:
            fw = io.FlowWriter(f)
            t = tflow.tflow(resp=True)
            fw.add(t)

            

Reported by Pylint.

              import time
from typing import Optional, Tuple

from h11._receivebuffer import ReceiveBuffer

from mitmproxy import http, connection
from mitmproxy.net.http import http1
from mitmproxy.proxy import commands, context, layer, tunnel
from mitmproxy.proxy.layers.http._hooks import HttpConnectUpstreamHook

            

Reported by Pylint.

                      self.buf += data
        response_head = self.buf.maybe_extract_lines()
        if response_head:
            response_head = [bytes(x) for x in response_head]  # TODO: Make url.parse compatible with bytearrays
            try:
                response = http1.read_response_head(response_head)
            except ValueError as e:
                proxyaddr = human.format_address(self.tunnel_connection.address)
                yield commands.Log(f"{proxyaddr}: {e}")

            

Reported by Pylint.

              import time
from typing import Optional, Tuple

from h11._receivebuffer import ReceiveBuffer

from mitmproxy import http, connection
from mitmproxy.net.http import http1
from mitmproxy.proxy import commands, context, layer, tunnel
from mitmproxy.proxy.layers.http._hooks import HttpConnectUpstreamHook

            

Reported by Pylint.

              from mitmproxy.utils import human


class HttpUpstreamProxy(tunnel.TunnelLayer):
    buf: ReceiveBuffer
    send_connect: bool
    conn: connection.Server
    tunnel_connection: connection.Server


            

Reported by Pylint.

                      self.send_connect = send_connect

    @classmethod
    def make(cls, ctx: context.Context, send_connect: bool) -> tunnel.LayerStack:
        spec = ctx.server.via
        assert spec
        assert spec.scheme in ("http", "https")

        http_proxy = connection.Server(spec.address)

            

Reported by Pylint.

                  @classmethod
    def make(cls, ctx: context.Context, send_connect: bool) -> tunnel.LayerStack:
        spec = ctx.server.via
        assert spec
        assert spec.scheme in ("http", "https")

        http_proxy = connection.Server(spec.address)

        stack = tunnel.LayerStack()

            

Reported by Bandit.

                  def make(cls, ctx: context.Context, send_connect: bool) -> tunnel.LayerStack:
        spec = ctx.server.via
        assert spec
        assert spec.scheme in ("http", "https")

        http_proxy = connection.Server(spec.address)

        stack = tunnel.LayerStack()
        if spec.scheme == "https":

            

Reported by Bandit.

              
        return stack

    def start_handshake(self) -> layer.CommandGenerator[None]:
        if not self.send_connect:
            return (yield from super().start_handshake())
        assert self.conn.address
        flow = http.HTTPFlow(self.context.client, self.tunnel_connection)
        flow.request = http.Request(

            

Reported by Pylint.

                  def start_handshake(self) -> layer.CommandGenerator[None]:
        if not self.send_connect:
            return (yield from super().start_handshake())
        assert self.conn.address
        flow = http.HTTPFlow(self.context.client, self.tunnel_connection)
        flow.request = http.Request(
            host=self.conn.address[0],
            port=self.conn.address[1],
            method=b"CONNECT",

            

Reported by Bandit.

                      raw = http1.assemble_request(flow.request)
        yield commands.SendData(self.tunnel_connection, raw)

    def receive_handshake_data(self, data: bytes) -> layer.CommandGenerator[Tuple[bool, Optional[str]]]:
        if not self.send_connect:
            return (yield from super().receive_handshake_data(data))
        self.buf += data
        response_head = self.buf.maybe_extract_lines()
        if response_head:

            

Reported by Pylint.