Merge pull request #1075 from qilingframework/libfuzzer

Fuzzing evovled

Author: xwings

examples/fuzzing/dlink_dir815/dir815_mips32el_linux.py

@@ -7,15 +7,10 @@
 
 import os,sys
 
-# This is new. Instead of unicorn, we import unicornafl. It's the same Uc with some new `afl_` functions
-import unicornafl
-
-# Make sure Qiling uses our patched unicorn instead of it's own, second so without instrumentation!
-unicornafl.monkeypatch()
-
 sys.path.append("../../..")
 from qiling import *
 from qiling.const import QL_VERBOSE
+from qiling.extensions.afl import ql_afl_fuzz
 
 def main(input_file, enable_trace=False):
 
@@ -32,32 +27,17 @@ def main(input_file, enable_trace=False):
                 verbose=QL_VERBOSE.DEBUG, env=env_vars,
                 console = True if enable_trace else False)
 
-    def place_input_callback(uc, input, _, data):
+    def place_input_callback(ql: Qiling, input: bytes, _: int):
         env_var = ("HTTP_COOKIE=uid=1234&password=").encode()
         env_vars = env_var + input + b"\x00" + (ql.path).encode() + b"\x00"
         ql.mem.write(ql.target_addr, env_vars)
 
-
     def start_afl(_ql: Qiling):
 
         """
         Callback from inside
         """
-        # We start our AFL forkserver or run once if AFL is not available.
-        # This will only return after the fuzzing stopped.
-        try:
-            print("Starting afl_fuzz().")
-            if not _ql.uc.afl_fuzz(input_file=input_file,
-                        place_input_callback=place_input_callback,
-                        exits=[ql.os.exit_point]):
-                print("Ran once without AFL attached.")
-                os._exit(0)  # that's a looot faster than tidying up.
-        except unicornafl.UcAflError as ex:
-            # This hook trigers more than once in this example.
-            # If this is the exception cause, we don't care.
-            # TODO: Chose a better hook position :)
-            if ex != unicornafl.UC_AFL_RET_CALLED_TWICE:
-                raise
+        ql_afl_fuzz(_ql, input_file=input_file, place_input_callback=place_input_callback, exits=[ql.os.exit_point])
 
     addr = ql.mem.search("HTTP_COOKIE=uid=1234&password=".encode())
     ql.target_addr = addr[0]

examples/fuzzing/linux_x8664/fuzz_x8664_linux.py

@@ -17,9 +17,7 @@
     $ rm -fr afl_outputs/default/
 """
 
-# This uses the new unicornafl, which no longer provides any Unicorn stuff so we have to import by our own.
-from unicornafl import *
-from unicorn import *
+# No more need for importing unicornafl, try ql.afl_fuzz instead!
 
 import os
 import sys
@@ -30,6 +28,7 @@
 from qiling import Qiling
 from qiling.const import QL_VERBOSE
 from qiling.extensions import pipe
+from qiling.extensions.afl import ql_afl_fuzz
 
 def main(input_file: str):
     mock_stdin = pipe.SimpleInStream(sys.stdin.fileno())
@@ -41,31 +40,18 @@ def main(input_file: str):
             stdout=None,
             stderr=None)
 
-    def place_input_callback(uc: Uc, input: bytes, persistent_round: int, data: Any) -> Optional[bool]:
+    def place_input_callback(ql: Qiling, input: bytes, persistent_round: int) -> Optional[bool]:
         """Called with every newly generated input.
         """
 
         ql.os.stdin.write(input)
 
+        return True
+
     def start_afl(_ql: Qiling):
         """Callback from inside.
         """
-
-        # We start our AFL forkserver or run once if AFL is not available.
-        # This will only return after the fuzzing stopped.
-        try:
-            # _ql.uc.afl_fuzz shall also work, but just for compatibility with old unicornafl
-            if not uc_afl_fuzz(_ql.uc, input_file=input_file, place_input_callback=place_input_callback, exits=[ql.os.exit_point]):
-                _ql.log.warning("Ran once without AFL attached")
-                os._exit(0)
-
-        except UcAflError as ex:
-            # This hook triggers more than once in this example.
-            # If this is the exception cause, we don't care.
-
-            # TODO: choose a better hook position :)
-            if ex.errno != UC_AFL_RET_CALLED_TWICE:
-                raise
+        ql_afl_fuzz(_ql, input_file=input_file, place_input_callback=place_input_callback, exits=[ql.os.exit_point])
 
     # get image base address
     ba = ql.loader.images[0].base

examples/fuzzing/linux_x8664/libfuzzer_x8664_linux.py

@@ -0,0 +1,45 @@
+#!/usr/bin/python3
+
+from fuzzercorn import *
+from unicorn import *
+from qiling import Qiling
+from qiling.extensions import pipe
+
+import sys, os, ctypes
+
+class SimpleFuzzer:
+
+    def run(self):
+        ql = Qiling(["./x8664_fuzz"], "../../rootfs/x8664_linux",
+            console=False, # No output
+            stdin=None,
+            stdout=None,
+            stderr=None)
+        ba = ql.loader.images[0].base
+        try:
+            # Only instrument the function `fun`, so we don't need to instrument libc and ld
+            FuzzerCornFuzz(ql.uc, sys.argv, [ql.os.exit_point], self.place_input, self.init, UserData=ql, Ranges=[(ba + 0x1179, ba + 0x122B)], CountersCount=4096)
+        except Exception as ex:
+            os.abort() # Quick exit
+    
+    def place_input(self, uc: Uc, data: ctypes.Array, ql: Qiling):
+        ql.restore(self.snapshot)
+        ql.os.stdin = pipe.SimpleInStream(1)
+        ql.os.stdin.write(bytes(data))
+        return 1
+
+    def init(self, uc: Uc, argv: list, ql: Qiling):
+        ba = ql.loader.images[0].base
+        ql.hook_address(callback=lambda x: os.abort(), address=ba + 0x1225) # ___stack_chk_fail
+        ql.run(end=ba + 0x122c) # Run to main.
+
+        # Save a snapshot.
+        self.snapshot = ql.save()
+        return 0
+    
+
+if __name__ == "__main__":
+    # chmod +x ./libfuzzer_x8664_linux.py
+    # ./libfuzzer_x8664_linux.py -jobs=6 -workers=6
+    SimpleFuzzer().run()
+

examples/fuzzing/qnx_arm/fuzz_arm_qnx.py

@@ -10,18 +10,15 @@
 afl-fuzz -i ./afl_inputs -o ./afl_outputs -m none -U -- python3 ./fuzz_x8664_linux.py @@
 """
 
-# This is new. Instead of unicorn, we import unicornafl. It's the same Uc with some new `afl_` functions
-import unicornafl
-
-# Make sure Qiling uses our patched unicorn instead of it's own, second so without instrumentation!
-unicornafl.monkeypatch()
+# No more need for importing unicornafl, try ql.afl_fuzz instead!
 
 import sys, os
 from binascii import hexlify
 
 sys.path.append("../../..")
 from qiling import *
 from qiling.extensions import pipe
+from qiling.extensions.afl import ql_afl_fuzz
 
 def main(input_file, enable_trace=False):
     mock_stdin = pipe.SimpleInStream(sys.stdin.fileno())
@@ -35,33 +32,17 @@ def main(input_file, enable_trace=False):
     # or this for output:
     # ... stdout=sys.stdout, stderr=sys.stderr)
 
-    def place_input_callback(uc, input, _, data):
+    def place_input_callback(ql: Qiling, input: bytes, _: int):
         ql.os.stdin.write(input)
+        return True
 
     def start_afl(_ql: Qiling):
-        """
-        Callback from inside
-        """
-        # We start our AFL forkserver or run once if AFL is not available.
-        # This will only return after the fuzzing stopped.
-        try:
-            #print("Starting afl_fuzz().")
-            if not _ql.uc.afl_fuzz(input_file=input_file,
-                        place_input_callback=place_input_callback,
-                        exits=[ql.os.exit_point]):
-                print("Ran once without AFL attached.")
-                os._exit(0)  # that's a looot faster than tidying up.
-        except unicornafl.UcAflError as ex:
-            # This hook trigers more than once in this example.
-            # If this is the exception cause, we don't care.
-            # TODO: Chose a better hook position :)
-            if ex != unicornafl.UC_AFL_RET_CALLED_TWICE:
-                raise
+        ql_afl_fuzz(_ql, input_file=input_file, place_input_callback=place_input_callback, exits=[ql.os.exit_point])
 
     LIBC_BASE = int(ql.profile.get("OS32", "interp_address"), 16)
 
     # crash in case we reach SignalKill
-    ql.hook_address(callback=lambda x: os.abort(), address=LIBC_BASE + 0x456d4)
+    ql.hook_address(callback=lambda x: os.abort(), address=LIBC_BASE + 0x38170)
 
     # Add hook at main() that will fork Unicorn and start instrumentation.
     main_addr = 0x08048aa0

examples/fuzzing/tenda_ac15/fuzz_tendaac15_httpd.py

@@ -13,12 +13,10 @@
 
 import os, pickle, socket, sys, threading
 
-import unicornafl
-unicornafl.monkeypatch()
-
 sys.path.append("../../../")
 from qiling import *
 from qiling.const import QL_VERBOSE
+from qiling.extensions.afl import ql_afl_fuzz
 
 def patcher(ql):
     br0_addr = ql.mem.search("br0".encode() + b'\x00')
@@ -40,25 +38,10 @@ def place_input_callback(uc, input, _, data):
         ql.mem.write(target_address, input)
 
     def start_afl(_ql: Qiling):
-
         """
         Callback from inside
         """
-        # We start our AFL forkserver or run once if AFL is not available.
-        # This will only return after the fuzzing stopped.
-        try:
-            #print("Starting afl_fuzz().")
-            if not _ql.uc.afl_fuzz(input_file=input_file,
-                        place_input_callback=place_input_callback,
-                        exits=[ql.os.exit_point]):
-                print("Ran once without AFL attached.")
-                os._exit(0)  # that's a looot faster than tidying up.
-        except unicornafl.UcAflError as ex:
-            # This hook trigers more than once in this example.
-            # If this is the exception cause, we don't care.
-            # TODO: Chose a better hook position :)
-            if ex != unicornafl.UC_AFL_RET_CALLED_TWICE:
-                raise
+        ql_afl_fuzz(_ql, input_file=input_file, place_input_callback=place_input_callback, exits=[ql.os.exit_point])
 
     ql.hook_address(callback=start_afl, address=0x10930+8)
 

qiling/core.py

@@ -7,10 +7,11 @@
 import ntpath, os, pickle, platform
 
 # See https://stackoverflow.com/questions/39740632/python-type-hinting-without-cyclic-imports
-from typing import Dict, List, Union
+from typing import Callable, Dict, List, Union
 from typing import TYPE_CHECKING
 
 from unicorn.unicorn import Uc
+
 if TYPE_CHECKING:
     from .arch.register import QlRegisterManager
     from .arch.arch import QlArch
@@ -840,7 +841,6 @@ def stack_read(self, offset):
     def stack_write(self, offset, data):
         return self.arch.stack_write(offset, data)
 
-
     # Assembler/Diassembler API
     @property
     def assembler(self):

qiling/extensions/pipe.py

@@ -48,11 +48,25 @@ def writable(self) -> bool:
     def readable(self) -> bool:
         return True
 
+    def seek(self, offset: int, origin: int = 0) -> int:
+        # Imitate os.lseek
+        raise OSError("Illega Seek")
+
+    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 +98,73 @@ 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 seek(self, offset: int, origin: int = 0) -> int:
+        if origin == 0: # SEEK_SET
+            base = 0
+        elif origin == 1: # SEEK_CUR
+            base = self.cur
+        else: # SEEK_END
+            base = len(self.buff)
+
+        if base + offset >= len(self.buff):
+            self.cur = len(self.buff)
+        elif base + offset <= 0:
+            self.cur = 0
+        else:
+            self.cur = base + offset
+
+        return self.cur
+    
+    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):
+                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:
+        self.buff = self.buff[:self.cur]
+        self.buff.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

qiling/os/filestruct.py

@@ -43,6 +43,9 @@ def write(self, write_buf: bytes) -> int:
     def fileno(self) -> int:
         return self.__fd
 
+    def seek(self, lseek_offset: int, lseek_origin: int = os.SEEK_SET) -> int:
+        return self.lseek(lseek_offset, lseek_origin)
+
     def lseek(self, lseek_offset: int, lseek_origin: int = os.SEEK_SET) -> int:
         return os.lseek(self.__fd, lseek_offset, lseek_origin)