第2天：网络的分类和拓扑结构

# day02_lan_scanner.py
# 功能：扫描局域网内的活跃设备

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

def get_local_ip_and_subnet():
    """
    获取本机IP地址和所在的子网

    返回：
        local_ip: 本机IP地址字符串
        subnet: 子网字符串（如 "192.168.1.0/24"）
    """
    try:
        # 创建socket连接到外部地址，获取本机IP
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.connect(("8.8.8.8", 80))
        local_ip = sock.getsockname()[0]
        sock.close()

        # 从IP地址推断子网（假设是/24子网）
        # 例如：192.168.1.100 -> 192.168.1.0/24
        ip_parts = local_ip.split('.')
        subnet = f"{ip_parts[0]}.{ip_parts[1]}.{ip_parts[2]}.0/24"

        return local_ip, subnet

    except Exception as e:
        print(f"获取本机IP失败：{e}")
        return None, None

def check_host(ip, timeout=0.5):
    """
    检查指定IP的主机是否在线

    参数：
        ip: 要检查的IP地址
        timeout: 超时时间（秒）

    返回：
        如果主机在线，返回(ip, hostname)
        如果主机不在线，返回None

    工作原理：
        尝试连接到目标IP的常用端口（80端口）
        如果能连接，说明主机在线
    """
    try:
        # 尝试连接到80端口（HTTP端口）
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)

        # 尝试连接
        result = sock.connect_ex((str(ip), 80))

        if result == 0:
            # 连接成功，尝试获取主机名
            try:
                hostname = socket.gethostbyaddr(str(ip))[0]
            except:
                hostname = "未知主机名"

            sock.close()
            return (str(ip), hostname)
        else:
            sock.close()
            return None

    except:
        return None

def ping_check(ip, timeout=0.5):
    """
    使用更简单的方法检测主机是否在线
    尝试多个常用端口
    """
    common_ports = [80, 443, 22, 445, 139]  # HTTP, HTTPS, SSH, SMB

    for port in common_ports:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)
            result = sock.connect_ex((str(ip), port))
            sock.close()

            if result == 0:
                # 尝试获取主机名
                try:
                    hostname = socket.gethostbyaddr(str(ip))[0]
                except:
                    hostname = "未知主机名"

                return (str(ip), hostname, port)

        except:
            continue

    return None

def scan_network(subnet, max_workers=50):
    """
    扫描整个子网

    参数：
        subnet: 子网地址（如 "192.168.1.0/24"）
        max_workers: 最大线程数

    返回：
        活跃主机列表
    """
    # 解析子网，获取所有IP地址
    network = ipaddress.ip_network(subnet, strict=False)
    total_hosts = network.num_addresses - 2  # 减去网络地址和广播地址

    print(f"🔍 开始扫描子网: {subnet}")
    print(f"📊 总共需要扫描 {total_hosts} 个IP地址")
    print(f"⏳ 请稍候...\n")

    active_hosts = []
    scanned = 0

    # 使用线程池并发扫描
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有扫描任务
        future_to_ip = {
            executor.submit(ping_check, ip): ip
            for ip in network.hosts()  # 排除网络地址和广播地址
        }

        # 处理完成的任务
        for future in as_completed(future_to_ip):
            scanned += 1
            result = future.result()

            if result:
                ip, hostname, port = result
                active_hosts.append((ip, hostname, port))
                print(f"✅ 发现设备: {ip:15} | {hostname:30} | 端口: {port}")

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

    return active_hosts

def display_topology(active_hosts, local_ip):
    """
    显示网络拓扑结构（简化版）
    """
    print("\n" + "=" * 70)
    print("网络拓扑结构（星型拓扑）".center(70))
    print("=" * 70)
    print()
    print("                      [路由器/交换机]")
    print("                             |")
    print("          " + "-" * 40)

    for ip, hostname, port in active_hosts:
        marker = " (本机)" if ip == local_ip else ""
        print(f"          |-- [{ip}] {hostname}{marker}")

    print()
    print("=" * 70)

def main():
    """
    主函数
    """
    print("=" * 70)
    print("局域网设备扫描器".center(70))
    print("=" * 70)
    print()

    # 获取本机IP和子网
    local_ip, subnet = get_local_ip_and_subnet()

    if not local_ip:
        print("❌ 无法获取本机IP地址，请检查网络连接！")
        return

    print(f"📍 本机IP: {local_ip}")
    print(f"🌐 所在子网: {subnet}")
    print()

    # 扫描网络
    active_hosts = scan_network(subnet)

    print("\n" + "=" * 70)
    print(f"✨ 扫描完成！共发现 {len(active_hosts)} 台活跃设备")
    print("=" * 70)

    # 显示详细信息
    if active_hosts:
        print("\n📋 活跃设备列表：")
        print(f"{'IP地址':<15} | {'主机名':<30} | {'开放端口'}")
        print("-" * 70)

        for ip, hostname, port in active_hosts:
            marker = " ⭐" if ip == local_ip else ""
            print(f"{ip:<15} | {hostname:<30} | {port}{marker}")

        # 显示拓扑结构
        display_topology(active_hosts, local_ip)

        # 分析网络类型
        print("\n💡 网络分析：")
        print(f"   - 这是一个局域网（LAN）")
        print(f"   - 使用星型拓扑结构")
        print(f"   - 共有 {len(active_hosts)} 台设备连接到同一个交换机/路由器")

    else:
        print("\n⚠️  未发现其他活跃设备")

if __name__ == "__main__":
    main()

cd code/day02
python day02_lan_scanner.py

import ipaddress

# 创建网络对象
network = ipaddress.ip_network("192.168.1.0/24")

# 获取网络中的所有主机IP
for ip in network.hosts():
    print(ip)  # 192.168.1.1, 192.168.1.2, ..., 192.168.1.254

from concurrent.futures import ThreadPoolExecutor

# 创建线程池，最多50个线程同时工作
with ThreadPoolExecutor(max_workers=50) as executor:
    # 提交任务
    futures = [executor.submit(scan_ip, ip) for ip in ip_list]

    # 等待完成
    for future in as_completed(futures):
        result = future.result()

result = sock.connect_ex((ip, port))

# 返回值：
# 0 = 连接成功
# 其他 = 连接失败（错误码）