Merge pull request #318 from haeter525/update_parser_for_rizin_v0.3.x

Update the analysis library for Rizin v0.3.x

Author: pulorsok

.github/workflows/pytest.yml

@@ -31,7 +31,7 @@ jobs:
         
         
         # Install Rizin
-        sudo git clone --branch v0.2.1 https://github.com/rizinorg/rizin /opt/rizin/
+        sudo git clone --branch v0.3.4 https://github.com/rizinorg/rizin /opt/rizin/
         cd /opt/rizin/
         meson build
         ninja -C build

README.md

@@ -150,7 +150,7 @@ quark -a Ahmyth.apk -s --multi-process 4
 ```
 
 ### Upcoming unstable feature
-Now Quark also supports [Rizin](https://github.com/rizinorg/rizin) (v0.2.0 or v0.2.1, please see [#305](https://github.com/quark-engine/quark-engine/issues/305) for more detail) as one of our Android analysis frameworks. You can use option `--core-library` with `rizin` to enable the Rizin-based analysis library.
+Now Quark also supports [Rizin](https://github.com/rizinorg/rizin) as one of our Android analysis frameworks. You can use option `--core-library` with `rizin` to enable the Rizin-based analysis library.
 ```bash
 quark -a Ahmyth.apk -s --core-library rizin
 ```

quark/core/quark.py

@@ -29,7 +29,7 @@
     output_parent_function_json,
     output_parent_function_table,
 )
-from quark.utils.pprint import print_info, print_success
+from quark.utils.pprint import print_info, print_success, print_warning
 from quark.utils.weight import Weight
 
 MAX_SEARCH_LAYER = 3
@@ -334,7 +334,7 @@ def find_api_usage(self, class_name, method_name, descriptor_name):
         for method in potential_method_list:
             current_class_set = {method.class_name}
 
-            while not current_class_set.intersection(
+            while current_class_set and not current_class_set.intersection(
                 {class_name, "Ljava/lang/Object;"}
             ):
                 next_class_set = set()
@@ -423,6 +423,17 @@ def run(self, rule_obj):
                     second_api_xref_from
                 )
 
+                if not first_api_xref_from:
+                    print_warning(
+                        f"Unable to find the upperfunc of {first_api}"
+                    )
+                    continue
+                if not second_api_xref_from:
+                    print_warning(
+                        f"Unable to find the upperfunc of{second_api}"
+                    )
+                    continue
+
                 mutual_parent_function_list = self.find_intersection(
                     first_api_xref_from, second_api_xref_from
                 )

quark/core/rzapkinfo.py

@@ -18,10 +18,33 @@
 from quark.core.interface.baseapkinfo import BaseApkinfo
 from quark.core.struct.bytecodeobject import BytecodeObject
 from quark.core.struct.methodobject import MethodObject
-from quark.utils.tools import descriptor_to_androguard_format
+from quark.utils.tools import descriptor_to_androguard_format, remove_dup_list
 
 RizinCache = namedtuple("rizin_cache", "address dexindex is_imported")
 
+PRIMITIVE_TYPE_MAPPING = {
+    "void": "V",
+    "boolean": "Z",
+    "byte": "B",
+    "char": "C",
+    "short": "S",
+    "int": "I",
+    "long": "J",
+    "float": "F",
+    "double": "D",
+    "Boolean": "Ljava/lang/Boolean;",
+    "Byte": "Ljava/lang/Byte;",
+    "Character": "Ljava/lang/Character;",
+    "Short": "Ljava/lang/Short;",
+    "Integer": "Ljava/lang/Integer;",
+    "Long": "Ljava/lang/Long;",
+    "Float": "Ljava/lang/Float;",
+    "Double": "Ljava/lang/Double;",
+    "String": "Ljava/lang/String;",
+}
+
+RIZIN_ESCAPE_CHAR_LIST = ["<", ">", "$"]
+
 
 class RizinImp(BaseApkinfo):
     def __init__(
@@ -65,6 +88,29 @@ def _get_rz(self, index):
         rz.cmd("aa")
         return rz
 
+    def _convert_type_to_type_signature(self, raw_type: str):
+        if raw_type.endswith("[]"):
+            return "[" + self._convert_type_to_type_signature(raw_type[:-2])
+
+        if raw_type.startswith("["):
+            return "[" + self._convert_type_to_type_signature(raw_type[1:])
+
+        if raw_type in PRIMITIVE_TYPE_MAPPING:
+            return PRIMITIVE_TYPE_MAPPING[raw_type]
+
+        if "." in raw_type or "_" in raw_type:
+            raw_type = raw_type.replace(".", "/")
+            raw_type = raw_type.replace("_", "$")
+            return "L" + raw_type + ";"
+
+        return raw_type
+
+    @staticmethod
+    def _escape_str_in_rizin_manner(raw_str: str):
+        for c in RIZIN_ESCAPE_CHAR_LIST:
+            raw_str = raw_str.replace(c, "_")
+        return raw_str
+
     @functools.lru_cache
     def _get_methods_classified(self, dexindex):
         rz = self._get_rz(dexindex)
@@ -75,25 +121,108 @@ def _get_methods_classified(self, dexindex):
             if json_obj.get("type") not in ["FUNC", "METH"]:
                 continue
 
-            full_name = json_obj["realname"]
-            class_name, method_descriptor = full_name.split(".method.", maxsplit=1)
-            class_name = class_name + ";"
+            # -- Descriptor --
+            full_method_name = json_obj["name"]
+            raw_argument_str = next(
+                re.finditer("\\(.*\\).*", full_method_name), None
+            )
+            if raw_argument_str is None:
+                continue
+            raw_argument_str = raw_argument_str.group(0)
+
+            if raw_argument_str.endswith(")"):
+                # Convert Java lauguage type to JVM type signature
+
+                # Parse the arguments
+                raw_argument_str = raw_argument_str[1:-1]
+                arguments = [
+                    self._convert_type_to_type_signature(arg)
+                    for arg in raw_argument_str.split(", ")
+                ]
 
-            delimiter = method_descriptor.index('(')
-            methodname = method_descriptor[:delimiter]
-            descriptor = method_descriptor[delimiter:]
-            descriptor = descriptor_to_androguard_format(descriptor)
+                # Parse the return type
+                return_type = next(
+                    re.finditer(
+                        "[A-Za-zL][A-Za-z0-9L/\\;[\\]$.]+ ", full_method_name
+                    ),
+                    None,
+                )
+                if return_type is None:
+                    print(f"Unresolved method signature: {full_method_name}")
+                    continue
+                return_type = return_type.group(0).strip()
+
+                # Convert
+                raw_argument_str = (
+                    "("
+                    + " ".join(arguments)
+                    + ")"
+                    + self._convert_type_to_type_signature(return_type)
+                )
 
+            descriptor = descriptor_to_androguard_format(raw_argument_str)
+
+            # -- Method name --
+            method_name = json_obj["realname"]
+
+            # -- Is imported --
             is_imported = json_obj["is_imported"]
 
+            # -- Class name --
+            # Test if the class name is truncated
+            escaped_method_name = self._escape_str_in_rizin_manner(method_name)
+            if escaped_method_name.endswith("_"):
+                escaped_method_name = escaped_method_name[:-1]
+
+            flag_name = json_obj["flagname"]
+
+            # sym.imp.clone doesn't belong to a class
+            if flag_name == "sym.imp.clone":
+                method = MethodObject(
+                    class_name="",
+                    name="clone",
+                    descriptor="()Ljava/lang/Object;",
+                    cache=RizinCache(json_obj["vaddr"], dexindex, is_imported),
+                )
+                method_dict[""].append(method)
+                continue
+
+            if escaped_method_name not in flag_name:
+                logging.warning(
+                    f"The class name may be truncated: {json_obj['flagname']}"
+                )
+
+            # Drop the method name
+            match = None
+            for match in re.finditer("_+[A-Za-z]+", flag_name):
+                pass
+            if match is None:
+                logging.warning(
+                    f"Skip the damaged flag: {json_obj['flagname']}"
+                )
+                continue
+            match = match.group(0)
+            flag_name = flag_name[: flag_name.rfind(match)]
+
+            # Drop the prefixes sym. and imp.
+            while flag_name.startswith("sym.") or flag_name.startswith("imp."):
+                flag_name = flag_name[4:]
+
+            class_name = self._convert_type_to_type_signature(flag_name)
+
+            # Append the method
             method = MethodObject(
                 class_name=class_name,
-                name=methodname,
+                name=method_name,
                 descriptor=descriptor,
                 cache=RizinCache(json_obj["vaddr"], dexindex, is_imported),
             )
             method_dict[class_name].append(method)
 
+        # Remove duplicates
+        for class_name, method_list in method_dict.items():
+            method_dict[class_name] = remove_dup_list(method_list)
+
         return method_dict
 
     @functools.cached_property
@@ -326,17 +455,12 @@ def superclass_relationships(self) -> Dict[str, Set[str]]:
 
             rz = self._get_rz(dex_index)
 
-            hierarchy_graph = rz.cmd("icg").split("\n")
-
-            for element in hierarchy_graph:
-                if element.startswith("age"):
-                    element_part = element.split()
-                    for index, class_name in enumerate(element_part):
-                        if not class_name.endswith(";"):
-                            element_part[index] = class_name + ";"
+            class_info_list = rz.cmdj("icj")
+            for class_info in class_info_list:
+                class_name = class_info["classname"]
+                super_class = class_info["super"]
 
-                    for subclass in element_part[2:]:
-                        hierarchy_dict[subclass].add(element_part[1])
+                hierarchy_dict[class_name].add(super_class)
 
         return hierarchy_dict
 
@@ -348,16 +472,12 @@ def subclass_relationships(self) -> Dict[str, Set[str]]:
 
             rz = self._get_rz(dex_index)
 
-            hierarchy_graph = rz.cmd("icg").split("\n")
-
-            for element in hierarchy_graph:
-                if element.startswith("age"):
-                    element_part = element.split()
-                    for index, class_name in enumerate(element_part):
-                        if not class_name.endswith(";"):
-                            element_part[index] = class_name + ";"
+            class_info_list = rz.cmdj("icj")
+            for class_info in class_info_list:
+                class_name = class_info["classname"]
+                super_class = class_info["super"]
 
-                    hierarchy_dict[element_part[1]].update(element_part[2:])
+                hierarchy_dict[super_class].add(class_name)
 
         return hierarchy_dict
 
@@ -380,8 +500,8 @@ def _parse_smali(smali: str) -> BytecodeObject:
         mnemonic, args = smali.split(maxsplit=1)  # Split into twe parts
 
         # invoke-kind instruction may left method index at the last
-        if mnemonic.startswith("invoke"):
-            args = args[: args.rfind(" ;")]
+        # if mnemonic.startswith("invoke"):
+        #     args = args[: args.rfind(" ;")]
 
         args = [arg.strip() for arg in re.split("[{},]+", args) if arg]
 
@@ -392,7 +512,7 @@ def _parse_smali(smali: str) -> BytecodeObject:
             args = args[:-1]
 
             if mnemonic.startswith("invoke"):
-                parameter = re.sub(r"\.", ";->", parameter, count=1)
+                parameter = re.sub(r"\.", "->", parameter, count=1)
 
         register_list = []
         # Ranged registers

tests/core/test_apkinfo.py

@@ -117,7 +117,10 @@ def test_android_apis(self, apkinfo):
             ),
         }
 
-        assert len(apkinfo.android_apis) == 1270
+        if apkinfo.core_library == "androguard":
+            assert len(apkinfo.android_apis) == 1270
+        elif apkinfo.core_library == "rizin":
+            assert len(apkinfo.android_apis) == 1269
         assert api.issubset(apkinfo.android_apis)
 
     def test_custom_methods(self, apkinfo):
@@ -133,7 +136,10 @@ def test_custom_methods(self, apkinfo):
                 "()V",
             ),
         }
-        assert len(apkinfo.custom_methods) == 3999
+        if apkinfo.core_library == "androguard":
+            assert len(apkinfo.custom_methods) == 3999
+        elif apkinfo.core_library == "rizin":
+            assert len(apkinfo.custom_methods) == 3990
         assert test_custom_method.issubset(apkinfo.custom_methods)
 
     def test_all_methods(self, apkinfo):
@@ -153,7 +159,7 @@ def test_all_methods(self, apkinfo):
         if apkinfo.core_library == "androguard":
             assert len(apkinfo.all_methods) == 5452
         elif apkinfo.core_library == "rizin":
-            assert len(apkinfo.all_methods) == 5273
+            assert len(apkinfo.all_methods) == 5260
 
         assert test_custom_method.issubset(apkinfo.all_methods)
 
@@ -168,20 +174,21 @@ def test_find_method(self, apkinfo):
         assert str(result.descriptor) == "(Z)V"
 
     def test_upperfunc(self, apkinfo):
-        api = apkinfo.find_method("Ljava/lang/reflect/Field;", "setAccessible", "(Z)V")
-
-        expect_function = MethodObject(
-            (
-                "Landroid/support/v4/widget/SlidingPaneLayout$"
-                "SlidingPanelLayoutImplJB;"
-            ),
+        api = apkinfo.find_method(
+            "Lcom/example/google/service/ContactsHelper;",
             "<init>",
-            "()V",
+            "(Landroid/content/Context;)V",
+        )
+
+        expect_function = apkinfo.find_method(
+            "Lcom/example/google/service/SMSReceiver;",
+            "isContact",
+            "(Ljava/lang/String;)Ljava/lang/Boolean;",
         )
 
-        upper = list(apkinfo.upperfunc(api))[0]
+        upper_methods = list(apkinfo.upperfunc(api))
 
-        assert upper == expect_function
+        assert expect_function in upper_methods
 
     def test_lowerfunc(self, apkinfo):
         method = apkinfo.find_method(
@@ -238,17 +245,17 @@ def test_get_method_bytecode(self, apkinfo):
 
     def test_lowerfunc(self, apkinfo):
         method = apkinfo.find_method(
-            "Lcom/example/google/service/WebServiceCalling;",
-            "Send",
-            "(Landroid/os/Handler; Ljava/lang/String;)V",
+            "Lcom/example/google/service/SMSReceiver;",
+            "isContact",
+            "(Ljava/lang/String;)Ljava/lang/Boolean;",
         )
 
         expect_method = apkinfo.find_method(
-            "Ljava/lang/StringBuilder;",
-            "append",
-            "(Ljava/lang/String;)Ljava/lang/StringBuilder;",
+            "Lcom/example/google/service/ContactsHelper;",
+            "<init>",
+            "(Landroid/content/Context;)V",
         )
-        expect_offset = 42
+        expect_offset = 10
 
         upper_methods = apkinfo.lowerfunc(method)