首页 > 代码库 > python mitmdump抓包与redis订阅消息
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 =http://www.mamicode.com/ { ‘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 =http://www.mamicode.com/ {} 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=http://www.mamicode.com/(.*)/‘$‘ 获得主要sdk请求数据 raw_data =http://www.mamicode.com/ str(body.content) data_patten = re.compile(‘data=http://www.mamicode.com/(.*)/‘$‘, 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=http://www.mamicode.com/(.*)/‘$‘ 数据失败 else: log_msg = "大陆联运[%s]:\n-->POST 方法中的BODY data=http://www.mamicode.com/内容获取失败!/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=http://www.mamicode.com/(.*)/‘$‘ 获得主要sdk请求数据 raw_data =http://www.mamicode.com/ str(body.content) data_patten = re.compile(‘data=http://www.mamicode.com/(.*)/‘$‘, 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 = http://www.mamicode.com/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=http://www.mamicode.com/(.*)/‘$‘ 数据失败 else: log_msg = "大陆sy[]:\n-->POST 方法中的BODY data=http://www.mamicode.com/内容获取失败!/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=http://www.mamicode.com/{"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订阅消息
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。