Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
trustmemore
#!/usr/bin/env python
import os
import sys
import pwd
import random
import struct
import paramiko
import tempfile
import SocketServer
GATEKEEPER_PW = "trust is 74u57 -- but l33t doesn't have a u"
CHALLENGE_ANSWER = "OnlyDDTEKWouldDoSomethingThisEvil"
class TrustMeRequestHandler(SocketServer.BaseRequestHandler):
def readstr(self, x='\n'):
rv = ""
while True:
ch = self.request.recv(1)
if ch == x:
return rv
rv += ch
return rv
def handle(self):
(seed, ) = struct.unpack("I", open("/dev/urandom", "rb").read(4))
random.seed(seed)
self.request.send("Password: ")
pw = self.readstr()
if pw != GATEKEEPER_PW:
self.request.send("Incorrect.\n")
self.request.close()
return
self.request.send("Accepted. Now send me some bytes.\n")
private_key = self.request.recv(2048)
if len(private_key) < 1024:
self.request.send("Don't be shy. I like bytes.\n")
self.request.close()
return
if "BEGIN RSA PRIVATE KEY" not in private_key:
# We'll give them a ~1 in 25 chance of getting a blatant hint
r = random.choice(xrange(25))
if r == 0:
resp = "Sorry, I was expecting something with 'BEGIN RSA PRIVATE KEY' in it.\n"
else:
r = r % 5
if r == 0:
resp = "I was hoping for something a little more intimate.\n"
elif r == 1:
resp = "I'm wanting a private experience.\n"
elif r == 2:
resp = "I want to tell you a secret, but I can't unless you trust me.\n"
elif r == 3:
resp = "I just want to read your logs...to know if you're still faithful.\n"
else:
resp = "The sooner you trust me with your keys, the sooner we can get down to the action.\n"
self.request.send(resp)
self.request.close()
return
pkey_fp = tempfile.TemporaryFile()
pkey_fp.write(private_key)
pkey_fp.flush()
pkey_fp.seek(0)
(client_ip, src_port) = self.request.getpeername()
print "[+] Trying to connect out to %s" % (client_ip, )
try:
private_key_obj = paramiko.RSAKey.from_private_key(pkey_fp)
except:
self.request.send("I don't understand why you're trying to hide from me.\n")
self.request.close()
return
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(client_ip, port=22, username="root", pkey=private_key_obj, allow_agent=False, look_for_keys=False)
except:
self.request.send("Your key doesn't work!\n")
self.request.close()
return
print "[+] Connected!"
# Unset all history variables
ssh.exec_command("unset HISTFILE")
# Make sure we're on linux
version = ssh.exec_command("lsb_release -i")[1].read()
if "Distributor" not in version:
self.request.send("Stop lying to me! I know you only use linux!\n")
self.request.close()
return
# Check to see if we're root (uid 0)
uid = ssh.exec_command("id -u")[1].read()
if uid.strip() != "0":
self.request.send("I'm going to need all the access!\n")
self.request.close()
return
# Check to see if we're in a chroot
init = ssh.exec_command("stat -c %d:%i /proc/1/root/.")[1].read()
root = ssh.exec_command("stat -c %d:%i /")[1].read()
if init != root:
self.request.send("Don't keep me in the dark. I know you're hiding things from me.\n")
self.request.close()
return
# Check to see if we're virtualized (vmware)
vmware = ssh.exec_command("vmware-toolbox --version")[1].read()
if "VMware" in vmware:
self.request.send("I know what VMWare is!\n")
self.request.close()
return
# Check to see if we're virtualized (qemu)
qemu = ssh.exec_command("dmesg | grep -i virtual")[1].read()
if "QEMU" in qemu:
self.request.send("QEMU? Wouldn't you pick something faster to lie to me with?\n")
self.request.close()
return
# Check to see if they're uids >=1000
etc_passwd = ssh.exec_command("cat /etc/passwd")[1].read()
etc_passwd = etc_passwd.split("\n")
etc_passwd = filter(None, etc_passwd)
entries = map(lambda x: x.split(":"), etc_passwd)
entries = map(lambda x: (int(x[2]), x[5]), entries)
entries = filter(lambda x: x[0] >= 1000 and x[0] < 65000, entries)
if len(entries) == 0:
self.request.send("Don't tell me you don't use this computer!\n")
self.request.close()
return
# See if there are bash_history entries
has_history = False
for (uid, homedir) in entries:
history = ssh.exec_command("cat %s/.bash_history" % (homedir, ))[1].read()
if len(history) > 5:
has_history = True
if not has_history:
self.request.send("I want to see proof you use this computer!\n")
self.request.close()
return
# See if there is browser history (firefox)
has_firefox = False
has_chromium = False
has_chrome = False
for (uid, homedir) in entries:
firefox = ssh.exec_command("ls -l %s/.mozilla" % (homedir, ))[1].read()
if len(firefox) > 5:
has_firefox = True
# See if there is browser history (chromium)
for (uid, homedir) in entries:
chromium = ssh.exec_command("ls -l %s/.cache/chromium" % (homedir, ))[1].read()
if len(chromium) > 5:
has_chromium = True
for (uid, homedir) in entries:
chrome = ssh.exec_command("ls -l %s/.cache/google-chrome" % (homedir, ))[1].read()
if len(chrome) > 5:
has_chrome = True
if not any((has_firefox, has_chromium, has_chrome)):
self.request.send("What the hell do you use this computer for?!\n")
self.request.close()
return
# Write key file to disk
ssh.exec_command("echo %s > /root/gits" % (CHALLENGE_ANSWER, ))
self.request.send("Okay. I trust you.\n")
self.request.close()
return
class ForkingTCPServer(SocketServer.ForkingMixIn, SocketServer.TCPServer):
pass
if __name__ == '__main__':
port = 7457
server = ForkingTCPServer(('', port), TrustMeRequestHandler)
print "Server running on 0.0.0.0:%d" % (port)
server.serve_forever()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment