使用Python實現(xiàn)高效的端口掃描器

更新時間：2025年01月28日 15:20:30 作者：牛肉胡辣湯

在網絡安全領域,端口掃描是一項基本而重要的技能,通過端口掃描,可以發(fā)現(xiàn)目標主機上開放的服務和端口,這對于安全評估、滲透測試等有著不可忽視的作用,本文將介紹如何使用Python來實現(xiàn)一個高效且簡單的端口掃描器,需要的朋友可以參考下

1. 端口掃描的基本原理

端口掃描的基本原理是向目標主機的指定端口發(fā)送數(shù)據(jù)包，并監(jiān)聽是否有來自該端口的響應。根據(jù)響應的不同，可以判斷該端口的狀態(tài)（如開放、關閉或過濾）。

常見的端口掃描類型包括：

TCP SYN掃描：發(fā)送SYN包，如果收到SYN-ACK，則端口開放；如果收到RST，則端口關閉。
TCP Connect掃描：完成三次握手，如果成功則端口開放。
UDP掃描：發(fā)送UDP包，如果收到ICMP錯誤消息，則端口關閉；如果沒有響應，則可能開放或過濾。

2. 使用Python實現(xiàn)端口掃描

2.1 安裝必要的庫

首先，我們需要安裝??scapy??庫，這是一個強大的網絡工具庫，支持創(chuàng)建、發(fā)送、捕獲和解析網絡數(shù)據(jù)包。

pip install scapy

2.2 編寫端口掃描腳本

下面是一個使用??scapy??實現(xiàn)的簡單端口掃描器示例：

from scapy.all import *
import ipaddress
 
def port_scan(ip, ports):
    """
    對指定IP地址的指定端口進行掃描
    :param ip: 目標IP地址
    :param ports: 需要掃描的端口號列表
    :return: 打印開放的端口
    """
    open_ports = []
    for port in ports:
        # 構造SYN包
        packet = IP(dst=ip)/TCP(dport=port, flags="S")
        response = sr1(packet, timeout=1, verbose=0)
        
        if response is not None and TCP in response:
            if response[TCP].flags == 0x12:  # 如果收到SYN-ACK
                # 發(fā)送RST包復位連接
                send_rst = sr(IP(dst=ip)/TCP(dport=port, flags="R"), timeout=1, verbose=0)
                open_ports.append(port)
            elif response[TCP].flags == 0x14:  # 如果收到RST
                pass  # 端口關閉
    return open_ports
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 替換為目標IP
    target_ports = [22, 80, 443, 8080]  # 指定需要掃描的端口
    open_ports = port_scan(target_ip, target_ports)
    
    if open_ports:
        print(f"Open ports on {target_ip}: {open_ports}")
    else:
        print(f"No open ports found on {target_ip}")

2.3 腳本解釋

構造SYN包：使用??scapy??構建一個TCP SYN包，目標是目標IP地址和指定端口。
發(fā)送并接收響應：使用??sr1??函數(shù)發(fā)送數(shù)據(jù)包并等待響應，超時時間為1秒。
分析響應：如果收到SYN-ACK響應，說明端口開放；如果收到RST響應，說明端口關閉。
復位連接：對于開放的端口，發(fā)送一個RST包以復位連接，避免建立完整的TCP連接。

3. 運行與測試

確保你有權限發(fā)送網絡數(shù)據(jù)包（通常需要root權限）。運行上述腳本后，它將輸出目標主機上開放的端口列表。

本文介紹了如何使用Python和??scapy??庫實現(xiàn)一個簡單的端口掃描器。雖然這個掃描器功能較為基礎，但它提供了一個良好的起點，可以根據(jù)實際需求進一步擴展和優(yōu)化。例如，可以添加多線程支持以提高掃描速度，或者實現(xiàn)更復雜的掃描策略來規(guī)避防火墻檢測。

這篇文章詳細介紹了如何使用Python和??scapy???庫實現(xiàn)一個簡單的端口掃描器，適合初學者學習和實踐。端口掃描是網絡安全領域中常用的技術之一，用于檢測目標主機上開放的服務和端口。Python 提供了多種庫來實現(xiàn)這一功能，其中 ??socket?? 庫是最基礎也是最靈活的選擇之一。為了提高效率，可以使用多線程或異步 I/O 技術。

下面是一個使用 ??socket?? 和 ??concurrent.futures??（多線程）實現(xiàn)的高效端口掃描器示例：

import socket
from concurrent.futures import ThreadPoolExecutor
 
def scan_port(ip, port):
    try:
        # 創(chuàng)建一個 socket 對象
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.settimeout(1)  # 設置超時時間
            result = sock.connect_ex((ip, port))
            if result == 0:
                print(f"Port {port} is open")
            else:
                print(f"Port {port} is closed")
    except socket.error as e:
        print(f"Socket error: {e}")
 
def main():
    target_ip = input("Enter the target IP address: ")
    start_port = int(input("Enter the start port: "))
    end_port = int(input("Enter the end port: "))
 
    # 使用 ThreadPoolExecutor 來并行掃描多個端口
    with ThreadPoolExecutor(max_workers=100) as executor:
        for port in range(start_port, end_port + 1):
            executor.submit(scan_port, target_ip, port)
 
if __name__ == "__main__":
    main()

代碼說明：

??scan_port??? 函數(shù)：這個函數(shù)負責檢查單個端口是否開放。它創(chuàng)建一個 ??socket?? 對象，并嘗試連接到指定的 IP 地址和端口。如果連接成功，則端口開放；否則，端口關閉。
??main??? 函數(shù)：這是程序的主入口。用戶輸入目標 IP 地址和要掃描的端口范圍。然后使用 ??ThreadPoolExecutor?? 來并行執(zhí)行多個 ??scan_port?? 任務，以提高掃描速度。
??ThreadPoolExecutor??：這是一個多線程池，可以同時執(zhí)行多個任務。這里設置的最大工作線程數(shù)為 100，可以根據(jù)實際情況調整。

注意事項：

性能與資源：多線程可以顯著提高掃描速度，但過多的線程可能會消耗大量系統(tǒng)資源，甚至導致目標系統(tǒng)拒絕服務（DoS）。因此，需要根據(jù)實際情況調整 ??max_workers?? 的值。
法律與道德：未經授權的端口掃描可能違反法律法規(guī)。在進行端口掃描之前，請確保你有合法的權限。

進一步優(yōu)化：

異步 I/O：可以使用 ??asyncio?? 和 ??aiohttp?? 等庫來實現(xiàn)更高效的異步端口掃描。
錯誤處理：增加更多的錯誤處理邏輯，以應對網絡不穩(wěn)定等情況。

希望這個示例對你有所幫助！如果你有任何問題或需要進一步的幫助，請告訴我。在Python中實現(xiàn)高效的端口掃描可以通過多種方式完成，其中最常見的是使用多線程或多進程來提高掃描速度。這里將介紹一種使用??socket??和??threading??模塊的簡單方法，以及更高級的方法，如使用??asyncio??進行異步編程。

1. 使用 ??socket?? 和 ??threading??

這種方法的基本思路是為每個要掃描的端口創(chuàng)建一個線程，每個線程負責檢查該端口是否開放。

import socket
import threading
 
def scan_port(ip, port):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(1)  # 設置超時時間
        result = sock.connect_ex((ip, port))
        if result == 0:
            print(f"Port {port} is open")
        sock.close()
    except Exception as e:
        print(f"Error scanning port {port}: {e}")
 
def main(ip, start_port, end_port):
    threads = []
    for port in range(start_port, end_port + 1):
        thread = threading.Thread(target=scan_port, args=(ip, port))
        threads.append(thread)
        thread.start()
    
    for thread in threads:
        thread.join()
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 目標IP地址
    start_port = 1  # 開始端口
    end_port = 1024  # 結束端口
    main(target_ip, start_port, end_port)

2. 使用 ??asyncio?? 進行異步掃描

??asyncio?? 是 Python 的異步 I/O 框架，可以顯著提高端口掃描的速度，因為它允許在一個線程中并發(fā)執(zhí)行多個任務。

import asyncio
import socket
 
async def scan_port(ip, port):
    try:
        conn = asyncio.open_connection(ip, port)
        reader, writer = await asyncio.wait_for(conn, timeout=1)
        print(f"Port {port} is open")
        writer.close()
        await writer.wait_closed()
    except (asyncio.TimeoutError, ConnectionRefusedError):
        pass
 
async def main(ip, start_port, end_port):
    tasks = []
    for port in range(start_port, end_port + 1):
        task = asyncio.create_task(scan_port(ip, port))
        tasks.append(task)
    
    await asyncio.gather(*tasks)
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 目標IP地址
    start_port = 1  # 開始端口
    end_port = 1024  # 結束端口
    asyncio.run(main(target_ip, start_port, end_port))

3. 使用 ??scapy?? 進行更復雜的掃描

??scapy?? 是一個強大的網絡工具包，可以用于發(fā)送和接收網絡數(shù)據(jù)包，包括進行端口掃描。使用 ??scapy?? 可以實現(xiàn)更復雜的掃描策略，如 SYN 掃描等。

from scapy.all import sr1, IP, TCP
 
def scan_port(ip, port):
    src_port = 1025  # 源端口
    syn_packet = IP(dst=ip) / TCP(sport=src_port, dport=port, flags='S')
    response = sr1(syn_packet, timeout=1, verbose=0)
    
    if response and response.haslayer(TCP) and response.getlayer(TCP).flags & 0x12:  # SYN-ACK
        print(f"Port {port} is open")
        # 發(fā)送 RST 包關閉連接
        rst_packet = IP(dst=ip) / TCP(sport=src_port, dport=port, flags='R')
        send(rst_packet, verbose=0)
 
def main(ip, start_port, end_port):
    for port in range(start_port, end_port + 1):
        scan_port(ip, port)
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 目標IP地址
    start_port = 1  # 開始端口
    end_port = 1024  # 結束端口
    main(target_ip, start_port, end_port)