Merge branch 'Te-k-stalkerware'

mvt/android/modules/adb/base.py

@@ -132,7 +132,7 @@ class AndroidExtraction(MVTModule):
 
         """
         return self._adb_command(f"su -c {command}")
-    
+
     def _adb_check_file_exists(self, file):
         """Verify that a file exists.
 
@@ -166,7 +166,7 @@ class AndroidExtraction(MVTModule):
                 self._adb_download_root(remote_path, local_path, progress_callback)
             else:
                 raise Exception(f"Unable to download file {remote_path}: {e}")
-    
+
     def _adb_download_root(self, remote_path, local_path, progress_callback=None):
         try:
             # Check if we have root, if not raise an Exception.
@@ -191,7 +191,7 @@ class AndroidExtraction(MVTModule):
 
             # Delete the copy on /sdcard/.
             self._adb_command(f"rm -rf {new_remote_path}")
-            
+
         except AdbCommandFailureException as e:
             raise Exception(f"Unable to download file {remote_path}: {e}")
 

mvt/android/modules/adb/chrome_history.py

@@ -33,6 +33,14 @@ class ChromeHistory(AndroidExtraction):
             "data": f"{record['id']} - {record['url']} (visit ID: {record['visit_id']}, redirect source: {record['redirect_source']})"
         }
 
+    def check_indicators(self):
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            if self.indicators.check_domain(result["url"]):
+                self.detected.append(result)
+
     def _parse_db(self, db_path):
         """Parse a Chrome History database file.
 

mvt/android/modules/adb/packages.py

@@ -44,16 +44,24 @@ class Packages(AndroidExtraction):
         root_packages_path = os.path.join("..", "..", "data", "root_packages.txt")
         root_packages_string = pkg_resources.resource_string(__name__, root_packages_path)
         root_packages = root_packages_string.decode("utf-8").split("\n")
+        root_packages = [rp.strip() for rp in root_packages]
 
-        for root_package in root_packages:
-            root_package = root_package.strip()
-            if not root_package:
-                continue
 
-            if root_package in self.results:
+        for result in self.results:
+            if result["package_name"] in root_packages:
                 self.log.warning("Found an installed package related to rooting/jailbreaking: \"%s\"",
-                                 root_package)
+                                result["package_name"])
-                self.detected.append(root_package)
+                self.detected.append(result)
+            if result["package_name"] in self.indicators.ioc_app_ids:
+                self.log.warning("Found a malicious package name: \"%s\"",
+                                 result["package_name"])
+                self.detected.append(result)
+                for file in result["files"]:
+                    if file["sha256"] in self.indicators.ioc_files_sha256:
+                        self.log.warning("Found a malicious APK: \"%s\" %s",
+                                         result["package_name"],
+                                         file["sha256"])
+                        self.detected.append(result)
 
     def _get_files_for_package(self, package_name):
         output = self._adb_command(f"pm path {package_name}")

mvt/common/indicators.py

@@ -23,6 +23,8 @@ class Indicators:
         self.ioc_processes = []
         self.ioc_emails = []
         self.ioc_files = []
+        self.ioc_files_sha256 = []
+        self.ioc_app_ids = []
         self.ioc_count = 0
 
     def _add_indicator(self, ioc, iocs_list):
@@ -66,6 +68,12 @@ class Indicators:
             elif key == "file:name":
                 self._add_indicator(ioc=value,
                                     iocs_list=self.ioc_files)
+            elif key == "app:id":
+                self._add_indicator(ioc=value,
+                                    iocs_list=self.ioc_app_ids)
+            elif key == "file:hashes.sha256":
+                self._add_indicator(ioc=value,
+                                    iocs_list=self.ioc_files_sha256)
 
     def check_domain(self, url) -> bool:
         """Check if a given URL matches any of the provided domain indicators.