Python實(shí)現(xiàn)的多線程端口掃描工具分享

更新時(shí)間：2015年01月21日 11:10:06 投稿：junjie

這篇文章主要介紹了Python實(shí)現(xiàn)的多線程端口掃描工具分享,工具實(shí)現(xiàn)了掃單IP和掃IP段功能,本文給出運(yùn)行效果和實(shí)現(xiàn)源碼,需要的朋友可以參考下

昨晚今晚寫了兩晚，總算把Py Port Scanner 寫完了，姑且稱之為0.1版本，算是一個(gè)Python多線程端口掃描工具。

水平有限，實(shí)話中間有一些困惑和不解的地方，代碼可能也寫的比較亂。有些問題并未找到很好的解決方法，還望大家諒解。速度大家自己試驗(yàn)，我感覺還行。

送上效果圖兩份，分別是掃單IP和掃IP段：

源碼：

復(fù)制代碼代碼如下:

# -*- coding: utf-8 -*-
__author__ = 'Phtih0n'
import threading, socket, sys, cmd, os, Queue

#掃描常用端口
PortList = [21, 22, 23, 25, 80, 135, 137, 139, 445, 1433, 1502, 3306, 3389, 8080, 9015]
#得到一個(gè)隊(duì)列
def GetQueue(list):
    PortQueue = Queue.Queue(65535)
    for p in list:
        PortQueue.put(p)
    return PortQueue

#單IP掃描線程個(gè)數(shù)
nThread = 20
#線程鎖
lock = threading.Lock()
#超時(shí)時(shí)間
Timeout = 3.0
#打開的端口列表
OpenPort = []

class ScanThread(threading.Thread):
    def __init__(self, scanIP):
        threading.Thread.__init__(self)
        self.IP = scanIP

    def Ping(self, Port):
        global OpenPort, lock, Timeout
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(Timeout)
        address = (self.IP, Port)
        try:
            sock.connect(address)
        except:
            sock.close()
            return False
        sock.close()
        OpenPort.append(Port)
        if lock.acquire():
            print "IP:%s Port:%d" % (self.IP, Port)
            lock.release()
        return True

class ScanThreadSingle(ScanThread):
    def __init__(self, scanIP, SingleQueue):
        ScanThread.__init__(self, scanIP)
        self.SingleQueue = SingleQueue

    def run(self):
        while not self.SingleQueue.empty():
            p = self.SingleQueue.get()
            self.Ping(p)

class ScanThreadMulti(ScanThread):
    def __init__(self, scanIP, PortList):
        ScanThread.__init__(self, scanIP)
        self.List = PortList[:]

    def run(self):
        for p in self.List:
            self.Ping(p)

class Shell(cmd.Cmd):
    u'''Py Port Scanner 0.1 使用說明：
    port [port..] 設(shè)置掃描的端口，用逗號(hào)分隔。
        默認(rèn)：21, 22, 23, 25, 80, 135, 137, 139, 445, 1433, 1502, 3306, 3389, 8080, 9015
        example：port 21,23,25
        example: port 1000..2000
        example: port 80,443,1000..1500
    scan [IP] 掃描某一IP地址
        example: scan 192.168.1.5
    search [IP begin]-[IP end] 掃描某一IP段
        example: search 192.168.1.1-192.168.1.100
    time [timeout] 設(shè)置超時(shí)時(shí)間，默認(rèn)為3秒
        example: time 5
    cls 清楚屏幕內(nèi)容
    listport 打印端口列表
    help 打開本幫助
        '''
    def __init__(self):
        cmd.Cmd.__init__(self)
        reload(sys)
        sys.setdefaultencoding('utf-8')
        self.prompt = "Port Scan >>"
        self.intro = "Py Port Scanner 0.1"

def do_EOF(self, line):
return True

def do_help(self, line):
print self.__doc__

    #設(shè)置端口
    def do_port(self, line):
        global PortList
        PortList = []
        ListTmp = line.split(',')
        for port in ListTmp:
            if port.find("..") < 0:
                if not port.isdigit():
                    print "輸入錯(cuò)誤"
                    return False
                PortList.append(int(port))
            else:
                RangeLst = port.split("..")
                if not (RangeLst[0].isdigit() and RangeLst[1].isdigit()):
                    raise ValueError
                    exit()
                for i in range(int(RangeLst[0]), int(RangeLst[1])):
                    PortList.append(i)

    def do_scan(self, line):
        global nThread, PortList
        ThreadList = []
        strIP = line
        SingleQueue = GetQueue(PortList)
        for i in range(0, nThread):
            t = ScanThreadSingle(strIP, SingleQueue)
            ThreadList.append(t)
        for t in ThreadList:
            t.start()
        for t in ThreadList:
            t.join()

    def do_search(self, line):
        global nThread, PortList
        ThreadList = []
        (BeginIP, EndIP) = line.split("-")
        try:
            socket.inet_aton(BeginIP)
            socket.inet_aton(EndIP)
        except:
            print "輸入錯(cuò)誤"
            return
        IPRange = BeginIP[0:BeginIP.rfind('.')]
        begin = BeginIP[BeginIP.rfind('.') + 1:]
        end = EndIP[EndIP.rfind('.') + 1:]
        for i in range(int(begin), int(end)):
            strIP = "%s.%s" % (IPRange, i)
            t = ScanThreadMulti(strIP, PortList)
            ThreadList.append(t)
        for t in ThreadList:
            t.start()
        for t in ThreadList:
            t.join()

    def do_listport(self, line):
        global PortList
        for p in PortList:
            print p,
        print '\n'

    def do_time(self, line):
        global Timeout
        try:
            Timeout = float(line)
        except:
            print u"參數(shù)錯(cuò)誤"

def do_cls(self, line):
os.system("cls")

if '__main__' == __name__:
    try:
        os.system("cls")
        shell = Shell()
        shell.cmdloop()
    except:
        exit()

您可能感興趣的文章: