Python 并行哈希暴力破解

Question

我已根据该站点上另一个用户编写的不同脚本修改了此脚本。我试图尽我所能理解它，但我遇到了困难。当我尝试使用 charset 作为可能仅小写字母来破解哈希时，它工作得很好。但是当我尝试破解包含数字和字母的哈希时，除非我将“spc”变量从 1000000 增加到 100000000，否则它将无法工作。

import itertools
import math
import string
import multiprocessing
import hashlib
import traceback
import sys

def hashstring(string, algorithm):
    return hashlib.new(algorithm, string).hexdigest()

def gen_product(prefix, charset, length):
    for string in itertools.product(charset, repeat=length):
        yield prefix + "".join(string)

def string_generator(prefix, hash, suffix_length, length, charset, hashalg):
    num_done = 0
    if length <= suffix_length:
        assert prefix == ""
        for possible in gen_product("", charset, length):
            if hashstring(possible, hashalg) == hash:
                return possible
    else:
        assert len(prefix) + suffix_length == length
        for possible in gen_product(prefix, charset, suffix_length):
            if hashstring(possible, hashalg) == hash:
                return possible

    return None

def run_string_generator(*args):
    try:
        return string_generator(*args)
    except:
        raise Exception("".join(traceback.format_exception(*sys.exc_info())))

def do_work(pool, hash, charset, length, hashalg, spc=100000000):
    n = len(charset)
    suffix_len = int(math.ceil(math.log(spc) / math.log(n)) - 1)

    max_short_len = min(suffix_len, length)
    for length in range(1, max_short_len + 1):
        result = pool.apply_async(run_string_generator, args = ("", hash, suffix_len, length, charset, hashalg))
        if result.get() != None:
            return result.get()
    for length in range(max_short_len + 1, length + 1):
        for prefix in gen_product("", charset, length - suffix_len):
            result = pool.apply_async(run_string_generator, args = (prefix, hash, suffix_len, length, charset, hashalg))    
            if result.get() != None:
                return result.get()

    return None


def parallel_bruteforce(hash, charset, length, hashalg="md5", spc=1000000, cores=None):
    pool = multiprocessing.Pool(cores)
    result = do_work(pool, hash, charset, length, hashalg, spc)

    pool.close()
    pool.join()

    return result

if __name__ == "__main__":
    print "Starting..."
    #The hash is an md5 encryption of "test1"
    print parallel_bruteforce("5a105e8b9d40e1329780d62ea2265d8a", string.ascii_lowercase +  string.digits, 5, spc=100000000)

编辑： 原始代码的另一篇文章的链接是https://stackoverflow.com/a/20135250/1769995

Answer 1

抱歉，我现在没有时间解释这一点。这是对我早期答案的编辑，它保留了并行性并在哈希被破解时“提前”停止所有工作人员。通常，您不希望传递在调用中从不改变的参数，因此我在模块级别做的比您做的要多。附件代码显示：

workers will cycle through the last 3 chars
. . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
[etc]
. . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
the plaintext is 'test1'

来了。请注意，它通过实现PAIRS（字母表中所有 2 字符字符串的列表）来发挥另一个低级效率技巧。从长远来看，这再次节省了数十亿的冗余连接。

import string
import hashlib
from itertools import product

CHARSET = string.ascii_lowercase +  string.digits
MAX_LENGTH = 5
NUM_PROCESSES = None # defaults to all available cores

HASHALG = "md5"
HASH = "5a105e8b9d40e1329780d62ea2265d8a"

PAIRS = ["".join(t) for t in product(CHARSET, repeat=2)]

def make_bases(count):
    bases = [PAIRS] * (count // 2)
    if count & 1:
        bases.insert(0, CHARSET)
    return bases

# string_gen is what the workers run.  Everything else
# runs in the main program.
def string_gen(prefix, suffix_len, length):
    # Generate all strings of length `length` starting with `prefix`.
    # If length > suffix_len, only the last suffix_len characters
    # need to be generated.
    if length <= suffix_len:
        assert prefix == ""
        bases = make_bases(length)
    else:
        assert len(prefix) + suffix_len == length
        bases = make_bases(suffix_len)
    for t in product(*bases):
        result = prefix + "".join(t)
        # do something with result
        if hashlib.new(HASHALG, result).hexdigest() == HASH:
            return result

def record_done(result):
    global all_done, the_secret
    print ".",
    if result is not None:
        print
        the_secret = result
        all_done = True
        pool.close()
        pool.terminate() # stop all workers! we're done

def do_work(pool, strings_per_chunk=1000000):
    global all_done, the_secret
    all_done = False
    the_secret = None
    # What's the most chars we can cycle through without
    # exceeding strings_per_chunk?
    N = len(CHARSET)
    suffix_len = 1
    while N**suffix_len <= strings_per_chunk:
        suffix_len += 1
    suffix_len -= 1
    print "workers will cycle through the last", suffix_len, "chars"

    # There's no point to splitting up very short strings.
    max_short_len = min(suffix_len, MAX_LENGTH)
    for length in range(1, max_short_len + 1):
        pool.apply_async(string_gen, args=("", suffix_len, length),
                         callback=record_done)
        if all_done:
            return
    # And now the longer strings.
    for length in range(max_short_len + 1, MAX_LENGTH + 1):
        for t in product(*make_bases(length - suffix_len)):
            prefix = "".join(t)
            pool.apply_async(string_gen, args=(prefix, suffix_len, length),
                             callback=record_done)
            if all_done:
                return

if __name__ == "__main__":
    import multiprocessing
    pool = multiprocessing.Pool(NUM_PROCESSES)
    do_work(pool)
    pool.close()
    pool.join()
    if the_secret is None:
        print "didn't crack it!"
    else:
        print "the plaintext is", repr(the_secret)

注意： 正如所写，对于较大的问题规模和/或非常小的问题strings_per_chunk，该代码“过于并行”。主程序触发apply_async() 调用的速度远远快于工作进程处理它们的速度，因此multiprocessing 机器最终可能会尝试将数十亿个工作项排队。然后，您可能会耗尽 RAM 或其他系统资源。当然这也是可以修复的；-)

解决这个问题

multiprocessing 没有公开任何限制其内部队列的方法，因此“自然”的解决方案是使用我们自己的队列添加一个层。这会在multiprocessing 的内部任务队列中为每个处理器保留多达 3 个待处理任务，但只要它自己的队列变得更长，就会阻止主程序生成更多前缀。还修改了在哈希被破解时引发EarlyExit 异常的逻辑；这比使用全局标志更容易和更清洁。以下内容旨在替换上面的所有内容，从record_done()向下：

class EarlyExit(Exception):
    def __init__(self, result):
        Exception.__init__(self)
        self.result = result

class CustomDispatcher:
    def __init__(self, pool):
        from collections import deque
        self.pool = pool
        self.q = deque()

    def queue_work(self, *args):
        while len(self.q) > NUM_PROCESSES * 3:
            # provided the workers have significant work to do,
            # it will "take a long time" to finish the work
            # already queued.  Rather than swamp the mp machinery
            # with even more pending tasks, wait for some to
            # finish first.
            self.unqueue()
        self.q.append(self.pool.apply_async(string_gen, args))

    def unqueue(self):
        if self.q:
            # note:  the main program spends most of its time
            # blocked on the .get(); that's because it can
            # generate prefixes far faster than workers can
            # process them
            result = self.q.popleft().get()
            print ".",
            if result is not None:
                print
                raise EarlyExit(result)

    def drain(self):
        while self.q:
            self.unqueue()

def do_work(dispatch, strings_per_chunk=10000000):
    # What's the most chars we can cycle through without
    # exceeding strings_per_chunk?
    N = len(CHARSET)
    suffix_len = 1
    while N**suffix_len <= strings_per_chunk:
        suffix_len += 1
    suffix_len -= 1
    print "workers will cycle through the last", suffix_len, "chars"
    print "so each dot represents", \
          format(len(CHARSET)**suffix_len, ","), "strings"

    # There's no point to splitting up very short strings.
    max_short_len = min(suffix_len, MAX_LENGTH)
    for length in range(1, max_short_len + 1):
        dispatch.queue_work("", suffix_len, length)
    # And now the longer strings.
    for length in range(max_short_len + 1, MAX_LENGTH + 1):
        for t in product(*make_bases(length - suffix_len)):
            dispatch.queue_work("".join(t), suffix_len, length)
    dispatch.drain()  # check remaining tasks for a winner

if __name__ == "__main__":
    import multiprocessing
    pool = multiprocessing.Pool(NUM_PROCESSES)
    dispatch = CustomDispatcher(pool)
    try:
        do_work(dispatch)
    except EarlyExit as e:
        print "the plaintext is", repr(e.result)
    else:
        print "didn't crack it!"
    pool.close()
    pool.terminate() # stop all workers! we're done
    pool.join()

随着字母表的大小和/或生成的最长字符串的大小增加，可能性的组合爆炸可能意味着您将永远等待结果，但至少通过此更改您不会耗尽 RAM -并且您将以接近 100% 的容量利用所有内核。

Answer 2

注意：

    result = pool.apply_async(run_string_generator, args = (prefix, hash, suffix_len, length, charset, hashalg))    
    if result.get() != None:
        return result.get()

破坏了所有的并行性。 result.get() 阻塞，直到工作进程完成其任务。那时，一次只能有一名工作人员处于活动状态。你没有注意到只有一个核心处于活动状态吗？

Answer 3

我首先要指出，spc 的名字充其量是不好的 - 如果它被描述，我们就不需要对其目的进行逆向工程。它从 main 通过 parallel_bruteforce 原封不动地传递到 do_work，它只在一行中使用：

suffix_len = int(math.ceil(math.log(spc) / math.log(n)) - 1)

这条线表明它与长度有关，但以相当复杂的数学方式。 n 是字符集大小。对数的使用适用于计算字符串的长度，该字符串需要一组特定的数字来表示多个不同的值。例如，log(50000)/log(16) 大约是 3.9 - 告诉我们需要 4 个十六进制数字来计算 50000 个值，这与 0xffff==65535 非常匹配。像这样划分对数会在对数底之间进行转换；通常只实现自然对数​​和以 10 为底的对数（python 中的 log 和 log10），但 logx(n)=log(n)/log(x)、regardless of the base of log。

这样，我们可以看到spc 很可能与搜索空间有关——尝试的尝试次数。它在这里使用的方式必须意味着该算法进行指定次数的尝试，而不是特定宽度所需的尽可能多的尝试。如果必须提高宽度，则需要提高 spc 以匹配：spc=len(charset)**width。这自然揭示了具有更广泛字符集的暴力破解的指数性质；它会很慢。