Skip to content

Instantly share code, notes, and snippets.

@socketz
Forked from neoplacer/fake_ftp.py
Last active May 18, 2019 14:59
Show Gist options
  • Save socketz/f8157db85c2255bca01aadd42876e851 to your computer and use it in GitHub Desktop.
Save socketz/f8157db85c2255bca01aadd42876e851 to your computer and use it in GitHub Desktop.
Fake FTP Server - Python 3.x
#!/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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment