Refactors dumpsys receiver parsing into an artifact

mvt/android/artifacts/dumpsys_accessibility.py

@@ -18,7 +18,7 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
                 self.detected.append(result)
                 continue
 
-    def parse(self, content: str):
+    def parse(self, content: str) -> None:
         """
         Parse the Dumpsys Accessibility section/
         Adds results to self.results (List[Dict[str, str]])

mvt/android/artifacts/dumpsys_appops.py

@@ -55,7 +55,7 @@ class DumpsysAppopsArtifact(AndroidArtifact):
                         result["package_name"],
                     )
 
-    def parse(self, output: str) -> List[Dict[str, Any]]:
+    def parse(self, output: str) -> None:
         self.results: List[Dict[str, Any]] = []
         perm = {}
         package = {}

mvt/android/artifacts/dumpsys_receivers.py

@@ -0,0 +1,116 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2023 Claudio Guarnieri.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+#   https://license.mvt.re/1.1/
+
+from .artifact import AndroidArtifact
+
+INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
+INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
+INTENT_DATA_SMS_RECEIVED = "android.intent.action.DATA_SMS_RECEIVED"
+INTENT_PHONE_STATE = "android.intent.action.PHONE_STATE"
+INTENT_NEW_OUTGOING_CALL = "android.intent.action.NEW_OUTGOING_CALL"
+
+
+class DumpsysReceiversArtifact(AndroidArtifact):
+    """
+    Parser for dumpsys receivers in the package section
+    """
+
+    def check_indicators(self) -> None:
+        for intent, receivers in self.results.items():
+            for receiver in receivers:
+                if intent == INTENT_NEW_OUTGOING_SMS:
+                    self.log.info(
+                        'Found a receiver to intercept outgoing SMS messages: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_SMS_RECEIVED:
+                    self.log.info(
+                        'Found a receiver to intercept incoming SMS messages: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_DATA_SMS_RECEIVED:
+                    self.log.info(
+                        'Found a receiver to intercept incoming data SMS message: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_PHONE_STATE:
+                    self.log.info(
+                        "Found a receiver monitoring "
+                        'telephony state/incoming calls: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_NEW_OUTGOING_CALL:
+                    self.log.info(
+                        'Found a receiver monitoring outgoing calls: "%s"',
+                        receiver["receiver"],
+                    )
+
+                if not self.indicators:
+                    continue
+
+                ioc = self.indicators.check_app_id(receiver["package_name"])
+                if ioc:
+                    receiver["matched_indicator"] = ioc
+                    self.detected.append({intent: receiver})
+                    continue
+
+    def parse(self, output: str) -> None:
+        self.results = {}
+
+        in_receiver_resolver_table = False
+        in_non_data_actions = False
+        intent = None
+        for line in output.splitlines():
+            if line.startswith("Receiver Resolver Table:"):
+                in_receiver_resolver_table = True
+                continue
+
+            if not in_receiver_resolver_table:
+                continue
+
+            if line.startswith("  Non-Data Actions:"):
+                in_non_data_actions = True
+                continue
+
+            if not in_non_data_actions:
+                continue
+
+            # If we hit an empty line, the Non-Data Actions section should be
+            # finished.
+            if line.strip() == "":
+                break
+
+            # We detect the action name.
+            if (
+                line.startswith(" " * 6)
+                and not line.startswith(" " * 8)
+                and ":" in line
+            ):
+                intent = line.strip().replace(":", "")
+                self.results[intent] = []
+                continue
+
+            # If we are not in an intent block yet, skip.
+            if not intent:
+                continue
+
+            # If we are in a block but the line does not start with 8 spaces
+            # it means the block ended a new one started, so we reset and
+            # continue.
+            if not line.startswith(" " * 8):
+                intent = None
+                continue
+
+            # If we got this far, we are processing receivers for the
+            # activities we are interested in.
+            receiver = line.strip().split(" ")[1]
+            package_name = receiver.split("/")[0]
+
+            self.results[intent].append(
+                {
+                    "package_name": package_name,
+                    "receiver": receiver,
+                }
+            )

mvt/android/modules/adb/dumpsys_receivers.py

@@ -6,18 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_receiver_resolver_table
+from mvt.android.artifacts.dumpsys_receivers import DumpsysReceiversArtifact
 
 from .base import AndroidExtraction
 
-INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
-INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
-INTENT_DATA_SMS_RECEIVED = "android.intent.action.DATA_SMS_RECEIVED"
-INTENT_PHONE_STATE = "android.intent.action.PHONE_STATE"
-INTENT_NEW_OUTGOING_CALL = "android.intent.action.NEW_OUTGOING_CALL"
 
-
-class DumpsysReceivers(AndroidExtraction):
+class DumpsysReceivers(DumpsysReceiversArtifact, AndroidExtraction):
     """This module extracts details on receivers for risky activities."""
 
     def __init__(
@@ -40,49 +34,11 @@ class DumpsysReceivers(AndroidExtraction):
 
         self.results = results if results else {}
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for intent, receivers in self.results.items():
-            for receiver in receivers:
-                if intent == INTENT_NEW_OUTGOING_SMS:
-                    self.log.info(
-                        'Found a receiver to intercept outgoing SMS messages: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_SMS_RECEIVED:
-                    self.log.info(
-                        'Found a receiver to intercept incoming SMS messages: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_DATA_SMS_RECEIVED:
-                    self.log.info(
-                        'Found a receiver to intercept incoming data SMS message: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_PHONE_STATE:
-                    self.log.info(
-                        "Found a receiver monitoring "
-                        'telephony state/incoming calls: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_NEW_OUTGOING_CALL:
-                    self.log.info(
-                        'Found a receiver monitoring outgoing calls: "%s"',
-                        receiver["receiver"],
-                    )
-
-                ioc = self.indicators.check_app_id(receiver["package_name"])
-                if ioc:
-                    receiver["matched_indicator"] = ioc
-                    self.detected.append({intent: receiver})
-                    continue
-
     def run(self) -> None:
         self._adb_connect()
 
         output = self._adb_command("dumpsys package")
-        self.results = parse_dumpsys_receiver_resolver_table(output)
+        self.parse(output)
 
         self._adb_disconnect()
+        self.log.info("Extracted receivers for %d intents", len(self.results))

mvt/android/modules/androidqf/dumpsys_receivers.py

@@ -6,19 +6,12 @@
 import logging
 from typing import Any, Dict, List, Optional, Union
 
-from mvt.android.modules.adb.dumpsys_receivers import (
-    INTENT_DATA_SMS_RECEIVED,
-    INTENT_NEW_OUTGOING_CALL,
-    INTENT_NEW_OUTGOING_SMS,
-    INTENT_PHONE_STATE,
-    INTENT_SMS_RECEIVED,
-)
-from mvt.android.parsers import parse_dumpsys_receiver_resolver_table
+from mvt.android.artifacts.dumpsys_receivers import DumpsysReceiversArtifact
 
 from .base import AndroidQFModule
 
 
-class DumpsysReceivers(AndroidQFModule):
+class DumpsysReceivers(DumpsysReceiversArtifact, AndroidQFModule):
     """This module analyse dumpsys receivers"""
 
     def __init__(
@@ -41,67 +34,16 @@ class DumpsysReceivers(AndroidQFModule):
 
         self.results = results if results else {}
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for intent, receivers in self.results.items():
-            for receiver in receivers:
-                if intent == INTENT_NEW_OUTGOING_SMS:
-                    self.log.info(
-                        'Found a receiver to intercept outgoing SMS messages: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_SMS_RECEIVED:
-                    self.log.info(
-                        'Found a receiver to intercept incoming SMS messages: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_DATA_SMS_RECEIVED:
-                    self.log.info(
-                        'Found a receiver to intercept incoming data SMS message: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_PHONE_STATE:
-                    self.log.info(
-                        "Found a receiver monitoring "
-                        'telephony state/incoming calls: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_NEW_OUTGOING_CALL:
-                    self.log.info(
-                        'Found a receiver monitoring outgoing calls: "%s"',
-                        receiver["receiver"],
-                    )
-
-                ioc = self.indicators.check_app_id(receiver["package_name"])
-                if ioc:
-                    receiver["matched_indicator"] = ioc
-                    self.detected.append({intent: receiver})
-
     def run(self) -> None:
         dumpsys_file = self._get_files_by_pattern("*/dumpsys.txt")
         if not dumpsys_file:
             return
-
-        in_receivers = False
-        lines = []
         data = self._get_file_content(dumpsys_file[0])
-        for line in data.decode("utf-8").split("\n"):
-            if line.strip() == "DUMP OF SERVICE package:":
-                in_receivers = True
-                continue
 
-            if not in_receivers:
-                continue
+        dumpsys_section = self.extract_dumpsys_section(
+            data.decode("utf-8", errors="replace"), "DUMP OF SERVICE package:"
+        )
 
-            if line.strip().startswith(
-                "------------------------------------------------------------------------------"
-            ):  # pylint: disable=line-too-long
-                break
-
-            lines.append(line.rstrip())
-
-        self.results = parse_dumpsys_receiver_resolver_table("\n".join(lines))
+        self.parse(dumpsys_section)
 
         self.log.info("Extracted receivers for %d intents", len(self.results))

mvt/android/modules/bugreport/receivers.py

@@ -6,18 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_receiver_resolver_table
+from mvt.android.artifacts.dumpsys_receivers import DumpsysReceiversArtifact
 
 from .base import BugReportModule
 
-INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
-INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
-INTENT_DATA_SMS_RECEIVED = "android.intent.action.DATA_SMS_RECEIVED"
-INTENT_PHONE_STATE = "android.intent.action.PHONE_STATE"
-INTENT_NEW_OUTGOING_CALL = "android.intent.action.NEW_OUTGOING_CALL"
 
-
-class Receivers(BugReportModule):
+class Receivers(DumpsysReceiversArtifact, BugReportModule):
     """This module extracts details on receivers for risky activities."""
 
     def __init__(
@@ -40,45 +34,6 @@ class Receivers(BugReportModule):
 
         self.results = results if results else {}
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for intent, receivers in self.results.items():
-            for receiver in receivers:
-                if intent == INTENT_NEW_OUTGOING_SMS:
-                    self.log.info(
-                        'Found a receiver to intercept outgoing SMS messages: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_SMS_RECEIVED:
-                    self.log.info(
-                        'Found a receiver to intercept incoming SMS messages: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_DATA_SMS_RECEIVED:
-                    self.log.info(
-                        'Found a receiver to intercept incoming data SMS message: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_PHONE_STATE:
-                    self.log.info(
-                        "Found a receiver monitoring "
-                        'telephony state/incoming calls: "%s"',
-                        receiver["receiver"],
-                    )
-                elif intent == INTENT_NEW_OUTGOING_CALL:
-                    self.log.info(
-                        'Found a receiver monitoring outgoing calls: "%s"',
-                        receiver["receiver"],
-                    )
-
-                ioc = self.indicators.check_app_id(receiver["package_name"])
-                if ioc:
-                    receiver["matched_indicator"] = ioc
-                    self.detected.append({intent: receiver})
-                    continue
-
     def run(self) -> None:
         content = self._get_dumpstate_file()
         if not content:
@@ -88,23 +43,9 @@ class Receivers(BugReportModule):
             )
             return
 
-        in_receivers = False
-        lines = []
-        for line in content.decode(errors="ignore").splitlines():
-            if line.strip() == "DUMP OF SERVICE package:":
-                in_receivers = True
-                continue
-
-            if not in_receivers:
-                continue
-
-            if line.strip().startswith(
-                "------------------------------------------------------------------------------"
-            ):  # pylint: disable=line-too-long
-                break
-
-            lines.append(line)
-
-        self.results = parse_dumpsys_receiver_resolver_table("\n".join(lines))
+        dumpsys_section = self.extract_dumpsys_section(
+            content.decode("utf-8", errors="replace"), "DUMP OF SERVICE package:"
+        )
 
+        self.parse(dumpsys_section)
         self.log.info("Extracted receivers for %d intents", len(self.results))

mvt/android/parsers/__init__.py

@@ -2,5 +2,3 @@
 # Copyright (c) 2021-2023 Claudio Guarnieri.
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
-
-from .dumpsys import parse_dumpsys_receiver_resolver_table

mvt/android/parsers/dumpsys.py

@@ -7,64 +7,6 @@ import re
 from typing import Any, Dict, List
 
 
-def parse_dumpsys_receiver_resolver_table(output: str) -> Dict[str, Any]:
-    results = {}
-
-    in_receiver_resolver_table = False
-    in_non_data_actions = False
-    intent = None
-    for line in output.splitlines():
-        if line.startswith("Receiver Resolver Table:"):
-            in_receiver_resolver_table = True
-            continue
-
-        if not in_receiver_resolver_table:
-            continue
-
-        if line.startswith("  Non-Data Actions:"):
-            in_non_data_actions = True
-            continue
-
-        if not in_non_data_actions:
-            continue
-
-        # If we hit an empty line, the Non-Data Actions section should be
-        # finished.
-        if line.strip() == "":
-            break
-
-        # We detect the action name.
-        if line.startswith(" " * 6) and not line.startswith(" " * 8) and ":" in line:
-            intent = line.strip().replace(":", "")
-            results[intent] = []
-            continue
-
-        # If we are not in an intent block yet, skip.
-        if not intent:
-            continue
-
-        # If we are in a block but the line does not start with 8 spaces
-        # it means the block ended a new one started, so we reset and
-        # continue.
-        if not line.startswith(" " * 8):
-            intent = None
-            continue
-
-        # If we got this far, we are processing receivers for the
-        # activities we are interested in.
-        receiver = line.strip().split(" ")[1]
-        package_name = receiver.split("/")[0]
-
-        results[intent].append(
-            {
-                "package_name": package_name,
-                "receiver": receiver,
-            }
-        )
-
-    return results
-
-
 def parse_dumpsys_package_for_details(output: str) -> Dict[str, Any]:
     """
     Parse one entry of a dumpsys package information