Fake FTP Server - Python 3.x

fake_ftp.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: @neoplacer
# Contributor: @socketz

"""
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."""
        response = '{} {}\r\n'.format(code, explanation)
        self.request.send(response.encode())

    def process_request(self):
        """Parse input into a command and an argument."""
        data = self.request.recv(64)
        parts = data.strip().decode().split(' ')
        return parts.pop(0), parts

    def log_auth(self, user, password):
        """Write username and password to logfile."""
        now = datetime.now().isoformat(' ')[:19]
        client = '{}:{}'.format(self.client_address[0], self.client_address[1])
        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 {0}:{1}.'.format(
            self.client_address[0], self.client_address[1]))
        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 "{}" has identified.'.format(user))
                self.respond(331, 'Password required for {}'.format(user))
                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 "{}" supplied password "{}", storing.'.format(user, password))
                self.log_auth(user, password)
                self.respond(430, 'Login incorrect.')
                break
            else:
                self.debug('Rejecting request "{}".'.format(' '.join(args)))
                self.respond(530, 'Please login with USER and PASS.')
                break
        self.request.close()
        self.debug('Connection with {}:{} closed.'.format(
            self.client_address[0], self.client_address[1]))


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()