使用Python和并发编程实现高效端口扫描

实现一个高效的端口扫描器可以使用Python的socket库和concurrent.futures模块来实现并发扫描。以下是一个简单的端口扫描器示例：

import socket
from concurrent.futures import ThreadPoolExecutor
def scan_port(host, port):
    try:
        # 创建一个socket对象
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # 设置超时时间
        sock.settimeout(1)
        # 尝试连接目标端口
        result = sock.connect_ex((host, port))
        if result == 0:
            print(f"Port {port} is open")
        # 关闭socket连接
        sock.close()
    except Exception as e:
        print(f"Error scanning port {port}: {e}")
def port_scan(host, start_port, end_port, max_threads=100):
    print(f"Scanning {host} from port {start_port} to {end_port}...")
    with ThreadPoolExecutor(max_workers=max_threads) as executor:
        # 使用线程池并发扫描端口
        for port in range(start_port, end_port + 1):
            executor.submit(scan_port, host, port)
if __name__ == "__main__":
    host = input("Enter the host to scan: ")
    start_port = int(input("Enter the start port: "))
    end_port = int(input("Enter the end port: "))
    max_threads = int(input("Enter the number of threads: "))
    port_scan(host, start_port, end_port, max_threads)
import socket
from concurrent.futures import ThreadPoolExecutor

def scan_port(host, port):
    try:
        # 创建一个socket对象
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # 设置超时时间
        sock.settimeout(1)
        # 尝试连接目标端口
        result = sock.connect_ex((host, port))
        if result == 0:
            print(f"Port {port} is open")
        # 关闭socket连接
        sock.close()
    except Exception as e:
        print(f"Error scanning port {port}: {e}")

def port_scan(host, start_port, end_port, max_threads=100):
    print(f"Scanning {host} from port {start_port} to {end_port}...")
    with ThreadPoolExecutor(max_workers=max_threads) as executor:
        # 使用线程池并发扫描端口
        for port in range(start_port, end_port + 1):
            executor.submit(scan_port, host, port)

if __name__ == "__main__":
    host = input("Enter the host to scan: ")
    start_port = int(input("Enter the start port: "))
    end_port = int(input("Enter the end port: "))
    max_threads = int(input("Enter the number of threads: "))

    port_scan(host, start_port, end_port, max_threads)
import socket
from concurrent.futures import ThreadPoolExecutor

def scan_port(host, port):
    try:
        # 创建一个socket对象
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # 设置超时时间
        sock.settimeout(1)
        # 尝试连接目标端口
        result = sock.connect_ex((host, port))
        if result == 0:
            print(f"Port {port} is open")
        # 关闭socket连接
        sock.close()
    except Exception as e:
        print(f"Error scanning port {port}: {e}")

def port_scan(host, start_port, end_port, max_threads=100):
    print(f"Scanning {host} from port {start_port} to {end_port}...")
    with ThreadPoolExecutor(max_workers=max_threads) as executor:
        # 使用线程池并发扫描端口
        for port in range(start_port, end_port + 1):
            executor.submit(scan_port, host, port)

if __name__ == "__main__":
    host = input("Enter the host to scan: ")
    start_port = int(input("Enter the start port: "))
    end_port = int(input("Enter the end port: "))
    max_threads = int(input("Enter the number of threads: "))

    port_scan(host, start_port, end_port, max_threads)

代码说明：

1. scan_port函数：用于扫描单个端口。它尝试连接到指定的主机和端口，如果连接成功（返回值为0），则说明该端口是开放的。
2. port_scan函数：使用ThreadPoolExecutor来并发扫描多个端口。max_threads参数控制并发线程的数量。
3. 主程序：从用户输入中获取要扫描的主机、起始端口、结束端口以及并发线程数，然后调用port_scan函数进行扫描。

使用说明：

• 运行程序后，输入要扫描的主机IP或域名。
• 输入要扫描的端口范围（起始端口和结束端口）。
• 输入并发线程数（建议根据网络状况和系统性能进行调整）。

注意事项：

• 端口扫描可能会对目标主机造成一定的负载，请确保你有权限扫描目标主机。
• 扫描速度过快可能会导致网络阻塞或被防火墙拦截，建议合理设置超时时间和并发线程数。

优化建议：

• 可以进一步优化扫描策略，例如使用异步I/O（如asyncio）来提高效率。
• 可以添加更多的错误处理和日志记录功能，以便更好地调试和分析扫描结果。

这个端口扫描器是一个基础版本，适合学习和简单的网络探测任务。对于更复杂的需求，可以考虑使用更专业的工具如nmap。