import re
import typing
import base64

from mitmproxy import exceptions
from mitmproxy import ctx
from mitmproxy import http
from mitmproxy.utils import strutils


            

Reported by Pylint.

              from mitmproxy.utils import strutils


def parse_upstream_auth(auth: str) -> bytes:
    pattern = re.compile(".+:")
    if pattern.search(auth) is None:
        raise exceptions.OptionsError(
            "Invalid upstream auth specification: %s" % auth
        )

            

Reported by Pylint.

                  """
    auth: typing.Optional[bytes] = None

    def load(self, loader):
        loader.add_option(
            "upstream_auth", typing.Optional[str], None,
            """
            Add HTTP Basic authentication to upstream proxy and reverse proxy
            requests. Format: username:password.

            

Reported by Pylint.

                  """
    auth: typing.Optional[bytes] = None

    def load(self, loader):
        loader.add_option(
            "upstream_auth", typing.Optional[str], None,
            """
            Add HTTP Basic authentication to upstream proxy and reverse proxy
            requests. Format: username:password.

            

Reported by Pylint.

                          """
        )

    def configure(self, updated):
        if "upstream_auth" in updated:
            if ctx.options.upstream_auth is None:
                self.auth = None
            else:
                self.auth = parse_upstream_auth(ctx.options.upstream_auth)

            

Reported by Pylint.

                          else:
                self.auth = parse_upstream_auth(ctx.options.upstream_auth)

    def http_connect_upstream(self, f: http.HTTPFlow):
        if self.auth:
            f.request.headers["Proxy-Authorization"] = self.auth

    def requestheaders(self, f: http.HTTPFlow):
        if self.auth:

            

Reported by Pylint.

                          else:
                self.auth = parse_upstream_auth(ctx.options.upstream_auth)

    def http_connect_upstream(self, f: http.HTTPFlow):
        if self.auth:
            f.request.headers["Proxy-Authorization"] = self.auth

    def requestheaders(self, f: http.HTTPFlow):
        if self.auth:

            

Reported by Pylint.

                      if self.auth:
            f.request.headers["Proxy-Authorization"] = self.auth

    def requestheaders(self, f: http.HTTPFlow):
        if self.auth:
            if ctx.options.mode.startswith("upstream") and f.request.scheme == "http":
                f.request.headers["Proxy-Authorization"] = self.auth
            elif ctx.options.mode.startswith("reverse"):
                f.request.headers["Authorization"] = self.auth

            

Reported by Pylint.

                      if self.auth:
            f.request.headers["Proxy-Authorization"] = self.auth

    def requestheaders(self, f: http.HTTPFlow):
        if self.auth:
            if ctx.options.mode.startswith("upstream") and f.request.scheme == "http":
                f.request.headers["Proxy-Authorization"] = self.auth
            elif ctx.options.mode.startswith("reverse"):
                f.request.headers["Authorization"] = self.auth

            

Reported by Pylint.

                  stream_id: StreamId


class HttpConnection(layer.Layer):
    conn: Connection

    def __init__(self, context: Context, conn: Connection):
        super().__init__(context)
        self.conn = conn

            

Reported by Pylint.

              import html
import textwrap
from dataclasses import dataclass

from mitmproxy import http
from mitmproxy.connection import Connection
from mitmproxy.proxy import commands, events, layer
from mitmproxy.proxy.context import Context


            

Reported by Pylint.

              

@dataclass
class HttpEvent(events.Event):
    # we need stream ids on every event to avoid race conditions
    stream_id: StreamId


class HttpConnection(layer.Layer):

            

Reported by Pylint.

                  stream_id: StreamId


class HttpConnection(layer.Layer):
    conn: Connection

    def __init__(self, context: Context, conn: Connection):
        super().__init__(context)
        self.conn = conn

            

Reported by Pylint.

                      self.conn = conn


class HttpCommand(commands.Command):
    pass


class ReceiveHttp(HttpCommand):
    event: HttpEvent

            

Reported by Pylint.

                      self.conn = conn


class HttpCommand(commands.Command):
    pass


class ReceiveHttp(HttpCommand):
    event: HttpEvent

            

Reported by Pylint.

                  pass


class ReceiveHttp(HttpCommand):
    event: HttpEvent

    def __init__(self, event: HttpEvent):
        self.event = event


            

Reported by Pylint.

                  pass


class ReceiveHttp(HttpCommand):
    event: HttpEvent

    def __init__(self, event: HttpEvent):
        self.event = event


            

Reported by Pylint.

                      return f"Receive({self.event})"


def format_error(status_code: int, message: str) -> bytes:
    reason = http.status_codes.RESPONSES.get(status_code, "Unknown")
    return textwrap.dedent(f"""
    <html>
    <head>
        <title>{status_code} {reason}</title>

            

Reported by Pylint.

              """
import random
import sys
from mitmproxy import io, http
import typing  # noqa


class Writer:
    def __init__(self, path: str) -> None:

            

Reported by Pylint.

              """
Generate a mitmproxy dump file.

This script demonstrates how to generate a mitmproxy dump file,
as it would also be generated by passing `-w` to mitmproxy.
In contrast to `-w`, this gives you full control over which
flows should be saved and also allows you to rotate files or log
to multiple files in parallel.
"""

            

Reported by Pylint.

              import random
import sys
from mitmproxy import io, http
import typing  # noqa


class Writer:
    def __init__(self, path: str) -> None:
        self.f: typing.IO[bytes] = open(path, "wb")

            

Reported by Pylint.

              import typing  # noqa


class Writer:
    def __init__(self, path: str) -> None:
        self.f: typing.IO[bytes] = open(path, "wb")
        self.w = io.FlowWriter(self.f)

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

            

Reported by Pylint.

              
class Writer:
    def __init__(self, path: str) -> None:
        self.f: typing.IO[bytes] = open(path, "wb")
        self.w = io.FlowWriter(self.f)

    def response(self, flow: http.HTTPFlow) -> None:
        if random.choice([True, False]):
            self.w.add(flow)

            

Reported by Pylint.

              class Writer:
    def __init__(self, path: str) -> None:
        self.f: typing.IO[bytes] = open(path, "wb")
        self.w = io.FlowWriter(self.f)

    def response(self, flow: http.HTTPFlow) -> None:
        if random.choice([True, False]):
            self.w.add(flow)


            

Reported by Pylint.

                      self.f: typing.IO[bytes] = open(path, "wb")
        self.w = io.FlowWriter(self.f)

    def response(self, flow: http.HTTPFlow) -> None:
        if random.choice([True, False]):
            self.w.add(flow)

    def done(self):
        self.f.close()

            

Reported by Pylint.

                      self.w = io.FlowWriter(self.f)

    def response(self, flow: http.HTTPFlow) -> None:
        if random.choice([True, False]):
            self.w.add(flow)

    def done(self):
        self.f.close()


            

Reported by Bandit.

                      if random.choice([True, False]):
            self.w.add(flow)

    def done(self):
        self.f.close()


addons = [Writer(sys.argv[1])]

            

Reported by Pylint.

              import webbrowser

from mitmproxy import ctx
from typing import Sequence


class WebAddon:
    def load(self, loader):
        loader.add_option(

            

Reported by Pylint.

              import webbrowser

from mitmproxy import ctx
from typing import Sequence


class WebAddon:
    def load(self, loader):
        loader.add_option(

            

Reported by Pylint.

              from typing import Sequence


class WebAddon:
    def load(self, loader):
        loader.add_option(
            "web_open_browser", bool, True,
            "Start a browser."
        )

            

Reported by Pylint.

              

class WebAddon:
    def load(self, loader):
        loader.add_option(
            "web_open_browser", bool, True,
            "Start a browser."
        )
        loader.add_option(

            

Reported by Pylint.

              

class WebAddon:
    def load(self, loader):
        loader.add_option(
            "web_open_browser", bool, True,
            "Start a browser."
        )
        loader.add_option(

            

Reported by Pylint.

                          "Web UI host."
        )
        loader.add_option(
            "web_columns", Sequence[str], ["tls", "icon", "path", "method", "status", "size", "time"],
            "Columns to show in the flow list"
        )

    def running(self):
        if hasattr(ctx.options, "web_open_browser") and ctx.options.web_open_browser:

            

Reported by Pylint.

                          "Columns to show in the flow list"
        )

    def running(self):
        if hasattr(ctx.options, "web_open_browser") and ctx.options.web_open_browser:
            web_url = f"http://{ctx.options.web_host}:{ctx.options.web_port}/"
            success = open_browser(web_url)
            if not success:
                ctx.log.info(

            

Reported by Pylint.

                          "Columns to show in the flow list"
        )

    def running(self):
        if hasattr(ctx.options, "web_open_browser") and ctx.options.web_open_browser:
            web_url = f"http://{ctx.options.web_host}:{ctx.options.web_port}/"
            success = open_browser(web_url)
            if not success:
                ctx.log.info(

            

Reported by Pylint.

                  )
    for browser in browsers:
        try:
            b = webbrowser.get(browser)
        except webbrowser.Error:
            pass
        else:
            if b.open(url):
                return True

            

Reported by Pylint.

              import os
import shlex
import shutil
import subprocess
import textwrap

ROOT_CA = "trusted-root"
SUBJECT = "example.mitmproxy.org"


            

Reported by Bandit.

              SUBJECT = "example.mitmproxy.org"


def do(args):
    print("> %s" % args)
    args = shlex.split(args)
    output = subprocess.check_output(args)
    return output


            

Reported by Pylint.

              SUBJECT = "example.mitmproxy.org"


def do(args):
    print("> %s" % args)
    args = shlex.split(args)
    output = subprocess.check_output(args)
    return output


            

Reported by Pylint.

              def do(args):
    print("> %s" % args)
    args = shlex.split(args)
    output = subprocess.check_output(args)
    return output


def genrsa(cert: str):
    do(f"openssl genrsa -out {cert}.key 2048")

            

Reported by Bandit.

                  return output


def genrsa(cert: str):
    do(f"openssl genrsa -out {cert}.key 2048")


def sign(cert: str, subject: str):
    with open(f"openssl-{cert}.conf", "w") as f:

            

Reported by Pylint.

                  do(f"openssl genrsa -out {cert}.key 2048")


def sign(cert: str, subject: str):
    with open(f"openssl-{cert}.conf", "w") as f:
        f.write(textwrap.dedent(f"""
        authorityKeyIdentifier=keyid,issuer
        basicConstraints=CA:FALSE
        keyUsage = digitalSignature, keyEncipherment

            

Reported by Pylint.

              

def sign(cert: str, subject: str):
    with open(f"openssl-{cert}.conf", "w") as f:
        f.write(textwrap.dedent(f"""
        authorityKeyIdentifier=keyid,issuer
        basicConstraints=CA:FALSE
        keyUsage = digitalSignature, keyEncipherment
        subjectAltName = DNS:{subject}

            

Reported by Pylint.

                  os.remove(f"openssl-{cert}.conf")


def mkcert(cert, subject):
    genrsa(cert)
    do(f"openssl req -new -nodes -batch "
       f"-key {cert}.key "
       f"-subj /CN={subject}/O=mitmproxy "
       f"-addext \"subjectAltName = DNS:{subject}\" "

            

Reported by Pylint.

                 "-days 7300 "
   "-out trusted-root.crt"
   )
h = do("openssl x509 -hash -noout -in trusted-root.crt").decode("ascii").strip()
shutil.copyfile("trusted-root.crt", f"{h}.0")

# create trusted leaf cert.
mkcert("trusted-leaf", SUBJECT)


            

Reported by Pylint.

              def assemble_request(request):
    if request.data.content is None:
        raise ValueError("Cannot assemble flow with missing content")
    head = assemble_request_head(request)
    body = b"".join(assemble_body(request.data.headers, [request.data.content], request.data.trailers))
    return head + body


def assemble_request_head(request):

            

Reported by Pylint.

              def assemble_request(request):
    if request.data.content is None:
        raise ValueError("Cannot assemble flow with missing content")
    head = assemble_request_head(request)
    body = b"".join(assemble_body(request.data.headers, [request.data.content], request.data.trailers))
    return head + body


def assemble_request_head(request):

            

Reported by Pylint.

                  if request.data.content is None:
        raise ValueError("Cannot assemble flow with missing content")
    head = assemble_request_head(request)
    body = b"".join(assemble_body(request.data.headers, [request.data.content], request.data.trailers))
    return head + body


def assemble_request_head(request):
    first_line = _assemble_request_line(request.data)

            

Reported by Pylint.

                  return head + body


def assemble_request_head(request):
    first_line = _assemble_request_line(request.data)
    headers = _assemble_request_headers(request.data)
    return b"%s\r\n%s\r\n" % (first_line, headers)



            

Reported by Pylint.

                  return b"%s\r\n%s\r\n" % (first_line, headers)


def assemble_response(response):
    if response.data.content is None:
        raise ValueError("Cannot assemble flow with missing content")
    head = assemble_response_head(response)
    body = b"".join(assemble_body(response.data.headers, [response.data.content], response.data.trailers))
    return head + body

            

Reported by Pylint.

                  if response.data.content is None:
        raise ValueError("Cannot assemble flow with missing content")
    head = assemble_response_head(response)
    body = b"".join(assemble_body(response.data.headers, [response.data.content], response.data.trailers))
    return head + body


def assemble_response_head(response):
    first_line = _assemble_response_line(response.data)

            

Reported by Pylint.

                  return head + body


def assemble_response_head(response):
    first_line = _assemble_response_line(response.data)
    headers = _assemble_response_headers(response.data)
    return b"%s\r\n%s\r\n" % (first_line, headers)



            

Reported by Pylint.

                  return b"%s\r\n%s\r\n" % (first_line, headers)


def assemble_body(headers, body_chunks, trailers):
    if "chunked" in headers.get("transfer-encoding", "").lower():
        for chunk in body_chunks:
            if chunk:
                yield b"%x\r\n%s\r\n" % (len(chunk), chunk)
        if trailers:

            

Reported by Pylint.

                  Args:
        request_data (mitmproxy.net.http.request.RequestData)
    """
    if request_data.method.upper() == b"CONNECT":
        return b"%s %s %s" % (
            request_data.method,
            request_data.authority,
            request_data.http_version
        )

            

Reported by Pylint.

              import pytest
from mitmproxy.utils.spec import parse_spec


def test_parse_spec():
    flow_filter, subject, replacement = parse_spec("/foo/bar/voing")
    assert flow_filter.pattern == "foo"
    assert subject == "bar"
    assert replacement == "voing"

            

Reported by Pylint.

              import pytest
from mitmproxy.utils.spec import parse_spec


def test_parse_spec():
    flow_filter, subject, replacement = parse_spec("/foo/bar/voing")
    assert flow_filter.pattern == "foo"
    assert subject == "bar"
    assert replacement == "voing"

            

Reported by Pylint.

              from mitmproxy.utils.spec import parse_spec


def test_parse_spec():
    flow_filter, subject, replacement = parse_spec("/foo/bar/voing")
    assert flow_filter.pattern == "foo"
    assert subject == "bar"
    assert replacement == "voing"


            

Reported by Pylint.

              
def test_parse_spec():
    flow_filter, subject, replacement = parse_spec("/foo/bar/voing")
    assert flow_filter.pattern == "foo"
    assert subject == "bar"
    assert replacement == "voing"

    flow_filter, subject, replacement = parse_spec("/bar/voing")
    assert flow_filter(1) is True

            

Reported by Bandit.

              def test_parse_spec():
    flow_filter, subject, replacement = parse_spec("/foo/bar/voing")
    assert flow_filter.pattern == "foo"
    assert subject == "bar"
    assert replacement == "voing"

    flow_filter, subject, replacement = parse_spec("/bar/voing")
    assert flow_filter(1) is True
    assert subject == "bar"

            

Reported by Bandit.

                  flow_filter, subject, replacement = parse_spec("/foo/bar/voing")
    assert flow_filter.pattern == "foo"
    assert subject == "bar"
    assert replacement == "voing"

    flow_filter, subject, replacement = parse_spec("/bar/voing")
    assert flow_filter(1) is True
    assert subject == "bar"
    assert replacement == "voing"

            

Reported by Bandit.

                  assert replacement == "voing"

    flow_filter, subject, replacement = parse_spec("/bar/voing")
    assert flow_filter(1) is True
    assert subject == "bar"
    assert replacement == "voing"

    with pytest.raises(ValueError, match="Invalid number of parameters"):
        parse_spec("/")

            

Reported by Bandit.

              
    flow_filter, subject, replacement = parse_spec("/bar/voing")
    assert flow_filter(1) is True
    assert subject == "bar"
    assert replacement == "voing"

    with pytest.raises(ValueError, match="Invalid number of parameters"):
        parse_spec("/")


            

Reported by Bandit.

                  flow_filter, subject, replacement = parse_spec("/bar/voing")
    assert flow_filter(1) is True
    assert subject == "bar"
    assert replacement == "voing"

    with pytest.raises(ValueError, match="Invalid number of parameters"):
        parse_spec("/")

    with pytest.raises(ValueError, match="Invalid filter pattern"):

            

Reported by Bandit.

              import sys
import pytest

from mitmproxy.addons import termlog
from mitmproxy import log
from mitmproxy.test import taddons


class TestTermLog:

            

Reported by Pylint.

              import sys
import pytest

from mitmproxy.addons import termlog
from mitmproxy import log
from mitmproxy.test import taddons


class TestTermLog:

            

Reported by Pylint.

              from mitmproxy.test import taddons


class TestTermLog:
    @pytest.mark.usefixtures('capfd')
    @pytest.mark.parametrize('outfile, expected_out, expected_err', [
        (None, ['one', 'three'], ['four']),
        (sys.stdout, ['one', 'three', 'four'], []),
        (sys.stderr, [], ['one', 'three', 'four']),

            

Reported by Pylint.

              from mitmproxy.test import taddons


class TestTermLog:
    @pytest.mark.usefixtures('capfd')
    @pytest.mark.parametrize('outfile, expected_out, expected_err', [
        (None, ['one', 'three'], ['four']),
        (sys.stdout, ['one', 'three', 'four'], []),
        (sys.stderr, [], ['one', 'three', 'four']),

            

Reported by Pylint.

                      (None, ['one', 'three'], ['four']),
        (sys.stdout, ['one', 'three', 'four'], []),
        (sys.stderr, [], ['one', 'three', 'four']),
    ])
    def test_output(self, outfile, expected_out, expected_err, capfd):
        t = termlog.TermLog(outfile=outfile)
        with taddons.context(t) as tctx:
            tctx.options.termlog_verbosity = "info"
            tctx.configure(t)

            

Reported by Pylint.

                      (None, ['one', 'three'], ['four']),
        (sys.stdout, ['one', 'three', 'four'], []),
        (sys.stderr, [], ['one', 'three', 'four']),
    ])
    def test_output(self, outfile, expected_out, expected_err, capfd):
        t = termlog.TermLog(outfile=outfile)
        with taddons.context(t) as tctx:
            tctx.options.termlog_verbosity = "info"
            tctx.configure(t)

            

Reported by Pylint.

                      (sys.stderr, [], ['one', 'three', 'four']),
    ])
    def test_output(self, outfile, expected_out, expected_err, capfd):
        t = termlog.TermLog(outfile=outfile)
        with taddons.context(t) as tctx:
            tctx.options.termlog_verbosity = "info"
            tctx.configure(t)
            t.add_log(log.LogEntry("one", "info"))
            t.add_log(log.LogEntry("two", "debug"))

            

Reported by Pylint.

                          t.add_log(log.LogEntry("three", "warn"))
            t.add_log(log.LogEntry("four", "error"))
        out, err = capfd.readouterr()
        assert out.strip().splitlines() == expected_out
        assert err.strip().splitlines() == expected_err

            

Reported by Bandit.

                          t.add_log(log.LogEntry("four", "error"))
        out, err = capfd.readouterr()
        assert out.strip().splitlines() == expected_out
        assert err.strip().splitlines() == expected_err

            

Reported by Bandit.

              from mitmproxy.contentviews import raw
from . import full_eval


def test_view_raw():
    v = full_eval(raw.ViewRaw())
    assert v(b"foo")



            

Reported by Pylint.

              from mitmproxy.contentviews import raw
from . import full_eval


def test_view_raw():
    v = full_eval(raw.ViewRaw())
    assert v(b"foo")



            

Reported by Pylint.

              from . import full_eval


def test_view_raw():
    v = full_eval(raw.ViewRaw())
    assert v(b"foo")


def test_render_priority():

            

Reported by Pylint.

              

def test_view_raw():
    v = full_eval(raw.ViewRaw())
    assert v(b"foo")


def test_render_priority():
    v = raw.ViewRaw()

            

Reported by Pylint.

              
def test_view_raw():
    v = full_eval(raw.ViewRaw())
    assert v(b"foo")


def test_render_priority():
    v = raw.ViewRaw()
    assert v.render_priority(b"anything")

            

Reported by Bandit.

                  assert v(b"foo")


def test_render_priority():
    v = raw.ViewRaw()
    assert v.render_priority(b"anything")
    assert not v.render_priority(b"")

            

Reported by Pylint.

              

def test_render_priority():
    v = raw.ViewRaw()
    assert v.render_priority(b"anything")
    assert not v.render_priority(b"")

            

Reported by Pylint.

              
def test_render_priority():
    v = raw.ViewRaw()
    assert v.render_priority(b"anything")
    assert not v.render_priority(b"")

            

Reported by Bandit.

              def test_render_priority():
    v = raw.ViewRaw()
    assert v.render_priority(b"anything")
    assert not v.render_priority(b"")

            

Reported by Bandit.

              from mitmproxy.test import tflow

from mitmproxy.addons import anticomp
from mitmproxy.test import taddons


class TestAntiComp:
    def test_simple(self):
        sa = anticomp.AntiComp()

            

Reported by Pylint.

              from mitmproxy.test import taddons


class TestAntiComp:
    def test_simple(self):
        sa = anticomp.AntiComp()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            sa.request(f)

            

Reported by Pylint.

              from mitmproxy.test import taddons


class TestAntiComp:
    def test_simple(self):
        sa = anticomp.AntiComp()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            sa.request(f)

            

Reported by Pylint.

              

class TestAntiComp:
    def test_simple(self):
        sa = anticomp.AntiComp()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            sa.request(f)


            

Reported by Pylint.

              

class TestAntiComp:
    def test_simple(self):
        sa = anticomp.AntiComp()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            sa.request(f)


            

Reported by Pylint.

              
class TestAntiComp:
    def test_simple(self):
        sa = anticomp.AntiComp()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            sa.request(f)

            tctx.configure(sa, anticomp=True)

            

Reported by Pylint.

                  def test_simple(self):
        sa = anticomp.AntiComp()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            sa.request(f)

            tctx.configure(sa, anticomp=True)
            f = tflow.tflow(resp=True)


            

Reported by Pylint.

                          sa.request(f)

            tctx.configure(sa, anticomp=True)
            f = tflow.tflow(resp=True)

            f.request.headers["Accept-Encoding"] = "foobar"
            sa.request(f)
            assert f.request.headers["Accept-Encoding"] == "identity"

            

Reported by Pylint.

              
            f.request.headers["Accept-Encoding"] = "foobar"
            sa.request(f)
            assert f.request.headers["Accept-Encoding"] == "identity"

            

Reported by Bandit.