diff --git a/qiling/extensions/pipe.py b/qiling/extensions/pipe.py index 60729388e..1fcbaeb8e 100644 --- a/qiling/extensions/pipe.py +++ b/qiling/extensions/pipe.py @@ -48,11 +48,28 @@ def writable(self) -> bool: def readable(self) -> bool: return True + def seek(self, offset: int, origin: int) -> int: + # Imitate os.lseek + raise OSError("Illega Seek") + + def lseek(self, offset: int, origin: int) -> int: + return self.seek(offset, origin) + + def seekable(self) -> bool: + return False + class SimpleStreamBase: def __init__(self, fd: int, *args): super().__init__(*args) self.__fd = fd + self.__closed = False + + def close(self) -> None: + self.__closed = True + + def closed(self) -> bool: + return self.__closed def fileno(self) -> int: return self.__fd @@ -84,3 +101,82 @@ def flush(self) -> None: def writable(self) -> bool: return True + +class SimpleBufferedStream(TextIO): + """Simple buffered IO. + """ + + def __init__(self): + self.buff = bytearray() + self.cur = 0 + + def lseek(self, offset: int, origin: int) -> int: + if origin == 0: # SEEK_SET + base = 0 + elif origin == 1: # SEEK_CUR + base = self.cur + else: # SEEK_END + base = len(self.buff) - 1 + + if base + offset >= len(self.buff): + self.cur = base + offset - 1 + else: + self.cur = base + offset + + return self.cur + + def seek(self, offset: int, origin: int) -> int: + return self.lseek(offset, origin) + + def tell(self) -> int: + return self.cur + + def read(self, n: int = -1) -> bytes: + if n == -1: + ret = self.buff[self.cur:] + self.cur = len(self.buff) + else: + ret = self.buff[self.cur:self.cur + n] + + if self.cur + n >= len(self.buff) - 1: + self.cur = len(self.buff) + else: + self.cur = self.cur + n + + return bytes(ret) + + def readline(self, limit: int = -1) -> bytes: + ret = bytearray() + + while not (ret.endswith(b'\n') or len(ret) == limit): + ret.extend(self.read(1)) + + return bytes(ret) + + def write(self, s: bytes) -> int: + if self.cur == 0: + self.buff.extend(s) + else: + self.buff[:self.cur].extend(s) + + self.cur += len(s) + + return len(s) + + def flush(self) -> None: + pass + + def writable(self) -> bool: + return True + + def readable(self) -> bool: + return True + + def seekable(self) -> bool: + return True + + # This method simply extends internal buffer with self.cur untouched. + def extend(self, s: bytes) -> int: + self.buff.extend(s) + + return len(s) \ No newline at end of file