import typing

from mitmproxy import exceptions


class OptionAddon:
    def load(self, loader):
        loader.add_option(
            name = "optionaddon",

            

Reported by Pylint.

              import typing

from mitmproxy import exceptions


class OptionAddon:
    def load(self, loader):
        loader.add_option(
            name = "optionaddon",

            

Reported by Pylint.

              from mitmproxy import exceptions


class OptionAddon:
    def load(self, loader):
        loader.add_option(
            name = "optionaddon",
            typespec = typing.Optional[int],
            default = None,

            

Reported by Pylint.

              

class OptionAddon:
    def load(self, loader):
        loader.add_option(
            name = "optionaddon",
            typespec = typing.Optional[int],
            default = None,
            help = "Option Addon",

            

Reported by Pylint.

              

class OptionAddon:
    def load(self, loader):
        loader.add_option(
            name = "optionaddon",
            typespec = typing.Optional[int],
            default = None,
            help = "Option Addon",

            

Reported by Pylint.

                          help = "Option Addon",
        )

    def configure(self, updates):
        raise exceptions.OptionsError("Options Error")

addons = [
    OptionAddon()
]

            

Reported by Pylint.

                          help = "Option Addon",
        )

    def configure(self, updates):
        raise exceptions.OptionsError("Options Error")

addons = [
    OptionAddon()
]

            

Reported by Pylint.

              addons = [
    OptionAddon()
]


            

Reported by Pylint.

                          port = port or url.default_port(scheme)
            if not port:
                raise ValueError
            # TODO: we can probably get rid of this check?
            url.parse(target)

        raise_if_http_version_unknown(http_version)
    except ValueError as e:
        raise ValueError(f"Bad HTTP request line: {line!r}") from e

            

Reported by Pylint.

                                  raise ValueError()
                ret.append((name, value))
            except ValueError:
                raise ValueError(f"Invalid header line: {line!r}")
    return Headers(ret)


def read_request_head(lines: List[bytes]) -> Request:
    """

            

Reported by Pylint.

              import re
import time
from typing import List, Tuple, Iterable, Optional

from mitmproxy.http import Request, Headers, Response
from mitmproxy.net.http import url


def get_header_tokens(headers, key):

            

Reported by Pylint.

                  """
    if "connection" in headers:
        tokens = get_header_tokens(headers, "connection")
        if "close" in tokens:
            return True
        elif "keep-alive" in tokens:
            return False

    return http_version not in (

            

Reported by Pylint.

                  )


def expected_http_body_size(
        request: Request,
        response: Optional[Response] = None
) -> Optional[int]:
    """
        Returns:

            

Reported by Pylint.

                  return -1


def raise_if_http_version_unknown(http_version: bytes) -> None:
    if not re.match(br"^HTTP/\d\.\d$", http_version):
        raise ValueError(f"Unknown HTTP version: {http_version!r}")


def _read_request_line(line: bytes) -> Tuple[str, int, bytes, bytes, bytes, bytes, bytes]:

            

Reported by Pylint.

                          url.parse(target)

        raise_if_http_version_unknown(http_version)
    except ValueError as e:
        raise ValueError(f"Bad HTTP request line: {line!r}") from e

    return host, port, method, scheme, authority, path, http_version



            

Reported by Pylint.

                      http_version, status_code_str, reason = parts
        status_code = int(status_code_str)
        raise_if_http_version_unknown(http_version)
    except ValueError as e:
        raise ValueError(f"Bad HTTP response line: {line!r}") from e

    return http_version, status_code, reason



            

Reported by Pylint.

              import pytest
from mitmproxy.coretypes import bidi


def test_bidi():
    b = bidi.BiDi(a=1, b=2)
    assert b.a == 1
    assert b.get_name(1) == "a"
    assert b.get_name(5) is None

            

Reported by Pylint.

              import pytest
from mitmproxy.coretypes import bidi


def test_bidi():
    b = bidi.BiDi(a=1, b=2)
    assert b.a == 1
    assert b.get_name(1) == "a"
    assert b.get_name(5) is None

            

Reported by Pylint.

              from mitmproxy.coretypes import bidi


def test_bidi():
    b = bidi.BiDi(a=1, b=2)
    assert b.a == 1
    assert b.get_name(1) == "a"
    assert b.get_name(5) is None
    with pytest.raises(AttributeError):

            

Reported by Pylint.

              

def test_bidi():
    b = bidi.BiDi(a=1, b=2)
    assert b.a == 1
    assert b.get_name(1) == "a"
    assert b.get_name(5) is None
    with pytest.raises(AttributeError):
        getattr(b, "c")

            

Reported by Pylint.

              
def test_bidi():
    b = bidi.BiDi(a=1, b=2)
    assert b.a == 1
    assert b.get_name(1) == "a"
    assert b.get_name(5) is None
    with pytest.raises(AttributeError):
        getattr(b, "c")
    with pytest.raises(ValueError):

            

Reported by Bandit.

              def test_bidi():
    b = bidi.BiDi(a=1, b=2)
    assert b.a == 1
    assert b.get_name(1) == "a"
    assert b.get_name(5) is None
    with pytest.raises(AttributeError):
        getattr(b, "c")
    with pytest.raises(ValueError):
        bidi.BiDi(one=1, two=1)

            

Reported by Bandit.

                  b = bidi.BiDi(a=1, b=2)
    assert b.a == 1
    assert b.get_name(1) == "a"
    assert b.get_name(5) is None
    with pytest.raises(AttributeError):
        getattr(b, "c")
    with pytest.raises(ValueError):
        bidi.BiDi(one=1, two=1)

            

Reported by Bandit.

              body.
"""

from mitmproxy import http
from mitmproxy.http import Headers


def request(flow: http.HTTPFlow):
    if flow.request.trailers:

            

Reported by Pylint.

              """

from mitmproxy import http
from mitmproxy.http import Headers


def request(flow: http.HTTPFlow):
    if flow.request.trailers:
        print("HTTP Trailers detected! Request contains:", flow.request.trailers)

            

Reported by Pylint.

              """
This script simply prints all received HTTP Trailers.

HTTP requests and responses can container trailing headers which are sent after
the body is fully transmitted. Such trailers need to be announced in the initial
headers by name, so the receiving endpoint can wait and read them after the
body.
"""


            

Reported by Pylint.

              from mitmproxy.http import Headers


def request(flow: http.HTTPFlow):
    if flow.request.trailers:
        print("HTTP Trailers detected! Request contains:", flow.request.trailers)

    if flow.request.path == "/inject_trailers":
        if flow.request.is_http10:

            

Reported by Pylint.

                      print("HTTP Trailers detected! Request contains:", flow.request.trailers)

    if flow.request.path == "/inject_trailers":
        if flow.request.is_http10:
            # HTTP/1.0 doesn't support trailers
            return
        elif flow.request.is_http11:
            if not flow.request.content:
                # Avoid sending a body on GET requests or a 0 byte chunked body with trailers.

            

Reported by Pylint.

                      print("Injected a new request trailer...", flow.request.headers["trailer"])


def response(flow: http.HTTPFlow):
    if flow.response.trailers:
        print("HTTP Trailers detected! Response contains:", flow.response.trailers)

    if flow.request.path == "/inject_trailers":
        if flow.request.is_http10:

            

Reported by Pylint.

                      print("HTTP Trailers detected! Response contains:", flow.response.trailers)

    if flow.request.path == "/inject_trailers":
        if flow.request.is_http10:
            return
        elif flow.request.is_http11:
            if not flow.response.content:
                return
            flow.response.headers["transfer-encoding"] = "chunked"

            

Reported by Pylint.

                  - mitmdump -s custom_next_layer.py
    - curl -x localhost:8080 -k https://example.com
"""
from mitmproxy import ctx
from mitmproxy.proxy import layer, layers


def running():
    # We change the connection strategy to lazy so that next_layer happens before we actually connect upstream.

            

Reported by Pylint.

                  - curl -x localhost:8080 -k https://example.com
"""
from mitmproxy import ctx
from mitmproxy.proxy import layer, layers


def running():
    # We change the connection strategy to lazy so that next_layer happens before we actually connect upstream.
    # Alternatively we could also change the server address in `server_connect`.

            

Reported by Pylint.

              from mitmproxy.proxy import layer, layers


def running():
    # We change the connection strategy to lazy so that next_layer happens before we actually connect upstream.
    # Alternatively we could also change the server address in `server_connect`.
    ctx.options.connection_strategy = "lazy"



            

Reported by Pylint.

              

def running():
    # We change the connection strategy to lazy so that next_layer happens before we actually connect upstream.
    # Alternatively we could also change the server address in `server_connect`.
    ctx.options.connection_strategy = "lazy"


def next_layer(nextlayer: layer.NextLayer):

            

Reported by Pylint.

                  ctx.options.connection_strategy = "lazy"


def next_layer(nextlayer: layer.NextLayer):
    ctx.log(
        f"{nextlayer.context=}\n"
        f"{nextlayer.data_client()[:70]=}\n"
        f"{nextlayer.data_server()[:70]=}\n"
    )

            

Reported by Pylint.

                      # which our example server here does not accept for plaintext connections.
        nextlayer.context.client.alpn = b""

        # We know all layers that come next: First negotiate TLS with the client, then do simple TCP passthrough.
        # Setting only one layer here would also work, in that case next_layer would be called again after TLS establishment.
        nextlayer.layer = layers.ClientTLSLayer(nextlayer.context)
        nextlayer.layer.child_layer = layers.TCPLayer(nextlayer.context)

            

Reported by Pylint.

                      nextlayer.context.client.alpn = b""

        # We know all layers that come next: First negotiate TLS with the client, then do simple TCP passthrough.
        # Setting only one layer here would also work, in that case next_layer would be called again after TLS establishment.
        nextlayer.layer = layers.ClientTLSLayer(nextlayer.context)
        nextlayer.layer.child_layer = layers.TCPLayer(nextlayer.context)

            

Reported by Pylint.

              import json
import requests

from mitmproxy import ctx

FILE_WORKERS = 1
HTTP_WORKERS = 10



            

Reported by Pylint.

              
        self._init_transformations()

        for i in range(FILE_WORKERS if self.outfile else HTTP_WORKERS):
            t = Thread(target=self.worker)
            t.daemon = True
            t.start()

    def response(self, flow):

            

Reported by Pylint.

                      self.auth = None
        self.queue = Queue()

    def done(self):
        self.queue.join()
        if self.outfile:
            self.outfile.close()

    fields = {

            

Reported by Pylint.

                      """
        Recursively convert all list/dict elements of type `bytes` into strings.
        """
        if isinstance(obj, dict):
            return {cls.convert_to_strings(key): cls.convert_to_strings(value)
                    for key, value in obj.items()}
        elif isinstance(obj, list) or isinstance(obj, tuple):
            return [cls.convert_to_strings(element) for element in obj]
        elif isinstance(obj, bytes):

            

Reported by Pylint.

                      if isinstance(obj, dict):
            return {cls.convert_to_strings(key): cls.convert_to_strings(value)
                    for key, value in obj.items()}
        elif isinstance(obj, list) or isinstance(obj, tuple):
            return [cls.convert_to_strings(element) for element in obj]
        elif isinstance(obj, bytes):
            return str(obj)[2:-1]
        return obj


            

Reported by Pylint.

                          return str(obj)[2:-1]
        return obj

    def worker(self):
        while True:
            frame = self.queue.get()
            self.dump(frame)
            self.queue.task_done()


            

Reported by Pylint.

                      self._init_transformations()

        for i in range(FILE_WORKERS if self.outfile else HTTP_WORKERS):
            t = Thread(target=self.worker)
            t.daemon = True
            t.start()

    def response(self, flow):
        """

            

Reported by Pylint.

              

def load(l):
    import pydevd_pycharm
    pydevd_pycharm.settrace("localhost", port=5678, stdoutToServer=True, stderrToServer=True, suspend=False)

            

Reported by Pylint.

              """


def load(l):
    import pydevd_pycharm
    pydevd_pycharm.settrace("localhost", port=5678, stdoutToServer=True, stderrToServer=True, suspend=False)

            

Reported by Pylint.

              """
This script enables remote debugging of the mitmproxy console *UI* with PyCharm.
For general debugging purposes, it is easier to just debug mitmdump within PyCharm.

Usage:
    - pip install pydevd on the mitmproxy machine
    - Open the Run/Debug Configuration dialog box in PyCharm, and select the
      Python Remote Debug configuration type.
    - Debugging works in the way that mitmproxy connects to the debug server

            

Reported by Pylint.

              """


def load(l):
    import pydevd_pycharm
    pydevd_pycharm.settrace("localhost", port=5678, stdoutToServer=True, stderrToServer=True, suspend=False)

            

Reported by Pylint.

              """


def load(l):
    import pydevd_pycharm
    pydevd_pycharm.settrace("localhost", port=5678, stdoutToServer=True, stderrToServer=True, suspend=False)

            

Reported by Pylint.

              

def load(l):
    import pydevd_pycharm
    pydevd_pycharm.settrace("localhost", port=5678, stdoutToServer=True, stderrToServer=True, suspend=False)

            

Reported by Pylint.

              
def load(l):
    import pydevd_pycharm
    pydevd_pycharm.settrace("localhost", port=5678, stdoutToServer=True, stderrToServer=True, suspend=False)

            

Reported by Pylint.

              import urllib.parse
import typing  # noqa

from mitmproxy import http

# set of SSL/TLS capable hosts
secure_hosts: typing.Set[str] = set()



            

Reported by Pylint.

              secure_hosts: typing.Set[str] = set()


def request(flow: http.HTTPFlow) -> None:
    flow.request.headers.pop('If-Modified-Since', None)
    flow.request.headers.pop('Cache-Control', None)

    # do not force https redirection
    flow.request.headers.pop('Upgrade-Insecure-Requests', None)

            

Reported by Pylint.

                      flow.request.port = 443

        # We need to update the request destination to whatever is specified in the host header:
        # Having no TLS Server Name Indication from the client and just an IP address as request.host
        # in transparent mode, TLS server name certificate validation would fail.
        flow.request.host = flow.request.pretty_host


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

            

Reported by Pylint.

                      flow.request.host = flow.request.pretty_host


def response(flow: http.HTTPFlow) -> None:
    assert flow.response
    flow.response.headers.pop('Strict-Transport-Security', None)
    flow.response.headers.pop('Public-Key-Pins', None)

    # strip links in response body

            

Reported by Pylint.

              

def response(flow: http.HTTPFlow) -> None:
    assert flow.response
    flow.response.headers.pop('Strict-Transport-Security', None)
    flow.response.headers.pop('Public-Key-Pins', None)

    # strip links in response body
    flow.response.content = flow.response.content.replace(b'https://', b'http://')

            

Reported by Bandit.

                  flow.response.content = flow.response.content.replace(b'https://', b'http://')

    # strip meta tag upgrade-insecure-requests in response body
    csp_meta_tag_pattern = br'<meta.*http-equiv=["\']Content-Security-Policy[\'"].*upgrade-insecure-requests.*?>'
    flow.response.content = re.sub(csp_meta_tag_pattern, b'', flow.response.content, flags=re.IGNORECASE)

    # strip links in 'Location' header
    if flow.response.headers.get('Location', '').startswith('https://'):
        location = flow.response.headers['Location']

            

Reported by Pylint.

              
    # strip meta tag upgrade-insecure-requests in response body
    csp_meta_tag_pattern = br'<meta.*http-equiv=["\']Content-Security-Policy[\'"].*upgrade-insecure-requests.*?>'
    flow.response.content = re.sub(csp_meta_tag_pattern, b'', flow.response.content, flags=re.IGNORECASE)

    # strip links in 'Location' header
    if flow.response.headers.get('Location', '').startswith('https://'):
        location = flow.response.headers['Location']
        hostname = urllib.parse.urlparse(location).hostname

            

Reported by Pylint.

              Without this script, if the web application under test crashes, mitmproxy will send 502 Bad Gateway responses.
These responses are irritating the web application scanner since they obfuscate the actual problem.
"""
from mitmproxy import http
from mitmproxy.exceptions import HttpSyntaxException


def error(self, flow: http.HTTPFlow):
    """Kills the flow if it has an error different to HTTPSyntaxException.

            

Reported by Pylint.

              These responses are irritating the web application scanner since they obfuscate the actual problem.
"""
from mitmproxy import http
from mitmproxy.exceptions import HttpSyntaxException


def error(self, flow: http.HTTPFlow):
    """Kills the flow if it has an error different to HTTPSyntaxException.
            Sometimes, web scanners generate malformed HTTP syntax on purpose and we do not want to kill these requests.

            

Reported by Pylint.

              from mitmproxy.exceptions import HttpSyntaxException


def error(self, flow: http.HTTPFlow):
    """Kills the flow if it has an error different to HTTPSyntaxException.
            Sometimes, web scanners generate malformed HTTP syntax on purpose and we do not want to kill these requests.
    """
    if flow.error is not None and not isinstance(flow.error, HttpSyntaxException):
        flow.kill()

            

Reported by Pylint.

              """
This script suppresses the 502 Bad Gateway messages, mitmproxy sends if the server is not responsing correctly.
For example, this functionality can be helpful if mitmproxy is used in between a web scanner and a web application.
Without this script, if the web application under test crashes, mitmproxy will send 502 Bad Gateway responses.
These responses are irritating the web application scanner since they obfuscate the actual problem.
"""
from mitmproxy import http
from mitmproxy.exceptions import HttpSyntaxException


            

Reported by Pylint.

              """
This script suppresses the 502 Bad Gateway messages, mitmproxy sends if the server is not responsing correctly.
For example, this functionality can be helpful if mitmproxy is used in between a web scanner and a web application.
Without this script, if the web application under test crashes, mitmproxy will send 502 Bad Gateway responses.
These responses are irritating the web application scanner since they obfuscate the actual problem.
"""
from mitmproxy import http
from mitmproxy.exceptions import HttpSyntaxException


            

Reported by Pylint.

              """
This script suppresses the 502 Bad Gateway messages, mitmproxy sends if the server is not responsing correctly.
For example, this functionality can be helpful if mitmproxy is used in between a web scanner and a web application.
Without this script, if the web application under test crashes, mitmproxy will send 502 Bad Gateway responses.
These responses are irritating the web application scanner since they obfuscate the actual problem.
"""
from mitmproxy import http
from mitmproxy.exceptions import HttpSyntaxException


            

Reported by Pylint.

              
def error(self, flow: http.HTTPFlow):
    """Kills the flow if it has an error different to HTTPSyntaxException.
            Sometimes, web scanners generate malformed HTTP syntax on purpose and we do not want to kill these requests.
    """
    if flow.error is not None and not isinstance(flow.error, HttpSyntaxException):
        flow.kill()

            

Reported by Pylint.

              import shutil
import subprocess
import tempfile
import typing

from mitmproxy import command
from mitmproxy import ctx



            

Reported by Pylint.

              import shutil
import subprocess
import tempfile
import typing

from mitmproxy import command
from mitmproxy import ctx



            

Reported by Bandit.

              from mitmproxy import ctx


def get_chrome_executable() -> typing.Optional[str]:
    for browser in (
            "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
            # https://stackoverflow.com/questions/40674914/google-chrome-path-in-windows-10
            r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
            r"C:\Program Files (x86)\Google\Application\chrome.exe",

            

Reported by Pylint.

                  return None


class Browser:
    browser: typing.List[subprocess.Popen] = []
    tdir: typing.List[tempfile.TemporaryDirectory] = []

    @command.command("browser.start")
    def start(self) -> None:

            

Reported by Pylint.

              
        tdir = tempfile.TemporaryDirectory()
        self.tdir.append(tdir)
        self.browser.append(subprocess.Popen(
            [
                cmd,
                "--user-data-dir=%s" % str(tdir.name),
                "--proxy-server={}:{}".format(
                    ctx.options.listen_host or "127.0.0.1",

            

Reported by Bandit.

                          stderr = subprocess.DEVNULL,
        ))

    def done(self):
        for browser in self.browser:
            browser.kill()
        for tdir in self.tdir:
            tdir.cleanup()
        self.browser = []

            

Reported by Pylint.

                      for tdir in self.tdir:
            tdir.cleanup()
        self.browser = []
        self.tdir = []
            

Reported by Pylint.