python mitmdump抓包与redis订阅消息

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python mitmdump抓包与redis订阅消息相关的知识,希望对你有一定的参考价值。

本实例实现需求

django项目,后端采用python mitmdump 扩展脚本“sdk_log.py”实时抓取与过滤4399SDK 客户端日志,并且使用redis发布。

前端使用websocket连接,订阅某频道信息,实时输出对应游戏的客户端日志到页面中。

开发环境

win7,python3,

安装redis_server

参考 在windows x64上部署使用Redis

安装python redis

python3 -m pip install redis

安装python mitmproxy

python3 -m pip install mitmproxy

代码实现

一、客户端日志抓包处理脚本 sdk_log.py:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from mitmproxy import http
import urllib
import re
import json
import logging
import redis

# 对日志的输出格式及方式做相关配置
logging.basicConfig(level=logging.DEBUG,
                    format=%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s,
                    datefmt=%a, %d %b %Y %H:%M:%S,
                    filename=sdk.log,
                    filemode=a)

# 过滤目标host的日志
host_filter = [
    udpdcs.4399sy.com,
    sdkdcs.4399sy.com,
    dpdcs.4399sy.com.hk,
    dpdcs.4399en.com,
    dpdcs.4399th.com,
    dpdcs.4399sy.ru,
    dpdcs.moregame.vn,
]


# 大陆联运操作名(从path中获取,如activity_open.php)
udpdcs_action_from_path = {
    init_info: 初始化日志,
    activity_open: 打开游戏日志,
    activity_before_login: 登录界面前日志,
    user_login: 登录日志,
    user_server_login: 选服日志,
    user_online: 在线日志,
    enter_game: 进入游戏日志,
    share: share,
    share_log: share_log,
    photo_share: photo_share,
    event: event
}
# 从path中获取操作类型,如activity_open.php
path_filter = {
    # ‘load_start_before_login‘: u‘登录前加载开始日志‘,
    # ‘load_finish_before_login‘: u‘登录前加载结束日志‘,
    # ‘click_enter‘: u‘进入游戏日志‘,
    # ‘enter_game‘:u‘进入游戏‘,
    # ‘get_user_server_login‘: u‘选服日志‘,
    # ‘user_create_role‘: u‘创角日志‘,
    # ‘role_login‘: u‘角色登录日志‘,
    # ‘enter_success‘: u‘成功进入游戏日志‘,
    # ‘role_level‘: u‘角色升级日志‘,
    # ‘exit_success‘: u‘退出游戏日志‘,
}
# 大陆sy操作名(从body中获取)
sdkdcs_action_from_data = {
    open_game: 打开游戏日志,
    open_login: 登录界面前日志,
    select_server: 选服日志,
    create_role: 创角日志,
    role_level_change: 等级日志,

}

pool = redis.ConnectionPool(host=127.0.0.1, port=6379, db=1)
r = redis.StrictRedis(connection_pool=pool)


def response(flow):
    host = flow.request.host
    method = flow.request.method
    body = flow.request.data
    url = urllib.parse.unquote(flow.request.url)
    path = flow.request.path_components
    querystring = flow.request._get_query()
    gameId = None

    # 大陆联运日志,Android使用GET方法,iOS使用POST方法
    if host == "udpdcs.4399sy.com":
        # GET 方式 (联运Android)
        if method == "GET":
            # 从path中获取操作类型
            action_type = path[0].rstrip(".php")
            action_name = udpdcs_action_from_path.get(action_type, "udpdcs_action_from_path[%s]" % action_type)
            # 从URL参数data中获取主要sdk请求数据
            data = {}
            for eachp in querystring:
                if eachp[0] == "data":
                    data = eachp[1]
                    try:
                        data_send = json.loads(data)
                        gameId = data_send.get(gameId)
                        print("gameId:", gameId)
                        log_msg = "大陆联运[%s]:\\n-->GET URL:%s" % (action_name, url)
                        logging.info(log_msg)
                    except Exception as e:
                        log_msg = "大陆联运[%s]:\\n-->GET 方法中的data内容转换json格式失败\\n-->GET URL:%s" % (action_name, url)
                        logging.error(log_msg)
            # 从URL参数中匹配data参数失败
            if not data:
                log_msg = "大陆联运[%s]:\\n-->GET 方法中的data内容获取失败,\\n-->GET URL:%s" % (action_name, url)
                logging.warning(log_msg)
        # POST 方式(联运iOS)
        else:
            # 从path中获取操作类型
            action_type = path[0].rstrip(".php")
            action_name = udpdcs_action_from_path.get(action_type, "udpdcs_action_from_path[%s]" % action_type)
            # 从POST BODY 中匹配‘data=(.*)\\‘$‘ 获得主要sdk请求数据
            raw_data = str(body.content)
            data_patten = re.compile(data=(.*)\\‘$, re.S)
            data_search = re.search(data_patten, raw_data)
            if data_search:
                data_send = urllib.parse.unquote(data_search.group(1))
                data_send = data_send.replace("\\n", "").replace("\\t", "").replace(" ", "")
                try:
                    data_send = json.loads(
                        data_send)  # eg. data_send={"did":"83F6B45E-6DA9-4B78-9DC0-000278B44F84","gameId":"1483512079389590" ...}
                    gameId = data_send.get(gameId)
                    log_msg = "大陆联运[%s]:\\n-->POST URL:%s,\\n-->POST DATA:%s" % (action_name, url, data_send)
                    logging.info(log_msg)
                except Exception as e:
                    log_msg = "大陆联运[%s]:\\n-->POST 方法中的POST DATA 内容转换json格式失败!\\n-->POST URL:%s\\n-->POST DATA:%s" % (
                        action_name, url, data_send)
                    logging.error(log_msg)
            # 从POST BODY 中匹配‘data=(.*)\\‘$‘ 数据失败
            else:
                log_msg = "大陆联运[%s]:\\n-->POST 方法中的BODY data=内容获取失败!\\n-->POST URL:%s\\n-->POST DATA:%s" % (
                    action_name, url, urllib.parse.unquote(raw_data))
                logging.warning(log_msg)
        print(log_msg)

    # 大陆sy日志,Android,iOS使用POST方法
    if host == "sdkdcs.4399sy.com":
        if method == "POST":
            # 从POST BODY 中匹配‘data=(.*)\\‘$‘ 获得主要sdk请求数据
            raw_data = str(body.content)
            data_patten = re.compile(data=(.*)\\‘$, re.S)
            data_search = re.search(data_patten, raw_data)
            if data_search:
                data_send = urllib.parse.unquote(data_search.group(1))
                data_send = data_send.replace("\\n", "").replace("\\t", "").replace(" ", "")
                try:
                    data_send = json.loads(data_send)  # eg. data_send={"data":{"common":{},"open_game":{}}}
                    data = data_send.get(data)
                    if data:
                        key_list = list(data.keys())
                        if "common" in key_list and len(key_list) == 2:
                            gameId = data.get("common").get("gameId")
                            print("gameId:", gameId)
                            # 获取操作类型
                            key_list.remove("common")
                            action_type = key_list.pop(0)
                            action_name = sdkdcs_action_from_data.get(action_type,
                                                                      "sdkdcs_action_from_data[%s]" % action_type)
                            log_msg = "大陆sy[%s]:\\n-->POST URL:%s \\n-->POST DATA:%s" % (action_name, url, data_send)
                            logging.info(log_msg)
                        else:
                            log_msg = "大陆sy[]:\\n-->BODY 字典中没有common这个key\\n-->POST URL:%\\n-->POST DATA:%s" % (url, data_send)
                            logging.warning(log_msg)
                    else:
                        log_msg = "大陆sy[]:\\n-->BODY 字典中没有data这个key \\n-->POST URL:% \\n-->POST DATA:%s" % (url, data_send)
                        logging.warning(log_msg)
                except json.decoder.JSONDecodeError as e:
                    log_msg = "大陆sy[]:\\n-->POST 方法中的POST DATA 内容转换json格式失败!\\n-->POST URL:%s \\n-->POST DATA:%s" % (url, data_send)
                    logging.error(log_msg)
                except Exception as e:
                    print(e)

            # 从POST BODY 中匹配‘data=(.*)\\‘$‘ 数据失败
            else:
                log_msg = "大陆sy[]:\\n-->POST 方法中的BODY data=内容获取失败!\\n-->POST URL:%s\\n-->POST DATA:%s" % (
                    url, urllib.parse.unquote(raw_data))
                logging.warning(log_msg)
        else:
            log_msg = "大陆sy[]:\\n-->大陆日志使用了GET方式?\\n-->GET URL:%s" % url
            logging.warning(log_msg)
        print(log_msg)
    # redis 发布
    if gameId:
        r.publish(gameId, log_msg)
  • 启动抓包脚本

在cmd中输入命令

mitmdump -s sdk_log.py ~u abc.com

正确启动后如下

E:\\workspace_python\\tmp>mitmdump -s sdk_log.py ~u abc.com
Loading script: sdk_log.py
Proxy server listening at http://0.0.0.0:8080
192.168.1.103:39247: clientconnect
  • 手机连接代理

手机连上与电脑相同局域网wifi,并设置代理。如电脑端ip为192.168.1.104,则设置代理为 192.168.1.104:8080 ,端口可以在mitmdump中添加参数修改,默认为8080

安装证书:手机访问mitm.it 下载安装对应证书即可。

  • 启动手机游戏

启动任意一个四三九九游戏,观察控制台日志输出,本实例以在4399sy.com中下载的安卓“翻滚球球”为例子。

192.168.1.103:40586: clientconnect
大陆联运[初始化日志]:
-->GET 方法中的data内容获取失败,
-->GET URL:http://udpdcs.4399sy.com/init_info.php?time=1496146854&flag=ee16ef51b6aee287a4f87be08aee2d6e
gameId: 1461722512884260
大陆联运[打开游戏日志]:
-->GET URL:http://udpdcs.4399sy.com/activity_open.php?time=1496146854&flag=cab238218ccf3c3e7f8cf103d389e621&data={"eventId":"0","ip":"0","did":"861744030244058","appVersion":"1.4.4.0","sdkVersion":"2.6.9.3","platformId":"274","gameId":"1461722512884260","areaId":"0","serverId":"0","os":"android","osVersion":"5.1","device":"OPPO+R9m","deviceType":"android","screen":"1920*1080","mno":"中国电信","nm":"WIFI","eventTime":"0","channel":"4399_cz","channelOld":"4399_cz","channelSy":"113","sim":"0","kts":"23c023c0acefd77a11ca131f092bf85d","pkgName":"com.jingmo.ball3d"}
gameId: 1461722512884260
大陆sy[打开游戏日志]:
-->POST URL:http://sdkdcs.4399sy.com/?time=1496146854&flag=92aae983b4afc6b7236e29cbe19411f1
-->POST DATA:{data: {open_game: [{ip: 0, appVersion: 1.4.4.0, sdkVersion: 3.7.20.0, channelId: 113, nm: WIFI, retry: 1, roleLevel: 0, roleName: ‘‘, nickname: ‘‘, serverId: 0, uid: 0, eventTime: 1496146854, msgId: 0, ic: MMHxiazowfGNkOjA=}], common: {eventId: 0, did: 861744030244058, gameId: 1461722512884260, os: android, osVersion: 5.1, device: OPPO+R9m, deviceType: android, screen: 1920*1080, mno: 中国电信, areaId: 1}}}
192.168.1.103:41319: clientconnect

 二、安装dwebsocket

下载dwebsocket https://github.com/duanhongyi/dwebsocket 后,进行安装

python setup.py install

三、django项目编写

  • 创建项目"dj_websocket",创建app"demo"
django-admin startproject dj_websocket
cd dj_websocket
django-admin startapp demo
  • dj_websocket/url.py
from django.conf.urls import url
from demo import views as v

urlpatterns = [
    # url(r‘^admin/‘, admin.site.urls),
    url(r^index/, 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 () {
        $(#connect_websocket).click(function () {
            if (window.s) {
                window.s.close()
            }
            /*创建socket连接*/
            var socket = new WebSocket("ws://" + window.location.host + "/echo");
            socket.onopen = function () {
                console.log(WebSocket open);//成功连接上Websocket
                window.s.send($(#message).val());//通过websocket发送数据
            };
            socket.onmessage = function (e) {
                console.log([SDK]:  + e.data);//打印出服务端返回过来的数据
                $(#messagecontainer).append(<p> + e.data + </p>);
            };
            // Call onopen directly if socket is already open
            if (socket.readyState == WebSocket.OPEN) socket.onopen();
            window.s = socket;
        });
        $(#send_message).click(function () {
            //如果未连接到websocket
            if (!window.s) {
                alert("websocket未连接.");
            } else {
                window.s.send($(#message).val());//通过websocket发送数据
            }
        });
        $(#close_websocket).click(function () {
            if (window.s) {
                window.s.close();//关闭websocket
                console.log(websocket已关闭);
            }
        });

    });
    //]]></script>
</head>
<body>
<br>
<input type="text" id="message" value="1461722512884260"/>
<button type="button" id="connect_websocket">连接 websocket</button>
<button type="button" id="send_message" style="display: none">发送 message</button>
<button type="button" id="close_websocket" style="display: none">关闭 websocket</button>
<h1>SDK 客户端实时日志</h1>

<div id="messagecontainer" style="word-break: break-all">
</div>
</body>
</html>
  • demo/views.py
# -*- coding: utf-8 -*-
from django.shortcuts import render
from dwebsocket.decorators import accept_websocket
from django.http import HttpResponse
import redis


def index(request):
    return render(request, index.html)


@accept_websocket
def echo(request):
    if not request.is_websocket():  # 判断是不是websocket连接
        try:  # 如果是普通的http方法
            message = request.GET[message]
            return HttpResponse(message)
        except:
            return render(request, index.html)
    else:
        for message in request.websocket:
            print("gameId:", message)
            pool = redis.ConnectionPool(host=127.0.0.1, port=6379, db=1)
            r = redis.StrictRedis(connection_pool=pool)
            p = r.pubsub()
            p.subscribe(message)
            for item in p.listen():
                # request.websocket.send(json.dumps(item))  # 发送消息到客户端
                if item[type] == message:
                    data = item[data]
                    print("sdk_log:", data)
                    request.websocket.send(data)
                    if item[data] == over:
                        break;
  • 运行django后,访问页面localhost:8000
python manage.py runserver
  • 点击页面中的按钮”连接 websocket“后,控制台输出”WebSocket open“
  • 启动手机中的游戏“翻滚球球”,则页面中实时输出抓包记录(所订阅频道根据输入框中的gameId值)

技术分享

 

以上是关于python mitmdump抓包与redis订阅消息的主要内容,如果未能解决你的问题,请参考以下文章

Mac下mitmproxy和mitmdump的使用

App爬虫神器mitmproxy和mitmdump的使用

使用SElinux抓包与扫描

Fiddler抓包与设备设置代理

以羊了个羊为例,浅谈小程序抓包与响应报文篡改

Apache网页优化:网页压缩,网页缓存(内含源码包与抓包工具)