利用Python進行端口掃描

Python端口掃描

　　在Linux中判斷一台主機是否可達，可以使用ping命令，而判斷端口是否打開，可以使用telnet命令，但是telnet命令沒有超時時間的參數，使用起來不是很方便，那么可以利用Python來完成一個端口掃描的功能

#!/usr/bin/env python

import socket

def get_ip_status(ip,port):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        server.connect((ip,port))
        print('{0} port {1} is open'.format(ip, port))
    except Exception as err:
        print('{0} port {1} is not open'.format(ip,port))
    finally:
        server.close()

if __name__ == '__main__':
    host = '10.0.0.11'
    for port in range(20,100):
        get_ip_status(host,port)

　上面使用的是python的socket模塊完成的端口檢測， 其實python的內置模塊telnetlib也可以完成端口檢測任務

import telnetlib

def get_ip_status(ip,port):
    server = telnetlib.Telnet()      # 創建一個Telnet對象
    try:
        server.open(ip,port)         # 利用Telnet對象的open方法進行tcp鏈接
        print('{0} port {1} is open'.format(ip, port))
    except Exception as err:
        print('{0} port {1} is not open'.format(ip,port))
    finally:
        server.close()

if __name__ == '__main__':
    host = '10.0.0.11'
    for port in range(20,100):
        get_ip_status(host,port)

　　當然上面這兩種方式都是串行執行的，這在多IP多端口的情況下是非常慢得，所以引入多線程threading模塊。

#!/usr/bin/env python

import telnetlib
import threading

def get_ip_status(ip,port):
    server = telnetlib.Telnet()
    try:
        server.open(ip,port)
        print('{0} port {1} is open'.format(ip, port))
    except Exception as err:
        print('{0} port {1} is not open'.format(ip,port))
    finally:
        server.close()

if __name__ == '__main__':
    host = '10.0.0.11'
    threads = []
    for port in range(20,100):
        t = threading.Thread(target=get_ip_status,args=(host,port))
        t.start()
        threads.append(t)

    for t in threads:
        t.join()

　　前面的例子中我們的程序為每個ip的每個端口創建一個線程，在IP和Port較多的時候，會暴露出各種問題(比如頻繁的上下文切換),因此，我們需要限制進程數量，那么可以利用Queue模塊。

#!/usr/bin/env python

import telnetlib
import threading
import queue

def get_ip_status(ip):
    server = telnetlib.Telnet()
    for port in range(20,100):
        try:
            server.open(ip,port)
            print('{0} port {1} is open'.format(ip, port))
        except Exception as err:
            print('{0} port {1} is not open'.format(ip,port))
        finally:
            server.close()

def check_open(q):
    try:
        while True:
            ip = q.get_nowait()
            get_ip_status(ip)
    except queue.Empty as e:
        pass

if __name__ == '__main__':
    host = ['10.0.0.10','10.0.0.11','10.0.0.12']     # 這里模擬多IP地址的情況，也可以從文件中讀取IP——list
    q = queue.Queue()
    for ip in host:
        q.put(ip)
    threads = []
    for i in range(10):
        t = threading.Thread(target=check_open,args=(q,))
        t.start()
        threads.append(t)

    for t in threads:
        t.join()

　　這里使用了Queue，那么就會引出生產者和消費者模型，生產者只需要把信息存入到隊列中，消費者消費時只需要看隊列中有沒有，這樣極大程度了解耦了我們的程序。