from . import modes
from .http import HttpLayer
from .tcp import TCPLayer
from .tls import ClientTLSLayer, ServerTLSLayer
from .websocket import WebsocketLayer

__all__ = [
    "modes",
    "HttpLayer",

            

Reported by Pylint.

              from . import modes
from .http import HttpLayer
from .tcp import TCPLayer
from .tls import ClientTLSLayer, ServerTLSLayer
from .websocket import WebsocketLayer

__all__ = [
    "modes",
    "HttpLayer",

            

Reported by Pylint.

              from . import modes
from .http import HttpLayer
from .tcp import TCPLayer
from .tls import ClientTLSLayer, ServerTLSLayer
from .websocket import WebsocketLayer

__all__ = [
    "modes",
    "HttpLayer",

            

Reported by Pylint.

              from .http import HttpLayer
from .tcp import TCPLayer
from .tls import ClientTLSLayer, ServerTLSLayer
from .websocket import WebsocketLayer

__all__ = [
    "modes",
    "HttpLayer",
    "TCPLayer",

            

Reported by Pylint.

              from . import modes
from .http import HttpLayer
from .tcp import TCPLayer
from .tls import ClientTLSLayer, ServerTLSLayer
from .websocket import WebsocketLayer

__all__ = [
    "modes",
    "HttpLayer",

            

Reported by Pylint.

              from . import modes
from .http import HttpLayer
from .tcp import TCPLayer
from .tls import ClientTLSLayer, ServerTLSLayer
from .websocket import WebsocketLayer

__all__ = [
    "modes",
    "HttpLayer",

            

Reported by Pylint.

              from typing import IO, Optional

import click

from mitmproxy import log
from mitmproxy import ctx


class TermLog:

            

Reported by Pylint.

              from mitmproxy import ctx


class TermLog:
    def __init__(self, outfile=None):
        self.outfile: Optional[IO] = outfile

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

            

Reported by Pylint.

                  def __init__(self, outfile=None):
        self.outfile: Optional[IO] = outfile

    def load(self, loader):
        loader.add_option(
            "termlog_verbosity", str, 'info',
            "Log verbosity.",
            choices=log.LogTierOrder
        )

            

Reported by Pylint.

                  def __init__(self, outfile=None):
        self.outfile: Optional[IO] = outfile

    def load(self, loader):
        loader.add_option(
            "termlog_verbosity", str, 'info',
            "Log verbosity.",
            choices=log.LogTierOrder
        )

            

Reported by Pylint.

                          choices=log.LogTierOrder
        )

    def add_log(self, e):
        if log.log_tier(ctx.options.termlog_verbosity) >= log.log_tier(e.level):
            click.secho(
                e.msg,
                file=self.outfile,
                fg=dict(error="red", warn="yellow",

            

Reported by Pylint.

                          choices=log.LogTierOrder
        )

    def add_log(self, e):
        if log.log_tier(ctx.options.termlog_verbosity) >= log.log_tier(e.level):
            click.secho(
                e.msg,
                file=self.outfile,
                fg=dict(error="red", warn="yellow",

            

Reported by Pylint.

              #!/usr/bin/env python3
import shutil
import subprocess
from pathlib import Path


here = Path(__file__).parent

for script in sorted((here / "scripts").glob("*.py")):

            

Reported by Pylint.

              #!/usr/bin/env python3
import shutil
import subprocess
from pathlib import Path


here = Path(__file__).parent

for script in sorted((here / "scripts").glob("*.py")):

            

Reported by Bandit.

              
for script in sorted((here / "scripts").glob("*.py")):
    print(f"Generating output for {script.name}...")
    out = subprocess.check_output(["python3", script.absolute()], cwd=here, text=True)
    if out:
        (here / "src" / "generated" / f"{script.stem}.html").write_text(out, encoding="utf8")

if (here / "public").exists():
    shutil.rmtree(here / "public")

            

Reported by Bandit.

              
for script in sorted((here / "scripts").glob("*.py")):
    print(f"Generating output for {script.name}...")
    out = subprocess.check_output(["python3", script.absolute()], cwd=here, text=True)
    if out:
        (here / "src" / "generated" / f"{script.stem}.html").write_text(out, encoding="utf8")

if (here / "public").exists():
    shutil.rmtree(here / "public")

            

Reported by Bandit.

              
if (here / "public").exists():
    shutil.rmtree(here / "public")
subprocess.run(["hugo"], cwd=here / "src", check=True)

            

Reported by Bandit.

              
if (here / "public").exists():
    shutil.rmtree(here / "public")
subprocess.run(["hugo"], cwd=here / "src", check=True)

            

Reported by Bandit.

              """

if re.match(r"linux(?:2)?", sys.platform):
    from . import linux

    original_addr = linux.original_addr
elif sys.platform == "darwin" or sys.platform.startswith("freebsd"):
    from . import osx


            

Reported by Pylint.

              
    original_addr = linux.original_addr
elif sys.platform == "darwin" or sys.platform.startswith("freebsd"):
    from . import osx

    original_addr = osx.original_addr
elif sys.platform.startswith("openbsd"):
    from . import openbsd


            

Reported by Pylint.

              
    original_addr = osx.original_addr
elif sys.platform.startswith("openbsd"):
    from . import openbsd

    original_addr = openbsd.original_addr
elif sys.platform == "win32":
    from . import windows


            

Reported by Pylint.

              
    original_addr = openbsd.original_addr
elif sys.platform == "win32":
    from . import windows

    resolver = windows.Resolver()
    init_transparent_mode = resolver.setup  # noqa
    original_addr = resolver.original_addr
else:

            

Reported by Pylint.

              import re
import socket
import sys
from typing import Callable, Optional, Tuple


def init_transparent_mode() -> None:
    """
    Initialize transparent mode.

            

Reported by Pylint.

                  init_transparent_mode = resolver.setup  # noqa
    original_addr = resolver.original_addr
else:
    original_addr = None

__all__ = [
    "original_addr",
    "init_transparent_mode"
]

            

Reported by Pylint.

              """
Read a mitmproxy dump file.
"""
from mitmproxy import io, http
from mitmproxy.exceptions import FlowReadException
import pprint
import sys

with open(sys.argv[1], "rb") as logfile:

            

Reported by Pylint.

              Read a mitmproxy dump file.
"""
from mitmproxy import io, http
from mitmproxy.exceptions import FlowReadException
import pprint
import sys

with open(sys.argv[1], "rb") as logfile:
    freader = io.FlowReader(logfile)

            

Reported by Pylint.

              #!/usr/bin/env python
"""
Read a mitmproxy dump file.
"""
from mitmproxy import io, http
from mitmproxy.exceptions import FlowReadException
import pprint
import sys


            

Reported by Pylint.

              """
from mitmproxy import io, http
from mitmproxy.exceptions import FlowReadException
import pprint
import sys

with open(sys.argv[1], "rb") as logfile:
    freader = io.FlowReader(logfile)
    pp = pprint.PrettyPrinter(indent=4)

            

Reported by Pylint.

              from mitmproxy import io, http
from mitmproxy.exceptions import FlowReadException
import pprint
import sys

with open(sys.argv[1], "rb") as logfile:
    freader = io.FlowReader(logfile)
    pp = pprint.PrettyPrinter(indent=4)
    try:

            

Reported by Pylint.

                              print(f.request.host)
            pp.pprint(f.get_state())
            print("")
    except FlowReadException as e:
        print(f"Flow file corrupted: {e}")

            

Reported by Pylint.

              from mitmproxy import ctx
from mitmproxy import hooks


class Recorder:
    call_log = []

    def __init__(self, name="recorder"):
        self.name = name

            

Reported by Pylint.

              from mitmproxy import ctx
from mitmproxy import hooks


class Recorder:
    call_log = []

    def __init__(self, name="recorder"):
        self.name = name

            

Reported by Pylint.

              from mitmproxy import ctx
from mitmproxy import hooks


class Recorder:
    call_log = []

    def __init__(self, name="recorder"):
        self.name = name

            

Reported by Pylint.

              from mitmproxy import hooks


class Recorder:
    call_log = []

    def __init__(self, name="recorder"):
        self.name = name


            

Reported by Pylint.

              from mitmproxy import hooks


class Recorder:
    call_log = []

    def __init__(self, name="recorder"):
        self.name = name


            

Reported by Pylint.

                  def __getattr__(self, attr):
        if attr in hooks.all_hooks:
            def prox(*args, **kwargs):
                lg = (self.name, attr, args, kwargs)
                if attr != "add_log":
                    ctx.log.info(str(lg))
                    self.call_log.append(lg)
                    ctx.log.debug(f"{self.name} {attr}")


            

Reported by Pylint.

              
    mitmproxy -s options-simple.py --set addheader true
"""
from mitmproxy import ctx


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

            

Reported by Pylint.

              """
Add a new mitmproxy option.

Usage:

    mitmproxy -s options-simple.py --set addheader true
"""
from mitmproxy import ctx


            

Reported by Pylint.

              from mitmproxy import ctx


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

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

            

Reported by Pylint.

                  def __init__(self):
        self.num = 0

    def load(self, loader):
        loader.add_option(
            name = "addheader",
            typespec = bool,
            default = False,
            help = "Add a count header to responses",

            

Reported by Pylint.

                  def __init__(self):
        self.num = 0

    def load(self, loader):
        loader.add_option(
            name = "addheader",
            typespec = bool,
            default = False,
            help = "Add a count header to responses",

            

Reported by Pylint.

                          help = "Add a count header to responses",
        )

    def response(self, flow):
        if ctx.options.addheader:
            self.num = self.num + 1
            flow.response.headers["count"] = str(self.num)



            

Reported by Pylint.

              
from mitmproxy.utils import data

import pytest

pytest_plugins = ('test.full_coverage_plugin',)

skip_windows = pytest.mark.skipif(
    os.name == "nt",

            

Reported by Pylint.

              import os
import socket

from mitmproxy.utils import data

import pytest

pytest_plugins = ('test.full_coverage_plugin',)


            

Reported by Pylint.

              
from mitmproxy.utils import data

import pytest

pytest_plugins = ('test.full_coverage_plugin',)

skip_windows = pytest.mark.skipif(
    os.name == "nt",

            

Reported by Pylint.

                  s.bind(("::1", 0))
    s.close()
except OSError:
    no_ipv6 = True
else:
    no_ipv6 = False

skip_no_ipv6 = pytest.mark.skipif(
    no_ipv6,

            

Reported by Pylint.

              except OSError:
    no_ipv6 = True
else:
    no_ipv6 = False

skip_no_ipv6 = pytest.mark.skipif(
    no_ipv6,
    reason='Host has no IPv6 support'
)

            

Reported by Pylint.

              

@pytest.fixture()
def tdata():
    return data.Data(__name__)

            

Reported by Pylint.

              import os
import re
from codecs import open

from setuptools import find_packages, setup

# Based on https://github.com/pypa/sampleproject/blob/main/setup.py
# and https://python-packaging-user-guide.readthedocs.org/


            

Reported by Pylint.

              import os
import re
from codecs import open

from setuptools import find_packages, setup

# Based on https://github.com/pypa/sampleproject/blob/main/setup.py
# and https://python-packaging-user-guide.readthedocs.org/


            

Reported by Pylint.

              
with open(os.path.join(here, 'README.md'), encoding='utf-8') as f:
    long_description = f.read()
long_description_content_type = "text/markdown"

with open(os.path.join(here, "mitmproxy", "version.py")) as f:
    match = re.search(r'VERSION = "(.+?)"', f.read())
    assert match
    VERSION = match.group(1)

            

Reported by Pylint.

              
with open(os.path.join(here, "mitmproxy", "version.py")) as f:
    match = re.search(r'VERSION = "(.+?)"', f.read())
    assert match
    VERSION = match.group(1)

setup(
    name="mitmproxy",
    version=VERSION,

            

Reported by Bandit.

              setup(
    name="mitmproxy",
    version=VERSION,
    description="An interactive, SSL/TLS-capable intercepting proxy for HTTP/1, HTTP/2, and WebSockets.",
    long_description=long_description,
    long_description_content_type=long_description_content_type,
    url="http://mitmproxy.org",
    author="Aldo Cortesi",
    author_email="aldo@corte.si",

            

Reported by Pylint.

                  },
    python_requires='>=3.8',
    # https://packaging.python.org/en/latest/requirements/#install-requires
    # It is not considered best practice to use install_requires to pin dependencies to specific versions.
    install_requires=[
        "asgiref>=3.2.10,<3.5",
        "blinker>=1.4, <1.5",
        "Brotli>=1.0,<1.1",
        "certifi>=2019.9.11",  # no semver here - this should always be on the last release!

            

Reported by Pylint.

              
import pyparsing

# TODO: There is a lot of work to be done here.
# The current implementation is written in a way that _any_ input is valid,
# which does not make sense once things get more complex.

PartialQuotedString = pyparsing.Regex(
    re.compile(

            

Reported by Pylint.

              import re

import pyparsing

# TODO: There is a lot of work to be done here.
# The current implementation is written in a way that _any_ input is valid,
# which does not make sense once things get more complex.

PartialQuotedString = pyparsing.Regex(

            

Reported by Pylint.

              ).leaveWhitespace()


def quote(val: str) -> str:
    if val and all(char not in val for char in "'\" \r\n\t"):
        return val
    if '"' not in val:
        return f'"{val}"'
    if "'" not in val:

            

Reported by Pylint.

                  return '"' + val.replace('"', r"\x22") + '"'


def unquote(x: str) -> str:
    if len(x) > 1 and x[0] in "'\"" and x[0] == x[-1]:
        return x[1:-1]
    else:
        return x

            

Reported by Pylint.

                  return '"' + val.replace('"', r"\x22") + '"'


def unquote(x: str) -> str:
    if len(x) > 1 and x[0] in "'\"" and x[0] == x[-1]:
        return x[1:-1]
    else:
        return x

            

Reported by Pylint.

              

def unquote(x: str) -> str:
    if len(x) > 1 and x[0] in "'\"" and x[0] == x[-1]:
        return x[1:-1]
    else:
        return x

            

Reported by Pylint.