Source code for cle.patched_stream
[docs]class PatchedStream:
"""
An object that wraps a readable stream, performing passthroughs on seek and read operations,
except to make it seem like the data has actually been patched by the given patches.
"""
[docs] def __init__(self, stream, patches):
"""
:param stream: The stream to patch
:param patches: A list of tuples of (addr, patch data)
"""
if type(stream) is PatchedStream:
patches = stream.patches + patches
stream = stream.stream
self.stream = stream
self.patches = patches
self._pos = stream.tell()
[docs] def read(self, *args, **kwargs):
data = self.stream.read(*args, **kwargs)
pos = self._pos
newpos = pos + len(data)
# AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
for addr, patch in self.patches:
if (addr >= self._pos and addr < newpos) or (self._pos >= addr and self._pos < addr + len(patch)):
inject_start = max(0, addr - pos)
inject_end = min(addr - pos + len(patch), len(data))
patch_start = max(0, pos - addr)
patch_end = min(newpos - addr, len(patch))
data = data[:inject_start] + patch[patch_start:patch_end] + data[inject_end:]
self._pos = newpos
return data
[docs] def seek(self, *args, **kwargs):
self.stream.seek(*args, **kwargs)
self._pos = self.stream.tell()