AI 助理
备案控制台
开发者社区 数据库 文章 正文

Python 运用Dpkt库解析数据包

2022-12-21 667
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云解析 DNS，旗舰版 1个月
全局流量管理 GTM，标准版 1个月
公共DNS（含HTTPDNS解析），每月1000万次HTTP解析
简介： dpkt项目是一个python模块，用于快速、简单的数据包解析，并定义了基本TCP/IP协议，使用该库可以快速解析通过各类抓包工具抓到的数据包，从而提取分析包内的参数。

dpkt项目是一个python模块，用于快速、简单的数据包解析，并定义了基本TCP/IP协议，使用该库可以快速解析通过各类抓包工具抓到的数据包，从而提取分析包内的参数。

使用Dpkt分析数据包: 使用Dpkt发现URL中存在的.zip字样链接

#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解析本机数据包中是否包含后门。

#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攻击。

#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动态抓包,然后调用不同的函数对抓到的数据包进行处理提取出想要的数据.

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("[+] 源地址: %10s --> 访问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地址的地理位置.

# 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.com\n")

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()
文章标签：
DDoS防护
云解析DNS
Python
定位技术
网络安全
网络协议
数据库
关键词：
Python库
python云解析DNS
Python解析
云解析DNS库
库云解析DNS
aliyun_11084
目录
相关文章
vohelon
|
4天前
|
XML JSON 数据库
Python的标准库
Python的标准库
vohelon
116 77
vohelon
|
5天前
|
XML JSON 数据库
Python的标准库
Python的标准库
vohelon
30 11
蓝易云
|
5天前
|
数据可视化 Python
以下是一些常用的图表类型及其Python代码示例，使用Matplotlib和Seaborn库。
通过这些思维导图和分析说明表，您可以更直观地理解和选择适合的数据可视化图表类型，帮助更有效地展示和分析数据。
蓝易云
37 8
肥猪肥猪-17824
|
10天前
|
存储 缓存 Python
Python中的装饰器深度解析与实践
在Python的世界里，装饰器如同一位神秘的魔法师，它拥有改变函数行为的能力。本文将揭开装饰器的神秘面纱，通过直观的代码示例，引导你理解其工作原理，并掌握如何在实际项目中灵活运用这一强大的工具。从基础到进阶，我们将一起探索装饰器的魅力所在。
肥猪肥猪-17824
26 4
左诗右码
|
14天前
|
Android开发 开发者 Python
通过标签清理微信好友：Python自动化脚本解析
微信已成为日常生活中的重要社交工具，但随着使用时间增长，好友列表可能变得臃肿。本文介绍了一个基于 Python 的自动化脚本，利用 `uiautomator2` 库，通过模拟用户操作实现根据标签批量清理微信好友的功能。脚本包括环境准备、类定义、方法实现等部分，详细解析了如何通过标签筛选并删除好友，适合需要批量管理微信好友的用户。
左诗右码
24 7
bruce_xiaowei
|
16天前
|
测试技术 开发者 Python
使用Python解析和分析源代码
本文介绍了如何使用Python的`ast`模块解析和分析Python源代码，包括安装准备、解析源代码、分析抽象语法树（AST）等步骤，展示了通过自定义`NodeVisitor`类遍历AST并提取信息的方法，为代码质量提升和自动化工具开发提供基础。
bruce_xiaowei
30 8
站大爷
|
13天前
|
安全 API 文件存储
Yagmail邮件发送库：如何用Python实现自动化邮件营销？
本文详细介绍了如何使用Yagmail库实现自动化邮件营销。Yagmail是一个简洁强大的Python库，能简化邮件发送流程，支持文本、HTML邮件及附件发送，适用于数字营销场景。文章涵盖了Yagmail的基本使用、高级功能、案例分析及最佳实践，帮助读者轻松上手。
站大爷
25 4
小王老师呀
|
16天前
|
XML 数据采集 数据格式
Python 爬虫必备杀器，xpath 解析 HTML
【11月更文挑战第17天】XPath 是一种用于在 XML 和 HTML 文档中定位节点的语言，通过路径表达式选取节点或节点集。它不仅适用于 XML，也广泛应用于 HTML 解析。基本语法包括标签名、属性、层级关系等的选择，如 `//p` 选择所有段落标签，`//a[@href=&#39;example.com&#39;]` 选择特定链接。在 Python 中，常用 lxml 库结合 XPath 进行网页数据抓取，支持高效解析与复杂信息提取。高级技巧涵盖轴的使用和函数应用，如 `contains()` 用于模糊匹配。
小王老师呀
48 7
武子康
|
2月前
|
Python
pip批量安装Python库 requirement.txt 离线环境无互联网环境下pip安装Python库
pip批量安装Python库 requirement.txt 离线环境无互联网环境下pip安装Python库
武子康
145 3
东方睿赢
|
6月前
|
开发工具 git Python
安装和使用`libnum`是一个用于数字理论函数的Python库
【6月更文挑战第19天】`libnum`是Python的数字理论函数库。安装可通过`git clone`，进入目录后运行`python setup.py install`，也可用`pip install libnum`。示例：使用`int_to_hex`将十进制数42转换为十六进制字符串&#39;2a&#39;。注意，信息可能已过时，应查最新文档以确保准确性。如遇问题，参考GitHub仓库或寻求社区帮助。
东方睿赢
127 1

热门文章

最新文章

  • 1
    C#操作IIS完整解析
  • 2
    uboot移植之uboot命令体系解析
  • 3
    Fresco源码解析 - 创建一个ImagePipeline（一）
  • 4
    在注册表中找到你的dns服务器
  • 5
    Spring Boot 2.0 新特性（一）：配置绑定 2.0 全解析
  • 6
    企业域名更换操作系列3：增加新域名所需DNS相关记录
  • 7
    Android Service完全解析，关于服务你所需知道的一切(下)
  • 8
    SpringMVC介绍之视图解析器ViewResolver
  • 9
    C# 面试题解析-请遍历页面上所有的TextBox控件并给它赋值为string.Empty
  • 10
    Linux--IP基础知识 网关设定 dns服务
  • 1
    如何使用Python的pyodps库来进行跨项目空间重命名表名？
    54
  • 2
    Python大佬耗费13年，始成400页《Python工匠》手册
    57
  • 3
    MaxCompute产品使用合集之PyODPS Python类的开发如何用MC的资源
    56
  • 4
    python并发执行request请求
    85
  • 5
    Python函数式编程入门窥探
    43
  • 6
    【Azure Function】发布 Python Function 到 Azure 成功，但是无法显示Function列表
    57
  • 7
    MaxCompute产品使用合集之是否可以将5个资源包统一写到同一个python UDF脚本
    30
  • 8
    深入Scikit-learn：掌握Python最强大的机器学习库
    164
  • 9
    告别低效！Python并查集：数据结构界的超级英雄，拯救你的编程人生！
    54
  • 10
    Python并查集大揭秘：让你在算法界呼风唤雨，秒杀一切复杂场景！
    51

    • 相关课程

    更多
  • 第八届大学生创新创业大赛阿里命题IoT赛题解析
  • 云计算工程师解析与实战-网络专家篇（体验版）
  • 深入解析Docker容器化技术
  • Java面试疑难点解析 - 面试技巧及语言基础
  • Java面试疑难点解析 - Java Web开发
  • Java面试疑难点解析 - 系统架构及项目设计

    • 相关电子书

    更多
  • 神龙云服务器产品及技术深度解析
  • 弹性创造价值：基于ECS的最佳性价比实践解析
  • 又快又稳：阿里云下一代虚拟交换机解析

    • 相关实验场景

    更多
  • 用Python画圣诞树
  • 通过云拨测对指定服务器进行Ping/DNS监测
  • Python选择及循环结构
  • 云端Python及基本操作
  • Python网络通信程序典型应用
  • Python新手入门（Anolis OS）

    • 推荐镜像

    更多
  • DNS
  • python-release
  • kali-security
    • 下一篇
    阿里云对象存储OSS收费标准：存储、流量和请求等多个计费项详解