Add files via upload

2026-07-19 18:38:00 +02:00 · 2026-07-16 11:03:49 +08:00
parent d244943019
commit b39fefd2d0
1 changed files with 306 additions and 107 deletions
@@ -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"