Adds androidqf files module (#541)

* Adds androidqf files module

* Add new files module to module list

---------

Co-authored-by: Donncha Ó Cearbhaill <donncha.ocearbhaill@amnesty.org>

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

@@ -16,6 +16,7 @@ from .packages import Packages
 from .processes import Processes
 from .settings import Settings
 from .sms import SMS
+from .files import Files
 
 ANDROIDQF_MODULES = [
     DumpsysActivities,
@@ -31,4 +32,5 @@ ANDROIDQF_MODULES = [
     Settings,
     SMS,
     DumpsysPackages,
+    Files,
 ]

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

@@ -0,0 +1,133 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2023 The MVT Authors.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+#   https://license.mvt.re/1.1/
+
+import datetime
+import json
+import logging
+from typing import Optional, Union
+
+from mvt.android.modules.androidqf.base import AndroidQFModule
+from mvt.common.utils import convert_datetime_to_iso
+
+SUSPICIOUS_PATHS = [
+    "/data/local/tmp/",
+]
+
+
+class Files(AndroidQFModule):
+    """This module analyse list of files"""
+
+    def __init__(
+        self,
+        file_path: Optional[str] = None,
+        target_path: Optional[str] = None,
+        results_path: Optional[str] = None,
+        module_options: Optional[dict] = None,
+        log: logging.Logger = logging.getLogger(__name__),
+        results: Optional[list] = None,
+    ) -> None:
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
+
+    def serialize(self, record: dict) -> Union[dict, list]:
+        records = []
+
+        for ts in set(
+            [record["access_time"], record["changed_time"], record["modified_time"]]
+        ):
+            macb = ""
+            macb += "M" if ts == record["modified_time"] else "-"
+            macb += "A" if ts == record["access_time"] else "-"
+            macb += "C" if ts == record["changed_time"] else "-"
+            macb += "-"
+
+            msg = record["path"]
+            if record["context"]:
+                msg += f" ({record['context']})"
+
+            records.append(
+                {
+                    "timestamp": ts,
+                    "module": self.__class__.__name__,
+                    "event": macb,
+                    "data": msg,
+                }
+            )
+
+        return records
+
+    def file_is_executable(self, mode_string):
+        return "x" in mode_string
+
+    def check_indicators(self) -> None:
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_file_path(result["path"])
+            if ioc:
+                result["matched_indicator"] == ioc
+                self.detected.append(result)
+                continue
+
+            # NOTE: Update with final path used for Android collector.
+            if result["path"] == "/data/local/tmp/collector":
+                continue
+
+            for suspicious_path in SUSPICIOUS_PATHS:
+                if result["path"].startswith(suspicious_path):
+                    file_type = ""
+                    if self.file_is_executable(result["mode"]):
+                        file_type = "executable "
+
+                    self.log.warning(
+                        'Found %sfile at suspicious path "%s".',
+                        file_type,
+                        result["path"],
+                    )
+                    self.detected.append(result)
+
+            if result.get("sha256", "") == "":
+                continue
+
+            ioc = self.indicators.check_file_hash(result["sha256"])
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+
+            # TODO: adds SHA1 and MD5 when available in MVT
+
+    def run(self) -> None:
+        for file in self._get_files_by_pattern("*/files.json"):
+            rawdata = self._get_file_content(file).decode("utf-8", errors="ignore")
+            try:
+                data = json.loads(rawdata)
+            except json.decoder.JSONDecodeError:
+                data = []
+                for line in rawdata.split("\n"):
+                    if line.strip() == "":
+                        continue
+                    data.append(json.loads(line))
+
+            for file_data in data:
+                for ts in ["access_time", "changed_time", "modified_time"]:
+                    if ts in file_data:
+                        file_data[ts] = convert_datetime_to_iso(
+                            datetime.datetime.fromtimestamp(
+                                file_data[ts], tz=datetime.timezone.utc
+                            )
+                        )
+
+                self.results.append(file_data)
+
+            break  # Only process the first matching file
+
+        self.log.info("Found a total of %d files", len(self.results))