import os import hmac import json from base64 import b64decode, b64encode from hashlib import sha256 from functools import cached_property import shutil from sys import version_info from packaging import version import sys sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import config if version_info < (3, 8): cached_property = property else: from functools import cached_property def version_tuple(version: 'str | None') -> tuple: if version is None: return (0, 0, 0) return tuple(map(int, version.split('.'))) def std_path(path: str) -> str: return path.replace('\\', '/') def bytes_format(b: int) -> str: for i in ['B', 'KB', 'MB', 'GB', 'TB']: if b < 1024: return f'{b:.3f} {i}' b /= 1024 return f'{b:.3f} PB' class FileParser: KEY = b'\xd4\x1f\xdb\xe37\xd0\x01h\x0c*MC\xaf\xe5p\xc7\x1f\xde\x85\xd8\xf3\xd4\xc4o7\x99\xc1\x8f\x1fP\x82w\xac\xa7\xabc2\x83q\x0c+\xb4\x1a\x07\x8e\xfb\xe7\xc1\x9c\xf0\x87\xa7\xe17u*\xb7X\x1c\x8d\x9c\x0e=\xe9' def __init__(self): self.file_path: str = None self.rel_path: str = None self.offset: int = None self.length: int = None self.part_index: int = 0 self.data: bytes = None def _get_file_data(self) -> bytes: with open(self.file_path, 'rb') as f: return f.read() def to_dict(self) -> dict: return { 'path': self.rel_path, 'byteOffset': self.offset, 'length': self.length, 'sha256HashBase64Encoded': self.file_hash_base64 } @cached_property def file_hash(self) -> bytes: return sha256(self.data).digest() @cached_property def file_hash_base64(self) -> str: return b64encode(self.file_hash).decode() @cached_property def detail_hash(self) -> bytes: return hmac.new(self.KEY, self.data, sha256).digest() @cached_property def detail_hash_base64(self) -> str: return b64encode(self.detail_hash).decode() @classmethod def from_bundle(cls, bundle_handler, file_abspath: str, offset: int, length: int, file_hash: bytes, part_index: int = 0): c = cls() c.file_path = file_abspath c.offset = offset c.length = length c.part_index = part_index bundle_handler.seek(offset) c.data = bundle_handler.read(length) if c.file_hash != file_hash: raise ValueError(f'File hash mismatch for `{file_abspath}`') return c def to_file(self) -> None: paths = os.path.split(self.file_path) if not os.path.isdir(paths[0]): os.makedirs(paths[0]) with open(self.file_path, 'wb') as f: f.write(self.data) @classmethod def from_file(cls, file_abspath: str, file_relpath: str, offset: int = 0): c = cls() c.file_path = file_abspath c.rel_path = file_relpath c.data = c._get_file_data() c.length = len(c.data) c.offset = offset return c class Debundler: def __init__(self, file_path, metadata_path, output_dir): self.file_path = file_path self.metadata_path = metadata_path self.output_dir = output_dir self.file_handlers = {} self._init_bundle_file_handlers() def __del__(self): for handler in self.file_handlers.values(): if handler is not None: handler.close() def _init_bundle_file_handlers(self): if not os.path.isfile(self.file_path): raise FileNotFoundError( f'Bundle file `{self.file_path}` not found') self.file_handlers[0] = open(self.file_path, 'rb') import re base, ext = os.path.splitext(self.file_path) root = re.sub(r'_\d+$', '', base) part_index = 1 while True: part_path = f'{root}_{part_index}{ext}' if os.path.isfile(part_path): self.file_handlers[part_index] = open(part_path, 'rb') part_index += 1 else: break @property def metadata(self) -> dict: if not os.path.isfile(self.metadata_path): raise FileNotFoundError( f'Metadata file `{self.metadata_path}` not found') try: with open(self.metadata_path, 'rb') as f: return json.load(f) except json.JSONDecodeError as e: raise ValueError( f'Metadata file `{self.metadata_path}` is not a valid JSON file') from e def parse(self): added: 'list[dict]' = self.metadata.get('added', None) if added is None: raise ValueError( f'Metadata file `{self.metadata_path}` does not contain `added` field') if not os.path.isdir(self.output_dir): print(f'Creating output directory `{self.output_dir}`') os.makedirs(self.output_dir) for file in added: part_index = file.get('partIndex', 0) handler = self.file_handlers.get(part_index) if handler is None: raise FileNotFoundError(f'Bundle part {part_index} not found for `{file["path"]}`') FileParser.from_bundle( handler, os.path.join(self.output_dir, file['path']), file['byteOffset'], file['length'], b64decode(file['sha256HashBase64Encoded']), part_index=part_index ).to_file() print(f'Debundling completed: {len(added)} files written') def Run(): if os.path.exists(config.scripts["ArcaseDebundler"]["output_path"]): shutil.rmtree(config.scripts["ArcaseDebundler"]["output_path"]) version_files = [] import re bundle_path = config.scripts["ArcaseDebundler"]["bundle_path"] seen_versions = set() for f in os.listdir(bundle_path): if not (os.path.isfile(os.path.join(bundle_path, f)) and f.lower().endswith('.cb')): continue base = f[:-3] m = re.match(r'^(.+?)_(\d+)$', base) if m: if int(m.group(2)) != 0: continue version_str = m.group(1) else: version_str = base if version_str in seen_versions: continue seen_versions.add(version_str) json_path = os.path.join(bundle_path, f"{version_str}.json") if os.path.exists(json_path): version_files.append({ 'version_str': version_str, 'version_obj': version.Version(version_str), 'cb_file': f }) version_files.sort(key=lambda x: x['version_obj']) for item in version_files: cb_file = item['cb_file'] version_str = item['version_str'] Debundler( file_path=os.path.join(bundle_path, cb_file), metadata_path=os.path.join(bundle_path, f"{version_str}.json"), output_dir=config.scripts["ArcaseDebundler"]["output_path"] ).parse() if __name__ == '__main__': Run()