1. 项目缘起为什么要在云函数里抓包最近在折腾一些小程序的数据分析发现很多核心逻辑和数据交互都跑在云函数上。直接在小程序端抓包看到的往往是加密后的数据流或者干脆就是一堆看不懂的二进制。这时候一个很自然的想法就冒出来了能不能直接在云函数的执行环境里把发出去的请求和收到的响应给“看”清楚这就是“云函数抓包”的核心诉求。听起来有点像在服务器上装个抓包软件但云函数的环境是临时的、无状态的而且权限限制通常很严格不可能让你去安装像 Wireshark 或 tcpdump 这样的系统级工具。所以我们必须换一种思路——在应用层用纯 Python 来实现网络请求的拦截和记录。这不仅仅是抓包更是一种针对特定应用小程序云函数的逆向分析手段目的是理解其数据流转的完整链条为后续的数据分析、接口调试甚至安全审计打下基础。这个方法特别适合开发者、安全研究员或者数据分析师。比如你想分析某个小程序商城的价格波动规律但数据接口被混淆了或者你需要调试自己编写的云函数想确认发出的 HTTP 请求格式是否正确再或者你在进行安全评估需要梳理云函数对外部服务的依赖情况。在这些场景下一个轻量级、可嵌入的 Python 抓包工具就显得非常实用。2. 核心思路在应用层实现请求拦截既然不能动系统那我们就从应用层入手。小程序云函数本质上是一段运行在云端的 Node.js 或 Python 代码它对外发起网络请求最常用的库无非是requestsPython或axios/原生http模块Node.js。我们的目标就是“劫持”这些库的请求发出和响应接收过程。2.1 方案选型猴子补丁 vs. 代理中间件要实现这个目标主要有两种技术路径猴子补丁在运行时动态替换掉requests库或http.client等底层模块的关键函数比如将requests.request或requests.Session.send替换成我们自己的函数。在这个自定义函数里我们先记录请求的详细信息然后调用原始函数完成实际请求再记录响应信息最后将响应返回。这种方式侵入性强但实现直接能捕获到最原始的请求。代理中间件不直接修改库函数而是创建一个代理对象或适配器。例如我们可以封装一个自己的Session类它内部使用真正的requests.Session但在request方法前后加入日志逻辑。或者利用requests库的hooks机制response钩子来记录响应。这种方式更优雅耦合度低但可能无法捕获到某些非常规方式发出的请求。对于云函数这种我们通常能控制代码的环境猴子补丁的彻底性更有优势。我们选择从requests.Session.send这个最底层、最核心的方法入手进行替换确保所有通过requests库发起的请求无论调用层级如何都能被我们捕获。2.2 技术要点确保捕获的完整性与性能在云函数中实现抓包有几个关键点需要考虑请求/响应的全信息记录不仅仅是 URL 和状态码。我们需要记录请求方法、完整的请求头、请求体可能是文本、JSON 或二进制文件以及响应的状态码、响应头和响应体。对于 JSON 数据最好能格式化输出以便阅读。对性能的影响要最小化云函数是按执行时间和内存消耗计费的。我们的抓包逻辑不能显著增加函数的运行开销。这意味着要避免深度拷贝大体积的请求/响应体对于二进制数据如图片可能只记录其大小和哈希值而非全部内容。输出目标的选择在云函数里我们不能把日志写到本地文件。最直接的输出方式是打印到标准输出print这些日志会被收集到云服务商提供的日志服务中。我们也可以将抓包数据结构化后发送到指定的日志收集接口或对象存储中但这会增加额外的网络请求。异步请求的支持越来越多的云函数开始支持异步编程。如果云函数中使用了aiohttp或httpx这样的异步 HTTP 客户端我们的抓包方案也需要有相应的异步版本。基于以上考量我们将设计一个核心的RequestSniffer类它使用猴子补丁技术并提供一个上下文管理器确保抓包行为在可控的范围内开启和关闭。3. 核心代码实现与逐行解析接下来我们动手实现这个纯 Python 的云函数抓包工具。我们将代码命名为cloud_function_sniffer.py。3.1 基础架构与导入import json import time import hashlib import requests from typing import Any, Dict, Optional, Union, Callable from urllib.parse import urlparse import functoolsjson: 用于格式化 JSON 请求/响应体。time: 记录请求耗时。hashlib: 用于计算大响应体如文件的哈希值避免记录全部内容。requests: 这是我们即将要“补丁”的目标库。typing: 提供类型注解让代码更清晰。urllib.parse: 用于解析 URL提取主机名等有用信息。functools: 使用wraps装饰器来保留被替换函数的元数据。3.2 核心类RequestSnifferclass RequestSniffer: HTTP请求嗅探器用于拦截和记录requests库发起的网络请求。 def __init__(self, max_body_length: int 1024 * 10, # 10KB skip_binary: bool True, output_handler: Optional[Callable[[Dict], None]] None): 初始化嗅探器。 Args: max_body_length: 记录请求/响应体的最大长度字节。超出部分将被截断。 skip_binary: 是否跳过记录二进制内容如图片、文件流。若为True则只记录大小和MD5。 output_handler: 自定义输出处理器。默认为None使用print输出到控制台。 self.max_body_length max_body_length self.skip_binary skip_binary self.output_handler output_handler or self._default_output self._original_send None # 保存原始的requests.Session.send方法 self._is_active Falsemax_body_length: 这是一个重要的性能和安全参数。云函数的响应体可能非常大比如下载一个文件。全部记录会消耗大量内存和日志额度。我们设置一个上限默认10KB超出的部分只记录摘要信息。skip_binary: 二进制数据如图片、PDF在日志里是一堆乱码没有可读性。开启此选项可以显著减少日志体积。output_handler: 提供灵活性。默认是打印到控制台但用户可以传入一个函数将抓包数据写入数据库、发送到日志平台等。_original_send: 保存requests.Session.send的原始引用这是猴子补丁的关键。_is_active: 状态标志防止重复激活。3.3 猴子补丁的核心自定义send方法这是整个工具最核心的部分我们创建一个新的_patched_send方法来替换原来的send。def _patched_send(self, request, **kwargs): 被替换后的Session.send方法包裹了日志记录逻辑。 start_time time.time() # 1. 在发送前记录请求信息 req_info self._capture_request(request, kwargs) # 2. 调用原始的send方法执行真正的网络请求 try: response self._original_send(request, **kwargs) elapsed time.time() - start_time # 3. 请求成功后记录响应信息 resp_info self._capture_response(response, elapsed) log_entry {**req_info, **resp_info, ‘status‘: ‘success‘} except Exception as e: elapsed time.time() - start_time # 4. 请求失败后记录异常信息 resp_info { ‘error‘: str(e), ‘elapsed‘: f‘{elapsed:.3f}s‘, ‘status_code‘: None, } log_entry {**req_info, **resp_info, ‘status‘: ‘error‘} raise # 将异常原样抛出不影响云函数原有逻辑 finally: # 5. 无论成功失败都输出日志条目 self.output_handler(log_entry) return response关键点解析计时开始在调用原始send前记录时间用于计算请求耗时。捕获请求调用_capture_request方法提取请求的详细信息。执行与捕获响应调用原始send这是实际网络发生的地方。成功后调用_capture_response记录响应。异常处理如果请求失败超时、网络错误等我们捕获异常记录错误信息但必须重新抛出异常raise。这是至关重要的否则会改变云函数的错误处理逻辑导致问题被掩盖。日志输出在finally块中确保日志一定会被输出。然后返回响应或让异常继续传播。3.4 请求与响应的信息捕获_capture_request和_capture_response是两个重要的工具方法负责从原始对象中提取我们需要的信息。def _capture_request(self, request, kwargs) - Dict[str, Any]: 从PreparedRequest对象和kwargs中提取请求信息。 # 解析URL parsed_url urlparse(request.url) # 获取请求体 body request.body body_preview ‘[Binary or Empty]‘ body_size 0 body_md5 None if body: body_size len(body) if isinstance(body, bytes) else len(body.encode(‘utf-8‘)) if self.skip_binary and isinstance(body, bytes): # 如果是二进制数据且设置了跳过则计算MD5 body_md5 hashlib.md5(body).hexdigest()[:8] body_preview f‘[Binary Data, size: {body_size}, md5: {body_md5}]‘ else: # 尝试解码为文本或截断预览 try: if isinstance(body, bytes): body_str body.decode(‘utf-8‘) else: body_str str(body) # 截断过长的内容 if len(body_str) self.max_body_length: body_preview body_str[:self.max_body_length] f‘... (truncated, total: {body_size})‘ else: body_preview body_str except UnicodeDecodeError: # 解码失败视为二进制 body_md5 hashlib.md5(body).hexdigest()[:8] body_preview f‘[Non-UTF8 Data, size: {body_size}, md5: {body_md5}]‘ req_info { ‘method‘: request.method, ‘url‘: request.url, ‘host‘: parsed_url.netloc, ‘path‘: parsed_url.path, ‘request_headers‘: dict(request.headers), ‘request_body_preview‘: body_preview, ‘request_body_size‘: body_size, ‘timestamp‘: time.strftime(‘%Y-%m-%d %H:%M:%S‘), } # 记录一些可能有用的kwargs如超时设置 if ‘timeout‘ in kwargs: req_info[‘timeout‘] kwargs[‘timeout‘] return req_info注意request.body可能是bytes、str或None。我们优先尝试将其作为文本处理以便在日志中可读。但对于图片上传等场景body是二进制流直接解码会出错。因此我们通过skip_binary开关和异常捕获来优雅处理。def _capture_response(self, response, elapsed: float) - Dict[str, Any]: 从Response对象中提取响应信息。 # 获取响应体内容注意这会消耗掉响应体需要做处理 resp_content response.content body_preview ‘[Empty]‘ body_size len(resp_content) body_md5 None if resp_content: if self.skip_binary and self._is_likely_binary(response.headers): body_md5 hashlib.md5(resp_content).hexdigest()[:8] body_preview f‘[Binary Response, size: {body_size}, md5: {body_md5}]‘ else: # 尝试解码为文本 try: content_str resp_content.decode(‘utf-8‘) # 如果是JSON可以尝试美化 if ‘application/json‘ in response.headers.get(‘Content-Type‘, ‘‘): try: parsed_json json.loads(content_str) content_str json.dumps(parsed_json, indent2, ensure_asciiFalse) except json.JSONDecodeError: pass # 保持原样 # 截断 if len(content_str) self.max_body_length: body_preview content_str[:self.max_body_length] f‘... (truncated, total: {body_size})‘ else: body_preview content_str except UnicodeDecodeError: body_md5 hashlib.md5(resp_content).hexdigest()[:8] body_preview f‘[Non-UTF8 Response, size: {body_size}, md5: {body_md5}]‘ resp_info { ‘status_code‘: response.status_code, ‘response_headers‘: dict(response.headers), ‘response_body_preview‘: body_preview, ‘response_body_size‘: body_size, ‘elapsed‘: f‘{elapsed:.3f}s‘, } return resp_info staticmethod def _is_likely_binary(headers: Dict) - bool: 根据Content-Type判断内容是否可能是二进制的。 content_type headers.get(‘Content-Type‘, ‘‘).lower() binary_types [‘image/‘, ‘application/octet-stream‘, ‘video/‘, ‘audio/‘, ‘application/pdf‘, ‘application/zip‘] return any(ct in content_type for ct in binary_types)关键点解析响应体消耗问题直接读取response.content会将响应体全部加载到内存。对于大文件这可能是个问题。但在抓包调试场景下我们通常关注的是接口返回的元数据和少量文本数据。skip_binary和max_body_length在这里起到了保护作用。切记在生产环境云函数中如果预期会下载大文件请务必将skip_binary设为True并谨慎设置max_body_length。智能内容处理我们通过Content-Type头来辅助判断是否为二进制数据。对于application/json我们尝试解析并美化让日志更易读。性能考量计算 MD5 和字符串解码都有开销。这就是为什么我们提供开关允许用户为性能而牺牲部分信息细节。3.5 激活、关闭与上下文管理器为了方便使用我们提供start()、stop()方法和一个上下文管理器。def start(self): 激活请求嗅探。 if self._is_active: return # 保存原始方法 self._original_send requests.Session.send # 创建绑定到当前实例的新函数 patched functools.partial(self._patched_send, self) # 替换类方法 requests.Session.send patched self._is_active True def stop(self): 停止请求嗅探恢复原状。 if self._is_active and self._original_send: requests.Session.send self._original_send self._original_send None self._is_active False def __enter__(self): 支持with语句。 self.start() return self def __exit__(self, exc_type, exc_val, exc_tb): 退出with语句块时自动停止。 self.stop() def _default_output(self, entry: Dict): 默认输出处理器格式化打印到控制台。 print(‘\n‘ ‘‘*60) print(f“[HTTP Request] {entry.get(‘method‘)} {entry.get(‘url‘)}“) print(f“Status: {entry.get(‘status‘).upper()} ({entry.get(‘status_code‘, ‘N/A‘)})“) print(f“Time: {entry.get(‘timestamp‘)} | Elapsed: {entry.get(‘elapsed‘, ‘N/A‘)}“) print(‘-‘*40) print(‘Request Headers:‘) for k, v in entry.get(‘request_headers‘, {}).items(): print(f‘ {k}: {v}‘) if entry.get(‘request_body_preview‘) and entry[‘request_body_preview‘] ! ‘[Binary or Empty]‘: print(‘Request Body Preview:‘) print(f‘ {entry[“request_body_preview“]}‘) print(‘-‘*40) print(‘Response Headers:‘) for k, v in entry.get(‘response_headers‘, {}).items(): print(f‘ {k}: {v}‘) if entry.get(‘response_body_preview‘) and entry[‘response_body_preview‘] ! ‘[Empty]‘: print(‘Response Body Preview:‘) # 对可能的多行内容进行缩进 preview entry[‘response_body_preview‘] lines preview.split(‘\n‘) for line in lines[:10]: # 最多打印前10行 print(f‘ {line}‘) if len(lines) 10: print(f‘ ... (and {len(lines)-10} more lines)‘) print(‘‘*60 ‘\n‘)start和stop这是标准的猴子补丁流程。functools.partial用于将实例方法_patched_send绑定到当前实例 (self)然后替换类的send方法。上下文管理器使用with RequestSniffer() as sniffer:的语法可以确保在任何情况下即使发生异常stop()都会被调用避免污染全局环境。这是在云函数中使用的最佳实践。_default_output将抓包数据格式化为易读的文本输出到控制台。在实际使用中你可以重写这个方法或者通过output_handler参数传入自定义函数将数据发送到云日志服务如腾讯云的CLS、阿里云的SLS或你自己的日志服务器。4. 在云函数中的实战应用现在我们来看如何将这个工具集成到一个真实的小程序云函数中。假设我们有一个云函数它的作用是获取天气信息并处理。4.1 示例云函数带抓包的天气查询# 文件名main.py (云函数入口文件) import json from cloud_function_sniffer import RequestSniffer def main_handler(event, context): 云函数主入口。 event: 触发事件的数据例如API网关传入的参数。 context: 运行上下文信息。 # 解析请求参数例如从event中获取城市名 city event.get(‘queryString‘, {}).get(‘city‘, ‘Beijing‘) # 关键步骤在需要抓包的代码块外包裹with语句 with RequestSniffer(max_body_length2048, skip_binaryTrue) as sniffer: # 以下所有通过requests发起的请求都会被记录 try: # 示例1: 调用一个公开的天气API import requests api_url f‘https://api.openweathermap.org/data/2.5/weather‘ params {‘q‘: city, ‘appid‘: ‘your_api_key_here‘, ‘units‘: ‘metric‘} resp requests.get(api_url, paramsparams, timeout5) weather_data resp.json() # 示例2: 将处理后的数据发送到自己的日志服务这个请求也会被捕获 log_payload {‘city‘: city, ‘temp‘: weather_data[‘main‘][‘temp‘]} # 假设我们有一个内部日志接口 # internal_log_url ‘https://your-log-service.com/log‘ # requests.post(internal_log_url, jsonlog_payload) # 这行被注释掉但若执行也会被抓包 except requests.exceptions.RequestException as e: return { ‘statusCode‘: 500, ‘body‘: json.dumps({‘error‘: f‘Weather API call failed: {str(e)}‘}) } # 返回处理结果 return { ‘statusCode‘: 200, ‘headers‘: {‘Content-Type‘: ‘application/json‘}, ‘body‘: json.dumps({ ‘city‘: city, ‘temperature‘: weather_data[‘main‘][‘temp‘], ‘description‘: weather_data[‘weather‘][0][‘description‘] }) }部署与查看日志将cloud_function_sniffer.py和main.py一起打包上传到你的云函数。触发云函数例如通过小程序调用或API网关测试。在云服务商提供的日志查询界面如腾讯云的“函数日志”、阿里云的“函数计算日志”你就能看到格式化输出的抓包信息了。4.2 输出日志示例在云函数日志中你可能会看到如下输出 [HTTP Request] GET https://api.openweathermap.org/data/2.5/weather?qBeijingappidxxxunitsmetric Status: SUCCESS (200) Time: 2023-10-27 10:30:00 | Elapsed: 0.452s ---------------------------------------- Request Headers: User-Agent: python-requests/2.28.2 Accept-Encoding: gzip, deflate Accept: */* Connection: keep-alive Request Body Preview: [Binary or Empty] ---------------------------------------- Response Headers: Content-Type: application/json; charsetutf-8 Content-Length: 456 Date: Thu, 27 Oct 2023 02:30:00 GMT Response Body Preview: { “coord“: { “lon“: 116.3972, “lat“: 39.9075 }, “weather“: [ { “id“: 801, “main“: “Clouds“, “description“: “few clouds“, “icon“: “02d“ } ], “main“: { “temp“: 15.6, “feels_like“: 14.8, ... } } 5. 高级技巧与疑难排查在实际使用中你可能会遇到一些特殊情况。这里分享几个踩坑后总结的经验。5.1 处理异步HTTP客户端如httpx, aiohttp现代云函数可能支持异步运行时。如果你的云函数使用了httpx或aiohttp我们需要为它们编写对应的嗅探器。原理类似但实现细节不同。以httpx为例import httpx from .request_sniffer_base import BaseSniffer # 假设有一个基础类 class HTTPXSniffer(BaseSniffer): def __init__(self, **kwargs): super().__init__(**kwargs) self._original_client_send None def _patched_client_send(self, request: httpx.Request) - httpx.Response: start_time time.time() req_info self._capture_httpx_request(request) try: response self._original_client_send(request) elapsed time.time() - start_time resp_info self._capture_httpx_response(response, elapsed) log_entry {**req_info, **resp_info, ‘status‘: ‘success‘} return response except Exception as e: elapsed time.time() - start_time # ... 错误处理 raise finally: self.output_handler(log_entry) def start(self): if self._is_active: return self._original_client_send httpx.Client.send httpx.Client.send self._patched_client_send # 同样需要处理httpx.AsyncClient如果用到 self._is_active True注意aiohttp的补丁会更复杂因为它涉及异步上下文。通常需要同时补丁ClientSession._request和aiohttp.client_reqrep.ClientResponse的相关方法。建议根据实际使用的库单独编写适配器。5.2 常见问题与解决方案问题现象可能原因解决方案抓包工具不生效没有日志输出1.RequestSniffer的with语句块没有覆盖到目标请求代码。2. 目标请求不是通过requests库发出的可能用了urllib3,http.client或其它自定义客户端。3. 云函数环境有多个Python进程或线程补丁未应用到所有环境。1. 检查with语句的位置确保目标请求发生在该上下文内。2. 确认项目使用的HTTP库。如果是httpx等需要使用对应的嗅探器。3. 在云函数入口处尽早激活嗅探器sniffer.start()并确保在函数结束时停止。使用with语句是最稳妥的方式。日志输出混乱或重复1. 猴子补丁被多次激活导致_patched_send被嵌套调用多次。2. 云函数被并发调用多个请求的日志交织在一起。1. 确保start()和stop()成对调用。使用with语句可避免此问题。2. 在日志条目中增加唯一请求ID可以从event或context中获取或使用uuid。修改_default_output或自定义output_handler来包含此ID。云函数内存暴涨或超时1. 抓包时没有跳过二进制响应且响应体巨大如文件下载。2.max_body_length设置过大导致大JSON响应被完整读入内存。1.务必设置skip_binaryTrue。这是生产环境最重要的安全设置。2. 根据实际需要调整max_body_length对于调试API2KB-10KB通常足够。如果必须分析大响应考虑只记录其元数据大小、哈希、前N个字节。补丁影响了其他不相关的请求猴子补丁是全局性的在with块外通过requests发起的请求也会被记录如果补丁未停止。严格使用上下文管理器 (with)。如果必须在全局激活确保在云函数所有逻辑结束后调用stop()。更好的架构是将需要抓包的逻辑隔离到一个单独的函数或模块中。无法捕获到某些请求头如Authorizationrequests库或目标服务器可能在特定阶段修改或重写请求头。我们捕获的是PreparedRequest对象它代表即将发送的请求。我们捕获的已经是最终要发送的版本。如果这里都没有那请求确实没带这个头。可以检查是否是库的bug或者请求头在更早的阶段被移除了。可以在_capture_request中打印request.headers来验证。5.3 性能优化与生产环境建议选择性抓包不要在所有云函数、所有请求上都开启抓包。可以通过环境变量来控制例如ENABLE_SNIFFERtrue。只在需要调试或审计的特定场景下开启。采样率可以在RequestSniffer类中添加一个sample_rate参数如0.01表示1%的请求被抓包避免产生海量日志。异步输出默认的print是同步操作可能会阻塞请求。如果对性能极其敏感可以考虑将日志条目放入一个内存队列由另一个后台线程异步消费并输出。但要注意云函数的运行时长限制。结构化日志与外部存储对于长期分析建议自定义output_handler将抓包数据以结构化格式如JSON行发送到云日志服务或对象存储。这样便于后续使用日志分析工具如ELK进行查询和统计。这个纯Python实现的云函数抓包工具本质上是应用层逆向的一个实用技巧。它剥离了复杂系统工具的外壳直击“看清数据流”这一核心需求。通过理解其原理并根据实际场景调整你可以将它灵活地应用于各种云端函数的调试、分析和安全审计工作中。 Python实现云函数抓包：应用层请求拦截与调试实践
1. 项目缘起为什么要在云函数里抓包最近在折腾一些小程序的数据分析发现很多核心逻辑和数据交互都跑在云函数上。直接在小程序端抓包看到的往往是加密后的数据流或者干脆就是一堆看不懂的二进制。这时候一个很自然的想法就冒出来了能不能直接在云函数的执行环境里把发出去的请求和收到的响应给“看”清楚这就是“云函数抓包”的核心诉求。听起来有点像在服务器上装个抓包软件但云函数的环境是临时的、无状态的而且权限限制通常很严格不可能让你去安装像 Wireshark 或 tcpdump 这样的系统级工具。所以我们必须换一种思路——在应用层用纯 Python 来实现网络请求的拦截和记录。这不仅仅是抓包更是一种针对特定应用小程序云函数的逆向分析手段目的是理解其数据流转的完整链条为后续的数据分析、接口调试甚至安全审计打下基础。这个方法特别适合开发者、安全研究员或者数据分析师。比如你想分析某个小程序商城的价格波动规律但数据接口被混淆了或者你需要调试自己编写的云函数想确认发出的 HTTP 请求格式是否正确再或者你在进行安全评估需要梳理云函数对外部服务的依赖情况。在这些场景下一个轻量级、可嵌入的 Python 抓包工具就显得非常实用。2. 核心思路在应用层实现请求拦截既然不能动系统那我们就从应用层入手。小程序云函数本质上是一段运行在云端的 Node.js 或 Python 代码它对外发起网络请求最常用的库无非是requestsPython或axios/原生http模块Node.js。我们的目标就是“劫持”这些库的请求发出和响应接收过程。2.1 方案选型猴子补丁 vs. 代理中间件要实现这个目标主要有两种技术路径猴子补丁在运行时动态替换掉requests库或http.client等底层模块的关键函数比如将requests.request或requests.Session.send替换成我们自己的函数。在这个自定义函数里我们先记录请求的详细信息然后调用原始函数完成实际请求再记录响应信息最后将响应返回。这种方式侵入性强但实现直接能捕获到最原始的请求。代理中间件不直接修改库函数而是创建一个代理对象或适配器。例如我们可以封装一个自己的Session类它内部使用真正的requests.Session但在request方法前后加入日志逻辑。或者利用requests库的hooks机制response钩子来记录响应。这种方式更优雅耦合度低但可能无法捕获到某些非常规方式发出的请求。对于云函数这种我们通常能控制代码的环境猴子补丁的彻底性更有优势。我们选择从requests.Session.send这个最底层、最核心的方法入手进行替换确保所有通过requests库发起的请求无论调用层级如何都能被我们捕获。2.2 技术要点确保捕获的完整性与性能在云函数中实现抓包有几个关键点需要考虑请求/响应的全信息记录不仅仅是 URL 和状态码。我们需要记录请求方法、完整的请求头、请求体可能是文本、JSON 或二进制文件以及响应的状态码、响应头和响应体。对于 JSON 数据最好能格式化输出以便阅读。对性能的影响要最小化云函数是按执行时间和内存消耗计费的。我们的抓包逻辑不能显著增加函数的运行开销。这意味着要避免深度拷贝大体积的请求/响应体对于二进制数据如图片可能只记录其大小和哈希值而非全部内容。输出目标的选择在云函数里我们不能把日志写到本地文件。最直接的输出方式是打印到标准输出print这些日志会被收集到云服务商提供的日志服务中。我们也可以将抓包数据结构化后发送到指定的日志收集接口或对象存储中但这会增加额外的网络请求。异步请求的支持越来越多的云函数开始支持异步编程。如果云函数中使用了aiohttp或httpx这样的异步 HTTP 客户端我们的抓包方案也需要有相应的异步版本。基于以上考量我们将设计一个核心的RequestSniffer类它使用猴子补丁技术并提供一个上下文管理器确保抓包行为在可控的范围内开启和关闭。3. 核心代码实现与逐行解析接下来我们动手实现这个纯 Python 的云函数抓包工具。我们将代码命名为cloud_function_sniffer.py。3.1 基础架构与导入import json import time import hashlib import requests from typing import Any, Dict, Optional, Union, Callable from urllib.parse import urlparse import functoolsjson: 用于格式化 JSON 请求/响应体。time: 记录请求耗时。hashlib: 用于计算大响应体如文件的哈希值避免记录全部内容。requests: 这是我们即将要“补丁”的目标库。typing: 提供类型注解让代码更清晰。urllib.parse: 用于解析 URL提取主机名等有用信息。functools: 使用wraps装饰器来保留被替换函数的元数据。3.2 核心类RequestSnifferclass RequestSniffer: HTTP请求嗅探器用于拦截和记录requests库发起的网络请求。 def __init__(self, max_body_length: int 1024 * 10, # 10KB skip_binary: bool True, output_handler: Optional[Callable[[Dict], None]] None): 初始化嗅探器。 Args: max_body_length: 记录请求/响应体的最大长度字节。超出部分将被截断。 skip_binary: 是否跳过记录二进制内容如图片、文件流。若为True则只记录大小和MD5。 output_handler: 自定义输出处理器。默认为None使用print输出到控制台。 self.max_body_length max_body_length self.skip_binary skip_binary self.output_handler output_handler or self._default_output self._original_send None # 保存原始的requests.Session.send方法 self._is_active Falsemax_body_length: 这是一个重要的性能和安全参数。云函数的响应体可能非常大比如下载一个文件。全部记录会消耗大量内存和日志额度。我们设置一个上限默认10KB超出的部分只记录摘要信息。skip_binary: 二进制数据如图片、PDF在日志里是一堆乱码没有可读性。开启此选项可以显著减少日志体积。output_handler: 提供灵活性。默认是打印到控制台但用户可以传入一个函数将抓包数据写入数据库、发送到日志平台等。_original_send: 保存requests.Session.send的原始引用这是猴子补丁的关键。_is_active: 状态标志防止重复激活。3.3 猴子补丁的核心自定义send方法这是整个工具最核心的部分我们创建一个新的_patched_send方法来替换原来的send。def _patched_send(self, request, **kwargs): 被替换后的Session.send方法包裹了日志记录逻辑。 start_time time.time() # 1. 在发送前记录请求信息 req_info self._capture_request(request, kwargs) # 2. 调用原始的send方法执行真正的网络请求 try: response self._original_send(request, **kwargs) elapsed time.time() - start_time # 3. 请求成功后记录响应信息 resp_info self._capture_response(response, elapsed) log_entry {**req_info, **resp_info, ‘status‘: ‘success‘} except Exception as e: elapsed time.time() - start_time # 4. 请求失败后记录异常信息 resp_info { ‘error‘: str(e), ‘elapsed‘: f‘{elapsed:.3f}s‘, ‘status_code‘: None, } log_entry {**req_info, **resp_info, ‘status‘: ‘error‘} raise # 将异常原样抛出不影响云函数原有逻辑 finally: # 5. 无论成功失败都输出日志条目 self.output_handler(log_entry) return response关键点解析计时开始在调用原始send前记录时间用于计算请求耗时。捕获请求调用_capture_request方法提取请求的详细信息。执行与捕获响应调用原始send这是实际网络发生的地方。成功后调用_capture_response记录响应。异常处理如果请求失败超时、网络错误等我们捕获异常记录错误信息但必须重新抛出异常raise。这是至关重要的否则会改变云函数的错误处理逻辑导致问题被掩盖。日志输出在finally块中确保日志一定会被输出。然后返回响应或让异常继续传播。3.4 请求与响应的信息捕获_capture_request和_capture_response是两个重要的工具方法负责从原始对象中提取我们需要的信息。def _capture_request(self, request, kwargs) - Dict[str, Any]: 从PreparedRequest对象和kwargs中提取请求信息。 # 解析URL parsed_url urlparse(request.url) # 获取请求体 body request.body body_preview ‘[Binary or Empty]‘ body_size 0 body_md5 None if body: body_size len(body) if isinstance(body, bytes) else len(body.encode(‘utf-8‘)) if self.skip_binary and isinstance(body, bytes): # 如果是二进制数据且设置了跳过则计算MD5 body_md5 hashlib.md5(body).hexdigest()[:8] body_preview f‘[Binary Data, size: {body_size}, md5: {body_md5}]‘ else: # 尝试解码为文本或截断预览 try: if isinstance(body, bytes): body_str body.decode(‘utf-8‘) else: body_str str(body) # 截断过长的内容 if len(body_str) self.max_body_length: body_preview body_str[:self.max_body_length] f‘... (truncated, total: {body_size})‘ else: body_preview body_str except UnicodeDecodeError: # 解码失败视为二进制 body_md5 hashlib.md5(body).hexdigest()[:8] body_preview f‘[Non-UTF8 Data, size: {body_size}, md5: {body_md5}]‘ req_info { ‘method‘: request.method, ‘url‘: request.url, ‘host‘: parsed_url.netloc, ‘path‘: parsed_url.path, ‘request_headers‘: dict(request.headers), ‘request_body_preview‘: body_preview, ‘request_body_size‘: body_size, ‘timestamp‘: time.strftime(‘%Y-%m-%d %H:%M:%S‘), } # 记录一些可能有用的kwargs如超时设置 if ‘timeout‘ in kwargs: req_info[‘timeout‘] kwargs[‘timeout‘] return req_info注意request.body可能是bytes、str或None。我们优先尝试将其作为文本处理以便在日志中可读。但对于图片上传等场景body是二进制流直接解码会出错。因此我们通过skip_binary开关和异常捕获来优雅处理。def _capture_response(self, response, elapsed: float) - Dict[str, Any]: 从Response对象中提取响应信息。 # 获取响应体内容注意这会消耗掉响应体需要做处理 resp_content response.content body_preview ‘[Empty]‘ body_size len(resp_content) body_md5 None if resp_content: if self.skip_binary and self._is_likely_binary(response.headers): body_md5 hashlib.md5(resp_content).hexdigest()[:8] body_preview f‘[Binary Response, size: {body_size}, md5: {body_md5}]‘ else: # 尝试解码为文本 try: content_str resp_content.decode(‘utf-8‘) # 如果是JSON可以尝试美化 if ‘application/json‘ in response.headers.get(‘Content-Type‘, ‘‘): try: parsed_json json.loads(content_str) content_str json.dumps(parsed_json, indent2, ensure_asciiFalse) except json.JSONDecodeError: pass # 保持原样 # 截断 if len(content_str) self.max_body_length: body_preview content_str[:self.max_body_length] f‘... (truncated, total: {body_size})‘ else: body_preview content_str except UnicodeDecodeError: body_md5 hashlib.md5(resp_content).hexdigest()[:8] body_preview f‘[Non-UTF8 Response, size: {body_size}, md5: {body_md5}]‘ resp_info { ‘status_code‘: response.status_code, ‘response_headers‘: dict(response.headers), ‘response_body_preview‘: body_preview, ‘response_body_size‘: body_size, ‘elapsed‘: f‘{elapsed:.3f}s‘, } return resp_info staticmethod def _is_likely_binary(headers: Dict) - bool: 根据Content-Type判断内容是否可能是二进制的。 content_type headers.get(‘Content-Type‘, ‘‘).lower() binary_types [‘image/‘, ‘application/octet-stream‘, ‘video/‘, ‘audio/‘, ‘application/pdf‘, ‘application/zip‘] return any(ct in content_type for ct in binary_types)关键点解析响应体消耗问题直接读取response.content会将响应体全部加载到内存。对于大文件这可能是个问题。但在抓包调试场景下我们通常关注的是接口返回的元数据和少量文本数据。skip_binary和max_body_length在这里起到了保护作用。切记在生产环境云函数中如果预期会下载大文件请务必将skip_binary设为True并谨慎设置max_body_length。智能内容处理我们通过Content-Type头来辅助判断是否为二进制数据。对于application/json我们尝试解析并美化让日志更易读。性能考量计算 MD5 和字符串解码都有开销。这就是为什么我们提供开关允许用户为性能而牺牲部分信息细节。3.5 激活、关闭与上下文管理器为了方便使用我们提供start()、stop()方法和一个上下文管理器。def start(self): 激活请求嗅探。 if self._is_active: return # 保存原始方法 self._original_send requests.Session.send # 创建绑定到当前实例的新函数 patched functools.partial(self._patched_send, self) # 替换类方法 requests.Session.send patched self._is_active True def stop(self): 停止请求嗅探恢复原状。 if self._is_active and self._original_send: requests.Session.send self._original_send self._original_send None self._is_active False def __enter__(self): 支持with语句。 self.start() return self def __exit__(self, exc_type, exc_val, exc_tb): 退出with语句块时自动停止。 self.stop() def _default_output(self, entry: Dict): 默认输出处理器格式化打印到控制台。 print(‘\n‘ ‘‘*60) print(f“[HTTP Request] {entry.get(‘method‘)} {entry.get(‘url‘)}“) print(f“Status: {entry.get(‘status‘).upper()} ({entry.get(‘status_code‘, ‘N/A‘)})“) print(f“Time: {entry.get(‘timestamp‘)} | Elapsed: {entry.get(‘elapsed‘, ‘N/A‘)}“) print(‘-‘*40) print(‘Request Headers:‘) for k, v in entry.get(‘request_headers‘, {}).items(): print(f‘ {k}: {v}‘) if entry.get(‘request_body_preview‘) and entry[‘request_body_preview‘] ! ‘[Binary or Empty]‘: print(‘Request Body Preview:‘) print(f‘ {entry[“request_body_preview“]}‘) print(‘-‘*40) print(‘Response Headers:‘) for k, v in entry.get(‘response_headers‘, {}).items(): print(f‘ {k}: {v}‘) if entry.get(‘response_body_preview‘) and entry[‘response_body_preview‘] ! ‘[Empty]‘: print(‘Response Body Preview:‘) # 对可能的多行内容进行缩进 preview entry[‘response_body_preview‘] lines preview.split(‘\n‘) for line in lines[:10]: # 最多打印前10行 print(f‘ {line}‘) if len(lines) 10: print(f‘ ... (and {len(lines)-10} more lines)‘) print(‘‘*60 ‘\n‘)start和stop这是标准的猴子补丁流程。functools.partial用于将实例方法_patched_send绑定到当前实例 (self)然后替换类的send方法。上下文管理器使用with RequestSniffer() as sniffer:的语法可以确保在任何情况下即使发生异常stop()都会被调用避免污染全局环境。这是在云函数中使用的最佳实践。_default_output将抓包数据格式化为易读的文本输出到控制台。在实际使用中你可以重写这个方法或者通过output_handler参数传入自定义函数将数据发送到云日志服务如腾讯云的CLS、阿里云的SLS或你自己的日志服务器。4. 在云函数中的实战应用现在我们来看如何将这个工具集成到一个真实的小程序云函数中。假设我们有一个云函数它的作用是获取天气信息并处理。4.1 示例云函数带抓包的天气查询# 文件名main.py (云函数入口文件) import json from cloud_function_sniffer import RequestSniffer def main_handler(event, context): 云函数主入口。 event: 触发事件的数据例如API网关传入的参数。 context: 运行上下文信息。 # 解析请求参数例如从event中获取城市名 city event.get(‘queryString‘, {}).get(‘city‘, ‘Beijing‘) # 关键步骤在需要抓包的代码块外包裹with语句 with RequestSniffer(max_body_length2048, skip_binaryTrue) as sniffer: # 以下所有通过requests发起的请求都会被记录 try: # 示例1: 调用一个公开的天气API import requests api_url f‘https://api.openweathermap.org/data/2.5/weather‘ params {‘q‘: city, ‘appid‘: ‘your_api_key_here‘, ‘units‘: ‘metric‘} resp requests.get(api_url, paramsparams, timeout5) weather_data resp.json() # 示例2: 将处理后的数据发送到自己的日志服务这个请求也会被捕获 log_payload {‘city‘: city, ‘temp‘: weather_data[‘main‘][‘temp‘]} # 假设我们有一个内部日志接口 # internal_log_url ‘https://your-log-service.com/log‘ # requests.post(internal_log_url, jsonlog_payload) # 这行被注释掉但若执行也会被抓包 except requests.exceptions.RequestException as e: return { ‘statusCode‘: 500, ‘body‘: json.dumps({‘error‘: f‘Weather API call failed: {str(e)}‘}) } # 返回处理结果 return { ‘statusCode‘: 200, ‘headers‘: {‘Content-Type‘: ‘application/json‘}, ‘body‘: json.dumps({ ‘city‘: city, ‘temperature‘: weather_data[‘main‘][‘temp‘], ‘description‘: weather_data[‘weather‘][0][‘description‘] }) }部署与查看日志将cloud_function_sniffer.py和main.py一起打包上传到你的云函数。触发云函数例如通过小程序调用或API网关测试。在云服务商提供的日志查询界面如腾讯云的“函数日志”、阿里云的“函数计算日志”你就能看到格式化输出的抓包信息了。4.2 输出日志示例在云函数日志中你可能会看到如下输出 [HTTP Request] GET https://api.openweathermap.org/data/2.5/weather?qBeijingappidxxxunitsmetric Status: SUCCESS (200) Time: 2023-10-27 10:30:00 | Elapsed: 0.452s ---------------------------------------- Request Headers: User-Agent: python-requests/2.28.2 Accept-Encoding: gzip, deflate Accept: */* Connection: keep-alive Request Body Preview: [Binary or Empty] ---------------------------------------- Response Headers: Content-Type: application/json; charsetutf-8 Content-Length: 456 Date: Thu, 27 Oct 2023 02:30:00 GMT Response Body Preview: { “coord“: { “lon“: 116.3972, “lat“: 39.9075 }, “weather“: [ { “id“: 801, “main“: “Clouds“, “description“: “few clouds“, “icon“: “02d“ } ], “main“: { “temp“: 15.6, “feels_like“: 14.8, ... } } 5. 高级技巧与疑难排查在实际使用中你可能会遇到一些特殊情况。这里分享几个踩坑后总结的经验。5.1 处理异步HTTP客户端如httpx, aiohttp现代云函数可能支持异步运行时。如果你的云函数使用了httpx或aiohttp我们需要为它们编写对应的嗅探器。原理类似但实现细节不同。以httpx为例import httpx from .request_sniffer_base import BaseSniffer # 假设有一个基础类 class HTTPXSniffer(BaseSniffer): def __init__(self, **kwargs): super().__init__(**kwargs) self._original_client_send None def _patched_client_send(self, request: httpx.Request) - httpx.Response: start_time time.time() req_info self._capture_httpx_request(request) try: response self._original_client_send(request) elapsed time.time() - start_time resp_info self._capture_httpx_response(response, elapsed) log_entry {**req_info, **resp_info, ‘status‘: ‘success‘} return response except Exception as e: elapsed time.time() - start_time # ... 错误处理 raise finally: self.output_handler(log_entry) def start(self): if self._is_active: return self._original_client_send httpx.Client.send httpx.Client.send self._patched_client_send # 同样需要处理httpx.AsyncClient如果用到 self._is_active True注意aiohttp的补丁会更复杂因为它涉及异步上下文。通常需要同时补丁ClientSession._request和aiohttp.client_reqrep.ClientResponse的相关方法。建议根据实际使用的库单独编写适配器。5.2 常见问题与解决方案问题现象可能原因解决方案抓包工具不生效没有日志输出1.RequestSniffer的with语句块没有覆盖到目标请求代码。2. 目标请求不是通过requests库发出的可能用了urllib3,http.client或其它自定义客户端。3. 云函数环境有多个Python进程或线程补丁未应用到所有环境。1. 检查with语句的位置确保目标请求发生在该上下文内。2. 确认项目使用的HTTP库。如果是httpx等需要使用对应的嗅探器。3. 在云函数入口处尽早激活嗅探器sniffer.start()并确保在函数结束时停止。使用with语句是最稳妥的方式。日志输出混乱或重复1. 猴子补丁被多次激活导致_patched_send被嵌套调用多次。2. 云函数被并发调用多个请求的日志交织在一起。1. 确保start()和stop()成对调用。使用with语句可避免此问题。2. 在日志条目中增加唯一请求ID可以从event或context中获取或使用uuid。修改_default_output或自定义output_handler来包含此ID。云函数内存暴涨或超时1. 抓包时没有跳过二进制响应且响应体巨大如文件下载。2.max_body_length设置过大导致大JSON响应被完整读入内存。1.务必设置skip_binaryTrue。这是生产环境最重要的安全设置。2. 根据实际需要调整max_body_length对于调试API2KB-10KB通常足够。如果必须分析大响应考虑只记录其元数据大小、哈希、前N个字节。补丁影响了其他不相关的请求猴子补丁是全局性的在with块外通过requests发起的请求也会被记录如果补丁未停止。严格使用上下文管理器 (with)。如果必须在全局激活确保在云函数所有逻辑结束后调用stop()。更好的架构是将需要抓包的逻辑隔离到一个单独的函数或模块中。无法捕获到某些请求头如Authorizationrequests库或目标服务器可能在特定阶段修改或重写请求头。我们捕获的是PreparedRequest对象它代表即将发送的请求。我们捕获的已经是最终要发送的版本。如果这里都没有那请求确实没带这个头。可以检查是否是库的bug或者请求头在更早的阶段被移除了。可以在_capture_request中打印request.headers来验证。5.3 性能优化与生产环境建议选择性抓包不要在所有云函数、所有请求上都开启抓包。可以通过环境变量来控制例如ENABLE_SNIFFERtrue。只在需要调试或审计的特定场景下开启。采样率可以在RequestSniffer类中添加一个sample_rate参数如0.01表示1%的请求被抓包避免产生海量日志。异步输出默认的print是同步操作可能会阻塞请求。如果对性能极其敏感可以考虑将日志条目放入一个内存队列由另一个后台线程异步消费并输出。但要注意云函数的运行时长限制。结构化日志与外部存储对于长期分析建议自定义output_handler将抓包数据以结构化格式如JSON行发送到云日志服务或对象存储。这样便于后续使用日志分析工具如ELK进行查询和统计。这个纯Python实现的云函数抓包工具本质上是应用层逆向的一个实用技巧。它剥离了复杂系统工具的外壳直击“看清数据流”这一核心需求。通过理解其原理并根据实际场景调整你可以将它灵活地应用于各种云端函数的调试、分析和安全审计工作中。