From b39fefd2d0d95f99de5fe51d6a3743fdeccd8655 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=85=AC=E6=98=8E?= <83812544+Ed1s0nZ@users.noreply.github.com> Date: Thu, 16 Jul 2026 11:03:49 +0800 Subject: [PATCH] Add files via upload --- tools/http-framework-test.yaml | 413 ++++++++++++++++++++++++--------- 1 file changed, 306 insertions(+), 107 deletions(-) diff --git a/tools/http-framework-test.yaml b/tools/http-framework-test.yaml index 8a29eb38..b1753e56 100644 --- a/tools/http-framework-test.yaml +++ b/tools/http-framework-test.yaml @@ -127,77 +127,263 @@ args: return urllib.parse.urlunsplit((parts.scheme, parts.netloc, path, query, fragment)) - def path_has_dot_segments(path: str) -> bool: - """True when path contains '.' / '..' segments that httpx would RFC-normalize away.""" - if not path: - return False - for segment in path.split("/"): - if segment in {".", ".."}: - return True - return False - - - def split_origin_and_target(url: str) -> Tuple[str, bytes]: - """Split URL into origin (scheme://netloc) and raw request-target bytes.""" + def origin_url(url: str) -> str: + """scheme://netloc/ — httpx build base; real request-target is set separately.""" parts = urllib.parse.urlsplit(url) - origin = urllib.parse.urlunsplit((parts.scheme, parts.netloc, "", "", "")) - if not origin.endswith("/"): - # httpx needs a valid URL string; path is supplied separately via raw target - pass - target = parts.path or "/" - if parts.query: - target = f"{target}?{parts.query}" - return origin.rstrip("/") + "/", target.encode("utf-8", errors="surrogateescape") + return urllib.parse.urlunsplit((parts.scheme, parts.netloc, "/", "", "")) - class RawPathURL: - """Wrap httpx.URL but keep literal request-target (preserves ../ for LFI tests).""" - - def __init__(self, base: "httpx.URL", raw_path: bytes): - self._base = base - self.raw_path = raw_path - - def __getattr__(self, name): - return getattr(self._base, name) - - def __str__(self): - host = self._base.host - scheme = self._base.scheme - port = self._base.port - default_port = 443 if scheme == "https" else 80 - netloc = host if port in (None, default_port) else f"{host}:{port}" - return f"{scheme}://{netloc}{self.raw_path.decode('utf-8', errors='replace')}" - - - def prepare_request_url(url: str): + def wire_request_target(url: str, absolute_form: bool = False) -> bytes: """ - Return (display_url, request_url, wire_target). - When path has '.'/'..' segments, request_url is a RawPathURL so httpx/httpcore - send the literal path instead of collapsing to e.g. /etc/passwd. + 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: + """Effective forward-proxy URL for absolute-form decisions (explicit > env).""" + if explicit_proxy: + return explicit_proxy + if not trust_env: + return "" + scheme = (urllib.parse.urlsplit(url).scheme or "").lower() + if scheme == "https": + return ( + os.environ.get("HTTPS_PROXY") + or os.environ.get("https_proxy") + or os.environ.get("ALL_PROXY") + or os.environ.get("all_proxy") + or "" + ) + return ( + os.environ.get("HTTP_PROXY") + or os.environ.get("http_proxy") + or os.environ.get("ALL_PROXY") + or os.environ.get("all_proxy") + or "" + ) + + + def needs_http_absolute_form(proxy: str, url: str) -> bool: + """HTTP(S) forward proxy expects absolute-form target for plaintext HTTP.""" + if not proxy: + return False try: - parts = urllib.parse.urlsplit(url) + proxy_scheme = urllib.parse.urlsplit(proxy).scheme.lower() + url_scheme = urllib.parse.urlsplit(url).scheme.lower() except ValueError: - return url, url, None - if not path_has_dot_segments(parts.path or ""): - return url, url, None - origin, raw_target = split_origin_and_target(url) - base = httpx.URL(origin) - wrapped = RawPathURL(base, raw_target) - return str(wrapped), wrapped, raw_target + return False + return url_scheme == "http" and proxy_scheme in {"http", "https"} + + + 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) -> str: text = str(exc) lowered = text.lower() - hints = [] if "unexpected_eof" in lowered or "eof occurred in violation of protocol" in lowered: - hints.append( - "服务器在 TLS 层直接断开（常见于 WAF/反代拒绝异常路径、URI 过长或敏感路径探测）。" + return ( + text + + "\nHint: 服务器在 TLS 层直接断开（常见于 WAF/反代拒绝异常路径、URI 过长或敏感路径探测）。" "可尝试：缩短 ../ 层数、改用 %2e%2e 编码、加 --allow-insecure、或经代理观察原始响应。" ) - if hints: - return text + "\nHint: " + " ".join(hints) return text @@ -521,7 +707,7 @@ args: 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: preserved '.'/'..' path segments (httpx would otherwise normalize them away).") + 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}") @@ -826,9 +1012,8 @@ args: delay_between = 0.0 encoded_url = smart_encode_url(args.url) if args.auto_encode_url else args.url - prepared_url, request_url, wire_target = prepare_request_url(encoded_url) method = (args.method or "GET").upper() - + trust_env = True # 处理 headers：支持字典（JSON字符串）和字符串格式 # 框架会将 object 类型序列化为 JSON 字符串传递 headers_list = [] @@ -885,29 +1070,49 @@ args: timeout_value = None timeout = httpx.Timeout(timeout_value or 60.0) - client_kwargs = { - "timeout": timeout, - "verify": not args.allow_insecure, - "follow_redirects": args.follow_redirects, - "cookies": cookie_jar, - } - if args.proxy: - client_kwargs["proxies"] = args.proxy - if "http2" in additional_options: - client_kwargs["http2"] = str_to_bool(additional_options["http2"]) - if "cert" in additional_options: - client_kwargs["cert"] = additional_options["cert"] - if "verify" in additional_options: - value = additional_options["verify"] - lowered = value.strip().lower() - client_kwargs["verify"] = str_to_bool(value) if lowered in {"true", "false", "1", "0", "yes", "no", "on", "off"} else value + max_redirects = 20 if "max_redirects" in additional_options: try: - client_kwargs["max_redirects"] = int(additional_options["max_redirects"]) + max_redirects = int(additional_options["max_redirects"]) except ValueError: pass if "trust_env" in additional_options: - client_kwargs["trust_env"] = str_to_bool(additional_options["trust_env"]) + 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 + 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 + if explicit_proxy: + transport_kwargs["proxy"] = explicit_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( @@ -915,7 +1120,7 @@ args: prepared_url, headers, body_meta, - wire_target=wire_target, + wire_target=initial_wire_target, input_url=args.url if args.url != prepared_url else None, ) @@ -925,9 +1130,8 @@ args: aggregate["wall_time"] = [] exit_code = 0 - verify_option = client_kwargs.get("verify", True) verify_tls = verify_option if isinstance(verify_option, bool) else True - skip_probe = bool(args.proxy) + skip_probe = bool(explicit_proxy) client = httpx.Client(**client_kwargs) try: @@ -942,33 +1146,23 @@ args: start = time.perf_counter() first_byte_time = None body_buffer = bytearray() + wire_target = initial_wire_target try: - if wire_target is not None: - origin, _ = split_origin_and_target(encoded_url) - build_kwargs = { - "method": method, - "url": origin, - "headers": headers, - } - else: - build_kwargs = { - "method": method, - "url": request_url, - "headers": headers, - } - if body_bytes is not None: - build_kwargs["content"] = body_bytes - - request = client.build_request(**build_kwargs) - if wire_target is not None: - request.url = RawPathURL(request.url, wire_target) - - response = client.send(request, stream=True) + 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(response.history) + metrics["redirects"] = len(redirect_history) http_version = response.http_version or "HTTP/1.1" for chunk in response.iter_bytes(): @@ -1066,11 +1260,16 @@ args: if probe_metrics: for key, value in probe_metrics.items(): print(f" Probe {key}: {value:.6f}s") - print(f" History length: {len(response.history)}") - if wire_target is not None: - print(f" Wire request-target: {wire_target.decode('utf-8', errors='replace')}") + 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 @@ -1116,7 +1315,7 @@ description: | **亮点：** - 纯 Python 实现：httpx 会话重用、HTTP/2/代理/certs 直接在脚本内配置，无外部二进制依赖 - - 路径穿越保真：自动保留 URL 中的 `.`/`..` 段（避免 httpx RFC 规范化把 LFI payload 折叠成 `/etc/passwd`） + - 路径保真（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 指标 @@ -1132,7 +1331,7 @@ description: | parameters: - name: "url" type: "string" - description: "目标URL（可自动编码路径/参数，避免特殊字符导致的请求失败如：https://www.target.com/s?wd=test）" + description: "目标URL（默认 path-as-is，保留 ../ 等字面量路径；可开 auto_encode_url 编码特殊字符）" required: true flag: "--url" - name: "method"