用Python编写一个高效的端口扫描器的方法

下面是用Python编写高效的端口扫描器的攻略：

1. 确定扫描范围

端口扫描器需要扫描哪些主机和端口号，一般需要提供两个参数：主机列表和端口范围。主机列表可以是一个IP地址列表或者一个网段；端口范围一般是一个起始端口和一个结束端口。在Python中，可以用ipaddress库来处理IP地址和网段，可以用range函数来处理端口范围。

示例一：扫描某个IP地址的所有端口

import socket

target_host = "192.168.1.100"
target_ports = range(1, 65536)

for port in target_ports:
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.settimeout(0.1)
    result = client.connect_ex((target_host, port))
    if result == 0:
        print(f"Port {port} is open")
    client.close()

示例二：扫描某个网段的所有主机的某个端口

import socket
import ipaddress

subnet = "192.168.1.0/24"
target_port = 8888

network = ipaddress.IPv4Network(subnet)

for ip in network.hosts():
    ip_str = str(ip)
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.settimeout(0.1)
    result = client.connect_ex((ip_str, target_port))
    if result == 0:
        print(f"Port {target_port} is open on {ip_str}")
    client.close()

2. 并发扫描

如果扫描的范围比较大，一条线程的效率会比较低，需要使用多线程或异步IO来提高并发度。Python中有多种实现并发扫描的方式，包括多线程、多进程、协程、异步IO等。

示例三：使用多线程实现并发扫描

import socket
import threading

target_host = "192.168.1.1"
target_ports = range(1, 65536)

def scan_port(port):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.settimeout(0.1)
    result = client.connect_ex((target_host, port))
    if result == 0:
        print(f"Port {port} is open")
    client.close()

for port in target_ports:
    thread = threading.Thread(target=scan_port, args=(port,))
    thread.start()

示例四：使用异步IO实现并发扫描

import socket
import asyncio

target_host = "192.168.1.1"
target_ports = range(1, 65536)

async def scan_port(port):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.settimeout(0.1)
    try:
        await asyncio.wait_for(
            loop.sock_connect(client, (target_host, port)),
            timeout=0.1,
        )
        print(f"Port {port} is open")
    except Exception:
        pass
    finally:
        client.close()

loop = asyncio.get_event_loop()
tasks = [loop.create_task(scan_port(port)) for port in target_ports]
loop.run_until_complete(asyncio.wait(tasks))