Modify examples to use the mock pipes

examples/crackme_x86_linux.py

@@ -7,39 +7,23 @@ import sys
 sys.path.append("..")
 
 import string
-from typing import TextIO
 
 from qiling import Qiling
-from qiling.os.posix import stat
+from qiling.const import QL_VERBOSE
+from qiling.extensions import pipe
 
 ROOTFS = r"rootfs/x86_linux"
 
-class MyPipe(TextIO):
-    def __init__(self):
-        self.buf = bytearray()
-
-    def write(self, s: bytes):
-        self.buf.extend(s)
-
-    def read(self, size: int) -> bytes:
-        ret = self.buf[:size]
-        self.buf = self.buf[size:]
-
-        return bytes(ret)
-
-    def fileno(self) -> int:
-        return sys.stdin.fileno()
-
-    def fstat(self):
-        return stat.Fstat(self.fileno())
-
 class Solver:
     def __init__(self, invalid: bytes):
+        mock_stdin = pipe.SimpleInStream(sys.stdin.fileno())
+        mock_stdout = pipe.NullOutStream(sys.stdout.fileno())
+
         # create a silent qiling instance
         self.ql = Qiling([rf"{ROOTFS}/bin/crackme_linux"], ROOTFS,
-            console=False,      # thwart qiling logger output
-            stdin=MyPipe(),     # take over the input to the program using a fake stdin
-            stdout=sys.stdout)  # thwart program output
+            verbose=QL_VERBOSE.OFF, # thwart qiling logger output
+            stdin=mock_stdin,       # take over the input to the program using a fake stdin
+            stdout=mock_stdout)     # disregard program output
 
         # execute program until it reaches the 'main' function
         self.ql.run(end=0x0804851b)
@@ -120,13 +104,13 @@ def main():
         for ch in string.printable:
             flag[idx] = ord(ch)
 
-            print(f'\rGuessing... [{"".join(chr(ch) if ch else "_" for ch in flag)}]', end='', file=sys.stderr, flush=True)
+            print(f'Guessing... [{"".join(chr(ch) if ch else "_" for ch in flag)}]', end='\r', file=sys.stderr, flush=True)
 
             if solver.replay(bytes(flag)):
                 break
 
         else:
-            print(f'No match found')
+            raise RuntimeError('no match found')
 
     print(f'\nFlag found!')
 

examples/crackme_x86_windows.py

@@ -8,41 +8,27 @@ sys.path.append("..")
 
 from qiling import Qiling
 from qiling.const import QL_VERBOSE
-
-class StringBuffer:
-    def __init__(self):
-        self.buffer = b''
-
-    def read(self, n):
-        ret = self.buffer[:n]
-        self.buffer = self.buffer[n:]
-        return ret
-
-    def read_all(self):
-        ret = self.buffer
-        self.buffer = b''
-        return ret
-
-    def write(self, string):
-        self.buffer += string
-        return len(string)
+from qiling.extensions import pipe
 
 def instruction_count(ql: Qiling, address: int, size: int, user_data):
     user_data[0] += 1
 
 def get_count(flag):
-    stdin = StringBuffer()
-    stdout = StringBuffer()
+    mock_stdin = pipe.SimpleStringBuffer()
+    mock_stdout = pipe.SimpleStringBuffer()
 
-    ql = Qiling(["rootfs/x86_windows/bin/crackme.exe"], "rootfs/x86_windows", verbose=QL_VERBOSE.OFF, stdin=stdin, stdout=stdout)
+    ql = Qiling(["rootfs/x86_windows/bin/crackme.exe"], "rootfs/x86_windows",
+        verbose=QL_VERBOSE.OFF,
+        stdin=mock_stdin,
+        stdout=mock_stdout)
 
-    stdin.write(bytes("".join(flag) + "\n", 'utf-8'))
+    ql.os.stdin.write(bytes("".join(flag) + "\n", 'utf-8'))
     count = [0]
 
     ql.hook_code(instruction_count, count)
     ql.run()
 
-    print(stdout.read_all().decode('utf-8'), end='')
+    print(ql.os.stdout.read().decode('utf-8'), end='')
     print(" ============ count: %d ============ " % count[0])
 
     return count[0]
@@ -69,6 +55,5 @@ def solve():
     except KeyboardInterrupt:
         print("STOP: KeyboardInterrupt")
 
-
 if __name__ == "__main__":
     solve()

examples/crackme_x86_windows_auto.py

@@ -7,28 +7,7 @@ import sys
 sys.path.append("..")
 
 from qiling import Qiling
-
-class StringBuffer:
-    def __init__(self):
-        self.buffer = b''
-
-    def read(self, n):
-        ret = self.buffer[:n]
-        self.buffer = self.buffer[n:]
-        return ret
-
-    def readline(self, end=b'\n'):
-        ret = b''
-        while True:
-            c = self.read(1)
-            ret += c
-            if c == end:
-                break
-        return ret
-
-    def write(self, string):
-        self.buffer += string
-        return len(string)
+from qiling.extensions import pipe
 
 def force_call_dialog_func(ql: Qiling):
     # get DialogFunc address
@@ -43,10 +22,9 @@ def force_call_dialog_func(ql: Qiling):
     ql.reg.eip = lpDialogFunc
 
 def our_sandbox(path, rootfs):
-    stdin = StringBuffer()
-    ql = Qiling(path, rootfs, stdin=stdin)
+    ql = Qiling(path, rootfs, stdin=pipe.SimpleInStream(sys.stdin.fileno()))
 
-    stdin.write(b"Ea5yR3versing\n")
+    ql.os.stdin.write(b"Ea5yR3versing\n")
     ql.hook_address(force_call_dialog_func, 0x00401016)
     ql.run()
 

examples/fuzzing/linux_x8664/fuzz_x8664_linux.py

@@ -31,52 +31,15 @@ from typing import Any, Optional
 sys.path.append("../../..")
 from qiling import Qiling
 from qiling.const import QL_VERBOSE
-from qiling.os.posix import stat
-
-class MyPipe():
-    """Fake stdin to handle incoming fuzzed keystrokes.
-    """
-
-    def __init__(self):
-        self.buf = b''
-
-    def write(self, s: bytes):
-        self.buf += s
-
-    def read(self, size: int) -> bytes:
-        ret = self.buf[:size]
-        self.buf = self.buf[size:]
-
-        return ret
-
-    def fileno(self) -> int:
-        return 0
-
-    def show(self):
-        pass
-
-    def clear(self):
-        pass
-
-    def flush(self):
-        pass
-
-    def close(self):
-        self.outpipe.close()
-
-    def lseek(self, offset: int, origin: int):
-        pass
-
-    def fstat(self):
-        return stat.Fstat(self.fileno())
+from qiling.extensions import pipe
 
 def main(input_file: str):
-    stdin = MyPipe()
+    mock_stdin = pipe.SimpleInStream(sys.stdin.fileno())
 
     ql = Qiling(["./x8664_fuzz"], "../../rootfs/x8664_linux",
             verbose=QL_VERBOSE.OFF, # keep qiling logging off
             console=False,          # thwart program output
-            stdin=stdin,            # redirect stdin to our fake one
+            stdin=mock_stdin,       # redirect stdin to our mock to feed it with incoming fuzzed keystrokes
             stdout=None,
             stderr=None)
 
@@ -84,7 +47,7 @@ def main(input_file: str):
         """Called with every newly generated input.
         """
 
-        stdin.write(input)
+        ql.os.stdin.write(input)
 
     def start_afl(_ql: Qiling):
         """Callback from inside.

examples/fuzzing/qnx_arm/fuzz_arm_qnx.py

@@ -19,54 +19,15 @@ unicornafl.monkeypatch()
 import sys, os
 from binascii import hexlify
 
-from capstone import *
-
 sys.path.append("../../..")
 from qiling import *
-
-# we cache this for some extra speed
-stdin_fstat = os.fstat(sys.stdin.fileno())
-
-# This is mostly taken from the crackmes
-class MyPipe():
-    def __init__(self):
-        self.buf = b''
-
-    def write(self, s):
-        self.buf += s
-
-    def read(self, size):
-        if size <= len(self.buf):
-            ret = self.buf[: size]
-            self.buf = self.buf[size:]
-        else:
-            ret = self.buf
-            self.buf = ''
-        return ret
-
-    def fileno(self):
-        return 0
-
-    def show(self):
-        pass
-
-    def clear(self):
-        pass
-
-    def flush(self):
-        pass
-
-    def close(self):
-        self.outpipe.close()
-
-    def fstat(self):
-        return stdin_fstat
-
+from qiling.extensions import pipe
 
 def main(input_file, enable_trace=False):
-    stdin = MyPipe()
+    mock_stdin = pipe.SimpleInStream(sys.stdin.fileno())
+
     ql = Qiling(["./arm_fuzz"], "../../rootfs/arm_qnx",
-                stdin=stdin,
+                stdin=mock_stdin,
                 stdout=1 if enable_trace else None,
                 stderr=1 if enable_trace else None,
                 console = True if enable_trace else False)
@@ -75,7 +36,7 @@ def main(input_file, enable_trace=False):
     # ... stdout=sys.stdout, stderr=sys.stderr)
 
     def place_input_callback(uc, input, _, data):
-        stdin.write(input)
+        ql.os.stdin.write(input)
 
     def start_afl(_ql: Qiling):
         """