Improve and comment examples

Author: elicn

examples/crackme_x86_linux.py

@@ -89,21 +89,26 @@ def replay(self, input: bytes) -> bool:
 
         return False
 
+def progress(msg: str) -> None:
+    print(msg, end='\r', file=sys.stderr, flush=True)
+
 def main():
-    idx_list = (1, 4, 2, 0, 3)
-    flag = [0] * len(idx_list)
+    flag = bytearray(b'*****')
+    indices = (1, 4, 2, 0, 3)
 
-    solver = Solver(bytes(flag))
+    # all possible flag characters (may be reduced to uppercase and digits to save time)
+    charset = string.printable
 
-    for idx in idx_list:
+    progress('Initializing...')
+    solver = Solver(flag)
 
-        # bruteforce all possible flag characters
-        for ch in string.printable:
-            flag[idx] = ord(ch)
+    for i in indices:
+        for ch in charset:
+            flag[i] = ord(ch)
 
-            print(f'Guessing... [{"".join(chr(ch) if ch else "_" for ch in flag)}]', end='\r', file=sys.stderr, flush=True)
+            progress(f'Guessing... {flag.decode()}')
 
-            if solver.replay(bytes(flag)):
+            if solver.replay(flag):
                 break
 
         else:
@@ -113,3 +118,5 @@ def main():
 
 if __name__ == "__main__":
     main()
+
+# expected flag: L1NUX

examples/crackme_x86_windows.py

@@ -10,47 +10,111 @@
 from qiling.const import QL_VERBOSE
 from qiling.extensions import pipe
 
-def instruction_count(ql: Qiling, address: int, size: int, user_data):
-    user_data[0] += 1
-
-def get_count(flag):
-    ql = Qiling(["rootfs/x86_windows/bin/crackme.exe"], "rootfs/x86_windows", verbose=QL_VERBOSE.OFF)
-
-    ql.os.stdin = pipe.SimpleStringBuffer()
-    ql.os.stdout = pipe.SimpleStringBuffer()
-
-    ql.os.stdin.write(bytes("".join(flag) + "\n", 'utf-8'))
-    count = [0]
-
-    ql.hook_code(instruction_count, count)
-    ql.run()
-
-    print(ql.os.stdout.read().decode('utf-8'), end='')
-    print(" ============ count: %d ============ " % count[0])
-
-    return count[0]
-
-def solve():
-    # BJWXB_CTF{C5307D46-E70E-4038-B6F9-8C3F698B7C53}
-    prefix = list("BJWXB_CTF{C5307D46-E70E-4038-B6F9-8C3F698B7C")
-    flag = list("\x00" * 100)
-    base = get_count(prefix + flag)
-    i = 0
-
-    try:
-        for i in range(len(flag)):
-            for j in "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-{}":
-                flag[i] = j
-                data = get_count(prefix + flag)
-                if data > base:
-                    base = data
-                    print("\n\n\n>>> FLAG: " + "".join(prefix + flag) + "\n\n\n")
-                    break
-            if flag[i] == "}":
+ROOTFS = r"rootfs/x86_windows"
+
+class Solver:
+    def __init__(self, invalid: bytes):
+        # create a silent qiling instance
+        self.ql = Qiling([rf"{ROOTFS}/bin/crackme.exe"], ROOTFS, verbose=QL_VERBOSE.OFF)
+
+        self.ql.os.stdin = pipe.SimpleInStream(sys.stdin.fileno())  # take over the input to the program using a fake stdin
+        self.ql.os.stdout = pipe.NullOutStream(sys.stdout.fileno()) # disregard program output
+
+        # execute program until it reaches the part that asks for input
+        self.ql.run(end=0x401030)
+
+        # record replay starting and ending points.
+        #
+        # since the emulation halted upon entering a function, its return address is there on
+        # the stack. we use it to limit the emulation till function returns
+        self.replay_starts = self.ql.arch.regs.arch_pc
+        self.replay_ends = self.ql.stack_read(0)
+
+        # instead of restarting the whole program every time a new flag character is guessed,
+        # we will restore its state to the latest point possible, fast-forwarding a good
+        # amount of start-up code that is not affected by the input.
+        #
+        # here we save the state just when the read input function is about to be called so we
+        # could use it to jumpstart the initialization part and get to the read input immediately
+        self.jumpstart = self.ql.save() or {}
+
+        # calibrate the replay instruction count by running the code with an invalid input
+        # first. the instruction count returned from the calibration process will be then
+        # used as a baseline for consequent replays
+        self.best_icount = self.__run(invalid)
+
+    def __run(self, input: bytes) -> int:
+        icount = [0]
+
+        def __count_instructions(ql: Qiling, address: int, size: int):
+            icount[0] += 1
+
+        # set a hook to fire up every time an instruction is about to execute
+        hobj = self.ql.hook_code(__count_instructions)
+
+        # feed stdin with input
+        self.ql.os.stdin.write(input + b'\n')
+
+        # resume emulation till function returns
+        self.ql.run(begin=self.replay_starts, end=self.replay_ends)
+
+        hobj.remove()
+
+        return icount[0]
+
+    def replay(self, input: bytes) -> bool:
+        """Restore state and replay with a new input.
+
+        Returns an indication to execution progress: `True` if a progress
+        was made, `False` otherwise
+        """
+
+        # restore program's state back to the starting point
+        self.ql.restore(self.jumpstart)
+
+        # resume emulation and count emulated instructions
+        curr_icount = self.__run(input)
+
+        # the larger part of the input is correct, the more instructions are expected to be executed. this is true
+        # for traditional loop-based validations like strcmp or memcmp which bails as soon as a mismatch is found:
+        # more correct characters mean more loop iterations - thus more executed instructions.
+        #
+        # if we got a higher instruction count, it means we made a progress in the right direction
+        if curr_icount > self.best_icount:
+            self.best_icount = curr_icount
+
+            return True
+
+        return False
+
+def progress(msg: str) -> None:
+    print(msg, end='\r', file=sys.stderr, flush=True)
+
+def main():
+    flag = bytearray(b'BJWXB_CTF{********-****-****-****-************}')
+    indices = (i for i, ch in enumerate(flag) if ch == ord('*'))
+
+    # uppercase hex digits
+    charset = '0123456789ABCDEF'
+
+    progress('Initializing...')
+    solver = Solver(flag)
+
+    for i in indices:
+        for ch in charset:
+            flag[i] = ord(ch)
+
+            progress(f'Guessing... {flag.decode()}')
+
+            if solver.replay(flag):
                 break
-        print("SOLVED!!!")
-    except KeyboardInterrupt:
-        print("STOP: KeyboardInterrupt")
+
+        else:
+            raise RuntimeError('no match found')
+
+    print(f'\nFlag found!')
 
 if __name__ == "__main__":
-    solve()
+    main()
+
+# expected flag: BJWXB_CTF{C5307D46-E70E-4038-B6F9-8C3F698B7C53}

examples/hello_arm_linux_custom_syscall.py

@@ -9,30 +9,37 @@
 from qiling import Qiling
 from qiling.const import QL_VERBOSE
 
-def my_syscall_write(ql: Qiling, write_fd, write_buf, write_count, *args, **kw):
-    regreturn = 0
-
+# customized system calls always use the same arguments list as the original ones,
+# but with a Qiling instance as first argument
+def my_syscall_write(ql: Qiling, fd: int, buf: int, count: int):
     try:
-        buf = ql.mem.read(write_buf, write_count)
-        ql.log.info("\n+++++++++\nmy write(%d,%x,%i) = %d\n+++++++++" % (write_fd, write_buf, write_count, regreturn))
-        ql.os.fd[write_fd].write(buf)
-        regreturn = write_count
+        # read data from emulated memory
+        data = ql.mem.read(buf, count)
+
+        # select the emulated file object that corresponds to the requested
+        # file descriptor
+        fobj = ql.os.fd[fd]
+
+        # write the data into the file object, if it supports write operations
+        if hasattr(fobj, 'write'):
+            fobj.write(data)
     except:
-        regreturn = -1
-        ql.log.info("\n+++++++++\nmy write(%d,%x,%i) = %d\n+++++++++" % (write_fd, write_buf, write_count, regreturn))
+        ret = -1
+    else:
+        ret = count
 
-    return regreturn
+    ql.log.info(f'my_syscall_write({fd}, {buf:#x}, {count}) = {ret}')
 
+    return ret
 
 if __name__ == "__main__":
-    ql = Qiling(["rootfs/arm_linux/bin/arm_hello"], "rootfs/arm_linux", verbose=QL_VERBOSE.DEBUG)
-    # Custom syscall handler by syscall name or syscall number.
-    # Known issue: If the syscall func is not be implemented in qiling, qiling does
-    # not know which func should be replaced.
-    # In that case, you must specify syscall by its number.
+    ql = Qiling([r'rootfs/arm_linux/bin/arm_hello'], r'rootfs/arm_linux', verbose=QL_VERBOSE.DEBUG)
+
+    # replacing a system call with a custom implementation.
+    # note that system calls may be referred to either by their name or number.
     ql.os.set_syscall(0x04, my_syscall_write)
 
-    # set syscall by syscall name
-    #ql.os.set_syscall("write", my_syscall_write)
+    # an possible alternative: refer to a syscall by its name
+    #ql.os.set_syscall('write', my_syscall_write)
 
     ql.run()

examples/hello_arm_linux_debug.py

@@ -9,12 +9,13 @@
 from qiling import Qiling
 from qiling.const import QL_VERBOSE
 
-def run_sandbox(path, rootfs, verbose):
-    ql = Qiling(path, rootfs, verbose = verbose)
+if __name__ == "__main__":
+    ql = Qiling([r'rootfs/arm_linux/bin/arm_hello'], r'rootfs/arm_linux', verbose=QL_VERBOSE.DEBUG)
+
     ql.debugger = "qdb" # enable qdb without options
+
+    # other possible alternatives:
     # ql.debugger = "qdb::rr" # switch on record and replay with rr
     # ql.debugger = "qdb:0x1030c" # enable qdb and setup breakpoin at 0x1030c
-    ql.run()
 
-if __name__ == "__main__":
-    run_sandbox(["rootfs/arm_linux/bin/arm_hello"], "rootfs/arm_linux", QL_VERBOSE.DEBUG)
+    ql.run()