第4天：TCP/IP四层模型

# day04_packet_analyzer.py
# 功能：捕获并分析网络数据包（TCP/IP各层信息）

import socket
import struct
import textwrap

def ethernet_frame(data):
    """
    解析以太网帧（网络接口层）

    以太网帧格式：
    [目标MAC(6字节) | 源MAC(6字节) | 类型(2字节) | 数据 | CRC(4字节)]
    """
    # 提取MAC地址和类型
    dest_mac, src_mac, eth_type = struct.unpack('! 6s 6s H', data[:14])

    # 格式化MAC地址
    dest_mac = ':'.join('{:02x}'.format(b) for b in dest_mac)
    src_mac = ':'.join('{:02x}'.format(b) for b in src_mac)

    # 以太网类型
    # 0x0800 = IPv4, 0x86DD = IPv6, 0x0806 = ARP
    eth_type = socket.htons(eth_type)

    # 返回MAC地址、类型和剩余数据
    return dest_mac, src_mac, eth_type, data[14:]

def ipv4_packet(data):
    """
    解析IPv4数据包（网际层）

    IPv4头部格式（20字节）：
    [版本+头长(1) | 服务类型(1) | 总长度(2) | 标识(2) | 标志+偏移(2) |
     TTL(1) | 协议(1) | 校验和(2) | 源IP(4) | 目标IP(4)]
    """
    # 解析IPv4头部的前20字节
    version_header_length = data[0]
    version = version_header_length >> 4  # 高4位是版本
    header_length = (version_header_length & 15) * 4  # 低4位是头长（单位：4字节）

    # 提取TTL、协议、源IP、目标IP
    ttl, proto, src, target = struct.unpack('! 8x B B 2x 4s 4s', data[:20])

    # 格式化IP地址
    src_ip = '.'.join(map(str, src))
    target_ip = '.'.join(map(str, target))

    # 协议号对应的协议名
    protocols = {1: 'ICMP', 6: 'TCP', 17: 'UDP'}
    proto_name = protocols.get(proto, f'Other({proto})')

    return version, header_length, ttl, proto_name, src_ip, target_ip, data[header_length:]

def tcp_segment(data):
    """
    解析TCP段（传输层）

    TCP头部格式（至少20字节）：
    [源端口(2) | 目标端口(2) | 序号(4) | 确认号(4) |
     头长+标志(2) | 窗口(2) | 校验和(2) | 紧急指针(2)]
    """
    src_port, dest_port, sequence, acknowledgment, offset_reserved_flags = struct.unpack(
        '! H H L L H', data[:14]
    )

    # 提取头部长度（高4位）
    offset = (offset_reserved_flags >> 12) * 4

    # 提取TCP标志位
    flag_urg = (offset_reserved_flags & 32) >> 5
    flag_ack = (offset_reserved_flags & 16) >> 4
    flag_psh = (offset_reserved_flags & 8) >> 3
    flag_rst = (offset_reserved_flags & 4) >> 2
    flag_syn = (offset_reserved_flags & 2) >> 1
    flag_fin = offset_reserved_flags & 1

    return src_port, dest_port, sequence, acknowledgment, flag_urg, flag_ack, \
           flag_psh, flag_rst, flag_syn, flag_fin, data[offset:]

def udp_segment(data):
    """
    解析UDP段（传输层）

    UDP头部格式（8字节）：
    [源端口(2) | 目标端口(2) | 长度(2) | 校验和(2)]
    """
    src_port, dest_port, length = struct.unpack('! H H 2x H', data[:8])
    return src_port, dest_port, length, data[8:]

def format_multi_line(prefix, string, size=80):
    """格式化多行输出"""
    size -= len(prefix)
    if isinstance(string, bytes):
        string = ''.join(r'\x{:02x}'.format(byte) for byte in string)
        if size % 2:
            size -= 1

    return '\n'.join([prefix + line for line in textwrap.wrap(string, size)])

def main():
    """
    主函数：捕获并分析数据包
    """
    print("=" * 80)
    print("TCP/IP数据包分析器".center(80))
    print("=" * 80)
    print("\n📡 正在监听网络接口...")
    print("💡 提示：访问一个网站或ping一个地址来生成流量")
    print("⚠️  按Ctrl+C停止捕获\n")
    print("=" * 80)

    # 创建原始套接字
    # 注意：在某些系统上可能需要root权限
    try:
        # Windows
        conn = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP)
        conn.bind(('', 0))
        conn.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
        conn.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
    except AttributeError:
        # Linux/Mac
        conn = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(3))

    packet_count = 0

    try:
        while True:
            # 接收数据
            raw_data, addr = conn.recvfrom(65535)
            packet_count += 1

            print(f"\n{'▼' * 40}")
            print(f"📦 数据包 #{packet_count}")
            print(f"{'▼' * 40}\n")

            # 1. 解析以太网帧（网络接口层）
            try:
                dest_mac, src_mac, eth_type, data = ethernet_frame(raw_data)
                print(f"🔗 网络接口层（以太网帧）")
                print(f"   源MAC地址: {src_mac}")
                print(f"   目标MAC地址: {dest_mac}")
                print(f"   协议类型: 0x{eth_type:04x}")
            except:
                # 某些系统可能不包含以太网头
                eth_type = 0x0800
                data = raw_data

            # 2. 解析IP数据包（网际层）
            if eth_type == 0x0800:  # IPv4
                version, header_length, ttl, proto, src_ip, target_ip, data = ipv4_packet(data)
                print(f"\n🌐 网际层（IP数据包）")
                print(f"   IP版本: IPv{version}")
                print(f"   头部长度: {header_length}字节")
                print(f"   TTL: {ttl}")
                print(f"   协议: {proto}")
                print(f"   源IP地址: {src_ip}")
                print(f"   目标IP地址: {target_ip}")

                # 3. 解析传输层
                if proto == 'TCP':
                    src_port, dest_port, sequence, acknowledgment, flag_urg, flag_ack, \
                    flag_psh, flag_rst, flag_syn, flag_fin, data = tcp_segment(data)

                    print(f"\n🚚 传输层（TCP段）")
                    print(f"   源端口: {src_port}")
                    print(f"   目标端口: {dest_port}")
                    print(f"   序列号: {sequence}")
                    print(f"   确认号: {acknowledgment}")
                    print(f"   标志位: ", end="")

                    flags = []
                    if flag_syn: flags.append("SYN")
                    if flag_ack: flags.append("ACK")
                    if flag_fin: flags.append("FIN")
                    if flag_rst: flags.append("RST")
                    if flag_psh: flags.append("PSH")
                    if flag_urg: flags.append("URG")

                    print(", ".join(flags) if flags else "无")

                    # 4. 应用层数据
                    if len(data) > 0:
                        print(f"\n📱 应用层数据")
                        print(f"   数据长度: {len(data)}字节")

                        # 尝试解析HTTP
                        try:
                            data_str = data.decode('utf-8', errors='ignore')
                            if data_str.startswith('HTTP') or data_str.startswith('GET') or \
                               data_str.startswith('POST'):
                                print(f"   协议: HTTP")
                                lines = data_str.split('\\r\\n')[:5]  # 只显示前5行
                                for line in lines:
                                    if line:
                                        print(f"   {line}")
                        except:
                            print(f"   数据预览: {data[:50]}")

                elif proto == 'UDP':
                    src_port, dest_port, length, data = udp_segment(data)

                    print(f"\n📦 传输层（UDP段）")
                    print(f"   源端口: {src_port}")
                    print(f"   目标端口: {dest_port}")
                    print(f"   长度: {length}字节")

                    # DNS查询/响应
                    if src_port == 53 or dest_port == 53:
                        print(f"   应用协议: DNS")

                elif proto == 'ICMP':
                    print(f"\n📡 传输层（ICMP）")
                    print(f"   这是一个ping或traceroute数据包")

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

            # 只捕获前10个包作为演示
            if packet_count >= 10:
                print(f"\n✅ 已捕获{packet_count}个数据包，停止监听")
                break

    except KeyboardInterrupt:
        print(f"\n\n✋ 用户停止捕获")
        print(f"📊 总共捕获了 {packet_count} 个数据包")

    finally:
        # Windows需要关闭混杂模式
        try:
            conn.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)
        except:
            pass
        conn.close()

    print("=" * 80)

if __name__ == "__main__":
    import sys
    if sys.platform == 'win32' or sys.platform == 'linux' or sys.platform == 'darwin':
        main()
    else:
        print("⚠️  此程序可能需要管理员/root权限")
        print("💡 Linux/Mac: sudo python day04_packet_analyzer.py")
        print("💡 Windows: 以管理员身份运行")

sudo python code/day04/day04_packet_analyzer.py

python code\day04\day04_packet_analyzer.py