mirror of
https://github.com/rvtr/libWiiPy.git
synced 2025-06-18 10:45:35 -04:00
Merge pull request #20 from NinjaCheetah/ash-dec
Add support for decompressing ASH files
This commit is contained in:
commit
53f798e36a
1
.gitignore
vendored
1
.gitignore
vendored
@ -165,6 +165,7 @@ cython_debug/
|
||||
*.tmd
|
||||
*.wad
|
||||
*.arc
|
||||
*.ash
|
||||
out_prod/
|
||||
remakewad.pl
|
||||
|
||||
|
@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "libWiiPy"
|
||||
version = "0.3.1"
|
||||
version = "0.4.0"
|
||||
authors = [
|
||||
{ name="NinjaCheetah", email="ninjacheetah@ncxprogramming.com" },
|
||||
{ name="Lillian Skinner", email="lillian@randommeaninglesscharacters.com" }
|
||||
|
@ -1,4 +1,5 @@
|
||||
# "archive/__init__.py" from libWiiPy by NinjaCheetah & Contributors
|
||||
# https://github.com/NinjaCheetah/libWiiPy
|
||||
|
||||
from .ash import *
|
||||
from .u8 import *
|
||||
|
233
src/libWiiPy/archive/ash.py
Normal file
233
src/libWiiPy/archive/ash.py
Normal file
@ -0,0 +1,233 @@
|
||||
# "archive/ash.py" from libWiiPy by NinjaCheetah & Contributors
|
||||
# https://github.com/NinjaCheetah/libWiiPy
|
||||
#
|
||||
# This code in particular is a direct translation of "ash-dec" from ASH0-tools. ASH0-tools is written by Garhoogin and
|
||||
# co-authored by NinjaCheetah.
|
||||
# https://github.com/NinjaCheetah/ASH0-tools
|
||||
#
|
||||
# See <link pending> for details about the ASH archive format.
|
||||
|
||||
import io
|
||||
from dataclasses import dataclass as _dataclass
|
||||
|
||||
|
||||
@_dataclass
|
||||
class _ASHBitReader:
|
||||
"""
|
||||
An _ASHBitReader class used to parse individual words in an ASH file. Private class used by the ASH module.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
src_data : list[int]
|
||||
The entire data of the ASH file being parsed, as a list of integers for each byte.
|
||||
size : int
|
||||
The size of the ASH file.
|
||||
src_pos : int
|
||||
The position in the src_data list currently being accessed.
|
||||
word : int
|
||||
The word currently being decompressed.
|
||||
bit_capacity : int
|
||||
tree_type : str
|
||||
What tree this bit reader is being used with. Used exclusively for debugging, as this value is only used in
|
||||
error messages.
|
||||
"""
|
||||
src_data: list[int]
|
||||
size: int
|
||||
src_pos: int
|
||||
word: int
|
||||
bit_capacity: int
|
||||
tree_type: str
|
||||
|
||||
|
||||
def _ash_bit_reader_feed_word(bit_reader: _ASHBitReader):
|
||||
# Ensure that there's enough data to read en entire word, then if there is, read one.
|
||||
if not bit_reader.src_pos + 4 <= bit_reader.size:
|
||||
print(bit_reader.src_pos)
|
||||
raise ValueError("Invalid ASH data! Cannot decompress.")
|
||||
bit_reader.word = int.from_bytes(bit_reader.src_data[bit_reader.src_pos:bit_reader.src_pos + 4], 'big')
|
||||
bit_reader.bit_capacity = 0
|
||||
bit_reader.src_pos += 4
|
||||
|
||||
|
||||
def _ash_bit_reader_init(bit_reader: _ASHBitReader, src: list[int], size: int, start_pos: int):
|
||||
# Load data into a bit reader, then have it read its first word.
|
||||
bit_reader.src_data = src
|
||||
bit_reader.size = size
|
||||
bit_reader.src_pos = start_pos
|
||||
_ash_bit_reader_feed_word(bit_reader)
|
||||
|
||||
|
||||
def _ash_bit_reader_read_bit(bit_reader: _ASHBitReader):
|
||||
# Reads the starting bit of the current word in the provided bit reader. If the capacity is at 31, then we've
|
||||
# shifted through the entire word, so a new one should be fed. If not, increase the capacity by one and shift the
|
||||
# current word left.
|
||||
bit = bit_reader.word >> 31
|
||||
if bit_reader.bit_capacity == 31:
|
||||
_ash_bit_reader_feed_word(bit_reader)
|
||||
else:
|
||||
bit_reader.bit_capacity += 1
|
||||
bit_reader.word = (bit_reader.word << 1) & 0xFFFFFFFF # This simulates a 32-bit integer.
|
||||
|
||||
return bit
|
||||
|
||||
|
||||
def _ash_bit_reader_read_bits(bit_reader: _ASHBitReader, num_bits: int):
|
||||
# Reads a series of bytes from the current word in the supplied bit reader.
|
||||
bits: int
|
||||
next_bit = bit_reader.bit_capacity + num_bits
|
||||
|
||||
if next_bit <= 32:
|
||||
bits = bit_reader.word >> (32 - num_bits)
|
||||
if next_bit != 32:
|
||||
bit_reader.word = (bit_reader.word << num_bits) & 0xFFFFFFFF # This simulates a 32-bit integer (again).
|
||||
bit_reader.bit_capacity += num_bits
|
||||
else:
|
||||
_ash_bit_reader_feed_word(bit_reader)
|
||||
else:
|
||||
bits = bit_reader.word >> (32 - num_bits)
|
||||
_ash_bit_reader_feed_word(bit_reader)
|
||||
bits |= (bit_reader.word >> (64 - next_bit))
|
||||
bit_reader.word = (bit_reader.word << (next_bit - 32)) & 0xFFFFFFFF # Simulate 32-bit int.
|
||||
bit_reader.bit_capacity = next_bit - 32
|
||||
|
||||
return bits
|
||||
|
||||
|
||||
def _ash_read_tree(bit_reader: _ASHBitReader, width: int, left_tree: [int], right_tree: [int]):
|
||||
# Read either the symbol or distance tree from the ASH file, and return the root of that tree.
|
||||
work = [0] * (2 * (1 << width))
|
||||
work_pos = 0
|
||||
|
||||
r23 = 1 << width
|
||||
tree_root = 0
|
||||
num_nodes = 0
|
||||
|
||||
while True:
|
||||
if _ash_bit_reader_read_bit(bit_reader) != 0:
|
||||
work[work_pos] = (r23 | 0x80000000)
|
||||
work_pos += 1
|
||||
work[work_pos] = (r23 | 0x40000000)
|
||||
work_pos += 1
|
||||
num_nodes += 2
|
||||
r23 += 1
|
||||
else:
|
||||
tree_root = _ash_bit_reader_read_bits(bit_reader, width)
|
||||
while True:
|
||||
work_pos -= 1
|
||||
node_value = work[work_pos]
|
||||
idx = node_value & 0x3FFFFFFF
|
||||
num_nodes -= 1
|
||||
try:
|
||||
if node_value & 0x80000000:
|
||||
right_tree[idx] = tree_root
|
||||
tree_root = idx
|
||||
else:
|
||||
left_tree[idx] = tree_root
|
||||
break
|
||||
except IndexError:
|
||||
raise ValueError("Decompression failed while reading " + bit_reader.tree_type + " tree! Incorrect "
|
||||
"leaf width may have been used. Try using a different number of bits for the " +
|
||||
bit_reader.tree_type + " tree leaves.")
|
||||
# Simulate a do-while loop.
|
||||
if num_nodes == 0:
|
||||
break
|
||||
# Also a do-while.
|
||||
if num_nodes == 0:
|
||||
break
|
||||
|
||||
return tree_root
|
||||
|
||||
|
||||
def _decompress_ash(input_data: list[int], size: int, sym_bits: int, dist_bits: int):
|
||||
# Get the size of the decompressed data by reading the second 4 bytes of the file and masking the first one out.
|
||||
decompressed_size = int.from_bytes(input_data[0x4:0x8]) & 0x00FFFFFF
|
||||
# Array of decompressed data and the position in that array that we're at. Mimics the memory pointer from the
|
||||
# original C source.
|
||||
out_buffer = [0] * decompressed_size
|
||||
out_buffer_pos = 0
|
||||
# Create two empty bit readers, and then initialize them at two different positions for the two trees.
|
||||
bit_reader1 = _ASHBitReader([0], 0, 0, 0, 0, "distance")
|
||||
_ash_bit_reader_init(bit_reader1, input_data, size, int.from_bytes(input_data[0x8:0xC], byteorder='big'))
|
||||
bit_reader2 = _ASHBitReader([0], 0, 0, 0, 0, "symbol")
|
||||
_ash_bit_reader_init(bit_reader2, input_data, size, 0xC)
|
||||
# Calculate the max for the symbol and distance trees based on the bit lengths that were passed. Then, allocate the
|
||||
# arrays for all the trees based on that maximum.
|
||||
sym_max = 1 << sym_bits
|
||||
dist_max = 1 << dist_bits
|
||||
sym_left_tree = [0] * (2 * sym_max - 1)
|
||||
sym_right_tree = [0] * (2 * sym_max - 1)
|
||||
dist_left_tree = [0] * (2 * dist_max - 1)
|
||||
dist_right_tree = [0] * (2 * dist_max - 1)
|
||||
# Read the trees to find the symbol and distance tree roots.
|
||||
sym_root = _ash_read_tree(bit_reader2, sym_bits, sym_left_tree, sym_right_tree)
|
||||
dist_root = _ash_read_tree(bit_reader1, dist_bits, dist_left_tree, dist_right_tree)
|
||||
# Main decompression loop.
|
||||
while True:
|
||||
sym = sym_root
|
||||
while sym >= sym_max:
|
||||
if _ash_bit_reader_read_bit(bit_reader2) != 0:
|
||||
sym = sym_right_tree[sym]
|
||||
else:
|
||||
sym = sym_left_tree[sym]
|
||||
if sym < 0x100:
|
||||
out_buffer[out_buffer_pos] = sym
|
||||
out_buffer_pos += 1
|
||||
decompressed_size -= 1
|
||||
else:
|
||||
dist_sym = dist_root
|
||||
while dist_sym >= dist_max:
|
||||
if _ash_bit_reader_read_bit(bit_reader1) != 0:
|
||||
dist_sym = dist_right_tree[dist_sym]
|
||||
else:
|
||||
dist_sym = dist_left_tree[dist_sym]
|
||||
copy_len = (sym - 0x100) + 3
|
||||
srcp_pos = out_buffer_pos - dist_sym - 1
|
||||
# Check to make sure we aren't going to exceed the specified decompressed size.
|
||||
if not copy_len <= decompressed_size:
|
||||
raise ValueError("Invalid ASH data! Cannot decompress.")
|
||||
|
||||
decompressed_size -= copy_len
|
||||
while copy_len > 0:
|
||||
out_buffer[out_buffer_pos] = out_buffer[srcp_pos]
|
||||
out_buffer_pos += 1
|
||||
srcp_pos += 1
|
||||
copy_len -= 1
|
||||
# Simulate a do-while loop.
|
||||
if decompressed_size == 0:
|
||||
break
|
||||
|
||||
return out_buffer
|
||||
|
||||
|
||||
def decompress_ash(ash_data: bytes, sym_tree_bits: int = 9, dist_tree_bits: int = 11) -> bytes:
|
||||
"""
|
||||
Decompresses the data of an ASH file and returns the decompressed data.
|
||||
|
||||
With the default parameters, this function can decompress ASH files found in the files of the Wii Menu and Animal
|
||||
Crossing: City Folk. Some ASH files, notably the ones found in the WiiWare title My Pokémon Ranch, require setting
|
||||
dist_tree_bits to 15 instead for a successful decompression. If an ASH file is failing to decompress with the
|
||||
default options, trying a dist_tree_bits value of 15 will likely fix it. No other leaf sizes are known to exist,
|
||||
however they might be out there.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
ash_data : bytes
|
||||
The data for the ASH file to decompress.
|
||||
sym_tree_bits : int, option
|
||||
Number of bits for each leaf in the symbol tree. Defaults to 9.
|
||||
dist_tree_bits : int, option
|
||||
Number of bits for each leaf in the distance tree. Defaults to 11.
|
||||
"""
|
||||
# Check the magic number to make sure this is an ASH file.
|
||||
with io.BytesIO(ash_data) as ash_data2:
|
||||
ash_magic = ash_data2.read(4)
|
||||
if ash_magic != b'\x41\x53\x48\x30':
|
||||
raise TypeError("This is not a valid ASH file!")
|
||||
# Begin decompression. Convert the compressed data to an array of ints for processing, then convert the returned
|
||||
# decompressed data back into bytes to return it.
|
||||
ash_size = len(ash_data)
|
||||
ash_data_int = [byte for byte in ash_data]
|
||||
decompressed_data = _decompress_ash(ash_data_int, ash_size, sym_tree_bits, dist_tree_bits)
|
||||
decompressed_data_bin = bytes(decompressed_data)
|
||||
|
||||
return decompressed_data_bin
|
@ -6,17 +6,17 @@
|
||||
import io
|
||||
import os
|
||||
import pathlib
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass as _dataclass
|
||||
from typing import List
|
||||
from ..shared import align_value
|
||||
from ..shared import _align_value
|
||||
|
||||
|
||||
@dataclass
|
||||
class U8Node:
|
||||
@_dataclass
|
||||
class _U8Node:
|
||||
"""
|
||||
A U8Node object that contains the data of a single node in a U8 file header. Each node keeps track of whether this
|
||||
node is for a file or directory, the offset of the name of the file/directory, the offset of the data for the file/
|
||||
directory, and the size of the data.
|
||||
directory, and the size of the data. Private class used by functions and methods in the U8 module.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
@ -44,7 +44,7 @@ class U8Archive:
|
||||
----------
|
||||
"""
|
||||
self.u8_magic = b''
|
||||
self.u8_node_list: List[U8Node] = [] # All the nodes in the header of a U8 file.
|
||||
self.u8_node_list: List[_U8Node] = [] # All the nodes in the header of a U8 file.
|
||||
self.file_name_list: List[str] = []
|
||||
self.file_data_list: List[bytes] = []
|
||||
self.u8_file_structure = dict
|
||||
@ -86,7 +86,7 @@ class U8Archive:
|
||||
node_name_offset = int.from_bytes(u8_data.read(2))
|
||||
node_data_offset = int.from_bytes(u8_data.read(4))
|
||||
node_size = int.from_bytes(u8_data.read(4))
|
||||
self.u8_node_list.append(U8Node(node_type, node_name_offset, node_data_offset, node_size))
|
||||
self.u8_node_list.append(_U8Node(node_type, node_name_offset, node_data_offset, node_size))
|
||||
# Iterate over all loaded nodes and create a list of file names and a list of file data.
|
||||
name_base_offset = u8_data.tell()
|
||||
for node in self.u8_node_list:
|
||||
@ -121,7 +121,7 @@ class U8Archive:
|
||||
for file_name in self.file_name_list:
|
||||
header_size += len(file_name) + 1
|
||||
# The initial data offset is equal to the file header (32 bytes) + node data aligned to 16 bytes.
|
||||
data_offset = align_value(header_size + 32, 16)
|
||||
data_offset = _align_value(header_size + 32, 16)
|
||||
# Adjust all nodes to place file data in the same order as the nodes. Why isn't it already like this?
|
||||
current_data_offset = data_offset
|
||||
for node in range(len(self.u8_node_list)):
|
||||
@ -241,7 +241,7 @@ def _pack_u8_dir(u8_archive: U8Archive, current_path, node_count, name_offset):
|
||||
node_count += 1
|
||||
u8_archive.file_name_list.append(file)
|
||||
u8_archive.file_data_list.append(open(current_path.joinpath(file), "rb").read())
|
||||
u8_archive.u8_node_list.append(U8Node(0, name_offset, 0, len(u8_archive.file_data_list[-1])))
|
||||
u8_archive.u8_node_list.append(_U8Node(0, name_offset, 0, len(u8_archive.file_data_list[-1])))
|
||||
name_offset = name_offset + len(file) + 1 # Add 1 to accommodate the null byte at the end of the name.
|
||||
# For directories, add their name to the file name list, add empty data to the file data list (since they obviously
|
||||
# wouldn't have any), find the total number of files and directories inside the directory to calculate the final
|
||||
@ -251,7 +251,7 @@ def _pack_u8_dir(u8_archive: U8Archive, current_path, node_count, name_offset):
|
||||
u8_archive.file_name_list.append(directory)
|
||||
u8_archive.file_data_list.append(b'')
|
||||
max_node = node_count + sum(1 for _ in current_path.joinpath(directory).rglob('*'))
|
||||
u8_archive.u8_node_list.append(U8Node(256, name_offset, 0, max_node))
|
||||
u8_archive.u8_node_list.append(_U8Node(256, name_offset, 0, max_node))
|
||||
name_offset = name_offset + len(directory) + 1 # Add 1 to accommodate the null byte at the end of the name.
|
||||
u8_archive, node_count, name_offset = _pack_u8_dir(u8_archive, current_path.joinpath(directory), node_count,
|
||||
name_offset)
|
||||
@ -280,7 +280,7 @@ def pack_u8(input_path) -> bytes:
|
||||
u8_archive = U8Archive()
|
||||
u8_archive.file_name_list.append("")
|
||||
u8_archive.file_data_list.append(b'')
|
||||
u8_archive.u8_node_list.append(U8Node(256, 0, 0, sum(1 for _ in input_path.rglob('*')) + 1))
|
||||
u8_archive.u8_node_list.append(_U8Node(256, 0, 0, sum(1 for _ in input_path.rglob('*')) + 1))
|
||||
# Call the private function _pack_u8_dir() on the root note, which will recursively call itself to pack every
|
||||
# subdirectory and file. Discard node_count and name_offset since we don't care about them here, as they're
|
||||
# really only necessary for the directory recursion.
|
||||
@ -300,8 +300,8 @@ def pack_u8(input_path) -> bytes:
|
||||
u8_archive.file_data_list.append(b'')
|
||||
u8_archive.file_data_list.append(file_data)
|
||||
# Append generic U8Node for the root, followed by the actual file's node.
|
||||
u8_archive.u8_node_list.append(U8Node(256, 0, 0, 2))
|
||||
u8_archive.u8_node_list.append(U8Node(0, 1, 0, len(file_data)))
|
||||
u8_archive.u8_node_list.append(_U8Node(256, 0, 0, 2))
|
||||
u8_archive.u8_node_list.append(_U8Node(0, 1, 0, len(file_data)))
|
||||
return u8_archive.dump()
|
||||
else:
|
||||
raise FileNotFoundError("Input file/directory: \"" + str(input_path) + "\" does not exist!")
|
||||
|
@ -4,12 +4,10 @@
|
||||
# This file defines general functions that may be useful in other modules of libWiiPy. Putting them here cuts down on
|
||||
# clutter in other files.
|
||||
|
||||
import binascii
|
||||
|
||||
|
||||
def align_value(value, alignment=64) -> int:
|
||||
def _align_value(value, alignment=64) -> int:
|
||||
"""
|
||||
Aligns the provided value to the set alignment (defaults to 64).
|
||||
Aligns the provided value to the set alignment (defaults to 64). Private function used by other libWiiPy modules.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
@ -29,9 +27,10 @@ def align_value(value, alignment=64) -> int:
|
||||
return value
|
||||
|
||||
|
||||
def pad_bytes(data, alignment=64) -> bytes:
|
||||
def _pad_bytes(data, alignment=64) -> bytes:
|
||||
"""
|
||||
Pads the provided bytes object to the provided alignment (defaults to 64).
|
||||
Pads the provided bytes object to the provided alignment (defaults to 64). Private function used by other libWiiPy
|
||||
modules.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
@ -48,24 +47,3 @@ def pad_bytes(data, alignment=64) -> bytes:
|
||||
while (len(data) % alignment) != 0:
|
||||
data += b'\x00'
|
||||
return data
|
||||
|
||||
|
||||
def convert_tid_to_iv(title_id: str) -> bytes:
|
||||
title_key_iv = b''
|
||||
if type(title_id) is bytes:
|
||||
# This catches the format b'0000000100000002'
|
||||
if len(title_id) == 16:
|
||||
title_key_iv = binascii.unhexlify(title_id)
|
||||
# This catches the format b'\x00\x00\x00\x01\x00\x00\x00\x02'
|
||||
elif len(title_id) == 8:
|
||||
pass
|
||||
# If it isn't one of those lengths, it cannot possibly be valid, so reject it.
|
||||
else:
|
||||
raise ValueError("Title ID is not valid!")
|
||||
# Allow for a string like "0000000100000002"
|
||||
elif type(title_id) is str:
|
||||
title_key_iv = binascii.unhexlify(title_id)
|
||||
# If the Title ID isn't bytes or a string, it isn't valid and is rejected.
|
||||
else:
|
||||
raise TypeError("Title ID type is not valid! It must be either type str or bytes.")
|
||||
return title_key_iv
|
||||
|
@ -2,10 +2,32 @@
|
||||
# https://github.com/NinjaCheetah/libWiiPy
|
||||
|
||||
import struct
|
||||
import binascii
|
||||
from .commonkeys import get_common_key
|
||||
from ..shared import convert_tid_to_iv
|
||||
from Crypto.Cipher import AES as _AES
|
||||
|
||||
from Crypto.Cipher import AES
|
||||
|
||||
def _convert_tid_to_iv(title_id: str) -> bytes:
|
||||
# Converts a Title ID in various formats into the format required to act as an IV. Private function used by other
|
||||
# crypto functions.
|
||||
title_key_iv = b''
|
||||
if type(title_id) is bytes:
|
||||
# This catches the format b'0000000100000002'
|
||||
if len(title_id) == 16:
|
||||
title_key_iv = binascii.unhexlify(title_id)
|
||||
# This catches the format b'\x00\x00\x00\x01\x00\x00\x00\x02'
|
||||
elif len(title_id) == 8:
|
||||
pass
|
||||
# If it isn't one of those lengths, it cannot possibly be valid, so reject it.
|
||||
else:
|
||||
raise ValueError("Title ID is not valid!")
|
||||
# Allow for a string like "0000000100000002"
|
||||
elif type(title_id) is str:
|
||||
title_key_iv = binascii.unhexlify(title_id)
|
||||
# If the Title ID isn't bytes or a string, it isn't valid and is rejected.
|
||||
else:
|
||||
raise TypeError("Title ID type is not valid! It must be either type str or bytes.")
|
||||
return title_key_iv
|
||||
|
||||
|
||||
def decrypt_title_key(title_key_enc: bytes, common_key_index: int, title_id: bytes | str) -> bytes:
|
||||
@ -31,11 +53,11 @@ def decrypt_title_key(title_key_enc: bytes, common_key_index: int, title_id: byt
|
||||
# Load the correct common key for the title.
|
||||
common_key = get_common_key(common_key_index)
|
||||
# Convert the IV into the correct format based on the type provided.
|
||||
title_key_iv = convert_tid_to_iv(title_id)
|
||||
title_key_iv = _convert_tid_to_iv(title_id)
|
||||
# The IV will always be in the same format by this point, so add the last 8 bytes.
|
||||
title_key_iv = title_key_iv + (b'\x00' * 8)
|
||||
# Create a new AES object with the values provided.
|
||||
aes = AES.new(common_key, AES.MODE_CBC, title_key_iv)
|
||||
aes = _AES.new(common_key, _AES.MODE_CBC, title_key_iv)
|
||||
# Decrypt the Title Key using the AES object.
|
||||
title_key = aes.decrypt(title_key_enc)
|
||||
return title_key
|
||||
@ -64,11 +86,11 @@ def encrypt_title_key(title_key_dec: bytes, common_key_index: int, title_id: byt
|
||||
# Load the correct common key for the title.
|
||||
common_key = get_common_key(common_key_index)
|
||||
# Convert the IV into the correct format based on the type provided.
|
||||
title_key_iv = convert_tid_to_iv(title_id)
|
||||
title_key_iv = _convert_tid_to_iv(title_id)
|
||||
# The IV will always be in the same format by this point, so add the last 8 bytes.
|
||||
title_key_iv = title_key_iv + (b'\x00' * 8)
|
||||
# Create a new AES object with the values provided.
|
||||
aes = AES.new(common_key, AES.MODE_CBC, title_key_iv)
|
||||
aes = _AES.new(common_key, _AES.MODE_CBC, title_key_iv)
|
||||
# Encrypt Title Key using the AES object.
|
||||
title_key = aes.encrypt(title_key_dec)
|
||||
return title_key
|
||||
@ -105,7 +127,7 @@ def decrypt_content(content_enc, title_key, content_index, content_length) -> by
|
||||
if (len(content_enc) % 16) != 0:
|
||||
content_enc = content_enc + (b'\x00' * (16 - (len(content_enc) % 16)))
|
||||
# Create a new AES object with the values provided, with the content's unique ID as the IV.
|
||||
aes = AES.new(title_key, AES.MODE_CBC, content_index_bin)
|
||||
aes = _AES.new(title_key, _AES.MODE_CBC, content_index_bin)
|
||||
# Decrypt the content using the AES object.
|
||||
content_dec = aes.decrypt(content_enc)
|
||||
# Trim additional bytes that may have been added so the content is the correct size.
|
||||
@ -144,7 +166,7 @@ def encrypt_content(content_dec, title_key, content_index) -> bytes:
|
||||
if (len(content_dec) % 16) != 0:
|
||||
content_dec = content_dec + (b'\x00' * (16 - (len(content_dec) % 16)))
|
||||
# Create a new AES object with the values provided, with the content's unique ID as the IV.
|
||||
aes = AES.new(title_key, AES.MODE_CBC, content_index_bin)
|
||||
aes = _AES.new(title_key, _AES.MODE_CBC, content_index_bin)
|
||||
# Encrypt the content using the AES object.
|
||||
content_enc = aes.encrypt(content_dec)
|
||||
# Trim down the encrypted content.
|
||||
|
@ -10,7 +10,7 @@ from .title import Title
|
||||
from .tmd import TMD
|
||||
from .ticket import Ticket
|
||||
|
||||
nus_endpoint = ["http://nus.cdn.shop.wii.com/ccs/download/", "http://ccs.cdn.wup.shop.nintendo.net/ccs/download/"]
|
||||
_nus_endpoint = ["http://nus.cdn.shop.wii.com/ccs/download/", "http://ccs.cdn.wup.shop.nintendo.net/ccs/download/"]
|
||||
|
||||
|
||||
def download_title(title_id: str, title_version: int = None, wiiu_endpoint: bool = False) -> Title:
|
||||
@ -68,9 +68,9 @@ def download_tmd(title_id: str, title_version: int = None, wiiu_endpoint: bool =
|
||||
# Build the download URL. The structure is download/<TID>/tmd for latest and download/<TID>/tmd.<version> for
|
||||
# when a specific version is requested.
|
||||
if wiiu_endpoint is False:
|
||||
tmd_url = nus_endpoint[0] + title_id + "/tmd"
|
||||
tmd_url = _nus_endpoint[0] + title_id + "/tmd"
|
||||
else:
|
||||
tmd_url = nus_endpoint[1] + title_id + "/tmd"
|
||||
tmd_url = _nus_endpoint[1] + title_id + "/tmd"
|
||||
# Add the version to the URL if one was specified.
|
||||
if title_version is not None:
|
||||
tmd_url += "." + str(title_version)
|
||||
@ -109,9 +109,9 @@ def download_ticket(title_id: str, wiiu_endpoint: bool = False) -> bytes:
|
||||
# Build the download URL. The structure is download/<TID>/cetk, and cetk will only exist if this is a free
|
||||
# title.
|
||||
if wiiu_endpoint is False:
|
||||
ticket_url = nus_endpoint[0] + title_id + "/cetk"
|
||||
ticket_url = _nus_endpoint[0] + title_id + "/cetk"
|
||||
else:
|
||||
ticket_url = nus_endpoint[1] + title_id + "/cetk"
|
||||
ticket_url = _nus_endpoint[1] + title_id + "/cetk"
|
||||
# Make the request.
|
||||
ticket_request = requests.get(url=ticket_url, headers={'User-Agent': 'wii libnup/1.0'}, stream=True)
|
||||
if ticket_request.status_code != 200:
|
||||
@ -142,11 +142,11 @@ def download_cert(wiiu_endpoint: bool = False) -> bytes:
|
||||
"""
|
||||
# Download the TMD and cetk for the System Menu 4.3U.
|
||||
if wiiu_endpoint is False:
|
||||
tmd_url = nus_endpoint[0] + "0000000100000002/tmd.513"
|
||||
cetk_url = nus_endpoint[0] + "0000000100000002/cetk"
|
||||
tmd_url = _nus_endpoint[0] + "0000000100000002/tmd.513"
|
||||
cetk_url = _nus_endpoint[0] + "0000000100000002/cetk"
|
||||
else:
|
||||
tmd_url = nus_endpoint[1] + "0000000100000002/tmd.513"
|
||||
cetk_url = nus_endpoint[1] + "0000000100000002/cetk"
|
||||
tmd_url = _nus_endpoint[1] + "0000000100000002/tmd.513"
|
||||
cetk_url = _nus_endpoint[1] + "0000000100000002/cetk"
|
||||
tmd = requests.get(url=tmd_url, headers={'User-Agent': 'wii libnup/1.0'}, stream=True).content
|
||||
cetk = requests.get(url=cetk_url, headers={'User-Agent': 'wii libnup/1.0'}, stream=True).content
|
||||
# Assemble the certificate.
|
||||
@ -186,9 +186,9 @@ def download_content(title_id: str, content_id: int, wiiu_endpoint: bool = False
|
||||
if len(content_id_hex) < 2:
|
||||
content_id_hex = "0" + content_id_hex
|
||||
if wiiu_endpoint is False:
|
||||
content_url = nus_endpoint[0] + title_id + "/000000" + content_id_hex
|
||||
content_url = _nus_endpoint[0] + title_id + "/000000" + content_id_hex
|
||||
else:
|
||||
content_url = nus_endpoint[1] + title_id + "/000000" + content_id_hex
|
||||
content_url = _nus_endpoint[1] + title_id + "/000000" + content_id_hex
|
||||
# Make the request.
|
||||
content_request = requests.get(url=content_url, headers={'User-Agent': 'wii libnup/1.0'}, stream=True)
|
||||
if content_request.status_code != 200:
|
||||
|
@ -5,11 +5,33 @@
|
||||
|
||||
import io
|
||||
import binascii
|
||||
from dataclasses import dataclass as _dataclass
|
||||
from .crypto import decrypt_title_key
|
||||
from ..types import TitleLimit
|
||||
from typing import List
|
||||
|
||||
|
||||
@_dataclass
|
||||
class _TitleLimit:
|
||||
"""
|
||||
A TitleLimit object that contains the type of restriction and the limit. The limit type can be one of the following:
|
||||
0 = None, 1 = Time Limit, 3 = None, or 4 = Launch Count. The maximum usage is then either the time in minutes the
|
||||
title can be played or the maximum number of launches allowed for that title, based on the type of limit applied.
|
||||
Private class used only by the Ticket class.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
limit_type : int
|
||||
The type of play limit applied.
|
||||
maximum_usage : int
|
||||
The maximum value for the type of play limit applied.
|
||||
"""
|
||||
# The type of play limit applied.
|
||||
# 0 = None, 1 = Time Limit, 3 = None, 4 = Launch Count
|
||||
limit_type: int
|
||||
# The maximum value of the limit applied.
|
||||
maximum_usage: int
|
||||
|
||||
|
||||
class Ticket:
|
||||
"""
|
||||
A Ticket object that allows for either loading and editing an existing Ticket or creating one manually if desired.
|
||||
@ -47,12 +69,14 @@ class Ticket:
|
||||
self.unknown1: bytes = b'' # Some unknown data, not always the same so reading it just in case.
|
||||
self.title_version: int = 0 # Version of the ticket's associated title.
|
||||
self.permitted_titles: bytes = b'' # Permitted titles mask
|
||||
self.permit_mask: bytes = b'' # "Permit mask. The current disc title is ANDed with the inverse of this mask to see if the result matches the Permitted Titles Mask."
|
||||
# "Permit mask. The current disc title is ANDed with the inverse of this mask to see if the result matches the
|
||||
# Permitted Titles Mask."
|
||||
self.permit_mask: bytes = b''
|
||||
self.title_export_allowed: int = 0 # Whether title export is allowed with a PRNG key or not.
|
||||
self.common_key_index: int = 0 # Which common key should be used. 0 = Common Key, 1 = Korean Key, 2 = vWii Key
|
||||
self.unknown2: bytes = b'' # More unknown data. Varies for VC/non-VC titles so reading it to ensure it matches.
|
||||
self.content_access_permissions: bytes = b'' # "Content access permissions (one bit for each content)"
|
||||
self.title_limits_list: List[TitleLimit] = [] # List of play limits applied to the title.
|
||||
self.title_limits_list: List[_TitleLimit] = [] # List of play limits applied to the title.
|
||||
# v1 ticket data
|
||||
# TODO: Write in v1 ticket attributes here. This code can currently only handle v0 tickets, and will reject v1.
|
||||
|
||||
@ -134,7 +158,7 @@ class Ticket:
|
||||
for limit in range(0, 8):
|
||||
limit_type = int.from_bytes(ticket_data.read(4))
|
||||
limit_value = int.from_bytes(ticket_data.read(4))
|
||||
self.title_limits_list.append(TitleLimit(limit_type, limit_value))
|
||||
self.title_limits_list.append(_TitleLimit(limit_type, limit_value))
|
||||
|
||||
def dump(self) -> bytes:
|
||||
"""
|
||||
|
@ -5,7 +5,7 @@
|
||||
|
||||
import io
|
||||
import binascii
|
||||
from ..shared import align_value, pad_bytes
|
||||
from ..shared import _align_value, _pad_bytes
|
||||
|
||||
|
||||
class WAD:
|
||||
@ -102,12 +102,12 @@ class WAD:
|
||||
# ====================================================================================
|
||||
wad_cert_offset = self.wad_hdr_size
|
||||
# crl isn't ever used, however an entry for its size exists in the header, so its calculated just in case.
|
||||
wad_crl_offset = align_value(wad_cert_offset + self.wad_cert_size)
|
||||
wad_tik_offset = align_value(wad_crl_offset + self.wad_crl_size)
|
||||
wad_tmd_offset = align_value(wad_tik_offset + self.wad_tik_size)
|
||||
wad_crl_offset = _align_value(wad_cert_offset + self.wad_cert_size)
|
||||
wad_tik_offset = _align_value(wad_crl_offset + self.wad_crl_size)
|
||||
wad_tmd_offset = _align_value(wad_tik_offset + self.wad_tik_size)
|
||||
# meta isn't guaranteed to be used, but some older SDK titles use it, and not reading it breaks things.
|
||||
wad_meta_offset = align_value(wad_tmd_offset + self.wad_tmd_size)
|
||||
wad_content_offset = align_value(wad_meta_offset + self.wad_meta_size)
|
||||
wad_meta_offset = _align_value(wad_tmd_offset + self.wad_tmd_size)
|
||||
wad_content_offset = _align_value(wad_meta_offset + self.wad_meta_size)
|
||||
# ====================================================================================
|
||||
# Load data for each WAD section based on the previously calculated offsets.
|
||||
# ====================================================================================
|
||||
@ -159,25 +159,25 @@ class WAD:
|
||||
wad_data += int.to_bytes(self.wad_content_size, 4)
|
||||
# WAD meta size.
|
||||
wad_data += int.to_bytes(self.wad_meta_size, 4)
|
||||
wad_data = pad_bytes(wad_data)
|
||||
wad_data = _pad_bytes(wad_data)
|
||||
# Retrieve the cert data and write it out.
|
||||
wad_data += self.get_cert_data()
|
||||
wad_data = pad_bytes(wad_data)
|
||||
wad_data = _pad_bytes(wad_data)
|
||||
# Retrieve the crl data and write it out.
|
||||
wad_data += self.get_crl_data()
|
||||
wad_data = pad_bytes(wad_data)
|
||||
wad_data = _pad_bytes(wad_data)
|
||||
# Retrieve the ticket data and write it out.
|
||||
wad_data += self.get_ticket_data()
|
||||
wad_data = pad_bytes(wad_data)
|
||||
wad_data = _pad_bytes(wad_data)
|
||||
# Retrieve the TMD data and write it out.
|
||||
wad_data += self.get_tmd_data()
|
||||
wad_data = pad_bytes(wad_data)
|
||||
wad_data = _pad_bytes(wad_data)
|
||||
# Retrieve the meta/footer data and write it out.
|
||||
wad_data += self.get_meta_data()
|
||||
wad_data = pad_bytes(wad_data)
|
||||
wad_data = _pad_bytes(wad_data)
|
||||
# Retrieve the content data and write it out.
|
||||
wad_data += self.get_content_data()
|
||||
wad_data = pad_bytes(wad_data)
|
||||
wad_data = _pad_bytes(wad_data)
|
||||
return wad_data
|
||||
|
||||
def get_wad_type(self) -> str:
|
||||
|
@ -28,24 +28,3 @@ class ContentRecord:
|
||||
content_type: int # Type of content, possible values of: 0x0001: Normal, 0x4001: DLC, 0x8001: Shared.
|
||||
content_size: int
|
||||
content_hash: bytes
|
||||
|
||||
|
||||
@dataclass
|
||||
class TitleLimit:
|
||||
"""
|
||||
A TitleLimit object that contains the type of restriction and the limit. The limit type can be one of the following:
|
||||
0 = None, 1 = Time Limit, 3 = None, or 4 = Launch Count. The maximum usage is then either the time in minutes the
|
||||
title can be played or the maximum number of launches allowed for that title, based on the type of limit applied.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
limit_type : int
|
||||
The type of play limit applied.
|
||||
maximum_usage : int
|
||||
The maximum value for the type of play limit applied.
|
||||
"""
|
||||
# The type of play limit applied.
|
||||
# 0 = None, 1 = Time Limit, 3 = None, 4 = Launch Count
|
||||
limit_type: int
|
||||
# The maximum value of the limit applied.
|
||||
maximum_usage: int
|
||||
|
@ -3,18 +3,18 @@
|
||||
|
||||
import unittest
|
||||
|
||||
from libWiiPy import commonkeys
|
||||
from libWiiPy import title
|
||||
|
||||
|
||||
class TestCommonKeys(unittest.TestCase):
|
||||
def test_common(self):
|
||||
self.assertEqual(commonkeys.get_common_key(0), b'\xeb\xe4*"^\x85\x93\xe4H\xd9\xc5Es\x81\xaa\xf7')
|
||||
self.assertEqual(title.get_common_key(0), b'\xeb\xe4*"^\x85\x93\xe4H\xd9\xc5Es\x81\xaa\xf7')
|
||||
|
||||
def test_korean(self):
|
||||
self.assertEqual(commonkeys.get_common_key(1), b'c\xb8+\xb4\xf4aN.\x13\xf2\xfe\xfb\xbaL\x9b~')
|
||||
self.assertEqual(title.get_common_key(1), b'c\xb8+\xb4\xf4aN.\x13\xf2\xfe\xfb\xbaL\x9b~')
|
||||
|
||||
def test_vwii(self):
|
||||
self.assertEqual(commonkeys.get_common_key(2), b'0\xbf\xc7n|\x19\xaf\xbb#\x1630\xce\xd7\xc2\x8d')
|
||||
self.assertEqual(title.get_common_key(2), b'0\xbf\xc7n|\x19\xaf\xbb#\x1630\xce\xd7\xc2\x8d')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
Loading…
Reference in New Issue
Block a user