文章总结： 本文分析了httpx库中参数传递机制的安全风险，重点探讨了data与json参数共存时可能存在的混淆漏洞。作者通过跟踪源码发现参数边界处理不当可能导致伪造数据块的风险，并建议开发者注意参数规范对代码安全性的影响。文章提供了具体的代码审计方法和安全开发建议。 综合评分： 72 文章分类： 代码审计,漏洞分析,WEB安全,安全开发,其他
偶然探究httpx的链子，关于伪造data块的可能性
原创
YMsora YMsora
YMs0ra的安全漫路
2026年3月26日 21:15 浙江
在小说阅读器读本章
去阅读
因为一些原因重新去看了一个题的源码，算是时隔好久了
核心在于httpx。正常把库扒下来。接下来跟一根乱七八糟的api和参数
很多时候是因为参数更迭的混乱导致审计的困难，
但是好像随着时间的流逝，这点也慢慢学会了。
有时候就是什么问题就一直想着看源码，忽视了很多搜集信息的渠道
得稍微摆正自己的思维了
简单跟一下源码，来吧
我们先看看最前面的api逻辑。
def request( method: str, url: URL | str, *, params: QueryParamTypes | None = None, content: RequestContent | None = None, data: RequestData | None = None, files: RequestFiles | None = None, json: typing.Any | None = None, headers: HeaderTypes | None = None, cookies: CookieTypes | None = None, auth: AuthTypes | None = None, proxy: ProxyTypes | None = None, timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG, follow_redirects: bool = False, verify: ssl.SSLContext | str | bool = True, trust_env: bool = True,) -> Response: """ Sends an HTTP request.
**Parameters:**
* **method** - HTTP method for the new `Request` object: `GET`, `OPTIONS`, `HEAD`, `POST`, `PUT`, `PATCH`, or `DELETE`. * **url** - URL for the new `Request` object. * **params** - *(optional)* Query parameters to include in the URL, as a string, dictionary, or sequence of two-tuples. * **content** - *(optional)* Binary content to include in the body of the request, as bytes or a byte iterator. * **data** - *(optional)* Form data to include in the body of the request, as a dictionary. * **files** - *(optional)* A dictionary of upload files to include in the body of the request. * **json** - *(optional)* A JSON serializable object to include in the body of the request. * **headers** - *(optional)* Dictionary of HTTP headers to include in the request. * **cookies** - *(optional)* Dictionary of Cookie items to include in the request. * **auth** - *(optional)* An authentication class to use when sending the request. * **proxy** - *(optional)* A proxy URL where all the traffic should be routed. * **timeout** - *(optional)* The timeout configuration to use when sending the request. * **follow_redirects** - *(optional)* Enables or disables HTTP redirects. * **verify** - *(optional)* Either `True` to use an SSL context with the default CA bundle, `False` to disable verification, or an instance of `ssl.SSLContext` to use a custom context. * **trust_env** - *(optional)* Enables or disables usage of environment variables for configuration.
**Returns:** `Response`
Usage:
``` >>> import httpx >>> response = httpx.request('GET', 'https://httpbin.org/get') >>> response <Response [200 OK]> ``` """ with Client( cookies=cookies, proxy=proxy, verify=verify, timeout=timeout, trust_env=trust_env, ) as client: return client.request( method=method, url=url, content=content, data=data, files=files, json=json, params=params, headers=headers, auth=auth, follow_redirects=follow_redirects, )
虽然初见参数很多，而且前面引入了其他的模块，先且不论，这里也就是正常规定了一个response
跟着看看client的request方法，接收参数形式倒是默认的
因为接收的参数很多，所以有时候会想多参数引用的边界，参数的规范真的对于代码可读性有非常大的影响
继续跟进
def request( self, method: str, url: URL | str, *, content: RequestContent | None = None, data: RequestData | None = None, files: RequestFiles | None = None, json: typing.Any | None = None, params: QueryParamTypes | None = None, headers: HeaderTypes | None = None, cookies: CookieTypes | None = None, auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT, follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT, timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT, extensions: RequestExtensions | None = None,) -> Response: """ Build and send a request.
Equivalent to:
```python request = client.build_request(...) response = client.send(request, ...) ```
See `Client.build_request()`, `Client.send()` and [Merging of configuration][0] for how the various parameters are merged with client-level configuration.
[0]: /advanced/clients/#merging-of-configuration """ if cookies is not None: message = ( "Setting per-request cookies=<...> is being deprecated, because " "the expected behaviour on cookie persistence is ambiguous. Set " "cookies directly on the client instance instead." ) warnings.warn(message, DeprecationWarning, stacklevel=2)
request = self.build_request( method=method, url=url, content=content, data=data, files=files, json=json, params=params, headers=headers, cookies=cookies, timeout=timeout, extensions=extensions, ) return self.send(request, auth=auth, follow_redirects=follow_redirects)
没啥都，直接看build_request
def build_request( self, method: str, url: URL | str, *, content: RequestContent | None = None, data: RequestData | None = None, files: RequestFiles | None = None, json: typing.Any | None = None, params: QueryParamTypes | None = None, headers: HeaderTypes | None = None, cookies: CookieTypes | None = None, timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT, extensions: RequestExtensions | None = None,) -> Request: """ Build and return a request instance.
* The `params`, `headers` and `cookies` arguments are merged with any values set on the client. * The `url` argument is merged with any `base_url` set on the client.
See also: [Request instances][0]
[0]: /advanced/clients/#request-instances """ url = self._merge_url(url) headers = self._merge_headers(headers) cookies = self._merge_cookies(cookies) params = self._merge_queryparams(params) extensions = {} if extensions is None else extensions if "timeout" not in extensions: timeout = ( self.timeout if isinstance(timeout, UseClientDefault) else Timeout(timeout) ) extensions = dict(**extensions, timeout=timeout.as_dict()) return Request( method, url, content=content, data=data, files=files, json=json, params=params, headers=headers, cookies=cookies, extensions=extensions, )
我们可以看到_merge_queryparams和_merge_cookies，_merge_headers以及_merge_url 可以看到最后的extension，去验证了看看是否有扩展有危险拼入,果然不出所料，没有，继续追踪
class Request: def __init__( self, method: str, url: URL | str, *, params: QueryParamTypes | None = None, headers: HeaderTypes | None = None, cookies: CookieTypes | None = None, content: RequestContent | None = None, data: RequestData | None = None, files: RequestFiles | None = None, json: typing.Any | None = None, stream: SyncByteStream | AsyncByteStream | None = None, extensions: RequestExtensions | None = None, ) -> None: self.method = method.upper() self.url = URL(url) if params is None else URL(url, params=params) self.headers = Headers(headers) self.extensions = {} if extensions is None else dict(extensions)
if cookies: Cookies(cookies).set_cookie_header(self)
if stream is None: content_type: str | None = self.headers.get("content-type") headers, stream = encode_request( content=content, data=data, files=files, json=json, boundary=get_multipart_boundary_from_content_type( content_type=content_type.encode(self.headers.encoding) if content_type else None ), ) self._prepare(headers) self.stream = stream # Load the request body, except for streaming content. if isinstance(stream, ByteStream): self.read() else: # There's an important distinction between `Request(content=...)`, # and `Request(stream=...)`. # # Using `content=...` implies automatically populated `Host` and content # headers, of either `Content-Length: ...` or `Transfer-Encoding: chunked`. # # Using `stream=...` will not automatically include *any* # auto-populated headers. # # As an end-user you don't really need `stream=...`. It's only # useful when: # # * Preserving the request stream when copying requests, eg for redirects. # * Creating request instances on the *server-side* of the transport API. self.stream = stream
boundary=get_multipart_boundary_from_content_type( content_type=content_type.encode(self.headers.encoding) if content_type else None
这里的boundary分块可以被ct头指定，也就是说我们可以通过掌控CT头去增加form-data块
然后我们看回题目源码
action = request.files.get("action") act = json.loads(action.stream.read().decode()) if act["type"] == "echo": return content, 200 elif act["type"] == "debug": return content.format(app), 200 else: return 'unkown action', 400
可以看到读取了form-data的action，注意是file.get
act也就是action的数据流，如果里面的type的key是debug
就直接渲染app的上下文。我们也就是直接把context上下文渲染走format路线
然后后置的路线也就是我们控制了通过拿到了global的apikey，然后
就能访问admin去执行jinjia模板渲染
于是利用传输大于500kb的文件去达到写入临时文件，然后就可以加载jinjia模板了
当然我认为重要的点在前置。
OK了，exp可以在先知社区参考，这里就不放出了
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：YMs0ra的安全漫路 YMsora YMsora《偶然探究httpx的链子，关于伪造data块的可能性》
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论