mirror of
https://gitea.tendokyu.moe/Lannamokia/SEGAfs-GUI.git
synced 2026-09-22 21:27:54 +03:00
898 lines
37 KiB
Python
898 lines
37 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
SegaFS GUI - 图形化界面工具
|
|
用于创建加密的文件系统容器
|
|
"""
|
|
|
|
import tkinter as tk
|
|
from tkinter import ttk, filedialog, messagebox, scrolledtext
|
|
import threading
|
|
import os
|
|
from datetime import datetime
|
|
import traceback
|
|
import queue
|
|
import concurrent.futures
|
|
|
|
# 导入原始脚本的功能和依赖
|
|
from math import ceil
|
|
import secrets
|
|
import struct
|
|
import time
|
|
import zlib
|
|
|
|
from Crypto.Cipher import AES, PKCS1_OAEP
|
|
from Crypto.Hash import HMAC, SHA1
|
|
from Crypto.PublicKey import RSA
|
|
from construct import (
|
|
Bytes,
|
|
Const,
|
|
IfThenElse,
|
|
Int16ul,
|
|
Int32ul,
|
|
Int64ul,
|
|
Int8ul,
|
|
Struct,
|
|
)
|
|
|
|
# 从原脚本导入常量
|
|
BTKEY = bytes.fromhex("09ca5efd30c9aaef3804d0a7e3fa7120")
|
|
BTIV = bytes.fromhex("b155c22c2e7f0491fa7f0fdc217aff90")
|
|
SIGKEY = bytes.fromhex(
|
|
"e1bdcb2d5e9ed3b5de234364dfa4d126849edff769fc6c28fba5f43bc482bd7479d676afce8188e1d3a6852f4ebce45cde46bd15e8ee5fe84d197f945a54518f"
|
|
)
|
|
HEADER_META_PUBKEY = RSA.import_key("""-----BEGIN PUBLIC KEY-----
|
|
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAsRMLnJuczNpfoqPpHQ3o
|
|
5XNkjKXO6P3ToV/45Az5dNaHVL7uEu9vPI7a2KYFQnNYgD3UUHFahfTcljzLOkcH
|
|
1aVrhm8gaB/5mygjUJWcN+kKyB7sASqhL22RC7NlxtDY15ozli/b0MagVoaBAV5D
|
|
MytUCa73GPRGY0x9v/wTvtmFclYCWjJ9c2QzrCrQ9eNTVyETwh5q6qKEARHGZgCX
|
|
rWmdCsa/+oS+3pLbUGFlHCSZtCvvWCJgmgurlTGAGzoxrieO6XDEg2AGiRprWWL2
|
|
BGNh7gwgnSq6FWnKSf2Qe7xoFcTpV5QhNFBQjrq0KnBDRfz5EXJnMoKxNYL6reqR
|
|
uwIDAQAB
|
|
-----END PUBLIC KEY-----
|
|
""")
|
|
|
|
# 数据结构定义
|
|
Timestamp = Struct(
|
|
"year" / Int16ul,
|
|
"month" / Int8ul,
|
|
"day" / Int8ul,
|
|
"hour" / Int8ul,
|
|
"minute" / Int8ul,
|
|
"second" / Int8ul,
|
|
"milli" / Int8ul,
|
|
)
|
|
|
|
Version = Struct(
|
|
"release" / Int8ul,
|
|
"minor" / Int8ul,
|
|
"major" / Int16ul,
|
|
)
|
|
|
|
BootID = Struct(
|
|
"length" / Const(0x2800, Int32ul),
|
|
"magic" / Const(b"BTID", Bytes(4)),
|
|
"unk1" / Int8ul,
|
|
"type" / Int8ul,
|
|
"sequence_number" / Int8ul,
|
|
"derive_iv" / Int8ul,
|
|
"game_id" / Bytes(4),
|
|
"game_timestamp" / Timestamp,
|
|
"game_version" / IfThenElse(lambda ctx: ctx.type == 0x02, Bytes(4), Version),
|
|
"block_count" / Int64ul,
|
|
"block_size" / Int64ul,
|
|
"header_block_count" / Int64ul,
|
|
"unk2" / Int64ul,
|
|
"hw_family" / Bytes(3),
|
|
"hw_generation" / Int8ul,
|
|
"orig_timestamp" / Timestamp,
|
|
"orig_version" / Version,
|
|
"os_version" / Version,
|
|
"strings" / Bytes(0x27AC),
|
|
)
|
|
|
|
def get_page_iv(iv: bytes, offset: int):
|
|
return bytes(x ^ (offset >> (8 * (i % 8))) & 0xFF for (i, x) in enumerate(iv))
|
|
|
|
# 简化为单线程处理模式
|
|
|
|
class SegaFSGUI:
|
|
def __init__(self, root):
|
|
self.root = root
|
|
self.root.title("SegaFS 容器生成器")
|
|
self.root.geometry("800x700")
|
|
self.root.resizable(True, True)
|
|
|
|
# 线程控制
|
|
self.generation_thread = None
|
|
self.stop_generation = False
|
|
|
|
# 创建主框架
|
|
self.create_widgets()
|
|
|
|
# 初始化默认值
|
|
self.load_default_values()
|
|
|
|
def create_widgets(self):
|
|
# 创建笔记本控件（标签页）
|
|
notebook = ttk.Notebook(self.root)
|
|
notebook.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
|
|
|
|
# 基本设置标签页
|
|
self.basic_frame = ttk.Frame(notebook)
|
|
notebook.add(self.basic_frame, text="基本设置")
|
|
self.create_basic_tab()
|
|
|
|
# 高级设置标签页
|
|
self.advanced_frame = ttk.Frame(notebook)
|
|
notebook.add(self.advanced_frame, text="高级设置")
|
|
self.create_advanced_tab()
|
|
|
|
# 日志输出标签页
|
|
self.log_frame = ttk.Frame(notebook)
|
|
notebook.add(self.log_frame, text="日志输出")
|
|
self.create_log_tab()
|
|
|
|
# 底部按钮框架
|
|
self.button_frame = ttk.Frame(self.root)
|
|
self.button_frame.pack(fill=tk.X, padx=10, pady=(0, 10))
|
|
|
|
# 按钮容器
|
|
button_container = ttk.Frame(self.button_frame)
|
|
button_container.pack(side=tk.RIGHT)
|
|
|
|
# 生成按钮
|
|
self.generate_btn = ttk.Button(
|
|
button_container,
|
|
text="生成容器",
|
|
command=self.generate_container,
|
|
style="Accent.TButton"
|
|
)
|
|
self.generate_btn.pack(side=tk.LEFT, padx=(0, 5))
|
|
|
|
# 终止按钮
|
|
self.stop_btn = ttk.Button(
|
|
button_container,
|
|
text="终止生成",
|
|
command=self.stop_generation_process,
|
|
state='disabled'
|
|
)
|
|
self.stop_btn.pack(side=tk.LEFT)
|
|
|
|
# 进度条框架
|
|
progress_frame = ttk.Frame(self.button_frame)
|
|
progress_frame.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 10))
|
|
|
|
# 进度条
|
|
self.progress = ttk.Progressbar(
|
|
progress_frame,
|
|
mode='determinate'
|
|
)
|
|
self.progress.pack(fill=tk.X, pady=(0, 2))
|
|
|
|
# 进度标签
|
|
self.progress_label = ttk.Label(
|
|
progress_frame,
|
|
text="准备就绪",
|
|
font=('Arial', 8)
|
|
)
|
|
self.progress_label.pack()
|
|
|
|
def create_basic_tab(self):
|
|
# 文件选择区域
|
|
file_group = ttk.LabelFrame(self.basic_frame, text="文件设置", padding=10)
|
|
file_group.pack(fill=tk.X, padx=10, pady=5)
|
|
|
|
# 输入文件
|
|
ttk.Label(file_group, text="输入文件:").grid(row=0, column=0, sticky=tk.W, pady=2)
|
|
self.input_file_var = tk.StringVar()
|
|
input_frame = ttk.Frame(file_group)
|
|
input_frame.grid(row=0, column=1, sticky=tk.EW, padx=(10, 0), pady=2)
|
|
file_group.columnconfigure(1, weight=1)
|
|
|
|
self.input_entry = ttk.Entry(input_frame, textvariable=self.input_file_var)
|
|
self.input_entry.pack(side=tk.LEFT, fill=tk.X, expand=True)
|
|
ttk.Button(input_frame, text="浏览", command=self.browse_input_file).pack(side=tk.RIGHT, padx=(5, 0))
|
|
|
|
# 输出文件
|
|
ttk.Label(file_group, text="输出文件:").grid(row=1, column=0, sticky=tk.W, pady=2)
|
|
self.output_file_var = tk.StringVar()
|
|
output_frame = ttk.Frame(file_group)
|
|
output_frame.grid(row=1, column=1, sticky=tk.EW, padx=(10, 0), pady=2)
|
|
|
|
self.output_entry = ttk.Entry(output_frame, textvariable=self.output_file_var)
|
|
self.output_entry.pack(side=tk.LEFT, fill=tk.X, expand=True)
|
|
ttk.Button(output_frame, text="浏览", command=self.browse_output_file).pack(side=tk.RIGHT, padx=(5, 0))
|
|
|
|
# 加密设置区域
|
|
crypto_group = ttk.LabelFrame(self.basic_frame, text="加密设置", padding=10)
|
|
crypto_group.pack(fill=tk.X, padx=10, pady=5)
|
|
|
|
# 加密密钥
|
|
ttk.Label(crypto_group, text="加密密钥 (Hex):").grid(row=0, column=0, sticky=tk.W, pady=2)
|
|
self.encryption_key_var = tk.StringVar()
|
|
self.key_entry = ttk.Entry(crypto_group, textvariable=self.encryption_key_var, width=50)
|
|
self.key_entry.grid(row=0, column=1, sticky=tk.EW, padx=(10, 0), pady=2)
|
|
crypto_group.columnconfigure(1, weight=1)
|
|
|
|
# 加密IV
|
|
ttk.Label(crypto_group, text="加密 IV (Hex):").grid(row=1, column=0, sticky=tk.W, pady=2)
|
|
self.encryption_iv_var = tk.StringVar()
|
|
self.iv_entry = ttk.Entry(crypto_group, textvariable=self.encryption_iv_var, width=50)
|
|
self.iv_entry.grid(row=1, column=1, sticky=tk.EW, padx=(10, 0), pady=2)
|
|
|
|
# 游戏信息区域
|
|
game_group = ttk.LabelFrame(self.basic_frame, text="游戏信息", padding=10)
|
|
game_group.pack(fill=tk.X, padx=10, pady=5)
|
|
|
|
# 游戏ID
|
|
ttk.Label(game_group, text="游戏 ID:").grid(row=0, column=0, sticky=tk.W, pady=2)
|
|
self.game_id_var = tk.StringVar()
|
|
ttk.Entry(game_group, textvariable=self.game_id_var, width=10).grid(row=0, column=1, sticky=tk.W, padx=(10, 0), pady=2)
|
|
|
|
# 游戏版本
|
|
ttk.Label(game_group, text="游戏版本:").grid(row=0, column=2, sticky=tk.W, padx=(20, 0), pady=2)
|
|
self.game_version_var = tk.StringVar()
|
|
ttk.Entry(game_group, textvariable=self.game_version_var, width=10).grid(row=0, column=3, sticky=tk.W, padx=(10, 0), pady=2)
|
|
|
|
# 硬件信息
|
|
ttk.Label(game_group, text="硬件系列:").grid(row=1, column=0, sticky=tk.W, pady=2)
|
|
self.hw_family_var = tk.StringVar()
|
|
ttk.Entry(game_group, textvariable=self.hw_family_var, width=10).grid(row=1, column=1, sticky=tk.W, padx=(10, 0), pady=2)
|
|
|
|
ttk.Label(game_group, text="硬件代数:").grid(row=1, column=2, sticky=tk.W, padx=(20, 0), pady=2)
|
|
self.hw_generation_var = tk.IntVar()
|
|
ttk.Entry(game_group, textvariable=self.hw_generation_var, width=10).grid(row=1, column=3, sticky=tk.W, padx=(10, 0), pady=2)
|
|
|
|
def create_advanced_tab(self):
|
|
# 容器类型设置
|
|
type_group = ttk.LabelFrame(self.advanced_frame, text="容器类型", padding=10)
|
|
type_group.pack(fill=tk.X, padx=10, pady=5)
|
|
|
|
ttk.Label(type_group, text="类型:").grid(row=0, column=0, sticky=tk.W, pady=2)
|
|
self.container_type_var = tk.IntVar()
|
|
type_frame = ttk.Frame(type_group)
|
|
type_frame.grid(row=0, column=1, sticky=tk.W, padx=(10, 0), pady=2)
|
|
|
|
ttk.Radiobutton(type_frame, text="OS (0)", variable=self.container_type_var, value=0).pack(side=tk.LEFT)
|
|
ttk.Radiobutton(type_frame, text="App (1)", variable=self.container_type_var, value=1).pack(side=tk.LEFT, padx=(10, 0))
|
|
ttk.Radiobutton(type_frame, text="Option (2)", variable=self.container_type_var, value=2).pack(side=tk.LEFT, padx=(10, 0))
|
|
|
|
# 时间戳设置
|
|
timestamp_group = ttk.LabelFrame(self.advanced_frame, text="游戏时间戳", padding=10)
|
|
timestamp_group.pack(fill=tk.X, padx=10, pady=5)
|
|
|
|
# 年月日
|
|
date_frame = ttk.Frame(timestamp_group)
|
|
date_frame.pack(fill=tk.X, pady=2)
|
|
|
|
ttk.Label(date_frame, text="年:").pack(side=tk.LEFT)
|
|
self.year_var = tk.IntVar()
|
|
ttk.Entry(date_frame, textvariable=self.year_var, width=8).pack(side=tk.LEFT, padx=(5, 10))
|
|
|
|
ttk.Label(date_frame, text="月:").pack(side=tk.LEFT)
|
|
self.month_var = tk.IntVar()
|
|
ttk.Entry(date_frame, textvariable=self.month_var, width=6).pack(side=tk.LEFT, padx=(5, 10))
|
|
|
|
ttk.Label(date_frame, text="日:").pack(side=tk.LEFT)
|
|
self.day_var = tk.IntVar()
|
|
ttk.Entry(date_frame, textvariable=self.day_var, width=6).pack(side=tk.LEFT, padx=(5, 10))
|
|
|
|
# 时分秒
|
|
time_frame = ttk.Frame(timestamp_group)
|
|
time_frame.pack(fill=tk.X, pady=2)
|
|
|
|
ttk.Label(time_frame, text="时:").pack(side=tk.LEFT)
|
|
self.hour_var = tk.IntVar()
|
|
ttk.Entry(time_frame, textvariable=self.hour_var, width=6).pack(side=tk.LEFT, padx=(5, 10))
|
|
|
|
ttk.Label(time_frame, text="分:").pack(side=tk.LEFT)
|
|
self.minute_var = tk.IntVar()
|
|
ttk.Entry(time_frame, textvariable=self.minute_var, width=6).pack(side=tk.LEFT, padx=(5, 10))
|
|
|
|
ttk.Label(time_frame, text="秒:").pack(side=tk.LEFT)
|
|
self.second_var = tk.IntVar()
|
|
ttk.Entry(time_frame, textvariable=self.second_var, width=6).pack(side=tk.LEFT, padx=(5, 10))
|
|
|
|
# 当前时间按钮
|
|
ttk.Button(timestamp_group, text="使用当前时间", command=self.set_current_time).pack(pady=5)
|
|
|
|
# 版本信息设置
|
|
version_group = ttk.LabelFrame(self.advanced_frame, text="版本信息", padding=10)
|
|
version_group.pack(fill=tk.X, padx=10, pady=5)
|
|
|
|
# OS版本
|
|
os_frame = ttk.Frame(version_group)
|
|
os_frame.pack(fill=tk.X, pady=2)
|
|
|
|
ttk.Label(os_frame, text="OS版本 - 主版本:").pack(side=tk.LEFT)
|
|
self.os_major_var = tk.IntVar()
|
|
ttk.Entry(os_frame, textvariable=self.os_major_var, width=8).pack(side=tk.LEFT, padx=(5, 10))
|
|
|
|
ttk.Label(os_frame, text="次版本:").pack(side=tk.LEFT)
|
|
self.os_minor_var = tk.IntVar()
|
|
ttk.Entry(os_frame, textvariable=self.os_minor_var, width=8).pack(side=tk.LEFT, padx=(5, 10))
|
|
|
|
ttk.Label(os_frame, text="发布版本:").pack(side=tk.LEFT)
|
|
self.os_release_var = tk.IntVar()
|
|
ttk.Entry(os_frame, textvariable=self.os_release_var, width=8).pack(side=tk.LEFT, padx=(5, 0))
|
|
|
|
# 其他设置
|
|
other_group = ttk.LabelFrame(self.advanced_frame, text="其他设置", padding=10)
|
|
other_group.pack(fill=tk.X, padx=10, pady=5)
|
|
|
|
# 块大小
|
|
ttk.Label(other_group, text="块大小 (Hex):").grid(row=0, column=0, sticky=tk.W, pady=2)
|
|
self.block_size_var = tk.StringVar()
|
|
ttk.Entry(other_group, textvariable=self.block_size_var, width=15).grid(row=0, column=1, sticky=tk.W, padx=(10, 0), pady=2)
|
|
|
|
# 序列号
|
|
ttk.Label(other_group, text="序列号:").grid(row=0, column=2, sticky=tk.W, padx=(20, 0), pady=2)
|
|
self.sequence_number_var = tk.IntVar()
|
|
ttk.Entry(other_group, textvariable=self.sequence_number_var, width=10).grid(row=0, column=3, sticky=tk.W, padx=(10, 0), pady=2)
|
|
|
|
def create_log_tab(self):
|
|
# 日志文本区域
|
|
self.log_text = scrolledtext.ScrolledText(
|
|
self.log_frame,
|
|
wrap=tk.WORD,
|
|
height=20,
|
|
font=('Consolas', 9)
|
|
)
|
|
self.log_text.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
|
|
|
|
# 清除日志按钮
|
|
clear_btn = ttk.Button(
|
|
self.log_frame,
|
|
text="清除日志",
|
|
command=self.clear_log
|
|
)
|
|
clear_btn.pack(pady=(0, 10))
|
|
|
|
def load_default_values(self):
|
|
"""加载默认配置值"""
|
|
# 设置默认值
|
|
self.encryption_key_var.set("f272e5016863af2ba0337f50de686f6e")
|
|
self.encryption_iv_var.set("5327e132631e7f71b61be7cc0df382ce")
|
|
|
|
self.game_id_var.set("SDGS")
|
|
self.game_version_var.set("A041")
|
|
self.hw_family_var.set("ACA")
|
|
self.hw_generation_var.set(0)
|
|
|
|
self.container_type_var.set(1) # App
|
|
|
|
# 默认时间戳
|
|
self.year_var.set(2023)
|
|
self.month_var.set(12)
|
|
self.day_var.set(28)
|
|
self.hour_var.set(16)
|
|
self.minute_var.set(34)
|
|
self.second_var.set(43)
|
|
|
|
# OS版本
|
|
self.os_major_var.set(80)
|
|
self.os_minor_var.set(54)
|
|
self.os_release_var.set(1)
|
|
|
|
# 其他
|
|
self.block_size_var.set("0x40000")
|
|
self.sequence_number_var.set(0)
|
|
|
|
def browse_input_file(self):
|
|
"""浏览输入文件"""
|
|
filename = filedialog.askopenfilename(
|
|
title="选择输入文件",
|
|
filetypes=[("所有文件", "*.*")]
|
|
)
|
|
if filename:
|
|
self.input_file_var.set(filename)
|
|
|
|
def browse_output_file(self):
|
|
"""浏览输出文件"""
|
|
filename = filedialog.asksaveasfilename(
|
|
title="选择输出文件",
|
|
defaultextension=".bin",
|
|
filetypes=[("二进制文件", "*.bin"), ("所有文件", "*.*")]
|
|
)
|
|
if filename:
|
|
self.output_file_var.set(filename)
|
|
|
|
def set_current_time(self):
|
|
"""设置当前时间"""
|
|
now = datetime.now()
|
|
self.year_var.set(now.year)
|
|
self.month_var.set(now.month)
|
|
self.day_var.set(now.day)
|
|
self.hour_var.set(now.hour)
|
|
self.minute_var.set(now.minute)
|
|
self.second_var.set(now.second)
|
|
|
|
def log_message(self, message):
|
|
"""添加日志消息"""
|
|
self.log_text.insert(tk.END, f"[{datetime.now().strftime('%H:%M:%S')}] {message}\n")
|
|
self.log_text.see(tk.END)
|
|
self.root.update_idletasks()
|
|
|
|
def clear_log(self):
|
|
"""清除日志"""
|
|
self.log_text.delete(1.0, tk.END)
|
|
|
|
def validate_inputs(self):
|
|
"""验证输入参数"""
|
|
if not self.input_file_var.get():
|
|
raise ValueError("请选择输入文件")
|
|
|
|
if not self.output_file_var.get():
|
|
raise ValueError("请选择输出文件")
|
|
|
|
if not os.path.exists(self.input_file_var.get()):
|
|
raise ValueError("输入文件不存在")
|
|
|
|
# 验证十六进制字符串
|
|
try:
|
|
bytes.fromhex(self.encryption_key_var.get())
|
|
except ValueError:
|
|
raise ValueError("加密密钥格式错误，请输入有效的十六进制字符串")
|
|
|
|
try:
|
|
bytes.fromhex(self.encryption_iv_var.get())
|
|
except ValueError:
|
|
raise ValueError("加密IV格式错误，请输入有效的十六进制字符串")
|
|
|
|
# 验证块大小
|
|
try:
|
|
int(self.block_size_var.get(), 16)
|
|
except ValueError:
|
|
raise ValueError("块大小格式错误，请输入有效的十六进制数值")
|
|
|
|
def get_config(self):
|
|
"""获取当前配置"""
|
|
config = {
|
|
'ENCRYPTION_KEY': bytes.fromhex(self.encryption_key_var.get()),
|
|
'ENCRYPTION_IV': bytes.fromhex(self.encryption_iv_var.get()),
|
|
'INPUT_FILE': self.input_file_var.get(),
|
|
'OUTPUT_FILE': self.output_file_var.get(),
|
|
'BOOTID': {
|
|
'unk1': 0x01,
|
|
'type': self.container_type_var.get(),
|
|
'sequence_number': self.sequence_number_var.get(),
|
|
'derive_iv': 0,
|
|
'game_id': self.game_id_var.get().encode('utf-8')[:4].ljust(4, b'\x00'),
|
|
'game_timestamp': {
|
|
'year': self.year_var.get(),
|
|
'month': self.month_var.get(),
|
|
'day': self.day_var.get(),
|
|
'hour': self.hour_var.get(),
|
|
'minute': self.minute_var.get(),
|
|
'second': self.second_var.get(),
|
|
'milli': 0,
|
|
},
|
|
'game_version': self._parse_game_version() if self.container_type_var.get() != 2 else self.game_version_var.get().encode('utf-8')[:4].ljust(4, b'\x00'),
|
|
'block_size': int(self.block_size_var.get(), 16),
|
|
'header_block_count': 8,
|
|
'unk2': 0,
|
|
'hw_family': self.hw_family_var.get().encode('utf-8')[:3].ljust(3, b'\x00'),
|
|
'hw_generation': self.hw_generation_var.get(),
|
|
'orig_timestamp': {
|
|
'year': 0, 'month': 0, 'day': 0,
|
|
'hour': 0, 'minute': 0, 'second': 0, 'milli': 0,
|
|
},
|
|
'orig_version': {'release': 0, 'minor': 0, 'major': 0},
|
|
'os_version': {
|
|
'release': self.os_release_var.get(),
|
|
'minor': self.os_minor_var.get(),
|
|
'major': self.os_major_var.get(),
|
|
},
|
|
'strings': b"\x00" * 0x27AC,
|
|
}
|
|
}
|
|
return config
|
|
|
|
def _parse_game_version(self):
|
|
"""解析游戏版本号为Version结构
|
|
|
|
支持多种格式:
|
|
- "A041" -> major=0x41, minor=0x0A, release=0x00 (字母+数字格式)
|
|
- "1.2.3" -> major=1, minor=2, release=3 (点分格式)
|
|
- "123" -> major=123, minor=0, release=0 (纯数字格式)
|
|
"""
|
|
version_str = self.game_version_var.get().strip()
|
|
|
|
if not version_str:
|
|
return {'release': 0, 'minor': 0, 'major': 0}
|
|
|
|
# 处理字母+数字格式 (如 "A041")
|
|
if len(version_str) >= 2 and version_str[0].isalpha():
|
|
try:
|
|
# 第一个字符作为minor版本 (A=10, B=11, etc.)
|
|
minor = ord(version_str[0].upper()) - ord('A') + 10
|
|
# 剩余数字作为major版本
|
|
major_str = version_str[1:]
|
|
if major_str.isdigit():
|
|
major = int(major_str)
|
|
else:
|
|
# 如果不是纯数字，尝试解析为十六进制
|
|
major = int(major_str, 16) if all(c in '0123456789ABCDEFabcdef' for c in major_str) else 0
|
|
return {'release': 0, 'minor': minor, 'major': major}
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
# 处理点分格式 (如 "1.2.3")
|
|
if '.' in version_str:
|
|
try:
|
|
parts = version_str.split('.')
|
|
major = int(parts[0]) if len(parts) > 0 and parts[0].isdigit() else 0
|
|
minor = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 0
|
|
release = int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 0
|
|
return {'release': release, 'minor': minor, 'major': major}
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
# 处理纯数字格式
|
|
if version_str.isdigit():
|
|
return {'release': 0, 'minor': 0, 'major': int(version_str)}
|
|
|
|
# 尝试解析为十六进制
|
|
try:
|
|
if all(c in '0123456789ABCDEFabcdef' for c in version_str):
|
|
major = int(version_str, 16)
|
|
return {'release': 0, 'minor': 0, 'major': major}
|
|
except ValueError:
|
|
pass
|
|
|
|
# 默认返回
|
|
return {'release': 0, 'minor': 0, 'major': 0}
|
|
|
|
def generate_container_thread(self):
|
|
"""在后台线程中生成容器"""
|
|
try:
|
|
self.log_message("开始生成容器...")
|
|
|
|
# 验证输入
|
|
self.validate_inputs()
|
|
self.log_message("输入验证通过")
|
|
|
|
# 获取配置
|
|
config = self.get_config()
|
|
self.log_message(f"输入文件: {config['INPUT_FILE']}")
|
|
self.log_message(f"输出文件: {config['OUTPUT_FILE']}")
|
|
|
|
# 验证输入文件是否存在
|
|
if not os.path.exists(config['INPUT_FILE']):
|
|
raise FileNotFoundError(f"输入文件不存在: {config['INPUT_FILE']}")
|
|
|
|
# 验证输出目录是否存在，如果不存在则创建
|
|
output_dir = os.path.dirname(config['OUTPUT_FILE'])
|
|
if output_dir and not os.path.exists(output_dir):
|
|
os.makedirs(output_dir)
|
|
self.log_message(f"创建输出目录: {output_dir}")
|
|
|
|
# 开始生成容器
|
|
self.log_message("配置已准备完成，开始处理文件...")
|
|
self.update_progress(5, "分析输入文件...")
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
# 获取文件大小并计算块数
|
|
filesize = os.stat(config['INPUT_FILE']).st_size
|
|
self.log_message(f"输入文件大小: {filesize/1024/1024/1024:.2f} GB ({filesize} 字节)")
|
|
|
|
if filesize == 0:
|
|
raise ValueError("输入文件为空")
|
|
|
|
config['BOOTID']['block_count'] = ceil(filesize / config['BOOTID']['block_size']) + 8
|
|
self.log_message(f"计算块数: {config['BOOTID']['block_count']}")
|
|
self.update_progress(10, "准备加密参数...")
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
# 生成头部元数据
|
|
self.log_message("生成头部元数据...")
|
|
self.update_progress(15, "生成头部元数据...")
|
|
header_key = secrets.token_bytes(16)
|
|
header_iv = secrets.token_bytes(16)
|
|
encrypted_keypair = PKCS1_OAEP.new(HEADER_META_PUBKEY).encrypt(header_key + header_iv)
|
|
header_meta = struct.pack("<Q", int(time.time())) + os.path.abspath(config['INPUT_FILE']).encode("utf-8") + b"\x00"
|
|
header_meta += secrets.token_bytes(config['BOOTID']['block_size'] - len(header_meta) - len(encrypted_keypair))
|
|
header_meta = encrypted_keypair + AES.new(header_key, AES.MODE_CBC, header_iv).encrypt(header_meta)
|
|
header_meta_crc32 = zlib.crc32(header_meta)
|
|
|
|
block_crc32s = [0, header_meta_crc32, header_meta_crc32, header_meta_crc32,
|
|
header_meta_crc32, header_meta_crc32, header_meta_crc32, header_meta_crc32]
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
# 开始写入文件
|
|
self.log_message("开始写入输出文件...")
|
|
self.update_progress(20, "创建输出文件...")
|
|
with open(config['INPUT_FILE'], "rb") as fin, open(config['OUTPUT_FILE'], "w+b") as fout:
|
|
# 写入BootID
|
|
self.log_message("写入BootID头部...")
|
|
self.update_progress(25, "写入BootID头部...")
|
|
cipher = AES.new(BTKEY, AES.MODE_CBC, BTIV)
|
|
bootid = BootID.build(config['BOOTID'])
|
|
bootid_crc32 = zlib.crc32(bootid)
|
|
bootid_bytes = cipher.encrypt(struct.pack("<I", bootid_crc32) + bootid)
|
|
fout.write(bootid_bytes)
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
# 写入随机字节填充
|
|
fout.write(secrets.token_bytes(config['BOOTID']['block_size'] - 0x2800))
|
|
|
|
# 写入头部元数据块
|
|
self.log_message("写入头部元数据块...")
|
|
self.update_progress(30, "写入头部元数据...")
|
|
for i in range(config['BOOTID']['header_block_count'] - 1):
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
fout.write(header_meta)
|
|
|
|
# 加密文件内容 - 单线程处理
|
|
self.log_message("开始加密文件内容...")
|
|
self.update_progress(35, "开始加密文件内容...")
|
|
|
|
# 使用单线程处理
|
|
self.log_message(f"文件大小: {filesize/1024/1024/1024:.2f}GB，使用单线程处理")
|
|
total_written, block_crc32s = self._process_file_singlethreaded(
|
|
fin, fout, config, filesize, block_crc32s
|
|
)
|
|
|
|
# 获取当前块的CRC32值（用于后续填充）
|
|
block_crc32 = 0 # 重置为0，因为_process_file_singlethreaded已处理完整块
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
# 填充未完成的块
|
|
if (total_written % config['BOOTID']['block_size']) != 0:
|
|
self.log_message("填充未完成的块...")
|
|
self.update_progress(86, "填充数据块...")
|
|
null_byte_count = config['BOOTID']['block_size'] - (total_written % config['BOOTID']['block_size'])
|
|
|
|
while null_byte_count > 0:
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
page_iv = get_page_iv(config['ENCRYPTION_IV'], total_written)
|
|
cipher = AES.new(config['ENCRYPTION_KEY'], AES.MODE_CBC, page_iv)
|
|
encrypted = cipher.encrypt(b"\x00" * 4096)
|
|
total_written += fout.write(encrypted)
|
|
block_crc32 = zlib.crc32(encrypted, block_crc32)
|
|
null_byte_count -= 4096
|
|
|
|
block_crc32s.append(block_crc32)
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
# 写入CRC32校验和
|
|
self.log_message("写入CRC32校验和...")
|
|
self.update_progress(88, "写入CRC32校验和...")
|
|
fout.seek(0x2A04)
|
|
for crc32 in block_crc32s[1:]:
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
fout.write(struct.pack("<I", crc32))
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
# 计算第一个块的CRC32
|
|
self.update_progress(92, "计算块校验和...")
|
|
fout.seek(0)
|
|
block_0_crc32 = zlib.crc32(fout.read(0x2800))
|
|
fout.seek(0x204, os.SEEK_CUR)
|
|
block_0_crc32 = zlib.crc32(fout.read(config['BOOTID']['block_size'] - 0x2800 - 0x204), block_0_crc32)
|
|
|
|
fout.seek(0x2A00)
|
|
fout.write(struct.pack("<I", block_0_crc32))
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
# 计算HMAC签名
|
|
self.log_message("计算HMAC签名...")
|
|
self.update_progress(95, "计算HMAC签名...")
|
|
fout.seek(0x2A00)
|
|
hmac = HMAC.new(SIGKEY, digestmod=SHA1)
|
|
to_read = config['BOOTID']['block_size'] * 8 - 0x2A00
|
|
|
|
while to_read > 0:
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
block = fout.read(min(0x1000, to_read))
|
|
if len(block) == 0:
|
|
raise Exception("计算签名时数据不足")
|
|
hmac.update(block)
|
|
to_read -= len(block)
|
|
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return
|
|
|
|
fout.seek(0x2800)
|
|
fout.write(hmac.digest())
|
|
|
|
# 完成进度更新
|
|
self.update_progress(100, "容器生成完成！")
|
|
self.log_message("容器生成完成！")
|
|
self.log_message(f"输出文件: {config['OUTPUT_FILE']}")
|
|
output_size = os.path.getsize(config['OUTPUT_FILE'])
|
|
self.log_message(f"文件大小: {output_size/1024/1024/1024:.2f} GB ({output_size} 字节)")
|
|
|
|
# 在主线程中显示完成消息
|
|
self.root.after(0, lambda: messagebox.showinfo("成功", "容器生成完成！"))
|
|
|
|
except FileNotFoundError as e:
|
|
error_msg = f"文件错误: {str(e)}"
|
|
self.log_message(error_msg)
|
|
self.root.after(0, lambda: messagebox.showerror("文件错误", error_msg))
|
|
except ValueError as e:
|
|
error_msg = f"参数错误: {str(e)}"
|
|
self.log_message(error_msg)
|
|
self.root.after(0, lambda: messagebox.showerror("参数错误", error_msg))
|
|
except PermissionError as e:
|
|
error_msg = f"权限错误: {str(e)}"
|
|
self.log_message(error_msg)
|
|
self.root.after(0, lambda: messagebox.showerror("权限错误", error_msg))
|
|
except Exception as e:
|
|
error_msg = f"生成容器时发生未知错误: {str(e)}"
|
|
self.log_message(error_msg)
|
|
self.log_message(f"错误详情: {traceback.format_exc()}")
|
|
self.root.after(0, lambda: messagebox.showerror("错误", error_msg))
|
|
|
|
finally:
|
|
# 如果生成过程被终止或发生异常，清理未完成的输出文件
|
|
if self.stop_generation or 'config' in locals():
|
|
output_file = self.output_file_var.get()
|
|
if output_file and os.path.exists(output_file):
|
|
try:
|
|
# 检查文件是否完整（通过检查文件大小或其他标志）
|
|
if self.stop_generation:
|
|
os.remove(output_file)
|
|
self.log_message(f"已清理未完成的输出文件: {output_file}")
|
|
except Exception as e:
|
|
self.log_message(f"清理输出文件失败: {str(e)}")
|
|
|
|
# 在主线程中停止进度条并启用按钮
|
|
self.root.after(0, self.generation_finished)
|
|
|
|
def stop_generation_process(self):
|
|
"""终止生成过程"""
|
|
self.stop_generation = True
|
|
self.log_message("用户请求终止生成过程...")
|
|
|
|
# 如果有正在运行的生成线程，等待其结束
|
|
if hasattr(self, 'generation_thread') and self.generation_thread and self.generation_thread.is_alive():
|
|
self.log_message("正在终止生成线程...")
|
|
self.generation_thread.join(timeout=2.0) # 等待最多2秒
|
|
|
|
# 删除可能已创建的输出文件
|
|
output_file = self.output_file_var.get()
|
|
if output_file and os.path.exists(output_file):
|
|
try:
|
|
os.remove(output_file)
|
|
self.log_message(f"已删除未完成的输出文件: {output_file}")
|
|
except Exception as e:
|
|
self.log_message(f"删除输出文件失败: {str(e)}")
|
|
|
|
# 重置状态
|
|
self.progress['value'] = 0
|
|
self.progress_label.config(text="已停止")
|
|
self.generate_btn.config(state='normal')
|
|
self.stop_btn.config(state='disabled')
|
|
|
|
def update_progress(self, value, text=""):
|
|
"""更新进度条和标签"""
|
|
def update():
|
|
self.progress['value'] = value
|
|
if text:
|
|
self.progress_label.config(text=text)
|
|
self.root.after(0, update)
|
|
|
|
def _process_file_singlethreaded(self, fin, fout, config, filesize, block_crc32s):
|
|
"""单线程文件处理方法"""
|
|
total_written = 0
|
|
to_read = filesize
|
|
block_crc32 = 0
|
|
processed_bytes = 0
|
|
last_update_time = time.time()
|
|
|
|
while to_read > 0:
|
|
# 检查是否被终止
|
|
if self.stop_generation:
|
|
self.log_message("生成过程已被用户终止")
|
|
return total_written, block_crc32s
|
|
|
|
page_iv = get_page_iv(config['ENCRYPTION_IV'], total_written)
|
|
contents = fin.read(4096)
|
|
contents_len = len(contents)
|
|
to_read -= contents_len
|
|
processed_bytes += contents_len
|
|
|
|
# 更新进度 - 每秒更新一次
|
|
current_time = time.time()
|
|
if current_time - last_update_time >= 1.0 or to_read == 0: # 每秒更新一次
|
|
current_progress = 35 + (processed_bytes / filesize) * 50
|
|
progress_text = f"加密文件: {processed_bytes/1024/1024/1024:.2f}/{filesize/1024/1024/1024:.2f} GB ({processed_bytes/filesize*100:.1f}%)"
|
|
self.update_progress(current_progress, progress_text)
|
|
self.log_message(f"处理进度: {processed_bytes/filesize*100:.1f}% ({processed_bytes/1024/1024/1024:.2f}/{filesize/1024/1024/1024:.2f} GB)")
|
|
last_update_time = current_time
|
|
|
|
if contents_len < 4096:
|
|
contents += b"\x00" * (4096 - contents_len)
|
|
|
|
cipher = AES.new(config['ENCRYPTION_KEY'], AES.MODE_CBC, page_iv)
|
|
encrypted = cipher.encrypt(contents)
|
|
total_written += fout.write(encrypted)
|
|
block_crc32 = zlib.crc32(encrypted, block_crc32)
|
|
|
|
if (total_written % config['BOOTID']['block_size']) == 0:
|
|
block_crc32s.append(block_crc32)
|
|
block_crc32 = 0
|
|
|
|
return total_written, block_crc32s
|
|
|
|
# 移除多进程处理方法
|
|
|
|
def generation_finished(self):
|
|
"""生成完成后的清理工作"""
|
|
self.progress['value'] = 0
|
|
self.progress_label.config(text="准备就绪")
|
|
self.generate_btn.config(state='normal', text='生成容器')
|
|
self.stop_btn.config(state='disabled')
|
|
self.stop_generation = False
|
|
self.generation_thread = None
|
|
|
|
def generate_container(self):
|
|
"""生成容器"""
|
|
# 禁用生成按钮，启用终止按钮
|
|
self.generate_btn.config(state='disabled', text='生成中...')
|
|
self.stop_btn.config(state='normal')
|
|
self.progress['value'] = 0
|
|
self.progress_label.config(text="初始化...")
|
|
self.stop_generation = False
|
|
|
|
# 在后台线程中执行生成
|
|
self.generation_thread = threading.Thread(target=self.generate_container_thread)
|
|
self.generation_thread.daemon = True
|
|
self.generation_thread.start()
|
|
|
|
def main():
|
|
root = tk.Tk()
|
|
app = SegaFSGUI(root)
|
|
root.mainloop()
|
|
|
|
if __name__ == "__main__":
|
|
main() |