if client_alpn in options:
            return client_alpn
        else:
            return SSL.NO_OVERLAPPING_PROTOCOLS
    if server_alpn and server_alpn in options:
        return server_alpn
    if server_alpn == b"":
        # We do have a server connection, but the remote server refused to negotiate a protocol:
        # We need to mirror this on the client connection.

            

Reported by Pylint.

                  if server_alpn == b"":
        # We do have a server connection, but the remote server refused to negotiate a protocol:
        # We need to mirror this on the client connection.
        return SSL.NO_OVERLAPPING_PROTOCOLS
    http_alpns = tls.HTTP_ALPNS if http2 else tls.HTTP1_ALPNS
    for alpn in options:  # client sends in order of preference, so we are nice and respect that.
        if alpn in http_alpns:
            return alpn
    else:

            

Reported by Pylint.

                      if alpn in http_alpns:
            return alpn
    else:
        return SSL.NO_OVERLAPPING_PROTOCOLS


class TlsConfig:
    """
    This addon supplies the proxy core with the desired OpenSSL connection objects to negotiate TLS.

            

Reported by Pylint.

                  for alpn in options:  # client sends in order of preference, so we are nice and respect that.
        if alpn in http_alpns:
            return alpn
    else:
        return SSL.NO_OVERLAPPING_PROTOCOLS


class TlsConfig:
    """

            

Reported by Pylint.

                  """
    certstore: certs.CertStore = None  # type: ignore

    # TODO: We should support configuring TLS 1.3 cipher suites (https://github.com/mitmproxy/mitmproxy/issues/4260)
    # TODO: We should re-use SSL.Context options here, if only for TLS session resumption.
    #       This may require patches to pyOpenSSL, as some functionality is only exposed on contexts.
    # TODO: This addon should manage the following options itself, which are current defined in mitmproxy/options.py:
    #  - upstream_cert
    #  - add_upstream_certs_to_client_chain

            

Reported by Pylint.

                  certstore: certs.CertStore = None  # type: ignore

    # TODO: We should support configuring TLS 1.3 cipher suites (https://github.com/mitmproxy/mitmproxy/issues/4260)
    # TODO: We should re-use SSL.Context options here, if only for TLS session resumption.
    #       This may require patches to pyOpenSSL, as some functionality is only exposed on contexts.
    # TODO: This addon should manage the following options itself, which are current defined in mitmproxy/options.py:
    #  - upstream_cert
    #  - add_upstream_certs_to_client_chain
    #  - ciphers_client

            

Reported by Pylint.

                  # TODO: We should support configuring TLS 1.3 cipher suites (https://github.com/mitmproxy/mitmproxy/issues/4260)
    # TODO: We should re-use SSL.Context options here, if only for TLS session resumption.
    #       This may require patches to pyOpenSSL, as some functionality is only exposed on contexts.
    # TODO: This addon should manage the following options itself, which are current defined in mitmproxy/options.py:
    #  - upstream_cert
    #  - add_upstream_certs_to_client_chain
    #  - ciphers_client
    #  - ciphers_server
    #  - key_size

            

Reported by Pylint.

                          typespec=str,
            default=net_tls.DEFAULT_MIN_VERSION.name,
            choices=[x.name for x in net_tls.Version],
            help=f"Set the minimum TLS version for client connections.",
        )
        loader.add_option(
            name="tls_version_client_max",
            typespec=str,
            default=net_tls.DEFAULT_MAX_VERSION.name,

            

Reported by Pylint.

                          typespec=str,
            default=net_tls.DEFAULT_MAX_VERSION.name,
            choices=[x.name for x in net_tls.Version],
            help=f"Set the maximum TLS version for client connections.",
        )
        loader.add_option(
            name="tls_version_server_min",
            typespec=str,
            default=net_tls.DEFAULT_MIN_VERSION.name,

            

Reported by Pylint.

                          typespec=str,
            default=net_tls.DEFAULT_MIN_VERSION.name,
            choices=[x.name for x in net_tls.Version],
            help=f"Set the minimum TLS version for server connections.",
        )
        loader.add_option(
            name="tls_version_server_max",
            typespec=str,
            default=net_tls.DEFAULT_MAX_VERSION.name,

            

Reported by Pylint.

                      else:
            handler = asyncio_utils.create_task(
                self.handle_connection(self.client),
                name=f"client connection handler",
                client=self.client.peername,
            )
            if not handler:
                return   # this should not be needed, see asyncio_utils.create_task
            self.transports[self.client].handler = handler

            

Reported by Pylint.

              
    async def open_connection(self, command: commands.OpenConnection) -> None:
        if not command.connection.address:
            self.log(f"Cannot open connection, no hostname given.")
            self.server_event(events.OpenConnectionCompleted(command, f"Cannot open connection, no hostname given."))
            return

        hook_data = server_hooks.ServerConnectionHookData(
            client=self.client,

            

Reported by Pylint.

                  async def open_connection(self, command: commands.OpenConnection) -> None:
        if not command.connection.address:
            self.log(f"Cannot open connection, no hostname given.")
            self.server_event(events.OpenConnectionCompleted(command, f"Cannot open connection, no hostname given."))
            return

        hook_data = server_hooks.ServerConnectionHookData(
            client=self.client,
            server=command.connection

            

Reported by Pylint.

                  async def handle_hook(self, hook: commands.StartHook) -> None:
        pass

    def log(self, message: str, level: str = "info") -> None:
        print(message)

    def server_event(self, event: events.Event) -> None:
        self.timeout_watchdog.register_activity()
        try:

            

Reported by Pylint.

                                  self.log(command.message, command.level)
                else:
                    raise RuntimeError(f"Unexpected command: {command}")
        except Exception:
            self.log(f"mitmproxy has crashed!\n{traceback.format_exc()}", level="error")

    def close_connection(self, connection: Connection, half_close: bool = False) -> None:
        if half_close:
            if not connection.state & ConnectionState.CAN_WRITE:

            

Reported by Pylint.

                          # lambda ctx: layers.HttpLayer(ctx, HTTPMode.regular),
            # lambda ctx: setattr(ctx.server, "tls", True) or layers.ServerTLSLayer(ctx),
            # lambda ctx: layers.ClientTLSLayer(ctx),
            lambda ctx: layers.modes.ReverseProxy(ctx),
            lambda ctx: layers.HttpLayer(ctx, HTTPMode.transparent)
        ]

        def next_layer(nl: layer.NextLayer):
            l = layer_stack.pop(0)(nl.context)

            

Reported by Pylint.

              from mitmproxy.utils.data import pkg_data


class TimeoutWatchdog:
    last_activity: float
    CONNECTION_TIMEOUT = 10 * 60
    can_timeout: asyncio.Event
    blocker: int


            

Reported by Pylint.

                      self.can_timeout.set()
        self.blocker = 0

    def register_activity(self):
        self.last_activity = time.time()

    async def watch(self):
        while True:
            await self.can_timeout.wait()

            

Reported by Pylint.

                  def register_activity(self):
        self.last_activity = time.time()

    async def watch(self):
        while True:
            await self.can_timeout.wait()
            await asyncio.sleep(self.CONNECTION_TIMEOUT - (time.time() - self.last_activity))
            if self.last_activity + self.CONNECTION_TIMEOUT < time.time():
                await self.callback()

            

Reported by Pylint.

                              return

    @contextmanager
    def disarm(self):
        self.can_timeout.clear()
        self.blocker += 1
        try:
            yield
        finally:

            

Reported by Pylint.

              import typing
import os

import ruamel.yaml
import ruamel.yaml.error

from mitmproxy import command
from mitmproxy.tools.console import commandexecutor
from mitmproxy.tools.console import signals

            

Reported by Pylint.

              import os

import ruamel.yaml
import ruamel.yaml.error

from mitmproxy import command
from mitmproxy.tools.console import commandexecutor
from mitmproxy.tools.console import signals
from mitmproxy import ctx

            

Reported by Pylint.

              

class Binding:
    def __init__(self, key, command, contexts, help):
        self.key, self.command, self.contexts = key, command, sorted(contexts)
        self.help = help

    def keyspec(self):
        """

            

Reported by Pylint.

              

class Binding:
    def __init__(self, key, command, contexts, help):
        self.key, self.command, self.contexts = key, command, sorted(contexts)
        self.help = help

    def keyspec(self):
        """

            

Reported by Pylint.

                  def add(
        self,
        key: str,
        command: str,
        contexts: typing.Sequence[str],
        help=""
    ) -> None:
        """
            Add a key to the key map.

            

Reported by Pylint.

                      key: str,
        command: str,
        contexts: typing.Sequence[str],
        help=""
    ) -> None:
        """
            Add a key to the key map.
        """
        self._check_contexts(contexts)

            

Reported by Pylint.

                              try:
                    txt = f.read()
                except UnicodeDecodeError as e:
                    raise KeyBindingError(
                        f"Encoding error - expected UTF8: {p}: {e}"
                    )
            try:
                vals = self.parse(txt)
            except KeyBindingError as e:

            

Reported by Pylint.

                          for v in vals:
                user_ctxs = v.get("ctx", ["global"])
                try:
                    km._check_contexts(user_ctxs)
                    km.remove(v["key"], user_ctxs)
                    km.add(
                        key = v["key"],
                        command = v["cmd"],
                        contexts = user_ctxs,

            

Reported by Pylint.

                      except ruamel.yaml.error.MarkedYAMLError as v:
            if hasattr(v, "problem_mark"):
                snip = v.problem_mark.get_snippet()
                raise KeyBindingError(
                    "Key binding config error at line %s:\n%s\n%s" %
                    (v.problem_mark.line + 1, snip, v.problem)
                )
            else:
                raise KeyBindingError("Could not parse key bindings.")

            

Reported by Pylint.

                                  (v.problem_mark.line + 1, snip, v.problem)
                )
            else:
                raise KeyBindingError("Could not parse key bindings.")
        if not data:
            return []
        if not isinstance(data, list):
            raise KeyBindingError("Invalid keybinding config - expected a list of keys")


            

Reported by Pylint.

              import io
import struct

from mitmproxy.io import tnetstring

MAXINT = 2 ** (struct.Struct('i').size * 8 - 1) - 1

FORMAT_EXAMPLES = {
    b'0:}': {},

            

Reported by Pylint.

              }


def get_random_object(random=random, depth=0):
    """Generate a random serializable object."""
    #  The probability of generating a scalar value increases as the depth increase.
    #  This ensures that we bottom out eventually.
    if random.randint(depth, 10) <= 4:
        what = random.randint(0, 1)

            

Reported by Pylint.

              
def suite():
    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    suite.addTest(loader.loadTestsFromTestCase(Test_Format))
    suite.addTest(loader.loadTestsFromTestCase(Test_FileLoading))
    return suite

            

Reported by Pylint.

              import unittest
import random
import math
import io
import struct

from mitmproxy.io import tnetstring

MAXINT = 2 ** (struct.Struct('i').size * 8 - 1) - 1

            

Reported by Pylint.

                  b'10:\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00,': b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
    b'24:5:12345#5:67890#5:xxxxx,]': [12345, 67890, b'xxxxx'],
    b'18:3:0.1^3:0.2^3:0.3^]': [0.1, 0.2, 0.3],
    b'243:238:233:228:223:218:213:208:203:198:193:188:183:178:173:168:163:158:153:148:143:138:133:128:123:118:113:108:103:99:95:91:87:83:79:75:71:67:63:59:55:51:47:43:39:35:31:27:23:19:15:11:hello-there,]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]': [[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[[b'hello-there']]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]  # noqa
}


def get_random_object(random=random, depth=0):
    """Generate a random serializable object."""

            

Reported by Pylint.

              }


def get_random_object(random=random, depth=0):
    """Generate a random serializable object."""
    #  The probability of generating a scalar value increases as the depth increase.
    #  This ensures that we bottom out eventually.
    if random.randint(depth, 10) <= 4:
        what = random.randint(0, 1)

            

Reported by Pylint.

              }


def get_random_object(random=random, depth=0):
    """Generate a random serializable object."""
    #  The probability of generating a scalar value increases as the depth increase.
    #  This ensures that we bottom out eventually.
    if random.randint(depth, 10) <= 4:
        what = random.randint(0, 1)

            

Reported by Pylint.

                  """Generate a random serializable object."""
    #  The probability of generating a scalar value increases as the depth increase.
    #  This ensures that we bottom out eventually.
    if random.randint(depth, 10) <= 4:
        what = random.randint(0, 1)
        if what == 0:
            n = random.randint(0, 10)
            l = []
            for _ in range(n):

            

Reported by Bandit.

                  #  The probability of generating a scalar value increases as the depth increase.
    #  This ensures that we bottom out eventually.
    if random.randint(depth, 10) <= 4:
        what = random.randint(0, 1)
        if what == 0:
            n = random.randint(0, 10)
            l = []
            for _ in range(n):
                l.append(get_random_object(random, depth + 1))

            

Reported by Bandit.

                  if random.randint(depth, 10) <= 4:
        what = random.randint(0, 1)
        if what == 0:
            n = random.randint(0, 10)
            l = []
            for _ in range(n):
                l.append(get_random_object(random, depth + 1))
            return l
        if what == 1:

            

Reported by Pylint.

              from mitmproxy.utils import human
from mitmproxy.utils import strutils
from mitmproxy.websocket import WebSocketMessage, WebSocketData
from wsproto.frame_protocol import CloseReason


def indent(n: int, text: str) -> str:
    l = str(text).strip().splitlines()
    pad = " " * n

            

Reported by Pylint.

                  def _echo_message(
        self,
        message: Union[http.Message, TCPMessage, WebSocketMessage],
        flow: Union[http.HTTPFlow, TCPFlow]
    ):
        _, lines, error = contentviews.get_message_content_view(
            ctx.options.dumper_default_contentview,
            message,
            flow

            

Reported by Pylint.

                      if ctx.options.flow_detail >= 2:
            self.echo("")

    def _echo_request_line(self, flow: http.HTTPFlow) -> None:
        if flow.client_conn:
            client = click.style(
                strutils.escape_control_characters(
                    human.format_address(flow.client_conn.peername)
                )

            

Reported by Pylint.

              
        self.echo(f"{client}: {method} {url}{http_version}")

    def _echo_response_line(self, flow: http.HTTPFlow) -> None:
        if flow.is_replay == "response":
            replay_str = "[replay]"
            replay = click.style(replay_str, fg="yellow", bold=True)
        else:
            replay_str = ""

            

Reported by Pylint.

              import itertools
import shutil
from typing import IO, Optional, Union

import click

from mitmproxy import contentviews
from mitmproxy import ctx
from mitmproxy import exceptions

            

Reported by Pylint.

              from wsproto.frame_protocol import CloseReason


def indent(n: int, text: str) -> str:
    l = str(text).strip().splitlines()
    pad = " " * n
    return "\n".join(pad + i for i in l)



            

Reported by Pylint.

              from wsproto.frame_protocol import CloseReason


def indent(n: int, text: str) -> str:
    l = str(text).strip().splitlines()
    pad = " " * n
    return "\n".join(pad + i for i in l)



            

Reported by Pylint.

              

def indent(n: int, text: str) -> str:
    l = str(text).strip().splitlines()
    pad = " " * n
    return "\n".join(pad + i for i in l)


def colorful(line, styles):

            

Reported by Pylint.

                  return "\n".join(pad + i for i in l)


def colorful(line, styles):
    yield "    "  # we can already indent here
    for (style, text) in line:
        yield click.style(text, **styles.get(style, {}))



            

Reported by Pylint.

                      yield click.style(text, **styles.get(style, {}))


class Dumper:
    def __init__(self, outfile=None):
        self.filter: Optional[flowfilter.TFilter] = None
        self.outfp: Optional[IO] = outfile

    def load(self, loader):

            

Reported by Pylint.

              import time
import pytest
from mitmproxy.utils import human


def test_format_timestamp():
    assert human.format_timestamp(time.time())



            

Reported by Pylint.

                  assert human.format_address(("127.0.0.1", "54010")) == "127.0.0.1:54010"
    assert human.format_address(("example.com", "54010")) == "example.com:54010"
    assert human.format_address(("::", "8080")) == "*:8080"
    assert human.format_address(("0.0.0.0", "8080")) == "*:8080"
    assert human.format_address(None) == "<no address>"

            

Reported by Bandit.

              import time
import pytest
from mitmproxy.utils import human


def test_format_timestamp():
    assert human.format_timestamp(time.time())



            

Reported by Pylint.

              from mitmproxy.utils import human


def test_format_timestamp():
    assert human.format_timestamp(time.time())


def test_format_timestamp_with_milli():
    assert human.format_timestamp_with_milli(time.time())

            

Reported by Pylint.

              

def test_format_timestamp():
    assert human.format_timestamp(time.time())


def test_format_timestamp_with_milli():
    assert human.format_timestamp_with_milli(time.time())


            

Reported by Bandit.

                  assert human.format_timestamp(time.time())


def test_format_timestamp_with_milli():
    assert human.format_timestamp_with_milli(time.time())


def test_parse_size():
    assert human.parse_size("0") == 0

            

Reported by Pylint.

              

def test_format_timestamp_with_milli():
    assert human.format_timestamp_with_milli(time.time())


def test_parse_size():
    assert human.parse_size("0") == 0
    assert human.parse_size("0b") == 0

            

Reported by Bandit.

                  assert human.format_timestamp_with_milli(time.time())


def test_parse_size():
    assert human.parse_size("0") == 0
    assert human.parse_size("0b") == 0
    assert human.parse_size("1") == 1
    assert human.parse_size("1k") == 1024
    assert human.parse_size("1m") == 1024**2

            

Reported by Pylint.

              

def test_parse_size():
    assert human.parse_size("0") == 0
    assert human.parse_size("0b") == 0
    assert human.parse_size("1") == 1
    assert human.parse_size("1k") == 1024
    assert human.parse_size("1m") == 1024**2
    assert human.parse_size("1g") == 1024**3

            

Reported by Bandit.

              
def test_parse_size():
    assert human.parse_size("0") == 0
    assert human.parse_size("0b") == 0
    assert human.parse_size("1") == 1
    assert human.parse_size("1k") == 1024
    assert human.parse_size("1m") == 1024**2
    assert human.parse_size("1g") == 1024**3
    with pytest.raises(ValueError):

            

Reported by Bandit.

              import asyncio
from contextlib import asynccontextmanager

import pytest

from mitmproxy import exceptions
from mitmproxy.addons.proxyserver import Proxyserver
from mitmproxy.connection import Address
from mitmproxy.proxy import layers, server_hooks

            

Reported by Pylint.

                  def __init__(self):
        self.flows = []
        self.layers = [
            lambda ctx: layers.modes.HttpProxy(ctx),
            lambda ctx: layers.HttpLayer(ctx, HTTPMode.regular),
            lambda ctx: layers.TCPLayer(ctx),
        ]

    def request(self, f):

            

Reported by Pylint.

                      self.layers = [
            lambda ctx: layers.modes.HttpProxy(ctx),
            lambda ctx: layers.HttpLayer(ctx, HTTPMode.regular),
            lambda ctx: layers.TCPLayer(ctx),
        ]

    def request(self, f):
        self.flows.append(f)


            

Reported by Pylint.

                          assert state.flows[0].response.status_code == 204

            # Waiting here until everything is really torn down... takes some effort.
            conn_handler = list(ps._connections.values())[0]
            client_handler = conn_handler.transports[conn_handler.client].handler
            writer.close()
            await writer.wait_closed()
            try:
                await client_handler

            

Reported by Pylint.

              import asyncio
from contextlib import asynccontextmanager

import pytest

from mitmproxy import exceptions
from mitmproxy.addons.proxyserver import Proxyserver
from mitmproxy.connection import Address
from mitmproxy.proxy import layers, server_hooks

            

Reported by Pylint.

              from mitmproxy.test.tflow import tclient_conn, tserver_conn


class HelperAddon:
    def __init__(self):
        self.flows = []
        self.layers = [
            lambda ctx: layers.modes.HttpProxy(ctx),
            lambda ctx: layers.HttpLayer(ctx, HTTPMode.regular),

            

Reported by Pylint.

                          lambda ctx: layers.TCPLayer(ctx),
        ]

    def request(self, f):
        self.flows.append(f)

    def tcp_start(self, f):
        self.flows.append(f)


            

Reported by Pylint.

                          lambda ctx: layers.TCPLayer(ctx),
        ]

    def request(self, f):
        self.flows.append(f)

    def tcp_start(self, f):
        self.flows.append(f)


            

Reported by Pylint.

                  def request(self, f):
        self.flows.append(f)

    def tcp_start(self, f):
        self.flows.append(f)

    def next_layer(self, nl):
        nl.layer = self.layers.pop(0)(nl.context)


            

Reported by Pylint.

                  def request(self, f):
        self.flows.append(f)

    def tcp_start(self, f):
        self.flows.append(f)

    def next_layer(self, nl):
        nl.layer = self.layers.pop(0)(nl.context)


            

Reported by Pylint.

                          cmd.append(name)
        with self.uistopped():
            try:
                subprocess.call(cmd, shell=shell)
            except:
                signals.status_message.send(
                    message="Can't start external viewer: %s" % " ".join(c)
                )
        # add a small delay before deletion so that the file is not removed before being loaded by the viewer

            

Reported by Bandit.

              import contextlib
import threading

import urwid

from mitmproxy import addons
from mitmproxy import master
from mitmproxy import log
from mitmproxy.addons import intercept

            

Reported by Pylint.

                          keymap.KeymapConfig(),
        )

        def sigint_handler(*args, **kwargs):
            self.prompt_for_exit()

        signal.signal(signal.SIGINT, sigint_handler)

        self.window = None

            

Reported by Pylint.

                          keymap.KeymapConfig(),
        )

        def sigint_handler(*args, **kwargs):
            self.prompt_for_exit()

        signal.signal(signal.SIGINT, sigint_handler)

        self.window = None

            

Reported by Pylint.

                      super().__setattr__(name, value)
        signals.update_settings.send(self)

    def options_error(self, opts, exc):
        signals.status_message.send(
            message=str(exc),
            expire=1
        )


            

Reported by Pylint.

                          callback = self.quit,
        )

    def sig_add_log(self, event_store, entry: log.LogEntry):
        if log.log_tier(self.options.console_eventlog_verbosity) < log.log_tier(entry.level):
            return
        if entry.level in ("error", "warn", "alert"):
            signals.status_message.send(
                message = (

            

Reported by Pylint.

                              expire=5
            )

    def sig_call_in(self, sender, seconds, callback, args=()):
        def cb(*_):
            return callback(*args)
        self.loop.set_alarm_in(seconds, cb)

    @contextlib.contextmanager

            

Reported by Pylint.

                      with self.uistopped():
            try:
                subprocess.call(cmd)
            except:
                signals.status_message.send(
                    message="Can't start editor: %s" % c
                )
            else:
                with open(name, "r" if text else "rb") as f:

            

Reported by Pylint.

                      with self.uistopped():
            try:
                subprocess.call(cmd, shell=shell)
            except:
                signals.status_message.send(
                    message="Can't start external viewer: %s" % " ".join(c)
                )
        # add a small delay before deletion so that the file is not removed before being loaded by the viewer
        t = threading.Timer(1.0, os.unlink, args=[name])

            

Reported by Pylint.

                      t = threading.Timer(1.0, os.unlink, args=[name])
        t.start()

    def set_palette(self, opts, updated):
        self.ui.register_palette(
            palettes.palettes[opts.console_palette].palette(
                opts.console_palette_transparent
            )
        )

            

Reported by Pylint.

                          f.resume()
        ctx.master.addons.trigger(hooks.UpdateHook(intercepted))

    # FIXME: this will become view.mark later
    @command.command("flow.mark")
    def mark(self, flows: typing.Sequence[flow.Flow], marker: mitmproxy.types.Marker) -> None:
        """
            Mark flows.
        """

            

Reported by Pylint.

                      """
        updated = []
        if marker not in emoji.emoji:
            raise exceptions.CommandError(f"invalid marker value")

        for i in flows:
            i.marked = marker
            updated.append(i)
        ctx.master.addons.trigger(hooks.UpdateHook(updated))

            

Reported by Pylint.

                          updated.append(i)
        ctx.master.addons.trigger(hooks.UpdateHook(updated))

    # FIXME: this will become view.mark.toggle later
    @command.command("flow.mark.toggle")
    def mark_toggle(self, flows: typing.Sequence[flow.Flow]) -> None:
        """
            Toggle mark for flows.
        """

            

Reported by Pylint.

                      ctx.log.alert("Killed %s flows." % len(updated))
        ctx.master.addons.trigger(hooks.UpdateHook(updated))

    # FIXME: this will become view.revert later
    @command.command("flow.revert")
    def revert(self, flows: typing.Sequence[flow.Flow]) -> None:
        """
            Revert flow changes.
        """

            

Reported by Pylint.

                                          "URL {} is invalid: {}".format(repr(val), e)
                        ) from e
                else:
                    self.rupdate = False

            resp = getattr(f, "response", None)
            supdate = True
            if resp:
                if attr == "status_code":

            

Reported by Pylint.

              import typing

import os

from mitmproxy.utils import emoji
from mitmproxy import ctx, hooks
from mitmproxy import exceptions
from mitmproxy import command
from mitmproxy import flow

            

Reported by Pylint.

              LISTEN_PORT = 8080


class Core:
    def configure(self, updated):
        opts = ctx.options
        if opts.add_upstream_certs_to_client_chain and not opts.upstream_cert:
            raise exceptions.OptionsError(
                "add_upstream_certs_to_client_chain requires the upstream_cert option to be enabled."

            

Reported by Pylint.

              

class Core:
    def configure(self, updated):
        opts = ctx.options
        if opts.add_upstream_certs_to_client_chain and not opts.upstream_cert:
            raise exceptions.OptionsError(
                "add_upstream_certs_to_client_chain requires the upstream_cert option to be enabled."
            )

            

Reported by Pylint.

              

class Core:
    def configure(self, updated):
        opts = ctx.options
        if opts.add_upstream_certs_to_client_chain and not opts.upstream_cert:
            raise exceptions.OptionsError(
                "add_upstream_certs_to_client_chain requires the upstream_cert option to be enabled."
            )

            

Reported by Pylint.

                      opts = ctx.options
        if opts.add_upstream_certs_to_client_chain and not opts.upstream_cert:
            raise exceptions.OptionsError(
                "add_upstream_certs_to_client_chain requires the upstream_cert option to be enabled."
            )
        if "mode" in updated:
            mode = opts.mode
            if mode.startswith("reverse:") or mode.startswith("upstream:"):
                try:

            

Reported by Pylint.

              from pathlib import Path
from unittest import mock

from mitmproxy.connections import ServerConnection
from mitmproxy.exceptions import HttpSyntaxException
from mitmproxy.test import tflow
from mitmproxy.test import tutils
import multiprocessing


            

Reported by Pylint.

              from unittest import mock

from mitmproxy.connections import ServerConnection
from mitmproxy.exceptions import HttpSyntaxException
from mitmproxy.test import tflow
from mitmproxy.test import tutils
import multiprocessing

from examples.contrib.webscanner_helper.watchdog import WatchdogAddon, logger

            

Reported by Pylint.

              
from mitmproxy.connections import ServerConnection
from mitmproxy.exceptions import HttpSyntaxException
from mitmproxy.test import tflow
from mitmproxy.test import tutils
import multiprocessing

from examples.contrib.webscanner_helper.watchdog import WatchdogAddon, logger


            

Reported by Pylint.

              from mitmproxy.connections import ServerConnection
from mitmproxy.exceptions import HttpSyntaxException
from mitmproxy.test import tflow
from mitmproxy.test import tutils
import multiprocessing

from examples.contrib.webscanner_helper.watchdog import WatchdogAddon, logger



            

Reported by Pylint.

              from mitmproxy.test import tutils
import multiprocessing

from examples.contrib.webscanner_helper.watchdog import WatchdogAddon, logger


class TestWatchdog:

    def test_init_file(self, tmpdir):

            

Reported by Pylint.

              import time
from pathlib import Path
from unittest import mock

from mitmproxy.connections import ServerConnection
from mitmproxy.exceptions import HttpSyntaxException
from mitmproxy.test import tflow
from mitmproxy.test import tutils
import multiprocessing

            

Reported by Pylint.

              from mitmproxy.exceptions import HttpSyntaxException
from mitmproxy.test import tflow
from mitmproxy.test import tutils
import multiprocessing

from examples.contrib.webscanner_helper.watchdog import WatchdogAddon, logger


class TestWatchdog:

            

Reported by Pylint.

              from examples.contrib.webscanner_helper.watchdog import WatchdogAddon, logger


class TestWatchdog:

    def test_init_file(self, tmpdir):
        tmpfile = tmpdir.join("tmpfile")
        with open(tmpfile, "w") as tfile:
            tfile.write("")

            

Reported by Pylint.

              
class TestWatchdog:

    def test_init_file(self, tmpdir):
        tmpfile = tmpdir.join("tmpfile")
        with open(tmpfile, "w") as tfile:
            tfile.write("")
        event = multiprocessing.Event()
        try:

            

Reported by Pylint.

              
class TestWatchdog:

    def test_init_file(self, tmpdir):
        tmpfile = tmpdir.join("tmpfile")
        with open(tmpfile, "w") as tfile:
            tfile.write("")
        event = multiprocessing.Event()
        try:

            

Reported by Pylint.