Skip to content

Instantly share code, notes, and snippets.

@neoplacer
Created November 7, 2017 09:55
Show Gist options
  • Save neoplacer/6504e34a35a6df323b9c4a80da1a1180 to your computer and use it in GitHub Desktop.
Save neoplacer/6504e34a35a6df323b9c4a80da1a1180 to your computer and use it in GitHub Desktop.
Fake FTP Server
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Fake FTP Server
~~~~~~~~~~~~~~~
This is a simple fake FTP daemon. It stores the login data (username and
password) given to it and then terminates the connection.
It was made to easily recover forgotten passwords which are still stored (but
scrambled) by an FTP client without the need for using a sniffer tool.
Beware: Although the script reacts on ``<Ctrl-C>``, it won't exit until
another connect has happened because ``socket.accept()`` is blocking. Also,
due to the use of threads, it may take some time until the port will be
available again.
Some useful resources for implementing this were the IETF's `RFC 959`_ and the
`FTP reference`_ by D. J. Bernstein.
.. _RFC 959: http://tools.ietf.org/rfc/rfc959.txt
.. _FTP reference: http://cr.yp.to/ftp.html
:Copyright: 2007 Jochen Kupperschmidt
:Date: 13-Jul-2007
:License: MIT
"""
from datetime import datetime
from optparse import OptionParser
from SocketServer import BaseRequestHandler, ThreadingTCPServer
from sys import stdout
class FTPLoginHandler(BaseRequestHandler):
"""Handler for FTP authentication."""
def debug(self, message):
"""Show log message."""
if self.server.debug:
print '***', message
def respond(self, code, explanation):
"""Send a response to the client."""
self.request.send('%d %s\r\n' % (code, explanation))
def process_request(self):
"""Parse input into a command and an argument."""
data = self.request.recv(64)
parts = data.strip().split(' ')
return parts.pop(0), parts
def log_auth(self, user, password):
"""Write username and password to logfile."""
now = datetime.now().isoformat(' ')[:19]
client = '%s:%d' % self.client_address
line = ' '.join((now, client, user, password))
self.server.logfile.write(line + '\n')
self.server.logfile.flush()
def handle(self):
"""Handle incoming data."""
self.debug('Connection from %s:%d.' % self.client_address)
self.respond(220, self.server.banner)
user = None
while True:
cmd, args = self.process_request()
if cmd == 'USER':
if user is not None:
self.respond(503, 'Incorrect sequence of'
' commands: PASS required after USER.')
continue
user = (args and args[0] or '*missing*')
self.debug('User "%s" has identified.' % user)
self.respond(331, 'Please specify the password.')
continue
elif cmd == 'PASS':
if user is None:
self.respond(503, 'Incorrect sequence of'
' commands: USER required before PASS.')
continue
password = (args and args[0] or '*missing*')
self.debug('User "%s" supplied password "%s", storing.'
% (user, password))
self.log_auth(user, password)
self.respond(530, 'Login incorrect.')
break
else:
self.debug('Rejecting request "%s".' % ' '.join(args))
self.respond(530, 'Please login with USER and PASS.')
break
self.request.close()
self.debug('Connection with %s:%d closed.' % self.client_address)
class FTPLoginServer(ThreadingTCPServer):
def __init__(self, host='', port=21, banner='', debug=False, logfile=None,
append=False):
ThreadingTCPServer.__init__(self, (host, port), FTPLoginHandler)
self.banner = banner
self.debug = debug
mode = (append and 'a' or 'w')
self.logfile = (logfile and open(logfile, mode) or stdout)
def server_close(self):
ThreadingTCPServer.server_close(self)
self.logfile.close()
if __name__ == '__main__':
parser = OptionParser(usage='%prog [options] <port>')
parser.add_option('-a', '--append', dest='append', action='store_true',
help='append to LOGFILE ')
parser.add_option('-b', '--banner', dest='banner',
help='custom banner string')
parser.add_option('-d', '--debug', dest='debug', action='store_true',
help='show debugging messages')
parser.add_option('-l', '--logfile', dest='logfile',
help='write collected user/password data to LOGFILE')
opts, args = parser.parse_args()
# Parse arguments.
if len(args) != 1:
parser.print_help()
parser.exit()
try:
port = int(args[0])
except ValueError:
parser.print_help()
parser.exit()
# Serve.
server = FTPLoginServer(port=port, **opts.__dict__)
try:
server.serve_forever()
except KeyboardInterrupt:
print 'Ctrl-C pressed, exiting...'
server.server_close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment