Upload app.py

app.py

@@ -3,51 +3,88 @@ from pywallet import wallet
 import time
 import itertools
 import string
+import threading
 
 # --- 配置 ---
-# 密码字符集（仅用于基本暴力破解，非常不推荐用于实际钱包）
-DEFAULT_CHARSET = string.ascii_lowercase + string.digits # a-z, 0-9
-MAX_BRUTEFORCE_LEN = 4 # 限制暴力破解长度，否则会永远运行
+DEFAULT_CHARSET = string.ascii_lowercase + string.digits
+MAX_BRUTEFORCE_LEN = 4
 
-# --- 辅助函数 ---
+# --- 辅助函数 (保持不变) ---
 def try_decrypt_wallet(wallet_data_bytes, password_attempt):
-    """
-    尝试用给定的密码解密钱包。
-    返回 True 如果成功, False 如果失败。
-    """
     try:
-        # pywallet 需要一个文件名或类文件对象，但我们可以从字节加载
-        # 为了安全和简单，我们避免将钱包数据写入临时文件
-        # 但 pywallet.Wallet.from_binary() 似乎是存在的
         w = wallet.Wallet.from_binary(wallet_data_bytes)
-
         if w.is_encrypted():
-            if w.decrypt(password_attempt):
-                # 解密成功！
-                return True
-            else:
-                # 密码错误
-                return False
+            return w.decrypt(password_attempt)
         else:
-            # 钱包未加密
-            st.info("钱包文件未加密。")
-            return True # 可以认为是一种“成功”
-    except Exception as e:
-        # st.error(f"解密时发生错误: {e}") # 调试时可以取消注释
-        # 常见的错误可能是密码错误导致解密失败，或者钱包文件损坏
+            st.session_state.info_message = "钱包文件未加密。"
+            return True
+    except Exception:
         return False
 
 def generate_passwords_bruteforce(charset, max_len):
-    """
-    (非常不推荐) 基本的暴力破解密码生成器
-    """
     for length in range(1, max_len + 1):
         for item in itertools.product(charset, repeat=length):
             yield "".join(item)
 
+# --- 后台工作函数 ---
+def background_cracker_task(wallet_bytes_data, passwords_iterable, task_id, method_name):
+    """
+    在后台线程中运行破解过程。
+    更新 st.session_state 来反映进度和结果。
+    """
+    st.session_state[f'task_{task_id}_status'] = 'running'
+    st.session_state[f'task_{task_id}_found_password'] = None
+    st.session_state[f'task_{task_id}_error_message'] = None
+    st.session_state[f'task_{task_id}_attempts'] = 0
+    st.session_state[f'task_{task_id}_time_taken'] = 0
+
+    start_time = time.time()
+    found = False
+    processed_passwords = 0
+
+    # 特殊处理暴力破解，因为它是一个生成器，无法预知总数
+    is_bruteforce = (method_name == "bruteforce")
+    total_passwords_for_progress = 0
+    if not is_bruteforce and isinstance(passwords_iterable, list):
+        total_passwords_for_progress = len(passwords_iterable)
+
+    for pwd in passwords_iterable:
+        # 检查是否应该停止 (如果用户启动了新任务)
+        if st.session_state.get('current_task_id') != task_id:
+            st.session_state[f'task_{task_id}_error_message'] = "任务被另一个操作中止。"
+            st.session_state[f'task_{task_id}_status'] = 'failed'
+            return
+
+        processed_passwords += 1
+        st.session_state[f'task_{task_id}_attempts'] = processed_passwords
+
+        # 为了避免过于频繁地更新session_state (可能导致性能问题)
+        # 可以在这里添加一些逻辑，例如每隔N次尝试或每隔X秒更新一次用于显示的尝试计数
+        # 不过对于最终结果的设置，这个不是大问题
+
+        if try_decrypt_wallet(wallet_bytes_data, pwd):
+            st.session_state[f'task_{task_id}_found_password'] = pwd
+            st.session_state[f'task_{task_id}_status'] = 'success'
+            found = True
+            break
+        # time.sleep(0.0001) # 极小的延时，以允许GIL切换，但对CPU密集型任务效果不大
+
+    end_time = time.time()
+    st.session_state[f'task_{task_id}_time_taken'] = end_time - start_time
+
+    if not found and st.session_state[f'task_{task_id}_status'] == 'running': # 确保不是因为中止而退出
+        st.session_state[f'task_{task_id}_error_message'] = "未找到密码。"
+        st.session_state[f'task_{task_id}_status'] = 'failed'
+
+    # 主动触发一次 Streamlit 刷新，让UI能立刻感知到状态变化
+    # 这个在线程中调用需要小心，但对于更新session_state然后让Streamlit自己刷新是安全的
+    # st.experimental_rerun() # 旧版，如果需要可以尝试
+    # Streamlit 通常会自动检测 session_state 的变化并重新运行脚本
+
+
 # --- Streamlit UI ---
-st.set_page_config(page_title="Wallet.dat Cracker", layout="wide")
-st.title("Bitcoin Wallet.dat 密码破解尝试工具 (教育目的)")
+st.set_page_config(page_title="Wallet.dat Cracker (Background)", layout="wide")
+st.title("Bitcoin Wallet.dat 密码破解尝试 (后台运行)")
 
 st.warning(
     """
@@ -59,6 +96,15 @@ st.warning(
     """
 )
 
+# 初始化 session_state 变量
+if 'current_task_id' not in st.session_state:
+    st.session_state.current_task_id = None
+if 'task_counter' not in st.session_state: # 用于生成唯一的 task_id
+    st.session_state.task_counter = 0
+if 'info_message' not in st.session_state:
+    st.session_state.info_message = None
+
+
 # --- 文件上传 ---
 uploaded_wallet_file = st.file_uploader("1. 上传你的 wallet.dat 文件", type=["dat"])
 
@@ -69,108 +115,131 @@ if uploaded_wallet_file:
     # --- 破解方法选择 ---
     crack_method = st.radio(
         "2. 选择破解方法:",
-        ('使用密码列表', '尝试单个密码', '基本暴力破解 (非常不推荐，极慢)')
+        ('使用密码列表', '尝试单个密码', '基本暴力破解 (极慢，后台运行)')
     )
 
-    # --- 密码列表方法 ---
-    if crack_method == '使用密码列表':
-        uploaded_password_list = st.file_uploader(
-            "上传密码列表文件 ( .txt, 每行一个密码)", type=["txt"]
-        )
-        if uploaded_password_list:
-            passwords_to_try = [
-                line.strip() for line in uploaded_password_list.read().decode().splitlines() if line.strip()
-            ]
-            st.info(f"密码列表加载了 {len(passwords_to_try)} 个密码。")
-
-            if st.button("开始使用列表破解", key="list_crack"):
-                if not passwords_to_try:
-                    st.error("密码列表为空。")
+    current_task_id = st.session_state.get('current_task_id')
+    task_status = None
+    if current_task_id:
+        task_status = st.session_state.get(f'task_{current_task_id}_status')
+
+    # --- 显示任务状态和结果 ---
+    if task_status == 'running':
+        attempts = st.session_state.get(f'task_{current_task_id}_attempts', 0)
+        with st.spinner(f"正在后台破解... 已尝试 {attempts} 个密码。请勿关闭此页面。"):
+            # 由于 Streamlit 的特性，spinner 会在脚本重新运行时自动更新。
+            # 我们需要一种方式让脚本在线程运行时定期重新运行以更新spinner的文本。
+            # 一个简单的方法是使用 st.empty() 和 time.sleep() 在主线程中轮询，
+            # 但这又会阻塞主线程。
+            # 更好的方式是依赖 Streamlit 对 session_state 变化的自动刷新。
+            # 为了让 spinner 中的数字更新，后台线程可能需要更频繁地更新一个 session_state 变量
+            # 但为了简单，我们这里只显示一个通用的 "正在后台破解..."
+            # 并且依赖于任务完成后 session_state 的变化来刷新整个页面。
+            # 或者，我们可以在这里放一个占位符，后台线程完成后，其设置的session_state会导致页面刷新
+            st.empty() # 只是为了保持 spinner 可见
+            time.sleep(1) # 让streamlit有机会重跑脚本并显示spinner
+            st.rerun() # 强制重跑以更新spinner中的计数（如果做了计数更新）
+
+
+    elif task_status == 'success':
+        found_pwd = st.session_state.get(f'task_{current_task_id}_found_password')
+        time_taken = st.session_state.get(f'task_{current_task_id}_time_taken', 0)
+        attempts = st.session_state.get(f'task_{current_task_id}_attempts', 0)
+        st.success(f"🎉 密码找到: **{found_pwd}** 🎉")
+        st.balloons()
+        st.info(f"总耗时: {time_taken:.2f} 秒。尝试密码数: {attempts}。")
+        if st.button("清除结果并重试", key="clear_success"):
+            st.session_state.current_task_id = None # 清除任务标记
+            st.rerun()
+
+
+    elif task_status == 'failed':
+        error_msg = st.session_state.get(f'task_{current_task_id}_error_message')
+        time_taken = st.session_state.get(f'task_{current_task_id}_time_taken', 0)
+        attempts = st.session_state.get(f'task_{current_task_id}_attempts', 0)
+        st.error(f"破解失败: {error_msg}")
+        st.info(f"总耗时: {time_taken:.2f} 秒。尝试密码数: {attempts}。")
+        if st.button("清除结果并重试", key="clear_failed"):
+            st.session_state.current_task_id = None # 清除任务标记
+            st.rerun()
+
+    if st.session_state.get('info_message'):
+        st.info(st.session_state.info_message)
+        st.session_state.info_message = None # 显示后清除
+
+    # --- 启动按钮 (仅当没有任务运行时显示) ---
+    if task_status not in ['running', 'success', 'failed'] or not task_status :
+        passwords_to_try_list = []
+        brute_force_params = {}
+        can_start = False
+        start_button_label = "开始破解"
+        method_name_for_task = ""
+
+        if crack_method == '使用密码列表':
+            uploaded_password_list = st.file_uploader(
+                "上传密码列表文件 ( .txt, 每行一个密码)", type=["txt"], key="pwdlist_uploader"
+            )
+            if uploaded_password_list:
+                passwords_to_try_list = [
+                    line.strip() for line in uploaded_password_list.read().decode().splitlines() if line.strip()
+                ]
+                if passwords_to_try_list:
+                    st.info(f"密码列表加载了 {len(passwords_to_try_list)} 个密码。")
+                    can_start = True
+                    start_button_label = "开始使用列表破解"
+                    method_name_for_task = "list"
                 else:
-                    st.info("开始破解尝试...")
-                    progress_bar = st.progress(0)
-                    status_text = st.empty()
-                    found_password = None
-                    start_time = time.time()
-
-                    for i, pwd in enumerate(passwords_to_try):
-                        status_text.text(f"尝试密码 ({i+1}/{len(passwords_to_try)}): {pwd}")
-                        if try_decrypt_wallet(wallet_bytes, pwd):
-                            found_password = pwd
-                            break
-                        progress_bar.progress((i + 1) / len(passwords_to_try))
-                        # 在 Streamlit 中添加少量延迟以允许 UI 更新，尤其是在快速循环中
-                        # time.sleep(0.001)
-
-
-                    end_time = time.time()
-                    status_text.text("破解尝试完成。")
-                    elapsed_time = end_time - start_time
-
-                    if found_password:
-                        st.success(f"🎉 密码找到: **{found_password}** 🎉")
-                        st.balloons()
-                    else:
-                        st.error("未能在密码列表中找到正确的密码。")
-                    st.write(f"总耗时: {elapsed_time:.2f} 秒。")
-                    st.write(f"平均速度: {len(passwords_to_try) / elapsed_time if elapsed_time > 0 else float('inf'):.2f} 密码/秒。")
-
-
-    # --- 单个密码方法 ---
-    elif crack_method == '尝试单个密码':
-        single_password = st.text_input("输入要尝试的密码:", type="password")
-        if st.button("尝试此密码", key="single_crack"):
-            if not single_password:
-                st.warning("请输入密码。")
+                    st.warning("上传的密码列表为空或无效。")
             else:
-                st.info(f"正在尝试密码: {single_password}")
-                if try_decrypt_wallet(wallet_bytes, single_password):
-                    st.success(f"🎉 密码正确: **{single_password}** 🎉")
-                    st.balloons()
-                else:
-                    st.error("密码错误。")
-
-    # --- 基本暴力破解方法 ---
-    elif crack_method == '基本暴力破解 (非常不推荐，极慢)':
-        charset_input = st.text_input("字符集:", DEFAULT_CHARSET)
-        max_len_input = st.number_input("最大密码长度 (警告: 长度大于4会非常慢):", min_value=1, max_value=8, value=MAX_BRUTEFORCE_LEN)
-
-        st.warning(f"使用字符集 '{charset_input}' 和最大长度 {max_len_input} 进行暴力破解，"
-                   f"可能需要尝试 {sum(len(charset_input)**i for i in range(1, max_len_input + 1))} 个密码。")
-
-        if st.button("开始暴力破解 (请耐心等待)", key="brute_crack"):
-            st.info("开始暴力破解尝试...")
-            # 无法预估总数，因此不使用标准进度条
-            status_text = st.empty()
-            found_password = None
-            start_time = time.time()
-            count = 0
-            estimated_total = sum(len(charset_input)**i for i in range(1, max_len_input + 1))
-            progress_bar = st.progress(0)
-
+                st.info("请上传密码列表文件。")
 
-            for pwd in generate_passwords_bruteforce(charset_input, max_len_input):
-                count += 1
-                status_text.text(f"尝试密码 #{count}: {pwd}")
-                if try_decrypt_wallet(wallet_bytes, pwd):
-                    found_password = pwd
-                    break
-                if count % 100 == 0: # 每100次更新一次进度条，避免过于频繁
-                    progress_bar.progress(min(1.0, count / estimated_total if estimated_total > 0 else 0))
-                    # time.sleep(0.001) # 允许UI更新
 
+        elif crack_method == '尝试单个密码':
+            single_password = st.text_input("输入要尝试的密码:", type="password", key="single_pwd_input")
+            if single_password:
+                passwords_to_try_list = [single_password]
+                can_start = True
+                start_button_label = "尝试此密码"
+                method_name_for_task = "single"
+            else:
+                st.info("请输入要尝试的密码。")
 
-            end_time = time.time()
-            status_text.text("暴力破解尝试完成。")
-            elapsed_time = end_time - start_time
 
-            if found_password:
-                st.success(f"🎉 密码找到: **{found_password}** 🎉")
-                st.balloons()
+        elif crack_method == '基本暴力破解 (极慢，后台运行)':
+            charset_input = st.text_input("字符集:", DEFAULT_CHARSET, key="charset_input")
+            max_len_input = st.number_input(
+                "最大密码长度 (警告: 长度大于4会非常慢):",
+                min_value=1, max_value=8, value=MAX_BRUTEFORCE_LEN, key="maxlen_input"
+            )
+            estimated_total = sum(len(charset_input)**i for i in range(1, max_len_input + 1))
+            st.warning(f"使用字符集 '{charset_input}' 和最大长度 {max_len_input} 进行暴力破解，"
+                       f"可能需要尝试 {estimated_total} 个密码。这会非常耗时。")
+            brute_force_params = {'charset': charset_input, 'max_len': max_len_input}
+            can_start = True
+            start_button_label = "开始暴力破解 (后台)"
+            method_name_for_task = "bruteforce"
+
+
+        if can_start and st.button(start_button_label, key=f"start_button_{method_name_for_task}"):
+            st.session_state.task_counter += 1
+            new_task_id = st.session_state.task_counter
+            st.session_state.current_task_id = new_task_id
+
+            if method_name_for_task == "bruteforce":
+                passwords_iterable = generate_passwords_bruteforce(
+                    brute_force_params['charset'], brute_force_params['max_len']
+                )
             else:
-                st.error(f"在指定的字符集和长度内未能找到密码 (尝试了 {count} 个)。")
-            st.write(f"总耗时: {elapsed_time:.2f} 秒。")
-            st.write(f"平均速度: {count / elapsed_time if elapsed_time > 0 else float('inf'):.2f} 密码/秒。")
+                passwords_iterable = passwords_to_try_list
+
+            # 启动后台线程
+            thread = threading.Thread(
+                target=background_cracker_task,
+                args=(wallet_bytes, passwords_iterable, new_task_id, method_name_for_task)
+            )
+            thread.daemon = True # 允许主程序退出时线程也退出
+            thread.start()
+            st.rerun() # 立即重新运行脚本以显示 spinner
 
 else:
     st.info("请先上传 wallet.dat 文件。")