本实例实现需求
在游戏SDK测试中，经常需要测试游戏中SDK的埋点日志是否接入正确。本实例通过抓包（客户端http/https 请求）来判定埋点日志是是否接入正确。
实现细节：使用django项目，后端采用python mitmdump 扩展脚本“log_handler.py”实时抓取与过滤4399SDK 客户端日志，将数据处理成约定需要的格式，保存和发布到redis中。
前端使用websocket连接，websocket服务端通过redis订阅对应游戏频道信息，实时输出游戏的客户端日志到web页面中。
开发环境
win7，python3,
安装redis_server
安装python redis
python3 -m pip install redis
安装python mitmproxy
python3 -m pip install mitmproxy
代码实现
一、客户端日志抓包处理脚本 log_handler.py:
#!/usr/bin/env python # -*- coding: utf-8 -*- import urllib import json import logging import redis import datetime from mitmproxy import http # 定义联运日志类型，根据抓包分析可知，类型可通过接口名获得 udpdcs_action_from_path = { 'init_info': '初始化日志', 'activity_open': '打开游戏日志', 'activity_before_login': '登录界面前日志', 'user_login': '登录日志', 'enter_game': '进入游戏日志', 'user_server_login': '选服日志', 'user_create_role': '创角日志', 'user_online': '在线日志', 'role_level': '等级日志', } # redis连接池类，返回一个redis链接 class RedisPool: def __init__(self, host="127.0.0.1", port=6379, db=0): self.host = host self.port = port self.db = db def redis_connect(self): pool = redis.ConnectionPool(host=self.host, port=self.port, db=self.db) return redis.StrictRedis(connection_pool=pool) pool = RedisPool("127.0.0.1", 6379, 1) r = pool.redis_connect() def response(flow: http.HTTPFlow): # ly 日志处理 game_id, dict_msg = ly_log_filter(flow) # 日志保存与发布 if game_id and dict_msg: publish_log(game_id, dict_msg) # 联运日志处理 def ly_log_filter(flow): # 根据域名，过滤大陆联运日志 """ 日志请求示例：GET http://udpdcs.4399sy.com/activity_open.php?time=1515481517&flag=cb8001e31777347ba4e2608620e45091&data={"eventId":"0","ip":"0","did":"863726035876487","appVersion":"1.2.3","sdkVersion":"3.1.4.0","platformId":"1","gameId":"1499130088511390","areaId":"0","serverId":"0","os":"android","osVersion":"6.0","device":"M5","deviceType":"android","screen":"1280*720","mno":"","nm":"WIFI","eventTime":"0","channel":"4399","channelOld":"4399","channelSy":"270","sim":"0","kts":"f409b38a02f14aafd1063d6bd30fa636","pkgName":"com.sy4399.xxtjd"}""" host = flow.request.host method = flow.request.method url = urllib.parse.unquote(flow.request.url) dict_msg = None data_send = None game_id = None if host in ["udpdcs.4399sy.com"]: plat = "大陆联运" status_code = flow.response.status_code ret = flow.response.content.decode('utf-8') try: ret = json.loads(ret) except Exception as e: print(e) if method == "GET": # 从path中获取操作类型 path = flow.request.path_components action_type = path[-1].rstrip(".php") action_name = udpdcs_action_from_path.get(action_type) if action_name: # 从URL参数data中获取主要sdk请求数据 querystring = flow.request._get_query() for eachp in querystring: if eachp[0] == "data": data = eachp[1] try: # 将关键的请求参数字符串转为字典，便于数据操作 """参数示例：{"eventId":"0","ip":"0","did":"863726035876487","appVersion":"1.2.3","sdkVersion":"3.1.4.0","platformId":"1","gameId":"1499130088511390","areaId":"0","serverId":"0","os":"android","osVersion":"6.0","device":"M5","deviceType":"android","screen":"1280*720","mno":"","nm":"WIFI","eventTime":"0","channel":"4399","channelOld":"4399","channelSy":"270","sim":"0","kts":"f409b38a02f14aafd1063d6bd30fa636","pkgName":"com.sy4399.xxtjd"}""" data_send = json.loads(data) game_id = data_send.get('gameId') except Exception as e: logging.error(e) dict_msg = { "plat": plat, "host": host, "method": method, "url": url, "action_type": action_type, "action_name": action_name, "data": data_send, "action_time": datetime.datetime.now().strftime( '%Y-%m-%d %H:%M:%S.%f'), "status_code": status_code, "response": ret } else: print("action_type=%s,操作类型为定义" % action_type) else: body = flow.request.content print("使用了POST方式:%s" % url) print("POST DATA:%s" % body) print("*" * 200) return game_id, dict_msg # 发布日志 def publish_log(game_id, dict_msg): if game_id: print("game_id:%s" % game_id) print("dict_msg:", dict_msg) if dict_msg: # 发布到redis频道game_id r.publish(game_id, json.dumps(dict_msg)) # 保存到redis列表中，数据持久化 key = game_id + "_" + str( datetime.datetime.now().strftime("%Y%m%d")) r.lpush(key, json.dumps(dict_msg))
- 启动抓包脚本
在cmd中输入命令
mitmdump -s log_handler.py ~u abc.com
正确启动后如下
E:workspacesdk_monitordemo>mitmdump -s log_handler.py ~u abc.com
Loading script: log_handler.py
Proxy server listening at http://0.0.0.0:8080
- 手机连接代理
手机连上与电脑相同局域网wifi，并设置代理。如电脑端ip为192.168.1.104，则设置代理为 192.168.1.104:8080 ,端口可以在mitmdump中添加参数修改，默认为8080
安装证书：手机访问mitm.it 下载安装对应证书即可。
- 启动手机游戏
启动任意一个四三九九游戏，观察控制台日志输出，本实例以在4399sy.com中下载的安卓“小小突击队”为例子。
二、安装dwebsocket
下载dwebsocket https://github.com/duanhongyi/dwebsocket 后，进行安装
python setup.py install
三、django项目编写
- 创建项目"sdk_monitor",创建app"demo"
django-admin startproject sdk_monitor
cd sdk_monitor
django-admin startapp demo
- sdk_monitor/url.py
from django.conf.urls import url from demo import views as v urlpatterns = [ # url(r'^admin/', admin.site.urls), url(r'^$', v.index), url(r'^echo$', v.echo), ]
- templates/index.html
<!DOCTYPE html> <html> <head> <title>django-websocket</title> <script src="http://code.jquery.com/jquery-1.11.1.min.js"></script> <script type="text/javascript">//<![CDATA[ $(function () { <!-- socket 连接函数,socket 连接时，传入游戏id，在socket服务端订阅redis频道消息 --> function socket_connect(gameId) { if (window.s) { window.s.close() } /*创建socket连接*/ var socket = new WebSocket("ws://" + window.location.host + "/echo"); socket.onopen = function () { console.log('WebSocket open');//成功连接上Websocket send_socket_message(gameId); //通过websocket发送数据 }; socket.onmessage = function (e) { var data = JSON.parse(e.data); console.log(data.plat + ":" + data.action_name + " 上报时间：" + data.action_time);//打印出服务端返回过来的数据 console.log(data); console.log(""); //$('#messagecontainer').append('<p>' + JSON.stringify(data.url) + '</p>'); $('#messagecontainer').append("<p>" + data.plat + " :[" + data.action_name + "] 上报时间：" + data.action_time + "<br>" + data.method + " : " + data.url + "<br>status_code : " + data.status_code + "<br>response : " + data.response + "<br><hr></p>"); }; // Call onopen directly if socket is already open if (socket.readyState == WebSocket.OPEN) socket.onopen(); window.s = socket; } <!-- 发送socket信息函数 --> function send_socket_message(msg) { if (window.s) { if (window.s.readyState == 1) { window.s.send(msg) } else { alert("websocket已关闭."); } } else { alert("websocket未连接."); } } <!-- 关闭socket连接函数 --> function close_socket() { if (window.s) { if (window.s.OPEN == 1) { window.s.close();//关闭websocket } } console.log('websocket closed'); } $('#connect_websocket').click(function () { var gameId = $('#gameId').val(); socket_connect(gameId); }); }); //]]></script> </head> <body> <br> <input type="text" id="gameId" value="1499130088511390"/> <button type="button" id="connect_websocket">订阅游戏日志</button> <h1>SDK 客户端实时日志</h1> <div id="messagecontainer" style="word-break: break-all"> </div> </body> </html>
- demo/views.py
from django.shortcuts import render from dwebsocket.decorators import accept_websocket from django.http import HttpResponse import redis def index(request): """ socket 订阅消息显示页面 :param request: :return: """ return render(request, 'index.html') @accept_websocket def echo(request): """ socket 服务端接口，根据socket连接时发送的游戏id，进行redis消息订阅，然后将订阅消息返回客户端""" if not request.is_websocket(): # 判断是不是websocket连接 try: # 如果是普通的http方法 gameId = request.GET['gameId'] return HttpResponse(gameId) except: return render(request, 'index.html') else: for gameId in request.websocket: # redis 消息订阅 pool = redis.ConnectionPool(host='127.0.0.1', port=6379, db=1) r = redis.StrictRedis(connection_pool=pool) p = r.pubsub() p.subscribe(gameId) for item in p.listen(): # socket消息为message类型时，将消息发送到socket客户端 if item['type'] == 'message': data = item['data'].decode() request.websocket.send(data) if item['data'] == 'over': break
- 运行django后，访问页面localhost:8000
python manage.py runserver
- 点击页面中的按钮”连接 websocket“后，控制台输出”WebSocket open“
- 启动手机中的游戏“小小突击队”，则页面中实时输出抓包记录（所订阅频道根据输入框中的gameId值）
最终结果优化
稍微美化下前端，梳理对应测试点后，测试过程如下gif动图
