Python实现云函数抓包：应用层请求拦截与调试实践

1. 项目缘起：为什么要在云函数里抓包？

最近在折腾一些小程序的数据分析，发现很多核心逻辑和数据交互都跑在云函数上。直接在小程序端抓包，看到的往往是加密后的数据流，或者干脆就是一堆看不懂的二进制。这时候，一个很自然的想法就冒出来了：能不能直接在云函数的执行环境里，把发出去的请求和收到的响应给“看”清楚？这就是“云函数抓包”的核心诉求。

听起来有点像在服务器上装个抓包软件，但云函数的环境是临时的、无状态的，而且权限限制通常很严格，不可能让你去安装像 Wireshark 或 tcpdump 这样的系统级工具。所以，我们必须换一种思路——在应用层，用纯 Python 来实现网络请求的拦截和记录。这不仅仅是抓包，更是一种针对特定应用（小程序云函数）的逆向分析手段，目的是理解其数据流转的完整链条，为后续的数据分析、接口调试甚至安全审计打下基础。

这个方法特别适合开发者、安全研究员或者数据分析师。比如，你想分析某个小程序商城的价格波动规律，但数据接口被混淆了；或者你需要调试自己编写的云函数，想确认发出的 HTTP 请求格式是否正确；再或者，你在进行安全评估，需要梳理云函数对外部服务的依赖情况。在这些场景下，一个轻量级、可嵌入的 Python 抓包工具就显得非常实用。

2. 核心思路：在应用层实现请求拦截

既然不能动系统，那我们就从应用层入手。小程序云函数本质上是一段运行在云端的 Node.js 或 Python 代码，它对外发起网络请求，最常用的库无非是requests（Python）或axios/原生http模块（Node.js）。我们的目标就是“劫持”这些库的请求发出和响应接收过程。

2.1 方案选型：猴子补丁 vs. 代理中间件

要实现这个目标，主要有两种技术路径：

1. 猴子补丁：在运行时动态替换掉requests库（或http.client等底层模块）的关键函数，比如将requests.request或requests.Session.send替换成我们自己的函数。在这个自定义函数里，我们先记录请求的详细信息，然后调用原始函数完成实际请求，再记录响应信息，最后将响应返回。这种方式侵入性强，但实现直接，能捕获到最原始的请求。
2. 代理中间件：不直接修改库函数，而是创建一个代理对象或适配器。例如，我们可以封装一个自己的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 functools

• json: 用于格式化 JSON 请求/响应体。
• time: 记录请求耗时。
• hashlib: 用于计算大响应体（如文件）的哈希值，避免记录全部内容。
• requests: 这是我们即将要“补丁”的目标库。
• typing: 提供类型注解，让代码更清晰。
• urllib.parse: 用于解析 URL，提取主机名等有用信息。
• functools: 使用wraps装饰器来保留被替换函数的元数据。

3.2 核心类：RequestSniffer

class 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 = False

• max_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

关键点解析：

1. 计时开始：在调用原始send前记录时间，用于计算请求耗时。
2. 捕获请求：调用_capture_request方法提取请求的详细信息。
3. 执行与捕获响应：调用原始send，这是实际网络发生的地方。成功后，调用_capture_response记录响应。
4. 异常处理：如果请求失败（超时、网络错误等），我们捕获异常，记录错误信息，但必须重新抛出异常（raise）。这是至关重要的，否则会改变云函数的错误处理逻辑，导致问题被掩盖。
5. 日志输出：在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, indent=2, ensure_ascii=False) 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)

关键点解析：

1. 响应体消耗问题：直接读取response.content会将响应体全部加载到内存。对于大文件，这可能是个问题。但在抓包调试场景下，我们通常关注的是接口返回的元数据和少量文本数据。skip_binary和max_body_length在这里起到了保护作用。切记，在生产环境云函数中，如果预期会下载大文件，请务必将skip_binary设为True，并谨慎设置max_body_length。
2. 智能内容处理：我们通过Content-Type头来辅助判断是否为二进制数据。对于application/json，我们尝试解析并美化，让日志更易读。
3. 性能考量：计算 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_length=2048, skip_binary=True) 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, params=params, timeout=5) 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, json=log_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‘] }) }

部署与查看日志：

1. 将cloud_function_sniffer.py和main.py一起打包上传到你的云函数。
2. 触发云函数（例如通过小程序调用或API网关测试）。
3. 在云服务商提供的日志查询界面（如腾讯云的“函数日志”、阿里云的“函数计算日志”），你就能看到格式化输出的抓包信息了。

4.2 输出日志示例

在云函数日志中，你可能会看到如下输出：

============================================================ [HTTP Request] GET https://api.openweathermap.org/data/2.5/weather?q=Beijing&appid=xxx&units=metric 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; charset=utf-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 常见问题与解决方案

5.3 性能优化与生产环境建议

1. 选择性抓包：不要在所有云函数、所有请求上都开启抓包。可以通过环境变量来控制，例如ENABLE_SNIFFER=true。只在需要调试或审计的特定场景下开启。
2. 采样率：可以在RequestSniffer类中添加一个sample_rate参数（如0.01表示1%的请求被抓包），避免产生海量日志。
3. 异步输出：默认的print是同步操作，可能会阻塞请求。如果对性能极其敏感，可以考虑将日志条目放入一个内存队列，由另一个后台线程异步消费并输出。但要注意云函数的运行时长限制。
4. 结构化日志与外部存储：对于长期分析，建议自定义output_handler，将抓包数据以结构化格式（如JSON行）发送到云日志服务或对象存储。这样便于后续使用日志分析工具（如ELK）进行查询和统计。

这个纯Python实现的云函数抓包工具，本质上是应用层逆向的一个实用技巧。它剥离了复杂系统工具的外壳，直击“看清数据流”这一核心需求。通过理解其原理并根据实际场景调整，你可以将它灵活地应用于各种云端函数的调试、分析和安全审计工作中。