本實例實現需求
在游戲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:\workspace\sdk_monitor\demo>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動圖
