python+mitmproxy抓包過濾+redis消息訂閱+websocket實時消息發送，日誌實時輸出到web界面

時間 2019-11-18

標籤 python+mitmproxy python mitmproxy 抓包過濾 redis 消息訂閱 websocket 實時發送日誌輸出 web 界面欄目 Python 简体版

原文原文鏈接

本實例實現需求

在遊戲SDK測試中，常常須要測試遊戲中SDK的埋點日誌是否接入正確。本實例經過抓包（客戶端http/https 請求）來斷定埋點日誌是是否接入正確。javascript

實現細節：使用django項目，後端採用python mitmdump 擴展腳本「log_handler.py」實時抓取與過濾4399SDK 客戶端日誌，將數據處理成約定須要的格式，保存和發佈到redis中。php

前端使用websocket鏈接，websocket服務端經過redis訂閱對應遊戲頻道信息，實時輸出遊戲的客戶端日誌到web頁面中。html

開發環境

win7，python3,前端

安裝redis_serverjava

參考在windows x64上部署使用Redispython

安裝python redisjquery

python3 -m pip install redis

安裝python mitmproxyandroid

python3 -m pip install mitmproxy

代碼實現

1、客戶端日誌抓包處理腳本 log_handler.py:git

#!/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中輸入命令github

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中下載的安卓「小小突擊隊」爲例子。

2、安裝dwebsocket

下載dwebsocket https://github.com/duanhongyi/dwebsocket 後，進行安裝

python setup.py install

3、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動圖

***微信掃一掃，關注「python測試開發圈」，瞭解更多測試教程！***

相關標籤/搜索