res = beautify(data)
        return "JavaScript", base.format_text(res)

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

            

Reported by Pylint.

              import io
import re
from typing import Optional

from mitmproxy.utils import strutils
from mitmproxy.contentviews import base

DELIMITERS = '{};\n'
SPECIAL_AREAS = (

            

Reported by Pylint.

              )


def beautify(data):
    data = strutils.escape_special_areas(
        data,
        SPECIAL_AREAS,
        DELIMITERS
    )

            

Reported by Pylint.

                  return data


class ViewJavaScript(base.View):
    name = "JavaScript"
    __content_types = (
        "application/x-javascript",
        "application/javascript",
        "text/javascript"

            

Reported by Pylint.

                      res = beautify(data)
        return "JavaScript", base.format_text(res)

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

            

Reported by Pylint.

              from mitmproxy import ctx

def modify(chunks):
    for chunk in chunks:
        yield chunk.replace(b"foo", b"bar")


def running():
    ctx.log.info("stream_modify running")

            

Reported by Pylint.

              from mitmproxy import ctx

def modify(chunks):
    for chunk in chunks:
        yield chunk.replace(b"foo", b"bar")


def running():
    ctx.log.info("stream_modify running")

            

Reported by Pylint.

              from mitmproxy import ctx

def modify(chunks):
    for chunk in chunks:
        yield chunk.replace(b"foo", b"bar")


def running():
    ctx.log.info("stream_modify running")

            

Reported by Pylint.

                      yield chunk.replace(b"foo", b"bar")


def running():
    ctx.log.info("stream_modify running")


def responseheaders(flow):
    flow.response.stream = modify

            

Reported by Pylint.

                  ctx.log.info("stream_modify running")


def responseheaders(flow):
    flow.response.stream = modify

            

Reported by Pylint.

              a single simplest-possible page.
"""
from flask import Flask
from mitmproxy.addons import asgiapp

app = Flask("proxapp")


@app.route('/')

            

Reported by Pylint.

              """
Host a WSGI app in mitmproxy.

This example shows how to graft a WSGI app onto mitmproxy. In this
instance, we're using the Flask framework (http://flask.pocoo.org/) to expose
a single simplest-possible page.
"""
from flask import Flask
from mitmproxy.addons import asgiapp

            

Reported by Pylint.

              

@app.route('/')
def hello_world() -> str:
    return 'Hello World!'


addons = [
    # Host app at the magic domain "example.com" on port 80. Requests to this

            

Reported by Pylint.

                  # Host app at the magic domain "example.com" on port 80. Requests to this
    # domain and port combination will now be routed to the WSGI app instance.
    asgiapp.WSGIApp(app, "example.com", 80)
    # SSL works too, but the magic domain needs to be resolvable from the mitmproxy machine due to mitmproxy's design.
    # mitmproxy will connect to said domain and use serve its certificate (unless --no-upstream-cert is set)
    # but won't send any data.
    # mitmproxy.ctx.master.apps.add(app, "example.com", 443)
]

            

Reported by Pylint.

                  # domain and port combination will now be routed to the WSGI app instance.
    asgiapp.WSGIApp(app, "example.com", 80)
    # SSL works too, but the magic domain needs to be resolvable from the mitmproxy machine due to mitmproxy's design.
    # mitmproxy will connect to said domain and use serve its certificate (unless --no-upstream-cert is set)
    # but won't send any data.
    # mitmproxy.ctx.master.apps.add(app, "example.com", 443)
]

            

Reported by Pylint.

              
Run as follows: mitmproxy -s anatomy.py
"""
from mitmproxy import ctx


class Counter:
    def __init__(self):
        self.num = 0

            

Reported by Pylint.

                  def __init__(self):
        self.num = 0

    def request(self, flow):
        self.num = self.num + 1
        ctx.log.info("We've seen %d flows" % self.num)


addons = [

            

Reported by Pylint.

              from mitmproxy import ctx


class Counter:
    def __init__(self):
        self.num = 0

    def request(self, flow):
        self.num = self.num + 1

            

Reported by Pylint.

              from mitmproxy import ctx


class Counter:
    def __init__(self):
        self.num = 0

    def request(self, flow):
        self.num = self.num + 1

            

Reported by Pylint.

                  def __init__(self):
        self.num = 0

    def request(self, flow):
        self.num = self.num + 1
        ctx.log.info("We've seen %d flows" % self.num)


addons = [

            

Reported by Pylint.

              import time
import sys
from mitmproxy.script import concurrent


@concurrent
def request(flow):
    time.sleep(0.1)

            

Reported by Pylint.

              import time
import sys
from mitmproxy.script import concurrent


@concurrent
def request(flow):
    time.sleep(0.1)

            

Reported by Pylint.

              

@concurrent
def request(flow):
    time.sleep(0.1)

            

Reported by Pylint.

              import time
import sys
from mitmproxy.script import concurrent


@concurrent
def request(flow):
    time.sleep(0.1)

            

Reported by Pylint.

              

@concurrent
def request(flow):
    time.sleep(0.1)

            

Reported by Pylint.

              import socketserver
from time import sleep


class service(socketserver.BaseRequestHandler):

    def handle(self):
        data = 'dummy'
        print("Client connected with ", self.client_address)

            

Reported by Pylint.

              from time import sleep


class service(socketserver.BaseRequestHandler):

    def handle(self):
        data = 'dummy'
        print("Client connected with ", self.client_address)
        while True:

            

Reported by Pylint.

              from time import sleep


class service(socketserver.BaseRequestHandler):

    def handle(self):
        data = 'dummy'
        print("Client connected with ", self.client_address)
        while True:

            

Reported by Pylint.

                          self.request.send(
                "HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 7\r\n\r\ncontent")
            data = self.request.recv(1024)
            if not len(data):
                print("Connection closed by remote: ", self.client_address)
                sleep(3600)


class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):

            

Reported by Pylint.

                              sleep(3600)


class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    pass


server = ThreadedTCPServer(('', 1520), service)
server.serve_forever()

            

Reported by Pylint.

              
def test_basethread():
    t = basethread.BaseThread('foobar')
    assert re.match(r'foobar - age: \d+s', t._threadinfo())

            

Reported by Pylint.

              import re
from mitmproxy.coretypes import basethread


def test_basethread():
    t = basethread.BaseThread('foobar')
    assert re.match(r'foobar - age: \d+s', t._threadinfo())

            

Reported by Pylint.

              from mitmproxy.coretypes import basethread


def test_basethread():
    t = basethread.BaseThread('foobar')
    assert re.match(r'foobar - age: \d+s', t._threadinfo())

            

Reported by Pylint.

              

def test_basethread():
    t = basethread.BaseThread('foobar')
    assert re.match(r'foobar - age: \d+s', t._threadinfo())

            

Reported by Pylint.

              
def test_basethread():
    t = basethread.BaseThread('foobar')
    assert re.match(r'foobar - age: \d+s', t._threadinfo())

            

Reported by Bandit.

              """Post messages to mitmproxy's event log."""
from mitmproxy import ctx


def load(l):
    ctx.log.info("This is some informative text.")
    ctx.log.warn("This is a warning.")
    ctx.log.error("This is an error.")

            

Reported by Pylint.

              from mitmproxy import ctx


def load(l):
    ctx.log.info("This is some informative text.")
    ctx.log.warn("This is a warning.")
    ctx.log.error("This is an error.")

            

Reported by Pylint.

              """Post messages to mitmproxy's event log."""
from mitmproxy import ctx


def load(l):
    ctx.log.info("This is some informative text.")
    ctx.log.warn("This is a warning.")
    ctx.log.error("This is an error.")

            

Reported by Pylint.

              from mitmproxy import ctx


def load(l):
    ctx.log.info("This is some informative text.")
    ctx.log.warn("This is a warning.")
    ctx.log.error("This is an error.")

            

Reported by Pylint.

              from mitmproxy import ctx


def load(l):
    ctx.log.info("This is some informative text.")
    ctx.log.warn("This is a warning.")
    ctx.log.error("This is an error.")

            

Reported by Pylint.

              import os

import pytest
from hypothesis import settings

from mitmproxy import connection, options
from mitmproxy.addons.proxyserver import Proxyserver
from mitmproxy.addons.termlog import TermLog
from mitmproxy.proxy import context

            

Reported by Pylint.

              import os

import pytest
from hypothesis import settings

from mitmproxy import connection, options
from mitmproxy.addons.proxyserver import Proxyserver
from mitmproxy.addons.termlog import TermLog
from mitmproxy.proxy import context

            

Reported by Pylint.

              import os

import pytest
from hypothesis import settings

from mitmproxy import connection, options
from mitmproxy.addons.proxyserver import Proxyserver
from mitmproxy.addons.termlog import TermLog
from mitmproxy.proxy import context

            

Reported by Pylint.

              

@pytest.fixture
def tctx() -> context.Context:
    opts = options.Options()
    Proxyserver().load(opts)
    TermLog().load(opts)
    return context.Context(
        connection.Client(

            

Reported by Pylint.

              import socket
import struct
import typing

# Python's socket module does not have these constants
SO_ORIGINAL_DST = 80
SOL_IPV6 = 41



            

Reported by Pylint.

              SOL_IPV6 = 41


def original_addr(csock: socket.socket) -> typing.Tuple[str, int]:
    # Get the original destination on Linux.
    # In theory, this can be done using the following syscalls:
    #     sock.getsockopt(socket.SOL_IP, SO_ORIGINAL_DST, 16)
    #     sock.getsockopt(SOL_IPV6, SO_ORIGINAL_DST, 28)
    #

            

Reported by Pylint.

                      # with buflen=8 doesn't work.
        dst = csock.getsockopt(socket.SOL_IP, SO_ORIGINAL_DST, 16)
        port, raw_ip = struct.unpack_from("!2xH4s", dst)
        ip = socket.inet_ntop(socket.AF_INET, raw_ip)
    else:
        dst = csock.getsockopt(SOL_IPV6, SO_ORIGINAL_DST, 28)
        port, raw_ip = struct.unpack_from("!2xH4x16s", dst)
        ip = socket.inet_ntop(socket.AF_INET6, raw_ip)
    return ip, port

            

Reported by Pylint.

                  else:
        dst = csock.getsockopt(SOL_IPV6, SO_ORIGINAL_DST, 28)
        port, raw_ip = struct.unpack_from("!2xH4x16s", dst)
        ip = socket.inet_ntop(socket.AF_INET6, raw_ip)
    return ip, port

            

Reported by Pylint.