file upload completed

This commit is contained in:
marcos.luna
2025-10-30 11:28:05 +01:00
parent 6a009fecc1
commit d18a773265
@@ -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 <file_id> <path>")
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 <file> <path>"
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