From 4791e28bc9b62350d6557377989776e79ca7881b Mon Sep 17 00:00:00 2001 From: Sthornberry9 <46094434+Sthornberry9@users.noreply.github.com> Date: Tue, 11 Aug 2026 07:24:53 -0400 Subject: [PATCH 1/3] feat: add fragmented STFS parsing and verification --- backup_manager.py | 34 +- tests.py | 56 ++- unityscraper/domains/packages/__init__.py | 18 +- unityscraper/domains/packages/commands.py | 27 +- unityscraper/domains/packages/errors.py | 18 + unityscraper/domains/packages/executables.py | 85 ++++ unityscraper/domains/packages/inspectors.py | 9 +- unityscraper/domains/packages/models.py | 113 ++++- unityscraper/domains/packages/service.py | 11 +- unityscraper/domains/packages/stfs.py | 484 +++++++++++++++++++ 10 files changed, 830 insertions(+), 25 deletions(-) create mode 100644 unityscraper/domains/packages/errors.py create mode 100644 unityscraper/domains/packages/executables.py create mode 100644 unityscraper/domains/packages/stfs.py diff --git a/backup_manager.py b/backup_manager.py index 46ef53e..000abf9 100644 --- a/backup_manager.py +++ b/backup_manager.py @@ -17,6 +17,20 @@ from pathlib import Path, PurePosixPath from typing import Any, Callable, Iterable, Iterator, Optional +from unityscraper.domains.packages.errors import ( + InvalidPackageError, + PackageError as BackupError, + UnsafeArchiveError, +) +from unityscraper.domains.packages.executables import ( + inspect_xbe as _domain_inspect_xbe, + inspect_xex as _domain_inspect_xex, +) +from unityscraper.domains.packages.stfs import ( + extract_stfs_files as _domain_extract_stfs_files, + inspect_stfs as _domain_inspect_stfs, + list_stfs_entries as _domain_list_stfs_entries, +) STFS_MAGICS = {b"CON ", b"LIVE", b"PIRS"} CONTENT_TYPES = { @@ -38,18 +52,6 @@ FATX_INVALID_RE = re.compile(r'[<>:"/\\|?*]') -class BackupError(RuntimeError): - """Base error for backup operations.""" - - -class InvalidPackageError(BackupError): - """Raised when a file is not a supported Xbox package.""" - - -class UnsafeArchiveError(BackupError): - """Raised when an archive attempts to escape its extraction directory.""" - - class ConflictError(BackupError): """Raised when a destination conflict cannot be resolved automatically.""" @@ -496,6 +498,14 @@ def format_version(value: int) -> str: ) +# Compatibility exports. Package-format ownership lives in the packages domain. +inspect_stfs = _domain_inspect_stfs +list_stfs_entries = _domain_list_stfs_entries +extract_stfs_files = _domain_extract_stfs_files +inspect_xbe = _domain_inspect_xbe +inspect_xex = _domain_inspect_xex + + def sha256_file(path: str | Path) -> str: digest = hashlib.sha256() with Path(path).open("rb") as handle: diff --git a/tests.py b/tests.py index 0967200..bdbebce 100644 --- a/tests.py +++ b/tests.py @@ -90,6 +90,7 @@ from unityscraper.domains.library.models import GameSummary as ModularGameSummary from unityscraper.domains.library.service import LibraryService as ModularLibraryService from unityscraper.domains.packages.commands import InspectStfsPackage, InventoryStfsFileTable +from unityscraper.domains.packages.stfs import verify_stfs from unityscraper.domains.tools.catalog import ToolCatalog as ModularToolCatalog from unityscraper.domains.tools.models import ToolDefinition as ModularToolDefinition from unityscraper.domains.tools.runner import ExternalToolRunner as ModularToolRunner @@ -1603,7 +1604,8 @@ def test_stfs_file_table_is_inventoried_read_only(self): payload[0x360:0x364] = bytes.fromhex("53510804") payload[0x379] = 0x24 payload[0x37B] = 1 - payload[0x37C:0x37E] = (1).to_bytes(2, "big") + payload[0x37C:0x37E] = (1).to_bytes(2, "little") + payload[0x395:0x399] = (2).to_bytes(4, "big") name = b"savegame.dat" entry = 0xB000 payload[entry:entry + len(name)] = name @@ -1620,7 +1622,6 @@ def test_stfs_file_table_is_inventoried_read_only(self): self.assertEqual(entries[0].size, 123) self.assertTrue(entries[0].consecutive) - payload[0x379 + 0x1C:0x379 + 0x20] = (2).to_bytes(4, "big") payload[0xC000:0xC004] = b"data" package_path.write_bytes(payload) destination = self.temp_dir / "extracted" @@ -1629,6 +1630,57 @@ def test_stfs_file_table_is_inventoried_read_only(self): self.assertEqual(result["extracted"][0]["size"], 123) self.assertTrue(Path(result["manifest"]).is_file()) + def test_fragmented_stfs_extraction_and_integrity_verification(self): + payload = bytearray(0xF000) + payload[:4] = b"LIVE" + payload[0x340:0x344] = (0xA000).to_bytes(4, "big") + payload[0x344:0x348] = (1).to_bytes(4, "big") + payload[0x360:0x364] = bytes.fromhex("53510804") + payload[0x379] = 0x24 + payload[0x37B] = 1 + payload[0x37C:0x37E] = (1).to_bytes(2, "little") + payload[0x395:0x399] = (4).to_bytes(4, "big") + + name = b"fragmented.bin" + entry = 0xB000 + payload[entry:entry + len(name)] = name + payload[entry + 0x28] = len(name) + payload[entry + 0x29:entry + 0x2C] = (2).to_bytes(3, "little") + payload[entry + 0x2F:entry + 0x32] = (1).to_bytes(3, "little") + payload[entry + 0x32:entry + 0x34] = (0xFFFF).to_bytes(2, "big") + payload[entry + 0x34:entry + 0x38] = (0x1004).to_bytes(4, "big") + payload[0xC000:0xD000] = b"A" * 0x1000 + payload[0xE000:0xE004] = b"tail" + + for block, offset in enumerate((0xB000, 0xC000, 0xD000, 0xE000)): + record = 0xA000 + block * 0x18 + payload[record:record + 0x14] = hashlib.sha1( + payload[offset:offset + 0x1000] + ).digest() + next_block = 3 if block == 1 else 0xFFFFFF + payload[record + 0x14:record + 0x18] = ( + (2 << 30) | next_block + ).to_bytes(4, "big") + + package_path = self.temp_dir / "fragmented.stfs" + package_path.write_bytes(payload) + entries = list_stfs_entries(package_path) + self.assertEqual(entries[0].blocks, (1, 3)) + + destination = self.temp_dir / "fragmented-output" + extract_stfs_files(package_path, destination) + self.assertEqual( + (destination / "fragmented.bin").read_bytes(), + b"A" * 0x1000 + b"tail", + ) + self.assertTrue(verify_stfs(package_path).valid) + + payload[0xC000] ^= 0xFF + package_path.write_bytes(payload) + report = verify_stfs(package_path) + self.assertFalse(report.valid) + self.assertEqual(report.mismatched_blocks, 1) + def test_rejects_unknown_stfs_content_type(self): with self.assertRaises(InvalidPackageError): inspect_stfs(self._stfs(content_type=0xDEADBEEF)) diff --git a/unityscraper/domains/packages/__init__.py b/unityscraper/domains/packages/__init__.py index d7822bb..334fea0 100644 --- a/unityscraper/domains/packages/__init__.py +++ b/unityscraper/domains/packages/__init__.py @@ -2,26 +2,40 @@ from __future__ import annotations -from .commands import InspectStfsPackage, InventoryStfsFileTable +from .commands import InspectStfsPackage, InventoryStfsFileTable, VerifyStfsPackage from .inspectors import ( extract_stfs_files, inspect_stfs, inspect_xbe, inspect_xex, list_stfs_entries, + verify_stfs, +) +from .models import ( + StfsBlockVerification, + StfsEntry, + StfsHashRecord, + StfsIntegrityReport, + StfsPackage, + XbePackage, + XexPackage, ) -from .models import StfsEntry, StfsPackage, XbePackage, XexPackage __all__ = [ "StfsEntry", + "StfsBlockVerification", + "StfsHashRecord", + "StfsIntegrityReport", "StfsPackage", "XbePackage", "XexPackage", "InspectStfsPackage", "InventoryStfsFileTable", + "VerifyStfsPackage", "extract_stfs_files", "inspect_stfs", "inspect_xbe", "inspect_xex", "list_stfs_entries", + "verify_stfs", ] diff --git a/unityscraper/domains/packages/commands.py b/unityscraper/domains/packages/commands.py index 6528414..1aa5a7b 100644 --- a/unityscraper/domains/packages/commands.py +++ b/unityscraper/domains/packages/commands.py @@ -7,7 +7,7 @@ from unityscraper.core.jobs import JobResult -from .inspectors import inspect_stfs, list_stfs_entries +from .inspectors import inspect_stfs, list_stfs_entries, verify_stfs class InspectStfsPackage: @@ -51,4 +51,27 @@ def run(self, source: str | Path, *, max_entries: int = 100_000) -> JobResult: ) -__all__ = ["InspectStfsPackage", "InventoryStfsFileTable"] +class VerifyStfsPackage: + """Check every allocated STFS data block against its stored SHA-1.""" + + def run(self, source: str | Path, *, max_issues: int = 10_000) -> JobResult: + path = Path(source) + try: + report = verify_stfs(path, max_issues=max_issues) + except Exception as exc: + return JobResult.failed( + "STFS verification failed", + source=str(path), + error=str(exc), + ) + details = asdict(report) + details["source"] = str(report.source) + details["issues"] = [asdict(issue) for issue in report.issues] + return JobResult.completed( + "STFS verification completed", + valid=report.valid, + **details, + ) + + +__all__ = ["InspectStfsPackage", "InventoryStfsFileTable", "VerifyStfsPackage"] diff --git a/unityscraper/domains/packages/errors.py b/unityscraper/domains/packages/errors.py new file mode 100644 index 0000000..92410af --- /dev/null +++ b/unityscraper/domains/packages/errors.py @@ -0,0 +1,18 @@ +"""Package-domain exceptions shared by inspection and editing workflows.""" + +from __future__ import annotations + + +class PackageError(RuntimeError): + """Base error for malformed or unsupported Xbox package operations.""" + + +class InvalidPackageError(PackageError): + """Raised when package metadata or block allocation is invalid.""" + + +class UnsafeArchiveError(PackageError): + """Raised when a package path could escape its selected destination.""" + + +__all__ = ["InvalidPackageError", "PackageError", "UnsafeArchiveError"] diff --git a/unityscraper/domains/packages/executables.py b/unityscraper/domains/packages/executables.py new file mode 100644 index 0000000..6d77959 --- /dev/null +++ b/unityscraper/domains/packages/executables.py @@ -0,0 +1,85 @@ +"""Read-only XBE and XEX metadata inspection.""" + +from __future__ import annotations + +from pathlib import Path + +from .errors import InvalidPackageError +from .models import XbePackage, XexPackage + + +def inspect_xbe(path: str | Path) -> XbePackage: + package_path = Path(path) + with package_path.open("rb") as handle: + header = handle.read(0x11C) + if len(header) < 0x11C or header[:4] != b"XBEH": + raise InvalidPackageError(f"{package_path.name} is not an XBE executable") + base_address = int.from_bytes(header[0x104:0x108], "little") + certificate_address = int.from_bytes(header[0x118:0x11C], "little") + certificate_offset = certificate_address - base_address + if certificate_offset < 0: + raise InvalidPackageError("XBE certificate address is invalid") + handle.seek(certificate_offset) + certificate = handle.read(0xD0) + if len(certificate) < 0xD0: + raise InvalidPackageError("XBE certificate is incomplete") + title_id = f"{int.from_bytes(certificate[0x8:0xC], 'little'):08X}" + title_name = certificate[0xC:0x5C].decode("utf-16-le", errors="ignore") + title_name = title_name.split("\x00", 1)[0].strip() + return XbePackage( + package_path, + title_id, + title_name, + package_path.stat().st_size, + int.from_bytes(certificate[0x9C:0xA0], "little"), + int.from_bytes(certificate[0xA0:0xA4], "little"), + int.from_bytes(certificate[0xA8:0xAC], "little"), + int.from_bytes(certificate[0xAC:0xB0], "little"), + ) + + +def inspect_xex(path: str | Path) -> XexPackage: + package_path = Path(path) + with package_path.open("rb") as handle: + header = handle.read(0x4000) + if len(header) < 0x18 or header[:4] != b"XEX2": + raise InvalidPackageError(f"{package_path.name} is not an XEX2 executable") + + module_flags = int.from_bytes(header[4:8], "big") + optional_count = int.from_bytes(header[0x14:0x18], "big") + if optional_count > 4096 or 0x18 + optional_count * 8 > len(header): + raise InvalidPackageError("XEX optional-header table is invalid") + + execution_offset = 0 + for index in range(optional_count): + entry = 0x18 + index * 8 + key = int.from_bytes(header[entry:entry + 4], "big") + value = int.from_bytes(header[entry + 4:entry + 8], "big") + if key == 0x00040006: + execution_offset = value + break + if not execution_offset or execution_offset + 24 > len(header): + raise InvalidPackageError("XEX execution metadata is unavailable") + + info = header[execution_offset:execution_offset + 24] + + def format_version(value: int) -> str: + return ( + f"{(value >> 28) & 0xF}.{(value >> 24) & 0xF}." + f"{(value >> 8) & 0xFFFF}.{value & 0xFF}" + ) + + return XexPackage( + path=package_path, + title_id=info[12:16].hex().upper(), + media_id=info[0:4].hex().upper(), + version=format_version(int.from_bytes(info[4:8], "big")), + base_version=format_version(int.from_bytes(info[8:12], "big")), + disc_number=info[18], + disc_count=info[19], + module_flags=module_flags, + size=package_path.stat().st_size, + ) + + +__all__ = ["inspect_xbe", "inspect_xex"] diff --git a/unityscraper/domains/packages/inspectors.py b/unityscraper/domains/packages/inspectors.py index 995a65b..d0c08ef 100644 --- a/unityscraper/domains/packages/inspectors.py +++ b/unityscraper/domains/packages/inspectors.py @@ -2,12 +2,13 @@ from __future__ import annotations -from backup_manager import ( +from .executables import inspect_xbe, inspect_xex +from .stfs import ( extract_stfs_files, inspect_stfs, - inspect_xbe, - inspect_xex, list_stfs_entries, + read_stfs_layout, + verify_stfs, ) __all__ = [ @@ -16,4 +17,6 @@ "inspect_xbe", "inspect_xex", "list_stfs_entries", + "read_stfs_layout", + "verify_stfs", ] diff --git a/unityscraper/domains/packages/models.py b/unityscraper/domains/packages/models.py index 8ac17c0..fa8a779 100644 --- a/unityscraper/domains/packages/models.py +++ b/unityscraper/domains/packages/models.py @@ -1,7 +1,114 @@ -"""Package inspection data models.""" +"""Data contracts for Xbox package inspection and integrity workflows.""" from __future__ import annotations -from backup_manager import StfsEntry, StfsPackage, XbePackage, XexPackage +from dataclasses import dataclass +from pathlib import Path -__all__ = ["StfsEntry", "StfsPackage", "XbePackage", "XexPackage"] + +@dataclass(frozen=True) +class StfsPackage: + path: Path + magic: str + content_type: int + content_label: str + content_directory: str + title_id: str + media_id: str + disc_number: int + disc_count: int + display_name: str + title_name: str + size: int + save_game_id: str + console_id: str + profile_id: str + device_id: str + header_size: int = 0 + block_count: int = 0 + structure_type: int = 0 + + +@dataclass(frozen=True) +class StfsEntry: + index: int + path: str + name: str + is_directory: bool + consecutive: bool + allocated_blocks: int + starting_block: int + parent_index: int + size: int + blocks: tuple[int, ...] = () + + +@dataclass(frozen=True) +class StfsHashRecord: + block: int + level: int + stored_sha1: str + status: int + next_block: int + table_index: int + offset: int + + +@dataclass(frozen=True) +class StfsBlockVerification: + block: int + status: str + stored_sha1: str = "" + calculated_sha1: str = "" + message: str = "" + + +@dataclass(frozen=True) +class StfsIntegrityReport: + source: Path + block_count: int + checked: int + valid_blocks: int + mismatched_blocks: int + unverifiable_blocks: int + issues: tuple[StfsBlockVerification, ...] + + @property + def valid(self) -> bool: + return self.mismatched_blocks == 0 + + +@dataclass(frozen=True) +class XbePackage: + path: Path + title_id: str + title_name: str + size: int + allowed_media: int + region_flags: int + disc_number: int + version: int + + +@dataclass(frozen=True) +class XexPackage: + path: Path + title_id: str + media_id: str + version: str + base_version: str + disc_number: int + disc_count: int + module_flags: int + size: int + + +__all__ = [ + "StfsBlockVerification", + "StfsEntry", + "StfsHashRecord", + "StfsIntegrityReport", + "StfsPackage", + "XbePackage", + "XexPackage", +] diff --git a/unityscraper/domains/packages/service.py b/unityscraper/domains/packages/service.py index a7ad570..fea9a0f 100644 --- a/unityscraper/domains/packages/service.py +++ b/unityscraper/domains/packages/service.py @@ -2,16 +2,20 @@ from __future__ import annotations -from .commands import InspectStfsPackage, InventoryStfsFileTable +from .commands import InspectStfsPackage, InventoryStfsFileTable, VerifyStfsPackage from .inspectors import ( extract_stfs_files, inspect_stfs, inspect_xbe, inspect_xex, list_stfs_entries, + verify_stfs, ) from .models import ( + StfsBlockVerification, StfsEntry, + StfsHashRecord, + StfsIntegrityReport, StfsPackage, XbePackage, XexPackage, @@ -19,14 +23,19 @@ __all__ = [ "StfsEntry", + "StfsBlockVerification", + "StfsHashRecord", + "StfsIntegrityReport", "StfsPackage", "XbePackage", "XexPackage", "InspectStfsPackage", "InventoryStfsFileTable", + "VerifyStfsPackage", "extract_stfs_files", "inspect_stfs", "inspect_xbe", "inspect_xex", "list_stfs_entries", + "verify_stfs", ] diff --git a/unityscraper/domains/packages/stfs.py b/unityscraper/domains/packages/stfs.py new file mode 100644 index 0000000..64c4013 --- /dev/null +++ b/unityscraper/domains/packages/stfs.py @@ -0,0 +1,484 @@ +"""Bounded STFS parsing, fragmented block traversal, and integrity checks. + +The block geometry follows Dalavin's GPLv3 X360 library while using a new, +cross-platform Python implementation with explicit bounds at every file read. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Any, BinaryIO, Iterable + +from .errors import InvalidPackageError, UnsafeArchiveError +from .models import ( + StfsBlockVerification, + StfsEntry, + StfsHashRecord, + StfsIntegrityReport, + StfsPackage, +) + +STFS_MAGICS = {b"CON ", b"LIVE", b"PIRS"} +CONTENT_TYPES = { + 0x00000001: ("Saved Game", "00000001"), + 0x00000002: ("DLC", "00000002"), + 0x00005000: ("Original Xbox Game", "00005000"), + 0x00007000: ("Xbox 360 Game", "00007000"), + 0x000B0000: ("Title Update", "000B0000"), + 0x000D0000: ("Xbox Live Arcade", "000D0000"), + 0x00010000: ("Profile", "00010000"), +} +STFS_END = 0xFFFFFF +BLOCK_SIZE = 0x1000 +LEVEL0_BLOCKS = 0xAA +LEVEL1_BLOCKS = 0x70E4 +MAX_BLOCKS = 0x4AF768 + + +def _read_utf16be(data: bytes) -> str: + return data.decode("utf-16-be", errors="ignore").split("\x00", 1)[0].strip() + + +def _read_exact(handle: BinaryIO, offset: int, size: int, package_size: int) -> bytes: + if offset < 0 or size < 0 or offset + size > package_size: + raise InvalidPackageError("STFS structure points outside the package") + handle.seek(offset) + value = handle.read(size) + if len(value) != size: + raise InvalidPackageError("STFS package is truncated") + return value + + +@dataclass(frozen=True) +class StfsLayout: + magic: bytes + header_size: int + block_separation: int + table_blocks: int + table_start: int + block_count: int + package_size: int + shift: int + top_table_index: int + + @property + def base_offset(self) -> int: + return (self.header_size + 0xFFF) & 0xFFFFF000 + + @property + def structure_type(self) -> int: + return self.shift + + @property + def spaces(self) -> tuple[int, int]: + return (0xAB, 0x718F) if self.shift == 0 else (0xAC, 0x723A) + + def data_block(self, block: int) -> int: + self._validate_block(block) + result = (((block // LEVEL0_BLOCKS) + 1) << self.shift) + block + if block >= LEVEL0_BLOCKS: + result += ((block // LEVEL1_BLOCKS) + 1) << self.shift + if block >= LEVEL1_BLOCKS: + result += 1 << self.shift + return result + + def data_offset(self, block: int) -> int: + return self.base_offset + self.data_block(block) * BLOCK_SIZE + + def base_hash_block(self, block: int, level: int) -> int: + self._validate_block(block) + space0, space1 = self.spaces + if level == 0: + result = (block // LEVEL0_BLOCKS) * space0 + if block >= LEVEL0_BLOCKS: + result += ((block // LEVEL1_BLOCKS) + 1) << self.shift + if block >= LEVEL1_BLOCKS: + result += 1 << self.shift + return result + if level == 1: + if block < LEVEL1_BLOCKS: + return space0 + return space1 * (block // LEVEL1_BLOCKS) + (1 << self.shift) + if level == 2: + return space1 + raise InvalidPackageError("Unsupported STFS hash-tree level") + + def base_hash_offset(self, block: int, level: int) -> int: + entry = ( + block % LEVEL0_BLOCKS + if level == 0 + else (block // LEVEL0_BLOCKS) % LEVEL0_BLOCKS + if level == 1 + else (block // LEVEL1_BLOCKS) % LEVEL0_BLOCKS + ) + return self.base_offset + self.base_hash_block(block, level) * BLOCK_SIZE + entry * 0x18 + + def hash_record(self, handle: BinaryIO, block: int, level: int = 0) -> StfsHashRecord: + table_index = self._active_table_index(handle, block, level) + offset = self.base_hash_offset(block, level) + table_index * BLOCK_SIZE + raw = _read_exact(handle, offset, 0x18, self.package_size) + flags = int.from_bytes(raw[0x14:0x18], "big") + return StfsHashRecord( + block=block, + level=level, + stored_sha1=raw[:0x14].hex(), + status=(flags >> 30) & 0x3, + next_block=flags & 0xFFFFFF, + table_index=(flags >> 30) & 0x1, + offset=offset, + ) + + def block_chain( + self, + handle: BinaryIO, + start: int, + count: int, + *, + consecutive: bool = False, + ) -> tuple[int, ...]: + if count < 0 or count > self.block_count: + raise InvalidPackageError("STFS block-chain length is invalid") + if count == 0: + return () + self._validate_allocated_block(start) + if consecutive: + end = start + count + if end > self.block_count: + raise InvalidPackageError("STFS consecutive allocation exceeds the package") + return tuple(range(start, end)) + + blocks: list[int] = [] + visited: set[int] = set() + current = start + for index in range(count): + self._validate_allocated_block(current) + if current in visited: + raise InvalidPackageError("STFS block chain contains a loop") + visited.add(current) + blocks.append(current) + if index == count - 1: + break + record = self.hash_record(handle, current, 0) + if record.next_block == STFS_END: + raise InvalidPackageError("STFS block chain ends before its declared length") + current = record.next_block + return tuple(blocks) + + def _active_table_index(self, handle: BinaryIO, block: int, level: int) -> int: + if self.shift == 0: + return 0 + if level == 2: + return self.top_table_index + if level == 1: + if self.block_count > LEVEL1_BLOCKS: + return self.hash_record(handle, block, 2).table_index + return self.top_table_index + if level == 0: + if self.block_count > LEVEL0_BLOCKS: + return self.hash_record(handle, block, 1).table_index + return self.top_table_index + raise InvalidPackageError("Unsupported STFS hash-tree level") + + def _validate_block(self, block: int) -> None: + if block < 0 or block >= MAX_BLOCKS: + raise InvalidPackageError("STFS block number is outside the supported range") + + def _validate_allocated_block(self, block: int) -> None: + self._validate_block(block) + if block >= self.block_count: + raise InvalidPackageError("STFS block points beyond the allocated block count") + + +def read_stfs_layout(path: str | Path) -> StfsLayout: + package = Path(path) + package_size = package.stat().st_size + with package.open("rb") as handle: + header = _read_exact(handle, 0, 0x3AD, package_size) + if header[:4] not in STFS_MAGICS: + raise InvalidPackageError("Not a supported STFS package") + if int.from_bytes(header[0x3A9:0x3AD], "big"): + raise InvalidPackageError("SVOD packages do not contain an STFS file table") + if header[0x379] != 0x24 or header[0x37A] != 0: + raise InvalidPackageError("STFS volume descriptor is invalid") + + header_size = int.from_bytes(header[0x340:0x344], "big") + separation = header[0x37B] & 0x3 + block_count = int.from_bytes(header[0x395:0x399], "big") + if block_count <= 0 or block_count >= MAX_BLOCKS: + raise InvalidPackageError("STFS allocated block count is invalid") + aligned = (header_size + 0xFFF) & 0xFFFFF000 + shift = 0 if aligned == 0xB000 else 0 if separation & 1 else 1 + top_index = (separation >> 1) & 1 + table_start = int.from_bytes(header[0x37E:0x381], "little") + count_raw = header[0x37C:0x37E] + candidates = tuple(dict.fromkeys((int.from_bytes(count_raw, "little"), int.from_bytes(count_raw, "big")))) + table_blocks = 0 + for candidate in candidates: + if not 0 < candidate <= 0x3FF: + continue + layout = StfsLayout( + header[:4], header_size, separation, candidate, table_start, + block_count, package_size, shift, top_index, + ) + try: + if layout.data_offset(table_start) + BLOCK_SIZE <= package_size: + table_blocks = candidate + break + except InvalidPackageError: + continue + if not table_blocks: + raise InvalidPackageError("STFS file-table size is invalid") + return StfsLayout( + header[:4], header_size, separation, table_blocks, table_start, + block_count, package_size, shift, top_index, + ) + + +def inspect_stfs(path: str | Path) -> StfsPackage: + package_path = Path(path) + with package_path.open("rb") as handle: + header = handle.read(0x1791) + if len(header) < 0x1791 or header[:4] not in STFS_MAGICS: + raise InvalidPackageError(f"{package_path.name} is not a supported STFS package") + content_type = int.from_bytes(header[0x344:0x348], "big") + content = CONTENT_TYPES.get(content_type) + if content is None: + raise InvalidPackageError(f"Unsupported STFS content type 0x{content_type:08X}") + title_id = header[0x360:0x364].hex().upper() + if title_id == "00000000" or len(title_id) != 8: + raise InvalidPackageError("STFS package does not contain a usable TitleID") + header_size = int.from_bytes(header[0x340:0x344], "big") + block_count = int.from_bytes(header[0x395:0x399], "big") + aligned = (header_size + 0xFFF) & 0xFFFFF000 + separation = header[0x37B] & 0x3 + structure_type = 0 if aligned == 0xB000 else 0 if separation & 1 else 1 + return StfsPackage( + path=package_path, + magic=header[:4].decode("ascii").strip(), + content_type=content_type, + content_label=content[0], + content_directory=content[1], + title_id=title_id, + media_id=header[0x354:0x358].hex().upper(), + disc_number=header[0x366], + disc_count=header[0x367], + display_name=_read_utf16be(header[0x411:0x511]), + title_name=_read_utf16be(header[0x1691:0x1791]), + size=package_path.stat().st_size, + save_game_id=header[0x368:0x36C].hex().upper(), + console_id=header[0x36C:0x371].hex().upper(), + profile_id=header[0x371:0x379].hex().upper(), + device_id=header[0x3FD:0x411].hex().upper(), + header_size=header_size, + block_count=block_count, + structure_type=structure_type, + ) + + +def list_stfs_entries(path: str | Path, max_entries: int = 100_000) -> list[StfsEntry]: + package = Path(path) + layout = read_stfs_layout(package) + with package.open("rb") as handle: + table_chain = layout.block_chain( + handle, layout.table_start, layout.table_blocks, consecutive=False + ) + raw_entries: list[dict[str, Any]] = [] + for logical_block in table_chain: + table = _read_exact(handle, layout.data_offset(logical_block), BLOCK_SIZE, layout.package_size) + for entry_offset in range(0, BLOCK_SIZE, 0x40): + data = table[entry_offset:entry_offset + 0x40] + if not any(data): + continue + flags = data[0x28] + name_length = flags & 0x3F + if name_length == 0 or name_length > 0x28: + continue + raw_entries.append({ + "name": data[:name_length].decode("utf-8", errors="replace"), + "directory": bool(flags & 0x80), + "consecutive": bool(flags & 0x40), + "blocks": int.from_bytes(data[0x29:0x2C], "little"), + "start": int.from_bytes(data[0x2F:0x32], "little"), + "parent": int.from_bytes(data[0x32:0x34], "big"), + "size": int.from_bytes(data[0x34:0x38], "big"), + }) + if len(raw_entries) > max_entries: + raise InvalidPackageError("STFS file table exceeds the safety limit") + + entries: list[StfsEntry] = [] + for index, row in enumerate(raw_entries): + ancestors: list[str] = [] + parent = row["parent"] + visited: set[int] = set() + while parent != 0xFFFF: + if parent >= len(raw_entries) or parent in visited: + ancestors = ["[invalid-parent]"] + break + visited.add(parent) + ancestors.append(raw_entries[parent]["name"]) + parent = raw_entries[parent]["parent"] + full_path = "/".join(reversed(ancestors)) + full_path = f"{full_path}/{row['name']}" if full_path else row["name"] + block_count = row["blocks"] if not row["directory"] else 0 + blocks = layout.block_chain( + handle, + row["start"], + block_count, + consecutive=row["consecutive"], + ) if block_count else () + entries.append(StfsEntry( + index=index, + path=full_path, + name=row["name"], + is_directory=row["directory"], + consecutive=row["consecutive"], + allocated_blocks=row["blocks"], + starting_block=row["start"], + parent_index=row["parent"], + size=row["size"], + blocks=blocks, + )) + return entries + + +def verify_stfs(path: str | Path, *, max_issues: int = 10_000) -> StfsIntegrityReport: + package = Path(path).expanduser().resolve() + layout = read_stfs_layout(package) + checked = valid = mismatched = unverifiable = 0 + issues: list[StfsBlockVerification] = [] + with package.open("rb") as handle: + for block in range(layout.block_count): + checked += 1 + try: + record = layout.hash_record(handle, block) + data = _read_exact(handle, layout.data_offset(block), BLOCK_SIZE, layout.package_size) + except InvalidPackageError as exc: + mismatched += 1 + if len(issues) < max_issues: + issues.append(StfsBlockVerification(block, "invalid", message=str(exc))) + continue + calculated = hashlib.sha1(data).hexdigest() + if not record.stored_sha1 or set(record.stored_sha1) == {"0"}: + unverifiable += 1 + if len(issues) < max_issues: + issues.append(StfsBlockVerification( + block, "missing", record.stored_sha1, calculated, + "Hash record is empty", + )) + elif record.stored_sha1.lower() != calculated: + mismatched += 1 + if len(issues) < max_issues: + issues.append(StfsBlockVerification( + block, "mismatch", record.stored_sha1, calculated, + "Stored SHA-1 does not match the data block", + )) + else: + valid += 1 + return StfsIntegrityReport( + package, layout.block_count, checked, valid, mismatched, unverifiable, tuple(issues) + ) + + +def extract_stfs_files( + path: str | Path, + destination: str | Path, + selected_paths: Iterable[str] | None = None, + *, + max_output_size: int = 32 * 1024 * 1024 * 1024, +) -> dict[str, Any]: + package = Path(path).expanduser().resolve() + target = Path(destination).expanduser().resolve() + if package == target or target.is_relative_to(package): + raise InvalidPackageError("Extraction destination must be outside the package") + requested = {item.replace("\\", "/") for item in selected_paths or ()} + entries = list_stfs_entries(package) + files = [ + entry for entry in entries + if not entry.is_directory and (not requested or entry.path in requested) + ] + if requested - {entry.path for entry in files}: + missing = sorted(requested - {entry.path for entry in files}) + raise InvalidPackageError(f"STFS entries were not found: {', '.join(missing[:5])}") + if sum(entry.size for entry in files) > max_output_size: + raise InvalidPackageError("Selected STFS output exceeds the safety limit") + + layout = read_stfs_layout(package) + extracted: list[dict[str, Any]] = [] + skipped: list[dict[str, str]] = [] + target.mkdir(parents=True, exist_ok=True) + with package.open("rb") as handle: + for entry in files: + relative = _safe_member(entry.path) + output = (target / relative).resolve() + if not output.is_relative_to(target): + raise UnsafeArchiveError(f"STFS path escapes destination: {entry.path}") + if output.exists(): + skipped.append({"path": entry.path, "reason": "destination exists"}) + continue + output.parent.mkdir(parents=True, exist_ok=True) + partial = output.with_name(output.name + ".partial") + digest = hashlib.sha256() + remaining = entry.size + try: + with partial.open("xb") as destination_handle: + for block in entry.blocks: + size = min(remaining, BLOCK_SIZE) + chunk = _read_exact(handle, layout.data_offset(block), size, layout.package_size) + destination_handle.write(chunk) + digest.update(chunk) + remaining -= size + if remaining: + raise InvalidPackageError(f"STFS data is truncated: {entry.path}") + partial.replace(output) + finally: + partial.unlink(missing_ok=True) + extracted.append({ + "path": entry.path, + "output": str(output), + "size": entry.size, + "sha256": digest.hexdigest(), + "blocks": list(entry.blocks), + }) + manifest = target / "unityscraper-stfs-extraction.json" + manifest.write_text(json.dumps({ + "schema": 2, + "source": str(package), + "source_sha256": _sha256_file(package), + "read_only": True, + "supports_fragmented_files": True, + "extracted": extracted, + "skipped": skipped, + }, indent=2), encoding="utf-8") + return {"manifest": str(manifest), "extracted": extracted, "skipped": skipped} + + +def _safe_member(value: str) -> Path: + pure = PurePosixPath(value.replace("\\", "/")) + if pure.is_absolute() or not pure.parts or any(part in {"", ".", ".."} for part in pure.parts): + raise UnsafeArchiveError(f"Unsafe package path: {value}") + if ":" in pure.parts[0]: + raise UnsafeArchiveError(f"Unsafe package path: {value}") + return Path(*pure.parts) + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +__all__ = [ + "CONTENT_TYPES", + "STFS_MAGICS", + "StfsLayout", + "extract_stfs_files", + "inspect_stfs", + "list_stfs_entries", + "read_stfs_layout", + "verify_stfs", +] From 2383ef46283a86a49c12f8a87af37e664aa63d41 Mon Sep 17 00:00:00 2001 From: Sthornberry9 <46094434+Sthornberry9@users.noreply.github.com> Date: Tue, 11 Aug 2026 07:43:18 -0400 Subject: [PATCH 2/3] feat: add unified Xbox package lab --- ARCHITECTURE.md | 5 +- PACKAGE_LAB.md | 47 ++ PROFILES_AND_SAVES.md | 15 +- THIRD_PARTY_NOTICES.md | 12 +- modern_gui.py | 11 + package_gui.py | 465 +++++++++++++++++++ tests.py | 177 +++++++ unityscraper/domains/packages/__init__.py | 37 ++ unityscraper/domains/packages/executables.py | 11 +- unityscraper/domains/packages/fatx.py | 312 +++++++++++++ unityscraper/domains/packages/gdf.py | 234 ++++++++++ unityscraper/domains/packages/models.py | 13 + unityscraper/domains/packages/mutations.py | 245 ++++++++++ unityscraper/domains/packages/service.py | 37 ++ unityscraper/domains/packages/stfs.py | 181 +++++--- unityscraper/domains/packages/svod.py | 196 ++++++++ unityscraper/domains/profiles/__init__.py | 12 + unityscraper/domains/profiles/gpd.py | 206 ++++++++ unityscraper/domains/profiles/service.py | 12 + 19 files changed, 2146 insertions(+), 82 deletions(-) create mode 100644 PACKAGE_LAB.md create mode 100644 package_gui.py create mode 100644 unityscraper/domains/packages/fatx.py create mode 100644 unityscraper/domains/packages/gdf.py create mode 100644 unityscraper/domains/packages/mutations.py create mode 100644 unityscraper/domains/packages/svod.py create mode 100644 unityscraper/domains/profiles/gpd.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 5d338d3..233669d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -41,6 +41,7 @@ filesystem, FTP, package, or database behavior directly. - `knowledge_gui.py` renders knowledge search, imports, sources, and conflicts. - `backup_gui.py` renders inventory, package, FTP, and converter workflows. - `profile_gui.py` renders privacy-aware profile/save inventory and snapshots. +- `package_gui.py` renders STFS, SVOD, GDF/XISO, FATX-image, and GPD workflows. - `collection_gui.py` renders collection analysis, matching, reports, and repair previews. - `community_gui.py` renders unified search and the cross-domain community @@ -82,7 +83,9 @@ results to Tk's main loop. - `main.py` contains the XboxUnity collector and shared configuration. - `resume.py` handles partial download state and verification. -- `backup_manager.py` parses public STFS/XBE/XEX fields and performs safe +- `unityscraper/domains/packages` owns bounded package/image parsing, + verification, extraction, and transactional mutation. +- `backup_manager.py` retains compatibility exports and performs safe filesystem or FTP operations. - `knowledge_base.py` defines normalized knowledge records and resolution. - `consolemods_adapters.py`, `wiki_adapters.py`, and `dat_adapters.py` parse diff --git a/PACKAGE_LAB.md b/PACKAGE_LAB.md new file mode 100644 index 0000000..bef252f --- /dev/null +++ b/PACKAGE_LAB.md @@ -0,0 +1,47 @@ +# Package Lab + +Package Lab is UnityScraper's built-in Xbox package and image workspace. The +desktop page is available from the sidebar and the Tools menu. + +## Supported Workflows + +- **STFS (`CON`, `LIVE`, `PIRS`)**: inspect metadata, follow consecutive or + fragmented file chains, inventory and extract files, verify data-block + SHA-1 records, replace a file within its existing allocation, edit bounded + public text metadata, and rebuild the hash tree. +- **Games on Demand / SVOD**: inspect a package header and adjacent `.data` + directory, verify Data#### block hashes, and reconstruct the payload. +- **GDF/XISO**: inventory the directory tree and safely extract files from + supported Xbox disc images. +- **FATX images**: discover known partitions, follow FATX16/FATX32 chains, + inventory and extract files, and replace a file into a separate image when + it fits the existing allocation. +- **GPD/XDBF**: inspect achievements, settings, title history, and images; + update existing achievement or setting records into a separate output. + +## Write Safety + +Writes use a temporary file and atomic publication. FATX and GPD edits require +a separate output path. STFS replacement does not allocate new blocks: a +replacement must fit the file's existing chain. FATX replacement follows the +same rule. Source files remain unchanged when an operation fails. + +STFS rehashing is available without signing. A caller can provide a signer +through the packages-domain callback interface; UnityScraper does not ship or +store private signing keys. The callback must return the signature bytes in +the package-specific representation expected by the caller's lawful signing +material. + +## Deliberate Exclusions + +UnityScraper does not bundle X360's embedded key resources, Le Fluffie's +updater or artwork, account credential modification, or DLC license bypasses. +It also does not write directly to a physical device. FATX work targets image +files, and destructive changes require an explicit output image. + +## Attribution + +Format geometry and field layouts were informed by Dalavin's GPLv3 X360 +library and Le Fluffie lineage. UnityScraper uses a new Python implementation +with explicit bounds, transactional writes, and cross-platform paths. See +[Third-Party Notices](THIRD_PARTY_NOTICES.md). diff --git a/PROFILES_AND_SAVES.md b/PROFILES_AND_SAVES.md index dc31a31..f9c1e3e 100644 --- a/PROFILES_AND_SAVES.md +++ b/PROFILES_AND_SAVES.md @@ -81,19 +81,18 @@ device. ## Current Safety Boundary -This release deliberately does not: +Profile inventory and restore deliberately do not: -- edit achievements, GPD records, gamertags, or account blocks +- edit gamertags or account credential blocks - change profile, console, or device ownership fields -- rehash or resign modified CON packages +- automatically resign modified CON packages - authenticate to Xbox Live or Microsoft accounts - store CPU keys, account credentials, or signing material -- write raw FATX disks +- write physical FATX disks -Those operations can make a profile or save unusable when implemented -incorrectly. Future editing and migration support should only ship with -complete package verification, automatic pre-change snapshots, and -well-tested cross-platform signing support. +Package Lab separately provides transactional edits for existing GPD records +and STFS files. It writes to a chosen output, rebuilds STFS hashes, and accepts +only caller-supplied signing callbacks. See [Package Lab](PACKAGE_LAB.md). ## Profile Intelligence and Xenia diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index df43f04..35cc6b0 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -29,14 +29,16 @@ different lawfully obtained XeXTool build or another command-line utility. - **Archived source:** [mtolly/X360](https://github.com/mtolly/X360) - **License:** GNU General Public License version 3 -UnityScraper's profile and save implementation is new Python code informed by -the public package/profile model and field layout documented in the X360 -library and Le Fluffie source. The original GPL text is preserved at +UnityScraper's STFS, SVOD, GDF/XISO, FATX-image, GPD, profile, and save +implementation is new Python code informed by the public format geometry and +field layouts documented in the X360 library and Le Fluffie source. The +original GPL text is preserved at `assets/references/lefluffie/X360-GPL-3.0.txt`. UnityScraper does not include Le Fluffie's executable, updater, embedded key -resources, account-modification code, or artwork. The application credits -Dalavin prominently and links to the archived corresponding source. +resources, account credential modification, license-bypass behavior, or +artwork. The application credits Dalavin prominently and links to the archived +corresponding source. ## Tool Center Interoperability diff --git a/modern_gui.py b/modern_gui.py index 1f7b451..99ddc8a 100644 --- a/modern_gui.py +++ b/modern_gui.py @@ -51,6 +51,7 @@ from unityscraper.domains.library import GameSummary, LibraryService from platform_support import open_path from profile_gui import ProfileSavePage +from package_gui import PackageLabPage from profile_manager import ProfileSaveManager from setup_wizard import run_first_run_wizard from title_catalog import TitleSuggestion, XboxUnityTitleCatalog @@ -416,6 +417,7 @@ def _build_menubar(self) -> None: )), ("Tools", ( ("Backup Manager", self.show_backups), + ("Package Lab", self.show_package_lab), ("Tool Center", self.show_external_tools), ("Archive Health", self.show_health), ("Settings", self.show_settings), @@ -482,6 +484,7 @@ def _build_shell(self) -> None: (t("nav_downloads"), self.show_downloads), (t("nav_backups"), self.show_backups), (t("nav_profiles"), self.show_profiles), + ("Package Lab", self.show_package_lab), (t("nav_tools"), self.show_external_tools), (t("nav_collections"), self.show_collections), (t("nav_knowledge"), self.show_knowledge), @@ -562,6 +565,14 @@ def show_external_tools(self) -> None: CONFIG_PATH, ) + def show_package_lab(self) -> None: + self._clear_content() + self.package_lab_page = PackageLabPage( + self.root, + self.content, + self._page_header, + ) + def show_profiles(self) -> None: self._clear_content() self.profile_save_page = ProfileSavePage( diff --git a/package_gui.py b/package_gui.py new file mode 100644 index 0000000..a9a7c24 --- /dev/null +++ b/package_gui.py @@ -0,0 +1,465 @@ +"""Unified desktop work surface for Xbox package and image formats.""" + +from __future__ import annotations + +import tkinter as tk +from pathlib import Path +from tkinter import filedialog, messagebox, simpledialog, ttk +from typing import Callable + +from unityscraper.domains.packages import ( + edit_stfs_metadata, + extract_fatx, + extract_gdf, + extract_stfs_files, + extract_svod_payload, + inspect_fatx, + inspect_gdf, + inspect_stfs, + inspect_svod, + list_stfs_entries, + replace_fatx_file, + replace_stfs_file, + verify_stfs, + verify_svod, +) +from unityscraper.domains.profiles.gpd import ( + parse_gpd, + set_gpd_achievement_state, + update_gpd_setting, +) + + +class PackageLabPage: + """STFS, disc, FATX, and GPD workflows in one desktop page.""" + + def __init__( + self, + root: tk.Misc, + parent: ttk.Frame, + page_header: Callable[[str, str], None], + ) -> None: + self.root = root + self.parent = parent + self.status = tk.StringVar(value="Ready") + page_header( + "Package Lab", "Xbox package, disc image, device image, and profile database tools." + ) + self.notebook = ttk.Notebook(parent) + self.notebook.grid(row=1, column=0, sticky="nsew") + self._build_stfs_tab() + self._build_disc_tab() + self._build_fatx_tab() + self._build_gpd_tab() + ttk.Label(parent, textvariable=self.status, style="Statusbar.TLabel").grid( + row=2, column=0, sticky="ew", pady=(8, 0) + ) + + def _build_stfs_tab(self) -> None: + tab = ttk.Frame(self.notebook, padding=10) + self.notebook.add(tab, text="STFS") + tab.columnconfigure(0, weight=1) + tab.rowconfigure(2, weight=1) + self.stfs_path = tk.StringVar() + self.stfs_summary = tk.StringVar(value="No package loaded") + self._path_row(tab, self.stfs_path, self._open_stfs) + toolbar = ttk.Frame(tab) + toolbar.grid(row=1, column=0, sticky="ew", pady=(8, 8)) + for label, command in ( + ("Verify", self._verify_stfs), + ("Extract", self._extract_stfs), + ("Replace", self._replace_stfs), + ("Edit name", self._edit_stfs_name), + ): + ttk.Button(toolbar, text=label, command=command).pack(side=tk.LEFT, padx=(0, 6)) + ttk.Label(toolbar, textvariable=self.stfs_summary).pack(side=tk.RIGHT) + self.stfs_tree = self._tree(tab, ("kind", "size", "blocks"), (100, 110, 100), row=2) + + def _build_disc_tab(self) -> None: + tab = ttk.Frame(self.notebook, padding=10) + self.notebook.add(tab, text="Disc / GoD") + tab.columnconfigure(0, weight=1) + tab.rowconfigure(3, weight=1) + self.disc_path = tk.StringVar() + self._path_row(tab, self.disc_path, self._open_gdf) + actions = ttk.Frame(tab) + actions.grid(row=1, column=0, sticky="ew", pady=(8, 8)) + ttk.Button(actions, text="Open XISO", command=self._open_gdf).pack( + side=tk.LEFT, padx=(0, 6) + ) + ttk.Button(actions, text="Extract XISO", command=self._extract_gdf).pack( + side=tk.LEFT, padx=(0, 18) + ) + ttk.Button(actions, text="Open GoD", command=self._open_svod).pack( + side=tk.LEFT, padx=(0, 6) + ) + ttk.Button(actions, text="Verify GoD", command=self._verify_svod).pack( + side=tk.LEFT, padx=(0, 6) + ) + ttk.Button(actions, text="Extract GoD", command=self._extract_svod).pack(side=tk.LEFT) + self.disc_summary = tk.StringVar(value="No image loaded") + ttk.Label(tab, textvariable=self.disc_summary).grid( + row=2, column=0, sticky=tk.W, pady=(0, 6) + ) + self.disc_tree = self._tree(tab, ("kind", "size", "sector"), (100, 110, 100), row=3) + + def _build_fatx_tab(self) -> None: + tab = ttk.Frame(self.notebook, padding=10) + self.notebook.add(tab, text="FATX") + tab.columnconfigure(0, weight=1) + tab.rowconfigure(2, weight=1) + self.fatx_path = tk.StringVar() + self._path_row(tab, self.fatx_path, self._open_fatx) + actions = ttk.Frame(tab) + actions.grid(row=1, column=0, sticky="ew", pady=(8, 8)) + ttk.Button(actions, text="Extract", command=self._extract_fatx).pack( + side=tk.LEFT, padx=(0, 6) + ) + ttk.Button(actions, text="Replace", command=self._replace_fatx).pack(side=tk.LEFT) + self.fatx_tree = self._tree( + tab, ("partition", "kind", "size", "blocks"), (150, 90, 100, 90), row=2 + ) + + def _build_gpd_tab(self) -> None: + tab = ttk.Frame(self.notebook, padding=10) + self.notebook.add(tab, text="GPD") + tab.columnconfigure(0, weight=1) + tab.rowconfigure(2, weight=1) + self.gpd_path = tk.StringVar() + self._path_row(tab, self.gpd_path, self._open_gpd) + actions = ttk.Frame(tab) + actions.grid(row=1, column=0, sticky="ew", pady=(8, 8)) + ttk.Button(actions, text="Edit selected", command=self._edit_gpd).pack(side=tk.LEFT) + self.gpd_summary = tk.StringVar(value="No profile database loaded") + ttk.Label(actions, textvariable=self.gpd_summary).pack(side=tk.RIGHT) + self.gpd_tree = self._tree(tab, ("type", "state", "value"), (110, 150, 260), row=2) + + def _path_row( + self, parent: ttk.Frame, variable: tk.StringVar, command: Callable[[], None] + ) -> None: + row = ttk.Frame(parent) + row.grid(row=0, column=0, sticky="ew") + row.columnconfigure(0, weight=1) + ttk.Entry(row, textvariable=variable, state="readonly").grid(row=0, column=0, sticky="ew") + ttk.Button(row, text="Open", command=command).grid(row=0, column=1, padx=(8, 0)) + + def _tree( + self, + parent: ttk.Frame, + columns: tuple[str, ...], + widths: tuple[int, ...], + *, + row: int, + ) -> ttk.Treeview: + frame = ttk.Frame(parent) + frame.grid(row=row, column=0, sticky="nsew") + frame.columnconfigure(0, weight=1) + frame.rowconfigure(0, weight=1) + tree = ttk.Treeview(frame, columns=columns, show="tree headings", selectmode="browse") + tree.heading("#0", text="Path") + tree.column("#0", width=360, minwidth=180) + for name, width in zip(columns, widths): + tree.heading(name, text=name.title()) + tree.column(name, width=width, minwidth=70, stretch=name == columns[-1]) + scrollbar = ttk.Scrollbar(frame, orient=tk.VERTICAL, command=tree.yview) + tree.configure(yscrollcommand=scrollbar.set) + tree.grid(row=0, column=0, sticky="nsew") + scrollbar.grid(row=0, column=1, sticky="ns") + return tree + + def _run(self, action: Callable[[], None], success: str) -> None: + self.root.configure(cursor="watch") + self.status.set("Working...") + self.root.update_idletasks() + try: + action() + except Exception as exc: + self.status.set("Operation failed") + messagebox.showerror("Package Lab", str(exc), parent=self.root) + else: + self.status.set(success) + finally: + self.root.configure(cursor="") + + def _choose_file(self, title: str, patterns: tuple[tuple[str, str], ...]) -> str: + return filedialog.askopenfilename(parent=self.root, title=title, filetypes=patterns) + + def _open_stfs(self) -> None: + selected = self._choose_file("Open STFS package", (("Xbox packages", "*.*"),)) + if not selected: + return + self.stfs_path.set(selected) + + def action() -> None: + package = inspect_stfs(selected) + entries = list_stfs_entries(selected) + self._clear_tree(self.stfs_tree) + for entry in entries: + self.stfs_tree.insert( + "", + tk.END, + iid=f"stfs:{entry.index}", + text=entry.path, + values=( + "Folder" if entry.is_directory else "File", + entry.size, + len(entry.blocks), + ), + ) + self.stfs_summary.set( + f"{package.title_id} {package.content_label} {len(entries):,} entries" + ) + + self._run(action, "STFS package loaded") + + def _verify_stfs(self) -> None: + if not self.stfs_path.get(): + return + + def action() -> None: + report = verify_stfs(self.stfs_path.get()) + self.stfs_summary.set( + f"{report.valid_blocks:,} valid {report.mismatched_blocks:,} mismatched {report.unverifiable_blocks:,} missing" + ) + + self._run(action, "STFS verification completed") + + def _extract_stfs(self) -> None: + destination = filedialog.askdirectory(parent=self.root, title="Extract STFS") + if destination and self.stfs_path.get(): + selected = self._selected_path(self.stfs_tree) + self._run( + lambda: extract_stfs_files( + self.stfs_path.get(), destination, [selected] if selected else None + ), + "STFS extraction completed", + ) + + def _replace_stfs(self) -> None: + internal = self._selected_path(self.stfs_tree) + replacement = self._choose_file("Choose replacement", (("All files", "*.*"),)) + if not internal or not replacement or not self.stfs_path.get(): + return + output = filedialog.asksaveasfilename(parent=self.root, title="Save edited package") + if output: + self._run( + lambda: replace_stfs_file( + self.stfs_path.get(), internal, replacement, output=output + ), + "STFS replacement completed", + ) + + def _edit_stfs_name(self) -> None: + if not self.stfs_path.get(): + return + value = simpledialog.askstring("Display name", "Display name", parent=self.root) + if value is None: + return + output = filedialog.asksaveasfilename(parent=self.root, title="Save edited package") + if output: + self._run( + lambda: edit_stfs_metadata( + self.stfs_path.get(), {"display_name": value}, output=output + ), + "STFS metadata updated", + ) + + def _open_gdf(self) -> None: + selected = self._choose_file( + "Open XISO/GDF image", (("Disc images", "*.iso *.xiso *.gdf"), ("All files", "*.*")) + ) + if not selected: + return + self.disc_path.set(selected) + + def action() -> None: + image = inspect_gdf(selected) + self._clear_tree(self.disc_tree) + for index, entry in enumerate(image.entries): + self.disc_tree.insert( + "", + tk.END, + iid=f"gdf:{index}", + text=entry.path, + values=( + "Folder" if entry.is_directory else "File", + entry.size, + entry.start_sector, + ), + ) + self.disc_summary.set(f"XISO/GDF {len(image.entries):,} entries") + + self._run(action, "Disc image loaded") + + def _extract_gdf(self) -> None: + destination = filedialog.askdirectory(parent=self.root, title="Extract XISO/GDF") + if destination and self.disc_path.get(): + self._run( + lambda: extract_gdf(self.disc_path.get(), destination), "Disc extraction completed" + ) + + def _open_svod(self) -> None: + selected = self._choose_file("Open Games on Demand header", (("GoD headers", "*.*"),)) + if not selected: + return + self.disc_path.set(selected) + self._run( + lambda: self.disc_summary.set(self._svod_summary(inspect_svod(selected))), + "Games on Demand package loaded", + ) + + @staticmethod + def _svod_summary(package) -> str: + return f"GoD {package.title_id} {package.block_count:,} blocks {package.data_file_count:,} files" + + def _verify_svod(self) -> None: + if self.disc_path.get(): + + def action() -> None: + report = verify_svod(self.disc_path.get()) + self.disc_summary.set( + f"GoD {report.valid_blocks:,} valid {report.mismatched_blocks:,} mismatched" + ) + + self._run(action, "Games on Demand verification completed") + + def _extract_svod(self) -> None: + if not self.disc_path.get(): + return + output = filedialog.asksaveasfilename( + parent=self.root, title="Save GoD payload", defaultextension=".iso" + ) + if output: + self._run( + lambda: extract_svod_payload(self.disc_path.get(), output), "GoD payload extracted" + ) + + def _open_fatx(self) -> None: + selected = self._choose_file( + "Open FATX image", (("Device images", "*.img *.bin *.dd"), ("All files", "*.*")) + ) + if not selected: + return + self.fatx_path.set(selected) + + def action() -> None: + image = inspect_fatx(selected) + self._clear_tree(self.fatx_tree) + for index, entry in enumerate(image.entries): + self.fatx_tree.insert( + "", + tk.END, + iid=f"fatx:{index}", + text=entry.path, + values=( + entry.partition, + "Folder" if entry.is_directory else "File", + entry.size, + len(entry.blocks), + ), + ) + + self._run(action, "FATX image loaded") + + def _extract_fatx(self) -> None: + destination = filedialog.askdirectory(parent=self.root, title="Extract FATX") + if destination and self.fatx_path.get(): + selected = self._selected_path(self.fatx_tree) + self._run( + lambda: extract_fatx( + self.fatx_path.get(), destination, [selected] if selected else None + ), + "FATX extraction completed", + ) + + def _replace_fatx(self) -> None: + internal = self._selected_path(self.fatx_tree) + replacement = self._choose_file("Choose replacement", (("All files", "*.*"),)) + if not internal or not replacement or not self.fatx_path.get(): + return + output = filedialog.asksaveasfilename(parent=self.root, title="Save edited FATX image") + if output: + self._run( + lambda: replace_fatx_file( + self.fatx_path.get(), internal, replacement, output=output + ), + "FATX replacement completed", + ) + + def _open_gpd(self) -> None: + selected = self._choose_file( + "Open GPD", (("Xbox profile databases", "*.gpd"), ("All files", "*.*")) + ) + if not selected: + return + self.gpd_path.set(selected) + + def action() -> None: + report = parse_gpd(selected) + self._clear_tree(self.gpd_tree) + for item in report.achievements: + self.gpd_tree.insert( + "", + tk.END, + iid=f"achievement:{item.achievement_id}", + text=item.title, + values=("Achievement", item.state, item.gamerscore), + ) + for item in report.settings: + self.gpd_tree.insert( + "", + tk.END, + iid=f"setting:{item.setting_id}", + text=f"0x{item.setting_id:08X}", + values=("Setting", item.value_type, str(item.value)), + ) + self.gpd_summary.set( + f"{report.unlocked_count:,}/{len(report.achievements):,} achievements {len(report.settings):,} settings" + ) + + self._run(action, "GPD loaded") + + def _edit_gpd(self) -> None: + selection = self.gpd_tree.selection() + if not selection or not self.gpd_path.get(): + return + kind, raw_id = selection[0].split(":", 1) + output = filedialog.asksaveasfilename( + parent=self.root, title="Save edited GPD", defaultextension=".gpd" + ) + if not output: + return + if kind == "achievement": + state = simpledialog.askstring( + "Achievement state", + "locked, unlocked-offline, or unlocked-online", + parent=self.root, + ) + if state: + self._run( + lambda: set_gpd_achievement_state( + self.gpd_path.get(), int(raw_id), state, output=output + ), + "Achievement updated", + ) + else: + value = simpledialog.askstring("Setting value", "Value", parent=self.root) + if value is not None: + self._run( + lambda: update_gpd_setting( + self.gpd_path.get(), int(raw_id), value, output=output + ), + "Setting updated", + ) + + @staticmethod + def _clear_tree(tree: ttk.Treeview) -> None: + tree.delete(*tree.get_children()) + + @staticmethod + def _selected_path(tree: ttk.Treeview) -> str: + selection = tree.selection() + return str(tree.item(selection[0], "text")) if selection else "" + + +__all__ = ["PackageLabPage"] diff --git a/tests.py b/tests.py index bdbebce..f3708f2 100644 --- a/tests.py +++ b/tests.py @@ -2720,6 +2720,182 @@ def test_package_workspace_is_read_only_and_profile_tools_are_audited(self): self.assertFalse(comparison["identical"]) +class TestPackageLabDomains(unittest.TestCase): + """Package Lab format engines and transactional editing.""" + + def setUp(self): + self.temp_dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + shutil.rmtree(self.temp_dir) + + def test_fragmented_stfs_round_trip_and_tamper_detection(self): + from unityscraper.domains.packages import ( + edit_stfs_metadata, + extract_stfs_files, + inspect_stfs, + list_stfs_entries, + replace_stfs_file, + verify_stfs, + ) + + payload = bytearray(0xF000) + payload[:4] = b"LIVE" + payload[0x340:0x344] = (0xA000).to_bytes(4, "big") + payload[0x344:0x348] = (1).to_bytes(4, "big") + payload[0x360:0x364] = bytes.fromhex("53510804") + payload[0x379] = 0x24 + payload[0x37B] = 1 + payload[0x37C:0x37E] = (1).to_bytes(2, "little") + payload[0x395:0x399] = (4).to_bytes(4, "big") + entry = 0xB000 + name = b"fragmented.bin" + payload[entry:entry + len(name)] = name + payload[entry + 0x28] = len(name) + payload[entry + 0x29:entry + 0x2C] = (2).to_bytes(3, "little") + payload[entry + 0x2F:entry + 0x32] = (1).to_bytes(3, "little") + payload[entry + 0x32:entry + 0x34] = (0xFFFF).to_bytes(2, "big") + payload[entry + 0x34:entry + 0x38] = (0x1004).to_bytes(4, "big") + payload[0xC000:0xD000] = b"A" * 0x1000 + payload[0xE000:0xE004] = b"tail" + for block, offset in enumerate((0xB000, 0xC000, 0xD000, 0xE000)): + record = 0xA000 + block * 0x18 + payload[record:record + 0x14] = hashlib.sha1( + payload[offset:offset + 0x1000] + ).digest() + next_block = 3 if block == 1 else 0xFFFFFF + payload[record + 0x14:record + 0x18] = ( + (2 << 30) | next_block + ).to_bytes(4, "big") + source = self.temp_dir / "fragmented.stfs" + source.write_bytes(payload) + self.assertEqual(list_stfs_entries(source)[0].blocks, (1, 3)) + self.assertTrue(verify_stfs(source).valid) + + replacement = self.temp_dir / "replacement.bin" + replacement.write_bytes(b"replacement") + edited = self.temp_dir / "edited.stfs" + replace_stfs_file(source, "fragmented.bin", replacement, output=edited) + named = self.temp_dir / "named.stfs" + edit_stfs_metadata(edited, {"display_name": "Edited Save"}, output=named) + self.assertEqual(inspect_stfs(named).display_name, "Edited Save") + output = self.temp_dir / "stfs-output" + extract_stfs_files(named, output) + self.assertEqual((output / "fragmented.bin").read_bytes(), b"replacement") + self.assertTrue(verify_stfs(named).valid) + changed = bytearray(named.read_bytes()) + changed[0xC000] ^= 0xFF + named.write_bytes(changed) + self.assertFalse(verify_stfs(named).valid) + + def test_gdf_and_svod_inventory_verification_and_extraction(self): + from unityscraper.domains.packages import ( + extract_gdf, + extract_svod_payload, + inspect_gdf, + inspect_svod, + verify_svod, + ) + + image = bytearray(0x1800) + image[:20] = b"MICROSOFT*XBOX*MEDIA" + image[20:24] = (1).to_bytes(4, "little") + image[24:28] = (0x40).to_bytes(4, "little") + image[0x804:0x808] = (2).to_bytes(4, "little") + image[0x808:0x80C] = (7).to_bytes(4, "little") + image[0x80C] = 0x80 + image[0x80D] = 10 + image[0x80E:0x818] = b"readme.txt" + image[0x1000:0x1007] = b"xbox360" + gdf = self.temp_dir / "sample.iso" + gdf.write_bytes(image) + self.assertEqual(inspect_gdf(gdf).entries[0].path, "readme.txt") + gdf_output = self.temp_dir / "gdf-output" + extract_gdf(gdf, gdf_output) + self.assertEqual((gdf_output / "readme.txt").read_bytes(), b"xbox360") + + header = bytearray(0xB000) + header[:4] = b"LIVE" + header[0x340:0x344] = (0xB000).to_bytes(4, "big") + header[0x344:0x348] = (0x00007000).to_bytes(4, "big") + header[0x360:0x364] = bytes.fromhex("53510804") + header[0x379:0x37D] = b"\x24\x05\x05\x11" + header[0x392:0x395] = (1).to_bytes(3, "big") + header[0x39D:0x3A1] = (1).to_bytes(4, "big") + header[0x3A1:0x3A9] = (4).to_bytes(8, "big") + svod = self.temp_dir / "god-header" + svod.write_bytes(header) + data_dir = svod.with_name(svod.name + ".data") + data_dir.mkdir() + data = bytearray(0x3000) + data[0x2000:0x2004] = b"GOD!" + data[0x1000:0x1014] = hashlib.sha1(data[0x2000:0x3000]).digest() + (data_dir / "Data0000").write_bytes(data) + self.assertEqual(inspect_svod(svod).block_count, 1) + self.assertTrue(verify_svod(svod).valid) + svod_output = self.temp_dir / "payload.iso" + extract_svod_payload(svod, svod_output) + self.assertEqual(svod_output.read_bytes(), b"GOD!") + + def test_fatx_inventory_extraction_and_guarded_replacement(self): + from unityscraper.domains.packages import ( + extract_fatx, + inspect_fatx, + replace_fatx_file, + ) + + image = bytearray(0x6000) + image[:4] = b"FATX" + image[8:12] = (8).to_bytes(4, "big") + image[12:16] = (1).to_bytes(4, "big") + image[0x1002:0x1004] = (0xFFFF).to_bytes(2, "big") + image[0x1004:0x1006] = (0xFFFF).to_bytes(2, "big") + image[0x2000] = 8 + image[0x2002:0x200A] = b"game.bin" + image[0x202C:0x2030] = (2).to_bytes(4, "big") + image[0x2030:0x2034] = (4).to_bytes(4, "big") + image[0x3000:0x3004] = b"FATX" + source = self.temp_dir / "fatx.img" + source.write_bytes(image) + self.assertEqual(inspect_fatx(source).entries[0].path, "game.bin") + output = self.temp_dir / "fatx-output" + extract_fatx(source, output) + self.assertEqual((output / "game.bin").read_bytes(), b"FATX") + replacement = self.temp_dir / "replacement.bin" + replacement.write_bytes(b"EDIT") + edited = self.temp_dir / "edited.img" + replace_fatx_file(source, "game.bin", replacement, output=edited) + edited_output = self.temp_dir / "fatx-edited-output" + extract_fatx(edited, edited_output) + self.assertEqual((edited_output / "game.bin").read_bytes(), b"EDIT") + + def test_gpd_achievement_edit_is_transactional(self): + import struct + + from unityscraper.domains.profiles.gpd import set_gpd_achievement_state + + strings = b"".join( + value.encode("utf-16-be") + b"\0\0" + for value in ("First Steps", "Locked text", "Unlocked text") + ) + payload = bytearray(0x1C) + payload[4:8] = (7).to_bytes(4, "big", signed=True) + payload[8:12] = (42).to_bytes(4, "big", signed=True) + payload[12:16] = (25).to_bytes(4, "big") + entry_payload = bytes(payload) + strings + header = struct.pack(">4sIIIII", b"XDBF", 1, 1, 1, 0, 0) + entry = struct.pack(">Hqii", 1, 100, 0, len(entry_payload)) + source = self.temp_dir / "53510804.gpd" + original = header + entry + entry_payload + source.write_bytes(original) + output = self.temp_dir / "edited.gpd" + report = set_gpd_achievement_state( + source, 7, "unlocked-offline", output=output + ) + self.assertEqual(report.unlocked_count, 1) + self.assertEqual(source.read_bytes(), original) + + def run_tests(): """Run all tests""" # Create test suite @@ -2747,6 +2923,7 @@ def run_tests(): suite.addTests(loader.loadTestsFromTestCase(TestRoadmapFeatures)) suite.addTests(loader.loadTestsFromTestCase(TestUnifiedV1Foundation)) suite.addTests(loader.loadTestsFromTestCase(TestCommunityRoadmap)) + suite.addTests(loader.loadTestsFromTestCase(TestPackageLabDomains)) # Run tests runner = unittest.TextTestRunner(verbosity=2) diff --git a/unityscraper/domains/packages/__init__.py b/unityscraper/domains/packages/__init__.py index 334fea0..9696297 100644 --- a/unityscraper/domains/packages/__init__.py +++ b/unityscraper/domains/packages/__init__.py @@ -11,21 +11,47 @@ list_stfs_entries, verify_stfs, ) +from .gdf import GdfEntry, GdfImage, extract_gdf, inspect_gdf +from .fatx import ( + FatxEntry, + FatxImage, + FatxPartition, + extract_fatx, + inspect_fatx, + replace_fatx_file, +) from .models import ( StfsBlockVerification, StfsEntry, StfsHashRecord, StfsIntegrityReport, + StfsMutationResult, StfsPackage, XbePackage, XexPackage, ) +from .mutations import edit_stfs_metadata, rehash_stfs, replace_stfs_file +from .svod import ( + SvodIntegrityReport, + SvodPackage, + extract_svod_payload, + inspect_svod, + verify_svod, +) __all__ = [ "StfsEntry", + "GdfEntry", + "GdfImage", + "FatxEntry", + "FatxImage", + "FatxPartition", "StfsBlockVerification", "StfsHashRecord", "StfsIntegrityReport", + "StfsMutationResult", + "SvodIntegrityReport", + "SvodPackage", "StfsPackage", "XbePackage", "XexPackage", @@ -38,4 +64,15 @@ "inspect_xex", "list_stfs_entries", "verify_stfs", + "edit_stfs_metadata", + "rehash_stfs", + "replace_stfs_file", + "extract_gdf", + "inspect_gdf", + "extract_fatx", + "inspect_fatx", + "replace_fatx_file", + "extract_svod_payload", + "inspect_svod", + "verify_svod", ] diff --git a/unityscraper/domains/packages/executables.py b/unityscraper/domains/packages/executables.py index 6d77959..b57f488 100644 --- a/unityscraper/domains/packages/executables.py +++ b/unityscraper/domains/packages/executables.py @@ -53,21 +53,18 @@ def inspect_xex(path: str | Path) -> XexPackage: execution_offset = 0 for index in range(optional_count): entry = 0x18 + index * 8 - key = int.from_bytes(header[entry:entry + 4], "big") - value = int.from_bytes(header[entry + 4:entry + 8], "big") + key = int.from_bytes(header[entry : entry + 4], "big") + value = int.from_bytes(header[entry + 4 : entry + 8], "big") if key == 0x00040006: execution_offset = value break if not execution_offset or execution_offset + 24 > len(header): raise InvalidPackageError("XEX execution metadata is unavailable") - info = header[execution_offset:execution_offset + 24] + info = header[execution_offset : execution_offset + 24] def format_version(value: int) -> str: - return ( - f"{(value >> 28) & 0xF}.{(value >> 24) & 0xF}." - f"{(value >> 8) & 0xFFFF}.{value & 0xFF}" - ) + return f"{(value >> 28) & 0xF}.{(value >> 24) & 0xF}.{(value >> 8) & 0xFFFF}.{value & 0xFF}" return XexPackage( path=package_path, diff --git a/unityscraper/domains/packages/fatx.py b/unityscraper/domains/packages/fatx.py new file mode 100644 index 0000000..cc89756 --- /dev/null +++ b/unityscraper/domains/packages/fatx.py @@ -0,0 +1,312 @@ +"""Read-first FATX image browsing, extraction, and guarded replacement.""" + +from __future__ import annotations + +import os +import shutil +import tempfile +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import BinaryIO, Iterable + +from .errors import InvalidPackageError, UnsafeArchiveError + +FATX_MAGIC = b"FATX" +SECTOR_SIZE = 0x200 +ENTRY_SIZE = 0x40 +MAX_ENTRIES = 1_000_000 +MAX_DEPTH = 128 +KNOWN_PARTITIONS = ( + ("Memory Unit Content", 0x7FF000), + ("Hard Drive System", 0x118EB0000), + ("Hard Drive Compatibility", 0x120EB0000), + ("Hard Drive Content", 0x130EB0000), + ("USB Cache", 0x8000400), + ("USB Content", 0x20000000), + ("Image", 0), +) + + +@dataclass(frozen=True) +class FatxPartition: + name: str + offset: int + size: int + sectors_per_block: int + block_size: int + fat_entry_size: int + fat_size: int + data_offset: int + root_block: int + + +@dataclass(frozen=True) +class FatxEntry: + partition: str + path: str + name: str + is_directory: bool + start_block: int + size: int + entry_offset: int + blocks: tuple[int, ...] + + +@dataclass(frozen=True) +class FatxImage: + path: Path + partitions: tuple[FatxPartition, ...] + entries: tuple[FatxEntry, ...] + + +def inspect_fatx(path: str | Path) -> FatxImage: + source = Path(path).expanduser().resolve() + image_size = source.stat().st_size + offsets = [(name, offset) for name, offset in KNOWN_PARTITIONS if offset + 4 <= image_size] + partitions: list[FatxPartition] = [] + entries: list[FatxEntry] = [] + with source.open("rb") as handle: + valid_offsets = [] + for name, offset in offsets: + handle.seek(offset) + if handle.read(4) == FATX_MAGIC: + valid_offsets.append((name, offset)) + for index, (name, offset) in enumerate(valid_offsets): + next_offset = min( + (candidate for _, candidate in valid_offsets if candidate > offset), + default=image_size, + ) + partition = _read_partition(handle, name, offset, next_offset - offset) + partitions.append(partition) + entries.extend(_read_directory(handle, partition, partition.root_block, "", 0, set())) + if not partitions: + raise InvalidPackageError("No supported FATX partitions were found") + return FatxImage(source, tuple(partitions), tuple(entries)) + + +def extract_fatx( + path: str | Path, + destination: str | Path, + selected_paths: Iterable[str] | None = None, +) -> dict[str, object]: + image = inspect_fatx(path) + target = Path(destination).expanduser().resolve() + requested = {item.replace("\\", "/").strip("/") for item in selected_paths or ()} + files = [ + entry + for entry in image.entries + if not entry.is_directory and (not requested or entry.path in requested) + ] + missing = requested - {entry.path for entry in files} + if missing: + raise InvalidPackageError(f"FATX entries were not found: {', '.join(sorted(missing)[:5])}") + by_name = {partition.name: partition for partition in image.partitions} + target.mkdir(parents=True, exist_ok=True) + extracted: list[dict[str, object]] = [] + with image.path.open("rb") as handle: + for entry in files: + output = (target / _safe_member(entry.path)).resolve() + if not output.is_relative_to(target): + raise UnsafeArchiveError(f"FATX path escapes destination: {entry.path}") + output.parent.mkdir(parents=True, exist_ok=True) + temporary = output.with_name(output.name + ".partial") + partition = by_name[entry.partition] + remaining = entry.size + try: + with temporary.open("xb") as destination_handle: + for block in entry.blocks: + handle.seek(_block_offset(partition, block)) + chunk = handle.read(min(remaining, partition.block_size)) + if len(chunk) != min(remaining, partition.block_size): + raise InvalidPackageError(f"FATX file is truncated: {entry.path}") + destination_handle.write(chunk) + remaining -= len(chunk) + if remaining: + raise InvalidPackageError(f"FATX chain ends early: {entry.path}") + if output.exists(): + raise FileExistsError(output) + temporary.replace(output) + finally: + temporary.unlink(missing_ok=True) + extracted.append({"path": entry.path, "output": str(output), "size": entry.size}) + return {"source": str(image.path), "extracted": extracted} + + +def replace_fatx_file( + image_path: str | Path, + internal_path: str, + replacement: str | Path, + *, + output: str | Path, +) -> Path: + """Replace a FATX file only when it fits the existing chain, into a new image.""" + source = Path(image_path).expanduser().resolve() + incoming = Path(replacement).expanduser().resolve() + target = Path(output).expanduser().resolve() + if target == source: + raise InvalidPackageError("FATX writes require a separate output image") + image = inspect_fatx(source) + normalized = internal_path.replace("\\", "/").strip("/") + entry = next( + (item for item in image.entries if item.path == normalized and not item.is_directory), None + ) + if entry is None: + raise InvalidPackageError(f"FATX file was not found: {normalized}") + partition = next(item for item in image.partitions if item.name == entry.partition) + if incoming.stat().st_size > len(entry.blocks) * partition.block_size: + raise InvalidPackageError("Replacement exceeds the existing FATX allocation") + target.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{target.name}.", suffix=".partial", dir=target.parent + ) + os.close(descriptor) + temporary = Path(temporary_name) + try: + shutil.copy2(source, temporary) + with incoming.open("rb") as source_handle, temporary.open("r+b") as handle: + remaining = incoming.stat().st_size + for block in entry.blocks: + chunk = ( + source_handle.read(min(remaining, partition.block_size)) if remaining else b"" + ) + handle.seek(_block_offset(partition, block)) + handle.write(chunk.ljust(partition.block_size, b"\0")) + remaining -= len(chunk) + handle.seek(entry.entry_offset + 0x30) + handle.write(incoming.stat().st_size.to_bytes(4, "big")) + if target.exists(): + raise FileExistsError(target) + os.replace(temporary, target) + finally: + temporary.unlink(missing_ok=True) + return target + + +def _read_partition(handle: BinaryIO, name: str, offset: int, size: int) -> FatxPartition: + handle.seek(offset + 8) + values = handle.read(8) + if len(values) != 8: + raise InvalidPackageError("FATX partition header is truncated") + sectors_per_block = int.from_bytes(values[:4], "big") + root_block = int.from_bytes(values[4:], "big") + if sectors_per_block <= 0 or sectors_per_block > 0x10000: + raise InvalidPackageError("FATX sectors-per-block value is invalid") + block_size = sectors_per_block * SECTOR_SIZE + block_count = size // block_size + fat_entry_size = 2 if block_count < 0xFFF5 else 4 + raw_fat_size = block_count * fat_entry_size + fat_size = raw_fat_size + (0x1000 - raw_fat_size % 0x1000) + data_offset = offset + 0x1000 + fat_size + partition = FatxPartition( + name, + offset, + size, + sectors_per_block, + block_size, + fat_entry_size, + fat_size, + data_offset, + root_block, + ) + _block_offset(partition, root_block) + return partition + + +def _read_directory( + handle: BinaryIO, + partition: FatxPartition, + start_block: int, + parent: str, + depth: int, + visited_directories: set[int], +) -> list[FatxEntry]: + if depth > MAX_DEPTH or start_block in visited_directories: + raise InvalidPackageError("FATX directory graph is invalid") + visited_directories.add(start_block) + entries: list[FatxEntry] = [] + for directory_block in _block_chain(handle, partition, start_block): + base = _block_offset(partition, directory_block) + for index in range(partition.block_size // ENTRY_SIZE): + offset = base + index * ENTRY_SIZE + handle.seek(offset) + data = handle.read(ENTRY_SIZE) + name_length = data[0] if data else 0 + if name_length == 0xE5: + continue + if name_length in {0, 0xFF}: + break + if name_length > 0x2A or len(data) != ENTRY_SIZE: + raise InvalidPackageError("FATX directory entry is invalid") + name = data[2 : 2 + name_length].decode("ascii", errors="replace") + if any(character in name for character in "\\/\0") or name in {".", ".."}: + raise InvalidPackageError("FATX directory contains an unsafe name") + start = int.from_bytes(data[0x2C:0x30], "big") + size = int.from_bytes(data[0x30:0x34], "big") + is_directory = bool(data[1] & 0x10) + blocks = _block_chain(handle, partition, start) + full_path = f"{parent}/{name}" if parent else name + entry = FatxEntry( + partition.name, full_path, name, is_directory, start, size, offset, blocks + ) + entries.append(entry) + if len(entries) > MAX_ENTRIES: + raise InvalidPackageError("FATX directory exceeds the safety limit") + for entry in tuple(entries): + if entry.is_directory: + entries.extend( + _read_directory( + handle, + partition, + entry.start_block, + entry.path, + depth + 1, + visited_directories, + ) + ) + visited_directories.remove(start_block) + return entries + + +def _block_chain(handle: BinaryIO, partition: FatxPartition, start: int) -> tuple[int, ...]: + end = 0xFFFF if partition.fat_entry_size == 2 else 0xFFFFFFFF + blocks: list[int] = [] + visited: set[int] = set() + current = start + max_blocks = partition.size // partition.block_size + while current not in {0, end}: + if current >= max_blocks or current in visited: + raise InvalidPackageError("FATX allocation chain is invalid") + visited.add(current) + blocks.append(current) + handle.seek(partition.offset + 0x1000 + current * partition.fat_entry_size) + raw = handle.read(partition.fat_entry_size) + if len(raw) != partition.fat_entry_size: + raise InvalidPackageError("FATX allocation table is truncated") + current = int.from_bytes(raw, "big") + return tuple(blocks) + + +def _block_offset(partition: FatxPartition, block: int) -> int: + max_blocks = partition.size // partition.block_size + if block <= 0 or block >= max_blocks: + raise InvalidPackageError("FATX block points outside the partition") + return partition.data_offset + (block - 1) * partition.block_size + + +def _safe_member(value: str) -> Path: + pure = PurePosixPath(value.replace("\\", "/")) + if pure.is_absolute() or not pure.parts or any(part in {"", ".", ".."} for part in pure.parts): + raise UnsafeArchiveError(f"Unsafe FATX path: {value}") + if ":" in pure.parts[0]: + raise UnsafeArchiveError(f"Unsafe FATX path: {value}") + return Path(*pure.parts) + + +__all__ = [ + "FatxEntry", + "FatxImage", + "FatxPartition", + "extract_fatx", + "inspect_fatx", + "replace_fatx_file", +] diff --git a/unityscraper/domains/packages/gdf.py b/unityscraper/domains/packages/gdf.py new file mode 100644 index 0000000..a720ad1 --- /dev/null +++ b/unityscraper/domains/packages/gdf.py @@ -0,0 +1,234 @@ +"""Bounded GDF/XISO image browsing and extraction.""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import BinaryIO, Iterable + +from .errors import InvalidPackageError, UnsafeArchiveError + +GDF_MAGIC = b"MICROSOFT*XBOX*MEDIA" +SECTOR_SIZE = 0x800 +MAGIC_OFFSETS = (0, 0x10000, 0x1FB20, 0x30600, 0xFDA0000) +MAX_ENTRIES = 500_000 +MAX_DEPTH = 128 + + +@dataclass(frozen=True) +class GdfEntry: + path: str + name: str + is_directory: bool + start_sector: int + size: int + entry_offset: int + + +@dataclass(frozen=True) +class GdfImage: + path: Path + base_offset: int + deviation: int + root_sector: int + root_size: int + entries: tuple[GdfEntry, ...] + + +def inspect_gdf(path: str | Path, *, deviation: int = 0) -> GdfImage: + source = Path(path).expanduser().resolve() + size = source.stat().st_size + with source.open("rb") as handle: + base = _find_magic(handle, size) + handle.seek(base + len(GDF_MAGIC)) + descriptor = handle.read(17) + if len(descriptor) != 17: + raise InvalidPackageError("GDF volume descriptor is truncated") + root_sector = int.from_bytes(descriptor[:4], "little") + root_size = int.from_bytes(descriptor[4:8], "little", signed=True) + if root_size < 0: + raise InvalidPackageError("GDF root directory size is invalid") + entries = _read_directory( + handle, + size, + base, + deviation, + root_sector, + root_size, + "", + 0, + set(), + ) + if len(entries) > MAX_ENTRIES: + raise InvalidPackageError("GDF directory exceeds the safety limit") + return GdfImage(source, base, deviation, root_sector, root_size, tuple(entries)) + + +def extract_gdf( + path: str | Path, + destination: str | Path, + selected_paths: Iterable[str] | None = None, + *, + deviation: int = 0, + max_output_size: int = 128 * 1024 * 1024 * 1024, +) -> dict[str, object]: + image = inspect_gdf(path, deviation=deviation) + target = Path(destination).expanduser().resolve() + requested = {item.replace("\\", "/").strip("/") for item in selected_paths or ()} + files = [ + entry + for entry in image.entries + if not entry.is_directory and (not requested or entry.path in requested) + ] + missing = requested - {entry.path for entry in files} + if missing: + raise InvalidPackageError(f"GDF entries were not found: {', '.join(sorted(missing)[:5])}") + if sum(entry.size for entry in files) > max_output_size: + raise InvalidPackageError("Selected GDF output exceeds the safety limit") + target.mkdir(parents=True, exist_ok=True) + extracted: list[dict[str, object]] = [] + with image.path.open("rb") as handle: + for entry in files: + relative = _safe_member(entry.path) + output = (target / relative).resolve() + if not output.is_relative_to(target): + raise UnsafeArchiveError(f"GDF path escapes destination: {entry.path}") + output.parent.mkdir(parents=True, exist_ok=True) + temporary = output.with_name(output.name + ".partial") + digest = hashlib.sha256() + data_offset = _data_offset(image.base_offset, deviation, entry.start_sector) + if data_offset < 0 or data_offset + entry.size > image.path.stat().st_size: + raise InvalidPackageError(f"GDF data points outside the image: {entry.path}") + handle.seek(data_offset) + remaining = entry.size + try: + with temporary.open("xb") as destination_handle: + while remaining: + chunk = handle.read(min(1024 * 1024, remaining)) + if not chunk: + raise InvalidPackageError(f"GDF file is truncated: {entry.path}") + destination_handle.write(chunk) + digest.update(chunk) + remaining -= len(chunk) + if output.exists(): + raise FileExistsError(output) + temporary.replace(output) + finally: + temporary.unlink(missing_ok=True) + extracted.append( + { + "path": entry.path, + "output": str(output), + "size": entry.size, + "sha256": digest.hexdigest(), + } + ) + return {"source": str(image.path), "extracted": extracted} + + +def _find_magic(handle: BinaryIO, size: int) -> int: + for offset in MAGIC_OFFSETS: + if offset + len(GDF_MAGIC) > size: + continue + handle.seek(offset) + if handle.read(len(GDF_MAGIC)) == GDF_MAGIC: + return offset + raise InvalidPackageError("Not a supported GDF/XISO image") + + +def _read_directory( + handle: BinaryIO, + image_size: int, + base: int, + deviation: int, + sector: int, + size: int, + parent: str, + depth: int, + visited_directories: set[tuple[int, int]], +) -> list[GdfEntry]: + if depth > MAX_DEPTH: + raise InvalidPackageError("GDF directory nesting exceeds the safety limit") + directory_key = (sector, size) + if directory_key in visited_directories: + raise InvalidPackageError("GDF directory graph contains a loop") + visited_directories.add(directory_key) + directory_offset = _data_offset(base, deviation, sector) + if directory_offset < 0 or directory_offset + size > image_size: + raise InvalidPackageError("GDF directory points outside the image") + entries: list[GdfEntry] = [] + pending = [0] + visited_nodes: set[int] = set() + while pending: + node = pending.pop() + if node in visited_nodes: + continue + visited_nodes.add(node) + offset = directory_offset + node * 4 + if offset < directory_offset or offset + 14 > directory_offset + size: + raise InvalidPackageError("GDF directory node points outside its table") + handle.seek(offset) + header = handle.read(14) + left = int.from_bytes(header[:2], "little") + right = int.from_bytes(header[2:4], "little") + start_sector = int.from_bytes(header[4:8], "little") + entry_size = int.from_bytes(header[8:12], "little", signed=True) + attributes = header[12] + name_length = header[13] + if ( + entry_size < 0 + or name_length in {0, 0xFF} + or offset + 14 + name_length > directory_offset + size + ): + raise InvalidPackageError("GDF directory entry is invalid") + name = handle.read(name_length).decode("ascii", errors="replace") + if any(character in name for character in "\\/\0") or name in {".", ".."}: + raise InvalidPackageError("GDF directory contains an unsafe name") + full_path = f"{parent}/{name}" if parent else name + entry = GdfEntry(full_path, name, bool(attributes & 0x10), start_sector, entry_size, offset) + entries.append(entry) + if len(entries) > MAX_ENTRIES: + raise InvalidPackageError("GDF directory exceeds the safety limit") + if left: + pending.append(left) + if right: + pending.append(right) + for entry in tuple(entries): + if entry.is_directory: + entries.extend( + _read_directory( + handle, + image_size, + base, + deviation, + entry.start_sector, + entry.size, + entry.path, + depth + 1, + visited_directories, + ) + ) + visited_directories.remove(directory_key) + return entries + + +def _data_offset(base: int, deviation: int, sector: int) -> int: + result = sector * SECTOR_SIZE + if deviation: + result -= (deviation - 1) << 12 + if base > 0x10000: + result += base + return result + + +def _safe_member(value: str) -> Path: + pure = PurePosixPath(value.replace("\\", "/")) + if pure.is_absolute() or not pure.parts or any(part in {"", ".", ".."} for part in pure.parts): + raise UnsafeArchiveError(f"Unsafe GDF path: {value}") + if ":" in pure.parts[0]: + raise UnsafeArchiveError(f"Unsafe GDF path: {value}") + return Path(*pure.parts) + + +__all__ = ["GdfEntry", "GdfImage", "extract_gdf", "inspect_gdf"] diff --git a/unityscraper/domains/packages/models.py b/unityscraper/domains/packages/models.py index fa8a779..68016eb 100644 --- a/unityscraper/domains/packages/models.py +++ b/unityscraper/domains/packages/models.py @@ -41,6 +41,18 @@ class StfsEntry: parent_index: int size: int blocks: tuple[int, ...] = () + table_offset: int = 0 + + +@dataclass(frozen=True) +class StfsMutationResult: + source: Path + output: Path + operation: str + changed_paths: tuple[str, ...] + rehashed_blocks: int + signed: bool + sha256: str @dataclass(frozen=True) @@ -108,6 +120,7 @@ class XexPackage: "StfsEntry", "StfsHashRecord", "StfsIntegrityReport", + "StfsMutationResult", "StfsPackage", "XbePackage", "XexPackage", diff --git a/unityscraper/domains/packages/mutations.py b/unityscraper/domains/packages/mutations.py new file mode 100644 index 0000000..8016c19 --- /dev/null +++ b/unityscraper/domains/packages/mutations.py @@ -0,0 +1,245 @@ +"""Transactional STFS mutation and user-supplied signing interfaces.""" + +from __future__ import annotations + +import hashlib +import os +import shutil +import tempfile +from pathlib import Path +from typing import Callable, Mapping + +from .errors import InvalidPackageError +from .models import StfsMutationResult +from .stfs import ( + BLOCK_SIZE, + LEVEL0_BLOCKS, + LEVEL1_BLOCKS, + StfsLayout, + list_stfs_entries, + read_stfs_layout, +) + +StfsSigner = Callable[[str, bytes], bytes] + +METADATA_FIELDS: Mapping[str, tuple[int, int, str]] = { + "display_name": (0x411, 0x100, "utf-16-be"), + "title_name": (0x1691, 0x100, "utf-16-be"), + "publisher": (0x1611, 0x80, "utf-16-be"), +} + + +def replace_stfs_file( + package: str | Path, + internal_path: str, + replacement: str | Path, + *, + output: str | Path | None = None, + signer: StfsSigner | None = None, +) -> StfsMutationResult: + """Replace one file without reallocating its existing STFS block chain.""" + source = Path(package).expanduser().resolve() + incoming = Path(replacement).expanduser().resolve() + if not incoming.is_file(): + raise FileNotFoundError(incoming) + normalized = internal_path.replace("\\", "/").strip("/") + entry = next( + (item for item in list_stfs_entries(source) if item.path == normalized), + None, + ) + if entry is None or entry.is_directory: + raise InvalidPackageError(f"STFS file was not found: {normalized}") + replacement_size = incoming.stat().st_size + if replacement_size > entry.allocated_blocks * BLOCK_SIZE: + raise InvalidPackageError( + "Replacement exceeds the file's current allocation; package growth is not yet safe" + ) + + def mutate(working: Path) -> int: + layout = read_stfs_layout(working) + with incoming.open("rb") as source_handle, working.open("r+b") as handle: + remaining = replacement_size + for block in entry.blocks: + chunk = source_handle.read(min(remaining, BLOCK_SIZE)) if remaining else b"" + handle.seek(layout.data_offset(block)) + handle.write(chunk.ljust(BLOCK_SIZE, b"\0")) + remaining -= len(chunk) + if remaining: + raise InvalidPackageError("Replacement data did not fit its declared allocation") + handle.seek(entry.table_offset + 0x34) + handle.write(replacement_size.to_bytes(4, "big")) + return _rehash_in_place(working, signer) + + target, rehashed, signed = _transaction(source, output, mutate, signer is not None) + return StfsMutationResult( + source, target, "replace", (normalized,), rehashed, signed, _sha256(target) + ) + + +def edit_stfs_metadata( + package: str | Path, + updates: Mapping[str, str], + *, + output: str | Path | None = None, + signer: StfsSigner | None = None, +) -> StfsMutationResult: + """Edit bounded public text fields and rebuild package integrity data.""" + source = Path(package).expanduser().resolve() + unknown = set(updates) - set(METADATA_FIELDS) + if unknown: + raise InvalidPackageError(f"Unsupported STFS metadata fields: {', '.join(sorted(unknown))}") + + def mutate(working: Path) -> int: + read_stfs_layout(working) + with working.open("r+b") as handle: + for field, value in updates.items(): + offset, width, encoding = METADATA_FIELDS[field] + encoded = value.encode(encoding) + if len(encoded) > width - 2: + raise InvalidPackageError(f"{field} exceeds {width // 2 - 1} characters") + handle.seek(offset) + handle.write(encoded.ljust(width, b"\0")) + return _rehash_in_place(working, signer) + + target, rehashed, signed = _transaction(source, output, mutate, signer is not None) + return StfsMutationResult( + source, + target, + "metadata", + tuple(sorted(updates)), + rehashed, + signed, + _sha256(target), + ) + + +def rehash_stfs( + package: str | Path, + *, + output: str | Path | None = None, + signer: StfsSigner | None = None, +) -> StfsMutationResult: + """Rebuild the STFS hash tree and optionally invoke a caller-owned signer.""" + source = Path(package).expanduser().resolve() + + def mutate(working: Path) -> int: + return _rehash_in_place(working, signer) + + target, rehashed, signed = _transaction(source, output, mutate, signer is not None) + return StfsMutationResult(source, target, "rehash", (), rehashed, signed, _sha256(target)) + + +def _transaction( + source: Path, + output: str | Path | None, + mutation: Callable[[Path], int], + signed: bool, +) -> tuple[Path, int, bool]: + if not source.is_file(): + raise FileNotFoundError(source) + target = Path(output).expanduser().resolve() if output else source + target.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{target.name}.", suffix=".partial", dir=target.parent + ) + os.close(descriptor) + temporary = Path(temporary_name) + try: + shutil.copy2(source, temporary) + rehashed = mutation(temporary) + os.replace(temporary, target) + finally: + temporary.unlink(missing_ok=True) + return target, rehashed, signed + + +def _rehash_in_place(path: Path, signer: StfsSigner | None) -> int: + layout = read_stfs_layout(path) + with path.open("r+b") as handle: + for block in range(layout.block_count): + handle.seek(layout.data_offset(block)) + digest = hashlib.sha1(handle.read(BLOCK_SIZE)).digest() + record_offset = _record_offset(layout, handle, block, 0) + handle.seek(record_offset) + handle.write(digest) + + if layout.block_count > LEVEL0_BLOCKS: + groups = (layout.block_count + LEVEL0_BLOCKS - 1) // LEVEL0_BLOCKS + for group in range(groups): + block = group * LEVEL0_BLOCKS + table_offset = _active_table_offset(layout, handle, block, 0) + handle.seek(table_offset) + digest = hashlib.sha1(handle.read(BLOCK_SIZE)).digest() + handle.seek(_record_offset(layout, handle, block, 1)) + handle.write(digest) + + if layout.block_count > LEVEL1_BLOCKS: + groups = (layout.block_count + LEVEL1_BLOCKS - 1) // LEVEL1_BLOCKS + for group in range(groups): + block = group * LEVEL1_BLOCKS + table_offset = _active_table_offset(layout, handle, block, 1) + handle.seek(table_offset) + digest = hashlib.sha1(handle.read(BLOCK_SIZE)).digest() + handle.seek(_record_offset(layout, handle, block, 2)) + handle.write(digest) + + top_level = 0 if layout.block_count <= LEVEL0_BLOCKS else 1 + if layout.block_count > LEVEL1_BLOCKS: + top_level = 2 + top_offset = _active_table_offset(layout, handle, 0, top_level) + master_digest = hashlib.sha1(_read_block(handle, top_offset)).digest() + handle.seek(0x381) + handle.write(master_digest) + + header_hash_size = 0x9CBC if layout.base_offset == 0xA000 else 0xACBC + header_digest = hashlib.sha1(_read_range(handle, 0x344, header_hash_size)).digest() + handle.seek(0x32C) + handle.write(header_digest) + if signer is not None: + signing_digest = hashlib.sha1(_read_range(handle, 0x22C, 0x118)).digest() + signature = signer(layout.magic.decode("ascii").strip(), signing_digest) + expected = 0x80 if layout.magic == b"CON " else 0x100 + if len(signature) != expected: + raise InvalidPackageError( + f"Signer returned {len(signature)} bytes; {expected} are required" + ) + handle.seek(0x1AC if layout.magic == b"CON " else 4) + handle.write(signature) + return layout.block_count + + +def _record_offset(layout: StfsLayout, handle, block: int, level: int) -> int: + return layout.hash_record(handle, block, level).offset + + +def _active_table_offset(layout: StfsLayout, handle, block: int, level: int) -> int: + return layout.active_hash_table_offset(handle, block, level) + + +def _read_block(handle, offset: int) -> bytes: + return _read_range(handle, offset, BLOCK_SIZE) + + +def _read_range(handle, offset: int, size: int) -> bytes: + handle.seek(offset) + value = handle.read(size) + if len(value) != size: + raise InvalidPackageError("STFS package is truncated during rehashing") + return value + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +__all__ = [ + "METADATA_FIELDS", + "StfsSigner", + "edit_stfs_metadata", + "rehash_stfs", + "replace_stfs_file", +] diff --git a/unityscraper/domains/packages/service.py b/unityscraper/domains/packages/service.py index fea9a0f..6f83b15 100644 --- a/unityscraper/domains/packages/service.py +++ b/unityscraper/domains/packages/service.py @@ -11,21 +11,47 @@ list_stfs_entries, verify_stfs, ) +from .gdf import GdfEntry, GdfImage, extract_gdf, inspect_gdf +from .fatx import ( + FatxEntry, + FatxImage, + FatxPartition, + extract_fatx, + inspect_fatx, + replace_fatx_file, +) from .models import ( StfsBlockVerification, StfsEntry, StfsHashRecord, StfsIntegrityReport, + StfsMutationResult, StfsPackage, XbePackage, XexPackage, ) +from .mutations import edit_stfs_metadata, rehash_stfs, replace_stfs_file +from .svod import ( + SvodIntegrityReport, + SvodPackage, + extract_svod_payload, + inspect_svod, + verify_svod, +) __all__ = [ "StfsEntry", + "GdfEntry", + "GdfImage", + "FatxEntry", + "FatxImage", + "FatxPartition", "StfsBlockVerification", "StfsHashRecord", "StfsIntegrityReport", + "StfsMutationResult", + "SvodIntegrityReport", + "SvodPackage", "StfsPackage", "XbePackage", "XexPackage", @@ -38,4 +64,15 @@ "inspect_xex", "list_stfs_entries", "verify_stfs", + "edit_stfs_metadata", + "rehash_stfs", + "replace_stfs_file", + "extract_gdf", + "inspect_gdf", + "extract_fatx", + "inspect_fatx", + "replace_fatx_file", + "extract_svod_payload", + "inspect_svod", + "verify_svod", ] diff --git a/unityscraper/domains/packages/stfs.py b/unityscraper/domains/packages/stfs.py index 64c4013..41be195 100644 --- a/unityscraper/domains/packages/stfs.py +++ b/unityscraper/domains/packages/stfs.py @@ -116,6 +116,14 @@ def base_hash_offset(self, block: int, level: int) -> int: ) return self.base_offset + self.base_hash_block(block, level) * BLOCK_SIZE + entry * 0x18 + def active_hash_table_offset(self, handle: BinaryIO, block: int, level: int) -> int: + table_index = self._active_table_index(handle, block, level) + return ( + self.base_offset + + self.base_hash_block(block, level) * BLOCK_SIZE + + table_index * BLOCK_SIZE + ) + def hash_record(self, handle: BinaryIO, block: int, level: int = 0) -> StfsHashRecord: table_index = self._active_table_index(handle, block, level) offset = self.base_hash_offset(block, level) + table_index * BLOCK_SIZE @@ -214,14 +222,23 @@ def read_stfs_layout(path: str | Path) -> StfsLayout: top_index = (separation >> 1) & 1 table_start = int.from_bytes(header[0x37E:0x381], "little") count_raw = header[0x37C:0x37E] - candidates = tuple(dict.fromkeys((int.from_bytes(count_raw, "little"), int.from_bytes(count_raw, "big")))) + candidates = tuple( + dict.fromkeys((int.from_bytes(count_raw, "little"), int.from_bytes(count_raw, "big"))) + ) table_blocks = 0 for candidate in candidates: if not 0 < candidate <= 0x3FF: continue layout = StfsLayout( - header[:4], header_size, separation, candidate, table_start, - block_count, package_size, shift, top_index, + header[:4], + header_size, + separation, + candidate, + table_start, + block_count, + package_size, + shift, + top_index, ) try: if layout.data_offset(table_start) + BLOCK_SIZE <= package_size: @@ -232,8 +249,15 @@ def read_stfs_layout(path: str | Path) -> StfsLayout: if not table_blocks: raise InvalidPackageError("STFS file-table size is invalid") return StfsLayout( - header[:4], header_size, separation, table_blocks, table_start, - block_count, package_size, shift, top_index, + header[:4], + header_size, + separation, + table_blocks, + table_start, + block_count, + package_size, + shift, + top_index, ) @@ -287,24 +311,29 @@ def list_stfs_entries(path: str | Path, max_entries: int = 100_000) -> list[Stfs ) raw_entries: list[dict[str, Any]] = [] for logical_block in table_chain: - table = _read_exact(handle, layout.data_offset(logical_block), BLOCK_SIZE, layout.package_size) + table = _read_exact( + handle, layout.data_offset(logical_block), BLOCK_SIZE, layout.package_size + ) for entry_offset in range(0, BLOCK_SIZE, 0x40): - data = table[entry_offset:entry_offset + 0x40] + data = table[entry_offset : entry_offset + 0x40] if not any(data): continue flags = data[0x28] name_length = flags & 0x3F if name_length == 0 or name_length > 0x28: continue - raw_entries.append({ - "name": data[:name_length].decode("utf-8", errors="replace"), - "directory": bool(flags & 0x80), - "consecutive": bool(flags & 0x40), - "blocks": int.from_bytes(data[0x29:0x2C], "little"), - "start": int.from_bytes(data[0x2F:0x32], "little"), - "parent": int.from_bytes(data[0x32:0x34], "big"), - "size": int.from_bytes(data[0x34:0x38], "big"), - }) + raw_entries.append( + { + "name": data[:name_length].decode("utf-8", errors="replace"), + "directory": bool(flags & 0x80), + "consecutive": bool(flags & 0x40), + "blocks": int.from_bytes(data[0x29:0x2C], "little"), + "start": int.from_bytes(data[0x2F:0x32], "little"), + "parent": int.from_bytes(data[0x32:0x34], "big"), + "size": int.from_bytes(data[0x34:0x38], "big"), + "table_offset": layout.data_offset(logical_block) + entry_offset, + } + ) if len(raw_entries) > max_entries: raise InvalidPackageError("STFS file table exceeds the safety limit") @@ -323,24 +352,31 @@ def list_stfs_entries(path: str | Path, max_entries: int = 100_000) -> list[Stfs full_path = "/".join(reversed(ancestors)) full_path = f"{full_path}/{row['name']}" if full_path else row["name"] block_count = row["blocks"] if not row["directory"] else 0 - blocks = layout.block_chain( - handle, - row["start"], - block_count, - consecutive=row["consecutive"], - ) if block_count else () - entries.append(StfsEntry( - index=index, - path=full_path, - name=row["name"], - is_directory=row["directory"], - consecutive=row["consecutive"], - allocated_blocks=row["blocks"], - starting_block=row["start"], - parent_index=row["parent"], - size=row["size"], - blocks=blocks, - )) + blocks = ( + layout.block_chain( + handle, + row["start"], + block_count, + consecutive=row["consecutive"], + ) + if block_count + else () + ) + entries.append( + StfsEntry( + index=index, + path=full_path, + name=row["name"], + is_directory=row["directory"], + consecutive=row["consecutive"], + allocated_blocks=row["blocks"], + starting_block=row["start"], + parent_index=row["parent"], + size=row["size"], + blocks=blocks, + table_offset=row["table_offset"], + ) + ) return entries @@ -354,7 +390,9 @@ def verify_stfs(path: str | Path, *, max_issues: int = 10_000) -> StfsIntegrityR checked += 1 try: record = layout.hash_record(handle, block) - data = _read_exact(handle, layout.data_offset(block), BLOCK_SIZE, layout.package_size) + data = _read_exact( + handle, layout.data_offset(block), BLOCK_SIZE, layout.package_size + ) except InvalidPackageError as exc: mismatched += 1 if len(issues) < max_issues: @@ -364,17 +402,27 @@ def verify_stfs(path: str | Path, *, max_issues: int = 10_000) -> StfsIntegrityR if not record.stored_sha1 or set(record.stored_sha1) == {"0"}: unverifiable += 1 if len(issues) < max_issues: - issues.append(StfsBlockVerification( - block, "missing", record.stored_sha1, calculated, - "Hash record is empty", - )) + issues.append( + StfsBlockVerification( + block, + "missing", + record.stored_sha1, + calculated, + "Hash record is empty", + ) + ) elif record.stored_sha1.lower() != calculated: mismatched += 1 if len(issues) < max_issues: - issues.append(StfsBlockVerification( - block, "mismatch", record.stored_sha1, calculated, - "Stored SHA-1 does not match the data block", - )) + issues.append( + StfsBlockVerification( + block, + "mismatch", + record.stored_sha1, + calculated, + "Stored SHA-1 does not match the data block", + ) + ) else: valid += 1 return StfsIntegrityReport( @@ -396,7 +444,8 @@ def extract_stfs_files( requested = {item.replace("\\", "/") for item in selected_paths or ()} entries = list_stfs_entries(package) files = [ - entry for entry in entries + entry + for entry in entries if not entry.is_directory and (not requested or entry.path in requested) ] if requested - {entry.path for entry in files}: @@ -426,7 +475,9 @@ def extract_stfs_files( with partial.open("xb") as destination_handle: for block in entry.blocks: size = min(remaining, BLOCK_SIZE) - chunk = _read_exact(handle, layout.data_offset(block), size, layout.package_size) + chunk = _read_exact( + handle, layout.data_offset(block), size, layout.package_size + ) destination_handle.write(chunk) digest.update(chunk) remaining -= size @@ -435,23 +486,31 @@ def extract_stfs_files( partial.replace(output) finally: partial.unlink(missing_ok=True) - extracted.append({ - "path": entry.path, - "output": str(output), - "size": entry.size, - "sha256": digest.hexdigest(), - "blocks": list(entry.blocks), - }) + extracted.append( + { + "path": entry.path, + "output": str(output), + "size": entry.size, + "sha256": digest.hexdigest(), + "blocks": list(entry.blocks), + } + ) manifest = target / "unityscraper-stfs-extraction.json" - manifest.write_text(json.dumps({ - "schema": 2, - "source": str(package), - "source_sha256": _sha256_file(package), - "read_only": True, - "supports_fragmented_files": True, - "extracted": extracted, - "skipped": skipped, - }, indent=2), encoding="utf-8") + manifest.write_text( + json.dumps( + { + "schema": 2, + "source": str(package), + "source_sha256": _sha256_file(package), + "read_only": True, + "supports_fragmented_files": True, + "extracted": extracted, + "skipped": skipped, + }, + indent=2, + ), + encoding="utf-8", + ) return {"manifest": str(manifest), "extracted": extracted, "skipped": skipped} diff --git a/unityscraper/domains/packages/svod.py b/unityscraper/domains/packages/svod.py new file mode 100644 index 0000000..da1d8fd --- /dev/null +++ b/unityscraper/domains/packages/svod.py @@ -0,0 +1,196 @@ +"""SVOD/Games-on-Demand inspection, verification, and payload extraction.""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from pathlib import Path +from typing import BinaryIO + +from .errors import InvalidPackageError +from .stfs import STFS_MAGICS, inspect_stfs + +BLOCK_SIZE = 0x1000 +BLOCKS_PER_FILE = 0xA1C4 +HASHES_PER_TABLE = 0xCC +MAX_DATA_FILES = 9999 + + +@dataclass(frozen=True) +class SvodPackage: + header_path: Path + data_directory: Path | None + magic: str + title_id: str + display_name: str + block_count: int + data_file_count: int + data_size: int + shifted: bool + deviation: int + data_files: tuple[Path, ...] + + +@dataclass(frozen=True) +class SvodIntegrityReport: + checked_blocks: int + valid_blocks: int + mismatched_blocks: int + missing_blocks: int + + @property + def valid(self) -> bool: + return self.mismatched_blocks == 0 and self.missing_blocks == 0 + + +def inspect_svod( + header_path: str | Path, + data_directory: str | Path | None = None, +) -> SvodPackage: + source = Path(header_path).expanduser().resolve() + with source.open("rb") as handle: + header = handle.read(0x3AD) + if len(header) < 0x3AD or header[:4] not in STFS_MAGICS: + raise InvalidPackageError("Not a supported SVOD header") + if header[0x379:0x37D] != b"\x24\x05\x05\x11": + raise InvalidPackageError("Package does not contain an SVOD descriptor") + shifted = bool(header[0x391] & 0x40) + block_count = int.from_bytes(header[0x392:0x395], "big") + deviation = int.from_bytes(header[0x395:0x399], "little") if shifted else 0 + data_file_count = int.from_bytes(header[0x39D:0x3A1], "big") + data_size = int.from_bytes(header[0x3A1:0x3A9], "big") + if block_count <= 0: + raise InvalidPackageError("SVOD block count is invalid") + if data_file_count <= 0 or data_file_count > MAX_DATA_FILES: + raise InvalidPackageError("SVOD data-file count is invalid") + directory = _resolve_data_directory(source, data_directory) + data_files: tuple[Path, ...] = () + if directory is not None: + files = tuple(directory / f"Data{index:04d}" for index in range(data_file_count)) + missing = [item.name for item in files if not item.is_file()] + if missing: + raise InvalidPackageError(f"SVOD data files are missing: {', '.join(missing[:5])}") + data_files = files + metadata = inspect_stfs(source) + return SvodPackage( + source, + directory, + metadata.magic, + metadata.title_id, + metadata.display_name or metadata.title_name, + block_count, + data_file_count, + data_size, + shifted, + deviation, + data_files, + ) + + +def verify_svod( + header_path: str | Path, + data_directory: str | Path | None = None, +) -> SvodIntegrityReport: + package = inspect_svod(header_path, data_directory) + if not package.data_files: + raise InvalidPackageError("Choose the SVOD data directory to verify its payload") + checked = valid = mismatched = missing = 0 + handles = [path.open("rb") for path in package.data_files] + try: + for block in range(package.block_count): + checked += 1 + handle = handles[block // BLOCKS_PER_FILE] + try: + stored = _read_exact(handle, _hash_offset(block, 0), 0x14) + data = _read_exact(handle, _data_offset(block), BLOCK_SIZE) + except InvalidPackageError: + missing += 1 + continue + if hashlib.sha1(data).digest() == stored: + valid += 1 + else: + mismatched += 1 + finally: + for handle in handles: + handle.close() + return SvodIntegrityReport(checked, valid, mismatched, missing) + + +def extract_svod_payload( + header_path: str | Path, + destination: str | Path, + data_directory: str | Path | None = None, +) -> Path: + package = inspect_svod(header_path, data_directory) + if not package.data_files: + raise InvalidPackageError("Choose the SVOD data directory to extract its payload") + target = Path(destination).expanduser().resolve() + target.parent.mkdir(parents=True, exist_ok=True) + temporary = target.with_name(target.name + ".partial") + handles = [path.open("rb") for path in package.data_files] + try: + with temporary.open("xb") as output: + remaining = package.data_size or package.block_count * BLOCK_SIZE + for block in range(package.block_count): + handle = handles[block // BLOCKS_PER_FILE] + chunk = _read_exact(handle, _data_offset(block), BLOCK_SIZE) + write_size = min(remaining, BLOCK_SIZE) + output.write(chunk[:write_size]) + remaining -= write_size + if remaining <= 0: + break + if target.exists(): + raise FileExistsError(target) + temporary.replace(target) + finally: + for handle in handles: + handle.close() + temporary.unlink(missing_ok=True) + return target + + +def _resolve_data_directory(source: Path, value: str | Path | None) -> Path | None: + if value is not None: + directory = Path(value).expanduser().resolve() + if not directory.is_dir(): + raise FileNotFoundError(directory) + return directory + for candidate in ( + source.with_name(source.name + ".data"), + source.parent / f"{source.stem}.data", + source.parent / "data", + ): + if candidate.is_dir(): + return candidate.resolve() + return None + + +def _data_offset(block: int) -> int: + local = block % BLOCKS_PER_FILE + return 0x2000 + BLOCK_SIZE * local + BLOCK_SIZE * (local // HASHES_PER_TABLE) + + +def _hash_offset(block: int, level: int) -> int: + local = block % BLOCKS_PER_FILE + if level == 0: + return 0x1000 + 0xCD000 * (local // HASHES_PER_TABLE) + 0x14 * (local % HASHES_PER_TABLE) + if level == 1: + return 0x14 * (local // HASHES_PER_TABLE) + raise InvalidPackageError("Unsupported SVOD hash-tree level") + + +def _read_exact(handle: BinaryIO, offset: int, size: int) -> bytes: + handle.seek(offset) + value = handle.read(size) + if len(value) != size: + raise InvalidPackageError("SVOD data file is truncated") + return value + + +__all__ = [ + "SvodIntegrityReport", + "SvodPackage", + "extract_svod_payload", + "inspect_svod", + "verify_svod", +] diff --git a/unityscraper/domains/profiles/__init__.py b/unityscraper/domains/profiles/__init__.py index e1b5fde..f6c097b 100644 --- a/unityscraper/domains/profiles/__init__.py +++ b/unityscraper/domains/profiles/__init__.py @@ -3,6 +3,13 @@ from __future__ import annotations from .models import ProfileInfo, ProfileScanResult, RestoreResult, SaveInfo +from .gpd import ( + export_gpd_image, + parse_gpd, + parse_gpd_bytes, + set_gpd_achievement_state, + update_gpd_setting, +) from .operations import find_content_root, mask_identifier from .service import ( ProfileSaveConflict, @@ -22,4 +29,9 @@ "SaveInfo", "find_content_root", "mask_identifier", + "export_gpd_image", + "parse_gpd", + "parse_gpd_bytes", + "set_gpd_achievement_state", + "update_gpd_setting", ] diff --git a/unityscraper/domains/profiles/gpd.py b/unityscraper/domains/profiles/gpd.py new file mode 100644 index 0000000..e04299f --- /dev/null +++ b/unityscraper/domains/profiles/gpd.py @@ -0,0 +1,206 @@ +"""GPD/XDBF inspection exports and transactional record editing.""" + +from __future__ import annotations + +import os +import shutil +import struct +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from gpd_parser import ( + GpdAchievement, + GpdError, + GpdImage, + GpdReport, + GpdSetting, + GpdTitleHistory, + XDBF_ENTRY, + XDBF_FREE_ENTRY_SIZE, + XDBF_HEADER, + export_gpd_image, + parse_gpd, + parse_gpd_bytes, +) + +SETTING_TYPE_IDS = { + "context": 0, + "uint32": 1, + "int64": 2, + "double": 3, + "unicode": 4, + "float": 5, + "binary": 6, + "datetime": 7, + "null": 0xFF, +} + + +def update_gpd_setting( + source: str | Path, + setting_id: int, + value: Any, + *, + output: str | Path, +) -> GpdReport: + """Update one existing setting without resizing its XDBF allocation.""" + path = Path(source).expanduser().resolve() + + def mutate(data: bytearray) -> None: + offset, size = _find_entry(data, 3, setting_id) + if size < 0x18: + raise GpdError("Setting record is too small") + type_id = data[offset + 8] + area = offset + 16 + if type_id in {0, 1}: + data[area : area + 4] = int(value).to_bytes(4, "big", signed=False) + elif type_id == 2: + data[area : area + 8] = int(value).to_bytes(8, "big", signed=True) + elif type_id == 3: + data[area : area + 8] = struct.pack(">d", float(value)) + elif type_id == 5: + data[area : area + 4] = struct.pack(">f", float(value)) + elif type_id == 7: + filetime = _to_filetime(value) + data[area : area + 8] = filetime.to_bytes(8, "big", signed=True) + elif type_id in {4, 6}: + encoded = ( + str(value).encode("utf-16-be") + b"\0\0" + if type_id == 4 + else bytes.fromhex(str(value)) + ) + capacity = size - 0x18 + if len(encoded) > capacity: + raise GpdError("Replacement setting exceeds its existing allocation") + data[area : area + 4] = len(encoded).to_bytes(4, "big", signed=True) + data[offset + 0x18 : offset + size] = encoded.ljust(capacity, b"\0") + elif type_id == 0xFF: + raise GpdError("Null settings do not contain editable storage") + else: + raise GpdError(f"Unsupported setting type: {type_id}") + + target = _mutate_file(path, output, mutate) + return parse_gpd(target) + + +def set_gpd_achievement_state( + source: str | Path, + achievement_id: int, + state: str, + *, + output: str | Path, + unlocked_at: datetime | None = None, +) -> GpdReport: + """Set an existing achievement state in a separate GPD output.""" + states = {"locked": 0, "unlocked-offline": 0x12, "unlocked-online": 0x13} + if state not in states: + raise GpdError(f"Unsupported achievement state: {state}") + path = Path(source).expanduser().resolve() + + def mutate(data: bytearray) -> None: + offset, size = _find_achievement(data, achievement_id) + if size < 0x1C: + raise GpdError("Achievement record is too small") + data[offset + 17] = states[state] + timestamp = unlocked_at if state == "unlocked-online" else None + filetime = _to_filetime(timestamp or datetime.now(timezone.utc)) if timestamp else 0 + data[offset + 20 : offset + 28] = filetime.to_bytes(8, "big", signed=True) + + target = _mutate_file(path, output, mutate) + return parse_gpd(target) + + +def _find_achievement(data: bytearray, achievement_id: int) -> tuple[int, int]: + for offset, size in _entries(data, 1): + if ( + size >= 8 + and int.from_bytes(data[offset + 4 : offset + 8], "big", signed=True) == achievement_id + ): + return offset, size + raise KeyError(f"Achievement {achievement_id} was not found") + + +def _find_entry(data: bytearray, namespace: int, entry_id: int) -> tuple[int, int]: + for table_id, offset, size in _entries_with_ids(data, namespace): + if table_id == entry_id: + return offset, size + if size >= 4 and int.from_bytes(data[offset : offset + 4], "big", signed=True) == entry_id: + return offset, size + raise KeyError(f"XDBF entry {entry_id} was not found") + + +def _entries(data: bytearray, namespace: int): + for _entry_id, offset, size in _entries_with_ids(data, namespace): + yield offset, size + + +def _entries_with_ids(data: bytearray, namespace: int): + if len(data) < XDBF_HEADER.size: + raise GpdError("GPD is smaller than the XDBF header") + magic, _version, entry_max, entry_count, free_max, _free_count = XDBF_HEADER.unpack_from(data) + if magic != b"XDBF" or entry_count > entry_max: + raise GpdError("GPD has an invalid XDBF header") + header_size = XDBF_HEADER.size + entry_max * XDBF_ENTRY.size + free_max * XDBF_FREE_ENTRY_SIZE + if header_size > len(data): + raise GpdError("XDBF table extends beyond the file") + for index in range(entry_count): + entry_namespace, entry_id, relative, size = XDBF_ENTRY.unpack_from( + data, XDBF_HEADER.size + index * XDBF_ENTRY.size + ) + absolute = header_size + relative + if relative < 0 or size < 0 or absolute + size > len(data): + raise GpdError("XDBF entry extends beyond the file") + if entry_namespace == namespace: + yield entry_id, absolute, size + + +def _mutate_file(source: Path, output: str | Path, mutation) -> Path: + target = Path(output).expanduser().resolve() + if target == source: + raise GpdError("GPD edits require a separate output file") + target.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{target.name}.", suffix=".partial", dir=target.parent + ) + os.close(descriptor) + temporary = Path(temporary_name) + try: + shutil.copy2(source, temporary) + data = bytearray(temporary.read_bytes()) + mutation(data) + temporary.write_bytes(data) + parse_gpd(temporary) + if target.exists(): + raise FileExistsError(target) + os.replace(temporary, target) + finally: + temporary.unlink(missing_ok=True) + return target + + +def _to_filetime(value: Any) -> int: + if isinstance(value, str): + value = datetime.fromisoformat(value.replace("Z", "+00:00")) + if not isinstance(value, datetime): + raise GpdError("Datetime settings require an ISO timestamp or datetime") + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + epoch = datetime(1601, 1, 1, tzinfo=timezone.utc) + return int((value.astimezone(timezone.utc) - epoch).total_seconds() * 10_000_000) + + +__all__ = [ + "GpdAchievement", + "GpdError", + "GpdImage", + "GpdReport", + "GpdSetting", + "GpdTitleHistory", + "export_gpd_image", + "parse_gpd", + "parse_gpd_bytes", + "set_gpd_achievement_state", + "update_gpd_setting", +] diff --git a/unityscraper/domains/profiles/service.py b/unityscraper/domains/profiles/service.py index abb57a7..a58fe61 100644 --- a/unityscraper/domains/profiles/service.py +++ b/unityscraper/domains/profiles/service.py @@ -11,6 +11,13 @@ from .models import ProfileInfo, ProfileScanResult, RestoreResult, SaveInfo from .operations import find_content_root, mask_identifier +from .gpd import ( + export_gpd_image, + parse_gpd, + parse_gpd_bytes, + set_gpd_achievement_state, + update_gpd_setting, +) __all__ = [ "ProfileInfo", @@ -23,4 +30,9 @@ "SaveInfo", "find_content_root", "mask_identifier", + "export_gpd_image", + "parse_gpd", + "parse_gpd_bytes", + "set_gpd_achievement_state", + "update_gpd_setting", ] From 27c98e72bd12fe987e7d07a6b713040cbeea0a0b Mon Sep 17 00:00:00 2001 From: Sthornberry9 <46094434+Sthornberry9@users.noreply.github.com> Date: Tue, 11 Aug 2026 07:47:05 -0400 Subject: [PATCH 3/3] fix: honor STFS descriptor type and FATX partition paths --- tests.py | 56 ++++++++++++++------------- unityscraper/domains/packages/fatx.py | 11 +++++- unityscraper/domains/packages/stfs.py | 14 ++++--- 3 files changed, 48 insertions(+), 33 deletions(-) diff --git a/tests.py b/tests.py index f3708f2..8af587d 100644 --- a/tests.py +++ b/tests.py @@ -1597,17 +1597,17 @@ def test_inspect_stfs_reads_profile_ownership_fields(self): self.assertEqual(package.save_game_id, "12345678") def test_stfs_file_table_is_inventoried_read_only(self): - payload = bytearray(0xD000) + payload = bytearray(0xE000) payload[:4] = b"LIVE" - payload[0x340:0x344] = (0xA000).to_bytes(4, "big") + payload[0x340:0x344] = (0xB000).to_bytes(4, "big") payload[0x344:0x348] = (1).to_bytes(4, "big") payload[0x360:0x364] = bytes.fromhex("53510804") payload[0x379] = 0x24 - payload[0x37B] = 1 + payload[0x37B] = 0 payload[0x37C:0x37E] = (1).to_bytes(2, "little") payload[0x395:0x399] = (2).to_bytes(4, "big") name = b"savegame.dat" - entry = 0xB000 + entry = 0xC000 payload[entry:entry + len(name)] = name payload[entry + 0x28] = len(name) | 0x40 payload[entry + 0x29:entry + 0x2C] = (1).to_bytes(3, "little") @@ -1622,7 +1622,7 @@ def test_stfs_file_table_is_inventoried_read_only(self): self.assertEqual(entries[0].size, 123) self.assertTrue(entries[0].consecutive) - payload[0xC000:0xC004] = b"data" + payload[0xD000:0xD004] = b"data" package_path.write_bytes(payload) destination = self.temp_dir / "extracted" result = extract_stfs_files(package_path, destination) @@ -1631,29 +1631,29 @@ def test_stfs_file_table_is_inventoried_read_only(self): self.assertTrue(Path(result["manifest"]).is_file()) def test_fragmented_stfs_extraction_and_integrity_verification(self): - payload = bytearray(0xF000) + payload = bytearray(0x11000) payload[:4] = b"LIVE" - payload[0x340:0x344] = (0xA000).to_bytes(4, "big") + payload[0x340:0x344] = (0xB000).to_bytes(4, "big") payload[0x344:0x348] = (1).to_bytes(4, "big") payload[0x360:0x364] = bytes.fromhex("53510804") payload[0x379] = 0x24 - payload[0x37B] = 1 + payload[0x37B] = 0 payload[0x37C:0x37E] = (1).to_bytes(2, "little") payload[0x395:0x399] = (4).to_bytes(4, "big") name = b"fragmented.bin" - entry = 0xB000 + entry = 0xC000 payload[entry:entry + len(name)] = name payload[entry + 0x28] = len(name) payload[entry + 0x29:entry + 0x2C] = (2).to_bytes(3, "little") payload[entry + 0x2F:entry + 0x32] = (1).to_bytes(3, "little") payload[entry + 0x32:entry + 0x34] = (0xFFFF).to_bytes(2, "big") payload[entry + 0x34:entry + 0x38] = (0x1004).to_bytes(4, "big") - payload[0xC000:0xD000] = b"A" * 0x1000 - payload[0xE000:0xE004] = b"tail" + payload[0xD000:0xE000] = b"A" * 0x1000 + payload[0xF000:0xF004] = b"tail" - for block, offset in enumerate((0xB000, 0xC000, 0xD000, 0xE000)): - record = 0xA000 + block * 0x18 + for block, offset in enumerate((0xC000, 0xD000, 0xE000, 0xF000)): + record = 0xB000 + block * 0x18 payload[record:record + 0x14] = hashlib.sha1( payload[offset:offset + 0x1000] ).digest() @@ -1675,7 +1675,7 @@ def test_fragmented_stfs_extraction_and_integrity_verification(self): ) self.assertTrue(verify_stfs(package_path).valid) - payload[0xC000] ^= 0xFF + payload[0xD000] ^= 0xFF package_path.write_bytes(payload) report = verify_stfs(package_path) self.assertFalse(report.valid) @@ -2739,16 +2739,16 @@ def test_fragmented_stfs_round_trip_and_tamper_detection(self): verify_stfs, ) - payload = bytearray(0xF000) + payload = bytearray(0x11000) payload[:4] = b"LIVE" - payload[0x340:0x344] = (0xA000).to_bytes(4, "big") + payload[0x340:0x344] = (0xB000).to_bytes(4, "big") payload[0x344:0x348] = (1).to_bytes(4, "big") payload[0x360:0x364] = bytes.fromhex("53510804") payload[0x379] = 0x24 - payload[0x37B] = 1 + payload[0x37B] = 0 payload[0x37C:0x37E] = (1).to_bytes(2, "little") payload[0x395:0x399] = (4).to_bytes(4, "big") - entry = 0xB000 + entry = 0xC000 name = b"fragmented.bin" payload[entry:entry + len(name)] = name payload[entry + 0x28] = len(name) @@ -2756,10 +2756,10 @@ def test_fragmented_stfs_round_trip_and_tamper_detection(self): payload[entry + 0x2F:entry + 0x32] = (1).to_bytes(3, "little") payload[entry + 0x32:entry + 0x34] = (0xFFFF).to_bytes(2, "big") payload[entry + 0x34:entry + 0x38] = (0x1004).to_bytes(4, "big") - payload[0xC000:0xD000] = b"A" * 0x1000 - payload[0xE000:0xE004] = b"tail" - for block, offset in enumerate((0xB000, 0xC000, 0xD000, 0xE000)): - record = 0xA000 + block * 0x18 + payload[0xD000:0xE000] = b"A" * 0x1000 + payload[0xF000:0xF004] = b"tail" + for block, offset in enumerate((0xC000, 0xD000, 0xE000, 0xF000)): + record = 0xB000 + block * 0x18 payload[record:record + 0x14] = hashlib.sha1( payload[offset:offset + 0x1000] ).digest() @@ -2784,7 +2784,7 @@ def test_fragmented_stfs_round_trip_and_tamper_detection(self): self.assertEqual((output / "fragmented.bin").read_bytes(), b"replacement") self.assertTrue(verify_stfs(named).valid) changed = bytearray(named.read_bytes()) - changed[0xC000] ^= 0xFF + changed[0xD000] ^= 0xFF named.write_bytes(changed) self.assertFalse(verify_stfs(named).valid) @@ -2857,17 +2857,19 @@ def test_fatx_inventory_extraction_and_guarded_replacement(self): image[0x3000:0x3004] = b"FATX" source = self.temp_dir / "fatx.img" source.write_bytes(image) - self.assertEqual(inspect_fatx(source).entries[0].path, "game.bin") + self.assertEqual(inspect_fatx(source).entries[0].path, "Image/game.bin") output = self.temp_dir / "fatx-output" extract_fatx(source, output) - self.assertEqual((output / "game.bin").read_bytes(), b"FATX") + self.assertEqual((output / "Image" / "game.bin").read_bytes(), b"FATX") replacement = self.temp_dir / "replacement.bin" replacement.write_bytes(b"EDIT") edited = self.temp_dir / "edited.img" - replace_fatx_file(source, "game.bin", replacement, output=edited) + replace_fatx_file(source, "Image/game.bin", replacement, output=edited) edited_output = self.temp_dir / "fatx-edited-output" extract_fatx(edited, edited_output) - self.assertEqual((edited_output / "game.bin").read_bytes(), b"EDIT") + self.assertEqual( + (edited_output / "Image" / "game.bin").read_bytes(), b"EDIT" + ) def test_gpd_achievement_edit_is_transactional(self): import struct diff --git a/unityscraper/domains/packages/fatx.py b/unityscraper/domains/packages/fatx.py index cc89756..1bee5e4 100644 --- a/unityscraper/domains/packages/fatx.py +++ b/unityscraper/domains/packages/fatx.py @@ -78,7 +78,16 @@ def inspect_fatx(path: str | Path) -> FatxImage: ) partition = _read_partition(handle, name, offset, next_offset - offset) partitions.append(partition) - entries.extend(_read_directory(handle, partition, partition.root_block, "", 0, set())) + entries.extend( + _read_directory( + handle, + partition, + partition.root_block, + partition.name, + 0, + set(), + ) + ) if not partitions: raise InvalidPackageError("No supported FATX partitions were found") return FatxImage(source, tuple(partitions), tuple(entries)) diff --git a/unityscraper/domains/packages/stfs.py b/unityscraper/domains/packages/stfs.py index 41be195..b0bc281 100644 --- a/unityscraper/domains/packages/stfs.py +++ b/unityscraper/domains/packages/stfs.py @@ -217,8 +217,7 @@ def read_stfs_layout(path: str | Path) -> StfsLayout: block_count = int.from_bytes(header[0x395:0x399], "big") if block_count <= 0 or block_count >= MAX_BLOCKS: raise InvalidPackageError("STFS allocated block count is invalid") - aligned = (header_size + 0xFFF) & 0xFFFFF000 - shift = 0 if aligned == 0xB000 else 0 if separation & 1 else 1 + shift = separation & 1 top_index = (separation >> 1) & 1 table_start = int.from_bytes(header[0x37E:0x381], "little") count_raw = header[0x37C:0x37E] @@ -276,9 +275,8 @@ def inspect_stfs(path: str | Path) -> StfsPackage: raise InvalidPackageError("STFS package does not contain a usable TitleID") header_size = int.from_bytes(header[0x340:0x344], "big") block_count = int.from_bytes(header[0x395:0x399], "big") - aligned = (header_size + 0xFFF) & 0xFFFFF000 separation = header[0x37B] & 0x3 - structure_type = 0 if aligned == 0xB000 else 0 if separation & 1 else 1 + structure_type = separation & 1 return StfsPackage( path=package_path, magic=header[:4].decode("ascii").strip(), @@ -351,7 +349,13 @@ def list_stfs_entries(path: str | Path, max_entries: int = 100_000) -> list[Stfs parent = raw_entries[parent]["parent"] full_path = "/".join(reversed(ancestors)) full_path = f"{full_path}/{row['name']}" if full_path else row["name"] - block_count = row["blocks"] if not row["directory"] else 0 + block_count = (row["size"] + BLOCK_SIZE - 1) // BLOCK_SIZE + if not row["directory"] and block_count > row["blocks"]: + raise InvalidPackageError( + f"STFS entry exceeds its declared allocation: {full_path}" + ) + if row["directory"]: + block_count = 0 blocks = ( layout.block_chain( handle,