Merge branch 'qilingframework:dev' into dev

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,11 +17,7 @@
     $ rm -fr afl_outputs/default/
 """
 
-# This is new. Instead of unicorn, we import unicornafl. It's the same Uc with some new `afl_` functions
-import unicornafl as UcAfl
-
-# Make sure Qiling uses our patched unicorn instead of it's own, second so without instrumentation!
-UcAfl.monkeypatch()
+# No more need for importing unicornafl, try ql.afl_fuzz instead!
 
 import os
 import sys
@@ -32,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())
@@ -43,30 +40,18 @@ def main(input_file: str):
             stdout=None,
             stderr=None)
 
-    def place_input_callback(uc: UcAfl.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:
-            if not _ql.uc.afl_fuzz(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 UcAfl.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 != UcAfl.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)
 

examples/sality.py

@@ -75,7 +75,7 @@ def hook_CreateFileA(ql: Qiling, address: int, params):
         else:
             return (-1)
     else:
-        ret = _CreateFile(ql, address, params, "CreateFileA")
+        ret = _CreateFile(ql, address, params)
     return ret
 
 def _WriteFile(ql: Qiling, address: int, params):

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/afl/__init__.py

@@ -0,0 +1 @@
+from .afl import ql_afl_fuzz

qiling/extensions/afl/afl.py

@@ -0,0 +1,59 @@
+from typing import List, Callable
+from qiling.core import Qiling
+from unicornafl import *
+from qiling.exception import QlErrorNotImplemented
+
+def ql_afl_fuzz(ql: Qiling,
+                input_file: str,
+                place_input_callback: Callable[["Qiling", bytes, int], bool],
+                exits: List[int],
+                validate_crash_callback: Callable[["Qiling", bytes, int], bool] = None,
+                always_validate: bool = False,
+                persistent_iters: int = 1):
+        """ Fuzz a range of code with afl++.
+            This function wraps some common logic with unicornafl.uc_afl_fuzz.
+            NOTE: If no afl-fuzz instance is found, this function is almost identical to ql.run.
+            :param Qiling ql: The Qiling instance.
+            :param str input_file: This usually is the input file name provided by the command argument.
+            :param Callable place_input_callback: This callback is triggered every time a new child is
+            generated. It returns True if the input is accepted, or the input would be skipped.
+            :param list exits: All possible exits.
+            :param Callable validate_crash_callback: This callback is triggered every time to check if we are crashed.                     
+            :param bool always_validate: If this is set to False, validate_crash_callback will be only triggered if
+            uc_emu_start (which is called internally by afl_fuzz) returns an error. Or the validate_crash_callback will
+            be triggered every time.
+            :param int persistent_iters: Fuzz how many times before forking a new child.
+            :raises UcAflError: If something wrong happens with the fuzzer.
+        """
+
+        def _ql_afl_place_input_wrapper(uc, input_bytes, iters, data):
+            (ql, cb, _) = data
+
+            return cb(ql, input_bytes, iters)
+
+        def _ql_afl_validate_wrapper(uc, input_bytes, iters, data):
+            (ql, _, cb) = data
+
+            return cb(ql, input_bytes, iters)
+
+        data = (ql, place_input_callback, validate_crash_callback)
+        try:
+            # uc_afl_fuzz will never return non-zero value.
+            uc_afl_fuzz(ql.uc, 
+                        input_file=input_file, 
+                        place_input_callback=_ql_afl_place_input_wrapper, 
+                        exits=exits, 
+                        validate_crash_callback=_ql_afl_validate_wrapper, 
+                        always_validate=always_validate, 
+                        persistent_iters=persistent_iters,
+                        data=data)
+        except NameError as ex:
+            raise QlErrorNotImplemented("unicornafl is not installed or AFL++ is not supported on this platform") from ex
+        except UcAflError as ex:
+            if ex.errno != UC_AFL_RET_CALLED_TWICE:
+                # This one is special. Many fuzzing scripts start fuzzing in a Unicorn UC_HOOK_CODE callback and 
+                # starts execution on the current address, which results in a duplicate UC_HOOK_CODE callback. To 
+                # make unicornafl easy to use, we handle this siliently.
+                #
+                # For other exceptions, we raise them.
+                raise