1. 项目缘起当小程序云函数成为“黑盒”最近在折腾一个微信小程序的数据分析项目客户的小程序大量使用了云函数来处理核心业务逻辑比如用户签到、积分兑换、数据上报等。问题来了这些云函数的代码和逻辑都跑在云端我们作为开发者或者安全研究员想了解它具体发送了什么数据、接收了什么响应简直像隔着一堵墙。官方调试工具提供的日志有限很多时候只能看到“调用成功”或“失败”具体的请求体、响应头、甚至是第三方接口的交互细节完全是个黑盒。这让我想起了早些年做Web端和移动端逆向分析时抓包是基本功。Fiddler、Charles、Wireshark都是老朋友。但面对小程序云函数这些传统工具似乎有点“鞭长莫及”。云函数的执行环境在腾讯云服务器上不在本地你没法直接给那台远程服务器装个代理。网上搜了一圈发现讨论这个的并不多大多停留在理论或者使用一些商业化的、需要特定环境的抓包方案上。于是一个想法冒了出来能不能用最纯粹的Python写一个轻量级的工具模拟出类似“中间人”的效果把云函数发起的网络请求给“钓”出来这个工具的目标很明确无需复杂的环境配置不依赖特定操作系统或硬件比如AX210网卡纯粹在代码层面实现对一个运行中Python进程模拟云函数环境的网络流量拦截和解析。这就是今天要分享的“纯Python实现小程序云函数抓包”项目的由来。它本质上是一种针对特定运行环境Python进程的流量Hook技术对于分析小程序云函数、某些Python SDK的内部网络行为或者进行安全的黑盒测试会非常有用。2. 核心思路从“代理”到“内存Hook”的转变一开始我最自然的想法是走代理模式。既然不能直接抓远程服务器的包那能不能让云函数的代码把所有网络请求都发往我本地启动的一个HTTP/HTTPS代理服务器呢这样我就能在代理服务器上看到所有流量。理论上可行但实操问题很多代码侵入性你需要修改云函数的源代码显式地设置http_proxy或https_proxy环境变量或者为requests、aiohttp等库配置代理。对于已编译的、不可修改的第三方包或者代码逻辑复杂的项目这很困难。HTTPS解密如果云函数请求的是HTTPS接口代理服务器需要安装自签名CA证书并进行MITM中间人解密。这在小程序模拟环境或某些受限制的服务器环境里配置起来非常麻烦且可能触发证书校验错误。环境依赖你需要一个稳定的、能处理并发请求的代理服务器比如用mitmproxy整个工具链变重了。那么有没有更底层、更通用的方法答案是有的那就是内存函数Hook。我们不必关心云函数代码里用的是requests.get还是urllib.request.urlopen抑或是aiohttp.ClientSession。在Python中这些库最终都会调用Python标准库socket模块的底层函数来创建连接和收发数据。如果我们能在运行时动态地替换掉这些关键的socket函数比如socket.sendall和socket.recv那么所有经由这些函数进出网络的数据都会先流经我们编写的“钩子”函数。在这个“钩子”里我们就可以记录、修改、分析这些原始的网络数据包。这个思路的优势非常明显零侵入性无需修改目标云函数的一行源代码。只需要在目标Python进程启动时或通过调试器附加后注入我们的Hook代码即可。通用性强无论云函数内部使用何种HTTP库requests,httpx,urllib,aiohttp只要它们基于Python的socket就会被捕获。信息完整获取的是最原始的TCP Socket层数据包含了完整的TCP/IP载荷我们可以从中解析出HTTP/HTTPS的明文如果是HTTP或加密流量如果是HTTPS需要后续解密但至少能看见流量特征。当然这种方法也有挑战主要是对Python运行时和线程安全的要求较高。接下来我们就一步步拆解如何实现这个“内存Hook”抓包器。3. 技术选型与关键库sys.settrace与socket劫持要实现运行时函数替换我们需要一个“入口点”来执行我们的Hook代码。这里有两个主流方向使用调试器接口ptrace/sys.settracesys.settrace是Python标准库提供的调试钩子它可以设置一个跟踪函数在代码执行到每一行时被调用。虽然通常用于调试和性能分析但我们也可以利用它在函数被调用时执行我们的代码。另一种更底层的是使用ptraceLinux或类似的系统调用但这需要C扩展且跨平台兼容性差。使用代码注入injectpy或PyInstaller钩子通过LD_PRELOADLinux或DLL注入Windows等方式将我们的代码注入到目标Python解释器进程中。这更接近传统逆向工程但实现复杂对环境依赖强。为了追求“纯Python”和相对简单我们选择方案1并聚焦于sys.settrace。但请注意sys.settrace是单线程的并且对性能有影响它更适合用于分析、调试场景而非生产环境的高频抓包。对于我们的云函数抓包通常是短期、低频的调试任务它是完全可行的。核心库就是Python自带的sys和socket。我们的工作流程将是编写一个Hook函数用于替换socket.socket类的sendall和recv方法。编写一个Trace函数通过sys.settrace设置监控socket.socket对象的创建。一旦监控到socket.socket实例被创建立即将其sendall和recv方法替换为我们自定义的、带日志记录功能的版本。这样当云函数的代码调用socket.sendall(data)发送数据时数据会先传到我们的函数我们记录下它然后再调用原始的socket.sendall发送出去。接收过程同理。注意直接Hooksocket模块的函数如socket.send是全局的会影响进程内所有socket操作可能干扰其他正常功能。我们选择Hooksocket.socket实例的方法针对性更强可以通过判断目标地址如是否是云函数调用的特定API域名来决定是否记录减少噪音。4. 完整代码实现与逐行解析下面就是完整的抓包工具核心代码。我们将它保存为一个独立的Python模块比如叫cloud_function_sniffer.py。使用时只需要在运行你的云函数脚本之前先导入并启动这个抓包器即可。#!/usr/bin/env python3 # -*- coding: utf-8 -*- 纯Python实现的小程序云函数网络流量抓包工具 原理通过sys.settrace Hook socket.socket对象的sendall和recv方法 import sys import socket import threading import json import re from datetime import datetime from urllib.parse import urlparse import typing as t # 全局锁防止多线程下替换方法时出现竞争条件 _hook_lock threading.Lock() # 存储原始方法的字典key为socket对象idvalue为原始方法元组(sendall, recv) _original_methods {} # 抓包数据存储列表 captured_packets [] def _make_logged_sendall(original_sendall, sock_obj): 创建一个带日志记录的sendall函数 :param original_sendall: 原始的sendall方法 :param sock_obj: 对应的socket对象 :return: 新的sendall函数 def logged_sendall(data: bytes, *args, **kwargs) - t.Any: # 记录发送的数据 local_time datetime.now().strftime(%Y-%m-%d %H:%M:%S.%f)[:-3] peer_name sock_obj.getpeername() if hasattr(sock_obj, getpeername) else (unknown, 0) packet_info { type: SEND, time: local_time, socket_id: id(sock_obj), peer_address: f{peer_name[0]}:{peer_name[1]}, data_raw: data.hex(), # 原始字节的十六进制表示 data_preview: repr(data[:200]) if len(data) 200 else repr(data), # 预览前200字节 data_length: len(data) } # 尝试解析为HTTP请求简单判断 try: decoded data.decode(utf-8, errorsignore) if decoded.startswith((GET , POST , PUT , DELETE , HEAD , OPTIONS )): packet_info[protocol] HTTP # 提取首行和Headers lines decoded.split(\r\n) packet_info[http_request_line] lines[0] headers {} for line in lines[1:]: if line : break if : in line: k, v line.split(: , 1) headers[k] v packet_info[http_headers] headers # 尝试找Body body_start decoded.find(\r\n\r\n) if body_start ! -1: body decoded[body_start4:] if body: packet_info[http_body_preview] body[:500] except UnicodeDecodeError: pass # 非文本数据保持原始十六进制 captured_packets.append(packet_info) print(f[{local_time}] [SEND to {peer_name}] Length: {len(data)} bytes) # 调用原始方法发送数据 return original_sendall(data, *args, **kwargs) return logged_sendall def _make_logged_recv(original_recv, sock_obj): 创建一个带日志记录的recv函数 :param original_recv: 原始的recv方法 :param sock_obj: 对应的socket对象 :return: 新的recv函数 def logged_recv(bufsize: int, *args, **kwargs) - bytes: # 先调用原始方法接收数据 data original_recv(bufsize, *args, **kwargs) if data: # 只记录非空数据 local_time datetime.now().strftime(%Y-%m-%d %H:%M:%S.%f)[:-3] peer_name sock_obj.getpeername() if hasattr(sock_obj, getpeername) else (unknown, 0) packet_info { type: RECV, time: local_time, socket_id: id(sock_obj), peer_address: f{peer_name[0]}:{peer_name[1]}, data_raw: data.hex(), data_preview: repr(data[:200]) if len(data) 200 else repr(data), data_length: len(data) } # 尝试解析为HTTP响应 try: decoded data.decode(utf-8, errorsignore) if decoded.startswith(HTTP/): packet_info[protocol] HTTP lines decoded.split(\r\n) packet_info[http_status_line] lines[0] headers {} for line in lines[1:]: if line : break if : in line: k, v line.split(: , 1) headers[k] v packet_info[http_headers] headers body_start decoded.find(\r\n\r\n) if body_start ! -1: body decoded[body_start4:] if body: packet_info[http_body_preview] body[:500] except UnicodeDecodeError: pass captured_packets.append(packet_info) print(f[{local_time}] [RECV from {peer_name}] Length: {len(data)} bytes) return data return logged_recv def _hook_socket_creation(frame, event, arg): sys.settrace 设置的跟踪函数用于在socket对象创建时Hook其方法。 这个函数会在代码执行时被频繁调用因此逻辑要尽量轻量。 if event call: # 当有函数被调用时 # 检查调用的函数是否是 socket.socket.__init__ 或 socket.socket (构造函数) # 通过帧的代码对象和名字来判断 code_obj frame.f_code # 检查是否在socket模块中并且函数名与socket相关 # 注意这里是一个简化判断实际可能需要更精确的匹配 if socket in code_obj.co_filename and socket in code_obj.co_name: # 等待socket对象初始化完成再Hook比较困难这里我们采用一个折中方案 # 不在此处直接Hook而是通过监控socket.socket方法的返回来进行。 # 更可靠的方法是Hook socket.socket 这个类本身的 __init__ 或 __new__。 # 为了简化我们在另一个地方见下面的hook_all_existing_sockets进行批量Hook。 pass return _hook_socket_creation # 返回自身以继续跟踪 def _replace_socket_methods(sock): 替换单个socket对象的sendall和recv方法。 使用锁保证线程安全。 with _hook_lock: sock_id id(sock) if sock_id not in _original_methods: # 保存原始方法 orig_sendall sock.sendall orig_recv sock.recv _original_methods[sock_id] (orig_sendall, orig_recv) # 替换为我们的带日志版本 sock.sendall _make_logged_sendall(orig_sendall, sock) sock.recv _make_logged_recv(orig_recv, sock) # 注意实际应用中可能还需要替换 send, recv_into, sendto 等方法 # 这里只覆盖最常用的 sendall 和 recv def hook_all_existing_sockets(): 遍历所有已存在的socket对象并Hook它们。 由于通过sys.settrace精确捕获新建socket对象比较复杂 我们可以采用一个更直接但有点“暴力”的方法定期扫描所有对象。 但更优雅的方式是直接替换 socket.socket 类的 __init__ 方法。 这里我们采用类方法替换的方案。 original_init socket.socket.__init__ def logged_init(self, *args, **kwargs): # 先调用原始的 __init__ result original_init(self, *args, **kwargs) # 然后替换这个新实例的方法 _replace_socket_methods(self) return result # 替换类的 __init__ 方法 socket.socket.__init__ logged_init # 同时Hook当前已经存在的socket对象如果有的话 # 遍历 gc.getobjects() 开销大这里仅作为示例实际可省略 # import gc # for obj in gc.getobjects(): # if isinstance(obj, socket.socket): # _replace_socket_methods(obj) print([*] Socket class __init__ method has been hooked.) def start_sniffing(): 启动抓包。 1. 替换socket.socket类的__init__方法确保之后新建的socket都被Hook。 2. 设置sys.settrace可选用于更复杂的场景本例中类替换已足够。 print([*] Starting Python socket sniffer...) hook_all_existing_sockets() # 对于已经运行的、在hook之前创建的socket上述方法可能无法捕获。 # 如果需要捕获所有可能需要更侵入性的方式如调试器注入。 # 本工具主要针对在hook启动后创建的socket连接。 print([*] Sniffer is active. All subsequent socket traffic will be logged.) print([*] Note: Traffic created before this point may not be captured.) def stop_sniffing(): 停止抓包并恢复原始方法可选。 注意恢复所有被修改的方法可能很困难因为原始引用可能已丢失。 通常用于短期调试不需要恢复。 print([*] Stopping sniffer.) # 尝试恢复这里仅作演示实际复杂环境可能无法完全恢复 with _hook_lock: for sock_id, (orig_sendall, orig_recv) in _original_methods.items(): # 我们需要通过sock_id找到对象这通常做不到。 # 所以此函数主要是一个示意。 pass # 更简单的方式是不恢复直接结束进程或忽略。 print([*] Sniffer stopped. Captured packets:, len(captured_packets)) def save_captured_data(filename: str captured_traffic.json): 将抓取到的数据保存为JSON文件 with open(filename, w, encodingutf-8) as f: json.dump(captured_packets, f, indent2, ensure_asciiFalse) print(f[*] Captured data saved to {filename}) def print_summary(): 打印抓包摘要 print(f\n{*60}) print(fCapture Summary) print(f{*60}) print(fTotal packets captured: {len(captured_packets)}) send_count sum(1 for p in captured_packets if p[type] SEND) recv_count sum(1 for p in captured_packets if p[type] RECV) print(f SEND: {send_count}) print(f RECV: {recv_count}) # 按目标地址统计 addr_stats {} for p in captured_packets: addr p.get(peer_address, unknown) addr_stats[addr] addr_stats.get(addr, 0) 1 if addr_stats: print(f\nTraffic by peer address:) for addr, count in sorted(addr_stats.items(), keylambda x: x[1], reverseTrue)[:10]: # 显示前10 print(f {addr}: {count} packets) # 显示最近的几个HTTP请求如果有 http_requests [p for p in captured_packets if p.get(type) SEND and p.get(protocol) HTTP] if http_requests: print(f\nRecent HTTP Requests:) for req in http_requests[-3:]: # 显示最后3个 print(f [{req[time]}] {req.get(http_request_line, N/A)}) print(f{*60}) # 提供一个便捷的使用方式作为上下文管理器 class SocketSnifferContext: def __enter__(self): start_sniffing() return self def __exit__(self, exc_type, exc_val, exc_tb): stop_sniffing() print_summary() # 自动保存可以根据需要开启 # save_captured_data() # 如果直接运行此脚本则进入一个简单的测试模式 if __name__ __main__: # 示例测试抓包器自身 print(Testing the sniffer on itself...) with SocketSnifferContext(): import urllib.request # 发起一个简单的HTTP请求观察是否被抓取 try: response urllib.request.urlopen(http://httpbin.org/get, timeout5) print(fTest request status: {response.status}) response.read() except Exception as e: print(fTest request failed (may be expected): {e}) print_summary() if captured_packets: save_captured_data(test_capture.json) print(\nFirst captured packet (SEND):) print(json.dumps(captured_packets[0], indent2))4.1 代码核心逻辑拆解这段代码虽然不长但有几个关键设计点值得深入探讨1. 函数工厂模式 (_make_logged_sendall和_make_logged_recv)这是实现Hook的核心。我们没有直接修改socket.socket类的原生方法而是为每一个socket对象实例动态创建了一个新的方法。这个新方法是一个闭包它捕获了原始的original_sendall方法和当前的sock_obj对象。这样做的好处是隔离性每个socket对象都有自己的日志函数互不干扰。信息丰富闭包可以访问创建时的上下文如sock_obj从而能记录对端地址(getpeername)等关联信息。可恢复性理论上我们存储了原始方法的引用可以在不需要时恢复尽管实操中较难。2. 类构造器Hook (hook_all_existing_sockets)这是确保“零侵入”的关键。我们直接替换了socket.socket.__init__这个类方法。这意味着从此以后任何代码无论是你的云函数还是它导入的第三方库在创建新的socket.socket对象时都会自动执行我们替换后的logged_init函数。在这个函数里我们先让原始__init__完成socket的初始化然后立即调用_replace_socket_methods来替换这个新生实例的sendall和recv方法。这是一种非常干净、彻底的Hook方式。3. 数据解析与记录策略在logged_sendall和logged_recv中我们不仅记录了原始字节(data.hex())还尝试进行简单的协议解析协议判断通过检查数据开头是否是GET、POST或HTTP/来识别HTTP流量。这对于分析云函数的REST API调用非常有用。头部解析使用split(‘\r\n’)按行分割并用:分割键值对来提取HTTP头部。这是一个简单的解析器对于标准的HTTP请求/响应足够用。数据预览我们同时存储了十六进制原始数据和文本预览截断前200字节既保证了数据的完整性可用于后续深度分析又方便人类阅读。结构化存储所有数据包信息都以字典形式存储在全局列表captured_packets中最后可以轻松导出为JSON方便用其他工具如Jupyter Notebook进行分析。4. 线程安全考虑网络请求常常涉及多线程。我们在_replace_socket_methods函数中使用了threading.Lock来确保在修改_original_methods这个全局字典时不会发生冲突。虽然Python的GIL在一定程度上保证了原子性但在涉及“检查-然后-操作”的场景下显式加锁是更安全的做法。5. 实战在小程序云函数调试场景中的应用理论说再多不如实际跑一遍。假设我们有一个云函数my_cloud_function它内部使用requests库向一个外部API发送请求。我们想抓取这个请求的详细内容。步骤一准备云函数模拟环境通常我们会在本地模拟云函数环境进行调试。创建一个测试脚本test_cloud_function.py# test_cloud_function.py import requests import json import time def main_handler(event, context): 模拟的云函数入口 print(Cloud function started.) # 模拟一个对外部API的请求 api_url https://httpbin.org/post payload { action: user_login, timestamp: int(time.time()), data: {user_id: test_123, nonce: abcxyz} } headers { User-Agent: MyCloudFunction/1.0, Content-Type: application/json, X-Custom-Header: debug-session } try: response requests.post(api_url, jsonpayload, headersheaders, timeout10) response.raise_for_status() result response.json() print(fAPI call succeeded. Status: {response.status_code}) # 模拟处理结果 return { code: 0, message: success, data: result.get(json) } except Exception as e: print(fAPI call failed: {e}) return { code: -1, message: str(e) } if __name__ __main__: # 本地测试 event {} context {} result main_handler(event, context) print(Cloud function result:, json.dumps(result, indent2))步骤二集成抓包器并运行我们修改测试脚本在云函数逻辑执行之前先启动我们的抓包器。# test_cloud_function_with_sniffer.py import sys import os # 将抓包器脚本所在目录加入路径或者直接复制代码过来 sys.path.insert(0, os.path.dirname(__file__)) # 首先导入并启动抓包器 from cloud_function_sniffer import start_sniffing, stop_sniffing, save_captured_data, print_summary print([*] 即将启动云函数并开始网络流量抓包...) start_sniffing() # 然后导入并运行你的云函数代码 # 注意必须在start_sniffing()之后导入以确保requests等库在socket被Hook后才加载 from test_cloud_function import main_handler import json import time event {} context {} print(\n[*] 执行云函数...) result main_handler(event, context) print([*] 云函数执行完毕。\n) # 停止抓包并输出结果 stop_sniffing() print_summary() # 保存抓包数据 timestamp int(time.time()) filename fcloud_function_trace_{timestamp}.json save_captured_data(filename) print(f[*] 详细抓包数据已保存至: {filename}) # 可选打印出捕获到的HTTP请求和响应详情 from cloud_function_sniffer import captured_packets print(\n[*] 详细的HTTP交互记录) for i, pkt in enumerate(captured_packets): if pkt.get(protocol) HTTP: print(f\n--- Packet {i1} [{pkt[type]}] {pkt[time]} ---) if pkt[type] SEND: print(fRequest: {pkt.get(http_request_line)}) headers pkt.get(http_headers, {}) print(Headers:, json.dumps(headers, indent2)) if http_body_preview in pkt: print(fBody (preview): {pkt[http_body_preview][:200]}...) else: # RECV print(fResponse: {pkt.get(http_status_line)}) headers pkt.get(http_headers, {}) print(Headers:, json.dumps(headers, indent2)) if http_body_preview in pkt: print(fBody (preview): {pkt[http_body_preview][:200]}...)运行这个脚本你会看到类似下面的输出[*] 即将启动云函数并开始网络流量抓包... [*] Starting Python socket sniffer... [*] Socket class __init__ method has been hooked. [*] Sniffer is active. All subsequent socket traffic will be logged. [*] 执行云函数... Cloud function started. [2024-05-27 10:30:15.123] [SEND to 34.206.85.169:443] Length: 258 bytes [2024-05-27 10:30:15.456] [RECV from 34.206.85.169:443] Length: 1256 bytes API call succeeded. Status: 200 [*] 云函数执行完毕。 [*] Stopping sniffer. [*] Sniffer stopped. Captured packets: 2 Capture Summary Total packets captured: 2 SEND: 1 RECV: 1 Traffic by peer address: 34.206.85.169:443: 2 packets Recent HTTP Requests: [2024-05-27 10:30:15.123] POST /post HTTP/1.1 [*] Captured data saved to cloud_function_trace_1716784215.json [*] 详细的HTTP交互记录 --- Packet 1 [SEND] 2024-05-27 10:30:15.123 --- Request: POST /post HTTP/1.1 Headers: { Host: httpbin.org, User-Agent: MyCloudFunction/1.0, Content-Type: application/json, X-Custom-Header: debug-session, Content-Length: 86, Accept-Encoding: gzip, deflate, Connection: keep-alive } Body (preview): {action: user_login, timestamp: 1716784215, data: {user_id: test_123, nonce: abcxyz}}... --- Packet 2 [RECV] 2024-05-27 10:30:15.456 --- Response: HTTP/1.1 200 OK Headers: { Date: Mon, 27 May 2024 02:30:15 GMT, Content-Type: application/json, Content-Length: 529, Connection: keep-alive, ... } Body (preview): { args: {}, data: {\action\: \user_login\, \timestamp\: 1716784215, \data\: {\user_id\: \test_123\, \nonce\: \abcxyz\}}, files: {}, form: {}, ... }...成功了我们清晰地看到了云函数发出的POST请求包括完整的请求头、请求体JSON内容以及服务器返回的响应头和响应体。这比云函数控制台里简单的“调用成功”日志要详细得多。6. 高级技巧与避坑指南在实际使用中你可能会遇到一些复杂情况。下面分享几个我踩过坑后总结的经验。6.1 处理HTTPS加密流量上面的代码捕获的是原始的TCP数据。如果云函数请求的是HTTPS接口大部分情况都是你看到data_preview将是乱码或不可读的加密数据。data_raw字段保存的十六进制字符串也是加密后的。要解密HTTPS需要在Hook层面实现TLS/SSL解密这非常复杂几乎需要重写Python的ssl模块。一个更实用的折中方案是记录元数据即使内容加密我们依然可以记录关键的元信息目标IP和端口、数据包大小、发送/接收时间。这有助于了解云函数的通信频率和数据量。结合其他工具如果必须看到明文一个可行的办法是在本地运行一个HTTPS代理如mitmproxy然后修改云函数代码或通过环境变量使其走这个代理。mitmproxy可以完成HTTPS的MITM解密。这时我们的Python抓包器可以Hook到的是云函数与本地代理之间的HTTP流量因为代理解密了从而看到明文。这牺牲了“零侵入性”但换来了可读性。Hook更上层与其Hook最底层的socket不如尝试Hook更上层的库比如requests的Session.send方法或urllib3的请求发送函数。在这些层面请求体可能还未被加密如果库是先组包再加密的话。但这需要对每个你想监控的库进行单独适配。对于小程序云函数逆向分析很多时候分析加密前的请求体构造逻辑参数如何拼接、签名如何计算比看加密后的流量更重要。因此即使看不到HTTPS明文这个抓包器在分析非加密流量或辅助理解通信模式上仍有很大价值。6.2 多线程与异步IO环境云函数环境或现代Python应用很可能使用asyncio、aiohttp或多线程。我们的Hook机制需要适应。多线程我们的代码使用了线程锁基本是安全的。但要确保start_sniffing()在所有工作线程启动之前被调用。如果工作线程在Hook启动前就创建了socket连接这些连接可能不会被捕获。异步IO (asyncio)这是一个挑战。asyncio使用自己的事件循环和传输层可能不直接使用标准的socket.socket对象而是使用更底层的loop.sock_*方法或asyncio.Transport。要Hookasyncio的流量你需要针对asyncio的协议层进行Hook例如替换asyncio.loop.create_connection返回的transport.write方法。这比同步socket Hook更复杂。应对策略如果你的云函数使用aiohttp可以尝试直接Hookaiohttp.ClientSession._request方法因为在这个层面请求信息URL、headers、body还是明文的。这需要你对目标库的源码有一定了解。6.3 性能影响与稳定性sys.settrace和函数替换特别是对每个socket操作都进行字符串解码和JSON序列化会带来性能开销。对于高频、低延迟的云函数这可能不适用。生产环境禁用这个工具绝对不要用于生产环境。它仅用于本地开发、调试和安全测试。选择性记录可以在_make_logged_sendall/recv函数开头增加过滤逻辑。例如只记录发送到特定域名或IP的流量或者当数据包大小超过一定阈值时才进行详细解析以减少开销。def logged_sendall(data, *args, **kwargs): peer_addr sock_obj.getpeername()[0] if hasattr(sock_obj, getpeername) else None # 只记录发送到特定IP段的流量 if peer_addr and not peer_addr.startswith(192.168.): # ... 进行记录操作 # 无论如何都调用原始方法 return original_sendall(data, *args, **kwargs)避免无限增长captured_packets列表会持续增长。在长时间运行的调试中可能需要定期清理或写入文件防止内存耗尽。6.4 与其他抓包工具的对比工具/方法原理优点缺点适用场景本Python抓包器Hook Python进程内的socket对象方法零侵入对目标代码纯Python可获取进程内精确流量无法解密HTTPS对异步IO支持弱有性能开销Python进程的本地调试分析库的内部网络行为Fiddler/Charles系统代理MITM功能强大支持HTTPS解密可视化好跨应用需要配置代理可能被应用检测或绕过对无GUI服务器不友好常规HTTP/HTTPS流量分析移动端抓包Wireshark网卡抓包解析协议栈最底层最全面支持所有协议信息过载需要专业知识过滤无法直接关联到特定进程/代码行网络故障排查协议学习分析非HTTP流量mitmproxy代理 可编程拦截脚本化能力强支持修改请求/响应适合自动化测试需要配置代理是独立进程自动化测试流量重放API调试选择哪种工具取决于你的具体需求。对于“理解小程序云函数这个特定Python进程到底发了什么网络请求”这个场景本Python抓包器提供了一种轻量级、直指核心的解决方案。7. 扩展思路从抓包到“云函数Hook”这个项目的意义不止于抓包。它演示了一种在Python运行时进行动态代码注入Hook的基本模式。基于这个模式我们可以做更多事情修改请求/响应在logged_sendall中你不仅可以记录data还可以修改它然后再传给original_sendall。同理在logged_recv中可以修改接收到的数据再返回。这就实现了简单的请求/响应篡改可用于测试服务器对异常数据的处理。模拟网络异常在Hook函数中随机抛出异常、延迟返回、或返回伪造数据用来测试云函数的容错性和重试机制。性能分析记录每个请求的耗时从sendall到recv完成统计不同API的响应时间找出性能瓶颈。依赖分析通过记录的对端地址和端口可以自动分析出云函数运行时依赖了哪些外部服务数据库、缓存、第三方API等。要实现更稳定、更强大的Hook可以考虑使用专门用于Python代码注入的库如pyringe已停止维护或pyrasite它们提供了更优雅的向运行中Python进程注入代码的方式。但对于快速验证和轻量级调试我们手写的这个纯Python版本已经足够强大和灵活。最后再次强调这类技术应仅用于合法的安全研究、调试和开发工作。在使用任何抓包或Hook工具时请务必遵守相关法律法规和服务条款。希望这个项目能为你打开一扇窗让你能更清晰地看到云函数内部的世界。 Python内存Hook技术实现小程序云函数网络流量抓包与逆向分析
1. 项目缘起当小程序云函数成为“黑盒”最近在折腾一个微信小程序的数据分析项目客户的小程序大量使用了云函数来处理核心业务逻辑比如用户签到、积分兑换、数据上报等。问题来了这些云函数的代码和逻辑都跑在云端我们作为开发者或者安全研究员想了解它具体发送了什么数据、接收了什么响应简直像隔着一堵墙。官方调试工具提供的日志有限很多时候只能看到“调用成功”或“失败”具体的请求体、响应头、甚至是第三方接口的交互细节完全是个黑盒。这让我想起了早些年做Web端和移动端逆向分析时抓包是基本功。Fiddler、Charles、Wireshark都是老朋友。但面对小程序云函数这些传统工具似乎有点“鞭长莫及”。云函数的执行环境在腾讯云服务器上不在本地你没法直接给那台远程服务器装个代理。网上搜了一圈发现讨论这个的并不多大多停留在理论或者使用一些商业化的、需要特定环境的抓包方案上。于是一个想法冒了出来能不能用最纯粹的Python写一个轻量级的工具模拟出类似“中间人”的效果把云函数发起的网络请求给“钓”出来这个工具的目标很明确无需复杂的环境配置不依赖特定操作系统或硬件比如AX210网卡纯粹在代码层面实现对一个运行中Python进程模拟云函数环境的网络流量拦截和解析。这就是今天要分享的“纯Python实现小程序云函数抓包”项目的由来。它本质上是一种针对特定运行环境Python进程的流量Hook技术对于分析小程序云函数、某些Python SDK的内部网络行为或者进行安全的黑盒测试会非常有用。2. 核心思路从“代理”到“内存Hook”的转变一开始我最自然的想法是走代理模式。既然不能直接抓远程服务器的包那能不能让云函数的代码把所有网络请求都发往我本地启动的一个HTTP/HTTPS代理服务器呢这样我就能在代理服务器上看到所有流量。理论上可行但实操问题很多代码侵入性你需要修改云函数的源代码显式地设置http_proxy或https_proxy环境变量或者为requests、aiohttp等库配置代理。对于已编译的、不可修改的第三方包或者代码逻辑复杂的项目这很困难。HTTPS解密如果云函数请求的是HTTPS接口代理服务器需要安装自签名CA证书并进行MITM中间人解密。这在小程序模拟环境或某些受限制的服务器环境里配置起来非常麻烦且可能触发证书校验错误。环境依赖你需要一个稳定的、能处理并发请求的代理服务器比如用mitmproxy整个工具链变重了。那么有没有更底层、更通用的方法答案是有的那就是内存函数Hook。我们不必关心云函数代码里用的是requests.get还是urllib.request.urlopen抑或是aiohttp.ClientSession。在Python中这些库最终都会调用Python标准库socket模块的底层函数来创建连接和收发数据。如果我们能在运行时动态地替换掉这些关键的socket函数比如socket.sendall和socket.recv那么所有经由这些函数进出网络的数据都会先流经我们编写的“钩子”函数。在这个“钩子”里我们就可以记录、修改、分析这些原始的网络数据包。这个思路的优势非常明显零侵入性无需修改目标云函数的一行源代码。只需要在目标Python进程启动时或通过调试器附加后注入我们的Hook代码即可。通用性强无论云函数内部使用何种HTTP库requests,httpx,urllib,aiohttp只要它们基于Python的socket就会被捕获。信息完整获取的是最原始的TCP Socket层数据包含了完整的TCP/IP载荷我们可以从中解析出HTTP/HTTPS的明文如果是HTTP或加密流量如果是HTTPS需要后续解密但至少能看见流量特征。当然这种方法也有挑战主要是对Python运行时和线程安全的要求较高。接下来我们就一步步拆解如何实现这个“内存Hook”抓包器。3. 技术选型与关键库sys.settrace与socket劫持要实现运行时函数替换我们需要一个“入口点”来执行我们的Hook代码。这里有两个主流方向使用调试器接口ptrace/sys.settracesys.settrace是Python标准库提供的调试钩子它可以设置一个跟踪函数在代码执行到每一行时被调用。虽然通常用于调试和性能分析但我们也可以利用它在函数被调用时执行我们的代码。另一种更底层的是使用ptraceLinux或类似的系统调用但这需要C扩展且跨平台兼容性差。使用代码注入injectpy或PyInstaller钩子通过LD_PRELOADLinux或DLL注入Windows等方式将我们的代码注入到目标Python解释器进程中。这更接近传统逆向工程但实现复杂对环境依赖强。为了追求“纯Python”和相对简单我们选择方案1并聚焦于sys.settrace。但请注意sys.settrace是单线程的并且对性能有影响它更适合用于分析、调试场景而非生产环境的高频抓包。对于我们的云函数抓包通常是短期、低频的调试任务它是完全可行的。核心库就是Python自带的sys和socket。我们的工作流程将是编写一个Hook函数用于替换socket.socket类的sendall和recv方法。编写一个Trace函数通过sys.settrace设置监控socket.socket对象的创建。一旦监控到socket.socket实例被创建立即将其sendall和recv方法替换为我们自定义的、带日志记录功能的版本。这样当云函数的代码调用socket.sendall(data)发送数据时数据会先传到我们的函数我们记录下它然后再调用原始的socket.sendall发送出去。接收过程同理。注意直接Hooksocket模块的函数如socket.send是全局的会影响进程内所有socket操作可能干扰其他正常功能。我们选择Hooksocket.socket实例的方法针对性更强可以通过判断目标地址如是否是云函数调用的特定API域名来决定是否记录减少噪音。4. 完整代码实现与逐行解析下面就是完整的抓包工具核心代码。我们将它保存为一个独立的Python模块比如叫cloud_function_sniffer.py。使用时只需要在运行你的云函数脚本之前先导入并启动这个抓包器即可。#!/usr/bin/env python3 # -*- coding: utf-8 -*- 纯Python实现的小程序云函数网络流量抓包工具 原理通过sys.settrace Hook socket.socket对象的sendall和recv方法 import sys import socket import threading import json import re from datetime import datetime from urllib.parse import urlparse import typing as t # 全局锁防止多线程下替换方法时出现竞争条件 _hook_lock threading.Lock() # 存储原始方法的字典key为socket对象idvalue为原始方法元组(sendall, recv) _original_methods {} # 抓包数据存储列表 captured_packets [] def _make_logged_sendall(original_sendall, sock_obj): 创建一个带日志记录的sendall函数 :param original_sendall: 原始的sendall方法 :param sock_obj: 对应的socket对象 :return: 新的sendall函数 def logged_sendall(data: bytes, *args, **kwargs) - t.Any: # 记录发送的数据 local_time datetime.now().strftime(%Y-%m-%d %H:%M:%S.%f)[:-3] peer_name sock_obj.getpeername() if hasattr(sock_obj, getpeername) else (unknown, 0) packet_info { type: SEND, time: local_time, socket_id: id(sock_obj), peer_address: f{peer_name[0]}:{peer_name[1]}, data_raw: data.hex(), # 原始字节的十六进制表示 data_preview: repr(data[:200]) if len(data) 200 else repr(data), # 预览前200字节 data_length: len(data) } # 尝试解析为HTTP请求简单判断 try: decoded data.decode(utf-8, errorsignore) if decoded.startswith((GET , POST , PUT , DELETE , HEAD , OPTIONS )): packet_info[protocol] HTTP # 提取首行和Headers lines decoded.split(\r\n) packet_info[http_request_line] lines[0] headers {} for line in lines[1:]: if line : break if : in line: k, v line.split(: , 1) headers[k] v packet_info[http_headers] headers # 尝试找Body body_start decoded.find(\r\n\r\n) if body_start ! -1: body decoded[body_start4:] if body: packet_info[http_body_preview] body[:500] except UnicodeDecodeError: pass # 非文本数据保持原始十六进制 captured_packets.append(packet_info) print(f[{local_time}] [SEND to {peer_name}] Length: {len(data)} bytes) # 调用原始方法发送数据 return original_sendall(data, *args, **kwargs) return logged_sendall def _make_logged_recv(original_recv, sock_obj): 创建一个带日志记录的recv函数 :param original_recv: 原始的recv方法 :param sock_obj: 对应的socket对象 :return: 新的recv函数 def logged_recv(bufsize: int, *args, **kwargs) - bytes: # 先调用原始方法接收数据 data original_recv(bufsize, *args, **kwargs) if data: # 只记录非空数据 local_time datetime.now().strftime(%Y-%m-%d %H:%M:%S.%f)[:-3] peer_name sock_obj.getpeername() if hasattr(sock_obj, getpeername) else (unknown, 0) packet_info { type: RECV, time: local_time, socket_id: id(sock_obj), peer_address: f{peer_name[0]}:{peer_name[1]}, data_raw: data.hex(), data_preview: repr(data[:200]) if len(data) 200 else repr(data), data_length: len(data) } # 尝试解析为HTTP响应 try: decoded data.decode(utf-8, errorsignore) if decoded.startswith(HTTP/): packet_info[protocol] HTTP lines decoded.split(\r\n) packet_info[http_status_line] lines[0] headers {} for line in lines[1:]: if line : break if : in line: k, v line.split(: , 1) headers[k] v packet_info[http_headers] headers body_start decoded.find(\r\n\r\n) if body_start ! -1: body decoded[body_start4:] if body: packet_info[http_body_preview] body[:500] except UnicodeDecodeError: pass captured_packets.append(packet_info) print(f[{local_time}] [RECV from {peer_name}] Length: {len(data)} bytes) return data return logged_recv def _hook_socket_creation(frame, event, arg): sys.settrace 设置的跟踪函数用于在socket对象创建时Hook其方法。 这个函数会在代码执行时被频繁调用因此逻辑要尽量轻量。 if event call: # 当有函数被调用时 # 检查调用的函数是否是 socket.socket.__init__ 或 socket.socket (构造函数) # 通过帧的代码对象和名字来判断 code_obj frame.f_code # 检查是否在socket模块中并且函数名与socket相关 # 注意这里是一个简化判断实际可能需要更精确的匹配 if socket in code_obj.co_filename and socket in code_obj.co_name: # 等待socket对象初始化完成再Hook比较困难这里我们采用一个折中方案 # 不在此处直接Hook而是通过监控socket.socket方法的返回来进行。 # 更可靠的方法是Hook socket.socket 这个类本身的 __init__ 或 __new__。 # 为了简化我们在另一个地方见下面的hook_all_existing_sockets进行批量Hook。 pass return _hook_socket_creation # 返回自身以继续跟踪 def _replace_socket_methods(sock): 替换单个socket对象的sendall和recv方法。 使用锁保证线程安全。 with _hook_lock: sock_id id(sock) if sock_id not in _original_methods: # 保存原始方法 orig_sendall sock.sendall orig_recv sock.recv _original_methods[sock_id] (orig_sendall, orig_recv) # 替换为我们的带日志版本 sock.sendall _make_logged_sendall(orig_sendall, sock) sock.recv _make_logged_recv(orig_recv, sock) # 注意实际应用中可能还需要替换 send, recv_into, sendto 等方法 # 这里只覆盖最常用的 sendall 和 recv def hook_all_existing_sockets(): 遍历所有已存在的socket对象并Hook它们。 由于通过sys.settrace精确捕获新建socket对象比较复杂 我们可以采用一个更直接但有点“暴力”的方法定期扫描所有对象。 但更优雅的方式是直接替换 socket.socket 类的 __init__ 方法。 这里我们采用类方法替换的方案。 original_init socket.socket.__init__ def logged_init(self, *args, **kwargs): # 先调用原始的 __init__ result original_init(self, *args, **kwargs) # 然后替换这个新实例的方法 _replace_socket_methods(self) return result # 替换类的 __init__ 方法 socket.socket.__init__ logged_init # 同时Hook当前已经存在的socket对象如果有的话 # 遍历 gc.getobjects() 开销大这里仅作为示例实际可省略 # import gc # for obj in gc.getobjects(): # if isinstance(obj, socket.socket): # _replace_socket_methods(obj) print([*] Socket class __init__ method has been hooked.) def start_sniffing(): 启动抓包。 1. 替换socket.socket类的__init__方法确保之后新建的socket都被Hook。 2. 设置sys.settrace可选用于更复杂的场景本例中类替换已足够。 print([*] Starting Python socket sniffer...) hook_all_existing_sockets() # 对于已经运行的、在hook之前创建的socket上述方法可能无法捕获。 # 如果需要捕获所有可能需要更侵入性的方式如调试器注入。 # 本工具主要针对在hook启动后创建的socket连接。 print([*] Sniffer is active. All subsequent socket traffic will be logged.) print([*] Note: Traffic created before this point may not be captured.) def stop_sniffing(): 停止抓包并恢复原始方法可选。 注意恢复所有被修改的方法可能很困难因为原始引用可能已丢失。 通常用于短期调试不需要恢复。 print([*] Stopping sniffer.) # 尝试恢复这里仅作演示实际复杂环境可能无法完全恢复 with _hook_lock: for sock_id, (orig_sendall, orig_recv) in _original_methods.items(): # 我们需要通过sock_id找到对象这通常做不到。 # 所以此函数主要是一个示意。 pass # 更简单的方式是不恢复直接结束进程或忽略。 print([*] Sniffer stopped. Captured packets:, len(captured_packets)) def save_captured_data(filename: str captured_traffic.json): 将抓取到的数据保存为JSON文件 with open(filename, w, encodingutf-8) as f: json.dump(captured_packets, f, indent2, ensure_asciiFalse) print(f[*] Captured data saved to {filename}) def print_summary(): 打印抓包摘要 print(f\n{*60}) print(fCapture Summary) print(f{*60}) print(fTotal packets captured: {len(captured_packets)}) send_count sum(1 for p in captured_packets if p[type] SEND) recv_count sum(1 for p in captured_packets if p[type] RECV) print(f SEND: {send_count}) print(f RECV: {recv_count}) # 按目标地址统计 addr_stats {} for p in captured_packets: addr p.get(peer_address, unknown) addr_stats[addr] addr_stats.get(addr, 0) 1 if addr_stats: print(f\nTraffic by peer address:) for addr, count in sorted(addr_stats.items(), keylambda x: x[1], reverseTrue)[:10]: # 显示前10 print(f {addr}: {count} packets) # 显示最近的几个HTTP请求如果有 http_requests [p for p in captured_packets if p.get(type) SEND and p.get(protocol) HTTP] if http_requests: print(f\nRecent HTTP Requests:) for req in http_requests[-3:]: # 显示最后3个 print(f [{req[time]}] {req.get(http_request_line, N/A)}) print(f{*60}) # 提供一个便捷的使用方式作为上下文管理器 class SocketSnifferContext: def __enter__(self): start_sniffing() return self def __exit__(self, exc_type, exc_val, exc_tb): stop_sniffing() print_summary() # 自动保存可以根据需要开启 # save_captured_data() # 如果直接运行此脚本则进入一个简单的测试模式 if __name__ __main__: # 示例测试抓包器自身 print(Testing the sniffer on itself...) with SocketSnifferContext(): import urllib.request # 发起一个简单的HTTP请求观察是否被抓取 try: response urllib.request.urlopen(http://httpbin.org/get, timeout5) print(fTest request status: {response.status}) response.read() except Exception as e: print(fTest request failed (may be expected): {e}) print_summary() if captured_packets: save_captured_data(test_capture.json) print(\nFirst captured packet (SEND):) print(json.dumps(captured_packets[0], indent2))4.1 代码核心逻辑拆解这段代码虽然不长但有几个关键设计点值得深入探讨1. 函数工厂模式 (_make_logged_sendall和_make_logged_recv)这是实现Hook的核心。我们没有直接修改socket.socket类的原生方法而是为每一个socket对象实例动态创建了一个新的方法。这个新方法是一个闭包它捕获了原始的original_sendall方法和当前的sock_obj对象。这样做的好处是隔离性每个socket对象都有自己的日志函数互不干扰。信息丰富闭包可以访问创建时的上下文如sock_obj从而能记录对端地址(getpeername)等关联信息。可恢复性理论上我们存储了原始方法的引用可以在不需要时恢复尽管实操中较难。2. 类构造器Hook (hook_all_existing_sockets)这是确保“零侵入”的关键。我们直接替换了socket.socket.__init__这个类方法。这意味着从此以后任何代码无论是你的云函数还是它导入的第三方库在创建新的socket.socket对象时都会自动执行我们替换后的logged_init函数。在这个函数里我们先让原始__init__完成socket的初始化然后立即调用_replace_socket_methods来替换这个新生实例的sendall和recv方法。这是一种非常干净、彻底的Hook方式。3. 数据解析与记录策略在logged_sendall和logged_recv中我们不仅记录了原始字节(data.hex())还尝试进行简单的协议解析协议判断通过检查数据开头是否是GET、POST或HTTP/来识别HTTP流量。这对于分析云函数的REST API调用非常有用。头部解析使用split(‘\r\n’)按行分割并用:分割键值对来提取HTTP头部。这是一个简单的解析器对于标准的HTTP请求/响应足够用。数据预览我们同时存储了十六进制原始数据和文本预览截断前200字节既保证了数据的完整性可用于后续深度分析又方便人类阅读。结构化存储所有数据包信息都以字典形式存储在全局列表captured_packets中最后可以轻松导出为JSON方便用其他工具如Jupyter Notebook进行分析。4. 线程安全考虑网络请求常常涉及多线程。我们在_replace_socket_methods函数中使用了threading.Lock来确保在修改_original_methods这个全局字典时不会发生冲突。虽然Python的GIL在一定程度上保证了原子性但在涉及“检查-然后-操作”的场景下显式加锁是更安全的做法。5. 实战在小程序云函数调试场景中的应用理论说再多不如实际跑一遍。假设我们有一个云函数my_cloud_function它内部使用requests库向一个外部API发送请求。我们想抓取这个请求的详细内容。步骤一准备云函数模拟环境通常我们会在本地模拟云函数环境进行调试。创建一个测试脚本test_cloud_function.py# test_cloud_function.py import requests import json import time def main_handler(event, context): 模拟的云函数入口 print(Cloud function started.) # 模拟一个对外部API的请求 api_url https://httpbin.org/post payload { action: user_login, timestamp: int(time.time()), data: {user_id: test_123, nonce: abcxyz} } headers { User-Agent: MyCloudFunction/1.0, Content-Type: application/json, X-Custom-Header: debug-session } try: response requests.post(api_url, jsonpayload, headersheaders, timeout10) response.raise_for_status() result response.json() print(fAPI call succeeded. Status: {response.status_code}) # 模拟处理结果 return { code: 0, message: success, data: result.get(json) } except Exception as e: print(fAPI call failed: {e}) return { code: -1, message: str(e) } if __name__ __main__: # 本地测试 event {} context {} result main_handler(event, context) print(Cloud function result:, json.dumps(result, indent2))步骤二集成抓包器并运行我们修改测试脚本在云函数逻辑执行之前先启动我们的抓包器。# test_cloud_function_with_sniffer.py import sys import os # 将抓包器脚本所在目录加入路径或者直接复制代码过来 sys.path.insert(0, os.path.dirname(__file__)) # 首先导入并启动抓包器 from cloud_function_sniffer import start_sniffing, stop_sniffing, save_captured_data, print_summary print([*] 即将启动云函数并开始网络流量抓包...) start_sniffing() # 然后导入并运行你的云函数代码 # 注意必须在start_sniffing()之后导入以确保requests等库在socket被Hook后才加载 from test_cloud_function import main_handler import json import time event {} context {} print(\n[*] 执行云函数...) result main_handler(event, context) print([*] 云函数执行完毕。\n) # 停止抓包并输出结果 stop_sniffing() print_summary() # 保存抓包数据 timestamp int(time.time()) filename fcloud_function_trace_{timestamp}.json save_captured_data(filename) print(f[*] 详细抓包数据已保存至: {filename}) # 可选打印出捕获到的HTTP请求和响应详情 from cloud_function_sniffer import captured_packets print(\n[*] 详细的HTTP交互记录) for i, pkt in enumerate(captured_packets): if pkt.get(protocol) HTTP: print(f\n--- Packet {i1} [{pkt[type]}] {pkt[time]} ---) if pkt[type] SEND: print(fRequest: {pkt.get(http_request_line)}) headers pkt.get(http_headers, {}) print(Headers:, json.dumps(headers, indent2)) if http_body_preview in pkt: print(fBody (preview): {pkt[http_body_preview][:200]}...) else: # RECV print(fResponse: {pkt.get(http_status_line)}) headers pkt.get(http_headers, {}) print(Headers:, json.dumps(headers, indent2)) if http_body_preview in pkt: print(fBody (preview): {pkt[http_body_preview][:200]}...)运行这个脚本你会看到类似下面的输出[*] 即将启动云函数并开始网络流量抓包... [*] Starting Python socket sniffer... [*] Socket class __init__ method has been hooked. [*] Sniffer is active. All subsequent socket traffic will be logged. [*] 执行云函数... Cloud function started. [2024-05-27 10:30:15.123] [SEND to 34.206.85.169:443] Length: 258 bytes [2024-05-27 10:30:15.456] [RECV from 34.206.85.169:443] Length: 1256 bytes API call succeeded. Status: 200 [*] 云函数执行完毕。 [*] Stopping sniffer. [*] Sniffer stopped. Captured packets: 2 Capture Summary Total packets captured: 2 SEND: 1 RECV: 1 Traffic by peer address: 34.206.85.169:443: 2 packets Recent HTTP Requests: [2024-05-27 10:30:15.123] POST /post HTTP/1.1 [*] Captured data saved to cloud_function_trace_1716784215.json [*] 详细的HTTP交互记录 --- Packet 1 [SEND] 2024-05-27 10:30:15.123 --- Request: POST /post HTTP/1.1 Headers: { Host: httpbin.org, User-Agent: MyCloudFunction/1.0, Content-Type: application/json, X-Custom-Header: debug-session, Content-Length: 86, Accept-Encoding: gzip, deflate, Connection: keep-alive } Body (preview): {action: user_login, timestamp: 1716784215, data: {user_id: test_123, nonce: abcxyz}}... --- Packet 2 [RECV] 2024-05-27 10:30:15.456 --- Response: HTTP/1.1 200 OK Headers: { Date: Mon, 27 May 2024 02:30:15 GMT, Content-Type: application/json, Content-Length: 529, Connection: keep-alive, ... } Body (preview): { args: {}, data: {\action\: \user_login\, \timestamp\: 1716784215, \data\: {\user_id\: \test_123\, \nonce\: \abcxyz\}}, files: {}, form: {}, ... }...成功了我们清晰地看到了云函数发出的POST请求包括完整的请求头、请求体JSON内容以及服务器返回的响应头和响应体。这比云函数控制台里简单的“调用成功”日志要详细得多。6. 高级技巧与避坑指南在实际使用中你可能会遇到一些复杂情况。下面分享几个我踩过坑后总结的经验。6.1 处理HTTPS加密流量上面的代码捕获的是原始的TCP数据。如果云函数请求的是HTTPS接口大部分情况都是你看到data_preview将是乱码或不可读的加密数据。data_raw字段保存的十六进制字符串也是加密后的。要解密HTTPS需要在Hook层面实现TLS/SSL解密这非常复杂几乎需要重写Python的ssl模块。一个更实用的折中方案是记录元数据即使内容加密我们依然可以记录关键的元信息目标IP和端口、数据包大小、发送/接收时间。这有助于了解云函数的通信频率和数据量。结合其他工具如果必须看到明文一个可行的办法是在本地运行一个HTTPS代理如mitmproxy然后修改云函数代码或通过环境变量使其走这个代理。mitmproxy可以完成HTTPS的MITM解密。这时我们的Python抓包器可以Hook到的是云函数与本地代理之间的HTTP流量因为代理解密了从而看到明文。这牺牲了“零侵入性”但换来了可读性。Hook更上层与其Hook最底层的socket不如尝试Hook更上层的库比如requests的Session.send方法或urllib3的请求发送函数。在这些层面请求体可能还未被加密如果库是先组包再加密的话。但这需要对每个你想监控的库进行单独适配。对于小程序云函数逆向分析很多时候分析加密前的请求体构造逻辑参数如何拼接、签名如何计算比看加密后的流量更重要。因此即使看不到HTTPS明文这个抓包器在分析非加密流量或辅助理解通信模式上仍有很大价值。6.2 多线程与异步IO环境云函数环境或现代Python应用很可能使用asyncio、aiohttp或多线程。我们的Hook机制需要适应。多线程我们的代码使用了线程锁基本是安全的。但要确保start_sniffing()在所有工作线程启动之前被调用。如果工作线程在Hook启动前就创建了socket连接这些连接可能不会被捕获。异步IO (asyncio)这是一个挑战。asyncio使用自己的事件循环和传输层可能不直接使用标准的socket.socket对象而是使用更底层的loop.sock_*方法或asyncio.Transport。要Hookasyncio的流量你需要针对asyncio的协议层进行Hook例如替换asyncio.loop.create_connection返回的transport.write方法。这比同步socket Hook更复杂。应对策略如果你的云函数使用aiohttp可以尝试直接Hookaiohttp.ClientSession._request方法因为在这个层面请求信息URL、headers、body还是明文的。这需要你对目标库的源码有一定了解。6.3 性能影响与稳定性sys.settrace和函数替换特别是对每个socket操作都进行字符串解码和JSON序列化会带来性能开销。对于高频、低延迟的云函数这可能不适用。生产环境禁用这个工具绝对不要用于生产环境。它仅用于本地开发、调试和安全测试。选择性记录可以在_make_logged_sendall/recv函数开头增加过滤逻辑。例如只记录发送到特定域名或IP的流量或者当数据包大小超过一定阈值时才进行详细解析以减少开销。def logged_sendall(data, *args, **kwargs): peer_addr sock_obj.getpeername()[0] if hasattr(sock_obj, getpeername) else None # 只记录发送到特定IP段的流量 if peer_addr and not peer_addr.startswith(192.168.): # ... 进行记录操作 # 无论如何都调用原始方法 return original_sendall(data, *args, **kwargs)避免无限增长captured_packets列表会持续增长。在长时间运行的调试中可能需要定期清理或写入文件防止内存耗尽。6.4 与其他抓包工具的对比工具/方法原理优点缺点适用场景本Python抓包器Hook Python进程内的socket对象方法零侵入对目标代码纯Python可获取进程内精确流量无法解密HTTPS对异步IO支持弱有性能开销Python进程的本地调试分析库的内部网络行为Fiddler/Charles系统代理MITM功能强大支持HTTPS解密可视化好跨应用需要配置代理可能被应用检测或绕过对无GUI服务器不友好常规HTTP/HTTPS流量分析移动端抓包Wireshark网卡抓包解析协议栈最底层最全面支持所有协议信息过载需要专业知识过滤无法直接关联到特定进程/代码行网络故障排查协议学习分析非HTTP流量mitmproxy代理 可编程拦截脚本化能力强支持修改请求/响应适合自动化测试需要配置代理是独立进程自动化测试流量重放API调试选择哪种工具取决于你的具体需求。对于“理解小程序云函数这个特定Python进程到底发了什么网络请求”这个场景本Python抓包器提供了一种轻量级、直指核心的解决方案。7. 扩展思路从抓包到“云函数Hook”这个项目的意义不止于抓包。它演示了一种在Python运行时进行动态代码注入Hook的基本模式。基于这个模式我们可以做更多事情修改请求/响应在logged_sendall中你不仅可以记录data还可以修改它然后再传给original_sendall。同理在logged_recv中可以修改接收到的数据再返回。这就实现了简单的请求/响应篡改可用于测试服务器对异常数据的处理。模拟网络异常在Hook函数中随机抛出异常、延迟返回、或返回伪造数据用来测试云函数的容错性和重试机制。性能分析记录每个请求的耗时从sendall到recv完成统计不同API的响应时间找出性能瓶颈。依赖分析通过记录的对端地址和端口可以自动分析出云函数运行时依赖了哪些外部服务数据库、缓存、第三方API等。要实现更稳定、更强大的Hook可以考虑使用专门用于Python代码注入的库如pyringe已停止维护或pyrasite它们提供了更优雅的向运行中Python进程注入代码的方式。但对于快速验证和轻量级调试我们手写的这个纯Python版本已经足够强大和灵活。最后再次强调这类技术应仅用于合法的安全研究、调试和开发工作。在使用任何抓包或Hook工具时请务必遵守相关法律法规和服务条款。希望这个项目能为你打开一扇窗让你能更清晰地看到云函数内部的世界。