第11天：端口和Socket深入

# 特点：
- 面向连接
- 可靠传输
- 数据按序到达
- 适用于需要可靠传输的场景

# 使用场景：
- 网页浏览（HTTP/HTTPS）
- 文件传输（FTP）
- 邮件发送（SMTP）
- 远程登录（SSH）

# Python创建：
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 特点：
- 无连接
- 不可靠（可能丢包）
- 速度快
- 适用于实时性要求高的场景

# 使用场景：
- 视频直播
- 语音通话
- 在线游戏
- DNS查询

# Python创建：
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# day11_port_scanner.py
# 功能：端口扫描器

import socket
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

class PortScanner:
    """端口扫描器"""

    def __init__(self):
        # 常用端口和服务的映射
        self.common_ports = {
            20: 'FTP-DATA',
            21: 'FTP',
            22: 'SSH',
            23: 'Telnet',
            25: 'SMTP',
            53: 'DNS',
            80: 'HTTP',
            110: 'POP3',
            143: 'IMAP',
            443: 'HTTPS',
            445: 'SMB',
            3306: 'MySQL',
            3389: 'RDP',
            5432: 'PostgreSQL',
            6379: 'Redis',
            8080: 'HTTP-Proxy',
            27017: 'MongoDB',
        }

    def scan_port(self, host, port, timeout=1):
        """
        扫描单个端口

        参数：
            host: 目标主机
            port: 端口号
            timeout: 超时时间（秒）

        返回：
            (port, is_open, service_name) 元组
        """
        try:
            # 创建TCP Socket
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)

            # 尝试连接
            result = sock.connect_ex((host, port))

            # 关闭连接
            sock.close()

            # result = 0 表示连接成功（端口开放）
            if result == 0:
                service = self.common_ports.get(port, 'Unknown')
                return (port, True, service)
            else:
                return (port, False, None)

        except socket.timeout:
            return (port, False, None)
        except socket.error:
            return (port, False, None)
        except Exception as e:
            return (port, False, None)

    def scan_port_with_banner(self, host, port, timeout=1):
        """
        扫描端口并尝试获取服务banner

        参数：
            host: 目标主机
            port: 端口号
            timeout: 超时时间

        返回：
            (port, is_open, service_name, banner)
        """
        port_result = self.scan_port(host, port, timeout)
        port_num, is_open, service = port_result

        if not is_open:
            return (port_num, False, None, None)

        # 尝试获取banner
        banner = None
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(2)
            sock.connect((host, port))

            # 发送简单的请求
            if port == 80 or port == 8080:
                sock.send(b'GET / HTTP/1.0\r\n\r\n')

            # 尝试接收响应
            banner = sock.recv(1024).decode('utf-8', errors='ignore').strip()
            sock.close()

            # 只保留第一行
            if banner:
                banner = banner.split('\n')[0][:50]

        except:
            pass

        return (port_num, is_open, service, banner)

    def scan_range(self, host, start_port, end_port, timeout=1, show_closed=False):
        """
        扫描端口范围

        参数：
            host: 目标主机
            start_port: 起始端口
            end_port: 结束端口
            timeout: 超时时间
            show_closed: 是否显示关闭的端口

        返回：
            开放端口列表
        """
        print(f"🔍 正在扫描 {host}:{start_port}-{end_port}")
        print(f"⏳ 请稍候...\n")

        open_ports = []
        total_ports = end_port - start_port + 1
        scanned = 0

        # 使用线程池并发扫描
        with ThreadPoolExecutor(max_workers=100) as executor:
            # 提交所有扫描任务
            futures = {
                executor.submit(self.scan_port, host, port, timeout): port
                for port in range(start_port, end_port + 1)
            }

            # 处理完成的任务
            for future in as_completed(futures):
                scanned += 1
                port_num, is_open, service = future.result()

                if is_open:
                    open_ports.append((port_num, service))
                    print(f"✅ 端口 {port_num:<6} 开放 - {service}")
                elif show_closed:
                    print(f"❌ 端口 {port_num:<6} 关闭")

                # 显示进度
                if scanned % 50 == 0:
                    progress = (scanned / total_ports) * 100
                    print(f"   进度: {progress:.1f}% ({scanned}/{total_ports})", end='\r')

        return open_ports

    def scan_common_ports(self, host, timeout=1):
        """
        扫描常用端口

        参数：
            host: 目标主机
            timeout: 超时时间

        返回：
            开放端口列表
        """
        print(f"🔍 正在扫描 {host} 的常用端口")
        print(f"📋 共 {len(self.common_ports)} 个常用端口\n")

        open_ports = []

        with ThreadPoolExecutor(max_workers=50) as executor:
            futures = {
                executor.submit(self.scan_port_with_banner, host, port, timeout): port
                for port in self.common_ports.keys()
            }

            for future in as_completed(futures):
                port_num, is_open, service, banner = future.result()

                if is_open:
                    open_ports.append((port_num, service, banner))
                    if banner:
                        print(f"✅ 端口 {port_num:<6} 开放 - {service:<15} | {banner}")
                    else:
                        print(f"✅ 端口 {port_num:<6} 开放 - {service}")

        return open_ports

    def display_results(self, host, open_ports):
        """
        显示扫描结果

        参数：
            host: 主机地址
            open_ports: 开放端口列表
        """
        print("\n" + "=" * 70)
        print(f"扫描结果 - {host}".center(70))
        print("=" * 70)

        if open_ports:
            print(f"\n发现 {len(open_ports)} 个开放端口:\n")
            print(f"{'端口':<10} {'服务':<20} {'备注'}")
            print("-" * 70)

            for port_info in sorted(open_ports):
                if len(port_info) == 2:
                    port, service = port_info
                    print(f"{port:<10} {service:<20}")
                elif len(port_info) == 3:
                    port, service, banner = port_info
                    banner_str = banner[:40] if banner else ""
                    print(f"{port:<10} {service:<20} {banner_str}")
        else:
            print("\n❌ 未发现开放端口")

        print("\n" + "=" * 70)

def main():
    """
    主函数
    """
    scanner = PortScanner()

    print("=" * 70)
    print("端口扫描器".center(70))
    print("=" * 70)
    print()

    print("请选择扫描模式:")
    print("1. 扫描常用端口（推荐）")
    print("2. 扫描端口范围")
    print("3. 扫描单个端口")
    print("0. 退出")

    choice = input("\n请输入选项: ").strip()

    if choice == '0':
        print("\n再见！")
        return

    # 获取目标主机
    host = input("\n请输入目标主机（IP或域名）: ").strip()
    if not host:
        print("❌ 主机不能为空")
        return

    # 验证主机是否可达
    print(f"\n🔍 正在检查主机 {host} 是否可达...")
    try:
        socket.gethostbyname(host)
        print("✅ 主机可达")
    except socket.gaierror:
        print("❌ 无法解析主机名")
        return

    print()
    start_time = time.time()

    if choice == '1':
        # 扫描常用端口
        open_ports = scanner.scan_common_ports(host)
        scanner.display_results(host, open_ports)

    elif choice == '2':
        # 扫描端口范围
        start_port = int(input("请输入起始端口: ").strip())
        end_port = int(input("请输入结束端口: ").strip())

        if start_port < 1 or end_port > 65535 or start_port > end_port:
            print("❌ 端口范围无效")
            return

        open_ports = scanner.scan_range(host, start_port, end_port)
        scanner.display_results(host, open_ports)

    elif choice == '3':
        # 扫描单个端口
        port = int(input("请输入端口号: ").strip())

        if port < 1 or port > 65535:
            print("❌ 端口号无效")
            return

        port_num, is_open, service, banner = scanner.scan_port_with_banner(host, port)

        if is_open:
            print(f"\n✅ 端口 {port} 开放")
            print(f"   服务: {service}")
            if banner:
                print(f"   Banner: {banner}")
        else:
            print(f"\n❌ 端口 {port} 关闭")

    # 显示耗时
    elapsed_time = time.time() - start_time
    print(f"\n⏱️  扫描耗时: {elapsed_time:.2f} 秒")

    # 安全提示
    print("\n" + "=" * 70)
    print("⚠️  安全提示:")
    print("   - 仅扫描您有权限扫描的主机")
    print("   - 未经授权的端口扫描可能违法")
    print("   - 请遵守网络安全法律法规")
    print("=" * 70)

if __name__ == "__main__":
    main()

python code/day11/day11_port_scanner.py

# connect_ex() 方法
result = sock.connect_ex((host, port))

# 返回值：
# 0 = 连接成功（端口开放）
# 其他 = 连接失败（端口关闭或被过滤）

# 使用线程池同时扫描多个端口
with ThreadPoolExecutor(max_workers=100) as executor:
    # 100个线程并发工作
    futures = {executor.submit(scan_port, host, port): port
               for port in range(start, end)}

# 连接成功后，尝试接收服务器的响应
sock.connect((host, port))
sock.send(b'GET / HTTP/1.0\r\n\r\n')  # 发送HTTP请求
banner = sock.recv(1024)  # 接收响应

# Banner可以帮助识别服务类型和版本

# 查看本机监听的端口
netstat -an | grep LISTEN    # Linux/Mac
netstat -an | findstr LISTEN # Windows

# 查看特定端口
lsof -i :8080               # Mac/Linux
netstat -ano | findstr :8080 # Windows

# 测试端口连通性
telnet host port
nc -zv host port            # Linux/Mac