python+pcap+dpkt 抓包小实例

Posted 2021-02-11 wangjq_china

  1 #!/usr/bin/env python

  2 # -*- coding: utf-8 -*-
  3 
  4 """ 网络数据包捕获与分析程序 """
  5 
  6 import pcap
  7 import dpkt
  8 import json
  9 import re
 10 import time
 11 from urllib import unquote
 12 
 13 # 过滤输出目标ip
 14 dst_lists = [
 15     \'203.66.1.212\',  # nslookup dpdcs.4399sy.com.hk
 16     \'52.74.10.186\',  # nslookup dpdcs.4399en.com
 17     \'52.58.69.212\',  # nslookup dpdcs.4399sy.ru
 18     \'220.241.11.3\',  # nslookup dpdcs.4399th.com
 19     \'124.243.195.63\',  # nslookup sdkdcs.4399sy.com
 20     \'42.62.106.216\',  # nslookup udpdcs.4399sy.com
 21     \'42.62.106.230\',  # nslookup udpdcs.4399sy.com
 22 ]
 23 
 24 req_data = ""
 25 times = 0
 26 
 27 
 28 def capt_data(eth_name="eth0", p_type=None):
 29     """
 30     捕获网卡数据包
 31     :param eth_name  网卡名，eg. eth0,eth3...
 32     :param p_type    日志捕获类型 1：sdk日志用例分析 2：目标域名过滤输出 3：原始数据包
 33     :return:
 34     """
 35 
 36     pc = pcap.pcap(eth_name)
 37     pc.setfilter(\'tcp port 80\')  # 设置监听过滤器
 38     print \'start capture....\'
 39     if pc:
 40         for p_time, p_data in pc:  # p_time为收到时间，p_data为收到数据
 41             anly_capt(p_time, p_data, p_type)
 42 
 43 
 44 def anly_capt(p_time, p_data, p_type):
 45     """
 46     解析数据包
 47     :param p_data  收到数据
 48     :param p_type  日志捕获类型 1：sdk日志用例分析 2：目标域名过滤输出 3：原始数据包
 49     :return:
 50     """
 51 
 52     p = dpkt.ethernet.Ethernet(p_data)
 53     if p.data.__class__.__name__ == \'IP\':
 54         ip_data = p.data
 55         src_ip = \'%d.%d.%d.%d\' % tuple(map(ord, list(ip_data.src)))
 56         dst_ip = \'%d.%d.%d.%d\' % tuple(map(ord, list(ip_data.dst)))
 57         if p.data.data.__class__.__name__ == \'TCP\':
 58             tcp_data = p.data.data
 59             if tcp_data.dport == 80:
 60                 # print tcp_data.data
 61                 if tcp_data.data:
 62                     # 调用日志模块，对日志进行处理
 63                     if p_type == 1:
 64                         # sdk日志用例分析
 65                         if dst_ip in dst_lists:
 66                             tmp = tcp_data.data.strip()
 67                             global req_data, times
 68                             if tmp.startswith("POST") or tmp.startswith("GET"):  # or times > 0
 69                                 if req_data:
 70                                     haiwai_log_case(req_data)
 71                                 req_data = tmp + "\\n"
 72                                 # times = 0
 73                             else:
 74                                 req_data = req_data + tmp
 75                                 # times = times + 1
 76 
 77                     elif p_type == 2:
 78                         # 目标域名过滤输出
 79                         if dst_ip in dst_lists:
 80                             print "tcp_data:", tcp_data.data
 81 
 82                     else:
 83                         # 无过滤条件输出
 84                         print "tcp_data:", tcp_data.data
 85 
 86 
 87 # android 日誌類型，从data中获取
 88 log_type_from_data = {
 89 
 90     \'open_game\': u\'[打开游戏]\',
 91     \'network_check\': u\'[网络监测]\',
 92     \'open_login\': u\'[登录界面前]\',
 93     \'select_server\': u\'[选服日志]\',
 94     \'create_role\': u\'[创角日志]\',
 95     \'role_level_change\': u\'[等级日志]\',
 96 
 97     # 海外,俄语
 98     \'activity_open\': u\'[打开游戏]\',
 99     \'load_start_before_login\': u\'[加载开始]\',
100     \'load_finish_before_login\': u\'[加载结束]\',
101     \'activity_before_login\': u\'[登录界面前]\',
102     \'click_enter\': u\'[进入游戏]\',
103     \'get_user_server_login\': u\'[选服日志]\',
104     \'user_create_role\': u\'[创角日志]\',
105     \'role_login\': u\'[角色登录]\',
106     \'enter_success\': u\'[成功进入游戏]\',
107     \'role_level\': u\'[等级日志]\',
108     \'user_online\': u\'[在线日志]\',
109     \'exit_success\': u\'[退出游戏]\',
110 
111 }
112 
113 # ios日誌類型，从请求资源路径获取
114 log_type_from_path = {
115     \'activity_open.php\': u\'[打开游戏]\',
116     \'load_start_before_login.php\': u\'[加载开始]\',
117     \'load_finish_before_login.php\': u\'[加载结束]\',
118     \'activity_before_login.php\': u\'[登录界面前]\',
119     \'click_enter.php\': u\'[进入游戏]\',
120     \'get_user_server_login.php\': u\'[选服日志]\',
121     \'user_create_role.php\': u\'[创角日志]\',
122     \'role_login.php\': u\'[角色登录]\',
123     \'enter_success.php\': u\'[成功进入游戏]\',
124     \'user_online.php\': u\'[在线日志]\',
125     \'role_level.php\': u\'[等级日志]\',
126     \'exit_success.php\': u\'[退出游戏]\',
127     \'share.php\': u\'[分享日志]\',
128     \'init_info.php\': u\'[初始化日志]\',
129     \'event.php\': u\'[事件日志]\',
130     \'user_login.php\': u\'[user_login]\',
131     \'user_server_login.php\': u\'[user_server_login]\',
132     \'enter_game.php\': u\'[enter_game]\',
133 }
134 
135 # 过滤path
136 filter_out_list = [
137     \'u/\',
138     \'plugin/error/check\',
139     \'service/version/get_info\',
140 ]
141 
142 # 过滤打印出属于列表中的host的日志。
143 host_list = [
144     \'dpdcs.4399sy.com.hk\',
145     \'dpdcs.4399en.com\',
146     \'dpdcs.4399sy.ru\',
147     \'dpdcs.4399th.com\',
148     \'sdkdcs.4399sy.com\',
149     \'udpdcs.4399sy.com\',
150 ]
151 
152 
153 def formattime(t):  # 日期字段格式化
154     return time.strftime(\'%c\', time.gmtime(t + 8 * 3600))
155 
156 
157 def req_to_dict(req_string):
158     """
159     将请求数据转换为dic
160     :param req_string:
161     :return:
162     """
163     req_dict = {}
164     req_string = req_string.strip()
165     if len(req_string) > 0:
166         req_string = unquote(req_string)
167         # print "req_string_after_unquote:",req_string
168         m1 = re.search("(GET|POST)(.*)\\?(.*)HTTP/1.1", req_string)  # (method,path,param)
169         m2 = re.search("Host:(.*)", req_string)  # (host,)
170         # m3 = re.search("\\sdata=(.*)\\s", req_string)  # (body,)
171         m4 = re.search("\\sdata=([\\s\\S]*)", req_string)  # (body,)
172         # m5 = re.search("eventTime\\":\\"(\\d+)", req_string)  # (eventTime,)
173         m5 = re.search("eventTime\\"\\s*:\\s*\\"(\\d+)", req_string)
174         if m1:
175             req_dict["method"] = m1.group(1).strip()
176             req_dict["path"] = m1.group(2).strip()[1:]
177             param_string = m1.group(3).strip()
178             if param_string:
179                 param_string = param_string.split("&")
180                 param_dict = {}
181                 for item in param_string:
182                     tmp_list = item.split("=")
183                     if len(tmp_list) > 1:
184                         param_dict[tmp_list[0]] = tmp_list[1]
185                 req_dict["param"] = param_dict
186         if m2:
187             req_dict["host"] = m2.group(1).strip()
188         if m4:
189             try:
190                 body = m4.group(1).replace("\\n", "")
191                 body = json.loads(body)
192             except ValueError:
193                 print "\\033[1;31;40m"
194                 print "m4:Error:body ValueError,req_string-->%s" % req_string
195                 print "\\033[0m"
196                 body = {}
197             req_dict["body"] = body
198 
199         if m5:
200             req_dict["eventTime"] = formattime(int(m5.group(1)))
201 
202     return req_dict
203 
204 
205 def haiwai_log_assert(req_dict):
206     """
207     日志断言处理,输出分析结果
208     :param req_dict:
209     :return:
210     """
211 
212     # 从 data 中获取日志类型
213     if isinstance(req_dict, dict) and req_dict.get("body"):
214         if req_dict.get("body").get("data"):
215             data_type = req_dict.get("body").get("data").keys()
216             data_type_set = set(data_type)
217             types_key_set = set(log_type_from_data.keys())
218             intersect = data_type_set.intersection(types_key_set)
219             if intersect:
220                 log_type = intersect.pop()
221                 print "\\033[1;31;40m %s log pass!--from body data || EventTime:-->[%s] \\033[0m" % (
222                     log_type_from_data.get(log_type), req_dict.get("eventTime"))
223                 print req_dict
224             else:
225                 if \'common\' in data_type and len(data_type) == 2:
226                     data_type.remove(\'common\')
227                     print "\\033[1;31;40m %s log not register!--from body data \\033[0m" % data_type
228 
229     # 从 path 中获取日志类型
230     path = req_dict.get("path")
231     host = req_dict.get("host")
232     if host in host_list:
233         if path in log_type_from_path.keys():
234             eventTime = ""
235             if req_dict.get("eventTime"):
236                 eventTime = req_dict.get("eventTime")
237             else:
238                 if req_dict.get("param"):
239                     eventTime = req_dict.get("param").get("time")
240                     if eventTime:
241                         eventTime = formattime(int(eventTime))
242             print "\\033[1;31;40m %s log pass--from url || EventTime:-->[%s] \\033[0m" % (
243                 log_type_from_path.get(path), eventTime)
244             print req_dict
245         elif path and path not in filter_out_list:
246             print "\\033[1;31;40m %s log not register!--from url! \\033[0m" % path
247             print req_dict
248 
249 
250 def client_log_check(log_type, req_dict, platform="sy"):
251     """
252     检查SDK客户端请求字段，返回测试结果集
253     :param log_type: 日志类型
254     :param req_dict: 日志字典
255     :param platform: 测试平台
256     :return:
257     """
258     pass
259 
260 
261 def haiwai_log_case(req_string):
262     """
263     日志用例集
264     一：将数据包转换为dict
265     二：对日志分析处理，输出测试结果
266     """
267 
268     req_dict = req_to_dict(req_string)
269     haiwai_log_assert(req_dict)
270 
271 
272 if __name__ == \'__main__\':
273     try:
274         capt_data("eth3", 1)
275     except TypeError:
276         capt_data("eth3", 1)

原文链接：python+pcap+dpkt抓包小实例