diff --git "a/Against\n/home/debian/Cazalla/Payload_Type/cazalla/cazalla/agent_functions/upload.py" "b/Against\n/home/debian/Cazalla/Payload_Type/cazalla/cazalla/agent_functions/upload.py" deleted file mode 100644 index 41514dd..0000000 --- "a/Against\n/home/debian/Cazalla/Payload_Type/cazalla/cazalla/agent_functions/upload.py" +++ /dev/null @@ -1,104 +0,0 @@ -from mythic_container.MythicCommandBase import * -from mythic_container.MythicRPC import * -import os - - -class UploadArguments(TaskArguments): - def __init__(self, command_line, **kwargs): - super().__init__(command_line, **kwargs) - self.args = [ - CommandParameter( - name="file", - type=ParameterType.File, - description="File to upload to the target", - ), - CommandParameter( - name="path", - type=ParameterType.String, - description="Path where the file should be saved on the target", - ), - ] - - async def parse_arguments(self): - if len(self.command_line) == 0: - raise ValueError("Must supply a file and path") - - parts = self.command_line.split(" ", 1) - if len(parts) != 2: - raise ValueError("Must supply both file and path: upload ") - - self.add_arg("file", parts[0]) - self.add_arg("path", parts[1]) - - async def parse_dictionary(self, dictionary_arguments): - if "file" in dictionary_arguments: - self.add_arg("file", dictionary_arguments["file"]) - if "path" in dictionary_arguments: - # Normalize path: replace double backslashes with single backslash - path = dictionary_arguments["path"] - if path: - path = path.replace("\\\\", "\\") - self.add_arg("path", path) - - -class UploadCommand(CommandBase): - cmd = "upload" - needs_admin = False - help_cmd = "upload " - description = "Upload a file from the Mythic server to the target" - version = 1 - author = "@KaseyaOFSTeam" - argument_class = UploadArguments - - async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse: - response = PTTaskCreateTaskingMessageResponse( - TaskID=taskData.Task.ID, - Success=True, - ) - - file_id = taskData.args.get_arg("file") - path = taskData.args.get_arg("path") - - # Get file information to extract filename - try: - file_search = await SendMythicRPCFileSearch(MythicRPCFileSearchMessage( - AgentFileId=file_id, - LimitByCallback=False, - TaskID=taskData.Task.ID - )) - - if file_search.Success and len(file_search.Files) > 0: - file_info = file_search.Files[0] - filename = file_info.Filename - - # If path is empty, use current directory - if not path or path.strip() == "": - path = ".\\" + filename - # If path ends with backslash or is a directory, append filename - elif path.endswith("\\") or path.endswith("/"): - # Path is a directory, append filename - path = path.rstrip("\\/") + "\\" + filename - # Check if path is just a directory (no filename extension and ends with no backslash) - elif not os.path.basename(path) or (not path.endswith("\\") and os.path.isabs(path) and not os.path.splitext(os.path.basename(path))[1]): - # Try to check if it's a directory by seeing if basename has no extension - # This is a heuristic - if path has no extension and looks like a directory, append filename - if not os.path.splitext(os.path.basename(path))[1]: - # No extension, might be a directory - path = path.rstrip("\\/") + "\\" + filename - - response.DisplayParams = f"file_id: {file_id}, path: {path}, filename: {filename}" - # Update the path argument with the final path (including filename if needed) - taskData.args.add_arg("path", path) - else: - # If we can't get file info, just use the path as-is - response.DisplayParams = f"file_id: {file_id}, path: {path}" - except Exception as e: - # If we can't get file info, just use the path as-is - response.DisplayParams = f"file_id: {file_id}, path: {path} (warning: could not get filename: {str(e)})" - - return response - - async def process_response(self, task: PTTaskMessageAllData, response: any) -> PTTaskProcessResponseMessageResponse: - resp = PTTaskProcessResponseMessageResponse(TaskID=task.Task.ID, Success=True) - return resp -