在Python 3中进行多重处理

Question

我正在搞一个zip文件破解程序，并决定使用多处理模块来加速进程。这是一个完整的痛苦，因为这是我第一次使用该模块，我甚至还没有完全理解它。但是，我得到它的工作。

问题是它没有完成单词列表;它只是在单词列表中随机停止放置，如果找到密码，它将继续通过单词列表，而不仅仅是停止该过程。

有谁知道它为什么会出现这种行为？

源代码的ZipFile饼干

#!/usr/bin/env python3 

import multiprocessing as mp 
import zipfile # Handeling the zipfile 
import sys # Command line arguments, and quiting application 
import time # To calculate runtime 

def usage(program_name): 
    print("Usage: {0} <path to zipfile> <dictionary>".format(program_name)) 
    sys.exit(1) 

def cracker(password): 
    try: 
     zFile.extractall(pwd=password) 
     print("[+] Password Found! : {0}".format(password.decode('utf-8'))) 
     pool.close() 
    except: 
     pass 

def main(): 
    global zFile 
    global pool 

    if len(sys.argv) < 3: 
     usage(sys.argv[0]) 

    zFile = zipfile.ZipFile(sys.argv[1]) 

    print("[*] Started Cracking") 

    startime = time.time() 
    pool = mp.Pool() 

    for i in open(sys.argv[2], 'r', errors='ignore'): 
     pswd = bytes(i.strip('\n'), 'utf-8') 
     pool.apply_async(cracker, (pswd,)) 

    print (pswd) 
    runtime = round(time.time() - startime, 5) 
    print ("[*] Runtime:", runtime, 'seconds') 
    sys.exit(0) 

if __name__ == "__main__": 
    main() 

Answer 1

下面是@phihag's意见的实施和@Equality 7-2521's answers：

#!/usr/bin/env python3 
"""Brute force zip password. 

Usage: brute-force-zip-password <zip archive> <passwords> 
""" 
import sys 
from multiprocessing import Pool 
from time import monotonic as timer 
from zipfile import ZipFile 

def init(archive): # run at the start of a worker process 
    global zfile 
    zfile = ZipFile(open(archive, 'rb')) # open file in each process once 

def check(password): 
    assert password 
    try: 
     with zfile.open(zfile.infolist()[0], pwd=password): 
      return password # assume success 
    except Exception as e: 
     if e.args[0] != 'Bad password for file': 
      # assume all other errors happen after the password was accepted 
      raise RuntimeError(password) from e 

def main(): 
    if len(sys.argv) != 3: 
     sys.exit(__doc__) # print usage 

    start = timer() 
    # decode passwords using the preferred locale encoding 
    with open(sys.argv[2], errors='ignore') as file, \ 
     Pool(initializer=init, initargs=[sys.argv[1]]) as pool: # use all CPUs 
     # check passwords encoded using utf-8 
     passwords = (line.rstrip('\n').encode('utf-8') for line in file) 
     passwords = filter(None, passwords) # filter empty passwords 
     for password in pool.imap_unordered(check, passwords, chunksize=100): 
      if password is not None: # found 
       print("Password: '{}'".format(password.decode('utf-8'))) 
       break 
     else: 
      sys.exit('Unable to find password') 
    print('Runtime: %.5f seconds' % (timer() - start,)) 

if __name__=="__main__": 
    main() 

注：

• 每个工作进程都有自己的ZipFile对象和压缩文件在每个进程中打开一次：它应该使它更加便携（Windows支持）并提高时间性能
• 内容未被提取：check(password)试图在成功时打开并立即关闭归档成员：它更安全并且应该提高时间性能（不需要创建目录等）
• 解密归档文件时除了'Bad password for file'之外的所有错误成员被假定为发生后密码被接受：理性是避免沉默意外的错误 - 每个异常应单独考虑
• check(password)预计非空密码
• chunksize参数可以大幅提高性能
• 难得for/else语法时，举报箱时，没有发现密码
• 的语句来为你

Answer 2

你太早期终止程序。要测试此方法，请在cracker方法中添加一个无害的time.sleep(10)，并观察您的程序仍在一秒钟内终止。

呼叫join等待池完成：

pool = mp.Pool() 
for i in open(sys.argv[2], 'r', errors='ignore'): 
    pswd = bytes(i.strip('\n'), 'utf-8') 
    pool.apply_async(cracker, (pswd,)) 

pool.close() # Indicate that no more data is coming 
pool.join() # Wait for pool to finish processing 

runtime = round(time.time() - startime, 5) 
print ("[*] Runtime:", runtime, 'seconds') 
sys.exit(0) 

此外，一旦你找到了正确的密码，称close只是表明没有更多的未来的任务来了 - 已经提交了所有的任务仍然会做。相反，请致电terminate来终止池而不处理任何更多任务。

此外，根据multiprocessing.Pool的实现细节，全局变量pool在您需要时可能无法使用（并且其值无法反序列化）。为了解决这个问题，你可以在

def cracker(password): 
    try: 
     zFile.extractall(pwd=password) 
    except RuntimeError: 
     return 
    return password 

def callback(found): 
    if found: 
     pool.terminate() 
... 
pool.apply_async(cracker, (pswd,), callback=cb) 

当然可以用一个回调，因为，因为你现在看结果的时候，apply是不正确的方式去。相反，你可以使用imap_unordered编写代码：

with open(sys.argv[2], 'r', errors='ignore') as passf, \ 
     multiprocessing.Pool() as pool: 
    passwords = (line.strip('\n').encode('utf-8') for line in passf) 
    for found in pool.imap_unordered(cracker, passwords): 
     if found: 
      break 

而不是使用全局变量，你可能还需要打开压缩文件（并创建一个ZipFile对象）中的每一个过程，通过使用initializer为池。更好（更快），放弃所有的I/O，只读一次你需要的字节，然后传给孩子们。

Answer 3

phihag的答案是正确的解决方案。

我只是想提供有关在找到正确密码时拨打terminate()的更多详细信息。我运行代码时未定义pool变量cracker()。所以，试图从那里调用它只是抛出一个异常：

NameError: name 'pool' is not defined 

（我fork()经验较弱，所以我不完全理解为什么全球zFile被复制到子进程成功而pool甚至不如果它被复制，它在父进程中不会是相同的pool，对吗？因此，调用它的任何方法都不会影响父进程中的真实池。无论如何，我更喜欢multiprocessing模块的编程指南中列出this建议部分：明确地传递资源，子进程）

我的建议是让cracker()回报，如果它是正确的密码，否则返回None。然后将回拨传递给apply_async()，记录正确的密码以及终止池。这是我采取在修改代码来做到这一点：

#!/usr/bin/env python3 

import multiprocessing as mp 
import zipfile # Handeling the zipfile 
import sys # Command line arguments, and quiting application 
import time # To calculate runtime 
import os 

def usage(program_name): 
    print("Usage: {0} <path to zipfile> <dictionary>".format(program_name)) 
    sys.exit(1) 

def cracker(zip_file_path, password): 
    print('[*] Starting new cracker (pid={0}, password="{1}")'.format(os.getpid(), password)) 

    try: 
     time.sleep(1) # XXX: to simulate the task taking a bit of time 
     with zipfile.ZipFile(zip_file_path) as zFile: 
      zFile.extractall(pwd=bytes(password, 'utf-8')) 
     return password 
    except: 
     return None 

def main(): 
    if len(sys.argv) < 3: 
     usage(sys.argv[0]) 

    print('[*] Starting main (pid={0})'.format(os.getpid())) 

    zip_file_path = sys.argv[1] 
    password_file_path = sys.argv[2] 
    startime = time.time() 
    actual_password = None 

    with mp.Pool() as pool: 
     def set_actual_password(password): 
      nonlocal actual_password 
      if password: 
       print('[*] Found password; stopping future tasks') 
       pool.terminate() 
       actual_password = password 

     with open(password_file_path, 'r', errors='ignore') as password_file: 
      for pswd in password_file: 
       pswd = pswd.strip('\n') 
       pool.apply_async(cracker, (zip_file_path, pswd,), callback=set_actual_password) 

     pool.close() 
     pool.join() 

    if actual_password: 
     print('[*] Cracked password: "{0}"'.format(actual_password)) 
    else: 
     print('[*] Unable to crack password') 
    runtime = round(time.time() - startime, 5) 
    print("[*] Runtime:", runtime, 'seconds') 
    sys.exit(0) 

if __name__ == "__main__": 
    main()