[网络协议]python多线程实现Nmap批量扫描tcp端口是否开放

import threading
import IPy
class MyThread(object):
    def __init__(self, func_list=None):
        # 所有线程函数的返回值汇总，如果最后为0，说明全部成功
        self.dev_list = []
        self.func_list = func_list
        self.threads = []

    def trace_func(self, func, *args, **kwargs):
        """
        @note:替代profile_func，新的跟踪线程返回值的函数，对真正执行的线程函数包一次函数，以获取返回值
        """
        dev_list=[]
        ret = func(*args, **kwargs)
        dev_list.append(ret)
        self.dev_list.extend(dev_list)



    def set_thread_func_list(self, func_list):
        """
        @note: func_list是一个list，每个元素是一个dict，有func和args两个参数
        """
        self.func_list = func_list

    def start(self):
        """
        @note: 启动多线程执行，并阻塞到结束
        """
        self.threads = []
        self.dev_list = []
        for func_dict in self.func_list:
            if func_dict["args"]:
                new_arg_list = []
                new_arg_list.append(func_dict["func"])
                for arg in func_dict["args"]:
                    new_arg_list.append(arg)
                new_arg_tuple = tuple(new_arg_list)
                t = threading.Thread(target=self.trace_func, args=new_arg_tuple)
            else:
                t = threading.Thread(target=self.trace_func, args=(func_dict["func"],))
            self.threads.append(t)

        for thread_obj in self.threads:
            thread_obj.start()

        for thread_obj in self.threads:
            thread_obj.join()

    def ret_value(self):
        """
        @note: 所有线程函数的返回值之和，如果为0那么表示所有函数执行成功
        """
        return self.dev_list
def port_scan(hosts,ports):
#扫描IP地址的端口是否方通的主函数
    #hosts:string for hosts as nmap use it 'scanme.nmap.org' or '198.116.0-255.1-127' or '216.163.128.20/20'
    #ports:string for ports as nmap use it '22,53,110,143-4564'
    nm = nmap.PortScannerYield()
    res = nm.scan(hosts, ports=ports, arguments="-sS")
    for host_res in res:
        each_ip_port_dict={}
        for k,v in host_res[1]["scan"].items():
            data=v['tcp']
            #获取扫描的TCP端口的返回信息
            open_port_list=[]
            for port ,port_state in data.items():
                #对探测的每个端口进行判断，如果端口是放开的，则state为open
                if port_state["state"]=="open":
                    open_port_list.append(port)
            each_ip_port_dict[k]=open_port_list
            #获取IP地址和可达端口的字典

        return each_ip_port_dict
        #print(host_res[0],data)
def IsIPv4(ip_addr):
    #判断输入是否是合法IO
    try:
        IPy.IP(ip_addr,make_net=True).version()
        return 1
    except Exception as e:
        return 0

if __name__ == '__main__':
    ip_net_flag=False
    while not ip_net_flag:
        ip_net=input("请输入IP地址段（eg:192.168.1.0/24）：")
        if IsIPv4(ip_net)==1:
            ip_net_flag=True
        else:
            print("IP地址输入不合法！")
    ports=input("请输入检测端口范围(eg:'22,53,110,143-4564')：")
    ip_net=IPy.IP(ip_net,make_net=True)
    ip_list=[]
    #将获取到的IP地址都放入到此列表中
    for ip1 in ip_net:
        ip_list.append(str(ip1))
    threads = []  # 多线程
    mt = MyThread()
    g_func_list = []
    for ip in ip_list:
        g_func_list.append({"func": port_scan, "args": (ip, ports)})
    mt.set_thread_func_list(g_func_list)
    mt.start()
    ip_port_dict = mt.dev_list
    print(ip_port_dict)

结果输出：
![在这里插入图片描述](https://img-blog.csdnimg.cn/4100b9eddc004570833b461e36587a51.png)