unrpa
unrpa copied to clipboard
Published
20 hours ago •
Lattyware
Lattyware
Random number added in the archive extractor
What did you try to open the archive with unrpa, and how did it fail?
Extracting files from C:\Temp\DSCS-0.1.1-win\game\dscs.rpa. [0.00%] modules\0005_core\keymap.rpyc
There was an error while trying to extract a file from the archive. If you wish to try and extract as much from the archive as possible, please use --continue-on-error. Error Detail: Traceback (most recent call last): File "C:\Users\omega\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.7_qbz5n2kfra8p0\LocalCache\local-packages\Python37\site-packages\unrpa_init_.py", line 134, in extract_files version.postprocess(file_view, output_file) File "C:\Users\omega\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.7_qbz5n2kfra8p0\LocalCache\local-packages\Python37\site-packages\unrpa\versions\version.py", line 24, in postprocess for segment in iter(source.read1, b""): File "C:\Users\omega\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.7_qbz5n2kfra8p0\LocalCache\local-packages\Python37\site-packages\unrpa\view.py", line 20, in read1 return self.base_read(lambda source: source.read1, amount) File "C:\Users\omega\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.7_qbz5n2kfra8p0\LocalCache\local-packages\Python37\site-packages\unrpa\view.py", line 34, in base_read return self.base_read(method, amount) File "C:\Users\omega\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.7_qbz5n2kfra8p0\LocalCache\local-packages\Python37\site-packages\unrpa\view.py", line 37, in base_read raise Exception("End of archive reached before the file should end.") Exception: End of archive reached before the file should end.
Files needed to add support
from __future__ import division, absolute_import, with_statement, print_function, unicode_literals
from renpy.compat import *
import renpy, os.path, sys, types, threading, zlib, re, io, unicodedata
from renpy.compat.pickle import loads
from renpy.webloader import DownloadNeeded
(b'').encode(b'utf-8')
def get_path(fn):
fn = os.path.join(renpy.config.gamedir, fn)
dn = os.path.dirname(fn)
try:
if not os.path.exists(dn):
os.makedirs(dn)
except:
pass
return fn
if renpy.android:
import android.apk
expansion = os.environ.get(b'ANDROID_EXPANSION', None)
if expansion is not None:
print(b'Using expansion file', expansion)
apks = [
android.apk.APK(apk=expansion, prefix=b'assets/x-game/'),
android.apk.APK(apk=expansion, prefix=b'assets/x-renpy/x-common/')]
game_apks = [
apks[0]]
else:
print(b'Not using expansion file.')
apks = [
android.apk.APK(prefix=b'assets/x-game/'),
android.apk.APK(prefix=b'assets/x-renpy/x-common/')]
game_apks = [
apks[0]]
else:
apks = []
game_apks = []
archives = []
old_config_archives = None
lower_map = {}
archive_handlers = []
class RPAv3ArchiveHandler(object):
@staticmethod
def get_supported_extensions():
return [b'.rpa']
@staticmethod
def get_supported_headers():
return [b'RPA-3.0 ']
@staticmethod
def read_index(infile):
l = infile.read(40)
offset = int(l[8:24], 16)
key = int(l[25:33], 16)
infile.seek(offset)
index = loads(zlib.decompress(infile.read()))
for k in index.keys():
if len(index[k][0]) == 2:
index[k] = [ (offset ^ key ^ 3735929054, dlen ^ key ^ 3735929054) for offset, dlen in index[k] ]
else:
index[k] = [ (offset ^ key ^ 3735929054, dlen ^ key ^ 3735929054, start) for offset, dlen, start in index[k] ]
return index
archive_handlers.append(RPAv3ArchiveHandler)
class RPAv2ArchiveHandler(object):
@staticmethod
def get_supported_extensions():
return [b'.rpa']
@staticmethod
def get_supported_headers():
return [b'RPA-2.0 ']
@staticmethod
def read_index(infile):
l = infile.read(24)
offset = int(l[8:], 16)
infile.seek(offset)
index = loads(zlib.decompress(infile.read()))
return index
archive_handlers.append(RPAv2ArchiveHandler)
class RPAv1ArchiveHandler(object):
@staticmethod
def get_supported_extensions():
return [b'.rpi']
@staticmethod
def get_supported_headers():
return [b'x\x9c']
@staticmethod
def read_index(infile):
return loads(zlib.decompress(infile.read()))
archive_handlers.append(RPAv1ArchiveHandler)
def index_archives():
global archives
global old_config_archives
if old_config_archives == renpy.config.archives:
return
else:
old_config_archives = renpy.config.archives[:]
lower_map.clear()
cleardirfiles()
archives = []
max_header_length = 0
for handler in archive_handlers:
for header in handler.get_supported_headers():
header_len = len(header)
if header_len > max_header_length:
max_header_length = header_len
archive_extensions = []
for handler in archive_handlers:
for ext in handler.get_supported_extensions():
if ext not in archive_extensions:
archive_extensions.append(ext)
for prefix in renpy.config.archives:
for ext in archive_extensions:
fn = None
f = None
try:
fn = transfn(prefix + ext)
f = open(fn, b'rb')
except:
continue
with f:
file_header = f.read(max_header_length)
for handler in archive_handlers:
try:
archive_handled = False
for header in handler.get_supported_headers():
if file_header.startswith(header):
f.seek(0, 0)
index = handler.read_index(f)
archives.append((prefix + ext, index))
archive_handled = True
break
if archive_handled == True:
break
except:
raise
for dir, fn in listdirfiles():
lower_map[unicodedata.normalize(b'NFC', fn.lower())] = fn
for fn in remote_files:
lower_map[unicodedata.normalize(b'NFC', fn.lower())] = fn
return
def walkdir(dir):
rv = []
if not os.path.exists(dir) and not renpy.config.developer:
return rv
for i in os.listdir(dir):
if i[0] == b'.':
continue
try:
i = renpy.exports.fsdecode(i)
except:
continue
if os.path.isdir(dir + b'/' + i):
for fn in walkdir(dir + b'/' + i):
rv.append(i + b'/' + fn)
else:
rv.append(i)
return rv
game_files = []
common_files = []
loadable_cache = {}
remote_files = {}
def cleardirfiles():
global common_files
global game_files
game_files = []
common_files = []
scandirfiles_callbacks = []
def scandirfiles():
seen = set()
def add(dn, fn, files, seen):
fn = unicode(fn)
if fn in seen:
return
if fn.startswith(b'cache/'):
return
if fn.startswith(b'saves/'):
return
files.append((dn, fn))
seen.add(fn)
loadable_cache[unicodedata.normalize(b'NFC', fn.lower())] = True
for i in scandirfiles_callbacks:
i(add, seen)
def scandirfiles_from_apk(add, seen):
for apk in apks:
if apk not in game_apks:
files = common_files
else:
files = game_files
for f in apk.list():
f = (b'/').join(i[2:] for i in f.split(b'/'))
add(None, f, files, seen)
return
if renpy.android:
scandirfiles_callbacks.append(scandirfiles_from_apk)
def scandirfiles_from_remote_file(add, seen):
index_filename = os.path.join(renpy.config.gamedir, b'renpyweb_remote_files.txt')
if os.path.exists(index_filename):
files = game_files
with open(index_filename, b'rb') as (remote_index):
while True:
f = remote_index.readline()
metadata = remote_index.readline()
if f == b'' or metadata == b'':
break
f = f.rstrip(b'\r\n')
metadata = metadata.rstrip(b'\r\n')
entry_type, entry_size = metadata.split(b' ')
if entry_type == b'image':
entry_size = [ int(i) for i in entry_size.split(b',') ]
add(b'/game', f, files, seen)
remote_files[f] = {b'type': entry_type, b'size': entry_size}
if renpy.emscripten or os.environ.get(b'RENPY_SIMULATE_DOWNLOAD', False):
scandirfiles_callbacks.append(scandirfiles_from_remote_file)
def scandirfiles_from_filesystem(add, seen):
for i in renpy.config.searchpath:
if renpy.config.commondir and i == renpy.config.commondir:
files = common_files
else:
files = game_files
i = os.path.join(renpy.config.basedir, i)
for j in walkdir(i):
add(i, j, files, seen)
scandirfiles_callbacks.append(scandirfiles_from_filesystem)
def scandirfiles_from_archives(add, seen):
files = game_files
for _prefix, index in archives:
for j in index:
add(None, j, files, seen)
return
scandirfiles_callbacks.append(scandirfiles_from_archives)
def listdirfiles(common=True):
if not game_files and not common_files:
scandirfiles()
if common:
return game_files + common_files
else:
return list(game_files)
class SubFile(object):
def __init__(self, fn, base, length, start):
self.fn = fn
self.f = None
self.base = base
self.offset = 0
self.length = length
self.start = start
if not self.start:
self.name = fn
else:
self.name = None
return
def open(self):
self.f = open(self.fn, b'rb')
self.f.seek(self.base)
def __enter__(self):
return self
def __exit__(self, _type, value, tb):
self.close()
return False
def read(self, length=None):
if self.f is None:
self.open()
maxlength = self.length - self.offset
if length is not None:
length = min(length, maxlength)
else:
length = maxlength
rv1 = self.start[self.offset:self.offset + length]
length -= len(rv1)
self.offset += len(rv1)
if length:
rv2 = self.f.read(length)
self.offset += len(rv2)
else:
rv2 = b''
return rv1 + rv2
def readline(self, length=None):
if self.f is None:
self.open()
maxlength = self.length - self.offset
if length is not None:
length = min(length, maxlength)
else:
length = maxlength
if self.offset < len(self.start):
rv = b''
while length:
c = self.read(1)
rv += c
if c == b'\n':
break
length -= 1
return rv
rv = self.f.readline(length)
self.offset += len(rv)
return rv
def readlines(self, length=None):
rv = []
while True:
l = self.readline(length)
if not l:
break
if length is not None:
length -= len(l)
if l < 0:
break
rv.append(l)
return rv
def xreadlines(self):
return self
def __iter__(self):
return self
def __next__(self):
rv = self.readline()
if not rv:
raise StopIteration()
return rv
next = __next__
def flush(self):
pass
def seek(self, offset, whence=0):
if self.f is None:
self.open()
if whence == 0:
offset = offset
elif whence == 1:
offset = self.offset + offset
elif whence == 2:
offset = self.length + offset
if offset > self.length:
offset = self.length
self.offset = offset
offset = offset - len(self.start)
if offset < 0:
offset = 0
self.f.seek(offset + self.base)
return
def tell(self):
return self.offset
def close(self):
if self.f is not None:
self.f.close()
self.f = None
return
def write(self, s):
raise Exception(b'Write not supported by SubFile')
open_file = open
if b'RENPY_FORCE_SUBFILE' in os.environ:
def open_file(name, mode):
f = open(name, mode)
f.seek(0, 2)
length = f.tell()
f.seek(0, 0)
return SubFile(f, 0, length, b'')
file_open_callbacks = []
def load_core(name):
name = lower_map.get(unicodedata.normalize(b'NFC', name.lower()), name)
for i in file_open_callbacks:
rv = i(name)
if rv is not None:
return rv
return
def load_from_file_open_callback(name):
if renpy.config.file_open_callback:
return renpy.config.file_open_callback(name)
else:
return
file_open_callbacks.append(load_from_file_open_callback)
def load_from_filesystem(name):
if not renpy.config.force_archives:
try:
fn = transfn(name)
return open_file(fn, b'rb')
except:
pass
return
file_open_callbacks.append(load_from_filesystem)
def load_from_apk(name):
for apk in apks:
prefixed_name = (b'/').join(b'x-' + i for i in name.split(b'/'))
try:
return apk.open(prefixed_name)
except IOError:
pass
return
if renpy.android:
file_open_callbacks.append(load_from_apk)
def load_from_archive(name):
for prefix, index in archives:
if name not in index:
continue
afn = transfn(prefix)
data = []
if len(index[name]) == 1:
t = index[name][0]
if len(t) == 2:
offset, dlen = t
start = b''
else:
offset, dlen, start = t
rv = SubFile(afn, offset, dlen, start)
else:
with open(afn, b'rb') as (f):
for offset, dlen in index[name]:
f.seek(offset)
data.append(f.read(dlen))
rv = io.BytesIO((b'').join(data))
return rv
return
file_open_callbacks.append(load_from_archive)
def load_from_remote_file(name):
if name in remote_files:
raise DownloadNeeded(relpath=name, rtype=remote_files[name][b'type'], size=remote_files[name][b'size'])
return
if renpy.emscripten or os.environ.get(b'RENPY_SIMULATE_DOWNLOAD', False):
file_open_callbacks.append(load_from_remote_file)
def check_name(name):
if renpy.config.reject_backslash and b'\\' in name:
raise Exception(b"Backslash in filename, use '/' instead: %r" % name)
if renpy.config.reject_relative:
split = name.split(b'/')
if b'.' in split or b'..' in split:
raise Exception(b"Filenames may not contain relative directories like '.' and '..': %r" % name)
def get_prefixes(tl=True):
rv = []
if tl:
language = renpy.game.preferences.language
else:
language = None
for prefix in renpy.config.search_prefixes:
if language is not None:
rv.append(renpy.config.tl_directory + b'/' + language + b'/' + prefix)
rv.append(prefix)
return rv
def load(name, tl=True):
if renpy.display.predict.predicting:
if threading.current_thread().name == b'MainThread':
if not (renpy.emscripten or os.environ.get(b'RENPY_SIMULATE_DOWNLOAD', False)):
raise Exception((b'Refusing to open {} while predicting.').format(name))
if renpy.config.reject_backslash and b'\\' in name:
raise Exception(b"Backslash in filename, use '/' instead: %r" % name)
name = re.sub(b'/+', b'/', name).lstrip(b'/')
for p in get_prefixes(tl):
rv = load_core(p + name)
if rv is not None:
return rv
raise IOError(b"Couldn't find file '%s'." % name)
return
def loadable_core(name):
name = lower_map.get(unicodedata.normalize(b'NFC', name.lower()), name)
if name in loadable_cache:
return loadable_cache[name]
try:
transfn(name)
loadable_cache[name] = True
return True
except:
pass
for apk in apks:
prefixed_name = (b'/').join(b'x-' + i for i in name.split(b'/'))
if prefixed_name in apk.info:
loadable_cache[name] = True
return True
for _prefix, index in archives:
if name in index:
loadable_cache[name] = True
return True
if name in remote_files:
loadable_cache[name] = True
return name
loadable_cache[name] = False
return False
def loadable(name):
name = name.lstrip(b'/')
if renpy.config.loadable_callback is not None and renpy.config.loadable_callback(name):
return True
else:
for p in get_prefixes():
if loadable_core(p + name):
return True
return False
def transfn(name):
name = name.lstrip(b'/')
if renpy.config.reject_backslash and b'\\' in name:
raise Exception(b"Backslash in filename, use '/' instead: %r" % name)
name = lower_map.get(unicodedata.normalize(b'NFC', name.lower()), name)
if isinstance(name, bytes):
name = name.decode(b'utf-8')
for d in renpy.config.searchpath:
fn = os.path.join(renpy.config.basedir, d, name)
add_auto(fn)
if os.path.isfile(fn):
return fn
raise Exception(b"Couldn't find file '%s'." % name)
hash_cache = dict()
def get_hash(name):
rv = hash_cache.get(name, None)
if rv is not None:
return rv
else:
rv = 0
try:
f = load(name)
while True:
data = f.read(1048576)
if not data:
break
rv = zlib.adler32(data, rv)
except:
pass
hash_cache[name] = rv
return rv
class RenpyImporter(object):
def __init__(self, prefix=b''):
self.prefix = prefix
def translate(self, fullname, prefix=None):
if prefix is None:
prefix = self.prefix
try:
if not isinstance(fullname, str):
fullname = fullname.decode(b'utf-8')
fn = prefix + fullname.replace(b'.', b'/')
except:
return
if loadable(fn + b'.py'):
return fn + b'.py'
else:
if loadable(fn + b'/__init__.py'):
return fn + b'/__init__.py'
return
def find_module(self, fullname, path=None):
if path is not None:
for i in path:
if self.translate(fullname, i):
return RenpyImporter(i)
if self.translate(fullname):
return self
else:
return
def load_module(self, fullname):
filename = self.translate(fullname, self.prefix)
pyname = pystr(fullname)
mod = sys.modules.setdefault(pyname, types.ModuleType(pyname))
mod.__name__ = pyname
mod.__file__ = filename
mod.__loader__ = self
if filename.endswith(b'__init__.py'):
mod.__path__ = [
filename[:-len(b'__init__.py')]]
for encoding in [b'utf-8', b'latin-1']:
try:
source = load(filename).read().decode(encoding)
if source and source[0] == b'\ufeff':
source = source[1:]
source = source.encode(b'raw_unicode_escape')
source = source.replace(b'\r', b'')
code = compile(source, filename, b'exec', renpy.python.old_compile_flags, 1)
break
except:
if encoding == b'latin-1':
raise
exec code in mod.__dict__
return sys.modules[fullname]
def get_data(self, filename):
return load(filename).read()
meta_backup = []
def add_python_directory(path):
if path and not path.endswith(b'/'):
path = path + b'/'
sys.meta_path.insert(0, RenpyImporter(path))
def init_importer():
meta_backup[:] = sys.meta_path
add_python_directory(b'python-packages/')
add_python_directory(b'')
def quit_importer():
sys.meta_path[:] = meta_backup
needs_autoreload = set()
auto_mtimes = {}
auto_thread = None
auto_quit_flag = True
auto_lock = threading.Condition()
auto_blacklisted = renpy.object.Sentinel(b'auto_blacklisted')
def auto_mtime(fn):
try:
return os.path.getmtime(fn)
except:
return
return
def add_auto(fn, force=False):
fn = fn.replace(b'\\', b'/')
if not renpy.autoreload:
return
if fn in auto_mtimes and not force:
return
for e in renpy.config.autoreload_blacklist:
if fn.endswith(e):
with auto_lock:
auto_mtimes[fn] = auto_blacklisted
return
mtime = auto_mtime(fn)
with auto_lock:
auto_mtimes[fn] = mtime
def auto_thread_function():
global auto_quit_flag
global needs_autoreload
while True:
with auto_lock:
auto_lock.wait(1.5)
if auto_quit_flag:
return
items = list(auto_mtimes.items())
for fn, mtime in items:
if mtime is auto_blacklisted:
continue
if auto_mtime(fn) != mtime:
with auto_lock:
if auto_mtime(fn) != auto_mtimes[fn]:
needs_autoreload.add(fn)
def check_autoreload():
while needs_autoreload:
fn = next(iter(needs_autoreload))
mtime = auto_mtime(fn)
with auto_lock:
needs_autoreload.discard(fn)
auto_mtimes[fn] = mtime
if not renpy.autoreload:
return
for regex, func in renpy.config.autoreload_functions:
if re.search(regex, fn, re.I):
fn = os.path.relpath(fn, renpy.config.gamedir).replace(b'\\', b'/')
func(fn)
break
else:
renpy.exports.reload_script()
def auto_init():
global auto_quit_flag
global auto_thread
global needs_autoreload
needs_autoreload = set()
if not renpy.autoreload:
return
auto_quit_flag = False
auto_thread = threading.Thread(target=auto_thread_function)
auto_thread.daemon = True
auto_thread.start()
def auto_quit():
global auto_quit_flag
if auto_thread is None:
return
else:
auto_quit_flag = True
with auto_lock:
auto_lock.notify_all()
auto_thread.join()
return
Additional context
It seems a random number is added to the key and offset to determine the index in the archive extractor. NSFW link to game
