python端口扫描代码源码

个人用python写的扫描主机端口的小程序，可用于扫描一台指定主机开放且正在监听哪些端口，请注意，在一台主机上netstat -an看到的正在listening的端口未必是开放的，还要看主机的防火墙以及路由器的防火墙的设置。脚本中可以配置的参数说明，可以自行修改: gd.Host = "127.0.0.1" #要被扫描的主机IP gd.ThreadSize = 1000 #并行线程数，建议值1000-3000 gd.ExitNum = 10 #扫描到多少个开放端口后退出 gd.Timer = 5 #超时时间，内网设1，公网不超过5 gd.PortList_1 = range(1,2000) #端口范围 gd.PortList_2 = [3389,6000] #特殊端口列表 脚本开始运行后会打印扫描到的主机正在监听且开放的端口，按ctrl+c随时结束扫描。

#!/usr/bin/python
import ossystimesocketthreadingcopy
 
class cGlobalData:
    pass
gd=cGlobalData（）

gd.Host = “127.0.0.1“
gd.ThreadSize = 1000
gd.ExitNum = 100
gd.Timer = 5
gd.PortList_1 = range（30000）
gd.PortList_2 = [33896000]

gd.OpenPorts = []
gd.NewPorts = []
gd.PrintLock = threading.Lock（）
gd.UpdateLock = threading.Lock（）
gd.OpenLock = threading.Lock（）
gd.ThreadLock = threading.Lock（）
gd.ExitLock = threading.Lock（）
gd.ScanPorts = []+gd.PortList_1 + gd.PortList_2
gd.ScanedNum = 0
gd.Exit = False
gd.ThreadNum=0
gd.GroupSize=int（（len（gd.ScanPorts））/gd.ThreadSize）


class cScanThread（threading.Thread）:
    def __init__（selfaIdxaLenaId）:
        threading.Thread.__init__（self）
	self.vId = aId
        self.vIdx=aIdx
	self.vLen=aLen
        return
    def run（self）:
        global gd
	gd.PrintLock.acquire（）
	#print “thread“self.vId“scan from“self.vIdx“to“self.vIdx+self.vLen
	gd.PrintLock.release（）
	for i in range（self.vIdxself.vIdx+self.vLen）:
	    gd.ExitLock.acquire（）
	    vExit=gd.Exit
	    gd.ExitLock.release（）
	    if（vExit）:
		gd.ThreadLock.acquire（）
		gd.ThreadNum = gd.ThreadNum -1
		gd.ThreadLock.release（）
		gd.PrintLock.acquire（）
		#print “thread“self.vId“exit because ports enough“
		gd.PrintLock.release（）
		return
	    sk = socket.socket（socket.AF_INET socket.SOCK_STREAM）
	    sk.settimeout（gd.Timer）
	    try:
		sk.connect（（gd.Hostgd.ScanPorts[i]））
	    except Exception:
		gd.UpdateLock.acquire（）
		gd.ScanedNum = gd.ScanedNum + 1
		gd.UpdateLock.release（）
		sk.close（）
		continue

	    sk.close（）
	    gd.PrintLock.acquire（）
	    print ‘******************‘+str（gd.ScanPorts[i]）+‘***‘
	    gd.PrintLock.release（）
	    gd.OpenLock.acquire（）
	    gd.NewPorts.append（gd.ScanPorts[i]）
	    gd.OpenLock.release（）

	gd.ThreadLock.acquire（）
	gd.ThreadNum = gd.ThreadNum -1
	gd.ThreadLock.release（）
	gd.PrintLock.acquire（）
	#print “thread“self.vId“scan over“
	gd.PrintLock.release（）
	return


def main（）:
    global gd

    try:
	pass
	for i in range（gd.ThreadSize）:
	    gd.ThreadLock.acquire（）
	    gd.ThreadNum = gd.ThreadNum + 1
	    vTrd = cScanThread（i*gd.GroupSizegd.GroupSizegd.ThreadNum）
	    gd.ThreadLock.release（）
	    vTrd.start（）
	if（len（gd.ScanPorts）-（gd.ThreadSize*gd.GroupSize） > 0）:
	    gd.ThreadLock.acquire（）
	    gd.ThreadNum = gd.ThreadNum + 1
	    vTrd = cScanThread（（gd.ThreadSize-1）*gd.GroupSizelen（gd.ScanPorts）-（gd.ThreadSize*gd.GroupSize）gd.ThreadNum）
	    gd.ThreadLock.release（）
	    vTrd.start（）

	gd.ThreadLock.acquire（）
	vThreadNum = gd.ThreadNum
	gd.ThreadLock.release（）
	gd.PrintLock.acquire（）
	#print vThreadNum“threads“
	#print gd.GroupSize“GroupSize“
	gd.PrintLock.release（）
	

	vOldNum=0
	vOpenNum=0
	while（True）:
	    time.sleep（2）
	    gd.ThreadLock.acquire（）
	    vThreadNum = gd.ThreadNum
	    gd.ThreadLock.release（）
	    gd.UpdateLock.acquire（）
	    vScanedNum = gd.ScanedNum
	    gd.UpdateLock.release（）
	    if（vOldNum == vScanedNum）:
		if（vThreadNum <= 0）:
		    break
		continue
	    print vScaned

 属性            大小     日期    时间   名称
----------- ---------  ---------- -----  ----

     文件       3720  2010-03-30 16:27  scanport.py

----------- ---------  ---------- -----  ----

                 3720                    1