Replace split("\n") with splitlines() for platform compatibility and other todos and chores

src/mvt/android/artifacts/dumpsys_adb.py

@@ -131,10 +131,17 @@ def parse(self, content: bytes) -> None:
             )
             return
 
-        # TODO: Parse AdbDebuggingManager line in output.
-        start_of_json = content.find(b"\n{") + 2
-        end_of_json = content.rfind(b"}\n") - 2
-        json_content = content[start_of_json:end_of_json].rstrip()
+        start_of_json = content.find(b"\n{")
+        if start_of_json == -1:
+            self.log.error("Unable to find ADB manager state in dumpsys output")
+            return
+
+        end_of_json = content.rfind(b"}\n")
+        if end_of_json == -1 or end_of_json <= start_of_json:
+            self.log.error("Unable to find complete ADB manager state in dumpsys output")
+            return
+
+        json_content = content[start_of_json + 2 : end_of_json - 2].rstrip()
 
         parsed = self.indented_dump_parser(json_content)
         if parsed.get("debugging_manager") is None:

src/mvt/android/artifacts/dumpsys_packages.py

@@ -14,9 +14,12 @@
 
 class DumpsysPackagesArtifact(AndroidArtifact):
     def check_indicators(self) -> None:
+        alerted_root_packages = set()
         for result in self.results:
-            # XXX: De-duplication Package detections
             if result["package_name"] in ROOT_PACKAGES:
+                if result["package_name"] in alerted_root_packages:
+                    continue
+                alerted_root_packages.add(result["package_name"])
                 self.alertstore.medium(
                     f'Found an installed package related to rooting/jailbreaking: "{result["package_name"]}"',
                     "",
@@ -188,7 +191,7 @@ def parse(self, content: str):
         package = []
 
         in_package_list = False
-        for line in content.split("\n"):
+        for line in content.splitlines():
             if line.startswith("Packages:"):
                 in_package_list = True
                 continue

src/mvt/android/artifacts/processes.py

@@ -8,7 +8,7 @@
 
 class Processes(AndroidArtifact):
     def parse(self, entry: str) -> None:
-        for line in entry.split("\n")[1:]:
+        for line in entry.splitlines()[1:]:
             proc = line.split()
 
             # Skip empty lines

src/mvt/android/artifacts/settings.py

@@ -67,11 +67,14 @@ def check_indicators(self) -> None:
                     # Check if one of the dangerous settings is using an unsafe
                     # value (different than the one specified).
                     if danger["key"] == key and danger["safe_value"] != value:
-                        self.log.warning(
-                            'Found suspicious "%s" setting "%s = %s" (%s)',
-                            namespace,
-                            key,
-                            value,
-                            danger["description"],
+                        self.alertstore.medium(
+                            f'Found suspicious "{namespace}" setting "{key} = {value}" ({danger["description"]})',
+                            "",
+                            {
+                                "namespace": namespace,
+                                "key": key,
+                                "value": value,
+                                "description": danger["description"],
+                            },
                         )
                         break

src/mvt/android/artifacts/tombstone_crashes.py

@@ -101,8 +101,7 @@ def check_indicators(self) -> None:
                 continue
 
             if result.get("command_line", []):
-                command_name = result.get("command_line")[0].split("/")[-1]
-                command_name = result["command_line"][0]
+                command_name = result["command_line"][0].split("/")[-1]
                 ioc_match = self.indicators.check_process(command_name)
                 if ioc_match:
                     self.alertstore.critical(
@@ -200,7 +199,7 @@ def _load_key_value_line(
             # eg. "Process uptime: 40s"
             tombstone[destination_key] = int(value_clean.rstrip("s"))
         elif destination_key == "command_line":
-            # XXX: Check if command line should be a single string in a list, or a list of strings.
+            # Wrap in list for consistency with protobuf format (repeated string).
             tombstone[destination_key] = [value_clean]
         else:
             tombstone[destination_key] = value_clean
@@ -262,7 +261,7 @@ def _load_timestamp_line(self, line: str, tombstone: dict) -> bool:
     @staticmethod
     def _parse_timestamp_string(timestamp: str) -> str:
         timestamp_parsed = parser.parse(timestamp)
-        # HACK: Swap the local timestamp to UTC, so keep the original time and avoid timezone conversion.
+        # Preserve the source wall-clock time while returning the project-wide ISO format.
         local_timestamp = timestamp_parsed.replace(tzinfo=datetime.timezone.utc)
         return convert_datetime_to_iso(local_timestamp)
 

src/mvt/android/cli.py

@@ -283,7 +283,13 @@ def check_androidqf(
 @click.argument("FOLDER", type=click.Path(exists=True))
 @click.pass_context
 def check_iocs(ctx, iocs, list_modules, module, folder):
-    cmd = CmdCheckIOCS(target_path=folder, ioc_files=iocs, module_name=module)
+    cmd = CmdCheckIOCS(
+        target_path=folder,
+        ioc_files=iocs,
+        module_name=module,
+        disable_version_check=_get_disable_flags(ctx)[0],
+        disable_indicator_check=_get_disable_flags(ctx)[1],
+    )
     cmd.modules = BACKUP_MODULES + BUGREPORT_MODULES + ANDROIDQF_MODULES
 
     if list_modules:

src/mvt/android/modules/androidqf/aqf_files.py

@@ -41,7 +41,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -107,16 +107,16 @@ def check_indicators(self) -> None:
                     msg = f'Found {file_type}file at suspicious path "{result["path"]}"'
                     self.alertstore.high(msg, "", result)
 
-            if result.get("sha256", "") == "":
-                continue
-
-            ioc_match = self.indicators.check_file_hash(result.get("sha256") or "")
-            if ioc_match:
-                self.alertstore.critical(
-                    ioc_match.message, "", result, matched_indicator=ioc_match.ioc
-                )
-
-            # TODO: adds SHA1 and MD5 when available in MVT
+            for hash_key in ("sha256", "sha1", "md5"):
+                file_hash = result.get(hash_key, "")
+                if not file_hash:
+                    continue
+                ioc_match = self.indicators.check_file_hash(file_hash)
+                if ioc_match:
+                    self.alertstore.critical(
+                        ioc_match.message, "", result, matched_indicator=ioc_match.ioc
+                    )
+                    break
 
     def run(self) -> None:
         if timezone := self._get_device_timezone():
@@ -131,7 +131,7 @@ def run(self) -> None:
                 data = json.loads(rawdata)
             except json.decoder.JSONDecodeError:
                 data = []
-                for line in rawdata.split("\n"):
+                for line in rawdata.splitlines():
                     if line.strip() == "":
                         continue
                     data.append(json.loads(line))
@@ -142,11 +142,11 @@ def run(self) -> None:
                         utc_timestamp = datetime.datetime.fromtimestamp(
                             file_data[ts], tz=datetime.timezone.utc
                         )
-                        # Convert the UTC timestamp to local tiem on Android device's local timezone
+                        # Convert the UTC timestamp to local time on Android device's local timezone
                         local_timestamp = utc_timestamp.astimezone(device_timezone)
 
-                        # HACK: We only output the UTC timestamp in convert_datetime_to_iso, we
-                        # set the timestamp timezone to UTC, to avoid the timezone conversion again.
+                        # Preserve the device-local wall-clock time while using
+                        # the project-wide ISO conversion helper.
                         local_timestamp = local_timestamp.replace(
                             tzinfo=datetime.timezone.utc
                         )

src/mvt/android/modules/androidqf/aqf_getprop.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -32,7 +32,7 @@ def __init__(
             log=log,
             results=results,
         )
-        self.results: list = []
+        self.results: list = [] if results is None else results
 
     def run(self) -> None:
         getprop_files = self._get_files_by_pattern("*/getprop.txt")

src/mvt/android/modules/androidqf/aqf_log_timestamps.py

@@ -27,7 +27,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/androidqf/aqf_packages.py

@@ -30,7 +30,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/androidqf/aqf_processes.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/androidqf/aqf_settings.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -32,15 +32,15 @@ def __init__(
             log=log,
             results=results,
         )
-        self.results: dict = {}
+        self.results: dict = results if results is not None else {}
 
     def run(self) -> None:
         for setting_file in self._get_files_by_pattern("*/settings_*.txt"):
             namespace = setting_file[setting_file.rfind("_") + 1 : -4]
 
             self.results[namespace] = {}
             data = self._get_file_content(setting_file)
-            for line in data.decode("utf-8").split("\n"):
+            for line in data.decode("utf-8").splitlines():
                 line = line.strip()
                 try:
                     key, value = line.split("=", 1)

src/mvt/android/modules/androidqf/base.py

@@ -23,7 +23,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/androidqf/mounts.py

@@ -32,7 +32,7 @@ def __init__(
             log=log,
             results=results,
         )
-        self.results: list = []
+        self.results: list = [] if results is None else results
 
     def run(self) -> None:
         """
@@ -66,6 +66,9 @@ def run(self) -> None:
                 # AndroidQF format: array of strings like
                 # "/dev/block/dm-12 on / type ext4 (ro,seclabel,noatime)"
                 mount_content = "\n".join(json_data)
+            else:
+                self.log.error("Expected mounts.json to contain a list of mount lines")
+                return
             self.parse(mount_content)
         except Exception as exc:
             self.log.error("Failed to parse mount information: %s", exc)

src/mvt/android/modules/androidqf/sms.py

@@ -21,10 +21,6 @@
 class SMS(AndroidQFModule):
     """
     This module analyse SMS file in backup
-
-    XXX: We should also de-duplicate this AQF module, but first we
-    need to add tests for loading encrypted SMS backups through the backup
-    sub-module.
     """
 
     def __init__(

src/mvt/android/modules/backup/base.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/backup/sms.py

@@ -20,7 +20,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/base.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_accessibility.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_activities.py

@@ -24,7 +24,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_adb_state.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_appops.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_battery_daily.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_battery_history.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_dbinfo.py

@@ -24,7 +24,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_getprop.py

@@ -22,7 +22,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_packages.py

@@ -23,7 +23,7 @@ def __init__(
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,