from dataclasses import dataclass
from typing import Optional

from mitmproxy import flow, tcp
from mitmproxy.proxy import commands, events, layer
from mitmproxy.proxy.commands import StartHook
from mitmproxy.connection import ConnectionState, Connection
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import MessageInjected

            

Reported by Pylint.

                  flow: tcp.TCPFlow


class TcpMessageInjected(MessageInjected[tcp.TCPMessage]):
    """
    The user has injected a custom TCP message.
    """



            

Reported by Pylint.

                          self.flow = tcp.TCPFlow(self.context.client, self.context.server, True)

    @expect(events.Start)
    def start(self, _) -> layer.CommandGenerator[None]:
        if self.flow:
            yield TcpStartHook(self.flow)

        if self.context.server.timestamp_start is None:
            err = yield commands.OpenConnection(self.context.server)

            

Reported by Pylint.

                  _handle_event = start

    @expect(events.DataReceived, events.ConnectionClosed, TcpMessageInjected)
    def relay_messages(self, event: events.Event) -> layer.CommandGenerator[None]:

        if isinstance(event, TcpMessageInjected):
            # we just spoof that we received data here and then process that regularly.
            event = events.DataReceived(
                self.context.client if event.message.from_client else self.context.server,

            

Reported by Pylint.

                  _handle_event = start

    @expect(events.DataReceived, events.ConnectionClosed, TcpMessageInjected)
    def relay_messages(self, event: events.Event) -> layer.CommandGenerator[None]:

        if isinstance(event, TcpMessageInjected):
            # we just spoof that we received data here and then process that regularly.
            event = events.DataReceived(
                self.context.client if event.message.from_client else self.context.server,

            

Reported by Pylint.

                              event.message.content,
            )

        assert isinstance(event, events.ConnectionEvent)

        from_client = event.connection == self.context.client
        send_to: Connection
        if from_client:
            send_to = self.context.server

            

Reported by Bandit.

                          raise AssertionError(f"Unexpected event: {event}")

    @expect(events.DataReceived, events.ConnectionClosed, TcpMessageInjected)
    def done(self, _) -> layer.CommandGenerator[None]:
        yield from ()

            

Reported by Pylint.

                          raise AssertionError(f"Unexpected event: {event}")

    @expect(events.DataReceived, events.ConnectionClosed, TcpMessageInjected)
    def done(self, _) -> layer.CommandGenerator[None]:
        yield from ()

            

Reported by Pylint.

              import urwid
from mitmproxy.tools.console.grideditor import base
from mitmproxy.tools.console import signals
from mitmproxy.net.http import cookies


class Column(base.Column):
    def __init__(self, heading, subeditor):
        super().__init__(heading)

            

Reported by Pylint.

              import urwid
from mitmproxy.tools.console.grideditor import base
from mitmproxy.tools.console import signals
from mitmproxy.net.http import cookies


class Column(base.Column):
    def __init__(self, heading, subeditor):
        super().__init__(heading)

            

Reported by Pylint.

              from mitmproxy.net.http import cookies


class Column(base.Column):
    def __init__(self, heading, subeditor):
        super().__init__(heading)
        self.subeditor = subeditor

    def Edit(self, data):

            

Reported by Pylint.

                  def blank(self):
        return []

    def keypress(self, key, editor):
        if key in "rRe":
            signals.status_message.send(
                self,
                message="Press enter to edit this field.",
                expire=1000

            

Reported by Pylint.

                      return []

    def keypress(self, key, editor):
        if key in "rRe":
            signals.status_message.send(
                self,
                message="Press enter to edit this field.",
                expire=1000
            )

            

Reported by Pylint.

                          return key


class Display(base.Cell):
    def __init__(self, data):
        p = cookies._format_pairs(data, sep="\n")
        w = urwid.Text(p)
        super().__init__(w)


            

Reported by Pylint.

              
class Display(base.Cell):
    def __init__(self, data):
        p = cookies._format_pairs(data, sep="\n")
        w = urwid.Text(p)
        super().__init__(w)

    def get_data(self):
        pass

            

Reported by Pylint.

              class Display(base.Cell):
    def __init__(self, data):
        p = cookies._format_pairs(data, sep="\n")
        w = urwid.Text(p)
        super().__init__(w)

    def get_data(self):
        pass

            

Reported by Pylint.

              import ipaddress
from mitmproxy import ctx


class Block:
    def load(self, loader):
        loader.add_option(
            "block_global", bool, True,
            """

            

Reported by Pylint.

              from mitmproxy import ctx


class Block:
    def load(self, loader):
        loader.add_option(
            "block_global", bool, True,
            """
            Block connections from public IP addresses.

            

Reported by Pylint.

              

class Block:
    def load(self, loader):
        loader.add_option(
            "block_global", bool, True,
            """
            Block connections from public IP addresses.
            """

            

Reported by Pylint.

              

class Block:
    def load(self, loader):
        loader.add_option(
            "block_global", bool, True,
            """
            Block connections from public IP addresses.
            """

            

Reported by Pylint.

                          """
        )

    def client_connected(self, client):
        parts = client.peername[0].rsplit("%", 1)
        address = ipaddress.ip_address(parts[0])
        if isinstance(address, ipaddress.IPv6Address):
            address = address.ipv4_mapped or address


            

Reported by Pylint.

                          """
        )

    def client_connected(self, client):
        parts = client.peername[0].rsplit("%", 1)
        address = ipaddress.ip_address(parts[0])
        if isinstance(address, ipaddress.IPv6Address):
            address = address.ipv4_mapped or address


            

Reported by Pylint.

                          return

        if ctx.options.block_private and address.is_private:
            ctx.log.warn(f"Client connection from {client.peername[0]} killed by block_private option.")
            client.error = "Connection killed by block_private."

        if ctx.options.block_global and address.is_global:
            ctx.log.warn(f"Client connection from {client.peername[0]} killed by block_global option.")
            client.error = "Connection killed by block_global."

            

Reported by Pylint.

                          client.error = "Connection killed by block_private."

        if ctx.options.block_global and address.is_global:
            ctx.log.warn(f"Client connection from {client.peername[0]} killed by block_global option.")
            client.error = "Connection killed by block_global."

            

Reported by Pylint.

              from mitmproxy import contentviews
from . import base


class ViewAuto(base.View):
    name = "Auto"

    def __call__(self, data, **metadata):
        # TODO: The auto view has little justification now that views implement render_priority,

            

Reported by Pylint.

                  name = "Auto"

    def __call__(self, data, **metadata):
        # TODO: The auto view has little justification now that views implement render_priority,
        # but we keep it around for now to not touch more parts.
        priority, view = max(
            (v.render_priority(data, **metadata), v)
            for v in contentviews.views
        )

            

Reported by Pylint.

                          return "No content", []
        return view(data, **metadata)

    def render_priority(self, data: bytes, **metadata) -> float:
        return -1  # don't recurse.

            

Reported by Pylint.

                          return "No content", []
        return view(data, **metadata)

    def render_priority(self, data: bytes, **metadata) -> float:
        return -1  # don't recurse.

            

Reported by Pylint.

              from mitmproxy import contentviews
from . import base


class ViewAuto(base.View):
    name = "Auto"

    def __call__(self, data, **metadata):
        # TODO: The auto view has little justification now that views implement render_priority,

            

Reported by Pylint.

              from . import base


class ViewAuto(base.View):
    name = "Auto"

    def __call__(self, data, **metadata):
        # TODO: The auto view has little justification now that views implement render_priority,
        # but we keep it around for now to not touch more parts.

            

Reported by Pylint.

                          return "No content", []
        return view(data, **metadata)

    def render_priority(self, data: bytes, **metadata) -> float:
        return -1  # don't recurse.

            

Reported by Pylint.

                          return "No content", []
        return view(data, **metadata)

    def render_priority(self, data: bytes, **metadata) -> float:
        return -1  # don't recurse.

            

Reported by Pylint.

              import subprocess

from . import pf

"""
    Doing this the "right" way by using DIOCNATLOOK on the pf device turns out
    to be a pain. Apple has made a number of modifications to the data
    structures returned, and compiling userspace tools to test and work with
    this turns out to be a pain in the ass. Parsing pfctl output is short,

            

Reported by Pylint.

              
from . import pf

"""
    Doing this the "right" way by using DIOCNATLOOK on the pf device turns out
    to be a pain. Apple has made a number of modifications to the data
    structures returned, and compiling userspace tools to test and work with
    this turns out to be a pain in the ass. Parsing pfctl output is short,
    simple, and works.

            

Reported by Pylint.

                      if "sudo: a password is required" in e.output.decode(errors="replace"):
            insufficient_priv = True
        else:
            raise RuntimeError("Error getting pfctl state: " + repr(e))
    else:
        insufficient_priv = "sudo: a password is required" in stxt.decode(errors="replace")

    if insufficient_priv:
        raise RuntimeError(

            

Reported by Pylint.

              import subprocess

from . import pf

"""
    Doing this the "right" way by using DIOCNATLOOK on the pf device turns out
    to be a pain. Apple has made a number of modifications to the data
    structures returned, and compiling userspace tools to test and work with
    this turns out to be a pain in the ass. Parsing pfctl output is short,

            

Reported by Pylint.

              import subprocess

from . import pf

"""
    Doing this the "right" way by using DIOCNATLOOK on the pf device turns out
    to be a pain. Apple has made a number of modifications to the data
    structures returned, and compiling userspace tools to test and work with
    this turns out to be a pain in the ass. Parsing pfctl output is short,

            

Reported by Bandit.

              STATECMD = ("sudo", "-n", "/sbin/pfctl", "-s", "state")


def original_addr(csock):
    peer = csock.getpeername()
    try:
        stxt = subprocess.check_output(STATECMD, stderr=subprocess.STDOUT)
    except subprocess.CalledProcessError as e:
        if "sudo: a password is required" in e.output.decode(errors="replace"):

            

Reported by Pylint.

              def original_addr(csock):
    peer = csock.getpeername()
    try:
        stxt = subprocess.check_output(STATECMD, stderr=subprocess.STDOUT)
    except subprocess.CalledProcessError as e:
        if "sudo: a password is required" in e.output.decode(errors="replace"):
            insufficient_priv = True
        else:
            raise RuntimeError("Error getting pfctl state: " + repr(e))

            

Reported by Bandit.

                  peer = csock.getpeername()
    try:
        stxt = subprocess.check_output(STATECMD, stderr=subprocess.STDOUT)
    except subprocess.CalledProcessError as e:
        if "sudo: a password is required" in e.output.decode(errors="replace"):
            insufficient_priv = True
        else:
            raise RuntimeError("Error getting pfctl state: " + repr(e))
    else:

            

Reported by Pylint.

              import pytest

import mitmproxy.types
from mitmproxy import command
from mitmproxy import ctx
from mitmproxy.test.tflow import tflow
from mitmproxy.tools.console import defaultkeys
from mitmproxy.tools.console import keymap
from mitmproxy.tools.console import master

            

Reported by Pylint.

              import pytest

import mitmproxy.types
from mitmproxy import command
from mitmproxy import ctx
from mitmproxy.test.tflow import tflow
from mitmproxy.tools.console import defaultkeys
from mitmproxy.tools.console import keymap
from mitmproxy.tools.console import master

            

Reported by Pylint.

              

@pytest.mark.asyncio
async def test_commands_exist():
    command_manager = command.CommandManager(ctx)

    km = keymap.Keymap(None)
    defaultkeys.map(km)
    assert km.bindings

            

Reported by Pylint.

              async def test_commands_exist():
    command_manager = command.CommandManager(ctx)

    km = keymap.Keymap(None)
    defaultkeys.map(km)
    assert km.bindings
    m = master.ConsoleMaster(None)
    await m.load_flow(tflow())


            

Reported by Pylint.

              
    km = keymap.Keymap(None)
    defaultkeys.map(km)
    assert km.bindings
    m = master.ConsoleMaster(None)
    await m.load_flow(tflow())

    for binding in km.bindings:
        try:

            

Reported by Bandit.

                  km = keymap.Keymap(None)
    defaultkeys.map(km)
    assert km.bindings
    m = master.ConsoleMaster(None)
    await m.load_flow(tflow())

    for binding in km.bindings:
        try:
            parsed, _ = command_manager.parse_partial(binding.command.strip())

            

Reported by Pylint.

                              if a.type != mitmproxy.types.Space
            ]

            assert cmd in m.commands.commands

            cmd_obj = m.commands.commands[cmd]
            cmd_obj.prepare_args(args)
        except Exception as e:
            raise ValueError(f"Invalid binding: {binding.command}") from e

            

Reported by Bandit.

              
            cmd_obj = m.commands.commands[cmd]
            cmd_obj.prepare_args(args)
        except Exception as e:
            raise ValueError(f"Invalid binding: {binding.command}") from e

            

Reported by Pylint.

              # (this script works best with --anticache)
from bs4 import BeautifulSoup
from mitmproxy import ctx, http


class Injector:
    def load(self, loader):
        loader.add_option(
            "iframe", str, "", "IFrame to inject"

            

Reported by Pylint.

              # (this script works best with --anticache)
from bs4 import BeautifulSoup
from mitmproxy import ctx, http


class Injector:
    def load(self, loader):
        loader.add_option(
            "iframe", str, "", "IFrame to inject"

            

Reported by Pylint.

              # (this script works best with --anticache)
from bs4 import BeautifulSoup
from mitmproxy import ctx, http


class Injector:
    def load(self, loader):
        loader.add_option(
            "iframe", str, "", "IFrame to inject"

            

Reported by Pylint.

              from mitmproxy import ctx, http


class Injector:
    def load(self, loader):
        loader.add_option(
            "iframe", str, "", "IFrame to inject"
        )


            

Reported by Pylint.

              

class Injector:
    def load(self, loader):
        loader.add_option(
            "iframe", str, "", "IFrame to inject"
        )

    def response(self, flow: http.HTTPFlow) -> None:

            

Reported by Pylint.

              

class Injector:
    def load(self, loader):
        loader.add_option(
            "iframe", str, "", "IFrame to inject"
        )

    def response(self, flow: http.HTTPFlow) -> None:

            

Reported by Pylint.

                          "iframe", str, "", "IFrame to inject"
        )

    def response(self, flow: http.HTTPFlow) -> None:
        if ctx.options.iframe:
            html = BeautifulSoup(flow.response.content, "html.parser")
            if html.body:
                iframe = html.new_tag(
                    "iframe",

            

Reported by Pylint.

                          "iframe", str, "", "IFrame to inject"
        )

    def response(self, flow: http.HTTPFlow) -> None:
        if ctx.options.iframe:
            html = BeautifulSoup(flow.response.content, "html.parser")
            if html.body:
                iframe = html.new_tag(
                    "iframe",

            

Reported by Pylint.

              
    mitmdump --rawtcp --tcp-hosts ".*" -s examples/tcp-simple.py
"""
from mitmproxy.utils import strutils
from mitmproxy import ctx
from mitmproxy import tcp


def tcp_message(flow: tcp.TCPFlow):

            

Reported by Pylint.

                  mitmdump --rawtcp --tcp-hosts ".*" -s examples/tcp-simple.py
"""
from mitmproxy.utils import strutils
from mitmproxy import ctx
from mitmproxy import tcp


def tcp_message(flow: tcp.TCPFlow):
    message = flow.messages[-1]

            

Reported by Pylint.

              """
from mitmproxy.utils import strutils
from mitmproxy import ctx
from mitmproxy import tcp


def tcp_message(flow: tcp.TCPFlow):
    message = flow.messages[-1]
    message.content = message.content.replace(b"foo", b"bar")

            

Reported by Pylint.

              """
Process individual messages from a TCP connection.

This script replaces full occurences of "foo" with "bar" and prints various details for each message.
Please note that TCP is stream-based and *not* message-based. mitmproxy splits stream contents into "messages"
as they are received by socket.recv(). This is pretty arbitrary and should not be relied on.
However, it is sometimes good enough as a quick hack.

Example Invocation:

            

Reported by Pylint.

              """
Process individual messages from a TCP connection.

This script replaces full occurences of "foo" with "bar" and prints various details for each message.
Please note that TCP is stream-based and *not* message-based. mitmproxy splits stream contents into "messages"
as they are received by socket.recv(). This is pretty arbitrary and should not be relied on.
However, it is sometimes good enough as a quick hack.

Example Invocation:

            

Reported by Pylint.

              Process individual messages from a TCP connection.

This script replaces full occurences of "foo" with "bar" and prints various details for each message.
Please note that TCP is stream-based and *not* message-based. mitmproxy splits stream contents into "messages"
as they are received by socket.recv(). This is pretty arbitrary and should not be relied on.
However, it is sometimes good enough as a quick hack.

Example Invocation:


            

Reported by Pylint.

              from mitmproxy import tcp


def tcp_message(flow: tcp.TCPFlow):
    message = flow.messages[-1]
    message.content = message.content.replace(b"foo", b"bar")

    ctx.log.info(
        f"tcp_message[from_client={message.from_client}), content={strutils.bytes_to_escaped_str(message.content)}]"

            

Reported by Pylint.

                  message.content = message.content.replace(b"foo", b"bar")

    ctx.log.info(
        f"tcp_message[from_client={message.from_client}), content={strutils.bytes_to_escaped_str(message.content)}]"
    )

            

Reported by Pylint.

              from mitmproxy.utils import typecheck


class StateObject(serializable.Serializable):
    """
    An object with serializable state.

    State attributes can either be serializable types(str, tuple, bool, ...)
    or StateObject instances themselves.

            

Reported by Pylint.

                      try:
            json.dumps(val)
        except TypeError:
            raise ValueError(f"Data not serializable: {val}")
        return val
    else:
        return typeinfo(val)



            

Reported by Pylint.

              import json
import typing

from mitmproxy.coretypes import serializable
from mitmproxy.utils import typecheck


class StateObject(serializable.Serializable):
    """

            

Reported by Pylint.

                          raise RuntimeWarning(f"Unexpected State in __setstate__: {state}")


def _process(typeinfo: typecheck.Type, val: typing.Any, make: bool) -> typing.Any:
    if val is None:
        return None
    elif make and hasattr(typeinfo, "from_state"):
        return typeinfo.from_state(val)
    elif not make and hasattr(val, "get_state"):

            

Reported by Pylint.

              

def _process(typeinfo: typecheck.Type, val: typing.Any, make: bool) -> typing.Any:
    if val is None:
        return None
    elif make and hasattr(typeinfo, "from_state"):
        return typeinfo.from_state(val)
    elif not make and hasattr(val, "get_state"):
        return val.get_state()

            

Reported by Pylint.

              
    typename = str(typeinfo)

    if typename.startswith("typing.List"):
        T = typecheck.sequence_type(typeinfo)
        return [_process(T, x, make) for x in val]
    elif typename.startswith("typing.Tuple"):
        Ts = typecheck.tuple_types(typeinfo)
        if len(Ts) != len(val):

            

Reported by Pylint.

                  typename = str(typeinfo)

    if typename.startswith("typing.List"):
        T = typecheck.sequence_type(typeinfo)
        return [_process(T, x, make) for x in val]
    elif typename.startswith("typing.Tuple"):
        Ts = typecheck.tuple_types(typeinfo)
        if len(Ts) != len(val):
            raise ValueError(f"Invalid data. Expected {Ts}, got {val}.")

            

Reported by Pylint.

                      T = typecheck.sequence_type(typeinfo)
        return [_process(T, x, make) for x in val]
    elif typename.startswith("typing.Tuple"):
        Ts = typecheck.tuple_types(typeinfo)
        if len(Ts) != len(val):
            raise ValueError(f"Invalid data. Expected {Ts}, got {val}.")
        return tuple(
            _process(T, x, make) for T, x in zip(Ts, val)
        )

            

Reported by Pylint.

              import mitmproxy


class DisableH2C:

    """
    We currently only support HTTP/2 over a TLS connection.

    Some clients try to upgrade a connection from HTTP/1.1 to h2c. We need to

            

Reported by Pylint.

                  by sending the connection preface. We just kill those flows.
    """

    def process_flow(self, f):
        if f.request.headers.get('upgrade', '') == 'h2c':
            mitmproxy.ctx.log.warn("HTTP/2 cleartext connections (h2c upgrade requests) are currently not supported.")
            del f.request.headers['upgrade']
            if 'connection' in f.request.headers:
                del f.request.headers['connection']

            

Reported by Pylint.

                  by sending the connection preface. We just kill those flows.
    """

    def process_flow(self, f):
        if f.request.headers.get('upgrade', '') == 'h2c':
            mitmproxy.ctx.log.warn("HTTP/2 cleartext connections (h2c upgrade requests) are currently not supported.")
            del f.request.headers['upgrade']
            if 'connection' in f.request.headers:
                del f.request.headers['connection']

            

Reported by Pylint.

                  by sending the connection preface. We just kill those flows.
    """

    def process_flow(self, f):
        if f.request.headers.get('upgrade', '') == 'h2c':
            mitmproxy.ctx.log.warn("HTTP/2 cleartext connections (h2c upgrade requests) are currently not supported.")
            del f.request.headers['upgrade']
            if 'connection' in f.request.headers:
                del f.request.headers['connection']

            

Reported by Pylint.

              
    def process_flow(self, f):
        if f.request.headers.get('upgrade', '') == 'h2c':
            mitmproxy.ctx.log.warn("HTTP/2 cleartext connections (h2c upgrade requests) are currently not supported.")
            del f.request.headers['upgrade']
            if 'connection' in f.request.headers:
                del f.request.headers['connection']
            if 'http2-settings' in f.request.headers:
                del f.request.headers['http2-settings']

            

Reported by Pylint.

                      )
        if is_connection_preface:
            f.kill()
            mitmproxy.ctx.log.warn("Initiating HTTP/2 connections with prior knowledge are currently not supported.")

    # Handlers

    def request(self, f):
        self.process_flow(f)

            

Reported by Pylint.

              
    # Handlers

    def request(self, f):
        self.process_flow(f)

            

Reported by Pylint.

              
    # Handlers

    def request(self, f):
        self.process_flow(f)

            

Reported by Pylint.