kitos/Angerona
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

OPSEC warnings completed

Browse Source
This commit is contained in:
marcos.luna
2025-11-05 13:37:03 +01:00
parent 30c2d6390f
commit 6cd3511962
20 changed files with 1924 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Payload_Type/cazalla/agent_capabilities.json
+33
View File
@@ -0,0 +1,33 @@
{
"os": ["Windows"],
"languages": ["C"],
"features": {
"mythic": [
"Artifacts",
"Credentials",
"Process Browser",
"File Browser",
"Keylogging",
"Tokens",
"Screenshots",
"SOCKS Proxy",
"Reverse Port Forward",
"Context Tracking",
"Download Files",
"Upload Files"
],
"custom": [
"AES-256-CBC Encryption with HMAC-SHA256",
"Process Termination",
"Token Impersonation",
"Dynamic Sleep Adjustment"
]
},
"payload_output": ["exe", "dll"],
"architectures": ["x64", "x86"],
"c2": ["http"],
"supported_wrappers": [],
"mythic_version": "3.3",
"agent_version": "1.0.0"
}
Payload_Type/cazalla/cazalla/agent_code/cazalla/SistemadeFicheros.c
+123 -14
View File
@@ -6,6 +6,91 @@
/* Forward decl to avoid implicit declaration on some toolchains */
BOOL PackageAddFormatPrintf(PPaquete package, BOOL copySize, char* fmt, ...);
/* Helper function to resolve relative paths to absolute paths */
/* Uses two static buffers to support resolving two paths simultaneously (e.g., cp source dest) */
static PCHAR resolveFilePath(PCHAR filePath, SIZE_T pathLen) {
static char resolvedPath1[MAX_PATH * 2] = {0};
static char resolvedPath2[MAX_PATH * 2] = {0};
static BOOL useFirstBuffer = TRUE;
// Select which buffer to use (alternate between calls)
char* resolvedPath = useFirstBuffer ? resolvedPath1 : resolvedPath2;
useFirstBuffer = !useFirstBuffer; // Toggle for next call
if (!filePath || pathLen == 0) {
return filePath;
}
// Check if path is already absolute
// Absolute paths in Windows:
// - Start with drive letter + colon (C:, D:, etc.)
// - Start with \\ (UNC path)
// - Start with \ (absolute from current drive root)
BOOL isAbsolute = FALSE;
if (pathLen >= 2) {
// Check for drive letter (C:, D:, etc.)
if (isalpha((unsigned char)filePath[0]) && filePath[1] == ':') {
isAbsolute = TRUE;
}
// Check for UNC path (\\)
else if (filePath[0] == '\\' && filePath[1] == '\\') {
isAbsolute = TRUE;
}
// Check for absolute from root (\)
else if (filePath[0] == '\\') {
isAbsolute = TRUE;
}
}
if (isAbsolute) {
// Path is already absolute, return as-is
return filePath;
}
// Path is relative, combine with current directory
PCHAR currentDir = getCurrentDirectoryContext();
// If current directory is empty, try to get it from system
if (!currentDir || currentDir[0] == '\0') {
static char systemCwd[MAX_PATH] = {0};
if (GetCurrentDirectoryA(sizeof(systemCwd), systemCwd) > 0) {
currentDir = systemCwd;
} else {
// If we can't get current directory, return original path
return filePath;
}
}
// Build resolved path: currentDir + \ + filePath
SIZE_T cwdLen = strlen(currentDir);
SIZE_T totalLen = cwdLen + 1 + pathLen; // +1 for backslash
if (totalLen >= sizeof(resolvedPath1)) {
// Path too long, return original
_err("[resolveFilePath] Resolved path too long (%zu bytes), using original", totalLen);
return filePath;
}
// Copy current directory
memcpy(resolvedPath, currentDir, cwdLen);
// Add backslash if current directory doesn't end with one
if (cwdLen > 0 && currentDir[cwdLen - 1] != '\\' && currentDir[cwdLen - 1] != '/') {
resolvedPath[cwdLen] = '\\';
cwdLen++;
}
// Copy file path
memcpy(resolvedPath + cwdLen, filePath, pathLen);
resolvedPath[cwdLen + pathLen] = '\0';
_dbg("[resolveFilePath] Resolved: \"%.*s\" -> \"%s\"", (int)pathLen, filePath, resolvedPath);
return resolvedPath;
}
#define _CRT_SECURE_NO_WARNINGS
#pragma warning(disable : 4996)
@@ -263,10 +348,18 @@ VOID CopiarFichero(PAnalizador argumentos) {
PCHAR ficheroExistente = getString(argumentos, &tamano);
PCHAR nuevoFichero = getString(argumentos, &tamano2);
_dbg("Copying file \"%s\" to \"%s\"", ficheroExistente, nuevoFichero);
_dbg("[cp] Original source: \"%.*s\" (len=%zu)", (int)tamano, ficheroExistente, tamano);
_dbg("[cp] Original dest: \"%.*s\" (len=%zu)", (int)tamano2, nuevoFichero, tamano2);
// Resolve relative paths to absolute paths
PCHAR resolvedSource = resolveFilePath(ficheroExistente, tamano);
PCHAR resolvedDest = resolveFilePath(nuevoFichero, tamano2);
_dbg("[cp] Resolved source: \"%s\"", resolvedSource);
_dbg("[cp] Resolved dest: \"%s\"", resolvedDest);
// Copy the file
if (!CopyFileA(ficheroExistente, nuevoFichero, FALSE)){
if (!CopyFileA(resolvedSource, resolvedDest, FALSE)){
DWORD error = GetLastError();
@@ -296,14 +389,14 @@ VOID CopiarFichero(PAnalizador argumentos) {
PPaquete respuestaTarea = nuevoPaquete(POST_RESPONSE, TRUE);
char frase2[512]; // Asume que el path no va a pasar de 512 caracteres
snprintf(frase2, sizeof(frase2), "[cp] Fichero copiado correctamente a: %s", nuevoFichero);
snprintf(frase2, sizeof(frase2), "[cp] Fichero copiado correctamente a: %s", resolvedDest);
addString(salida, frase2, FALSE);
addString(respuestaTarea, tareaUuid, FALSE);
addBytes(respuestaTarea, (PBYTE)salida->buffer, salida->length, TRUE);
// Add File Write artifact for the destination file
addArtifact(respuestaTarea, "File Write", nuevoFichero);
addArtifact(respuestaTarea, "File Write", resolvedDest);
mandarPaquete(respuestaTarea);
liberarPaquete(salida);
@@ -900,11 +993,21 @@ VOID LeerFichero(PAnalizador argumentos) {
SIZE_T tamano = 0;
PCHAR rutaFichero = getString(argumentos, &tamano);
_dbg("Reading file: \"%s\"", rutaFichero);
_dbg("[cat] Reading file: rutaFichero=%p, tamano=%zu", rutaFichero, tamano);
if (!rutaFichero || tamano == 0) {
_err("[cat] rutaFichero is NULL or tamano is 0!");
return;
}
// Resolve relative paths to absolute paths
PCHAR resolvedPath = resolveFilePath(rutaFichero, tamano);
_dbg("[cat] Original path: \"%.*s\" (len=%zu)", (int)tamano, rutaFichero, tamano);
_dbg("[cat] Resolved path: \"%s\"", resolvedPath);
// Abrir archivo para lectura
HANDLE hFile = CreateFileA(
rutaFichero,
resolvedPath,
GENERIC_READ,
FILE_SHARE_READ,
NULL,
@@ -1071,11 +1174,17 @@ VOID DescargarFichero(PAnalizador argumentos) {
SIZE_T tamano = 0;
PCHAR rutaFichero = getString(argumentos, &tamano);
_dbg("[download] Archivo a descargar: \"%s\"", rutaFichero);
_dbg("[download] Original path: \"%.*s\" (len=%zu)", (int)tamano, rutaFichero, tamano);
// Resolve relative paths to absolute paths
PCHAR resolvedPath = resolveFilePath(rutaFichero, tamano);
_dbg("[download] Resolved path: \"%s\"", resolvedPath);
_dbg("[download] Archivo a descargar: \"%s\"", resolvedPath);
// Abrir archivo para lectura
HANDLE hFile = CreateFileA(
rutaFichero,
resolvedPath,
GENERIC_READ,
FILE_SHARE_READ,
NULL,
@@ -1126,21 +1235,21 @@ VOID DescargarFichero(PAnalizador argumentos) {
PPaquete respuestaFinal = nuevoPaquete(POST_RESPONSE, TRUE);
addString(respuestaFinal, tareaUuid, FALSE);
PPaquete salidaFinal = nuevoPaquete(0, FALSE);
PackageAddFormatPrintf(salidaFinal, FALSE, "[download] Archivo vacío: %s\n", rutaFichero);
PackageAddFormatPrintf(salidaFinal, FALSE, "[download] Archivo vacío: %s\n", resolvedPath);
addBytes(respuestaFinal, (PBYTE)salidaFinal->buffer, salidaFinal->length, TRUE);
addArtifact(respuestaFinal, "File Read", rutaFichero);
addArtifact(respuestaFinal, "File Read", resolvedPath);
mandarPaquete(respuestaFinal);
liberarPaquete(salidaFinal);
liberarPaquete(respuestaFinal);
goto end;
}
// Obtener full path del archivo
// Obtener full path del archivo (use resolved path)
char fullPath[MAX_PATH];
DWORD pathLen = GetFullPathNameA(rutaFichero, MAX_PATH, fullPath, NULL);
DWORD pathLen = GetFullPathNameA(resolvedPath, MAX_PATH, fullPath, NULL);
if (pathLen == 0 || pathLen >= MAX_PATH) {
// Si falla, usar la ruta original
strncpy(fullPath, rutaFichero, MAX_PATH - 1);
// Si falla, usar la ruta resuelta
strncpy(fullPath, resolvedPath, MAX_PATH - 1);
fullPath[MAX_PATH - 1] = '\0';
}
Payload_Type/cazalla/cazalla/agent_code/cazalla/screenshot.c
+14 -2
View File
@@ -104,7 +104,8 @@ VOID CapturarPantalla(PAnalizador argumentos) {
// Copy screen to bitmap
if (!BitBlt(hMemoryDC, 0, 0, screenWidth, screenHeight, hScreenDC, 0, 0, SRCCOPY)) {
_err("[screenshot] Error copiando pantalla");
DWORD error = GetLastError();
_err("[screenshot] Error copiando pantalla con BitBlt, código de error: %lu", error);
SelectObject(hMemoryDC, hOldBitmap);
DeleteObject(hBitmap);
DeleteDC(hMemoryDC);
@@ -112,7 +113,18 @@ VOID CapturarPantalla(PAnalizador argumentos) {
PPaquete respuestaError = nuevoPaquete(POST_RESPONSE, TRUE);
addString(respuestaError, tareaUuid, FALSE);
PPaquete salidaError = nuevoPaquete(0, FALSE);
PackageAddFormatPrintf(salidaError, FALSE, "[screenshot] Error: No se pudo copiar la pantalla\n");
// Provide more specific error messages based on error code
if (error == ERROR_INVALID_HANDLE || error == 6) {
PackageAddFormatPrintf(salidaError, FALSE, "[screenshot] Error: Handle inválido (error: %lu). Puede que no haya sesión interactiva activa.\n", error);
} else if (error == ERROR_ACCESS_DENIED || error == 5) {
PackageAddFormatPrintf(salidaError, FALSE, "[screenshot] Error: Acceso denegado (error: %lu). Se requieren permisos de sesión interactiva.\n", error);
} else if (error == ERROR_INVALID_PARAMETER || error == 87) {
PackageAddFormatPrintf(salidaError, FALSE, "[screenshot] Error: Parámetros inválidos (error: %lu). Verifica dimensiones de pantalla.\n", error);
} else {
PackageAddFormatPrintf(salidaError, FALSE, "[screenshot] Error: No se pudo copiar la pantalla. Código de error: %lu\n", error);
}
addBytes(respuestaError, (PBYTE)salidaError->buffer, salidaError->length, TRUE);
mandarPaquete(respuestaError);
liberarPaquete(salidaError);
Payload_Type/cazalla/cazalla/agent_functions/cat.py
+104 -1
View File
@@ -9,7 +9,10 @@ class CatArguments(TaskArguments):
CommandParameter(
name="file",
type=ParameterType.String,
description="Ruta del archivo a leer"
description="Ruta del archivo a leer",
parameter_group_info=[ParameterGroupInfo(
required=True
)]
),
]
@@ -36,6 +39,106 @@ class CatCommand(CommandBase):
supported_os=[SupportedOS.Windows],
suggested_command=False
)
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about file reading risks, especially for sensitive files.
"""
file_path = taskData.args.get_arg("file") or ""
# Sensitive file patterns
sensitive_patterns = {
"sam": "SAM database - Credential extraction heavily monitored",
"system": "SYSTEM hive - Registry access logged",
"security": "SECURITY hive - Security database access logged",
"ntds.dit": "NTDS.dit - Active Directory database, highly monitored",
"lsass": "LSASS memory - Credential extraction heavily monitored",
"passwd": "Password files - Credential access monitored",
"shadow": "Shadow files - Credential access monitored",
"config": "Configuration files - May contain credentials",
}
warnings = []
file_lower = file_path.lower()
for pattern, reason in sensitive_patterns.items():
if pattern in file_lower:
warnings.append(f"🚨 {pattern}: {reason}")
message = "⚠️ OPSEC WARNING - File Read Detected\n\n"
message += "File reading is monitored by:\n"
message += " • EDR/XDR solutions\n"
message += " • Data loss prevention (DLP) systems\n"
message += " • File access monitoring tools\n"
message += " • File access auditing (if enabled)\n\n"
if file_path:
message += f"Target File: {file_path}\n\n"
if warnings:
message += "🚨 SENSITIVE FILE DETECTED:\n"
for warning in warnings:
message += f" {warning}\n"
message += "\n"
message += "🔍 DETECTION RISKS:\n"
message += " • File access events (if auditing enabled)\n"
message += " • Sensitive file patterns (DLP detection)\n"
message += " • Credential detection (if file contains credentials)\n"
message += " • Unusual file access patterns\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • Sensitive files (credentials, databases) are heavily monitored\n"
message += " • Credentials found in file will be automatically reported\n"
message += " • File content may be logged by DLP systems\n\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
file_path = taskData.args.get_arg("file") or ""
message = "⚠️ OPSEC POST-CHECK - File Read Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • File Access: Read operations on target file\n"
message += " • File Content: File data retrieved and analyzed\n"
message += " • Credential Detection: Automatic credential scanning (if found)\n"
message += " • File Access Logs: Event logs (if auditing enabled)\n\n"
if file_path:
message += f"Target File: {file_path}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR file access monitoring\n"
message += " • DLP systems (sensitive file detection)\n"
message += " • File access auditing (if enabled)\n"
message += " • Credential extraction detection\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • File contains credentials or sensitive data\n"
message += " • File access auditing is enabled\n"
message += " • DLP rules match file patterns\n"
message += " • File is a security database (SAM, NTDS, etc.)\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
Payload_Type/cazalla/cazalla/agent_functions/cp.py
+132 -5
View File
@@ -8,12 +8,20 @@ class CpArguments(TaskArguments):
CommandParameter(
name="fichero existente",
type=ParameterType.String,
description="Ruta del fichero a copiar"
description="Ruta del fichero a copiar",
parameter_group_info=[ParameterGroupInfo(
required=True,
ui_position=1
)]
),
CommandParameter(
name="fichero nuevo",
type=ParameterType.String,
description="Ruta del nuevo fichero"
description="Ruta del nuevo fichero",
parameter_group_info=[ParameterGroupInfo(
required=True,
ui_position=2
)]
)
]
@@ -40,9 +48,128 @@ class CpCommand(CommandBase):
suggested_command=False
)
# async def create_tasking(self, task: MythicTask) -> MythicTask:
# task.display_params = task.args.get_arg("command")
# return task
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about file copy risks, especially for sensitive files.
"""
source_path = taskData.args.get_arg("fichero existente") or ""
dest_path = taskData.args.get_arg("fichero nuevo") or ""
# Sensitive file patterns
sensitive_patterns = {
"sam": "SAM database - Credential extraction heavily monitored",
"system": "SYSTEM hive - Registry access logged",
"security": "SECURITY hive - Security database access logged",
"ntds.dit": "NTDS.dit - Active Directory database, highly monitored",
"lsass": "LSASS memory - Credential extraction heavily monitored",
}
# Sensitive locations
sensitive_locations = {
"system32": "System32 directory - Highly monitored",
"syswow64": "SysWOW64 directory - Highly monitored",
"windows": "Windows directory - Highly monitored",
}
warnings = []
source_lower = source_path.lower()
dest_lower = dest_path.lower()
for pattern, reason in sensitive_patterns.items():
if pattern in source_lower:
warnings.append(f"🚨 Source contains {pattern}: {reason}")
if pattern in dest_lower:
warnings.append(f"🚨 Destination contains {pattern}: {reason}")
for location, reason in sensitive_locations.items():
if location in source_lower:
warnings.append(f"⚠️ Source: {location} - {reason}")
if location in dest_lower:
warnings.append(f"⚠️ Destination: {location} - {reason}")
message = "⚠️ OPSEC WARNING - File Copy Detected\n\n"
message += "File copying is monitored by:\n"
message += " • EDR/XDR solutions (file operations monitoring)\n"
message += " • Data loss prevention (DLP) systems\n"
message += " • File access auditing (if enabled)\n"
message += " • File integrity monitoring\n\n"
if source_path:
message += f"Source File: {source_path}\n"
if dest_path:
message += f"Destination: {dest_path}\n"
if source_path or dest_path:
message += "\n"
if warnings:
message += "🚨 RISKS DETECTED:\n"
for warning in warnings:
message += f" {warning}\n"
message += "\n"
message += "🔍 DETECTION RISKS:\n"
message += " • File read/write events (if auditing enabled)\n"
message += " • Sensitive file patterns (DLP detection)\n"
message += " • File operations monitoring\n"
message += " • System file modification detection\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • Copying sensitive files (credentials, databases) is monitored\n"
message += " • System directories are highly monitored\n"
message += " • File operations may be logged\n\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
source_path = taskData.args.get_arg("fichero existente") or ""
dest_path = taskData.args.get_arg("fichero nuevo") or ""
message = "⚠️ OPSEC POST-CHECK - File Copy Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • File Read: Read operations on source file\n"
message += " • File Write: Write operations to destination\n"
message += " • File Copy: CopyFile API calls\n"
message += " • File Access Logs: Event logs (if auditing enabled)\n\n"
if source_path:
message += f"Source File: {source_path}\n"
if dest_path:
message += f"Destination: {dest_path}\n"
if source_path or dest_path:
message += "\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR file operation monitoring\n"
message += " • DLP systems (sensitive file detection)\n"
message += " • File access auditing (if enabled)\n"
message += " • File integrity monitoring\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • Source or destination contains sensitive files\n"
message += " • Copying to/from system directories\n"
message += " • File access auditing is enabled\n"
message += " • DLP rules match file patterns\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
Payload_Type/cazalla/cazalla/agent_functions/download.py
+101
View File
@@ -10,6 +10,9 @@ class DownloadArguments(TaskArguments):
name="file",
type=ParameterType.String,
description="Path to the file to download from the target",
parameter_group_info=[ParameterGroupInfo(
required=True
)]
),
]
@@ -33,6 +36,104 @@ class DownloadCommand(CommandBase):
author = "@KaseyaOFSTeam"
argument_class = DownloadArguments
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about file download risks.
"""
file_path = taskData.args.get_arg("file") or ""
# Sensitive file patterns
sensitive_patterns = {
"sam": "SAM database - Credential extraction heavily monitored",
"system": "SYSTEM hive - Registry access logged",
"security": "SECURITY hive - Security database access logged",
"ntds.dit": "NTDS.dit - Active Directory database, highly monitored",
"lsass": "LSASS memory - Credential extraction heavily monitored",
"passwd": "Password files - Credential access monitored",
"shadow": "Shadow files - Credential access monitored",
}
warnings = []
file_lower = file_path.lower()
for pattern, reason in sensitive_patterns.items():
if pattern in file_lower:
warnings.append(f"🚨 {pattern}: {reason}")
message = "⚠️ OPSEC WARNING - File Download Detected\n\n"
message += "File downloads are monitored by:\n"
message += " • EDR/XDR solutions\n"
message += " • Data loss prevention (DLP) systems\n"
message += " • File access monitoring tools\n"
message += " • Network monitoring (large file transfers)\n\n"
if file_path:
message += f"Target File: {file_path}\n\n"
if warnings:
message += "🚨 SENSITIVE FILE DETECTED:\n"
for warning in warnings:
message += f" {warning}\n"
message += "\n"
message += "🔍 DETECTION RISKS:\n"
message += " • File access events (if auditing enabled)\n"
message += " • Large file transfers (network monitoring)\n"
message += " • Sensitive file patterns (DLP detection)\n"
message += " • Unusual file access patterns\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • Sensitive files (credentials, databases) are heavily monitored\n"
message += " • Large files may trigger network alerts\n"
message += " • Multiple downloads may be suspicious\n\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
file_path = taskData.args.get_arg("file") or ""
message = "⚠️ OPSEC POST-CHECK - File Download Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • File Access: Read operations on target file\n"
message += " • Network Activity: File data transmission to Mythic\n"
message += " • File Access Logs: Event logs (if auditing enabled)\n\n"
if file_path:
message += f"Target File: {file_path}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR file access monitoring\n"
message += " • DLP systems (sensitive file detection)\n"
message += " • Network monitoring (large transfers)\n"
message += " • File access auditing (if enabled)\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • File contains credentials or sensitive data\n"
message += " • File is large (network anomaly detection)\n"
message += " • File access auditing is enabled\n"
message += " • DLP rules match file patterns\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Payload_Type/cazalla/cazalla/agent_functions/keylog_start.py
+73
View File
@@ -24,6 +24,79 @@ class KeylogStartCommand(CommandBase):
suggested_command=False,
)
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about keylogger risks - this is highly detectable.
"""
message = "🚨 CRITICAL OPSEC RISK - Keylogger Detected\n\n"
message += "Keyloggers are EXTREMELY detectable and monitored by:\n"
message += " • EDR/XDR solutions (high priority detection)\n"
message += " • Anti-malware solutions\n"
message += " • Behavioral analysis engines\n"
message += " • User activity monitoring tools\n"
message += " • Data loss prevention (DLP) systems\n\n"
message += "Detection mechanisms:\n"
message += " • Low-level keyboard hook detection\n"
message += " • API hooking detection (SetWindowsHookEx)\n"
message += " • Keyboard input monitoring\n"
message += " • Suspicious process behavior\n"
message += " • Memory scanning for keylogger patterns\n\n"
message += "⚠️ HIGH RISK FACTORS:\n"
message += " • Keyloggers are classified as malware by most security tools\n"
message += " • Immediate detection and removal is common\n"
message += " • Legal implications in many jurisdictions\n"
message += " • May trigger incident response procedures\n\n"
message += "Consider alternatives:\n"
message += " • Use credential theft techniques instead\n"
message += " • Use browser credential extraction\n"
message += " • Use clipboard monitoring (less detectable)\n"
message += " • Use screenshot-based credential capture\n\n"
message += "⚠️ WARNING: This command is highly detectable and should only be used\n"
message += "in controlled environments or with explicit authorization."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=True, # Always block keyloggers
OpsecPreBypassRole="lead", # Only lead can approve
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
message = "🚨 OPSEC POST-CHECK - Keylogger Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Keyboard Hook: Low-level keyboard hook installation\n"
message += " • API Hooking: SetWindowsHookEx/UnhookWindowsHookEx calls\n"
message += " • Process Behavior: Continuous keyboard monitoring\n"
message += " • Network Activity: Keylog data transmission\n"
message += " • Memory Artifacts: Keylogger code in memory\n\n"
message += "These artifacts are detected with HIGH CONFIDENCE by:\n"
message += " • EDR/XDR behavioral detection (99%+ detection rate)\n"
message += " • Anti-malware signature and heuristic detection\n"
message += " • Memory scanning tools\n"
message += " • API hooking detection mechanisms\n"
message += " • User activity monitoring systems\n\n"
message += "⚠️ CRITICAL WARNINGS:\n"
message += " • Detection is likely within minutes or hours\n"
message += " • May result in immediate agent termination\n"
message += " • May trigger full incident response\n"
message += " • Legal and compliance risks\n\n"
message += "⚠️ This is one of the most detectable operations in red teaming.\n"
message += "Use with extreme caution and only when absolutely necessary."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=True, # Block even in post
OpsecPostBypassRole="lead", # Only lead can approve
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
resp = PTTaskCreateTaskingMessageResponse(TaskID=taskData.Task.ID, Success=True)
resp.DisplayParams = "Start keylogger"
Payload_Type/cazalla/cazalla/agent_functions/kill.py
+99 -1
View File
@@ -11,7 +11,7 @@ class KillArguments(TaskArguments):
type=ParameterType.Number,
description="Process ID to kill",
parameter_group_info=[ParameterGroupInfo(
required=False,
required=True,
ui_position=1
)]
),
@@ -74,6 +74,104 @@ class KillCommand(CommandBase):
supported_ui_features = ["process_browser:kill"]
argument_class = KillArguments
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about process termination risks, especially for critical processes.
"""
pid = taskData.args.get_arg("pid") or taskData.args.get_arg("process_id")
# Critical system processes that should NEVER be killed
critical_pids = {
0: "Idle Process - Cannot be killed",
4: "System Process - Critical system process, NEVER kill",
8: "Memory Compression - Critical system process",
}
blocked = False
critical_warning = ""
if pid:
pid_int = int(pid) if pid and str(pid).isdigit() else None
if pid_int in critical_pids:
critical_warning = f"🚨 CRITICAL: {critical_pids[pid_int]}\n⚠️ WARNING: Killing this process may cause system instability or crash!\n\n"
blocked = True
message = "⚠️ OPSEC RISK - Process Termination Detected\n\n"
message += "Process termination is monitored by:\n"
message += " • EDR/XDR solutions\n"
message += " • Process monitoring tools\n"
message += " • Security auditing systems\n"
message += " • Endpoint protection platforms\n\n"
if pid:
message += f"Target Process ID: {pid}\n"
if critical_warning:
message += critical_warning
message += "\n"
message += "🔍 DETECTION MECHANISMS:\n"
message += " • TerminateProcess API calls\n"
message += " • Process termination events\n"
message += " • Suspicious process killing patterns\n"
message += " • Termination of security processes\n\n"
message += "💡 RECOMMENDATIONS:\n"
message += " • Verify the process is not critical using Process Browser\n"
message += " • Ensure you understand the impact before proceeding\n"
message += " • Check if process is a security tool (EDR, AV) - may trigger immediate alerts\n\n"
if blocked:
message += "⚠️ TO PROCEED: You need approval from another operator.\n"
message += "This command is blocked due to targeting a critical system process."
else:
message += "⚠️ WARNING: Killing security processes may trigger immediate alerts!\n"
message += "This is a WARNING only - command will proceed."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=blocked,
OpsecPreBypassRole="other_operator" if blocked else "operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
pid = taskData.args.get_arg("pid") or taskData.args.get_arg("process_id")
message = "⚠️ OPSEC POST-CHECK - Process Termination Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Process Termination: TerminateProcess API call\n"
message += " • Process Access: OpenProcess with PROCESS_TERMINATE\n"
message += " • Event Log: Process termination event (if logging enabled)\n\n"
if pid:
message += f"Target Process ID: {pid}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR behavioral detection\n"
message += " • Windows Event Logging (if enabled)\n"
message += " • Process monitoring tools\n"
message += " • Security information systems\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • Target is a security process (EDR, AV)\n"
message += " • Target is a critical system process\n"
message += " • Multiple processes are terminated rapidly\n"
message += " • Process is a protected service\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Payload_Type/cazalla/cazalla/agent_functions/list_tokens.py
+74
View File
@@ -53,6 +53,80 @@ class ListTokensCommand(CommandBase):
supported_ui_features = ["process_browser:list_tokens"]
argument_class = ListTokensArguments
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about token enumeration risks.
"""
pid = taskData.args.get_arg("process_id")
message = "⚠️ OPSEC WARNING - Token Enumeration Detected\n\n"
message += "Token enumeration is monitored by:\n"
message += " • EDR/XDR solutions (behavioral detection)\n"
message += " • Process monitoring tools\n"
message += " • Security auditing systems\n\n"
if pid:
message += f"Target Process ID: {pid}\n\n"
message += "🔍 DETECTION MECHANISMS:\n"
message += " • Token enumeration API calls\n"
message += " • Process access for token queries\n"
message += " • Suspicious enumeration patterns\n"
message += " • Security event logging (if enabled)\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • Enumerating tokens from processes may be logged\n"
message += " • Accessing sensitive processes (LSASS) is monitored\n"
message += " • Lower detection risk than token theft\n"
message += " • Useful for reconnaissance before token operations\n\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
pid = taskData.args.get_arg("process_id")
message = "⚠️ OPSEC POST-CHECK - Token Enumeration Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Process Access: OpenProcess for token queries\n"
message += " • Token Enumeration: Token enumeration API calls\n"
message += " • Process Information: Token details retrieved\n\n"
if pid:
message += f"Target Process ID: {pid}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR behavioral detection (low-moderate confidence)\n"
message += " • Process access monitoring\n"
message += " • Security event logging (if enabled)\n\n"
message += "⚠️ DETECTION PROBABILITY:\n"
message += " • Lower than token theft operations\n"
message += " • Process access may be logged\n"
message += " • Enumeration patterns may be detected\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Payload_Type/cazalla/cazalla/agent_functions/make_token.py
+86
View File
@@ -76,6 +76,92 @@ class MakeTokenCommand(CommandBase):
attackmapping = ["T1134.003"]
argument_class = MakeTokenArguments
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about token creation risks.
"""
domain = taskData.args.get_arg("domain") or ""
username = taskData.args.get_arg("username") or ""
logon_type = taskData.args.get_arg("logon_type") or 9
message = "⚠️ OPSEC WARNING - Token Creation Detected\n\n"
message += "Token creation operations are monitored by:\n"
message += " • EDR/XDR solutions (behavioral detection)\n"
message += " • Authentication monitoring tools\n"
message += " • Security auditing systems\n"
message += " • Windows Event Logging (if enabled)\n\n"
message += f"Configuration:\n"
message += f" • Domain: {domain or '(local)'}\n"
message += f" • Username: {username}\n"
message += f" • Logon Type: {logon_type}\n\n"
message += "🔍 DETECTION MECHANISMS:\n"
message += " • LogonUser API calls (authentication events)\n"
message += " • Token impersonation API calls\n"
message += " • Privilege escalation detection\n"
message += " • Suspicious authentication patterns\n"
message += " • Windows Event Logs (Security events)\n\n"
message += "💡 COMPARED TO steal_token:\n"
message += " • Less detectable than steal_token (no LSASS access)\n"
message += " • Requires valid credentials (domain/local)\n"
message += " • May generate authentication logs\n"
message += " • Network authentication may be logged\n\n"
message += "⚠️ WARNING: Authentication events may be logged.\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
domain = taskData.args.get_arg("domain") or ""
username = taskData.args.get_arg("username") or ""
message = "⚠️ OPSEC POST-CHECK - Token Creation Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Authentication Event: LogonUser API call\n"
message += " • Token Creation: New token created\n"
message += " • Token Impersonation: Token impersonation API calls\n"
message += " • Context Tracking: Impersonation context updated in Mythic UI\n\n"
message += f"Configuration:\n"
message += f" • Domain: {domain or '(local)'}\n"
message += f" • Username: {username}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR behavioral detection (moderate confidence)\n"
message += " • Windows Event Logging (Security events)\n"
message += " • Authentication monitoring tools\n"
message += " • Network authentication logs (if domain)\n\n"
message += "⚠️ DETECTION PROBABILITY:\n"
message += " • Lower than steal_token (no LSASS access)\n"
message += " • Authentication events may be logged\n"
message += " • Network authentication (domain) is logged\n"
message += " • Behavioral patterns may be detected\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Payload_Type/cazalla/cazalla/agent_functions/rev2self.py
+63
View File
@@ -24,6 +24,69 @@ class Rev2SelfCommand(CommandBase):
attackmapping = ["T1134.001"]
argument_class = Rev2SelfArguments
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about token reversion risks.
"""
message = "⚠️ OPSEC WARNING - Token Reversion Detected\n\n"
message += "Reverting token impersonation is monitored by:\n"
message += " • EDR/XDR solutions (behavioral detection)\n"
message += " • Security auditing systems\n"
message += " • Token manipulation monitoring\n\n"
message += "🔍 DETECTION MECHANISMS:\n"
message += " • Token reversion API calls\n"
message += " • Context change detection\n"
message += " • Privilege level changes\n"
message += " • Security event logging\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • Reverting from elevated token may reduce privileges\n"
message += " • Context tracking will update in Mythic UI\n"
message += " • Lower detection risk than token theft/creation\n\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the privilege implications."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
message = "⚠️ OPSEC POST-CHECK - Token Reversion Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Token Reversion: RevertToSelf API call\n"
message += " • Context Change: Privilege level change\n"
message += " • Context Tracking: Impersonation context cleared in Mythic UI\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR behavioral detection (low-moderate confidence)\n"
message += " • Security event logging (if enabled)\n"
message += " • Token manipulation monitoring\n\n"
message += "⚠️ DETECTION PROBABILITY:\n"
message += " • Lower than token theft/creation operations\n"
message += " • Context changes may be logged\n"
message += " • Privilege changes may trigger alerts\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Payload_Type/cazalla/cazalla/agent_functions/rm.py
+125 -4
View File
@@ -79,10 +79,131 @@ class RmCommand(CommandBase):
suggested_command=False
)
# async def create_tasking(self, task: MythicTask) -> MythicTask:
# task.display_params = task.args.get_arg("command")
# return task
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about file deletion risks, especially for system files.
"""
path = taskData.args.get_arg("path") or ""
# Critical system locations
critical_locations = {
"system32": "System32 directory - CRITICAL system files",
"syswow64": "SysWOW64 directory - CRITICAL system files",
"windows": "Windows directory - CRITICAL system files",
"program files": "Program Files - Application files",
"boot": "Boot directory - System boot files",
"winnt": "WinNT directory - System files",
}
# Critical file patterns
critical_files = {
".dll": "DLL files - System libraries",
".sys": "System driver files",
".exe": "Executable files - System binaries",
}
blocked = False
warnings = []
critical_warnings = []
path_lower = path.lower()
for location, reason in critical_locations.items():
if location in path_lower:
critical_warnings.append(f"🚨 {location}: {reason}")
if location in ["system32", "syswow64", "windows", "boot"]:
blocked = True
for file_ext, reason in critical_files.items():
if path_lower.endswith(file_ext) and any(loc in path_lower for loc in ["system32", "syswow64", "windows"]):
warnings.append(f"🚨 {file_ext}: {reason} - System file deletion")
blocked = True
message = "⚠️ OPSEC RISK - File/Directory Deletion Detected\n\n"
message += "File deletion is monitored by:\n"
message += " • EDR/XDR solutions (file deletion monitoring)\n"
message += " • Anti-malware solutions\n"
message += " • File access auditing (if enabled)\n"
message += " • System integrity monitoring\n\n"
if path:
message += f"Target Path: {path}\n\n"
if critical_warnings:
message += "🚨 CRITICAL LOCATION DETECTED:\n"
for warning in critical_warnings:
message += f" {warning}\n"
message += "\n"
if warnings:
message += "🚨 CRITICAL FILE DETECTED:\n"
for warning in warnings:
message += f" {warning}\n"
message += "\n"
message += "🔍 DETECTION RISKS:\n"
message += " • File deletion events (if auditing enabled)\n"
message += " • System file deletion alerts\n"
message += " • File integrity monitoring\n"
message += " • Suspicious deletion patterns\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • Deleting system files may cause system instability\n"
message += " • System file deletion triggers immediate alerts\n"
message += " • File deletion is often logged\n"
message += " • Use with extreme caution in system directories\n\n"
if blocked:
message += "⚠️ TO PROCEED: You need approval from another operator.\n"
message += "This command is blocked due to targeting critical system files."
else:
message += "⚠️ WARNING: File deletion may be logged.\n"
message += "✅ This is a WARNING only - command will proceed."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=blocked,
OpsecPreBypassRole="other_operator" if blocked else "operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
path = taskData.args.get_arg("path") or ""
message = "⚠️ OPSEC POST-CHECK - File Deletion Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • File Deletion: DeleteFile/RemoveDirectory API calls\n"
message += " • File Access: File system operations\n"
message += " • File Deletion Logs: Event logs (if auditing enabled)\n\n"
if path:
message += f"Target Path: {path}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR file deletion monitoring\n"
message += " • File access auditing (if enabled)\n"
message += " • System integrity monitoring\n"
message += " • File deletion event logging\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • Target is a system file or directory\n"
message += " • File deletion auditing is enabled\n"
message += " • System integrity monitoring is active\n"
message += " • Multiple files are deleted rapidly\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Payload_Type/cazalla/cazalla/agent_functions/rpfwd.py
+116
View File
@@ -91,6 +91,122 @@ class RpfwdCommand(CommandBase):
builtin=True,
)
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about reverse port forwarding risks.
"""
action = taskData.args.get_arg("action") or "start"
local_port = taskData.args.get_arg("port") or 8080
remote_host = taskData.args.get_arg("remote_host") or "127.0.0.1"
remote_port = taskData.args.get_arg("remote_port") or 80
if action == "stop":
# Stop action is low risk
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage="Stopping reverse port forwarding - low OPSEC risk."
)
message = "⚠️ OPSEC WARNING - Reverse Port Forwarding Detected\n\n"
message += "Reverse port forwarding is monitored by:\n"
message += " • Network monitoring tools\n"
message += " • Firewall logging\n"
message += " • Network intrusion detection systems (NIDS)\n"
message += " • EDR/XDR network monitoring\n"
message += " • Network flow analysis tools\n\n"
message += f"Configuration:\n"
message += f" • Local Port (Agent): {local_port}\n"
message += f" • Remote Host: {remote_host}\n"
message += f" • Remote Port: {remote_port}\n\n"
# Check for privileged ports
if local_port < 1024:
message += "⚠️ WARNING: Privileged port (<1024) requires administrator privileges!\n"
message += "This may trigger additional security alerts.\n\n"
message += "🔍 DETECTION MECHANISMS:\n"
message += " • Network connection monitoring\n"
message += " • Port binding detection\n"
message += " • Suspicious network patterns\n"
message += " • Connection to non-standard ports\n\n"
message += "💡 RECOMMENDATIONS:\n"
message += " • Use non-standard ports (avoid 80, 443, 22, 3389)\n"
message += " • Use ports commonly used by legitimate software\n"
message += " • Network traffic may be logged\n"
message += " • Firewall rules may block the connection\n\n"
message += "⚠️ DETECTION RISK: MODERATE\n"
message += "Network activity is visible but may blend in with normal traffic if configured properly.\n\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
action = taskData.args.get_arg("action") or "start"
if action == "stop":
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage="Stopping reverse port forwarding - no additional artifacts."
)
local_port = taskData.args.get_arg("port") or 8080
remote_host = taskData.args.get_arg("remote_host") or "127.0.0.1"
remote_port = taskData.args.get_arg("remote_port") or 80
message = "⚠️ OPSEC POST-CHECK - Reverse Port Forwarding Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Network Connection: Listener on agent port\n"
message += " • Port Binding: Socket bind on local port\n"
message += " • Network Activity: Connection forwarding\n"
message += " • Process Network Activity: Agent network connections\n\n"
message += f"Configuration:\n"
message += f" • Local Port: {local_port}\n"
message += f" • Remote: {remote_host}:{remote_port}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • Network monitoring tools\n"
message += " • Firewall logging\n"
message += " • NIDS/NIPS systems\n"
message += " • Network flow analysis\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • Using well-known ports (80, 443, 22, etc.)\n"
message += " • Network traffic is heavily monitored\n"
message += " • Firewall rules are strict\n"
message += " • Using privileged ports (<1024)\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_tasking(self, task: MythicTask) -> MythicTask:
action = task.args.get_arg("action")
if action == "start":
Payload_Type/cazalla/cazalla/agent_functions/screenshot.py
+72
View File
@@ -26,6 +26,78 @@ class ScreenshotCommand(CommandBase):
argument_class = ScreenshotArguments
attackmapping = ["T1113"]
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about screenshot capture risks.
"""
message = "⚠️ OPSEC WARNING - Screenshot Capture Detected\n\n"
message += "Screenshot capture is monitored by:\n"
message += " • EDR/XDR solutions (behavioral detection)\n"
message += " • Screen capture detection mechanisms\n"
message += " • Data loss prevention (DLP) systems\n"
message += " • User activity monitoring tools\n\n"
message += "🔍 DETECTION MECHANISMS:\n"
message += " • BitBlt/GDI screen capture API calls\n"
message += " • PrintWindow API calls\n"
message += " • Suspicious screen capture patterns\n"
message += " • Large data uploads (screenshot files)\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • Screenshots may be large and trigger network alerts\n"
message += " • Multiple screenshots may be suspicious\n"
message += " • Screen capture events may be logged\n"
message += " • Sensitive data in screenshots may trigger DLP alerts\n\n"
message += "⚠️ DETECTION RISK: MODERATE\n"
message += "Usually not immediately detected, but may be flagged by behavioral analysis over time.\n\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False, # Moderate risk, don't block
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
message = "⚠️ OPSEC POST-CHECK - Screenshot Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Screen Capture: BitBlt/PrintWindow API calls\n"
message += " • File Upload: Screenshot image file transmission\n"
message += " • Network Activity: Image data transfer\n"
message += " • Memory Artifacts: Screenshot data in memory\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR behavioral detection\n"
message += " • Network monitoring tools (large file uploads)\n"
message += " • DLP systems (sensitive data capture)\n"
message += " • User activity monitoring\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • Multiple screenshots are taken rapidly\n"
message += " • Screenshots contain sensitive data\n"
message += " • Network traffic is monitored\n"
message += " • Behavioral analysis is aggressive\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Payload_Type/cazalla/cazalla/agent_functions/shell.py
+87 -1
View File
@@ -10,7 +10,10 @@ class ShellArguments(TaskArguments):
CommandParameter(
name="command",
type=ParameterType.String,
description="Comando a ejecutar"
description="Comando a ejecutar",
parameter_group_info=[ParameterGroupInfo(
required=True
)]
),
]
@@ -36,6 +39,89 @@ class ShellCommand(CommandBase):
supported_os=[SupportedOS.MacOS, SupportedOS.Linux, SupportedOS.Windows]
)
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about dangerous commands that are highly detectable.
"""
command = (taskData.args.get_arg("command") or "").lower()
# Highly detectable commands
dangerous_commands = {
"powershell": "PowerShell execution is heavily monitored by EDR/XDR",
"pwsh": "PowerShell Core execution is heavily monitored",
"wmic": "WMI queries are logged and monitored",
"net.exe": "Network enumeration commands trigger alerts",
"nslookup": "DNS queries may be monitored",
"systeminfo": "System information gathering is logged",
"whoami": "Identity queries are logged",
"tasklist": "Process enumeration is monitored",
"netstat": "Network connection enumeration triggers alerts",
"ipconfig": "Network configuration queries are logged",
}
blocked = False
warnings = []
for dangerous_cmd, reason in dangerous_commands.items():
if dangerous_cmd in command:
warnings.append(f"⚠️ {dangerous_cmd}: {reason}")
blocked = True
# Check for cmd.exe spawning (always present in shell command)
warnings.append("⚠️ cmd.exe spawn: This command spawns cmd.exe which is highly detectable by EDR/XDR")
message = "⚠️ HIGH OPSEC RISK - Shell Command Detected\n\n"
message += "This command will spawn cmd.exe which is heavily monitored.\n\n"
if warnings:
message += "Additional risks detected:\n"
for warning in warnings:
message += f" • {warning}\n"
message += "\n"
message += "Consider using built-in Cazalla commands instead:\n"
message += " • Use 'ps' instead of 'tasklist'\n"
message += " • Use 'ls' instead of 'dir'\n"
message += " • Use 'cat' instead of 'type'\n"
message += " • Use 'whoami' command instead of 'whoami.exe'\n\n"
message += "If you must proceed, ensure you understand the detection risks."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=blocked,
OpsecPreBypassRole="other_operator", # Requires another operator to approve
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
command = taskData.args.get_arg("command") or ""
message = "⚠️ OPSEC POST-CHECK - Shell Command Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Process Create: cmd.exe spawned\n"
message += " • Command Line: The executed command will be logged\n"
message += " • Network Activity: If command makes network connections\n\n"
message += "These artifacts may be detected by:\n"
message += " • EDR/XDR solutions\n"
message += " • SIEM systems\n"
message += " • Process monitoring tools\n"
message += " • Command line auditing\n\n"
message += "Ensure you understand the detection risks before proceeding."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False, # Don't block, just warn
OpsecPostBypassRole="operator", # Any operator can bypass
OpsecPostMessage=message
)
async def create_tasking(self, task: MythicTask) -> MythicTask:
task.display_params = task.args.get_arg("command")
return task
Payload_Type/cazalla/cazalla/agent_functions/socks.py
+117
View File
@@ -69,6 +69,123 @@ class SocksCommand(CommandBase):
builtin=True,
)
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about SOCKS proxy risks.
"""
action = taskData.args.get_arg("action") or "start"
port = taskData.args.get_arg("port") or 7002
if action == "stop":
# Stop action is low risk
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage="Stopping SOCKS proxy - low OPSEC risk."
)
message = "⚠️ OPSEC WARNING - SOCKS Proxy Detected\n\n"
message += "SOCKS proxy is monitored by:\n"
message += " • Network monitoring tools\n"
message += " • Firewall logging\n"
message += " • Network intrusion detection systems (NIDS)\n"
message += " • EDR/XDR network monitoring\n"
message += " • Proxy detection mechanisms\n"
message += " • Network flow analysis tools\n\n"
message += f"Configuration:\n"
message += f" • SOCKS Port: {port}\n\n"
message += "🔍 DETECTION MECHANISMS:\n"
message += " • Network connection monitoring\n"
message += " • Proxy protocol detection\n"
message += " • High-volume network traffic patterns\n"
message += " • Suspicious network behavior\n"
message += " • Connection to non-standard ports\n\n"
message += "⚠️ HIGH TRAFFIC WARNING:\n"
message += " • SOCKS proxies generate continuous network activity\n"
message += " • High bandwidth usage may trigger alerts\n"
message += " • Network patterns may be analyzed for anomalies\n"
message += " • All traffic through proxy is visible\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • SOCKS proxies generate significant network traffic\n"
message += " • Using common proxy ports (1080, 8080) may blend better\n"
message += " • Firewall rules may block the connection\n"
message += " • Network patterns may be analyzed\n\n"
message += "⚠️ DETECTION RISK: MODERATE-HIGH\n"
message += "Network activity is highly visible and proxy traffic patterns may be detected by network analysis.\n\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
action = taskData.args.get_arg("action") or "start"
if action == "stop":
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage="Stopping SOCKS proxy - no additional artifacts."
)
port = taskData.args.get_arg("port") or 7002
message = "⚠️ OPSEC POST-CHECK - SOCKS Proxy Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Network Connection: SOCKS proxy listener\n"
message += " • Port Binding: Socket bind on SOCKS port\n"
message += " • Network Activity: Continuous proxy traffic\n"
message += " • Process Network Activity: Agent network connections\n"
message += " • High Bandwidth Usage: Proxy data transfer\n\n"
message += f"Configuration:\n"
message += f" • SOCKS Port: {port}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • Network monitoring tools\n"
message += " • Firewall logging\n"
message += " • NIDS/NIPS systems\n"
message += " • Network flow analysis\n"
message += " • Bandwidth monitoring systems\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • High volume of traffic is generated\n"
message += " • Network traffic patterns are analyzed\n"
message += " • SOCKS protocol is detected\n"
message += " • Firewall rules are strict\n"
message += " • Network monitoring is aggressive\n\n"
message += "⚠️ WARNING: SOCKS proxies generate significant network activity\n"
message += "that may be detected by network security tools.\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_tasking(self, task: MythicTask) -> MythicTask:
action = task.args.get_arg("action")
port = task.args.get_arg("port")
Payload_Type/cazalla/cazalla/agent_functions/steal_token.py
+155 -1
View File
@@ -11,7 +11,7 @@ class StealTokenArguments(TaskArguments):
type=ParameterType.Number,
description="Process ID to steal token from",
parameter_group_info=[ParameterGroupInfo(
required=False,
required=True,
ui_position=1
)]
),
@@ -74,6 +74,160 @@ class StealTokenCommand(CommandBase):
supported_ui_features = ["process_browser:steal_token"]
argument_class = StealTokenArguments
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about token theft risks, especially for high-value processes.
"""
# Debug: Print to see if this function is being called
print(f"[DEBUG opsec_pre] steal_token opsec_pre called")
print(f"[DEBUG opsec_pre] taskData.args: {taskData.args}")
if hasattr(taskData, 'Task'):
print(f"[DEBUG opsec_pre] taskData.Task: {taskData.Task}")
if hasattr(taskData.Task, 'Params'):
print(f"[DEBUG opsec_pre] taskData.Task.Params: {taskData.Task.Params}")
# Try to get pid from parsed args first, then from raw parameters
pid = taskData.args.get_arg("pid") or taskData.args.get_arg("process_id")
print(f"[DEBUG opsec_pre] pid from args: {pid}")
# If not found in parsed args, try to get from raw task parameters
# (needed because opsec_pre runs before parse_dictionary)
if not pid:
# Try multiple ways to access the parameters
if hasattr(taskData.Task, 'Params') and taskData.Task.Params:
import json
try:
raw_params = json.loads(taskData.Task.Params) if isinstance(taskData.Task.Params, str) else taskData.Task.Params
pid = raw_params.get("pid") or raw_params.get("process_id")
except:
pass
# Also try from taskData.Task directly if it has a params attribute
if not pid and hasattr(taskData.Task, 'params') and taskData.Task.params:
import json
try:
raw_params = json.loads(taskData.Task.params) if isinstance(taskData.Task.params, str) else taskData.Task.params
pid = raw_params.get("pid") or raw_params.get("process_id")
except:
pass
# Try from taskData directly
if not pid and hasattr(taskData, 'params') and taskData.params:
import json
try:
raw_params = json.loads(taskData.params) if isinstance(taskData.params, str) else taskData.params
pid = raw_params.get("pid") or raw_params.get("process_id")
except:
pass
# High-value processes that are heavily monitored
high_value_pids = {
4: "System (PID 4) - Critical system process, heavily monitored",
}
# Common high-value process names (if we could check them)
high_value_processes = {
"lsass.exe": "LSASS - Credential extraction is heavily monitored by EDR/XDR",
"winlogon.exe": "Winlogon - Authentication process, monitored",
"csrss.exe": "CSRSS - Critical system process, monitored",
"services.exe": "Services - System service manager, monitored",
}
blocked = False
critical_warning = ""
if pid:
pid_int = int(pid) if pid and str(pid).isdigit() else None
if pid_int in high_value_pids:
critical_warning = f"🚨 CRITICAL: {high_value_pids[pid_int]}\n⚠️ WARNING: Stealing from this process may cause system instability!\n\n"
blocked = True
# Always block steal_token to ensure OPSEC warning is shown
# (similar to how shell always blocks for cmd.exe spawn)
blocked = True
message = "🚨 HIGH OPSEC RISK - Token Theft Operation\n\n"
message += "Token theft operations are EXTREMELY monitored by:\n"
message += " • EDR/XDR solutions (Defender, CrowdStrike, SentinelOne)\n"
message += " • Credential protection (LSASS protection)\n"
message += " • Process monitoring tools\n"
message += " • Security auditing systems\n\n"
if pid:
message += f"Target Process ID: {pid}\n"
if critical_warning:
message += critical_warning
message += "\n"
message += "🔍 DETECTION MECHANISMS:\n"
message += " • OpenProcess(PROCESS_QUERY_INFORMATION) on sensitive processes\n"
message += " • Token impersonation API calls (behavioral detection)\n"
message += " • Privilege escalation detection\n"
message += " • Suspicious process access patterns\n"
message += " • LSASS memory access (if targeting LSASS)\n\n"
message += "💡 ALTERNATIVES:\n"
message += " • Use 'make_token' for domain credentials (less detectable)\n"
message += " • Ensure proper evasion techniques are in place\n"
message += " • Verify LSASS protection status before targeting LSASS\n\n"
if critical_warning:
message += "🚨 CRITICAL WARNING: Targeting a critical system process!\n"
message += "⚠️ TO PROCEED: You need approval from another operator.\n"
message += "This command is blocked due to targeting a critical system process."
else:
message += "⚠️ TO PROCEED: You need approval from another operator.\n"
message += "This operation is highly detectable and requires explicit approval."
# Always return a message, even if pid is not found
# This ensures opsec_pre always triggers
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=blocked,
OpsecPreBypassRole="other_operator", # Requires another operator to approve
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
pid = taskData.args.get_arg("pid") or taskData.args.get_arg("process_id")
message = "⚠️ OPSEC POST-CHECK - Token Theft Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Process Access: OpenProcess on target process\n"
message += " • Token Manipulation: Token theft API calls\n"
message += " • Privilege Escalation: Token impersonation\n"
message += " • Context Tracking: Impersonation context updated in Mythic UI\n\n"
if pid:
message += f"Target Process ID: {pid}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR behavioral detection (high confidence)\n"
message += " • Windows Event Logging (if enabled)\n"
message += " • Process monitoring tools\n"
message += " • Credential protection mechanisms\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • Target is LSASS (PID varies by system)\n"
message += " • LSASS protection is enabled\n"
message += " • EDR has behavioral monitoring enabled\n"
message += " • Process is a critical system process\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False, # Don't block, just warn
OpsecPostBypassRole="operator", # Any operator can bypass
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Payload_Type/cazalla/cazalla/agent_functions/upload.py
+120
View File
@@ -11,11 +11,19 @@ class UploadArguments(TaskArguments):
name="file",
type=ParameterType.File,
description="File to upload to the target",
parameter_group_info=[ParameterGroupInfo(
required=True,
ui_position=1
)]
),
CommandParameter(
name="path",
type=ParameterType.String,
description="Path where the file should be saved on the target",
parameter_group_info=[ParameterGroupInfo(
required=True,
ui_position=2
)]
),
]
@@ -59,6 +67,118 @@ class UploadCommand(CommandBase):
author = "@KaseyaOFSTeam"
argument_class = UploadArguments
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
Warns about file upload risks.
"""
path = taskData.args.get_arg("path") or ""
# Sensitive locations
sensitive_locations = {
"system32": "System32 directory - Highly monitored",
"syswow64": "SysWOW64 directory - Highly monitored",
"windows": "Windows directory - Highly monitored",
"program files": "Program Files - Application installation monitoring",
"temp": "Temporary directories - May be scanned",
}
# Suspicious file extensions
suspicious_extensions = {
".exe": "Executable files - May trigger AV/EDR",
".dll": "DLL files - May trigger AV/EDR",
".ps1": "PowerShell scripts - Heavily monitored",
".bat": "Batch files - May trigger behavioral detection",
".vbs": "VBScript files - May trigger behavioral detection",
}
warnings = []
path_lower = path.lower()
for location, reason in sensitive_locations.items():
if location in path_lower:
warnings.append(f"⚠️ {location}: {reason}")
for ext, reason in suspicious_extensions.items():
if path_lower.endswith(ext):
warnings.append(f"🚨 {ext}: {reason}")
message = "⚠️ OPSEC WARNING - File Upload Detected\n\n"
message += "File uploads are monitored by:\n"
message += " • EDR/XDR solutions (file write monitoring)\n"
message += " • Anti-malware solutions (file scanning)\n"
message += " • Application control mechanisms\n"
message += " • Network monitoring (file transfers)\n\n"
if path:
message += f"Target Path: {path}\n\n"
if warnings:
message += "🚨 RISKS DETECTED:\n"
for warning in warnings:
message += f" {warning}\n"
message += "\n"
message += "🔍 DETECTION MECHANISMS:\n"
message += " • File write events (if auditing enabled)\n"
message += " • File scanning (AV/EDR real-time protection)\n"
message += " • Suspicious file patterns (executables, scripts)\n"
message += " • Network transfers (if monitored)\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • Executable files (.exe, .dll) are heavily scanned\n"
message += " • Scripts (.ps1, .bat, .vbs) are monitored\n"
message += " • System directories are highly monitored\n"
message += " • Temporary directories may be scanned\n\n"
message += "✅ This is a WARNING only - command will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=False,
OpsecPreBypassRole="operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
Warns about artifacts that will be created.
"""
path = taskData.args.get_arg("path") or ""
message = "⚠️ OPSEC POST-CHECK - File Upload Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • File Creation: New file written to disk\n"
message += " • File Write: Data written to target path\n"
message += " • Network Activity: File data transmission from Mythic\n"
message += " • File Access Logs: Event logs (if auditing enabled)\n\n"
if path:
message += f"Target Path: {path}\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR file write monitoring\n"
message += " • Anti-malware real-time scanning\n"
message += " • Application control mechanisms\n"
message += " • File access auditing (if enabled)\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY IF:\n"
message += " • File is an executable (.exe, .dll)\n"
message += " • File is a script (.ps1, .bat, .vbs)\n"
message += " • Target is a system directory\n"
message += " • File matches AV/EDR signatures\n"
message += " • File access auditing is enabled\n\n"
message += "✅ This is a WARNING only - task will proceed.\n"
message += "Ensure you understand the detection risks before continuing."
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False,
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Payload_Type/cazalla/translator/commands_from_c2.py
+18 -11
View File
@@ -117,7 +117,8 @@ def responseTasking(tasks):
if task["parameters"] != "":
parameters = json.loads(task["parameters"])
# Normalize paths for upload and ls commands - replace double backslashes
if command_to_run in ["upload", "ls", "download", "rm", "cat", "mkdir", "cp"] and "path" in parameters:
# cat uses "file" parameter, others use "path"
if command_to_run in ["upload", "ls", "download", "rm", "mkdir", "cp"] and "path" in parameters:
path = parameters["path"]
if path:
# Replace all sequences of double backslashes with single backslash
@@ -127,19 +128,25 @@ def responseTasking(tasks):
path = path.rstrip('\x00')
parameters["path"] = path
print(f"[DEBUG] Normalized path for {command_to_run}: {repr(path)}")
# Also normalize "file" parameter for cat and download
if command_to_run in ["cat", "download"] and "file" in parameters:
file_path = parameters["file"]
if file_path:
# Replace all sequences of double backslashes with single backslash
while "\\\\" in file_path:
file_path = file_path.replace("\\\\", "\\")
# Ensure path doesn't have trailing nulls or invalid characters
file_path = file_path.rstrip('\x00')
parameters["file"] = file_path
print(f"[DEBUG] Normalized file path for {command_to_run}: {repr(file_path)}")
data += len(parameters).to_bytes(4,"big")
for param in parameters:
param_value = parameters[param]
# Ensure parameter is a string and doesn't contain null bytes
if isinstance(param_value, str):
param_value = param_value.rstrip('\x00')
param_bytes = param_value.encode('utf-8')
# Remove any null bytes from the encoded string
param_bytes = param_bytes.replace(b'\x00', b'')
else:
param_bytes = str(param_value).encode('utf-8')
data += len(param_bytes).to_bytes(4, "big")
data += param_bytes
# Convert parameter value to string if it's not already
if not isinstance(param_value, str):
param_value = str(param_value)
data += len(param_value).to_bytes(4, "big")
data += param_value.encode()
else:
data += b"\x00\x00\x00\x00"
README.md
+212
View File
@@ -31,6 +31,7 @@
- [Keylogging Support](#%EF%B8%8F-keylogging-support)
- [Token Support](#-token-support)
- [Context Tracking](#-context-tracking)
- [OPSEC Checking](#️-opsec-checking)
- [Development](#development)
- [Troubleshooting](#troubleshooting)
- [Credits](#credits)
@@ -1703,6 +1704,217 @@ For more information, see the [Mythic Context Tracking documentation](https://do
---
## 🛡️ OPSEC Checking
Cazalla includes comprehensive **OPSEC Checking** functionality that provides operational security warnings and blocking for commands based on their detection risks. This feature helps operators make informed decisions and prevents accidental OPSEC violations.
### Overview
OPSEC Checking is implemented using two types of checks:
1. **`opsec_pre`**: Runs **before** task creation (`create_tasking`). Can block the task if it detects high-risk operations.
2. **`opsec_post`**: Runs **after** task creation but **before** the agent picks up the task. Provides final warnings about artifacts that will be created.
### How It Works
#### OPSEC Pre-Check (`opsec_pre`)
The `opsec_pre` function evaluates the command parameters before creating the task:
- **Blocking**: Can set `OpsecPreBlocked=True` to prevent task execution
- **Warning**: Can provide detailed warnings without blocking
- **Bypass Roles**: Defines who can approve blocked tasks:
- `operator`: Any operator can bypass
- `other_operator`: Requires approval from a different operator
- `lead`: Only operation lead can approve
**Example Flow:**
```
1. Operator creates task: shell whoami
2. opsec_pre detects "whoami" has safer alternative
3. Task is BLOCKED with message explaining why
4. Another operator must approve to proceed
```
#### OPSEC Post-Check (`opsec_post`)
The `opsec_post` function runs after task creation but before agent execution:
- **Artifact Review**: Warns about artifacts that will be created
- **Final Warning**: Last chance to cancel before agent picks up task
- **Context-Aware**: Can review artifacts generated by `create_tasking`
**Example Flow:**
```
1. opsec_pre passes (or is bypassed)
2. create_tasking executes
3. opsec_post reviews artifacts
4. Task is submitted to agent (if approved)
```
### Commands with OPSEC Checking
#### High-Risk Commands (Blocking)
| Command | Blocking Conditions | Bypass Role |
|---------|---------------------|-------------|
| `shell` | Commands with safer Cazalla alternatives (`whoami`, `tasklist`, `powershell`, etc.) | `other_operator` |
| `steal_token` | Critical system processes (PID 4) | `other_operator` |
| `kill` | Critical system processes (PID 0, 4, 8) | `other_operator` |
| `keylog_start` | Always blocked (extremely detectable) | `lead` |
| `rm` | Critical system files/directories (System32, SysWOW64, Windows, Boot) | `other_operator` |
#### Warning-Only Commands
| Command | Warning Type | Bypass Role |
|---------|-------------|-------------|
| `screenshot` | Screen capture detection | `operator` |
| `rpfwd` | Network activity monitoring | `operator` |
| `socks` | High-volume network traffic | `operator` |
| `download` | Sensitive file access | `operator` |
| `upload` | Executable/script uploads | `operator` |
| `make_token` | Authentication event logging | `operator` |
| `rev2self` | Token reversion | `operator` |
| `list_tokens` | Token enumeration | `operator` |
| `cat` | Sensitive file reading | `operator` |
| `rm` | File deletion (blocks for system files) | `other_operator` if blocked, `operator` if warning |
| `cp` | File copying (sensitive files) | `operator` |
### Example Messages
#### Blocking Message (shell whoami)
```
🚨 OPSEC BLOCKED - Command Has Safer Alternative
This command contains operations that have safer built-in alternatives in Cazalla:
🚨 whoami: Use Cazalla's 'whoami' command instead (no cmd.exe spawn)
⚠️ cmd.exe spawn: This command would spawn cmd.exe (highly detectable)
💡 SAFER ALTERNATIVES:
→ Use: 'whoami' (Cazalla built-in, no cmd.exe)
⚠️ TO PROCEED: You need approval from another operator.
This command is blocked to prevent accidental OPSEC violations.
```
#### Warning Message (download)
```
⚠️ OPSEC WARNING - File Download Detected
File downloads are monitored by:
• EDR/XDR solutions
• Data loss prevention (DLP) systems
• File access monitoring tools
• Network monitoring (large file transfers)
Target File: C:\Windows\System32\config\SAM
🚨 SENSITIVE FILE DETECTED:
🚨 sam: SAM database - Credential extraction heavily monitored
🔍 DETECTION RISKS:
• File access events (if auditing enabled)
• Large file transfers (network monitoring)
• Sensitive file patterns (DLP detection)
• Unusual file access patterns
✅ This is a WARNING only - command will proceed.
Ensure you understand the detection risks before continuing.
```
### OPSEC Checking Logic
#### Conditional Blocking
OPSEC checking is **intelligent and conditional**:
- **Blocking**: Only blocks when there's a clear safer alternative or critical risk
- **Warning**: Provides informative warnings for moderate risks
- **Context-Aware**: Considers command parameters (PID, file paths, etc.)
#### Example: `shell` Command
```python
# Blocks if command contains:
- "whoami" → Use Cazalla's 'whoami' instead
- "tasklist" → Use Cazalla's 'ps' instead
- "powershell" → Highly monitored, use alternatives
# Warns if command contains:
- "netstat" → Network enumeration, proceed with caution
- "ipconfig" → Network queries, proceed with caution
```
### Bypass Roles
| Role | Description | Use Case |
|------|-------------|----------|
| `operator` | Any operator can bypass | Low-risk warnings |
| `other_operator` | Requires different operator approval | High-risk operations |
| `lead` | Only operation lead can approve | Critical operations (keyloggers) |
### Benefits
1. **Prevents Accidents**: Blocks dangerous commands with safer alternatives
2. **Informs Operators**: Provides detailed risk information
3. **Improves OPSEC**: Reduces accidental detection
4. **Educational**: Teaches operators about detection risks
5. **Context-Aware**: Adapts to specific command parameters
### Technical Implementation
OPSEC checking is implemented in each command's Python definition:
```python
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
"""
# Analyze command parameters
command = taskData.args.get_arg("command") or ""
# Determine if blocking is needed
blocked = False
if "dangerous_pattern" in command:
blocked = True
# Build informative message
message = "⚠️ OPSEC WARNING - ...\n\n"
message += "Detailed explanation...\n"
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=blocked,
OpsecPreBypassRole="other_operator" if blocked else "operator",
OpsecPreMessage=message
)
async def opsec_post(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPostTaskMessageResponse:
"""
OPSEC check after creating the task.
"""
# Review artifacts that will be created
message = "⚠️ OPSEC POST-CHECK - Artifacts\n\n"
message += "Artifacts that will be created...\n"
return PTTTaskOPSECPostTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPostBlocked=False, # Usually don't block in post
OpsecPostBypassRole="operator",
OpsecPostMessage=message
)
```
For more information, see the [Mythic OPSEC Checking documentation](https://docs.mythic-c2.net/customizing/payload-type-development/opsec-checking).
---
## 🔧 Development
### Project Structure