核心代码

import socket
import argparse
from concurrent.futures import ThreadPoolExecutor
import time


def parse_ports(ports):
    """
    解析用户输入的端口参数，支持三种格式：单个端口、端口范围、端口列表
    
    函数含义:
        该函数负责将命令行参数中的端口字符串转换为 Python 可迭代的端口号序列，
        支持灵活的输入格式，满足不同扫描场景的需求。
    
    参数:
        ports (str): 用户输入的端口字符串，支持以下格式：
                     - 单个端口："80"
                     - 端口范围："1-1024"
                     - 端口列表："80,443,8080"
    
    返回:
        range 或 list: 解析后的端口号序列
        - range 类型：当输入为端口范围时（如 "1-1024"）
        - list 类型：当输入为单个端口或端口列表时
    
    异常处理:
        当输入格式不合法时，打印错误信息并退出程序
    """
    if "-" in ports:
        # 处理端口范围格式，如 "1-1024"
        try:
            start, end = map(int, ports.split("-"))
            return range(start, end + 1)
        except ValueError:
            print("[!] Invalid port range specified.")
            exit()
    elif "," in ports:
        # 处理端口列表格式，如 "80,443,8080"
        try:
            return [int(p) for p in ports.split(",")]
        except ValueError:
            print("[!] Invalid port list specified.")
            exit()
    else:
        # 处理单个端口格式，如 "80"
        try:
            return [int(ports)]
        except ValueError:
            print("[!] Invalid port specified.")
            exit()


def scan_port(target_ip, port, timeout):
    """
    扫描单个端口是否开放的核心函数
    
    函数含义:
        通过 TCP Socket 连接尝试，检测目标主机的指定端口是否处于开放状态。
        基于 TCP 三次握手原理，如果 connect_ex() 返回 0 表示连接成功，端口开放。
    
    参数:
        target_ip (str): 目标主机的 IP 地址，如 "192.168.1.1"
        port (int): 要扫描的端口号，范围 1-65535
        timeout (float): Socket 连接超时时间（秒），控制每次扫描的最大等待时间
    
    返回:
        int 或 None: 
        - 如果端口开放，返回端口号（int）
        - 如果端口关闭或超时，返回 None
    
    工作原理:
        1. 创建 IPv4 TCP Socket（AF_INET, SOCK_STREAM）
        2. 设置超时时间，避免阻塞过久
        3. 使用 connect_ex() 尝试连接目标端口
        4. 返回值为 0 表示连接成功（端口开放），否则返回错误码
    """
    try:
        # 创建 IPv4 TCP Socket
        # AF_INET: 使用 IPv4 地址族
        # SOCK_STREAM: 使用 TCP 协议（流式 Socket）
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(timeout)  # 设置超时时间，避免阻塞过久
            # connect_ex() 尝试连接目标端口
            # 返回值：0 表示连接成功，其他值表示错误码
            if s.connect_ex((target_ip, port)) == 0:
                return port
    except Exception:
        pass
    return None


def main():
    """
    程序的主入口函数，协调整个端口扫描流程
    
    函数含义:
        作为程序的控制中心，负责：
        1. 解析命令行参数
        2. 解析目标主机（IP 或域名）
        3. 解析端口范围
        4. 初始化线程池并发执行扫描任务
        5. 收集扫描结果并输出
    
    执行流程:
        参数解析 → 目标解析 → 端口解析 → 并发扫描 → 结果输出
    """
    # --- 参数解析 ---
    # 创建 ArgumentParser 对象，用于解析命令行参数
    parser = argparse.ArgumentParser(
        description="Python TCP Port Scanner - A powerful tool for network reconnaissance."
    )
    # 添加必需参数：-t 或 --target，指定目标 IP 或域名
    parser.add_argument(
        "-t", "--target", 
        required=True, 
        help="Specify the target IP address or hostname."
    )
    # 添加可选参数：-p 或 --ports，指定端口范围或列表，默认扫描 1-1024
    parser.add_argument(
        "-p", "--ports", 
        default="1-1024",
        help="Specify port range (e.g., 1-1024) or list (e.g., 80,443)."
    )
    # 添加可选参数：-w 或 --workers，指定并发线程数，默认 100 个线程
    parser.add_argument(
        "-w", "--workers", 
        type=int, 
        default=100, 
        help="Number of concurrent threads."
    )
    # 添加可选参数：--timeout，指定 Socket 连接超时时间，默认 0.5 秒
    parser.add_argument(
        "--timeout", 
        type=float, 
        default=0.5, 
        help="Socket connection timeout."
    )
    # 解析命令行参数，将解析结果存储在 args 对象中
    args = parser.parse_args()

    try:
        # 将域名解析为 IP 地址（如 google.com → 142.250.204.46）
        target_ip = socket.gethostbyname(args.target)
    except socket.gaierror:
        # 域名解析失败，捕获异常并提示用户
        print(f"[!] Hostname could not be resolved: {args.target}")
        return

    # 调用 parse_ports() 函数解析端口参数
    ports_to_scan = parse_ports(args.ports)

    # 打印扫描开始信息
    print("\n" + "-" * 50)
    print(f"[*] Scanning Target: {args.target} ({target_ip})")
    print(f"[*] Time started: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    print("-" * 50)

    # 初始化空列表，用于存储开放的端口号
    open_port = []

    # 使用线程池上下文管理器，创建最多包含 args.workers 个工作线程的线程池
    # with 语句确保线程池在任务完成后正确关闭和释放资源
    with ThreadPoolExecutor(max_workers=args.workers) as executor:
        # 创建 Future 对象映射字典：键为 Future 对象，值为对应的端口号
        # executor.submit() 方法：
        #   - 将 scan_port 函数提交到线程池执行
        #   - 立即返回一个 Future 对象（代表异步执行的任务）
        #   - Future 对象保存了任务的执行状态和结果
        # 字典推导式为每个端口创建一个扫描任务，建立 Future→port 的映射关系
        future_to_port = {
            executor.submit(scan_port, target_ip, port, args.timeout): port 
            for port in ports_to_scan
        }
        
        # 遍历所有 Future 对象（按提交顺序）
        for future in future_to_port:
            # future.result() 方法：
            #   - 阻塞等待任务完成
            #   - 获取 scan_port() 函数的返回值（端口号或 None）
            result = future.result()
            # 如果结果不为 None，表示端口开放
            if result is not None:
                open_port.append(result)
    
    # 输出扫描结果
    print("\n" + "-" * 50)
    print(f"[*] Scan completed: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    if open_port:
        # 对开放端口列表排序后输出
        print(f"[+] Open ports: {sorted(open_port)}")
    else:
        print("[!] No open ports found.")
    print("-" * 50)


if __name__ == "__main__":
    # 当直接运行此脚本时，调用 main() 函数启动程序
    main()

ThreadPoolExecutor 详细用法

ThreadPoolExecutor 是 Python concurrent.futures 模块提供的线程池类，用于并发执行任务。

核心参数

ThreadPoolExecutor(
    max_workers=None,           # 最大工作线程数
    thread_name_prefix='',      # 线程名称前缀
    initializer=None,           # 每个工作线程启动时调用的可调用对象
    initargs=()                 # initializer 的参数
)
 

主要方法

**submit(fn, *args, kwargs)
提交任务到线程池，立即返回 Future 对象

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=5)

# 提交任务
future = executor.submit(scan_port, '192.168.1.1', 80, 0.5)

# 获取结果（会阻塞直到任务完成）
result = future.result()

# 检查任务状态
print(future.done())  # 是否完成
print(future.running())  # 是否正在运行

*map(func, iterables, timeout=None, chunksize=1)
类似内置 map()，但并发执行

with ThreadPoolExecutor(max_workers=10) as executor:
    ports = [80, 443, 8080, 3306]
    results = executor.map(scan_port, ['192.168.1.1']*len(ports), ports, [0.5]*len(ports))
    
    for port in results:
        if port:
            print(f"Open: {port}")

shutdown(wait=True, cancel_futures=False)
关闭线程池

executor.shutdown(wait=True)  # 等待所有任务完成

✅ 性能提升：并发扫描多个端口，比串行快数十倍
✅ 资源控制：max_workers 限制最大线程数，避免系统过载
✅ 简化代码：无需手动管理线程创建和销毁

argparse 详细用法

argparse 是 Python 标准库的命令行参数解析模块，用于编写用户友好的命令行界面。

核心步骤

import argparse

# 步骤 1: 创建解析器
parser = argparse.ArgumentParser(description='程序描述')

# 步骤 2: 添加参数
parser.add_argument('-t', '--target', required=True, help='目标地址')

# 步骤 3: 解析参数
args = parser.parse_args()

# 步骤 4: 使用参数
print(args.target)

add_argument() 重要参数

parser.add_argument(
    '-t', '--target',          # 短选项和长选项
    required=True,             # 是否必需
    type=str,                  # 参数类型 (int, float, str 等)
    default='1-1024',          # 默认值
    help='指定目标 IP 或域名',   # 帮助信息
    metavar='IP',              # 帮助信息中的参数名
    choices=['1-100', '1-1024'],  # 可选值列表
    action='store_true'        # 特殊动作（如标志位）
)

常见参数类型示例

位置参数（无 - 前缀）

parser.add_argument('target', help='目标 IP 地址')
# 使用：python scanner.py 192.168.1.1

可选参数（带 - 或 -- 前缀）

parser.add_argument('-p', '--ports', default='1-1024', help='端口范围')
# 使用：python scanner.py -t 192.168.1.1 -p 1-100

类型转换

parser.add_argument('-w', '--workers', type=int, default=100)
# 自动将字符串转换为整数

布尔标志

parser.add_argument('-v', '--verbose', action='store_true')
# 使用时：python scanner.py -v
# args.verbose 为 True

可选值

parser.add_argument('--mode', choices=['fast', 'safe', 'stealth'])
# 只能选择指定的值