<commit_before>#
# xbe.py - Xbox xbe decoder
#
# Based on: http://www.caustik.com/cxbx/download/xbe.htm
# TODO: A future version of the library could use ctypes.structures to overcome this.
# TODO: Add logging
import io
from collections import namedtuple
import struct
import hashlib
import datetime
import codecs

version = '1.0.0'

# Fields in an xbe header
XbeHeaderBase = namedtuple('XbeHeader',
                           'magic digital_signature base_addr size_of_headers size_of_image size_of_image_header '
                           'time_date certificate_addr sections section_headers_addr init_flags entry_addr tls_addr '
                           'pe_stack_commit pe_heap_reserve pe_heap_commit pe_base_addr pe_sizeof_image '
                           'pe_checksum pe_time_date debug_path_name_addr debug_file_name_addr '
                           'debug_unicode_file_name_addr kernel_image_thunk_addr non_kernel_image_import_dir_addr '
                           'library_versions library_versions_addr kernel_library_version_addr '
                           'xapi_library_version_addr logo_bitmap_addr sizeof_logo_bitmap'
                           )

# Fields in an xbe cert
XbeCertBase = namedtuple('XbeCert',
                         'size time_date title_id raw_title_name alternate_title_id allowed_media game_region '
                         'game_ratings disk_number version lan_key signature_key title_alternate_signature_key'
                         )

# Fields in an xbe section header
XbeSectionHeaderBase = namedtuple('XbeSectionHeader',
                                  'section_flags virtual_addr virtual_size raw_addr raw_size section_name_addr '
                                  'section_name_ref_count head_shared_page_ref_count_addr '
                                  'tail_shared_page_ref_count_addr section_digest'
                                  )

# Fields in the library metadata
XbeLibBase = namedtuple('XbeLib',
                        'library_name major_version minor_version build_version library_flags'
                        )

XbeTLSBase = namedtuple('XbeTLS',
                        'data_start_addr data_end_addr tls_index_addr tls_callback_addr size_of_zero_fill '
                        'characteristics'

                        )

XBOX_PUBLIC_KEY_M = int.from_bytes(
    b"\xd3\xd7N\xe5f=\xd7\xe6\xc2\xd4\xa3\xa1\xf2\x176\xd4.R\xf6\xd2\x02\x10\xf5d\x9c4{"
    b"\xff\xef\x7f\xc2\xee\xbd\x05\x8b\xdey\xb4w\x8e[\x8c\x14\x99\xe3\xae\xc6srs\xb5\xfb\x01["
    b"XFm\xfc\x8a\xd6\x95\xda\xed\x1b./\xa2)\xe1?\xf1\xb9["
    b"dQ.\xa2\xc0\xf7\xba\xb3>\x8au\xff\x06\x92\\\x07&uy\x10]G\xbe\xd1jR\x90\x0b\xaej\x0b3D\x93^\xf9"
    b"\x9d\xfb\x15\xd9\xa4\x1c\xcfo\xe4q\x94\xbe\x13\x00\xa8R\xca\x07\xbd'\x98\x01\xa1\x9eO\xa3\xed"
    b"\x9f\xa0\xaas\xc4q\xf3\xe9NrB\x9c\xf09\xce\xbe\x03v\xfa+\x89\x14\x9a\x81\x16\xc1\x80\x8c>k\xaa"
    b"\x05\xecgZ\xcf\xa5p\xbd`\x0c\xe87\x9d\xeb\xf4R\xeaN`\x9f\xe4i\xcfR\xdbh\xf5\x11\xcbW\x8f\x9d"
    b"\xa18\n\x0cG\x1b\xb4lZSn&\x98\xf1\x88\xae|\x96\xbc\xf6\xbf\xb0G\x9a\x8d\xe4\xb3\xe2\x98\x85a"
    b"\xb1\xca_\xf7\x98Q-\x83\x81v\x0c\x88\xba\xd4\xc2\xd5<\x14\xc7r\xda~\xbd\x1bK\xa4",
    "little")

XBOX_PUBLIC_KEY_E = 65537


def mod_exp(base, exponent, modulus):
    """
    Computes s = (base ^ exponent) mod modulus
    (<NAME>: "Applied Cryptography" pp. 244)
    """
    s = 1
    while exponent != 0:
        if exponent & 1:
            s = (s * base) % modulus
        exponent >>= 1
        base = (base * base) % modulus
    return s


class Xbe(object):
    def __init__(self, xbe_file=None, xbe_data=None):

        self._xbe_stream = None

        if xbe_data is not None:
            self._xbe_stream = io.BytesIO(xbe_data)
        elif xbe_file is not None:
            self._xbe_stream = xbe_file
        else:
            raise Exception('No xbe data provided.')

        self._header = None
        self._cert = None
        self._section_headers = None
        self._sections = None
        self._valid_signature = None
        self._libraries = None
        self._tls = None

        if not self.is_xbe(self._xbe_stream):
            raise Exception('Invalid file.')

    @property
    def header(self):
        return self._header if self._header else self._read_header()

    @property
    def certificate(self):
        return self._cert if self._cert else self._read_cert()

    @property
    def sections(self):
        return self._sections if self._sections else self._read_sections()

    def __getitem__(self, item):
        return self.sections[item]

    @property
    def tls(self):
        return self._tls if self._tls else self._read_tls()

    @property
    def libraries(self):
        return self._libraries if self._libraries else self._read_libraries()

    @property
    def valid_signature(self):
        return self.verify_signature() if self._valid_signature is None else self._valid_signature

    @property
    def debug_path_name(self):
        return self._read_string(self.header.debug_path_name_addr - self.header.base_addr)

    @property
    def debug_file_name(self):
        return self._read_string(self.header.debug_file_name_addr - self.header.base_addr)

    @property
    def debug_unicode_file_name(self):
        return self._read_string(self.header.debug_unicode_file_name_addr - self.header.base_addr, encoding='utf-16le')

    def _read_tls(self):
        # TODO: Fix TLS Address calculation.
        if self.header.tls_addr != 0 and False:
            self._xbe_stream.seek(self.header.tls_addr - self.header.base_addr)
            data = self._xbe_stream.read(XbeTLS.size)
            self._tls = XbeTLS.unpack(data)
        else:
            self._tls = XbeTLS.unpack(b'\0' * 24)
        return self._tls

    def _read_string(self, addr, encoding='ascii', rsize=1024):
        string_data = []
        self._xbe_stream.seek(addr)
        reader = codecs.getreader(encoding)(self._xbe_stream)
        c = reader.read(1)
        while c and c != '\0':
            string_data.append(c)
            c = reader.read(1)

        return ''.join(string_data)

    def _read_libraries(self):
        self._libraries = {}

        self._xbe_stream.seek(self.header.library_versions_addr - self.header.base_addr)<commit_msg>Removed useless parameter from Xbe._read_string<commit_after>#
# xbe.py - Xbox xbe decoder
#
# Based on: http://www.caustik.com/cxbx/download/xbe.htm
# TODO: A future version of the library could use ctypes.structures to overcome this.
# TODO: Add logging
import io
from collections import namedtuple
import struct
import hashlib
import datetime
import codecs

version = '1.0.0'

# Fields in an xbe header
XbeHeaderBase = namedtuple('XbeHeader',
                           'magic digital_signature base_addr size_of_headers size_of_image size_of_image_header '
                           'time_date certificate_addr sections section_headers_addr init_flags entry_addr tls_addr '
                           'pe_stack_commit pe_heap_reserve pe_heap_commit pe_base_addr pe_sizeof_image '
                           'pe_checksum pe_time_date debug_path_name_addr debug_file_name_addr '
                           'debug_unicode_file_name_addr kernel_image_thunk_addr non_kernel_image_import_dir_addr '
                           'library_versions library_versions_addr kernel_library_version_addr '
                           'xapi_library_version_addr logo_bitmap_addr sizeof_logo_bitmap'
                           )

# Fields in an xbe cert
XbeCertBase = namedtuple('XbeCert',
                         'size time_date title_id raw_title_name alternate_title_id allowed_media game_region '
                         'game_ratings disk_number version lan_key signature_key title_alternate_signature_key'
                         )

# Fields in an xbe section header
XbeSectionHeaderBase = namedtuple('XbeSectionHeader',
                                  'section_flags virtual_addr virtual_size raw_addr raw_size section_name_addr '
                                  'section_name_ref_count head_shared_page_ref_count_addr '
                                  'tail_shared_page_ref_count_addr section_digest'
                                  )

# Fields in the library metadata
XbeLibBase = namedtuple('XbeLib',
                        'library_name major_version minor_version build_version library_flags'
                        )

XbeTLSBase = namedtuple('XbeTLS',
                        'data_start_addr data_end_addr tls_index_addr tls_callback_addr size_of_zero_fill '
                        'characteristics'

                        )

XBOX_PUBLIC_KEY_M = int.from_bytes(
    b"\xd3\xd7N\xe5f=\xd7\xe6\xc2\xd4\xa3\xa1\xf2\x176\xd4.R\xf6\xd2\x02\x10\xf5d\x9c4{"
    b"\xff\xef\x7f\xc2\xee\xbd\x05\x8b\xdey\xb4w\x8e[\x8c\x14\x99\xe3\xae\xc6srs\xb5\xfb\x01["
    b"XFm\xfc\x8a\xd6\x95\xda\xed\x1b./\xa2)\xe1?\xf1\xb9["
    b"dQ.\xa2\xc0\xf7\xba\xb3>\x8au\xff\x06\x92\\\x07&uy\x10]G\xbe\xd1jR\x90\x0b\xaej\x0b3D\x93^\xf9"
    b"\x9d\xfb\x15\xd9\xa4\x1c\xcfo\xe4q\x94\xbe\x13\x00\xa8R\xca\x07\xbd'\x98\x01\xa1\x9eO\xa3\xed"
    b"\x9f\xa0\xaas\xc4q\xf3\xe9NrB\x9c\xf09\xce\xbe\x03v\xfa+\x89\x14\x9a\x81\x16\xc1\x80\x8c>k\xaa"
    b"\x05\xecgZ\xcf\xa5p\xbd`\x0c\xe87\x9d\xeb\xf4R\xeaN`\x9f\xe4i\xcfR\xdbh\xf5\x11\xcbW\x8f\x9d"
    b"\xa18\n\x0cG\x1b\xb4lZSn&\x98\xf1\x88\xae|\x96\xbc\xf6\xbf\xb0G\x9a\x8d\xe4\xb3\xe2\x98\x85a"
    b"\xb1\xca_\xf7\x98Q-\x83\x81v\x0c\x88\xba\xd4\xc2\xd5<\x14\xc7r\xda~\xbd\x1bK\xa4",
    "little")

XBOX_PUBLIC_KEY_E = 65537


def mod_exp(base, exponent, modulus):
    """
    Computes s = (base ^ exponent) mod modulus
    (<NAME>: "Applied Cryptography" pp. 244)
    """
    s = 1
    while exponent != 0:
        if exponent & 1:
            s = (s * base) % modulus
        exponent >>= 1
        base = (base * base) % modulus
    return s


class Xbe(object):
    def __init__(self, xbe_file=None, xbe_data=None):

        self._xbe_stream = None

        if xbe_data is not None:
            self._xbe_stream = io.BytesIO(xbe_data)
        elif xbe_file is not None:
            self._xbe_stream = xbe_file
        else:
            raise Exception('No xbe data provided.')

        self._header = None
        self._cert = None
        self._section_headers = None
        self._sections = None
        self._valid_signature = None
        self._libraries = None
        self._tls = None

        if not self.is_xbe(self._xbe_stream):
            raise Exception('Invalid file.')

    @property
    def header(self):
        return self._header if self._header else self._read_header()

    @property
    def certificate(self):
        return self._cert if self._cert else self._read_cert()

    @property
    def sections(self):
        return self._sections if self._sections else self._read_sections()

    def __getitem__(self, item):
        return self.sections[item]

    @property
    def tls(self):
        return self._tls if self._tls else self._read_tls()

    @property
    def libraries(self):
        return self._libraries if self._libraries else self._read_libraries()

    @property
    def valid_signature(self):
        return self.verify_signature() if self._valid_signature is None else self._valid_signature

    @property
    def debug_path_name(self):
        return self._read_string(self.header.debug_path_name_addr - self.header.base_addr)

    @property
    def debug_file_name(self):
        return self._read_string(self.header.debug_file_name_addr - self.header.base_addr)

    @property
    def debug_unicode_file_name(self):
        return self._read_string(self.header.debug_unicode_file_name_addr - self.header.base_addr, encoding='utf-16le')

    def _read_tls(self):
        # TODO: Fix TLS Address calculation.
        if self.header.tls_addr != 0 and False:
            self._xbe_stream.seek(self.header.tls_addr - self.header.base_addr)
            data = self._xbe_stream.read(XbeTLS.size)
            self._tls = XbeTLS.unpack(data)
        else:
            self._tls = XbeTLS.unpack(b'\0' * 24)
        return self._tls

    def _read_string(self, addr, encoding='ascii'):
        string_data = []
        self._xbe_stream.seek(addr)
        reader = codecs.getreader(encoding)(self._xbe_stream)
        c = reader.read(1)
        while c and c != '\0':
            string_data.append(c)
            c = reader.read(1)

        return ''.join(string_data)

    def _read_libraries(self):
        self._libraries = {}

        self._xbe_stream.seek(self.header.library_versions_addr - self.header.base_addr)
{'fraction_non_alphanumeric': 0.0980095541401274, 'fraction_numerical': 0.03184713375796178, 'mean_word_length': 2.3238952103731148, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5727564', 'n_tokens_mistral': 4603, 'n_tokens_neox': 4372, 'n_words': 728}