快捷導(dǎo)航

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

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

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

1. 端口掃描的基本原理

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

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

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

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

2.1 安裝必要的庫

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

pip install scapy

2.2 編寫端口掃描腳本

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

from scapy.all import *
import ipaddress
 
def port_scan(ip, ports):
    """
    對(duì)指定IP地址的指定端口進(jìn)行掃描
    :param ip: 目標(biāo)IP地址
    :param ports: 需要掃描的端口號(hào)列表
    :return: 打印開放的端口
    """
    open_ports = []
    for port in ports:
        # 構(gòu)造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包復(fù)位連接
                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  # 端口關(guān)閉
    return open_ports
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 替換為目標(biāo)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 腳本解釋

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

3. 運(yùn)行與測試

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

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

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

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

import socket
from concurrent.futures import ThreadPoolExecutor
 
def scan_port(ip, port):
    try:
        # 創(chuàng)建一個(gè) socket 對(duì)象
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.settimeout(1)  # 設(shè)置超時(shí)時(shí)間
            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 來并行掃描多個(gè)端口
    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ù)：這個(gè)函數(shù)負(fù)責(zé)檢查單個(gè)端口是否開放。它創(chuàng)建一個(gè) ??socket?? 對(duì)象，并嘗試連接到指定的 IP 地址和端口。如果連接成功，則端口開放；否則，端口關(guān)閉。
??main??? 函數(shù)：這是程序的主入口。用戶輸入目標(biāo) IP 地址和要掃描的端口范圍。然后使用 ??ThreadPoolExecutor?? 來并行執(zhí)行多個(gè) ??scan_port?? 任務(wù)，以提高掃描速度。
??ThreadPoolExecutor??：這是一個(gè)多線程池，可以同時(shí)執(zhí)行多個(gè)任務(wù)。這里設(shè)置的最大工作線程數(shù)為 100，可以根據(jù)實(shí)際情況調(diào)整。

注意事項(xiàng)：

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

進(jìn)一步優(yōu)化：

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

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

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

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

import socket
import threading
 
def scan_port(ip, port):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(1)  # 設(shè)置超時(shí)時(shí)間
        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"  # 目標(biāo)IP地址
    start_port = 1  # 開始端口
    end_port = 1024  # 結(jié)束端口
    main(target_ip, start_port, end_port)

2. 使用 ??asyncio?? 進(jìn)行異步掃描

??asyncio?? 是 Python 的異步 I/O 框架，可以顯著提高端口掃描的速度，因?yàn)樗试S在一個(gè)線程中并發(fā)執(zhí)行多個(gè)任務(wù)。

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"  # 目標(biāo)IP地址
    start_port = 1  # 開始端口
    end_port = 1024  # 結(jié)束端口
    asyncio.run(main(target_ip, start_port, end_port))

3. 使用 ??scapy?? 進(jìn)行更復(fù)雜的掃描

??scapy?? 是一個(gè)強(qiáng)大的網(wǎng)絡(luò)工具包，可以用于發(fā)送和接收網(wǎng)絡(luò)數(shù)據(jù)包，包括進(jìn)行端口掃描。使用 ??scapy?? 可以實(shí)現(xiàn)更復(fù)雜的掃描策略，如 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 包關(guān)閉連接
        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"  # 目標(biāo)IP地址
    start_port = 1  # 開始端口
    end_port = 1024  # 結(jié)束端口
    main(target_ip, start_port, end_port)