mirror of https://github.com/scrapy/scrapy.git
71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from subprocess import PIPE, Popen
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
|
|
class MitmProxy:
|
|
auth_user = "scrapy"
|
|
auth_pass = "scrapy"
|
|
|
|
def __init__(self, mode: str | None = None) -> None:
|
|
self.mode = mode
|
|
|
|
def start(self) -> str:
|
|
script = """
|
|
import sys
|
|
from mitmproxy.tools.main import mitmdump
|
|
sys.argv[0] = "mitmdump"
|
|
sys.exit(mitmdump())
|
|
"""
|
|
cert_path = Path(__file__).parent.parent.resolve() / "keys"
|
|
args = [
|
|
"--listen-host",
|
|
"127.0.0.1",
|
|
"--listen-port",
|
|
"0",
|
|
"--proxyauth",
|
|
f"{self.auth_user}:{self.auth_pass}",
|
|
"--set",
|
|
f"confdir={cert_path}",
|
|
"--ssl-insecure",
|
|
"-s",
|
|
str(Path(__file__).with_name("mitm_proxy_addon.py")),
|
|
]
|
|
if self.mode:
|
|
args += ["--mode", self.mode]
|
|
self.proc: Popen[str] = Popen(
|
|
[
|
|
sys.executable,
|
|
"-u",
|
|
"-c",
|
|
script,
|
|
*args,
|
|
],
|
|
stdout=PIPE,
|
|
text=True,
|
|
)
|
|
assert self.proc.stdout is not None
|
|
scheme = "socks5" if self.mode == "socks5" else "http"
|
|
line = ""
|
|
for line in self.proc.stdout:
|
|
m = re.search(r"listening at (?:\w+://)?([^:]+:\d+)", line)
|
|
if m:
|
|
host_port = m.group(1)
|
|
return f"{scheme}://{self.auth_user}:{self.auth_pass}@{host_port}"
|
|
self.stop()
|
|
raise RuntimeError(f"Failed to parse mitmdump output: {line}")
|
|
|
|
def stop(self) -> None:
|
|
self.proc.kill()
|
|
self.proc.communicate()
|
|
|
|
|
|
def wrong_credentials(proxy_url: str) -> str:
|
|
bad_auth_proxy = list(urlsplit(proxy_url))
|
|
bad_auth_proxy[1] = bad_auth_proxy[1].replace("scrapy:scrapy@", "wrong:wronger@")
|
|
return urlunsplit(bad_auth_proxy)
|