#!/usr/bin/env python3
'''
@author: David Shaw, shawd@vmware.com

Inspired by EAS Inspector for Fiddler
https://easinspectorforfiddler.codeplex.com

----- The MIT License (MIT) ----- 
Filename: GlobalTokens.py

            

Reported by Pylint.

              Inspired by EAS Inspector for Fiddler
https://easinspectorforfiddler.codeplex.com

----- The MIT License (MIT) ----- 
Filename: GlobalTokens.py
Copyright (c) 2014, David P. Shaw

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal

            

Reported by Pylint.

              OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
'''
class GlobalTokens:
    SWITCH_PAGE = 0x00
    END = 0x01
    ENTITY = 0x02
    STR_I = 0x03
    LITERAL = 0x04

            

Reported by Pylint.

              OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
'''
class GlobalTokens:
    SWITCH_PAGE = 0x00
    END = 0x01
    ENTITY = 0x02
    STR_I = 0x03
    LITERAL = 0x04

            

Reported by Pylint.

              import pytest
from mitmproxy.utils import data


def test_pkg_data():
    assert data.pkg_data.path("tools/console")
    with pytest.raises(ValueError):
        data.pkg_data.path("nonexistent")

            

Reported by Pylint.

              import pytest
from mitmproxy.utils import data


def test_pkg_data():
    assert data.pkg_data.path("tools/console")
    with pytest.raises(ValueError):
        data.pkg_data.path("nonexistent")

            

Reported by Pylint.

              from mitmproxy.utils import data


def test_pkg_data():
    assert data.pkg_data.path("tools/console")
    with pytest.raises(ValueError):
        data.pkg_data.path("nonexistent")

            

Reported by Pylint.

              

def test_pkg_data():
    assert data.pkg_data.path("tools/console")
    with pytest.raises(ValueError):
        data.pkg_data.path("nonexistent")

            

Reported by Bandit.

              import mitmproxy.log
import mitmproxy.master
import mitmproxy.options

log: "mitmproxy.log.Log"
master: "mitmproxy.master.Master"
options: "mitmproxy.options.Options"

            

Reported by Pylint.

              import mitmproxy.log
import mitmproxy.master
import mitmproxy.options

log: "mitmproxy.log.Log"
master: "mitmproxy.master.Master"
options: "mitmproxy.options.Options"

            

Reported by Pylint.

              import mitmproxy.log
import mitmproxy.master
import mitmproxy.options

log: "mitmproxy.log.Log"
master: "mitmproxy.master.Master"
options: "mitmproxy.options.Options"

            

Reported by Pylint.

              import mitmproxy.log
import mitmproxy.master
import mitmproxy.options

log: "mitmproxy.log.Log"
master: "mitmproxy.master.Master"
options: "mitmproxy.options.Options"

            

Reported by Pylint.

                      1 2 3
        2 3 None
    """
    # TODO: move into utils
    iters: List[Iterator[Optional[T]]] = list(itertools.tee(iterator, behind + 1 + ahead))
    for i in range(behind):
        iters[i] = itertools.chain((behind - i) * [None], iters[i])
    for i in range(ahead):
        iters[-1 - i] = itertools.islice(

            

Reported by Pylint.

              import itertools
from typing import TypeVar, Iterable, Iterator, Tuple, Optional, List

T = TypeVar('T')


def window(iterator: Iterable[T], behind: int = 0, ahead: int = 0) -> Iterator[Tuple[Optional[T], ...]]:
    """
    Sliding window for an iterator.

            

Reported by Pylint.

              import itertools
from typing import TypeVar, Iterable, Iterator, Tuple, Optional, List

T = TypeVar('T')


def window(iterator: Iterable[T], behind: int = 0, ahead: int = 0) -> Iterator[Tuple[Optional[T], ...]]:
    """
    Sliding window for an iterator.

            

Reported by Pylint.

              T = TypeVar('T')


def window(iterator: Iterable[T], behind: int = 0, ahead: int = 0) -> Iterator[Tuple[Optional[T], ...]]:
    """
    Sliding window for an iterator.

    Example:
        >>> for prev, i, nxt in window(range(10), 1, 1):

            

Reported by Pylint.

              import secrets
from pathlib import Path

import objgraph

from mitmproxy import certs

if __name__ == "__main__":
    store = certs.CertStore.from_store(path=Path("~/.mitmproxy/").expanduser(), basename="mitmproxy", key_size=2048)

            

Reported by Pylint.

              
import objgraph

from mitmproxy import certs

if __name__ == "__main__":
    store = certs.CertStore.from_store(path=Path("~/.mitmproxy/").expanduser(), basename="mitmproxy", key_size=2048)
    store.STORE_CAP = 5


            

Reported by Pylint.

              import secrets
from pathlib import Path

import objgraph

from mitmproxy import certs

if __name__ == "__main__":
    store = certs.CertStore.from_store(path=Path("~/.mitmproxy/").expanduser(), basename="mitmproxy", key_size=2048)

            

Reported by Pylint.

              from mitmproxy import certs

if __name__ == "__main__":
    store = certs.CertStore.from_store(path=Path("~/.mitmproxy/").expanduser(), basename="mitmproxy", key_size=2048)
    store.STORE_CAP = 5

    for _ in range(5):
        store.get_cert(commonname=secrets.token_hex(16).encode(), sans=[], organization=None)


            

Reported by Pylint.

              #!/usr/bin/env python3
"""
All of the emoji and characters that can be used as flow markers.
"""

# auto-generated. run this file to refresh.

emoji = {
    ":+1:": "👍",

            

Reported by Pylint.

                  for name, url in r.json().items():
        codepoints = url.rpartition("/")[2].partition(".png")[0].split("-")
        try:
            emoji_val = "\u200d".join(chr(int(c, 16)) for c in codepoints)
        except ValueError:
            continue  # some GitHub-specific emojis, e.g. "atom".
        print(EMOJI_SRC.format(name=name, emoji_val=emoji_val), file=out)

    # add the default marker

            

Reported by Pylint.

                      print(CHAR_SRC.format(name=c, emoji_val=c), file=out)

    Path(__file__).write_text(
        re.sub(r"(?<={\n)[\s\S]*(?=}\n)", lambda x: out.getvalue(), Path(__file__).read_text("utf8"), 1),
        "utf8"
    )
            

Reported by Pylint.

                  Path(__file__).write_text(
        re.sub(r"(?<={\n)[\s\S]*(?=}\n)", lambda x: out.getvalue(), Path(__file__).read_text("utf8"), 1),
        "utf8"
    )
            

Reported by Pylint.

              from mitmproxy import ctx


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


def request(flow):
    raise ValueError("Error!")

            

Reported by Pylint.

              from mitmproxy import ctx


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


def request(flow):
    raise ValueError("Error!")

            

Reported by Pylint.

              from mitmproxy import ctx


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


def request(flow):
    raise ValueError("Error!")

            

Reported by Pylint.

                  ctx.log.info("error running")


def request(flow):
    raise ValueError("Error!")

            

Reported by Pylint.

              """Add an HTTP header to each response."""


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

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

            

Reported by Pylint.

              """Add an HTTP header to each response."""


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

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

            

Reported by Pylint.

              """Add an HTTP header to each response."""


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

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

            

Reported by Pylint.

                  def __init__(self):
        self.num = 0

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


addons = [

            

Reported by Pylint.

              import asyncio

from mitmproxy.tools import main


shutdown_script = "mitmproxy/data/addonscripts/shutdown.py"


def test_mitmweb(event_loop, tdata):

            

Reported by Pylint.

              from mitmproxy.tools import main


shutdown_script = "mitmproxy/data/addonscripts/shutdown.py"


def test_mitmweb(event_loop, tdata):
    asyncio.set_event_loop(event_loop)
    main.mitmweb([

            

Reported by Pylint.

              shutdown_script = "mitmproxy/data/addonscripts/shutdown.py"


def test_mitmweb(event_loop, tdata):
    asyncio.set_event_loop(event_loop)
    main.mitmweb([
        "--no-web-open-browser",
        "-s", tdata.path(shutdown_script),
        "-q", "-p", "0", "--web-port", "0",

            

Reported by Pylint.

                  ])


def test_mitmdump(event_loop, tdata):
    asyncio.set_event_loop(event_loop)
    main.mitmdump([
        "-s", tdata.path(shutdown_script),
        "-q", "-p", "0",
    ])

            

Reported by Pylint.

              """
Modify a streamed response.

Generally speaking, we recommend *not* to stream messages you need to modify.
Modifying streamed responses is tricky and brittle:
    - If the transfer encoding isn't chunked, you cannot simply change the content length.
    - If you want to replace all occurrences of "foobar", make sure to catch the cases
      where one chunk ends with [...]foo" and the next starts with "bar[...].
"""

            

Reported by Pylint.

              
def modify(data: bytes) -> Union[bytes, Iterable[bytes]]:
    """
    This function will be called for each chunk of request/response body data that arrives at the proxy,
    and once at the end of the message with an empty bytes argument (b"").

    It may either return bytes or an iterable of bytes (which would result in multiple HTTP/2 data frames).
    """
    return data.replace(b"foo", b"bar")

            

Reported by Pylint.

                  This function will be called for each chunk of request/response body data that arrives at the proxy,
    and once at the end of the message with an empty bytes argument (b"").

    It may either return bytes or an iterable of bytes (which would result in multiple HTTP/2 data frames).
    """
    return data.replace(b"foo", b"bar")


def responseheaders(flow):

            

Reported by Pylint.

                  return data.replace(b"foo", b"bar")


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

            

Reported by Pylint.