(测试版)适用于AssetBundle的ArchiveVersion结构适配脚本

调试AssetBundle时,Vibe Coding了个小玩意[1],支持根据AssetBundle的ArchiveVersion对包结构进行适配调整,写回版本号,使得修改后的AssetBundle可以直接在AssetStudio加载。

顺便又增加了反向检查调整ArchiveVersion的功能和降低ArchiveVersion的功能(\x08→\x06)。

UnityPy版脚本

安装依赖库

python -m pip install UnityPy
代码内容
import os
import sys
import argparse
import UnityPy
from UnityPy.enums import ArchiveFlags
from io import BytesIO

def _find_valid_signature(data: bytes) -> int:
    """
    扫描所有 UnityFS\x00 签名,返回第一个通过验证的偏移。
    验证条件:
      1. UnityVersion 为 b'5.x.x\x00'
      2. 文件头中的 FileSize 与实际剩余数据长度一致
    """
    sig = b'UnityFS\x00'
    offset = 0
    while True:
        pos = data.find(sig, offset)
        if pos == -1:
            raise ValueError("未找到任何 UnityFS 签名")
        
        # 检查 UnityVersion
        ver_start = pos + len(sig) + 4  # 跳过 signature 和 ArchiveVersion
        if ver_start + 6 > len(data):
            offset = pos + 1
            continue
        unity_version = data[ver_start:ver_start+6]
        if unity_version != b'5.x.x\x00':
            offset = pos + 1
            continue
        
        # 解析 UnityRevision 长度(从 ver_start+6 到第一个 \x00)
        rev_start = ver_start + 6
        null_idx = data.find(b'\x00', rev_start)
        if null_idx == -1:
            offset = pos + 1
            continue
        rev_len = null_idx - rev_start + 1  # 包含 \x00
        
        # 读取 FileSize(8 字节,位于 revision 之后)
        fs_start = rev_start + rev_len
        if fs_start + 8 > len(data):
            offset = pos + 1
            continue
        file_size = int.from_bytes(data[fs_start:fs_start+8], 'big')
        
        # 计算从该签名到文件末尾的实际长度
        actual_size = len(data) - pos
        
        # 比较(允许微小偏差,例如某些打包工具会留尾)
        if file_size > actual_size :  # 允许 1KB 误差
            offset = pos + 1
            continue
        
        # 所有验证通过
        return pos

def _get_bundle(environment):
    """统一返回可修改的 BundleFile 对象。"""
    file_obj = getattr(environment, "file", None)
    if file_obj is not None and type(file_obj).__name__ == "BundleFile":
        return file_obj
    for value in environment.files.values():
        if type(value).__name__ == "BundleFile":
            return value
        inner_files = getattr(value, "files", None)
        if isinstance(inner_files, dict):
            for sub in inner_files.values():
                if type(sub).__name__ == "BundleFile":
                    return sub
    raise TypeError("未在 environment.files 中找到 BundleFile")

def _get_serialized_files(environment):
    """遍历 environment.files,返回所有 SerializedFile 对象。"""
    results = []
    seen_ids = set()
    for value in environment.files.values():
        class_name = type(value).__name__
        if class_name == "SerializedFile":
            results.append(value)
        elif class_name in ("BundleFile", "WebFile"):
            for attr_name in ("files", "serialized_files"):
                inner_dict = getattr(value, attr_name, None)
                if isinstance(inner_dict, dict):
                    for sub in inner_dict.values():
                        if type(sub).__name__ == "SerializedFile" and id(sub) not in seen_ids:
                            seen_ids.add(id(sub))
                            results.append(sub)
    return results

# 核心函数
def Ono_Manami(input_path: str, output_path: str, new_revision: str | None = None, downgrade: bool = False) -> None:
    """修改 AssetBundle 的 UnityRevision 或执行降版本操作。"""

    # ── 0. 数据清洗 ──
    with open(input_path, "rb") as f:
        raw = f.read()

    valid_offset = _find_valid_signature(raw)
    if valid_offset != 0:
        print(f"[Info] 检测到假签名,使用偏移 {hex(valid_offset)} 处的真实签名")
        working_data = bytearray(raw[valid_offset:])   # 截取后的数据
    else:
        working_data = bytearray(raw)

    # ── 1. 设置 fallback 版本 ──
    if new_revision:
        fallback = new_revision.rstrip("\x00")
    else:
        fallback = "2007.5.05f1"
        print("[Info] 未提供 new_revision")
    UnityPy.config.FALLBACK_UNITY_VERSION = fallback

    # ── 2. 加载文件 ──
    stream = BytesIO(bytes(working_data))
    environment = UnityPy.load(stream)
    stream.close()

    # ── 3. 获取 BundleFile ──
    try:
        bundle = _get_bundle(environment)
    except TypeError as error:
        print(f"[Error] {error}")
        sys.exit(1)

    # ── 4. 修改 BundleFile ( 含 SerializedFile ) 中的 UnityRevision ──
    if new_revision is not None:
        bundle.version_engine = new_revision.rstrip("\x00")
        serialized_files = _get_serialized_files(environment)
        for index, serialized_file in enumerate(serialized_files):
            if new_revision is not None:
                clean_revision = new_revision.rstrip("\x00")
                if hasattr(serialized_file, "set_version"):
                    serialized_file.set_version(clean_revision)
                else:
                    serialized_file.unique_version = clean_revision
        if bundle is not None:
            bundle.is_changed = True

        # ── 5.保存:优先原样压缩,失败回退 LZ4 ──
        original_compression = int(bundle.dataflags) & 0x3F
        compression_name_map = {0: "None", 1: "LZMA", 2: "LZ4", 3: "LZ4HC"}
        original_compression_name = compression_name_map.get(original_compression, f"Unknown({original_compression})")
        bundle_bytes = None
        try:
            if original_compression in (2, 3):
                bundle_bytes = environment.file.save(packer="lz4")
                print(f"使用原始压缩格式保存:{original_compression_name}")
            elif original_compression == 0:
                bundle_bytes = environment.file.save(packer="none")
                print("使用原始压缩格式保存:None")
            else:
                raise RuntimeError(f"原始压缩 {original_compression_name} 不在 UnityPy 安全写回范围内")
        except Exception as e:
            print(f"[Warning] 原样压缩保存失败:{e},回退 LZ4")

        if not bundle_bytes:
            bundle_bytes = environment.file.save(packer="lz4")
            print("已回退到 LZ4 压缩保存")
            if not bundle_bytes:
                raise RuntimeError("environment.file.save() 返回空字节")
        bundle_data = bundle_bytes
    else:
        bundle_data = working_data

    # ── 6. 二进制解析与降版本/反向检查 ──
    data = bytearray(bundle_data)
    signature = b'UnityFS\x00'
    sig_len = len(signature)
    pos = sig_len

    # 读取 ArchiveVersion
    ArchiveVersion = int.from_bytes(data[pos:pos+4], 'big')
    pos += 4

    # 读取 UnityVersion (6字节)
    unity_version = data[pos:pos+6]
    pos += 6

    # 找到 UnityRevision (从pos开始到第一个\x00, 包含\x00)
    null_idx = data.find(b'\x00', pos)
    if null_idx == -1:
        raise ValueError("未找到UnityRevision的终止符")
    revision_len = null_idx - pos + 1
    pos = null_idx + 1

    # 读取 FileSize (8字节)
    Filesize = int.from_bytes(data[pos:pos+8], 'big')
    pos += 8

    # 读取 compressedBlockInfoSize (4字节)
    compressed_block_info_size = int.from_bytes(data[pos:pos+4], 'big')
    pos += 4

    # 读取 decompressedBlockInfoSize (4字节)
    decompressed_block_info_size = int.from_bytes(data[pos:pos+4], 'big')
    pos += 4

    # 读取 flags (4字节)
    flags = int.from_bytes(data[pos:pos+4], 'big')
    pos += 4  # 此时 pos 指向 flags 结束后的位置

    # ── 7. 降版本模式 ──
    if downgrade and ArchiveVersion >= 7:
        print("-----🦀·🦀-----")
        ArchiveVersion = 6
        # 回写 version
        data[sig_len:sig_len+4] = ArchiveVersion.to_bytes(4, 'big')

        # 修改 flags:保留最后一个字节,其余置0
        last_byte = data[pos-1]
        new_flags = (0 << 24) | (0 << 16) | (0 << 8) | last_byte
        data[pos-4:pos] = new_flags.to_bytes(4, 'big')
        flags = new_flags

        # 删除 flags_end 后的填充
        flags_end = pos
        first_nonzero = None
        for i in range(flags_end, len(data)):
            if data[i] != 0:
                first_nonzero = i
                break
        if first_nonzero is not None:
            fill_len = first_nonzero - flags_end
            if fill_len > 0:
                data = data[:flags_end] + data[first_nonzero:]

        # 对 compressed data 起始位置做同样反向对齐
        cd_start = flags_end + compressed_block_info_size
        first_nonzero2 = None
        for i in range(cd_start, len(data)):
            if data[i] != 0:
                first_nonzero2 = i
                break
        if first_nonzero2 is not None:
            fill_len2 = first_nonzero2 - cd_start
            if fill_len2 > 0:
                data = data[:cd_start] + data[first_nonzero2:]

        # 降级完成,跳过后续对齐
        need_align = False

    # ── 8. 反向检查(仅当无版本变更且未降级时触发)──
    elif new_revision is None and ArchiveVersion < 7:
        need_align = False
        if ArchiveVersion >= 7:
            need_align = True
        else:
            # 反向检查:检查是否有对齐填充
            flags_end = pos
            first_nonzero = None
            for i in range(flags_end, len(data)):
                if data[i] != 0:
                    first_nonzero = i
                    break
            if first_nonzero is not None:
                Y_check = first_nonzero - flags_end
                if Y_check > 0:
                    ArchiveVersion = 8
                    need_align = True
                    data[sig_len:sig_len+4] = ArchiveVersion.to_bytes(4, 'big')
                    print("[Info] 反向检查检测到填充,已将 ArchiveVersion 提升至 8")

        if need_align:
            flags_end = pos
            X = (16 - (flags_end % 16)) % 16
            first_nonzero = None
            for i in range(flags_end, len(data)):
                if data[i] != 0:
                    first_nonzero = i
                    break
            if first_nonzero is None:
                raise ValueError("未找到compressedBlockInfo起始字节")
            Y = first_nonzero - flags_end
            data = data[:flags_end] + b'\x00' * X + data[first_nonzero:]
            pos = flags_end + X

            # 读取 compressedBlockInfo (不解析,仅确定结束位置)
            compressed_block_info_end = pos + compressed_block_info_size

            # 对 compressed data 起始做对齐调整
            cd_start = compressed_block_info_end
            X2 = (16 - (cd_start % 16)) % 16
            first_nonzero2 = None
            for i in range(cd_start, len(data)):
                if data[i] != 0:
                    first_nonzero2 = i
                    break
            if first_nonzero2 is None:
                raise ValueError("未找到compressed data起始字节")
            Y2 = first_nonzero2 - cd_start
            data = data[:cd_start] + b'\x00' * X2 + data[first_nonzero2:]

    # ── 9. 写入输出文件 ──
    with open(output_path, "wb") as f:
        f.write(data)

    # ── 10. 校验输出 ──
    if os.path.exists(output_path):
        size = os.path.getsize(output_path)
        print(f"修改完成。输出文件: {output_path}  ({size} 字节)")
        if size == 0:
            print("⚠ 警告:输出文件大小为 0")
    else:
        print(f"⚠ 警告:未生成输出文件 {output_path}")
# ──────────────────────────────────────────────
# 命令行入口
# ──────────────────────────────────────────────
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="AssetBundle UnityRevision 注入 / 降版本工具 (UnityPy 版)")
    parser.add_argument("input_file", help="输入文件路径")
    parser.add_argument("new_revision", nargs="?", default=None,help="一个新的 UnityRevision(可选)")
    parser.add_argument("--OnoManami", action="store_true",help="降版本模式(新锐测试中)(ArchiveVersion >= 7 -> 6)")
    args = parser.parse_args()

    input_file_path = args.input_file
    new_revision_value = args.new_revision
    downgrade_mode = args.OnoManami

    directory = os.path.dirname(input_file_path) or "."
    base_name = os.path.splitext(os.path.basename(input_file_path))[0]
    extension = os.path.splitext(input_file_path)[1]
    output_file_path = os.path.join(directory, f"{base_name}_test{extension}")

    Ono_Manami(input_file_path, output_file_path,new_revision_value, downgrade=downgrade_mode)
    print("感谢您参与测试!")

简单使用:在终端打开

python script.py <输入文件> [<新UnityRevision>]
lz4版脚本(实验性)

没写注入版本号的代码,甚至没有写重压缩的代码,因为lz4类型可以直接注入,而lzma需要解压后才能注入,从经验上讲,应该可以不需要导入lz4库支持。
安装依赖库

python -m pip install lz4
import os
import sys
import argparse
import struct
import lz4.block
import lzma

def find_valid_signature(data: bytes) -> int:
    """
    扫描所有 UnityFS\x00 签名,返回第一个通过验证的偏移。
    验证:UnityVersion == b'5.x.x\x00',且 FileSize 不超过实际剩余长度。
    """
    sig = b'UnityFS\x00'
    offset = 0
    while True:
        pos = data.find(sig, offset)
        if pos == -1:
            raise ValueError("未找到任何 UnityFS 签名")

        ver_start = pos + len(sig) + 4
        if ver_start + 6 > len(data):
            offset = pos + 1
            continue
        if data[ver_start:ver_start + 6] != b'5.x.x\x00':
            offset = pos + 1
            continue

        rev_start = ver_start + 6
        null_idx = data.find(b'\x00', rev_start)
        if null_idx == -1:
            offset = pos + 1
            continue
        rev_len = null_idx - rev_start + 1

        fs_start = rev_start + rev_len
        if fs_start + 8 > len(data):
            offset = pos + 1
            continue
        file_size = int.from_bytes(data[fs_start:fs_start + 8], 'big')
        actual_size = len(data) - pos
        if file_size > actual_size:
            offset = pos + 1
            continue
        return pos


def read_cstr(data: bytes, pos: int) -> tuple[str, int]:
    end = data.find(b'\x00', pos)
    if end == -1:
        raise ValueError("未找到字符串终止符")
    return data[pos:end].decode('utf-8', 'replace'), end + 1


def _read_extra_length_lz4ak(data: bytes, cur_pos: int, max_pos: int) -> tuple[int, int]:
    """读取 LZ4AK 的变长 extra length(0xFF 继续累加)"""
    l = 0
    while cur_pos < max_pos:
        b = data[cur_pos]
        l += b
        cur_pos += 1
        if b != 0xFF:
            break
    return l, cur_pos


def decompress_lz4ak(compressed_data: bytes, uncompressed_size: int) -> bytes:
    """LZ4AK -> 标准 LZ4 格式转换后解压"""
    ip = 0
    fixed = bytearray(compressed_data)
    compressed_size = len(fixed)

    while ip < compressed_size:
        # Token: 交换高低 4 位
        literal_length = fixed[ip] & 0xF
        match_length = (fixed[ip] >> 4) & 0xF
        fixed[ip] = (literal_length << 4) | match_length
        ip += 1

        # Literal extra length
        if literal_length == 0xF:
            l, ip = _read_extra_length_lz4ak(fixed, ip, compressed_size)
            literal_length += l

        # 跳过 literal 数据
        ip += literal_length
        if ip >= compressed_size:
            break

        # Offset: 大端 -> 小端
        if ip + 1 >= compressed_size:
            break
        offset = (fixed[ip] << 8) | fixed[ip + 1]
        fixed[ip] = offset & 0xFF
        fixed[ip + 1] = (offset >> 8) & 0xFF
        ip += 2

        # Match length extra
        if match_length == 0xF:
            l, ip = _read_extra_length_lz4ak(fixed, ip, compressed_size)
            # match_length += l  # 不需要写回 fixed,token 里已经是正确值

    return lz4.block.decompress(bytes(fixed), uncompressed_size=uncompressed_size)


def decompress_raw(comp: bytes, unc_size: int, comp_type: int) -> bytes:
    """UnityFS 裸块解压:0=None 1=LZMA 2=LZ4 3=LZ4HC 4=LZ4AK"""
    t = comp_type & 0x3F
    if t == 0:
        return bytes(comp[:unc_size])
    if t == 1:
        # Unity LZMA: 5-byte properties + raw LZMA data
        props = comp[:5]
        # 解析 LZMA 属性
        lc = props[0] % 9
        lp = (props[0] // 9) % 5
        pb = props[0] // 45
        dict_size = props[1] | (props[2] << 8) | (props[3] << 16) | (props[4] << 24)
        if dict_size == 0:
            dict_size = 1 << 20  # LZMA 规范:0 表示 1MB
        
        decompressor = lzma.LZMADecompressor(
            format=lzma.FORMAT_RAW,
            filters=[{
                'id': lzma.FILTER_LZMA1,
                'dict_size': dict_size,
                'lc': lc,
                'lp': lp,
                'pb': pb
            }]
        )
        # 跳过 5 字节属性头,解压剩余数据
        return decompressor.decompress(bytes(comp[5:]), max_length=unc_size)
    if t in (2, 3):
        return lz4.block.decompress(bytes(comp), uncompressed_size=unc_size)
    if t == 4:
        return decompress_lz4ak(bytes(comp), unc_size)
    raise NotImplementedError(f"压缩类型 {t} 未支持")


def parse_blocks_info(bi: bytes) -> tuple[list, list]:
    p = 16  # 跳过 16 字节 hash
    (block_count,) = struct.unpack_from('>i', bi, p); p += 4
    blocks = []
    for _ in range(block_count):
        unc, comp, fl = struct.unpack_from('>IIH', bi, p); p += 10
        blocks.append({"unc": unc, "comp": comp, "flags": fl})

    (node_count,) = struct.unpack_from('>i', bi, p); p += 4
    nodes = []
    for _ in range(node_count):
        off, size = struct.unpack_from('>qq', bi, p); p += 16
        (nfl,) = struct.unpack_from('>I', bi, p); p += 4
        name, p = read_cstr(bi, p)
        nodes.append({"offset": off, "size": size, "flags": nfl, "name": name})
    return blocks, nodes


def unpack_asset_bundle(input_path: str, output_path: str) -> None:
    with open(input_path, "rb") as f:
        raw = f.read()

    sig_offset = find_valid_signature(raw)
    data = bytearray(raw[sig_offset:] if sig_offset != 0 else raw)
    if sig_offset != 0:
        print(f"[Info] 使用偏移 {hex(sig_offset)} 处的真实签名")

    pos = 8  # 跳过 'UnityFS\x00'
    archive_version = struct.unpack_from('>I', data, pos)[0]; pos += 4

    unity_version, pos = read_cstr(data, pos)
    unity_revision, pos = read_cstr(data, pos)

    file_size = struct.unpack_from('>Q', data, pos)[0]; pos += 8
    comp_bi_size = struct.unpack_from('>I', data, pos)[0]; pos += 4
    unc_bi_size = struct.unpack_from('>I', data, pos)[0]; pos += 4
    bi_flags = struct.unpack_from('>I', data, pos)[0]; pos += 4

    print(f"[Header] ver={archive_version} unity={unity_version} "
          f"compBI={comp_bi_size} uncBI={unc_bi_size} flags=0x{bi_flags:X}")

    # ---- 对齐:header 结束到 BlockInfo 之间 ----
    header_end = pos
    if archive_version <= 6:
        pad_x = 0
    else:  # ver >= 7
        pad_x = (16 - (header_end % 16)) % 16

    bi_start = header_end + pad_x
    bi_comp = data[bi_start:bi_start + comp_bi_size]
    block_info = decompress_raw(bi_comp, unc_bi_size, bi_flags)
    print(f"[BlockInfo] {len(bi_comp)} -> {len(block_info)}")

    blocks, nodes = parse_blocks_info(block_info)
    for n in nodes:
        print(f"  node: {n['name']} off={n['offset']} size={n['size']}")

    # ---- 对齐:BlockInfo 结束到数据区之间 ----
    data_pos = bi_start + comp_bi_size
    if archive_version >= 8:
        pad_x2 = (16 - (data_pos % 16)) % 16
        data_pos += pad_x2
    else:
        pad_x2 = 0

    # ---- 解压所有数据块 ----
    decompressed_blocks = []
    for i, b in enumerate(blocks):
        chunk = data[data_pos:data_pos + b["comp"]]
        data_pos += b["comp"]
        dec = decompress_raw(chunk, b["unc"], b["flags"])
        decompressed_blocks.append(dec)
        print(f"  块[{i}]: comp={b['comp']} -> unc={len(dec)}")

    # ---- 重建未压缩的 BlockInfo ----
    new_bi = bytearray()
    new_bi.extend(b'\x00' * 16)  # hash 占位
    new_bi.extend(struct.pack('>i', len(blocks)))
    for i, b in enumerate(blocks):
        unc_size = len(decompressed_blocks[i])
        new_bi.extend(struct.pack('>IIH', unc_size, unc_size, 0))  # flags=0 未压缩
    new_bi.extend(struct.pack('>i', len(nodes)))
    for n in nodes:
        new_bi.extend(struct.pack('>qq', n["offset"], n["size"]))
        new_bi.extend(struct.pack('>I', n["flags"]))
        new_bi.extend(n["name"].encode('utf-8'))
        new_bi.extend(b'\x00')

    new_unc_bi_size = len(new_bi)
    new_comp_bi = bytes(new_bi)  # 未压缩,comp==unc
    new_comp_bi_size = len(new_comp_bi)

    # ---- 组装输出文件 ----
    output = bytearray()

    # Header
    output.extend(b'UnityFS\x00')
    output.extend(struct.pack('>I', archive_version))
    output.extend(unity_version.encode('utf-8'))
    output.extend(b'\x00')
    output.extend(unity_revision.encode('utf-8'))
    output.extend(b'\x00')

    file_size_pos = len(output)
    output.extend(struct.pack('>Q', 0))  # 占位 file_size
    output.extend(struct.pack('>I', new_comp_bi_size))
    output.extend(struct.pack('>I', new_unc_bi_size))
    output.extend(struct.pack('>I', 0))  # flags=0 未压缩

    # 前对齐(与输入一致)
    if archive_version <= 6:
        pad1 = 0
    else:
        pad1 = (16 - (len(output) % 16)) % 16
    output.extend(b'\x00' * pad1)
    output.extend(new_comp_bi)

    # 直接写回数据块(理论上由解压后的block info的offset指定,重新压缩则由archive flags)
    for dec in decompressed_blocks:
        output.extend(dec)

    # 回填 file_size
    actual_size = len(output)
    output[file_size_pos:file_size_pos + 8] = struct.pack('>Q', actual_size)

    with open(output_path, "wb") as f:
        f.write(output)
    print(f"[OK] 输出: {output_path} ({actual_size} 字节)")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="UnityFS 解压为未压缩归档流")
    parser.add_argument("input_file")
    parser.add_argument("-o", "--output", default=None)
    args = parser.parse_args()

    out = args.output or (os.path.splitext(args.input_file)[0] + "_unpacked.bin")
    unpack_asset_bundle(args.input_file, out)
    print("完成")
测试样本

测试样本中有些就是真实样本,有些是基于真实样本修改的。侵删。
1.test1.zip (3.7 MB)
修改了ArchiveVersion的ab包,用于测试结构修复。
2.test2.zip (5.2 KB)
用于测试FakeHeader的功能、降低ArchiveVersion加载到AssetStudio v0.16.47的功能。
3.test1.zip (1.1 MB)
测试注入版本号,推荐2022.3.62f2

注:

开始设计只改了文件头的版本号,忘了数据块里还有版本号信息 :innocent:。


  1. 主要体验一下创建新话题的功能 :+1:,这个是捎带的 ↩︎