dpkt项目是一个python模块，用于快速、简单的数据包解析，并定义了基本TCP/IP协议，使用该库可以快速解析通过各类抓包工具抓到的数据包，从而提取分析包内的参数。
使用Dpkt分析数据包: 使用Dpkt发现URL中存在的.zip字样链接
代码语言：javascript复制#coding=utf-8
import dpkt
import socket
def FindPcapWord(pcap,WordKey):
for ts,buf in pcap:
try:
eth = dpkt.ethernet.Ethernet(buf)
ip = eth.data
src = socket.inet_ntoa(ip.src)
dst = socket.inet_ntoa(ip.dst)
tcp = ip.data
http = dpkt.http.Request(tcp.data)
if(http.method == "GET"):
uri = http.uri.lower()
if WordKey in uri:
print("[ ] 源地址: {} --> 目标地址: {} 检索到URL中存在 {}".format(src,dst,uri))
except Exception:
pass
fp = open("D://aaa.pcap","rb")
pcap = dpkt.pcap.Reader(fp)
FindPcapWord(pcap,"wang.zip")也可以使用dpkt解析本机数据包中是否包含后门。
代码语言：javascript复制#coding=utf-8
import dpkt
import socket
def FindPcapWord(pcap,WordKey):
for timestamp,packet in pcap:
try:
eth = dpkt.ethernet.Ethernet(packet)
ip = eth.data
src = socket.inet_ntoa(ip.src)
dst = socket.inet_ntoa(ip.dst)
tcp = ip.data
http = dpkt.http.Request(tcp.data)
if(http.method == "GET"):
uri = http.uri.lower()
if WordKey in uri:
print("[ ] 源地址: {} --> 目标地址: {} 检索到URL中存在 {}".format(src,dst,uri))
except Exception:
pass
def Banner():
print(" _ ____ _ _ ")
print(" | | _ _/ ___|| |__ __ _ _ __| | __")
print(" | | | | | ___ | '_ / _` | '__| |/ /")
print(" | |__| |_| |___) | | | | (_| | | | < ")
print(" |_______, |____/|_| |_|__,_|_| |_|_\")
print(" |___/ n")
print("E-Mail: me@lyshark.com")
def FindHivemind(pcap):
for timestamp,packet in pcap:
try:
eth = dpkt.ethernet.Ethernet(packet)
ip = eth.data
tcp = ip.data
src = socket.inet_ntoa(ip.src)
dst = socket.inet_ntoa(ip.dst)
sport = tcp.sport
dport = tcp.dport
# print("[ ] 源地址: {}:{} --> 目标地址:{}:{}".format(src,sport,dst,dport))
if dport == 80 and dst == "125.39.247.226":
# 如果数据流中存在cmd等明文命令则说明可能存在后门
if '[cmd]# ' in tcp.data.lower():
print("[ ] {}:{}".format(dst,dport))
except Exception:
pass
Banner()
fp = open("D://aaa.pcap","rb")
pcap = dpkt.pcap.Reader(fp)
FindHivemind(pcap)实时检测DDoS攻击: 主要通过设置检测不正常数据包数量的阈值来判断是否存在DDoS攻击。
代码语言：javascript复制#coding=utf-8
import dpkt
import socket
def FindPcapWord(pcap,WordKey):
for timestamp,packet in pcap:
try:
eth = dpkt.ethernet.Ethernet(packet)
ip = eth.data
src = socket.inet_ntoa(ip.src)
dst = socket.inet_ntoa(ip.dst)
tcp = ip.data
http = dpkt.http.Request(tcp.data)
if(http.method == "GET"):
uri = http.uri.lower()
if WordKey in uri:
print("[ ] 源地址: {} --> 目标地址: {} 检索到URL中存在 {}".format(src,dst,uri))
except Exception:
pass
def FindHivemind(pcap):
for timestamp,packet in pcap:
try:
eth = dpkt.ethernet.Ethernet(packet)
ip = eth.data
tcp = ip.data
src = socket.inet_ntoa(ip.src)
dst = socket.inet_ntoa(ip.dst)
sport = tcp.sport
dport = tcp.dport
# print("[ ] 源地址: {}:{} --> 目标地址:{}:{}".format(src,sport,dst,dport))
if dport == 80 and dst == "125.39.247.226":
# 如果数据流中存在cmd等明文命令则说明可能存在后门
if '[cmd]# ' in tcp.data.lower():
print("[ ] {}:{}".format(dst,dport))
except Exception:
pass
def Banner():
print(" _ ____ _ _ ")
print(" | | _ _/ ___|| |__ __ _ _ __| | __")
print(" | | | | | ___ | '_ / _` | '__| |/ /")
print(" | |__| |_| |___) | | | | (_| | | | < ")
print(" |_______, |____/|_| |_|__,_|_| |_|_\")
print(" |___/ n")
print("E-Mail: me@lyshark.com")
def FindDDosAttack(pcap):
pktCount = {}
for timestamp,packet in pcap:
try:
eth = dpkt.ethernet.Ethernet(packet)
ip = eth.data
tcp = ip.data
src = socket.inet_ntoa(ip.src)
dst = socket.inet_ntoa(ip.dst)
sport = tcp.sport
# 累计判断各个src地址对目标地址80端口访问次数
if dport == 80:
stream = src ":" dst
if pktCount.has_key(stream):
pktCount[stream] = pktCount[stream] 1
else:
pktCount[stream] = 1
except Exception:
pass
for stream in pktCount:
pktSent = pktCount[stream]
# 如果超过设置的检测阈值500,则判断为DDOS攻击行为
if pktSent > 500:
src = stream.split(":")[0]
dst = stream.split(":")[1]
print("[ ] 源地址: {} 攻击: {} 流量: {} pkts.".format(src,dst,str(pktSent)))
if __name__ == "__main__":
Banner()
fp = open("D://data.pcap","rb")
pcap = dpkt.pcap.Reader(fp)
FindPcapWord(pcap,"wang.zip")DPKT动态抓包解析: 首先使用scapy动态抓包,然后调用不同的函数对抓到的数据包进行处理提取出想要的数据.
代码语言：javascript复制import os,argparse,dpkt
from scapy.all import *
pkts=[]
count=0
# 检查数据包的IP层,提取出IP和TTL字段的值
def Get_TTL(pkt):
try:
if pkt.haslayer(IP):
ip_src = pkt.getlayer(IP).src
ip_sport = pkt.getlayer(IP).sport
ip_dst = pkt.getlayer(IP).dst
ip_dport = pkt.getlayer(IP).dport
ip_ttl = str(pkt.ttl)
print("[ ] 源地址: %-15s:%-5s --> 目标地址: %-15s:%-5s --> TTL: %-5s"%(ip_src,ip_sport,ip_dst,ip_dport,ip_ttl))
except Exception:
pass
# 获取本机发送出去的DNS请求所对应的网站地址
def Get_DNSRR(pkt):
if pkt.haslayer(DNSRR):
rrname = pkt.getlayer(DNSRR).rrname
rdata = pkt.getlayer(DNSRR).rdata
ttl = pkt.getlayer(DNSRR).ttl
print("[ ] 域名: {} --> 别名: {} --> TTL: {}".format(rrname,rdata,ttl))
# 解析网页的DNS查询记录
def Get_DNSQR(pkt):
# 判断是否含有DNSRR且存在UDP端口53
if pkt.haslayer(DNSRR) and pkt.getlayer(UDP).sport == 53:
rcode = pkt.getlayer(DNS).rcode
qname = pkt.getlayer(DNSQR).qname
# 若rcode为3,则表示该域名不存在
if rcode == 3:
print("[-] 域名解析不存在")
else:
print("[ ] 解析DNSQR存在:" str(qname))
# 检测主机是否被DDOS攻击了
def FindDDosAttack(pcap):
pktCount = {}
for timestamp,packet in pcap:
try:
eth = dpkt.ethernet.Ethernet(packet)
ip = eth.data
tcp = ip.data
src = socket.inet_ntoa(ip.src)
dst = socket.inet_ntoa(ip.dst)
sport = tcp.sport
# 累计判断各个src地址对目标地址80端口访问次数
if dport == 80:
stream = src ":" dst
if pktCount.has_key(stream):
pktCount[stream] = pktCount[stream] 1
else:
pktCount[stream] = 1
except Exception:
pass
for stream in pktCount:
pktSent = pktCount[stream]
# 如果超过设置的检测阈值500,则判断为DDOS攻击行为
if pktSent > 500:
src = stream.split(":")[0]
dst = stream.split(":")[1]
print("[ ] 源地址: {} 攻击: {} 流量: {} pkts.".format(src,dst,str(pktSent)))
# FindPcapURL 监控提取数据包中的所有URL
def FindPcapURL(pcap):
Url = []
for timestamp,packet in pcap:
try:
eth = dpkt.ethernet.Ethernet(packet)
ip = eth.data
src = socket.inet_ntoa(ip.src)
tcp = ip.data
http = dpkt.http.Request(tcp.data)
if(http.method == "GET"):
UrlHead = http.headers
for key,value in UrlHead.items():
url = re.findall('^https*://.*',str(value))
if url:
print("[ ] 源地址: s --> 访问URL: %-80s"%(src, url[0]))
except Exception:
pass
return set(Url)
# 动态保存pcap文件(每1024字节保存一次pcap文件),并读取出其中的网址解析出来
def write_cap(pkt):
global pkts
global count
pkts.append(pkt)
count = 1
if count == 1024:
wrpcap("data.pcap",pkts)
fp = open("./data.pcap","rb")
pcap = dpkt.pcap.Reader(fp)
FindPcapURL(pcap)
fp.close()
pkts,count = [],0
def Banner():
print(" _ ____ _ _ ")
print(" | | _ _/ ___|| |__ __ _ _ __| | __")
print(" | | | | | ___ | '_ / _` | '__| |/ /")
print(" | |__| |_| |___) | | | | (_| | | | < ")
print(" |_______, |____/|_| |_|__,_|_| |_|_\")
print(" |___/ n")
print("E-Mail: me@lyshark.com")
if __name__ == "__main__":
Banner()
parser = argparse.ArgumentParser()
parser.add_argument("--mode",dest="mode",help="模式选择<TTL/DNSRR/DNSQR/URL>")
args = parser.parse_args()
if args.mode == "TTL":
print("[*] 开始抓取本机TTL流量")
sniff(prn=Get_TTL,store=0)
elif args.mode == "DNSRR":
print("[*] 开始抓取本机发送出去的DNS查询请求所对应的网站URL")
sniff(prn=Get_DNSRR,store=0)
elif args.mode == "DNSQR":
print("[*] 解析网页的DNS查询记录")
sniff(prn=Get_DNSQR,store=0)
elif args.mode == "URL":
print("[ ] 开始抓包,pcap文件并读取出其中的网址")
sniff(prn=write_cap,store=0)
else:
parser.print_help()Geoip2定位IP来源: 首先提取出Pcpa格式的数据包文件,然后通过使用离线数据库查询出指定IP地址的地理位置.
代码语言：javascript复制# pip install geoip2
# github地址下载：https://github.com/maxmind/GeoIP2-python
# 离线数据库：https://www.maxmind.com/en/accounts/current/geoip/downloads
import argparse
import socket,dpkt
import geoip2.database
def AnalysisPace(DpktPack,Filter):
respon = []
with open(DpktPack,"rb") as fp:
pcap = dpkt.pcap.Reader(fp)
for timestamp, packet in pcap:
try:
eth = dpkt.ethernet.Ethernet(packet)
# 解析过滤出网络层(三层)中的IP数据包
if eth.data.__class__.__name__ == "IP":
ip = eth.data
src = socket.inet_ntoa(ip.src)
dst = socket.inet_ntoa(ip.dst)
# 解析过滤出传输层(四层)中的TCP数据包
if eth.data.data.__class__.__name__ == "TCP":
sport = eth.data.data.sport
dport = eth.data.data.dport
# 过滤出源地址是192.168.1.2且目的端口是80或者443的流量
# if src == "192.168.1.2" and dport == 80 or dport == 443:
if eval(Filter):
dic = { "src":"None","sport":0 , "dst":"None","dport":0 }
#print("[ ] 时间戳: %-17s 源地址: %-14s:%-2s ---> 目标地址: %-16s:%-2s" %(timestamp,src, sport, dst, dport))
RecvData = eth.data.data.data
if len(RecvData) and b"GET" in RecvData:
#print("[*] 时间戳: {} 源地址: {} <--- 访问网页: {}".format(timestamp,src,bytes.decode(RecvData).split("n")[1]))
pass
dic['src'] = src
dic['dst'] = dst
dic['sport'] = sport
dic['dport'] = dport
respon.append(dic)
except Exception:
pass
return respon
def AnalysisIP_To_Address(PcapFile,MmdbFile):
IPDict = AnalysisPace(PcapFile,"dport ==80 or dport == 443")
NoRepeat = []
for item in range(len(IPDict)):
NoRepeat.append(IPDict[item].get("dst"))
NoRepeat = set(NoRepeat)
reader = geoip2.database.Reader(MmdbFile)
for item in NoRepeat:
try:
response = reader.city(item)
print("[ ] IP地址: %-16s --> " %item,end="")
print("网段: %-16s --> " %response.traits.network,end="")
print("经度: %-10s 纬度: %-10s --> " %(response.location.latitude, response.location.longitude),end="")
print("定位: {} {} {}".format(response.country.names["zh-CN"],response.subdivisions.most_specific.name,response.city.name),end="n")
except Exception:
print("定位: None None None")
pass
def Banner():
print(" _ ____ _ _ ")
print(" | | _ _/ ___|| |__ __ _ _ __| | __")
print(" | | | | | ___ | '_ / _` | '__| |/ /")
print(" | |__| |_| |___) | | | | (_| | | | < ")
print(" |_______, |____/|_| |_|__,_|_| |_|_\")
print(" |___/ n")
print("E-Mail: me@lyshark.comn")
if __name__ == '__main__':
Banner()
parser = argparse.ArgumentParser()
parser.add_argument("-p", "--pcap", dest="pcap", help="设置抓到的数据包 *.pcap")
parser.add_argument("-d", "--mmdb", dest="mmdb", help="设置城市数据库 GeoLite2-City.mmdb")
args = parser.parse_args()
# 使用方式: main.py -p data.pcap -d GeoLite2-City.mmdb (分析数据包中IP)
if args.pcap and args.mmdb:
AnalysisIP_To_Address(args.pcap,args.mmdb)
else:
parser.print_help()
