mirror of
https://github.com/Ed1s0nZ/CyberStrikeAI.git
synced 2026-07-18 18:07:32 +02:00
公明
1555 lines
62 KiB
YAML
1555 lines
62 KiB
YAML
name: "http-framework-test"
|
||
command: "python3"
|
||
args:
|
||
- "-c"
|
||
- |
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import shlex
|
||
import socket
|
||
import ssl
|
||
import sys
|
||
import time
|
||
import urllib.parse
|
||
import urllib.request
|
||
from typing import Dict, List, Tuple
|
||
|
||
try:
|
||
import httpx
|
||
except ImportError:
|
||
print("Missing dependency: httpx. Install it with `pip install httpx`.", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
try:
|
||
from charset_normalizer import from_bytes as charset_from_bytes
|
||
except ImportError:
|
||
charset_from_bytes = None
|
||
|
||
try:
|
||
import chardet
|
||
except ImportError:
|
||
chardet = None
|
||
|
||
METRIC_KEYS = [
|
||
"dns_lookup",
|
||
"tcp_connect",
|
||
"tls_handshake",
|
||
"pretransfer",
|
||
"ttfb",
|
||
"total",
|
||
"speed_download",
|
||
"size_download",
|
||
"http_code",
|
||
"redirects",
|
||
]
|
||
|
||
|
||
def parse_headers(raw: str) -> List[Tuple[str, str]]:
|
||
if not raw:
|
||
return []
|
||
raw = raw.strip()
|
||
if not raw:
|
||
return []
|
||
headers: List[Tuple[str, str]] = []
|
||
parsed = None
|
||
try:
|
||
parsed = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
parsed = None
|
||
if isinstance(parsed, dict):
|
||
for key, value in parsed.items():
|
||
headers.append((str(key).strip(), str(value).strip()))
|
||
return headers
|
||
if isinstance(parsed, list):
|
||
for item in parsed:
|
||
if isinstance(item, str) and ":" in item:
|
||
key, value = item.split(":", 1)
|
||
headers.append((key.strip(), value.strip()))
|
||
if headers:
|
||
return headers
|
||
for line in raw.replace(";", "\n").splitlines():
|
||
if ":" not in line:
|
||
continue
|
||
key, value = line.split(":", 1)
|
||
headers.append((key.strip(), value.strip()))
|
||
return headers
|
||
|
||
|
||
def parse_cookies(raw: str) -> Dict[str, str]:
|
||
cookies: Dict[str, str] = {}
|
||
if not raw:
|
||
return cookies
|
||
for part in raw.split(";"):
|
||
if "=" not in part:
|
||
continue
|
||
name, value = part.split("=", 1)
|
||
name = name.strip()
|
||
if not name:
|
||
continue
|
||
cookies[name] = value.strip()
|
||
return cookies
|
||
|
||
|
||
def parse_additional_options(raw: str) -> Dict[str, str]:
|
||
options: Dict[str, str] = {}
|
||
if not raw:
|
||
return options
|
||
try:
|
||
tokens = shlex.split(raw)
|
||
except ValueError:
|
||
tokens = raw.split()
|
||
for token in tokens:
|
||
if "=" in token:
|
||
key, value = token.split("=", 1)
|
||
options[key.strip()] = value.strip()
|
||
else:
|
||
options[token.strip()] = "true"
|
||
return options
|
||
|
||
|
||
def str_to_bool(value) -> bool:
|
||
if isinstance(value, bool):
|
||
return value
|
||
if value is None:
|
||
return False
|
||
return str(value).strip().lower() in {"1", "true", "yes", "on"}
|
||
|
||
|
||
def smart_encode_url(url: str, safe_path="/:@&=%+,$-~", safe_query="/:@&=%+,$-~"):
|
||
try:
|
||
parts = urllib.parse.urlsplit(url)
|
||
except ValueError:
|
||
return url
|
||
path = urllib.parse.quote(parts.path or "/", safe=safe_path)
|
||
query = urllib.parse.quote(parts.query, safe=safe_query)
|
||
fragment = urllib.parse.quote(parts.fragment, safe=safe_query)
|
||
return urllib.parse.urlunsplit((parts.scheme, parts.netloc, path, query, fragment))
|
||
|
||
|
||
def origin_url(url: str) -> str:
|
||
"""scheme://netloc/ — httpx build base; real request-target is set separately."""
|
||
parts = urllib.parse.urlsplit(url)
|
||
return urllib.parse.urlunsplit((parts.scheme, parts.netloc, "/", "", ""))
|
||
|
||
|
||
def wire_request_target(url: str, absolute_form: bool = False) -> bytes:
|
||
"""
|
||
HTTP request-target on the wire (curl --path-as-is semantics, no fragment).
|
||
absolute_form: HTTP forward-proxy + plaintext http:// requires absolute-form.
|
||
"""
|
||
parts = urllib.parse.urlsplit(url)
|
||
path = parts.path or "/"
|
||
if absolute_form:
|
||
target = urllib.parse.urlunsplit((parts.scheme, parts.netloc, path, parts.query, ""))
|
||
elif parts.query:
|
||
target = f"{path}?{parts.query}"
|
||
else:
|
||
target = path
|
||
return target.encode("utf-8", errors="surrogateescape")
|
||
|
||
|
||
def resolve_forward_proxy(explicit_proxy: str, trust_env: bool, url: str) -> str:
|
||
"""Return the proxy that the custom transport must actually use."""
|
||
if explicit_proxy:
|
||
return explicit_proxy
|
||
if not trust_env:
|
||
return ""
|
||
parsed = urllib.parse.urlsplit(url)
|
||
host = parsed.hostname or ""
|
||
# urllib handles NO_PROXY/no_proxy matching, including suffixes and '*'.
|
||
if host and urllib.request.proxy_bypass(host):
|
||
return ""
|
||
proxies = urllib.request.getproxies()
|
||
scheme = (parsed.scheme or "").lower()
|
||
return proxies.get(scheme) or proxies.get("all") or ""
|
||
|
||
|
||
def needs_http_absolute_form(proxy: str, url: str) -> bool:
|
||
"""
|
||
Keep the custom raw target in origin-form.
|
||
|
||
httpcore adds the scheme/authority itself when an HTTP forward proxy needs
|
||
absolute-form. Supplying an absolute raw_path here would duplicate the URL.
|
||
"""
|
||
return False
|
||
|
||
|
||
def display_url_for_target(url: str, wire_target: bytes) -> str:
|
||
target = wire_target.decode("utf-8", errors="replace")
|
||
if target.startswith("http://") or target.startswith("https://"):
|
||
return target
|
||
parts = urllib.parse.urlsplit(url)
|
||
return urllib.parse.urlunsplit((parts.scheme, parts.netloc, "", "", "")).rstrip("/") + (
|
||
target if target.startswith("/") else "/" + target
|
||
)
|
||
|
||
|
||
def resolve_url_path_as_is(base_url: str, location: str) -> str:
|
||
"""
|
||
Resolve a Location header without RFC remove_dot_segments.
|
||
Keeps literal '.' / '..' in both absolute and relative references.
|
||
"""
|
||
location = (location or "").strip()
|
||
if not location:
|
||
raise ValueError("Redirect missing Location header")
|
||
if "#" in location:
|
||
location = location.split("#", 1)[0]
|
||
loc = urllib.parse.urlsplit(location)
|
||
base = urllib.parse.urlsplit(base_url)
|
||
|
||
if loc.scheme:
|
||
path = loc.path or "/"
|
||
return urllib.parse.urlunsplit((loc.scheme, loc.netloc, path, loc.query, ""))
|
||
|
||
if location.startswith("//"):
|
||
# protocol-relative
|
||
rest = location[2:]
|
||
if "/" in rest:
|
||
netloc, path_and_more = rest.split("/", 1)
|
||
path_and_more = "/" + path_and_more
|
||
else:
|
||
netloc, path_and_more = rest, "/"
|
||
rel = urllib.parse.urlsplit(path_and_more)
|
||
return urllib.parse.urlunsplit((base.scheme, netloc, rel.path or "/", rel.query, ""))
|
||
|
||
if loc.path.startswith("/"):
|
||
path = loc.path or "/"
|
||
query = loc.query
|
||
else:
|
||
base_path = base.path or "/"
|
||
if not base_path.endswith("/"):
|
||
base_dir = base_path.rsplit("/", 1)[0] + "/"
|
||
else:
|
||
base_dir = base_path
|
||
path = base_dir + loc.path
|
||
query = loc.query if ("?" in location) else ""
|
||
|
||
return urllib.parse.urlunsplit((base.scheme, base.netloc, path or "/", query, ""))
|
||
|
||
|
||
def redirect_method(method: str, status_code: int) -> str:
|
||
"""Mirror httpx redirect method switching (301/302 POST->GET, 303 -> GET)."""
|
||
method = (method or "GET").upper()
|
||
if status_code == 303 and method != "HEAD":
|
||
return "GET"
|
||
if status_code in {301, 302} and method == "POST":
|
||
return "GET"
|
||
return method
|
||
|
||
|
||
REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308}
|
||
|
||
|
||
class PathAsIsTransport(httpx.HTTPTransport):
|
||
"""
|
||
Transport that honors extensions['path_as_is_target'] as the HTTP request-target.
|
||
Equivalent to curl --path-as-is (do not squash /./ or /../).
|
||
"""
|
||
|
||
def handle_request(self, request: "httpx.Request") -> "httpx.Response":
|
||
target = request.extensions.get("path_as_is_target")
|
||
if target is None:
|
||
return super().handle_request(request)
|
||
if isinstance(target, str):
|
||
target = target.encode("utf-8", errors="surrogateescape")
|
||
|
||
original_url = request.url
|
||
|
||
class _PathAsIsURL:
|
||
__slots__ = ("_base", "raw_path")
|
||
|
||
def __init__(self, base, raw_path):
|
||
self._base = base
|
||
self.raw_path = raw_path
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(self._base, name)
|
||
|
||
request.url = _PathAsIsURL(original_url, target)
|
||
try:
|
||
return super().handle_request(request)
|
||
finally:
|
||
request.url = original_url
|
||
|
||
|
||
def send_path_as_is(
|
||
client: "httpx.Client",
|
||
*,
|
||
method: str,
|
||
url: str,
|
||
headers: "httpx.Headers",
|
||
content: bytes = None,
|
||
follow_redirects: bool = False,
|
||
max_redirects: int = 20,
|
||
forward_proxy: str = "",
|
||
):
|
||
"""
|
||
Send request with path-as-is semantics. Optionally follow redirects the same way.
|
||
Returns (response, final_url, final_wire_target, history).
|
||
Caller must close the returned response.
|
||
"""
|
||
history = []
|
||
current_method = (method or "GET").upper()
|
||
current_url = url
|
||
current_content = content
|
||
# Copy so redirect Host stripping does not mutate caller's headers permanently
|
||
current_headers = httpx.Headers(headers)
|
||
|
||
for hop in range(max_redirects + 1):
|
||
absolute = needs_http_absolute_form(forward_proxy, current_url)
|
||
wire_target = wire_request_target(current_url, absolute_form=absolute)
|
||
build_kwargs = {
|
||
"method": current_method,
|
||
"url": origin_url(current_url),
|
||
"headers": current_headers,
|
||
}
|
||
if current_content is not None:
|
||
build_kwargs["content"] = current_content
|
||
|
||
request = client.build_request(**build_kwargs)
|
||
request.extensions["path_as_is_target"] = wire_target
|
||
response = client.send(request, stream=True)
|
||
|
||
is_redirect = response.status_code in REDIRECT_STATUS_CODES
|
||
if (not follow_redirects) or (not is_redirect):
|
||
try:
|
||
response.history = list(history)
|
||
except Exception:
|
||
pass
|
||
return response, current_url, wire_target, history
|
||
|
||
if hop >= max_redirects:
|
||
response.close()
|
||
for item in history:
|
||
item.close()
|
||
raise httpx.TooManyRedirects(
|
||
f"Exceeded maximum allowed redirects ({max_redirects})",
|
||
request=request,
|
||
)
|
||
|
||
location = response.headers.get("Location")
|
||
if not location:
|
||
try:
|
||
response.history = list(history)
|
||
except Exception:
|
||
pass
|
||
return response, current_url, wire_target, history
|
||
|
||
next_url = resolve_url_path_as_is(current_url, location)
|
||
next_method = redirect_method(current_method, response.status_code)
|
||
|
||
# Drain body so the connection can be reused, then keep in history
|
||
try:
|
||
response.read()
|
||
except Exception:
|
||
pass
|
||
history.append(response)
|
||
|
||
# Drop body when method switches to GET/HEAD (httpx behavior)
|
||
if next_method in {"GET", "HEAD"} and next_method != current_method:
|
||
current_content = None
|
||
elif response.status_code in {301, 302, 303} and next_method == "GET":
|
||
current_content = None
|
||
|
||
# Host header must match next authority
|
||
if "Host" in current_headers:
|
||
del current_headers["Host"]
|
||
|
||
current_method = next_method
|
||
current_url = next_url
|
||
|
||
raise httpx.TooManyRedirects(
|
||
f"Exceeded maximum allowed redirects ({max_redirects})",
|
||
request=request,
|
||
)
|
||
|
||
|
||
def explain_request_error(exc: BaseException, url: str = "", proxy: str = "") -> str:
|
||
text = str(exc)
|
||
lowered = text.lower()
|
||
if any(
|
||
marker in lowered
|
||
for marker in (
|
||
"temporary failure in name resolution",
|
||
"name or service not known",
|
||
"nodename nor servname provided",
|
||
"getaddrinfo failed",
|
||
)
|
||
):
|
||
target_host = urllib.parse.urlsplit(url).hostname or "目标主机"
|
||
if proxy:
|
||
proxy_host = urllib.parse.urlsplit(proxy).hostname or "代理服务器"
|
||
return (
|
||
text
|
||
+ f"\nHint: DNS 解析失败。当前请求经环境/显式代理 {proxy_host} 转发；"
|
||
"请确认代理进程可达，并检查代理地址本身能否解析。"
|
||
)
|
||
return (
|
||
text
|
||
+ f"\nHint: DNS 暂时无法解析 {target_host}。请检查域名、容器/主机 DNS 与网络；"
|
||
"若运行环境要求代理，请设置 HTTPS_PROXY/HTTP_PROXY，或通过 proxy 参数显式传入。"
|
||
)
|
||
if "unexpected_eof" in lowered or "eof occurred in violation of protocol" in lowered:
|
||
return (
|
||
text
|
||
+ "\nHint: 服务器在 TLS 层直接断开（常见于 WAF/反代拒绝异常路径、URI 过长或敏感路径探测）。"
|
||
"可尝试：缩短 ../ 层数、改用 %2e%2e 编码、加 --allow-insecure、或经代理观察原始响应。"
|
||
)
|
||
return text
|
||
|
||
|
||
def quote_form_component_preserving_pct(value: str) -> str:
|
||
normalized = urllib.parse.unquote(value)
|
||
return urllib.parse.quote_plus(normalized, safe="")
|
||
|
||
|
||
def encode_form_data(data: str) -> str:
|
||
if not data:
|
||
return data
|
||
|
||
def find_key_value_pairs(text):
|
||
pairs = []
|
||
i = 0
|
||
text_len = len(text)
|
||
|
||
while i < text_len:
|
||
while i < text_len and text[i] in " \t\n\r":
|
||
i += 1
|
||
if i >= text_len:
|
||
break
|
||
|
||
key_start = i
|
||
while i < text_len and text[i] != "=":
|
||
i += 1
|
||
|
||
if i >= text_len:
|
||
remaining = text[key_start:].strip()
|
||
if remaining:
|
||
pairs.append((None, remaining))
|
||
break
|
||
|
||
key = text[key_start:i].strip()
|
||
i += 1
|
||
|
||
if not key:
|
||
continue
|
||
|
||
value_start = i
|
||
value_end = text_len
|
||
|
||
j = value_start
|
||
while j < text_len:
|
||
if text[j] == "&":
|
||
k = j + 1
|
||
while k < text_len and text[k] in " \t\n\r":
|
||
k += 1
|
||
m = k
|
||
while m < text_len and text[m] not in "=&":
|
||
m += 1
|
||
if m < text_len and text[m] == "=":
|
||
value_end = j
|
||
i = j + 1
|
||
break
|
||
j += 1
|
||
|
||
value = text[value_start:value_end]
|
||
pairs.append((key, value))
|
||
|
||
if value_end < text_len:
|
||
i = value_end + 1
|
||
else:
|
||
break
|
||
|
||
return pairs
|
||
|
||
pairs = find_key_value_pairs(data)
|
||
parts = []
|
||
|
||
for key, value in pairs:
|
||
if key is None:
|
||
parts.append(quote_form_component_preserving_pct(value))
|
||
else:
|
||
encoded_key = quote_form_component_preserving_pct(key)
|
||
encoded_value = quote_form_component_preserving_pct(value)
|
||
parts.append(f"{encoded_key}={encoded_value}")
|
||
|
||
return "&".join(parts)
|
||
|
||
|
||
def should_encode_form(headers: httpx.Headers, data: str) -> bool:
|
||
if not data:
|
||
return False
|
||
if not headers:
|
||
return False
|
||
content_type = headers.get("Content-Type")
|
||
if not content_type:
|
||
return False
|
||
return "application/x-www-form-urlencoded" in content_type.lower()
|
||
|
||
|
||
def extract_charset_from_content_type(content_type: str) -> str:
|
||
if not content_type:
|
||
return ""
|
||
parts = content_type.split(";")
|
||
for part in parts[1:]:
|
||
lowered = part.strip().lower()
|
||
if lowered.startswith("charset="):
|
||
return part.split("=", 1)[1].strip().strip('"').strip("'")
|
||
return ""
|
||
|
||
|
||
def extract_declared_charset_from_body(data: bytes) -> str:
|
||
if not data:
|
||
return ""
|
||
sample = data[:16384]
|
||
try:
|
||
sample_text = sample.decode("iso-8859-1", errors="ignore")
|
||
except UnicodeDecodeError:
|
||
return ""
|
||
for line in sample_text.splitlines():
|
||
lowered = line.lower()
|
||
if "content-type:" in lowered and "charset=" in lowered:
|
||
charset_index = lowered.index("charset=") + len("charset=")
|
||
remainder = line[charset_index:]
|
||
for separator in [";", " ", "\t"]:
|
||
remainder = remainder.split(separator)[0]
|
||
return remainder.strip().strip('"').strip("'")
|
||
meta_match = re.search(r'charset=["\']?([a-zA-Z0-9_\-.:]+)', sample_text, re.IGNORECASE)
|
||
if meta_match:
|
||
return meta_match.group(1)
|
||
return ""
|
||
|
||
|
||
def decode_body_bytes(data: bytes, headers: httpx.Headers, user_encoding: str = ""):
|
||
attempts = []
|
||
if user_encoding:
|
||
attempts.append(("user", user_encoding))
|
||
header_declared = extract_charset_from_content_type(headers.get("Content-Type", "")) if headers else ""
|
||
if header_declared:
|
||
attempts.append(("header", header_declared))
|
||
body_declared = extract_declared_charset_from_body(data) if not header_declared else ""
|
||
if body_declared:
|
||
attempts.append(("body", body_declared))
|
||
for source, encoding in attempts:
|
||
enc = (encoding or "").strip()
|
||
if not enc:
|
||
continue
|
||
try:
|
||
return data.decode(enc), enc, source
|
||
except (LookupError, UnicodeDecodeError):
|
||
continue
|
||
if charset_from_bytes is not None and data:
|
||
best = charset_from_bytes(data).best()
|
||
if best and best.encoding:
|
||
try:
|
||
return data.decode(best.encoding), best.encoding, "detected"
|
||
except (LookupError, UnicodeDecodeError):
|
||
pass
|
||
if chardet is not None and data:
|
||
detection = chardet.detect(data)
|
||
encoding = detection.get("encoding")
|
||
if encoding:
|
||
try:
|
||
return data.decode(encoding), encoding, "detected"
|
||
except (LookupError, UnicodeDecodeError):
|
||
pass
|
||
try:
|
||
return data.decode("utf-8"), "utf-8", "fallback"
|
||
except UnicodeDecodeError:
|
||
return data.decode("utf-8", errors="replace"), "utf-8", "fallback"
|
||
|
||
|
||
def prepare_body(data: str, headers: httpx.Headers, debug: bool = False):
|
||
meta = {
|
||
"source": "inline",
|
||
"mode": "none",
|
||
"length": 0,
|
||
"charset": None,
|
||
"encoded": False,
|
||
}
|
||
if not data:
|
||
return None, meta
|
||
if data.startswith("@"):
|
||
path = data[1:]
|
||
if path == "-":
|
||
payload = sys.stdin.buffer.read()
|
||
meta["source"] = "stdin"
|
||
else:
|
||
path = os.path.expanduser(path)
|
||
if not os.path.isfile(path):
|
||
raise FileNotFoundError(f"Body file not found: {path}")
|
||
with open(path, "rb") as fh:
|
||
payload = fh.read()
|
||
meta["source"] = path
|
||
meta["mode"] = "binary"
|
||
meta["length"] = len(payload)
|
||
return payload, meta
|
||
stripped = data.strip()
|
||
content_type = headers.get("Content-Type")
|
||
if not content_type:
|
||
guessed = ""
|
||
if stripped.startswith("{") or stripped.startswith("["):
|
||
guessed = "application/json"
|
||
elif "=" in stripped and "&" in stripped:
|
||
guessed = "application/x-www-form-urlencoded"
|
||
elif stripped.startswith("<"):
|
||
guessed = "application/xml"
|
||
if guessed:
|
||
headers["Content-Type"] = guessed
|
||
content_type = guessed
|
||
processed = data
|
||
if should_encode_form(headers, data):
|
||
processed = encode_form_data(data)
|
||
meta["mode"] = "form-urlencoded"
|
||
meta["encoded"] = True
|
||
else:
|
||
normalized = (headers.get("Content-Type") or "").lower()
|
||
if normalized.startswith("application/json"):
|
||
meta["mode"] = "json"
|
||
elif normalized.startswith("multipart/form-data"):
|
||
meta["mode"] = "multipart"
|
||
elif normalized.startswith("text/") or "xml" in normalized:
|
||
meta["mode"] = "text"
|
||
else:
|
||
meta["mode"] = "raw"
|
||
content_type = headers.get("Content-Type")
|
||
charset = extract_charset_from_content_type(content_type) if content_type else ""
|
||
if not charset and meta["mode"] in ("json", "text", "form-urlencoded"):
|
||
charset = "utf-8"
|
||
base = (content_type or "text/plain").split(";")[0].strip()
|
||
headers["Content-Type"] = f"{base}; charset={charset}"
|
||
elif not charset:
|
||
charset = "utf-8"
|
||
body_bytes = processed.encode(charset, errors="surrogatepass")
|
||
meta["charset"] = charset
|
||
meta["length"] = len(body_bytes)
|
||
if debug:
|
||
print("\n===== Debug: Body Encoding =====")
|
||
print(f"Mode: {meta['mode']}")
|
||
print(f"Charset: {charset}")
|
||
print(f"Length: {meta['length']} bytes")
|
||
print(f"Source: {meta['source']}")
|
||
if meta["encoded"]:
|
||
print("Form data was URL-encoded prior to sending.")
|
||
return body_bytes, meta
|
||
|
||
|
||
def probe_connection(url: str, timeout: float, verify_tls: bool, skip: bool):
|
||
metrics = {}
|
||
if skip:
|
||
return metrics
|
||
parsed = urllib.parse.urlsplit(url)
|
||
host = parsed.hostname
|
||
if not host:
|
||
return metrics
|
||
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
||
dns_start = time.perf_counter()
|
||
try:
|
||
addr_info = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
|
||
except socket.gaierror:
|
||
return metrics
|
||
dns_time = time.perf_counter() - dns_start
|
||
family, socktype, proto, _, sockaddr = addr_info[0]
|
||
sock = socket.socket(family, socktype, proto)
|
||
sock.settimeout(timeout or 30.0)
|
||
try:
|
||
connect_start = time.perf_counter()
|
||
sock.connect(sockaddr)
|
||
connect_time = time.perf_counter() - connect_start
|
||
tls_time = 0.0
|
||
if parsed.scheme == "https":
|
||
ctx = ssl.create_default_context()
|
||
if not verify_tls:
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = ssl.CERT_NONE
|
||
try:
|
||
tls_start = time.perf_counter()
|
||
tls_sock = ctx.wrap_socket(sock, server_hostname=host, do_handshake_on_connect=False)
|
||
try:
|
||
tls_sock.do_handshake()
|
||
tls_time = time.perf_counter() - tls_start
|
||
finally:
|
||
tls_sock.close()
|
||
except ssl.SSLError:
|
||
tls_time = 0.0
|
||
else:
|
||
sock.close()
|
||
except OSError:
|
||
sock.close()
|
||
return {"dns_lookup": dns_time}
|
||
metrics["dns_lookup"] = dns_time
|
||
metrics["tcp_connect"] = connect_time
|
||
metrics["tls_handshake"] = tls_time
|
||
metrics["pretransfer"] = dns_time + connect_time + tls_time
|
||
return metrics
|
||
|
||
|
||
def format_metric_value(key: str, value):
|
||
if value is None:
|
||
return "n/a"
|
||
if key in {"http_code", "redirects", "size_download"}:
|
||
return str(int(value))
|
||
if key == "speed_download":
|
||
return f"{value:.2f} B/s"
|
||
return f"{value:.6f}s"
|
||
|
||
|
||
def summarize(values):
|
||
if not values:
|
||
return None
|
||
count = len(values)
|
||
total = sum(values)
|
||
return min(values), total / count, max(values)
|
||
|
||
|
||
def render_request_overview(
|
||
method: str,
|
||
url: str,
|
||
headers: httpx.Headers,
|
||
body_meta: Dict[str, str],
|
||
wire_target: bytes = None,
|
||
input_url: str = None,
|
||
):
|
||
items = list(headers.items())
|
||
print("\n===== Prepared Request =====")
|
||
print(f"Method: {method}")
|
||
print(f"URL: {url}")
|
||
if input_url and input_url != url:
|
||
print(f"Input URL: {input_url}")
|
||
if wire_target is not None:
|
||
print(f"Wire request-target: {wire_target.decode('utf-8', errors='replace')}")
|
||
print("Note: path-as-is enabled (literal path/query; equivalent to curl --path-as-is).")
|
||
print(f"Headers ({len(items)} total):")
|
||
for key, value in items:
|
||
print(f" {key}: {value}")
|
||
if body_meta["length"]:
|
||
charset = body_meta.get("charset") or "n/a"
|
||
print(f"Body: {body_meta['length']} bytes ({body_meta['mode']}, charset={charset}, source={body_meta['source']})")
|
||
else:
|
||
print("Body: <empty>")
|
||
|
||
|
||
def compile_response_filter(pattern: str, ignore_case: bool):
|
||
flags = 0
|
||
if ignore_case:
|
||
flags |= re.IGNORECASE
|
||
try:
|
||
return re.compile(pattern, flags)
|
||
except re.error as exc:
|
||
print(f"Invalid response_filter regex: {exc}", file=sys.stderr)
|
||
sys.exit(2)
|
||
|
||
|
||
def truncate_utf8(text: str, max_bytes: int) -> Tuple[str, bool]:
|
||
if max_bytes <= 0 or not text:
|
||
return text, False
|
||
encoded = text.encode("utf-8", errors="replace")
|
||
if len(encoded) <= max_bytes:
|
||
return text, False
|
||
truncated = encoded[:max_bytes].decode("utf-8", errors="ignore")
|
||
return truncated, True
|
||
|
||
|
||
def cap_line_entries(entries: List[Tuple[int, str]], max_lines: int) -> Tuple[List[Tuple[int, str]], bool]:
|
||
if max_lines <= 0 or len(entries) <= max_lines:
|
||
return entries, False
|
||
return entries[:max_lines], True
|
||
|
||
|
||
def expand_line_context(line_numbers: List[int], total_lines: int, context: int) -> List[int]:
|
||
if context <= 0:
|
||
return sorted(set(line_numbers))
|
||
included = set()
|
||
for num in line_numbers:
|
||
start = max(1, num - context)
|
||
end = min(total_lines, num + context)
|
||
for i in range(start, end + 1):
|
||
included.add(i)
|
||
return sorted(included)
|
||
|
||
|
||
def format_line_entries(lines: List[str], indices: List[int], ellipsis_gaps: bool = True) -> str:
|
||
if not indices:
|
||
return ""
|
||
chunks = []
|
||
prev = None
|
||
for num in indices:
|
||
if ellipsis_gaps and prev is not None and num > prev + 1:
|
||
chunks.append(" ...")
|
||
chunks.append(f" L{num}: {lines[num - 1]}")
|
||
prev = num
|
||
return "\n".join(chunks)
|
||
|
||
|
||
def filter_body_by_lines(
|
||
lines: List[str],
|
||
compiled: "re.Pattern",
|
||
invert: bool,
|
||
context_lines: int,
|
||
max_lines: int,
|
||
) -> Tuple[str, Dict[str, object]]:
|
||
matched_nums = []
|
||
for idx, line in enumerate(lines, start=1):
|
||
hit = compiled.search(line) is not None
|
||
if invert:
|
||
hit = not hit
|
||
if hit:
|
||
matched_nums.append(idx)
|
||
total = len(lines)
|
||
meta = {
|
||
"mode": "line",
|
||
"total_lines": total,
|
||
"matched_lines": len(matched_nums),
|
||
"invert": invert,
|
||
"truncated": False,
|
||
"byte_truncated": False,
|
||
}
|
||
if not matched_nums:
|
||
return "", meta
|
||
display_nums = expand_line_context(matched_nums, total, context_lines)
|
||
entries = [(n, lines[n - 1]) for n in display_nums]
|
||
entries, line_capped = cap_line_entries(entries, max_lines)
|
||
meta["truncated"] = line_capped
|
||
meta["display_lines"] = len(entries)
|
||
return format_line_entries(lines, [n for n, _ in entries], ellipsis_gaps=context_lines > 0), meta
|
||
|
||
|
||
def filter_body_multiline(
|
||
text: str,
|
||
compiled: "re.Pattern",
|
||
invert: bool,
|
||
max_lines: int,
|
||
dotall: bool,
|
||
) -> Tuple[str, Dict[str, object]]:
|
||
flags = compiled.flags
|
||
if dotall:
|
||
pattern = re.compile(compiled.pattern, flags | re.DOTALL | re.MULTILINE)
|
||
else:
|
||
pattern = re.compile(compiled.pattern, flags | re.MULTILINE)
|
||
matches = list(pattern.finditer(text))
|
||
if invert:
|
||
if matches:
|
||
return "", {"mode": "multiline" if not dotall else "full", "total_lines": text.count("\n") + (1 if text else 0), "matched_lines": 0, "invert": True, "truncated": False, "byte_truncated": False}
|
||
output = text
|
||
meta = {"mode": "multiline" if not dotall else "full", "matched_lines": 1, "invert": True, "truncated": False, "byte_truncated": False}
|
||
lines = text.splitlines()
|
||
if max_lines > 0 and len(lines) > max_lines:
|
||
output = "\n".join(lines[:max_lines])
|
||
meta["truncated"] = True
|
||
meta["total_lines"] = len(lines)
|
||
meta["display_lines"] = min(len(lines), max_lines) if max_lines > 0 else len(lines)
|
||
return output, meta
|
||
chunks = []
|
||
for match in matches:
|
||
snippet = match.group(0)
|
||
if "\n" in snippet:
|
||
snippet = snippet.replace("\n", "\\n")
|
||
start_line = text.count("\n", 0, match.start()) + 1
|
||
chunks.append((start_line, f" @{start_line}: {snippet}"))
|
||
entries, line_capped = cap_line_entries(chunks, max_lines if max_lines > 0 else len(chunks))
|
||
meta = {
|
||
"mode": "multiline" if not dotall else "full",
|
||
"total_lines": text.count("\n") + (1 if text else 0),
|
||
"matched_lines": len(matches),
|
||
"invert": False,
|
||
"truncated": line_capped,
|
||
"byte_truncated": False,
|
||
"display_lines": len(entries),
|
||
}
|
||
return "\n".join(line for _, line in entries), meta
|
||
|
||
|
||
def apply_body_limits_plain(text: str, max_lines: int) -> Tuple[str, Dict[str, object]]:
|
||
lines = text.splitlines()
|
||
meta = {
|
||
"mode": "plain",
|
||
"total_lines": len(lines),
|
||
"matched_lines": len(lines),
|
||
"invert": False,
|
||
"truncated": False,
|
||
"byte_truncated": False,
|
||
"display_lines": len(lines),
|
||
}
|
||
output = text
|
||
if max_lines > 0 and len(lines) > max_lines:
|
||
output = "\n".join(lines[:max_lines])
|
||
meta["truncated"] = True
|
||
meta["display_lines"] = max_lines
|
||
return output, meta
|
||
|
||
|
||
def format_response_body_output(
|
||
decoded_body: str,
|
||
filter_pattern: str,
|
||
filter_mode: str,
|
||
filter_invert: bool,
|
||
filter_ignore_case: bool,
|
||
max_lines: int,
|
||
max_bytes: int,
|
||
preview_lines: int,
|
||
context_lines: int,
|
||
compiled_filter=None,
|
||
) -> Tuple[str, Dict[str, object]]:
|
||
text = decoded_body.rstrip("\r\n")
|
||
if not text:
|
||
return "", {"mode": "empty", "total_lines": 0, "matched_lines": 0, "invert": filter_invert, "truncated": False, "byte_truncated": False, "display_lines": 0}
|
||
|
||
lines = text.splitlines()
|
||
mode = (filter_mode or "line").strip().lower()
|
||
if mode not in {"line", "multiline", "full"}:
|
||
mode = "line"
|
||
|
||
if filter_pattern:
|
||
compiled = compiled_filter or compile_response_filter(filter_pattern, filter_ignore_case)
|
||
if mode == "line":
|
||
output, meta = filter_body_by_lines(lines, compiled, filter_invert, context_lines, max_lines)
|
||
else:
|
||
output, meta = filter_body_multiline(text, compiled, filter_invert, max_lines, dotall=(mode == "full"))
|
||
meta["filter_pattern"] = filter_pattern
|
||
if not output and not filter_invert:
|
||
preview = min(max(preview_lines, 0), len(lines))
|
||
if preview > 0:
|
||
preview_text = format_line_entries(lines, list(range(1, preview + 1)), ellipsis_gaps=False)
|
||
preview_text, byte_truncated = truncate_utf8(preview_text, max_bytes)
|
||
return preview_text, {
|
||
**meta,
|
||
"preview": True,
|
||
"matched_lines": 0,
|
||
"display_lines": preview,
|
||
"byte_truncated": byte_truncated,
|
||
}
|
||
return "", {**meta, "preview": False, "matched_lines": 0, "display_lines": 0}
|
||
else:
|
||
output, meta = apply_body_limits_plain(text, max_lines)
|
||
|
||
output, byte_truncated = truncate_utf8(output, max_bytes)
|
||
if byte_truncated:
|
||
meta["byte_truncated"] = True
|
||
return output, meta
|
||
|
||
|
||
def print_response_body_summary(meta: Dict[str, object]):
|
||
mode = meta.get("mode")
|
||
if mode == "empty":
|
||
return
|
||
parts = [f"mode={mode}"]
|
||
if meta.get("filter_pattern"):
|
||
parts.append(f"pattern={meta['filter_pattern']!r}")
|
||
if meta.get("invert"):
|
||
parts.append("invert=true")
|
||
total = meta.get("total_lines")
|
||
matched = meta.get("matched_lines")
|
||
displayed = meta.get("display_lines")
|
||
if total is not None and matched is not None:
|
||
parts.append(f"matched {matched}/{total} lines")
|
||
if displayed is not None:
|
||
parts.append(f"showing {displayed}")
|
||
if meta.get("preview"):
|
||
parts.append("preview on zero match")
|
||
if meta.get("truncated"):
|
||
parts.append("line cap applied")
|
||
if meta.get("byte_truncated"):
|
||
parts.append("byte cap applied")
|
||
print(f"[body] {' | '.join(parts)}")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Pure Python HTTP testing helper powered by httpx")
|
||
parser.add_argument("--url", required=True)
|
||
parser.add_argument("--method", default="GET")
|
||
parser.add_argument("--data", default="")
|
||
parser.add_argument("--headers", default="", type=str)
|
||
parser.add_argument("--cookies", default="")
|
||
parser.add_argument("--user-agent", dest="user_agent", default="")
|
||
parser.add_argument("--proxy", default="")
|
||
parser.add_argument("--timeout", default="")
|
||
parser.add_argument("--repeat", type=int, default=1)
|
||
parser.add_argument("--delay", default="0")
|
||
parser.add_argument("--additional-args", dest="additional_args", default="")
|
||
parser.add_argument("--http2", dest="http2", action="store_true")
|
||
parser.add_argument("--no-http2", dest="http2", action="store_false")
|
||
parser.add_argument("--action", default="")
|
||
parser.add_argument("--include-headers", dest="include_headers", action="store_true")
|
||
parser.add_argument("--no-include-headers", dest="include_headers", action="store_false")
|
||
parser.add_argument("--auto-encode-url", dest="auto_encode_url", action="store_true")
|
||
parser.add_argument("--no-auto-encode-url", dest="auto_encode_url", action="store_false")
|
||
parser.add_argument("--follow-redirects", dest="follow_redirects", action="store_true")
|
||
parser.add_argument("--allow-insecure", dest="allow_insecure", action="store_true")
|
||
parser.add_argument("--verbose-output", dest="verbose_output", action="store_true")
|
||
parser.add_argument("--show-command", dest="show_command", action="store_true")
|
||
parser.add_argument("--show-summary", dest="show_summary", action="store_true")
|
||
parser.add_argument("--debug", dest="debug", action="store_true")
|
||
parser.add_argument("--response-encoding", dest="response_encoding", default="")
|
||
parser.add_argument("--download", dest="download", default="")
|
||
parser.add_argument("--response-filter", dest="response_filter", default="")
|
||
parser.add_argument("--response-filter-mode", dest="response_filter_mode", default="line")
|
||
parser.add_argument("--response-filter-invert", dest="response_filter_invert", action="store_true")
|
||
parser.add_argument("--no-response-filter-invert", dest="response_filter_invert", action="store_false")
|
||
parser.add_argument("--response-filter-ignore-case", dest="response_filter_ignore_case", action="store_true")
|
||
parser.add_argument("--no-response-filter-ignore-case", dest="response_filter_ignore_case", action="store_false")
|
||
parser.add_argument("--response-max-lines", dest="response_max_lines", type=int, default=0)
|
||
parser.add_argument("--response-max-bytes", dest="response_max_bytes", type=int, default=0)
|
||
parser.add_argument("--response-preview-lines", dest="response_preview_lines", type=int, default=5)
|
||
parser.add_argument("--response-context-lines", dest="response_context_lines", type=int, default=0)
|
||
parser.set_defaults(
|
||
include_headers=False,
|
||
auto_encode_url=False,
|
||
follow_redirects=False,
|
||
allow_insecure=False,
|
||
verbose_output=False,
|
||
show_command=False,
|
||
show_summary=False,
|
||
debug=False,
|
||
http2=False,
|
||
response_filter_invert=False,
|
||
response_filter_ignore_case=False,
|
||
)
|
||
# parse_known_args: tolerate legacy executor that appended bare "http2=true" tokens
|
||
args, unknown_args = parser.parse_known_args()
|
||
leaked_options = parse_additional_options(" ".join(unknown_args)) if unknown_args else {}
|
||
|
||
response_filter = (args.response_filter or "").strip()
|
||
response_max_lines = max(0, args.response_max_lines or 0)
|
||
response_max_bytes = max(0, args.response_max_bytes or 0)
|
||
response_preview_lines = max(0, args.response_preview_lines if args.response_preview_lines is not None else 5)
|
||
response_context_lines = max(0, args.response_context_lines or 0)
|
||
compiled_response_filter = None
|
||
if response_filter:
|
||
compiled_response_filter = compile_response_filter(
|
||
response_filter, args.response_filter_ignore_case
|
||
)
|
||
|
||
repeat = max(1, args.repeat)
|
||
try:
|
||
delay_between = float(args.delay or "0")
|
||
if delay_between < 0:
|
||
delay_between = 0.0
|
||
except ValueError:
|
||
delay_between = 0.0
|
||
|
||
encoded_url = smart_encode_url(args.url) if args.auto_encode_url else args.url
|
||
method = (args.method or "GET").upper()
|
||
trust_env = True
|
||
# 处理 headers：支持字典（JSON字符串）和字符串格式
|
||
# 框架会将 object 类型序列化为 JSON 字符串传递
|
||
headers_list = []
|
||
if args.headers:
|
||
headers_str = args.headers.strip()
|
||
# 优先尝试解析为 JSON（框架传递的字典会被序列化为 JSON）
|
||
if headers_str.startswith("{") or headers_str.startswith("["):
|
||
try:
|
||
parsed = json.loads(headers_str)
|
||
if isinstance(parsed, dict):
|
||
# 字典格式：直接转换为 (key, value) 元组列表
|
||
headers_list = [(str(k).strip(), str(v).strip()) for k, v in parsed.items()]
|
||
elif isinstance(parsed, list):
|
||
# 数组格式：使用原有的 parse_headers 函数处理
|
||
headers_list = parse_headers(headers_str)
|
||
else:
|
||
headers_list = parse_headers(headers_str)
|
||
except (json.JSONDecodeError, ValueError):
|
||
# JSON 解析失败，回退到原有的字符串解析逻辑
|
||
headers_list = parse_headers(headers_str)
|
||
else:
|
||
# 非 JSON 格式，使用原有的字符串解析逻辑（向后兼容）
|
||
headers_list = parse_headers(headers_str)
|
||
headers = httpx.Headers(headers_list)
|
||
if args.user_agent:
|
||
headers["User-Agent"] = args.user_agent
|
||
|
||
body_bytes = None
|
||
body_meta = {
|
||
"source": "inline",
|
||
"mode": "none",
|
||
"length": 0,
|
||
"charset": None,
|
||
"encoded": False,
|
||
}
|
||
try:
|
||
body_bytes, body_meta = prepare_body(args.data, headers, args.debug)
|
||
except FileNotFoundError as exc:
|
||
print(f"Body preparation failed: {exc}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
cookie_dict = parse_cookies(args.cookies)
|
||
cookie_jar = httpx.Cookies()
|
||
for name, value in cookie_dict.items():
|
||
cookie_jar.set(name, value)
|
||
|
||
additional_options = parse_additional_options(args.additional_args)
|
||
additional_options.update(leaked_options)
|
||
if args.http2:
|
||
additional_options["http2"] = "true"
|
||
|
||
timeout_value = None
|
||
if args.timeout:
|
||
try:
|
||
timeout_value = float(args.timeout)
|
||
except ValueError:
|
||
timeout_value = None
|
||
timeout = httpx.Timeout(timeout_value or 60.0)
|
||
|
||
max_redirects = 20
|
||
if "max_redirects" in additional_options:
|
||
try:
|
||
max_redirects = int(additional_options["max_redirects"])
|
||
except ValueError:
|
||
pass
|
||
if "trust_env" in additional_options:
|
||
trust_env = str_to_bool(additional_options["trust_env"])
|
||
|
||
verify_option = not args.allow_insecure
|
||
if "verify" in additional_options:
|
||
value = additional_options["verify"]
|
||
lowered = value.strip().lower()
|
||
verify_option = str_to_bool(value) if lowered in {"true", "false", "1", "0", "yes", "no", "on", "off"} else value
|
||
|
||
http2 = str_to_bool(additional_options["http2"]) if "http2" in additional_options else False
|
||
if http2:
|
||
try:
|
||
import h2 # noqa: F401
|
||
except ImportError:
|
||
print(
|
||
"HTTP/2 requires the h2 package. Install with: pip install 'httpx[http2]' (or: pip install h2)",
|
||
file=sys.stderr,
|
||
)
|
||
sys.exit(1)
|
||
cert = additional_options.get("cert")
|
||
explicit_proxy = (args.proxy or "").strip()
|
||
forward_proxy = resolve_forward_proxy(explicit_proxy, trust_env, encoded_url)
|
||
|
||
transport_kwargs = {
|
||
"verify": verify_option,
|
||
"trust_env": trust_env,
|
||
"http2": http2,
|
||
}
|
||
if cert:
|
||
transport_kwargs["cert"] = cert
|
||
# Supplying a custom transport makes httpx.Client stop installing proxies
|
||
# from the environment. Pass the resolved env/explicit proxy directly so
|
||
# trust_env=true retains its documented behavior.
|
||
if forward_proxy:
|
||
transport_kwargs["proxy"] = forward_proxy
|
||
|
||
# Always path-as-is (security-testing default; matches curl --path-as-is)
|
||
initial_absolute = needs_http_absolute_form(forward_proxy, encoded_url)
|
||
initial_wire_target = wire_request_target(encoded_url, absolute_form=initial_absolute)
|
||
prepared_url = display_url_for_target(encoded_url, initial_wire_target)
|
||
|
||
client_kwargs = {
|
||
"timeout": timeout,
|
||
"verify": verify_option,
|
||
"follow_redirects": False, # redirects handled by send_path_as_is
|
||
"cookies": cookie_jar,
|
||
"trust_env": trust_env,
|
||
"transport": PathAsIsTransport(**transport_kwargs),
|
||
}
|
||
|
||
if args.show_command:
|
||
render_request_overview(
|
||
method,
|
||
prepared_url,
|
||
headers,
|
||
body_meta,
|
||
wire_target=initial_wire_target,
|
||
input_url=args.url if args.url != prepared_url else None,
|
||
)
|
||
|
||
aggregate = None
|
||
if args.show_summary:
|
||
aggregate = {key: [] for key in METRIC_KEYS}
|
||
aggregate["wall_time"] = []
|
||
|
||
exit_code = 0
|
||
verify_tls = verify_option if isinstance(verify_option, bool) else True
|
||
skip_probe = bool(forward_proxy)
|
||
|
||
client = httpx.Client(**client_kwargs)
|
||
try:
|
||
for run_index in range(repeat):
|
||
if run_index > 0 and delay_between > 0:
|
||
time.sleep(delay_between)
|
||
|
||
metrics = {key: None for key in METRIC_KEYS}
|
||
probe_metrics = probe_connection(encoded_url, timeout_value or 60.0, verify_tls, skip_probe)
|
||
metrics.update(probe_metrics)
|
||
|
||
start = time.perf_counter()
|
||
first_byte_time = None
|
||
body_buffer = bytearray()
|
||
wire_target = initial_wire_target
|
||
|
||
try:
|
||
response, final_url, wire_target, redirect_history = send_path_as_is(
|
||
client,
|
||
method=method,
|
||
url=encoded_url,
|
||
headers=headers,
|
||
content=body_bytes,
|
||
follow_redirects=args.follow_redirects,
|
||
max_redirects=max_redirects,
|
||
forward_proxy=forward_proxy,
|
||
)
|
||
try:
|
||
status_code = response.status_code
|
||
metrics["http_code"] = status_code
|
||
metrics["redirects"] = len(redirect_history)
|
||
http_version = response.http_version or "HTTP/1.1"
|
||
|
||
for chunk in response.iter_bytes():
|
||
if first_byte_time is None:
|
||
first_byte_time = time.perf_counter()
|
||
body_buffer.extend(chunk)
|
||
|
||
total_elapsed = time.perf_counter() - start
|
||
metrics["total"] = total_elapsed
|
||
metrics["ttfb"] = (first_byte_time - start) if first_byte_time else total_elapsed
|
||
metrics["size_download"] = len(body_buffer)
|
||
metrics["speed_download"] = (len(body_buffer) / total_elapsed) if total_elapsed > 0 else 0.0
|
||
if metrics.get("pretransfer") is None:
|
||
metrics["pretransfer"] = (metrics.get("dns_lookup") or 0.0) + (metrics.get("tcp_connect") or 0.0) + (metrics.get("tls_handshake") or 0.0)
|
||
|
||
decoded_body, used_encoding, encoding_source = decode_body_bytes(bytes(body_buffer), response.headers, args.response_encoding)
|
||
|
||
print(f"\n===== Response #{run_index + 1} =====")
|
||
download_target = (args.download or "").strip()
|
||
if download_target:
|
||
# 支持多次请求时通过 {i} 占位符区分文件名
|
||
if repeat > 1 and "{i}" in download_target:
|
||
filename = download_target.format(i=run_index + 1)
|
||
else:
|
||
filename = download_target
|
||
try:
|
||
if os.path.dirname(filename):
|
||
os.makedirs(os.path.dirname(filename), exist_ok=True)
|
||
with open(filename, "wb") as fh:
|
||
fh.write(body_buffer)
|
||
print(f"[saved response body to file: {filename}]")
|
||
except OSError as exc:
|
||
print(f"[failed to save response body to {filename}: {exc}]", file=sys.stderr)
|
||
if args.include_headers:
|
||
status_line = f"{http_version} {status_code} {response.reason_phrase}"
|
||
print(status_line)
|
||
for key, value in response.headers.items():
|
||
print(f"{key}: {value}")
|
||
print("")
|
||
output_body, body_output_meta = format_response_body_output(
|
||
decoded_body,
|
||
response_filter,
|
||
args.response_filter_mode,
|
||
args.response_filter_invert,
|
||
args.response_filter_ignore_case,
|
||
response_max_lines,
|
||
response_max_bytes,
|
||
response_preview_lines,
|
||
response_context_lines,
|
||
compiled_filter=compiled_response_filter,
|
||
)
|
||
has_filter_or_cap = bool(
|
||
response_filter or response_max_lines > 0 or response_max_bytes > 0
|
||
)
|
||
if has_filter_or_cap and body_output_meta.get("mode") != "empty":
|
||
print_response_body_summary(body_output_meta)
|
||
if body_output_meta.get("preview") and not body_output_meta.get("matched_lines"):
|
||
print("[body] no regex match; showing preview:")
|
||
if output_body:
|
||
print(output_body)
|
||
if body_output_meta.get("truncated") or body_output_meta.get("byte_truncated"):
|
||
omitted = (body_output_meta.get("total_lines") or 0) - (
|
||
body_output_meta.get("display_lines") or 0
|
||
)
|
||
if omitted > 0:
|
||
print(f"[body] ... {omitted} more line(s) omitted (use --download for full body)")
|
||
elif body_output_meta.get("mode") == "empty":
|
||
print("[no body]")
|
||
elif response_filter and not body_output_meta.get("preview"):
|
||
print("[body] no lines matched filter")
|
||
else:
|
||
print("[no body]")
|
||
|
||
print(f"\n----- Meta #{run_index + 1} -----")
|
||
if args.show_summary:
|
||
for key in METRIC_KEYS:
|
||
label = key.replace("_", " ").title()
|
||
print(f"{label}: {format_metric_value(key, metrics.get(key))}")
|
||
value = metrics.get(key)
|
||
if aggregate is not None and isinstance(value, (int, float)):
|
||
aggregate[key].append(float(value))
|
||
print(f"Wall Time (client): {total_elapsed:.6f}s")
|
||
if aggregate is not None:
|
||
aggregate["wall_time"].append(total_elapsed)
|
||
else:
|
||
print("Summary disabled (--show-summary=false).")
|
||
print(f"Wall Time (client): {total_elapsed:.6f}s")
|
||
|
||
print(f"Encoding Used: {used_encoding} ({encoding_source})")
|
||
|
||
if args.verbose_output:
|
||
sent_bytes = len(body_bytes) if body_bytes else 0
|
||
print("\nVerbose diagnostics:")
|
||
print(f" Sent body bytes: {sent_bytes}")
|
||
if probe_metrics:
|
||
for key, value in probe_metrics.items():
|
||
print(f" Probe {key}: {value:.6f}s")
|
||
print(f" History length: {len(redirect_history)}")
|
||
print(f" Final URL: {final_url}")
|
||
print(f" Wire request-target: {wire_target.decode('utf-8', errors='replace')}")
|
||
finally:
|
||
response.close()
|
||
for item in redirect_history:
|
||
try:
|
||
item.close()
|
||
except Exception:
|
||
pass
|
||
|
||
except httpx.HTTPError as exc:
|
||
exit_code = 1
|
||
elapsed = time.perf_counter() - start
|
||
print(f"\n===== Response #{run_index + 1} =====")
|
||
print(
|
||
f"Request failed after {elapsed:.6f}s: "
|
||
f"{explain_request_error(exc, encoded_url, forward_proxy)}"
|
||
)
|
||
continue
|
||
|
||
if aggregate and repeat > 1:
|
||
print("\n===== Aggregate Timing =====")
|
||
for key, values in aggregate.items():
|
||
summary = summarize(values)
|
||
if not summary:
|
||
continue
|
||
label = key.replace("_", " ").title()
|
||
min_v, avg_v, max_v = summary
|
||
if key == "speed_download":
|
||
print(f"{label}: min {min_v:.2f} B/s | avg {avg_v:.2f} B/s | max {max_v:.2f} B/s")
|
||
elif key == "size_download":
|
||
print(f"{label}: min {min_v:.0f}B | avg {avg_v:.0f}B | max {max_v:.0f}B")
|
||
elif key in {"http_code", "redirects"}:
|
||
print(f"{label}: min {min_v:.0f} | avg {avg_v:.2f} | max {max_v:.0f}")
|
||
else:
|
||
print(f"{label}: min {min_v:.6f}s | avg {avg_v:.6f}s | max {max_v:.6f}s")
|
||
|
||
finally:
|
||
client.close()
|
||
|
||
sys.exit(exit_code)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print("\nInterrupted.", file=sys.stderr)
|
||
sys.exit(130)
|
||
enabled: true
|
||
short_description: "纯Python HTTP测试框架（httpx，会话复用+编码守护）"
|
||
description: |
|
||
纯 Python 构建的 HTTP 测试框架：基于 httpx 会话，结合 socket 探针补齐 DNS/TCP/TLS 观测，
|
||
提供全流程的请求体编码守护、重复播放与可观测能力，适配盲注延时、复杂 payload 调优等场景。
|
||
|
||
**亮点：**
|
||
- 纯 Python 实现：httpx 会话重用、HTTP/2/代理/certs 直接在脚本内配置，无外部二进制依赖
|
||
- 路径保真（curl --path-as-is）：自定义 Transport + 全请求/重定向链路保留字面量 `.`/`..`；HTTP 正向代理自动 absolute-form；兼容 httpx 0.28 `proxy=`
|
||
- 智能 Body 编码：支持 application/x-www-form-urlencoded 规范化编码、JSON/文本 charset 推断、
|
||
`@file`/`@-` 注入二进制、可视化调试
|
||
- 连接探针：在无代理场景下额外进行 DNS/TCP/TLS 探测，粗粒度复刻 curl -w 指标
|
||
- 可重复观测：repeat/delay + TTFB/total/speed_download 统计，便于盲注/时序测试
|
||
- 扩展开关：http2 / httpx_options 解析 http2、cert、verify、trust_env、max_redirects 等 httpx 选项
|
||
- 响应体瘦身：response_filter 按行/块正则提取，配合 max_lines/max_bytes 限制 stdout，降低 Agent token 消耗
|
||
|
||
**响应过滤最佳实践：**
|
||
- 大页面/HTML：用 `response_filter` 抓 error|exception|password|token|uid 等关键字行
|
||
- 无 filter 时：设 `response_max_lines=80` 或 `response_max_bytes=8192` 防止整页灌入上下文
|
||
- 0 命中：自动预览前 `response_preview_lines` 行，避免误判「空响应」
|
||
- 完整留存：大 body 用 `download` 落盘，stdout 只保留摘要行
|
||
parameters:
|
||
- name: "url"
|
||
type: "string"
|
||
description: "目标URL（默认 path-as-is，保留 ../ 等字面量路径；可开 auto_encode_url 编码特殊字符）"
|
||
required: true
|
||
flag: "--url"
|
||
- name: "method"
|
||
type: "string"
|
||
description: "HTTP方法（GET, POST, PUT, DELETE等）"
|
||
required: false
|
||
default: "GET"
|
||
flag: "--method"
|
||
- name: "data"
|
||
type: "string"
|
||
description: "请求数据/参数（支持JSON、表单、原始payload，前缀@file或@-可加载文件/STDIN）"
|
||
required: false
|
||
flag: "--data"
|
||
- name: "headers"
|
||
type: "object"
|
||
description: "自定义请求头（字典格式，如 {\"X-Custom\": \"value\"}）"
|
||
required: false
|
||
flag: "--headers"
|
||
- name: "cookies"
|
||
type: "string"
|
||
description: "自定义Cookie（格式：name1=value1; name2=value2）"
|
||
required: false
|
||
flag: "--cookies"
|
||
- name: "user_agent"
|
||
type: "string"
|
||
description: "自定义User-Agent"
|
||
required: false
|
||
flag: "--user-agent"
|
||
- name: "proxy"
|
||
type: "string"
|
||
description: "代理（http(s)://或socks5://地址，直接传递给httpx Client）"
|
||
required: false
|
||
flag: "--proxy"
|
||
- name: "timeout"
|
||
type: "string"
|
||
description: "最大超时时间（秒，支持小数）"
|
||
required: false
|
||
flag: "--timeout"
|
||
- name: "repeat"
|
||
type: "int"
|
||
description: "重复请求次数，用于盲注/稳定性观测（>=1）"
|
||
required: false
|
||
default: 1
|
||
flag: "--repeat"
|
||
- name: "delay"
|
||
type: "string"
|
||
description: "重复请求之间的延迟（秒，可为小数）"
|
||
required: false
|
||
default: "0"
|
||
flag: "--delay"
|
||
- name: "include_headers"
|
||
type: "bool"
|
||
description: "输出响应头（默认开启，可用 --no-include-headers 关闭）"
|
||
required: false
|
||
default: true
|
||
flag: "--include-headers"
|
||
- name: "auto_encode_url"
|
||
type: "bool"
|
||
description: "自动URL编码（默认开启，可通过 --no-auto-encode-url 关闭）"
|
||
required: false
|
||
default: true
|
||
flag: "--auto-encode-url"
|
||
- name: "follow_redirects"
|
||
type: "bool"
|
||
description: "跟随重定向（httpx follow_redirects）"
|
||
required: false
|
||
default: false
|
||
flag: "--follow-redirects"
|
||
- name: "allow_insecure"
|
||
type: "bool"
|
||
description: "忽略TLS证书错误（verify=False）"
|
||
required: false
|
||
default: false
|
||
flag: "--allow-insecure"
|
||
- name: "verbose_output"
|
||
type: "bool"
|
||
description: "输出额外调试信息（请求体字节数、探针延时、历史长度等）"
|
||
required: false
|
||
default: false
|
||
flag: "--verbose-output"
|
||
- name: "show_command"
|
||
type: "bool"
|
||
description: "打印请求概览（方法、URL、所有头、Body概况）"
|
||
required: false
|
||
default: true
|
||
flag: "--show-command"
|
||
- name: "show_summary"
|
||
type: "bool"
|
||
description: "打印指标摘要（默认开启，含DNS/TCP/TLS/TTFB/Total）"
|
||
required: false
|
||
default: true
|
||
flag: "--show-summary"
|
||
- name: "debug"
|
||
type: "bool"
|
||
description: "调试模式：显示Body编码处理细节（模式/charset/长度）"
|
||
required: false
|
||
default: false
|
||
flag: "--debug"
|
||
- name: "response_encoding"
|
||
type: "string"
|
||
description: "强制响应解码使用的编码（如GBK），覆盖自动探测"
|
||
required: false
|
||
flag: "--response-encoding"
|
||
- name: "response_filter"
|
||
type: "string"
|
||
description: |
|
||
响应体正则过滤（仅影响 stdout，不影响 --download 与指标）。
|
||
默认 line 模式按行匹配；示例：'(error|exception|SQL|password|token|uid)'。
|
||
与 response_max_lines/response_max_bytes 配合可显著减少 token 消耗。
|
||
required: false
|
||
flag: "--response-filter"
|
||
- name: "response_filter_mode"
|
||
type: "string"
|
||
description: "过滤模式：line（按行，默认）、multiline（跨行块）、full（整段 DOTALL 匹配）"
|
||
required: false
|
||
default: "line"
|
||
flag: "--response-filter-mode"
|
||
- name: "response_filter_invert"
|
||
type: "bool"
|
||
description: "反向过滤：输出不匹配 regex 的行（用于剔除 HTML 噪音）"
|
||
required: false
|
||
default: false
|
||
flag: "--response-filter-invert"
|
||
- name: "response_filter_ignore_case"
|
||
type: "bool"
|
||
description: "正则忽略大小写"
|
||
required: false
|
||
default: false
|
||
flag: "--response-filter-ignore-case"
|
||
- name: "response_max_lines"
|
||
type: "int"
|
||
description: "stdout 最多输出行数（0=不限制）；有 filter 时限制命中行数，无 filter 时截断全文"
|
||
required: false
|
||
default: 0
|
||
flag: "--response-max-lines"
|
||
- name: "response_max_bytes"
|
||
type: "int"
|
||
description: "stdout 响应体 UTF-8 字节上限（0=不限制），超出部分截断"
|
||
required: false
|
||
default: 0
|
||
flag: "--response-max-bytes"
|
||
- name: "response_preview_lines"
|
||
type: "int"
|
||
description: "filter 零命中时预览的前 N 行（默认 5，0=不预览）"
|
||
required: false
|
||
default: 5
|
||
flag: "--response-preview-lines"
|
||
- name: "response_context_lines"
|
||
type: "int"
|
||
description: "line 模式下命中行上下各保留 N 行上下文（类似 grep -C）"
|
||
required: false
|
||
default: 0
|
||
flag: "--response-context-lines"
|
||
- name: "action"
|
||
type: "string"
|
||
description: "保留字段：标识调用意图（request, spider等），脚本内部不使用"
|
||
required: false
|
||
default: "request"
|
||
flag: "--action"
|
||
- name: "download"
|
||
type: "string"
|
||
description: |
|
||
将响应体保存到本地文件，写入原始字节数据：
|
||
- "output.bin"：单次请求直接保存到指定文件
|
||
- "outputs/run-{i}.bin"：repeat>1 时按序号区分文件（{i} 将被替换为 1,2,...）
|
||
required: false
|
||
flag: "--download"
|
||
- name: "http2"
|
||
type: "bool"
|
||
description: "启用 HTTP/2（需安装 h2：pip install 'httpx[http2]'）。也可写在 httpx_options：http2=true"
|
||
required: false
|
||
default: false
|
||
flag: "--http2"
|
||
- name: "httpx_options"
|
||
type: "string"
|
||
description: |
|
||
额外的 httpx Client 选项（整段传给 --additional-args）。支持 "key=value" 或单词形式（等价于 key=true）：
|
||
- "http2=true"
|
||
- "cert=/path/to/client.pem"
|
||
- "verify=/path/to/ca.pem" 或 "verify=false"
|
||
- "trust_env=false"、"max_redirects=5" 等
|
||
注意：不要用框架通用字段 additional_args（会被拆成裸 argv）。请用本字段或 http2。
|
||
required: false
|
||
flag: "--additional-args"
|