Python程序，输入IP，扫描该IP哪些端口对外是开放的，输出端口列表

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
IP端口扫描程序
输入IP地址，扫描该IP哪些端口对外是开放的，输出端口列表
"""

import socket
import sys
import concurrent.futures
import ipaddress
from tabulate import tabulate
import time

def is_valid_ip(ip):
    """
    验证IP地址是否有效
    """
    try:
        ipaddress.ip_address(ip)
        return True
    except ValueError:
        return False

def scan_port(ip, port, timeout=1):
    """
    扫描单个端口是否开放
    
    参数:
        ip (str): 目标IP地址
        port (int): 要扫描的端口
        timeout (float): 连接超时时间(秒)
        
    返回:
        bool: 如果端口开放返回True，否则返回False
    """
    try:
        # 创建TCP套接字
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        
        # 尝试连接
        result = sock.connect_ex((ip, port))
        
        # 关闭套接字
        sock.close()
        
        # 如果连接成功，端口是开放的
        return result == 0
    
    except (socket.error, socket.timeout, OSError):
        return False

def scan_ports(ip, port_range=None, max_workers=100):
    """
    扫描IP地址的多个端口
    
    参数:
        ip (str): 目标IP地址
        port_range (tuple): 端口范围，格式为(起始端口, 结束端口)
        max_workers (int): 最大并发线程数
        
    返回:
        list: 开放端口列表
    """
    if port_range is None:
        # 默认扫描常见端口
        port_range = (1, 1024)
    
    start_port, end_port = port_range
    open_ports = []
    total_ports = end_port - start_port + 1
    
    print(f"开始扫描 {ip} 的端口 {start_port}-{end_port}...")
    start_time = time.time()
    
    # 使用线程池进行并发扫描
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 创建扫描任务
        future_to_port = {
            executor.submit(scan_port, ip, port): port 
            for port in range(start_port, end_port + 1)
        }
        
        # 处理完成的任务
        completed = 0
        for future in concurrent.futures.as_completed(future_to_port):
            port = future_to_port[future]
            completed += 1
            
            # 显示进度
            if completed % 100 == 0 or completed == total_ports:
                progress = (completed / total_ports) * 100
                elapsed = time.time() - start_time
                print(f"进度: {completed}/{total_ports} ({progress:.1f}%) - 已用时间: {elapsed:.1f}秒", end="\r")
            
            try:
                if future.result():
                    open_ports.append(port)
                    print(f"\n发现开放端口: {port}")
            except Exception as e:
                print(f"\n扫描端口 {port} 时出错: {e}")
    
    print(f"\n扫描完成! 总用时: {time.time() - start_time:.1f}秒")
    return open_ports

def display_open_ports(ip, open_ports):
    """
    显示开放端口列表
    """
    if not open_ports:
        print(f"\n{ip} 没有发现开放的端口")
        return
    
    # 尝试获取常见端口的服务名称
    port_info = []
    for port in sorted(open_ports):
        try:
            service = socket.getservbyport(port)
        except (socket.error, OSError):
            service = "未知"
        
        port_info.append([port, service])
    
    # 显示表格
    print(f"\n{ip} 的开放端口:")
    headers = ["端口", "可能的服务"]
    print(tabulate(port_info, headers=headers, tablefmt="grid"))

def main():
    """
    主函数
    """
    # 获取用户输入
    if len(sys.argv) > 1:
        target_ip = sys.argv[1]
    else:
        target_ip = input("请输入要扫描的IP地址: ")
    
    # 验证IP地址
    if not is_valid_ip(target_ip):
        print(f"错误: '{target_ip}' 不是有效的IP地址")
        sys.exit(1)
    
    # 获取端口范围
    try:
        custom_range = input("请输入要扫描的端口范围 (格式: 起始端口-结束端口) [默认: 1-1024]: ")
        if custom_range:
            start, end = map(int, custom_range.split('-'))
            if start < 1 or end > 65535 or start > end:
                raise ValueError
            port_range = (start, end)
        else:
            port_range = (1, 1024)
    except ValueError:
        print("错误: 无效的端口范围，使用默认范围 1-1024")
        port_range = (1, 1024)
    
    # 扫描端口
    open_ports = scan_ports(target_ip, port_range)
    
    # 显示结果
    display_open_ports(target_ip, open_ports)

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n\n扫描被用户中断")
        sys.exit(0)
    except Exception as e:
        print(f"\n程序执行出错: {e}")
        sys.exit(1)