import pytest

from mitmproxy.test import tflow
from mitmproxy.test import taddons

from mitmproxy.addons import stickyauth
from mitmproxy import exceptions



            

Reported by Pylint.

              import pytest

from mitmproxy.test import tflow
from mitmproxy.test import taddons

from mitmproxy.addons import stickyauth
from mitmproxy import exceptions



            

Reported by Pylint.

              from mitmproxy import exceptions


def test_configure():
    r = stickyauth.StickyAuth()
    with taddons.context(r) as tctx:
        tctx.configure(r, stickyauth="~s")
        with pytest.raises(exceptions.OptionsError):
            tctx.configure(r, stickyauth="~~")

            

Reported by Pylint.

              

def test_configure():
    r = stickyauth.StickyAuth()
    with taddons.context(r) as tctx:
        tctx.configure(r, stickyauth="~s")
        with pytest.raises(exceptions.OptionsError):
            tctx.configure(r, stickyauth="~~")


            

Reported by Pylint.

                          tctx.configure(r, stickyauth="~~")

        tctx.configure(r, stickyauth=None)
        assert not r.flt


def test_simple():
    r = stickyauth.StickyAuth()
    with taddons.context(r) as tctx:

            

Reported by Bandit.

                      assert not r.flt


def test_simple():
    r = stickyauth.StickyAuth()
    with taddons.context(r) as tctx:
        tctx.configure(r, stickyauth=".*")
        f = tflow.tflow(resp=True)
        f.request.headers["authorization"] = "foo"

            

Reported by Pylint.

              

def test_simple():
    r = stickyauth.StickyAuth()
    with taddons.context(r) as tctx:
        tctx.configure(r, stickyauth=".*")
        f = tflow.tflow(resp=True)
        f.request.headers["authorization"] = "foo"
        r.request(f)

            

Reported by Pylint.

                  r = stickyauth.StickyAuth()
    with taddons.context(r) as tctx:
        tctx.configure(r, stickyauth=".*")
        f = tflow.tflow(resp=True)
        f.request.headers["authorization"] = "foo"
        r.request(f)

        assert "address" in r.hosts


            

Reported by Pylint.

                      f.request.headers["authorization"] = "foo"
        r.request(f)

        assert "address" in r.hosts

        f = tflow.tflow(resp=True)
        r.request(f)
        assert f.request.headers["authorization"] == "foo"

            

Reported by Bandit.

              
        assert "address" in r.hosts

        f = tflow.tflow(resp=True)
        r.request(f)
        assert f.request.headers["authorization"] == "foo"

            

Reported by Pylint.

              import argparse

from mitmproxy import options
from mitmproxy.tools import cmdline, web, dump, console
from mitmproxy.tools import main


def test_common():
    parser = argparse.ArgumentParser()

            

Reported by Pylint.

              from mitmproxy.tools import main


def test_common():
    parser = argparse.ArgumentParser()
    opts = options.Options()
    cmdline.common_options(parser, opts)
    args = parser.parse_args(args=[])
    main.process_options(parser, opts, args)

            

Reported by Pylint.

                  main.process_options(parser, opts, args)


def test_mitmproxy():
    opts = options.Options()
    console.master.ConsoleMaster(opts)
    ap = cmdline.mitmproxy(opts)
    assert ap


            

Reported by Pylint.

              def test_mitmproxy():
    opts = options.Options()
    console.master.ConsoleMaster(opts)
    ap = cmdline.mitmproxy(opts)
    assert ap


def test_mitmdump():
    opts = options.Options()

            

Reported by Pylint.

                  opts = options.Options()
    console.master.ConsoleMaster(opts)
    ap = cmdline.mitmproxy(opts)
    assert ap


def test_mitmdump():
    opts = options.Options()
    dump.DumpMaster(opts)

            

Reported by Bandit.

                  assert ap


def test_mitmdump():
    opts = options.Options()
    dump.DumpMaster(opts)
    ap = cmdline.mitmdump(opts)
    assert ap


            

Reported by Pylint.

              def test_mitmdump():
    opts = options.Options()
    dump.DumpMaster(opts)
    ap = cmdline.mitmdump(opts)
    assert ap


def test_mitmweb():
    opts = options.Options()

            

Reported by Pylint.

                  opts = options.Options()
    dump.DumpMaster(opts)
    ap = cmdline.mitmdump(opts)
    assert ap


def test_mitmweb():
    opts = options.Options()
    web.master.WebMaster(opts)

            

Reported by Bandit.

                  assert ap


def test_mitmweb():
    opts = options.Options()
    web.master.WebMaster(opts)
    ap = cmdline.mitmweb(opts)
    assert ap

            

Reported by Pylint.

              def test_mitmweb():
    opts = options.Options()
    web.master.WebMaster(opts)
    ap = cmdline.mitmweb(opts)
    assert ap

            

Reported by Pylint.

              from mitmproxy.contentviews import hex
from . import full_eval


def test_view_hex():
    v = full_eval(hex.ViewHex())
    assert v(b"foo")



            

Reported by Pylint.

              from mitmproxy.contentviews import hex
from . import full_eval


def test_view_hex():
    v = full_eval(hex.ViewHex())
    assert v(b"foo")



            

Reported by Pylint.

              from mitmproxy.contentviews import hex
from . import full_eval


def test_view_hex():
    v = full_eval(hex.ViewHex())
    assert v(b"foo")



            

Reported by Pylint.

              from . import full_eval


def test_view_hex():
    v = full_eval(hex.ViewHex())
    assert v(b"foo")


def test_render_priority():

            

Reported by Pylint.

              

def test_view_hex():
    v = full_eval(hex.ViewHex())
    assert v(b"foo")


def test_render_priority():
    v = hex.ViewHex()

            

Reported by Pylint.

              
def test_view_hex():
    v = full_eval(hex.ViewHex())
    assert v(b"foo")


def test_render_priority():
    v = hex.ViewHex()
    assert not v.render_priority(b"ascii")

            

Reported by Bandit.

                  assert v(b"foo")


def test_render_priority():
    v = hex.ViewHex()
    assert not v.render_priority(b"ascii")
    assert v.render_priority(b"\xFF")
    assert not v.render_priority(b"")

            

Reported by Pylint.

              

def test_render_priority():
    v = hex.ViewHex()
    assert not v.render_priority(b"ascii")
    assert v.render_priority(b"\xFF")
    assert not v.render_priority(b"")

            

Reported by Pylint.

              
def test_render_priority():
    v = hex.ViewHex()
    assert not v.render_priority(b"ascii")
    assert v.render_priority(b"\xFF")
    assert not v.render_priority(b"")

            

Reported by Bandit.

              def test_render_priority():
    v = hex.ViewHex()
    assert not v.render_priority(b"ascii")
    assert v.render_priority(b"\xFF")
    assert not v.render_priority(b"")

            

Reported by Bandit.

              from mitmproxy import log
from mitmproxy import options
from mitmproxy.net import server_spec
from . import ctx as mitmproxy_ctx

# Conclusively preventing cross-thread races on proxy shutdown turns out to be
# very hard. We could build a thread sync infrastructure for this, or we could
# wait until we ditch threads and move all the protocols into the async loop.
# Until then, silence non-critical errors.

            

Reported by Pylint.

                      exc = None
        try:
            loop()
        except Exception:  # pragma: no cover
            exc = traceback.format_exc()
        finally:
            if not self.should_exit.is_set():  # pragma: no cover
                self.shutdown()
            loop = asyncio.get_event_loop()

            

Reported by Pylint.

              import asyncio
import logging
import sys
import threading
import traceback

from mitmproxy import addonmanager, hooks
from mitmproxy import command
from mitmproxy import controller

            

Reported by Pylint.

                      mitmproxy_ctx.log = self.log
        mitmproxy_ctx.options = self.options

    def start(self):
        self.should_exit.clear()

    async def running(self):
        self.addons.trigger(hooks.RunningHook())


            

Reported by Pylint.

                  def start(self):
        self.should_exit.clear()

    async def running(self):
        self.addons.trigger(hooks.RunningHook())

    def run_loop(self, loop):
        self.start()
        asyncio.ensure_future(self.running())

            

Reported by Pylint.

                  async def running(self):
        self.addons.trigger(hooks.RunningHook())

    def run_loop(self, loop):
        self.start()
        asyncio.ensure_future(self.running())

        exc = None
        try:

            

Reported by Pylint.

                              self.shutdown()
            loop = asyncio.get_event_loop()
            tasks = asyncio.all_tasks(loop)
            for p in tasks:
                p.cancel()
            loop.close()

        if exc:  # pragma: no cover
            print(exc, file=sys.stderr)

            

Reported by Pylint.

              
        self.addons.trigger(hooks.DoneHook())

    def run(self):
        loop = asyncio.get_event_loop()
        self.run_loop(loop.run_forever)

    async def _shutdown(self):
        self.should_exit.set()

            

Reported by Pylint.

                              except RuntimeError:
                    pass  # Event loop stopped before Future completed.

    def _change_reverse_host(self, f):
        """
        When we load flows in reverse proxy mode, we adjust the target host to
        the reverse proxy destination for all flows we load. This makes it very
        easy to replay saved flows against a different host.
        """

            

Reported by Pylint.

                          f.request.host, f.request.port = upstream_spec.address
            f.request.scheme = upstream_spec.scheme

    async def load_flow(self, f):
        """
        Loads a flow
        """

        if isinstance(f, http.HTTPFlow):

            

Reported by Pylint.

                          elif is_graphql_batch_query(data):
                return "GraphQL", base.format_text(format_query_list(data))

    def render_priority(self, data: bytes, *, content_type: typing.Optional[str] = None, **metadata) -> float:
        if content_type != "application/json":
            return 0

        data = parse_json(data)


            

Reported by Pylint.

              import json

import typing

from mitmproxy.contentviews import base
from mitmproxy.contentviews.json import parse_json, PARSE_ERROR


def format_graphql(data):

            

Reported by Pylint.

              from mitmproxy.contentviews.json import parse_json, PARSE_ERROR


def format_graphql(data):
    query = data["query"]
    header_data = data.copy()
    header_data["query"] = "..."
    return """{header}
---

            

Reported by Pylint.

              """.format(header=json.dumps(header_data, indent=2), query = query)


def format_query_list(data: typing.List[typing.Any]):
    num_queries = len(data) - 1
    result = ""
    for i, op in enumerate(data):
        result += "--- {i}/{num_queries}\n".format(i=i, num_queries=num_queries)
        result += format_graphql(op)

            

Reported by Pylint.

              def format_query_list(data: typing.List[typing.Any]):
    num_queries = len(data) - 1
    result = ""
    for i, op in enumerate(data):
        result += "--- {i}/{num_queries}\n".format(i=i, num_queries=num_queries)
        result += format_graphql(op)
    return result



            

Reported by Pylint.

                  return result


def is_graphql_query(data):
    return isinstance(data, dict) and "query" in data and "\n" in data["query"]


def is_graphql_batch_query(data):
    return isinstance(data, list) and isinstance(data[0], dict) and "query" in data[0]

            

Reported by Pylint.

                  return isinstance(data, dict) and "query" in data and "\n" in data["query"]


def is_graphql_batch_query(data):
    return isinstance(data, list) and isinstance(data[0], dict) and "query" in data[0]


class ViewGraphQL(base.View):
    name = "GraphQL"

            

Reported by Pylint.

                  return isinstance(data, list) and isinstance(data[0], dict) and "query" in data[0]


class ViewGraphQL(base.View):
    name = "GraphQL"

    def __call__(self, data, **metadata):
        data = parse_json(data)
        if data is not PARSE_ERROR:

            

Reported by Pylint.

              class ViewGraphQL(base.View):
    name = "GraphQL"

    def __call__(self, data, **metadata):
        data = parse_json(data)
        if data is not PARSE_ERROR:
            if is_graphql_query(data):
                return "GraphQL", base.format_text(format_graphql(data))
            elif is_graphql_batch_query(data):

            

Reported by Pylint.

                  def __call__(self, data, **metadata):
        data = parse_json(data)
        if data is not PARSE_ERROR:
            if is_graphql_query(data):
                return "GraphQL", base.format_text(format_graphql(data))
            elif is_graphql_batch_query(data):
                return "GraphQL", base.format_text(format_query_list(data))

    def render_priority(self, data: bytes, *, content_type: typing.Optional[str] = None, **metadata) -> float:

            

Reported by Pylint.

              import typing
import urwid

import mitmproxy.flow
from mitmproxy import http
from mitmproxy.tools.console import common, searchable
from mitmproxy.utils import human, strutils



            

Reported by Pylint.

                      return "–"


def flowdetails(state, flow: mitmproxy.flow.Flow):
    text = []

    sc = flow.server_conn
    cc = flow.client_conn
    req: typing.Optional[http.Request]

            

Reported by Pylint.

              import typing
import urwid

import mitmproxy.flow
from mitmproxy import http
from mitmproxy.tools.console import common, searchable
from mitmproxy.utils import human, strutils



            

Reported by Pylint.

              from mitmproxy.utils import human, strutils


def maybe_timestamp(base, attr):
    if base is not None and getattr(base, attr):
        return human.format_timestamp_with_milli(getattr(base, attr))
    else:
        # in mitmdump we serialize before a connection is closed.
        # loading those flows at a later point shouldn't display "active".

            

Reported by Pylint.

              

def maybe_timestamp(base, attr):
    if base is not None and getattr(base, attr):
        return human.format_timestamp_with_milli(getattr(base, attr))
    else:
        # in mitmdump we serialize before a connection is closed.
        # loading those flows at a later point shouldn't display "active".
        # We also use a ndash (and not a regular dash) so that it is sorted

            

Reported by Pylint.

                      return "–"


def flowdetails(state, flow: mitmproxy.flow.Flow):
    text = []

    sc = flow.server_conn
    cc = flow.client_conn
    req: typing.Optional[http.Request]

            

Reported by Pylint.

                      return "–"


def flowdetails(state, flow: mitmproxy.flow.Flow):
    text = []

    sc = flow.server_conn
    cc = flow.client_conn
    req: typing.Optional[http.Request]

            

Reported by Pylint.

                      return "–"


def flowdetails(state, flow: mitmproxy.flow.Flow):
    text = []

    sc = flow.server_conn
    cc = flow.client_conn
    req: typing.Optional[http.Request]

            

Reported by Pylint.

              def flowdetails(state, flow: mitmproxy.flow.Flow):
    text = []

    sc = flow.server_conn
    cc = flow.client_conn
    req: typing.Optional[http.Request]
    resp: typing.Optional[http.Response]
    if isinstance(flow, http.HTTPFlow):
        req = flow.request

            

Reported by Pylint.

                  text = []

    sc = flow.server_conn
    cc = flow.client_conn
    req: typing.Optional[http.Request]
    resp: typing.Optional[http.Response]
    if isinstance(flow, http.HTTPFlow):
        req = flow.request
        resp = flow.response

            

Reported by Pylint.