diff --git a/src/mvt/android/artifacts/dumpsys_accessibility.py b/src/mvt/android/artifacts/dumpsys_accessibility.py index 3d8fa6c2b..da46043c3 100644 --- a/src/mvt/android/artifacts/dumpsys_accessibility.py +++ b/src/mvt/android/artifacts/dumpsys_accessibility.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ import re +from typing import Any from .artifact import AndroidArtifact @@ -20,10 +21,10 @@ def check_indicators(self) -> None: continue self.alertstore.medium( - f'Found accessibility service: "{result["service"]}"', + f'Found accessibility service: "{result["component"]}"', "", result, - ) + ) def parse(self, content: str) -> None: """ @@ -33,41 +34,69 @@ def parse(self, content: str) -> None: :param content: content of the accessibility section (string) """ - # "Old" syntax - in_services = False + self.results: list[dict[str, Any]] = [] + services: dict[tuple[int | None, str], dict] = {} + user_id: int | None = None + state: str | None = None + for line in content.splitlines(): - if line.strip().startswith("installed services:"): - in_services = True + user_match = re.search(r"attributes:\{id=(\d+)", line) + if user_match: + user_id = int(user_match.group(1)) + + stripped = line.strip() + state_match = re.match( + r"(?i)(installed|enabled|binding|bound|crashed) services\s*:\s*\{(.*)", + stripped, + ) + if state_match: + state = state_match.group(1).lower() + inline = state_match.group(2) + for component in re.findall( + r"\{?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?\}?", inline + ): + service = services.setdefault( + (user_id, component), self._new_service(component, user_id) + ) + service[self._state_field(state)] = True + service["accessibility_tool"] = "(A11yTool)" in inline continue - if not in_services: + if not state: continue - - if line.strip() == "}": - # At end of installed services - break - - service = line.split(":")[1].strip() - - self.results.append( - { - "package_name": service.split("/")[0], - "service": service, - } + if stripped == "}" or stripped.startswith("AccessibilityInputFilter"): + state = None + continue + component_match = re.search( + r"(?:\d+\s*:\s*)?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?", + stripped, ) - - # "New" syntax - AOSP >= 14 (?) - # Looks like: - # Enabled services:{{com.azure.authenticator/com.microsoft.brooklyn.module.accessibility.BrooklynAccessibilityService}, {com.agilebits.onepassword/com.agilebits.onepassword.filling.accessibility.FillingAccessibilityService}} - - for line in content.splitlines(): - if line.strip().startswith("Enabled services:"): - matches = re.finditer(r"{([^{]+?)}", line) - - for match in matches: - # Each match is in format: / - package_name, _, service = match.group(1).partition("/") - - self.results.append( - {"package_name": package_name, "service": service} - ) + if component_match: + component = component_match.group(1) + service = services.setdefault( + (user_id, component), self._new_service(component, user_id) + ) + service[self._state_field(state)] = True + service["accessibility_tool"] = "(A11yTool)" in stripped + + self.results.extend(services.values()) + + @staticmethod + def _state_field(state: str) -> str: + return {"binding": "binding", "bound": "bound"}.get(state, state) + + @staticmethod + def _new_service(component: str, user_id: int | None) -> dict: + package_name, service_name = component.split("/", 1) + return { + "user_id": user_id, + "component": component, + "package_name": package_name, + "service_name": service_name, + "installed": False, + "enabled": False, + "binding": False, + "bound": False, + "crashed": False, + "accessibility_tool": False, + } diff --git a/src/mvt/android/artifacts/dumpsys_adb.py b/src/mvt/android/artifacts/dumpsys_adb.py index e5438c9bb..cbd80e21c 100644 --- a/src/mvt/android/artifacts/dumpsys_adb.py +++ b/src/mvt/android/artifacts/dumpsys_adb.py @@ -6,6 +6,7 @@ import base64 import binascii import hashlib +import re from .artifact import AndroidArtifact @@ -98,6 +99,34 @@ def parse_xml(self, xml_data): return keystore + def parse_binary_xml(self, data: bytes) -> list[dict]: + """Recover ADB key records from Android binary XML (ABX). + + Some dumpstate implementations embed ABX in a text stream and replace + binary token bytes. The public key remains intact, while unavailable + numeric metadata is represented as ``None`` rather than corrupt text. + """ + keystore = [] + seen = set() + for match in re.finditer( + rb"(? dict: if b" " in user_key: @@ -118,7 +147,7 @@ def calculate_key_info(user_key: bytes) -> dict: return { "user": user.decode("utf-8"), "fingerprint": key_fingerprint_colon, - "key": key_base64, + "key": key_base64.decode("ascii", errors="replace"), } def check_indicators(self) -> None: @@ -175,12 +204,23 @@ def parse(self, content: bytes) -> None: # Keystore is in XML format on some devices and we need to parse it if keystore_data and keystore_data.startswith(b" ModuleSerializedResult: continue for entry in perm["entries"]: - if "timestamp" in entry: + if entry.get("timestamp"): records.append( { "timestamp": entry["timestamp"], "module": self.__class__.__name__, - "event": entry["access"], + "event": entry["event"], "data": f"{result['package_name']} access to " - f"{perm['name']}: {entry['access']}", + f"{perm['name']}: {entry['event']}", } ) @@ -51,7 +52,7 @@ def check_indicators(self) -> None: continue # We use a placeholder entry to create a basic alert even without permission entries. - placeholder_entry = {"access": "Unknown", "timestamp": ""} + placeholder_entry = {"event": "unknown", "timestamp": ""} for perm in result["permissions"]: if ( @@ -60,12 +61,12 @@ def check_indicators(self) -> None: ): for entry in sorted( perm["entries"] or [placeholder_entry], - key=lambda x: x["timestamp"], + key=lambda x: x.get("timestamp") or "", ): cleaned_result = result.copy() cleaned_result["permissions"] = [perm] self.alertstore.medium( - f"Package '{result['package_name']}' had risky permission '{perm['name']}' set to '{entry['access']}' at {entry['timestamp']}", + f"Package '{result['package_name']}' had risky permission '{perm['name']}' set to '{entry['event']}' at {entry['timestamp']}", entry["timestamp"], cleaned_result, ) @@ -73,111 +74,151 @@ def check_indicators(self) -> None: elif result["package_name"] in RISKY_PACKAGES: for entry in sorted( perm["entries"] or [placeholder_entry], - key=lambda x: x["timestamp"], + key=lambda x: x.get("timestamp") or "", ): cleaned_result = result.copy() cleaned_result["permissions"] = [perm] self.alertstore.medium( - f"Risky package '{result['package_name']}' had '{perm['name']}' permission set to '{entry['access']}' at {entry['timestamp']}", + f"Risky package '{result['package_name']}' had '{perm['name']}' permission set to '{entry['event']}' at {entry['timestamp']}", entry["timestamp"], cleaned_result, ) def parse(self, output: str) -> None: - # self.results: List[Dict[str, Any]] = [] - perm: dict[str, Any] = {} - package: dict[str, Any] = {} - entry: dict[str, Any] = {} - uid = None + self.results: list[dict[str, Any]] = [] + permission: dict[str, Any] | None = None + package: dict[str, Any] | None = None + uid: str | None = None + uid_details: dict[str, Any] = {} + attribution: str | None = None in_packages = False + def finish_permission() -> None: + nonlocal permission + if package is not None and permission is not None: + package["permissions"].append(permission) + permission = None + + def finish_package() -> None: + nonlocal package + finish_permission() + if package is not None: + self.results.append(package) + package = None + for line in output.splitlines(): - if line.startswith(" Uid 0:"): + uid_match = re.match(r"^ Uid ([^:]+):$", line) + if uid_match: in_packages = True + finish_package() + uid = uid_match.group(1) + uid_details = { + "uid_state": None, + "capability": None, + "app_widget_visible": None, + "default_modes": {}, + } + continue if not in_packages: continue - if line.startswith(" Uid "): - uid = line[6:-1] - if entry: - perm["entries"].append(entry) - entry = {} - if package: - if perm: - package["permissions"].append(perm) - - perm = {} - self.results.append(package) - package = {} + uid_property = re.match( + r"^ (state|capability|appWidgetVisible)=(.*)$", line + ) + if uid_property: + key = { + "state": "uid_state", + "appWidgetVisible": "app_widget_visible", + }.get(uid_property.group(1), uid_property.group(1)) + value: Any = uid_property.group(2) + if value in ("true", "false"): + value = value == "true" + uid_details[key] = value continue - if line.startswith(" Package "): - if entry: - perm["entries"].append(entry) - entry = {} - - if package: - if perm: - package["permissions"].append(perm) - - perm = {} - self.results.append(package) + default_mode = re.match(r"^ ([A-Z0-9_]+): mode=([^\s]+)", line) + if default_mode and package is None: + uid_details["default_modes"][default_mode.group(1)] = ( + default_mode.group(2) + ) + continue + if line.startswith(" Package "): + finish_package() package = { "package_name": line[12:-1], "permissions": [], "uid": uid, + **uid_details, } continue - if package and line.startswith(" ") and line[6] != " ": - if entry: - perm["entries"].append(entry) - entry = {} - if perm: - package["permissions"].append(perm) - perm = {} - - perm["name"] = line.split()[0] - perm["entries"] = [] - if len(line.split()) > 1: - perm["access"] = line.split()[1][1:-2] - + operation_match = re.match( + r"^ ([A-Z0-9_]+)(?: \(([^)]+)\))?:\s*$", line + ) + if package is not None and operation_match: + finish_permission() + permission = { + "name": operation_match.group(1), + "mode": operation_match.group(2), + "entries": [], + } + attribution = None continue - if line.startswith(" "): - # Permission entry like: - # Reject: [fg-s]2021-05-19 22:02:52.054 (-314d1h25m2s33ms) - access_type = line.split(":")[0].strip() - if access_type not in ["Access", "Reject"]: - # Skipping invalid access type. Some entries are not in the format we expect - continue - - if entry: - perm["entries"].append(entry) - entry = {} - - entry["access"] = access_type - entry["type"] = line[line.find("[") + 1 : line.find("]")] + attribution_match = re.match(r"^\s{8,}([^=]+)=\[$", line) + if attribution_match: + attribution = attribution_match.group(1).strip() + continue + if line.strip() == "]": + attribution = None + continue - try: - entry["timestamp"] = convert_datetime_to_iso( - datetime.strptime( - line[line.find("]") + 1 : line.find("(")].strip(), - "%Y-%m-%d %H:%M:%S.%f", - ) + if permission is None: + continue + event_match = re.match( + r"^\s*(Access|Reject):\s*\[([^]]+)\]\s*" + r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+)\s*" + r"(\([^)]*\))?(?:\s+duration=([^\s]+))?", + line, + ) + running_match = re.match( + r"^\s*Running start at:\s*(\S+(?: \S+)?)", + line, + ) + if event_match: + entry = { + "event": event_match.group(1).lower(), + "access": event_match.group(1), + "uid_state": event_match.group(2), + "timestamp": convert_datetime_to_iso( + datetime.strptime(event_match.group(3), "%Y-%m-%d %H:%M:%S.%f") + ), + "relative_time": event_match.group(4), + "duration": event_match.group(5), + "attribution": attribution, + } + permission["entries"].append(entry) + elif running_match: + raw_start = running_match.group(1) + timestamp = None + if re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+", raw_start): + timestamp = convert_datetime_to_iso( + datetime.strptime(raw_start, "%Y-%m-%d %H:%M:%S.%f") ) - except ValueError: - # Invalid date format - pass - - if line.strip() == "": - break - - if entry: - perm["entries"].append(entry) - if perm: - package["permissions"].append(perm) - if package: - self.results.append(package) + permission["entries"].append( + { + "event": "running", + "access": "Running", + "uid_state": None, + "timestamp": timestamp, + "relative_time": raw_start + if raw_start.startswith("+") + else None, + "duration": None, + "attribution": attribution, + } + ) + + finish_package() diff --git a/src/mvt/android/artifacts/dumpsys_battery_daily.py b/src/mvt/android/artifacts/dumpsys_battery_daily.py index c0f2c5e0e..2ccb6b705 100644 --- a/src/mvt/android/artifacts/dumpsys_battery_daily.py +++ b/src/mvt/android/artifacts/dumpsys_battery_daily.py @@ -18,18 +18,18 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact): def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: action = record.get("action", "update") package_name = record["package_name"] - vers = record["vers"] + vers = record["version_code"] - if vers == "0": + if vers == 0: data = f"Recorded uninstall of package {package_name} (vers 0)" elif action == "downgrade": - prev_vers = record.get("previous_vers", "unknown") + prev_vers = record.get("previous_version_code", "unknown") data = f"Recorded downgrade of package {package_name} from vers {prev_vers} to vers {vers}" else: data = f"Recorded update of package {package_name} with vers {vers}" return { - "timestamp": record["from"], + "timestamp": record["period_start"], "module": self.__class__.__name__, "event": "battery_daily", "data": data, @@ -48,6 +48,7 @@ def check_indicators(self) -> None: continue def parse(self, output: str) -> None: + self.results = [] daily = None daily_updates: list[dict[str, Any]] = [] records: list[dict[str, Any]] = [] @@ -59,7 +60,10 @@ def parse(self, output: str) -> None: timeframe = line[13:].strip() date_from, date_to = timeframe.strip(":").split(" to ", 1) - daily = {"from": date_from[0:10], "to": date_to[0:10]} + daily = { + "period_start": self._format_daily_timestamp(date_from), + "period_end": self._format_daily_timestamp(date_to), + } continue if not daily: @@ -70,21 +74,30 @@ def parse(self, output: str) -> None: line = line.strip().replace("Update ", "") package_name, vers = line.split(" ", 1) - vers_nr = vers.split("=", 1)[1] + vers_raw = vers.split("=", 1)[1] + try: + version_code: int | str = int(vers_raw) + except ValueError: + version_code = vers_raw already_seen = False for update in daily_updates: - if package_name == update["package_name"] and vers_nr == update["vers"]: + if ( + package_name == update["package_name"] + and version_code == update["version_code"] + ): + update["occurrences"] += 1 already_seen = True break if not already_seen: update_record: dict[str, Any] = { "action": "update", - "from": daily["from"], - "to": daily["to"], + "period_start": daily["period_start"], + "period_end": daily["period_end"], "package_name": package_name, - "vers": vers_nr, + "version_code": version_code, + "occurrences": 1, } daily_updates.append(update_record) @@ -95,26 +108,31 @@ def parse(self, output: str) -> None: self._detect_uninstalls_and_downgrades(records) self.results.extend(records) - def _detect_uninstalls_and_downgrades( - self, records: list[dict[str, Any]] - ) -> None: + @staticmethod + def _format_daily_timestamp(value: str) -> str: + if len(value) >= 19 and value[10] == "-": + return f"{value[:10]} {value[11:].replace('-', ':')}" + return value + + def _detect_uninstalls_and_downgrades(self, records: list[dict[str, Any]]) -> None: package_versions: dict[str, int] = {} for record in sorted( records, key=lambda record: ( - record["from"], - record["to"], + record["period_start"], + record["period_end"], record["package_name"], ), ): package_name = record["package_name"] - vers_nr = record["vers"] + vers_nr = record["version_code"] - if vers_nr == "0": + if vers_nr == 0: + record["action"] = "uninstall" self.alertstore.medium( f"Detected uninstall of package {package_name} (vers 0)", - record["from"], + record["period_start"], record, ) package_versions.pop(package_name, None) @@ -128,11 +146,11 @@ def _detect_uninstalls_and_downgrades( previous_vers = package_versions.get(package_name) if previous_vers is not None and current_vers < previous_vers: record["action"] = "downgrade" - record["previous_vers"] = str(previous_vers) + record["previous_version_code"] = previous_vers self.alertstore.medium( f"Detected downgrade of package {package_name} " f"from vers {previous_vers} to vers {current_vers}", - record["from"], + record["period_start"], record, ) diff --git a/src/mvt/android/artifacts/dumpsys_battery_history.py b/src/mvt/android/artifacts/dumpsys_battery_history.py index 77d9d3c1b..6cb36f187 100644 --- a/src/mvt/android/artifacts/dumpsys_battery_history.py +++ b/src/mvt/android/artifacts/dumpsys_battery_history.py @@ -1,96 +1,177 @@ # Mobile Verification Toolkit (MVT) -# Copyright (c) 2021-2023 The MVT Authors. +# Copyright (c) 2021-2026 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 re + +from mvt.common.utils import convert_datetime_to_iso + from .artifact import AndroidArtifact class DumpsysBatteryHistoryArtifact(AndroidArtifact): - """ - Parser for dumpsys dattery history events. - """ + """Parser for package-related events in dumpsys batterystats history.""" def check_indicators(self) -> None: if not self.indicators: return - for result in self.results: - ioc_match = self.indicators.check_app_id(result["package_name"]) + package_name = result.get("package_name") + if not package_name: + continue + ioc_match = self.indicators.check_app_id(package_name) if ioc_match: self.alertstore.critical( ioc_match.message, "", result, matched_indicator=ioc_match.ioc ) - continue + + @staticmethod + def _parse_wall_time(value: str) -> datetime.datetime | None: + if re.fullmatch(r"\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+", value): + value = f"1900-{value}" + for date_format in ( + "%Y-%m-%d-%H-%M-%S-%f", + "%Y-%m-%d-%H-%M-%S", + "%Y-%m-%d %H:%M:%S.%f", + ): + try: + return datetime.datetime.strptime(value, date_format) + except ValueError: + pass + return None + + @staticmethod + def _elapsed_seconds(value: str) -> float | None: + if not value.startswith("+"): + return None + units = {"d": 86400, "h": 3600, "m": 60, "s": 1, "ms": 0.001} + total = 0.0 + for number, unit in re.findall(r"(\d+)(ms|d|h|m|s)", value): + total += int(number) * units[unit] + return total + + @staticmethod + def _package_from_name(name: str) -> str | None: + clean = name.removeprefix("*walarm*:").removeprefix("*alarm*:") + slash_parts = clean.split("/") + if len(slash_parts) > 1: + first = slash_parts[0].lstrip("@") + if first.startswith(("com.", "org.", "net.")): + return first + for part in reversed(slash_parts[1:]): + candidate = part.lstrip("@").split(":", 1)[0] + if candidate.startswith(("com.", "org.", "net.")): + return candidate + parts = clean.split(".") + package_parts = [] + for part in parts: + if part and part[0].islower(): + package_parts.append(part) + else: + break + return ".".join(package_parts) if len(package_parts) >= 2 else None + + @staticmethod + def _normalize_service(service: str) -> str: + # WorkManager decorates jobs with one or more scheduler prefixes. + if "@" in service: + candidates = [part for part in service.split("@") if "/" in part] + if candidates: + return candidates[-1] + return service def parse(self, data: str) -> None: + self.results: list[dict[str, str | None]] = [] + anchor_time: datetime.datetime | None = None + anchor_elapsed = 0.0 + has_history_heading = any( + line.startswith("Battery History") for line in data.splitlines() + ) + in_history = not has_history_heading + for line in data.splitlines(): - if line.startswith("Battery History "): + stripped = line.strip() + if line.startswith("Battery History"): + if in_history: + break + in_history = True continue - - if line.strip() == "": + if not in_history: + continue + if has_history_heading and not stripped: break + reset_match = re.search(r"(?:RESET:)?TIME:\s*(\S+)", stripped) + if reset_match: + parsed_time = self._parse_wall_time(reset_match.group(1)) + if parsed_time is not None: + elapsed_token = stripped.split()[0] + anchor_elapsed = self._elapsed_seconds(elapsed_token) or 0.0 + anchor_time = parsed_time + continue + + fields = stripped.split() + if not fields: + continue + if len(fields) > 1 and re.fullmatch(r"\d{2}-\d{2}", fields[0]): + time_elapsed = " ".join(fields[:2]) + line_time = self._parse_wall_time(time_elapsed) + elapsed = None + else: + time_elapsed = fields[0] + elapsed = self._elapsed_seconds(time_elapsed) + line_time = None + + timestamp = line_time + if timestamp is None and anchor_time is not None and elapsed is not None: + timestamp = anchor_time + datetime.timedelta( + seconds=elapsed - anchor_elapsed + ) + + def add( + event: str, uid: str, service: str, package_name: str | None + ) -> None: + self.results.append( + { + "time_elapsed": time_elapsed, + "timestamp": convert_datetime_to_iso(timestamp) + if timestamp + else None, + "event": event, + "uid": uid, + "package_name": package_name, + "service": service, + } + ) - time_parts = line.strip().split() - time_elapsed = time_parts[0] - if ( - len(time_parts) > 1 - and len(time_parts[0]) == 5 - and time_parts[0][2] == "-" - and ":" in time_parts[1] + for sign, uid, raw_service in re.findall( + r"([+-])job=([^:\s]+):\"([^\"]+)\"", line ): - time_elapsed = " ".join(time_parts[:2]) - - event = "" - if line.find("+job") > 0: - event = "start_job" - payload = line.split("+job=", 1)[1] - uid, separator, service = payload.partition(":") - if not separator: - continue - service = service.strip().strip('"') - package_name = service.split("/")[0] - elif line.find("-job") > 0: - event = "end_job" - payload = line.split("-job=", 1)[1] - uid, separator, service = payload.partition(":") - if not separator: - continue - service = service.strip().strip('"') - package_name = service.split("/")[0] - elif line.find("+running +wake_lock=") > 0: - payload = line.split("+running +wake_lock=", 1)[1] - uid, separator, _ = payload.partition(":") - if not separator: - continue - event = "wake" - service = ( - line[line.find("*walarm*:") + 9 :].split(" ")[0].strip('"').strip() + service = self._normalize_service(raw_service) + add( + "start_job" if sign == "+" else "end_job", + uid, + service, + self._package_from_name(service) or service.split("/", 1)[0], ) - if service == "" or "/" not in service: - continue - - package_name = service.split("/")[0] - elif (line.find("+top=") > 0) or (line.find("-top") > 0): - if line.find("+top=") > 0: - event = "start_top" - top_pos = line.find("+top=") - else: - event = "end_top" - top_pos = line.find("-top=") - colon_pos = top_pos + line[top_pos:].find(":") - uid = line[top_pos + 5 : colon_pos] - service = "" - package_name = line[colon_pos + 1 :].strip('"') - else: - continue - self.results.append( - { - "time_elapsed": time_elapsed, - "event": event, - "uid": uid, - "package_name": package_name, - "service": service, - } - ) + for sign, uid, package_name in re.findall( + r"([+-])top=([^:\s]+):\"([^\"]+)\"", line + ): + add( + "start_top" if sign == "+" else "end_top", + uid, + "", + package_name, + ) + + wake_match = re.search(r"\+wake_lock=([^:\s]+):\"([^\"]+)\"", line) + if wake_match: + wake_name = wake_match.group(2) + add( + "wake", + wake_match.group(1), + wake_name, + self._package_from_name(wake_name), + ) diff --git a/src/mvt/android/artifacts/dumpsys_dbinfo.py b/src/mvt/android/artifacts/dumpsys_dbinfo.py index 050582df2..0972b438d 100644 --- a/src/mvt/android/artifacts/dumpsys_dbinfo.py +++ b/src/mvt/android/artifacts/dumpsys_dbinfo.py @@ -29,16 +29,27 @@ def check_indicators(self) -> None: def parse(self, output: str) -> None: rxp = re.compile( - r".*\[((?:[0-9]{4}-)?[0-9]{2}-[0-9]{2} " - r"[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\]\s*" - r"(?:\[Pid:\((\d+)\)\])?([\w-]+).*?sql=\"(.+?)\"" - ) # pylint: disable=line-too-long + r"^\s*\d+:\s*\[((?:\d{4}-)?\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3})\]\s*" + r"(?:\[Pid:\((\d+)\)\])?\s*([\w-]+) took (\d+)ms - ([^,]+)," + r"\s*sql=\"(.*)\"(?:, path=(.*))?$" + ) - pool = None + pool: str | None = None + connection_number: int | None = None + is_primary: bool | None = None in_operations = False for line in output.splitlines(): if line.startswith("Connection pool for "): pool = line.replace("Connection pool for ", "").rstrip(":") + in_operations = False + + connection_match = re.match(r"\s+Connection #(\d+):", line) + if connection_match: + connection_number = int(connection_match.group(1)) + is_primary = None + + if line.strip().startswith("isPrimaryConnection:"): + is_primary = line.strip().split(":", 1)[1].strip() == "true" if not pool: continue @@ -52,7 +63,6 @@ def parse(self, output: str) -> None: if not line.startswith(" "): in_operations = False - pool = None continue match = rxp.match(line) @@ -60,11 +70,15 @@ def parse(self, output: str) -> None: continue result = { - "isodate": match.group(1), + "timestamp": match.group(1), + "pid": int(match.group(2)) if match.group(2) else None, "action": match.group(3), - "sql": match.group(4), - "path": pool, + "duration_ms": int(match.group(4)), + "status": match.group(5), + "sql": match.group(6), + "path": match.group(7) or pool, + "pool_path": pool, + "connection_number": connection_number, + "is_primary": is_primary, } - if match.group(2): - result["pid"] = match.group(2) self.results.append(result) diff --git a/src/mvt/android/artifacts/dumpsys_package_activities.py b/src/mvt/android/artifacts/dumpsys_package_activities.py index 31c6bbdb9..443d18fe1 100644 --- a/src/mvt/android/artifacts/dumpsys_package_activities.py +++ b/src/mvt/android/artifacts/dumpsys_package_activities.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ from .artifact import AndroidArtifact +from .package_resolvers import parse_resolver_table class DumpsysPackageActivitiesArtifact(AndroidArtifact): @@ -19,67 +20,11 @@ def check_indicators(self) -> None: ) continue - def parse(self, content: str): + def parse(self, content: str) -> None: """ Parse the Dumpsys Package section for activities Adds results to self.results :param content: content of the package section (string) """ - self.results = [] - - in_activity_resolver_table = False - in_non_data_actions = False - intent = None - for line in content.splitlines(): - if line.startswith("Activity Resolver Table:"): - in_activity_resolver_table = True - continue - - if not in_activity_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(":", "") - 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. - activity = line.strip().split(" ")[1] - package_name = activity.split("/")[0] - - self.results.append( - { - "intent": intent, - "package_name": package_name, - "activity": activity, - } - ) + self.results = parse_resolver_table(content, "Activity") diff --git a/src/mvt/android/artifacts/dumpsys_packages.py b/src/mvt/android/artifacts/dumpsys_packages.py index bef774446..1194a75dc 100644 --- a/src/mvt/android/artifacts/dumpsys_packages.py +++ b/src/mvt/android/artifacts/dumpsys_packages.py @@ -1,10 +1,10 @@ # Mobile Verification Toolkit (MVT) -# Copyright (c) 2021-2023 The MVT Authors. +# Copyright (c) 2021-2026 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 re -from typing import Any, Dict, List, Optional +from typing import Any from mvt.android.utils import ROOT_PACKAGES from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult @@ -12,222 +12,244 @@ from .artifact import AndroidArtifact +def _value(raw: str) -> Any: + if raw == "null": + return None + if raw in ("true", "false"): + return raw == "true" + try: + return int(raw) + except ValueError: + return raw + + class DumpsysPackagesArtifact(AndroidArtifact): def check_indicators(self) -> None: alerted_root_packages = set() for result in self.results: - if result["package_name"] in ROOT_PACKAGES: - if result["package_name"] in alerted_root_packages: - continue - alerted_root_packages.add(result["package_name"]) + package_name = result["package_name"] + if ( + package_name in ROOT_PACKAGES + and package_name not in alerted_root_packages + ): + alerted_root_packages.add(package_name) self.alertstore.medium( - f'Found an installed package related to rooting/jailbreaking: "{result["package_name"]}"', + f'Found an installed package related to rooting/jailbreaking: "{package_name}"', "", result, ) - continue - if not self.indicators: continue - - ioc_match = self.indicators.check_app_id(result.get("package_name", "")) + ioc_match = self.indicators.check_app_id(package_name) if ioc_match: self.alertstore.critical( ioc_match.message, "", result, matched_indicator=ioc_match.ioc ) def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: - records = [] timestamps = [ - {"event": "package_install", "timestamp": record["timestamp"]}, + ("package_install", record.get("timestamp")), + ("package_last_update", record.get("last_update_time")), + ] + timestamps.extend( + ("package_first_install", user.get("first_install_time")) + for user in record.get("users", []) + ) + return [ { - "event": "package_first_install", - "timestamp": record["first_install_time"], - }, - {"event": "package_last_update", "timestamp": record["last_update_time"]}, + "timestamp": timestamp, + "module": self.__class__.__name__, + "event": event, + "data": f"Install or update of package {record['package_name']}", + } + for event, timestamp in timestamps + if timestamp ] - for timestamp in timestamps: - records.append( - { - "timestamp": timestamp["timestamp"], - "module": self.__class__.__name__, - "event": timestamp["event"], - "data": f"Install or update of package {record['package_name']}", - } - ) - - return records - @staticmethod - def parse_dumpsys_package_for_details(output: str) -> Dict[str, Any]: - """ - Parse one entry of a dumpsys package information - """ - details: Dict[str, Any] = { - "uid": "", - "version_name": "", - "version_code": "", - "timestamp": "", - "first_install_time": "", - "last_update_time": "", - "installer": "", + def _permission(line: str, permission_type: str) -> dict: + name, _, details = line.strip().partition(":") + granted_match = re.search(r"granted=(true|false)", details) + flags_match = re.search(r"flags=\[\s*([^]]*)\]", details) + return { + "name": name, + "type": permission_type, + "granted": granted_match.group(1) == "true" if granted_match else None, + "flags": [ + flag.strip() + for flag in (flags_match.group(1).split("|") if flags_match else []) + if flag.strip() + ], + } + + @classmethod + def parse_dumpsys_package_for_details(cls, output: str) -> dict[str, Any]: + details: dict[str, Any] = { + "app_id": None, + "version_name": None, + "version_code": None, + "min_sdk": None, + "target_sdk": None, + "timestamp": None, + "last_update_time": None, + "installer": None, "system": False, - "permissions": list(), - "requested_permissions": list(), + "permissions": [], + "requested_permissions": [], + "users": [], } - in_install_permissions = False - in_runtime_permissions = False - in_declared_permissions = False - in_requested_permissions = True - current_user: Optional[int] = None - first_install_times: Dict[Optional[int], str] = {} - runtime_permissions: Dict[Optional[int], List[Dict[str, Any]]] = {} + permission_section: str | None = None + current_user: dict[str, Any] | None = None + legacy_first_install: str | None = None + for line in output.splitlines(): - user_match = re.match(r"User (\d+):", line.strip()) + stripped = line.strip() + user_match = re.match(r"User (\d+):\s*(.*)", stripped) if user_match: - current_user = int(user_match.group(1)) - - if in_install_permissions: - if line.startswith(" " * 4) and not line.startswith(" " * 6): - in_install_permissions = False - else: - lineinfo = line.strip().split(":") - permission = lineinfo[0] - granted = None - if "granted=" in lineinfo[1]: - granted = "granted=true" in lineinfo[1] - - details["permissions"].append( - {"name": permission, "granted": granted, "type": "install"} - ) - if in_runtime_permissions: - if not line.startswith(" " * 8): - in_runtime_permissions = False - else: - lineinfo = line.strip().split(":") - permission = lineinfo[0] - granted = None - if "granted=" in lineinfo[1]: - granted = "granted=true" in lineinfo[1] - - runtime_permissions.setdefault(current_user, []).append( - {"name": permission, "granted": granted, "type": "runtime"} - ) - if in_declared_permissions: - if not line.startswith(" " * 6): - in_declared_permissions = False - else: - permission = line.strip().split(":")[0] - details["permissions"].append( - {"name": permission, "type": "declared"} - ) - if in_requested_permissions: - if not line.startswith(" " * 6): - in_requested_permissions = False - else: - details["requested_permissions"].append(line.strip()) - if line.strip().startswith("userId="): - details["uid"] = line.split("=")[1].strip() - elif line.strip().startswith("versionName="): - details["version_name"] = line.split("=")[1].strip() - elif line.strip().startswith("versionCode="): - details["version_code"] = line.split("=", 1)[1].strip() - elif line.strip().startswith("timeStamp="): - details["timestamp"] = line.split("=")[1].strip() - elif line.strip().startswith("installerPackageName="): - details["installer"] = line.split("=", 1)[1].strip() - elif line.strip().startswith("pkgFlags="): - details["system"] = "SYSTEM" in line.split("=", 1)[1].split() - elif line.strip().startswith("firstInstallTime="): - first_install_times[current_user] = line.split("=", 1)[1].strip() - elif line.strip().startswith("lastUpdateTime="): - details["last_update_time"] = line.split("=")[1].strip() - elif line.strip() == "install permissions:": - in_install_permissions = True - elif line.strip() == "runtime permissions:": - in_runtime_permissions = True - elif line.strip() == "declared permissions:": - in_declared_permissions = True - elif line.strip() == "requested permissions:": - in_requested_permissions = True - - if 0 in first_install_times: - details["first_install_time"] = first_install_times[0] - elif None in first_install_times: - details["first_install_time"] = first_install_times[None] - elif first_install_times: - details["first_install_time"] = next(iter(first_install_times.values())) - - if 0 in runtime_permissions: - details["permissions"].extend(runtime_permissions[0]) - elif None in runtime_permissions: - details["permissions"].extend(runtime_permissions[None]) - elif runtime_permissions: - details["permissions"].extend(next(iter(runtime_permissions.values()))) - return details + current_user = {"user_id": int(user_match.group(1)), "permissions": []} + for key, raw in re.findall(r"(\w+)=([^\s]+)", user_match.group(2)): + clean_key = { + "notLaunched": "not_launched", + "installReason": "install_reason", + "uninstallReason": "uninstall_reason", + "dataDir": "data_dir", + }.get(key, re.sub(r"(? List[Dict[str, Any]]: - """ - Parse the dumpsys package service data - """ - pkg_rxp = re.compile(r" Package \[(.+?)\].*") + header = stripped.lower() + if header in { + "declared permissions:", + "install permissions:", + "requested permissions:", + "runtime permissions:", + }: + permission_section = header.split()[0] + continue - results = [] - package_name = None - package = {} - lines: list[str] = [] - for line in output.splitlines(): - if line.startswith(" Package ["): - if len(lines) > 0: - details = self.parse_dumpsys_package_for_details("\n".join(lines)) - package.update(details) - results.append(package) - lines = [] - package = {} - - matches = pkg_rxp.findall(line) - if not matches: + if current_user is not None: + user_property = re.match( + r"(installReason|uninstallReason|dataDir|firstInstallTime)=(.*)", + stripped, + ) + if user_property: + key = { + "installReason": "install_reason", + "uninstallReason": "uninstall_reason", + "dataDir": "data_dir", + "firstInstallTime": "first_install_time", + }[user_property.group(1)] + current_user[key] = _value(user_property.group(2)) continue - package_name = matches[0] - package["package_name"] = package_name + if permission_section == "requested" and line.startswith(" "): + details["requested_permissions"].append(stripped) continue - - if not package_name: + if permission_section in ("declared", "install") and line.startswith( + " " + ): + details["permissions"].append( + cls._permission(stripped, permission_section) + ) + continue + if ( + permission_section == "runtime" + and line.startswith(" ") + and current_user is not None + ): + current_user["permissions"].append(cls._permission(stripped, "runtime")) continue - lines.append(line) - - if len(lines) > 0: - details = self.parse_dumpsys_package_for_details("\n".join(lines)) - package.update(details) - results.append(package) + simple_match = re.match( + r"(appId|userId|versionName|timeStamp|lastUpdateTime|installerPackageName)=(.*)", + stripped, + ) + if simple_match: + key = { + "appId": "app_id", + "userId": "app_id", + "versionName": "version_name", + "timeStamp": "timestamp", + "lastUpdateTime": "last_update_time", + "installerPackageName": "installer", + }[simple_match.group(1)] + raw_value = simple_match.group(2) + details[key] = ( + _value(raw_value) + if key == "app_id" or raw_value == "null" + else raw_value + ) + continue + if stripped.startswith("pkgFlags="): + details["system"] = "SYSTEM" in stripped.split("=", 1)[1].split() + continue + version_match = re.match( + r"versionCode=([^\s]+)(?:\s+minSdk=([^\s]+))?(?:\s+targetSdk=([^\s]+))?", + stripped, + ) + if version_match: + details["version_code"] = _value(version_match.group(1)) + details["min_sdk"] = ( + _value(version_match.group(2)) if version_match.group(2) else None + ) + details["target_sdk"] = ( + _value(version_match.group(3)) if version_match.group(3) else None + ) + elif stripped.startswith("firstInstallTime="): + legacy_first_install = stripped.split("=", 1)[1] - return results + if legacy_first_install: + user_zero = next( + (user for user in details["users"] if user["user_id"] == 0), None + ) + if user_zero is None: + user_zero = {"user_id": 0, "permissions": []} + details["users"].append(user_zero) + user_zero.setdefault("first_install_time", legacy_first_install) + return details - def parse(self, content: str): - """ - Parse the Dumpsys Package section for activities - Adds results to self.results + def parse(self, content: str) -> None: + self.results: list[dict[str, Any]] = [] + category: str | None = None + package: dict[str, Any] | None = None + block: list[str] = [] - :param content: content of the package section (string) - """ - self.results = [] - package = [] + def finish() -> None: + nonlocal package, block + if package is not None: + package.update(self.parse_dumpsys_package_for_details("\n".join(block))) + self.results.append(package) + package = None + block = [] - in_package_list = False for line in content.splitlines(): - if line.startswith("Packages:"): - in_package_list = True + if line == "Packages:": + finish() + category = "active" continue - - if not in_package_list: + if line == "Hidden system packages:": + finish() + category = "hidden_system" continue - - if line.strip() == "": - break - - package.append(line) - - self.results = self.parse_dumpsys_packages("\n".join(package)) + package_match = re.match(r"^ Package \[([^]]+)\]", line) + if package_match and category: + finish() + package = { + "package_name": package_match.group(1), + "package_type": category, + } + continue + if ( + category + and line + and not line.startswith(" ") + and not line.endswith(" overlay paths:") + ): + finish() + category = None + continue + if package is not None: + block.append(line) + finish() diff --git a/src/mvt/android/artifacts/dumpsys_platform_compat.py b/src/mvt/android/artifacts/dumpsys_platform_compat.py index c01fbfd85..08cb46e37 100644 --- a/src/mvt/android/artifacts/dumpsys_platform_compat.py +++ b/src/mvt/android/artifacts/dumpsys_platform_compat.py @@ -3,6 +3,8 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +import re + from .artifact import AndroidArtifact @@ -25,19 +27,37 @@ def check_indicators(self) -> None: def parse(self, data: str) -> None: for line in data.splitlines(): - if not line.startswith("ChangeId(168419799; name=DOWNSCALED;"): + match = re.match(r"ChangeId\((\d+);\s*(.*)\)$", line.strip()) + if not match or "rawOverrides={" not in line: continue - - if line.strip() == "": - break - - # Look for rawOverrides field - if "rawOverrides={" in line: - # Extract the content inside the braces for rawOverrides - overrides_field = line.split("rawOverrides={", 1)[1].split("};", 1)[0] - - for entry in overrides_field.split(", "): - # Extract app name - uninstall_app = entry.split("=")[0].strip() - - self.results.append({"package_name": uninstall_app}) + body = match.group(2) + name_match = re.search(r"(?:^|;\s*)name=([^;]+)", body) + state = ( + "enabled" + if re.search(r"(?:^|;\s*)enabled(?:;|$)", body) + else "disabled" + ) + overridable = bool(re.search(r"(?:^|;\s*)overridable(?:;|$)", body)) + overrides_field = body.split("rawOverrides={", 1)[1].split("}", 1)[0] + for entry in overrides_field.split(","): + package_name, separator, raw_value = entry.strip().partition("=") + if not separator: + continue + value: bool | int | str + if raw_value in ("true", "false"): + value = raw_value == "true" + else: + try: + value = int(raw_value) + except ValueError: + value = raw_value + self.results.append( + { + "change_id": int(match.group(1)), + "change_name": name_match.group(1) if name_match else None, + "change_state": state, + "overridable": overridable, + "package_name": package_name, + "override_value": value, + } + ) diff --git a/src/mvt/android/artifacts/dumpsys_receivers.py b/src/mvt/android/artifacts/dumpsys_receivers.py index b4379309c..759433567 100644 --- a/src/mvt/android/artifacts/dumpsys_receivers.py +++ b/src/mvt/android/artifacts/dumpsys_receivers.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ from .artifact import AndroidArtifact +from .package_resolvers import parse_resolver_table INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS" INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED" @@ -18,115 +19,45 @@ class DumpsysReceiversArtifact(AndroidArtifact): """ 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_match = self.indicators.check_app_id(receiver["package_name"]) - if ioc_match: - self.alertstore.critical( - ioc_match.message, - "", - {intent: receiver}, - matched_indicator=ioc_match.ioc, - ) - continue + for receiver in self.results: + intent = receiver["key"] + if intent == INTENT_NEW_OUTGOING_SMS: + self.log.info( + 'Found a receiver to intercept outgoing SMS messages: "%s"', + receiver["component"], + ) + elif intent == INTENT_SMS_RECEIVED: + self.log.info( + 'Found a receiver to intercept incoming SMS messages: "%s"', + receiver["component"], + ) + elif intent == INTENT_DATA_SMS_RECEIVED: + self.log.info( + 'Found a receiver to intercept incoming data SMS message: "%s"', + receiver["component"], + ) + elif intent == INTENT_PHONE_STATE: + self.log.info( + 'Found a receiver monitoring telephony state/incoming calls: "%s"', + receiver["component"], + ) + elif intent == INTENT_NEW_OUTGOING_CALL: + self.log.info( + 'Found a receiver monitoring outgoing calls: "%s"', + receiver["component"], + ) + + if not self.indicators: + continue + + ioc_match = self.indicators.check_app_id(receiver["package_name"]) + if ioc_match: + self.alertstore.critical( + ioc_match.message, + "", + receiver, + matched_indicator=ioc_match.ioc, + ) def parse(self, output: str) -> None: - self.results: dict[str, list[dict[str, str]]] = {} - - 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 - - parts = line.strip().split(" ") - if len(parts) < 2: - # A single-token line here is not a receiver. Real dumpstate - # output can print an action header mis-indented (observed with - # 15 leading spaces instead of 6), which used to raise - # IndexError and abort the whole module. Treat a trailing-colon - # token as the next action, skip anything else. - if parts[0].endswith(":"): - intent = parts[0][:-1] - self.results.setdefault(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 = parts[1] - package_name = receiver.split("/")[0] - - self.results[intent].append( - { - "package_name": package_name, - "receiver": receiver, - } - ) + self.results = parse_resolver_table(output, "Receiver") diff --git a/src/mvt/android/artifacts/getprop.py b/src/mvt/android/artifacts/getprop.py index debcfc4cd..094eb5dc9 100644 --- a/src/mvt/android/artifacts/getprop.py +++ b/src/mvt/android/artifacts/getprop.py @@ -28,19 +28,16 @@ class GetProp(AndroidArtifact): def parse(self, entry: str) -> None: self.results: List[Dict[str, str]] = [] - rxp = re.compile(r"\[(.+?)\]: \[(.+?)\]") + # A property value may span several lines: persist.sys.boot.reason.history + # prints one boot per line. Matching the whole section instead of line by + # line lets a value run to the first closing bracket that ends a line. + rxp = re.compile( + r"^[ \t]*\[([^]]+)\]: \[(.*?)\][ \t\r]*$", + re.MULTILINE | re.DOTALL, + ) - for line in entry.splitlines(): - line = line.strip() - if line == "": - continue - - matches = re.findall(rxp, line) - if not matches or len(matches[0]) != 2: - continue - - prop_entry = {"name": matches[0][0], "value": matches[0][1]} - self.results.append(prop_entry) + for name, value in rxp.findall(entry): + self.results.append({"name": name, "value": value}) def get_device_timezone(self) -> str | None: """ diff --git a/src/mvt/android/artifacts/mounts.py b/src/mvt/android/artifacts/mounts.py index a35d889fa..5a5537b71 100644 --- a/src/mvt/android/artifacts/mounts.py +++ b/src/mvt/android/artifacts/mounts.py @@ -117,6 +117,51 @@ def parse(self, entry: str) -> None: # Skip lines that don't match expected format continue + @staticmethod + def parse_mountinfo(entry: str, process_id: int) -> list[dict[str, Any]]: + """Parse Linux /proc/PID/mountinfo records.""" + results = [] + for line in entry.splitlines(): + fields = line.split() + if "-" not in fields: + continue + separator = fields.index("-") + if separator < 6 or len(fields) < separator + 4: + continue + try: + mount_id = int(fields[0]) + parent_id = int(fields[1]) + except ValueError: + continue + mount_options = fields[5].split(",") + super_options = fields[separator + 3].split(",") + options = list(dict.fromkeys(mount_options + super_options)) + mount_point = fields[4].replace("\\040", " ") + device = fields[separator + 2].replace("\\040", " ") + filesystem_type = fields[separator + 1] + is_system = mount_point in SUSPICIOUS_MOUNT_POINTS or any( + mount_point.startswith(f"{prefix}/") + for prefix in SUSPICIOUS_MOUNT_POINTS + ) + results.append( + { + "mount_id": mount_id, + "parent_id": parent_id, + "major_minor": fields[2], + "root": fields[3].replace("\\040", " "), + "mount_point": mount_point, + "device": device, + "filesystem_type": filesystem_type, + "mount_options": ",".join(options), + "options_list": options, + "optional_fields": fields[6:separator], + "is_system_partition": is_system, + "is_read_write": "rw" in options, + "process_ids": [process_id], + } + ) + return results + def check_indicators(self) -> None: """ Check for suspicious mount configurations that may indicate root access diff --git a/src/mvt/android/artifacts/package_resolvers.py b/src/mvt/android/artifacts/package_resolvers.py new file mode 100644 index 000000000..60c221253 --- /dev/null +++ b/src/mvt/android/artifacts/package_resolvers.py @@ -0,0 +1,70 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 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 re + + +RESOLVER_TYPES = { + "Full MIME Types": "full_mime_type", + "Base MIME Types": "base_mime_type", + "Wild MIME Types": "wild_mime_type", + "Schemes": "scheme", + "Non-Data Actions": "non_data_action", + "MIME Typed Actions": "mime_typed_action", +} + +ENTRY_RE = re.compile( + r"^\s{8,}[0-9a-fA-F]+\s+(?P\S+)" + r"(?:\s+\((?P\d+)\s+filters?\))?\s*$" +) + + +def parse_resolver_table(content: str, table_name: str) -> list[dict]: + """Parse every resolver category from a dumpsys package table.""" + results: list[dict] = [] + in_table = False + resolver_type: str | None = None + key: str | None = None + + for line in content.splitlines(): + if line.startswith(f"{table_name} Resolver Table:"): + in_table = True + continue + if not in_table: + continue + + if line and not line.startswith(" "): + break + + section_match = re.match(r"^ {2}([^ ].*):\s*$", line) + if section_match: + resolver_type = RESOLVER_TYPES.get(section_match.group(1)) + key = None + continue + if resolver_type is None: + continue + + key_match = re.match(r"^ {6,}([^ ].*):\s*$", line) + if key_match: + key = key_match.group(1) + continue + if key is None: + continue + + entry_match = ENTRY_RE.match(line) + if not entry_match: + continue + component = entry_match.group("component") + results.append( + { + "resolver_type": resolver_type, + "key": key, + "package_name": component.split("/", 1)[0], + "component": component, + "filter_count": int(entry_match.group("filters") or 1), + } + ) + + return results diff --git a/src/mvt/android/artifacts/processes.py b/src/mvt/android/artifacts/processes.py index 5471eb29e..97e27a905 100644 --- a/src/mvt/android/artifacts/processes.py +++ b/src/mvt/android/artifacts/processes.py @@ -1,72 +1,91 @@ # Mobile Verification Toolkit (MVT) -# Copyright (c) 2021-2023 The MVT Authors. +# Copyright (c) 2021-2026 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/ from .artifact import AndroidArtifact +FIELD_NAMES = { + "LABEL": "label", + "USER": "user", + "PID": "pid", + "TID": "tid", + "PPID": "ppid", + "VSZ": "virtual_memory_size", + "RSS": "resident_set_size", + "WCHAN": "wchan", + "ADDR": "address", + "S": "state", + "PRI": "priority", + "NI": "nice", + "RTPRIO": "realtime_priority", + "SCH": "scheduler", + "PCY": "policy", + "TIME": "cpu_time", + "CMD": "command", + "NAME": "command", +} +INTEGER_FIELDS = { + "pid", + "tid", + "ppid", + "virtual_memory_size", + "resident_set_size", + "priority", + "nice", +} + + class Processes(AndroidArtifact): def parse(self, entry: str) -> None: - for line in entry.splitlines()[1:]: - proc = line.split() + self.results = [] + lines = [line for line in entry.splitlines() if line.strip()] + if not lines: + return + headers = lines[0].split() + if not all(header in FIELD_NAMES for header in headers): + return - # Skip empty lines - if len(proc) == 0: + for line in lines[1:]: + values = line.split(None, len(headers) - 1) + if len(values) != len(headers): continue - - # Sometimes WCHAN is empty. - if len(proc) == 8: - proc = proc[:5] + [""] + proc[5:] - - # Sometimes there is the security label. - if proc[0].startswith("u:r"): - label = proc[0] - proc = proc[1:] - else: - label = "" - - # Sometimes there is no WCHAN. - if len(proc) < 9: - proc = proc[:5] + [""] + proc[5:] - - self.results.append( - { - "user": proc[0], - "pid": int(proc[1]), - "ppid": int(proc[2]), - "virtual_memory_size": int(proc[3]), - "resident_set_size": int(proc[4]), - "wchan": proc[5], - "aprocress": proc[6], - "stat": proc[7], - "proc_name": proc[8].strip("[]"), - "label": label, - } - ) + result = {} + valid = True + for header, raw in zip(headers, values): + key = FIELD_NAMES[header] + value: str | int = raw.strip("[]") if key == "command" else raw + if key in INTEGER_FIELDS: + try: + value = int(value) + except ValueError: + valid = False + break + result[key] = value + if valid: + self.results.append(result) def check_indicators(self) -> None: if not self.indicators: return - for result in self.results: - proc_name = result.get("proc_name", "") - if not proc_name: - continue - - # Skipping this process because of false positives. - if result["proc_name"] == "gatekeeperd": + command = result.get("command", "") + if not isinstance(command, str): continue - - ioc_match = self.indicators.check_app_id(proc_name) - if ioc_match: - self.alertstore.critical( - ioc_match.message, "", result, matched_indicator=ioc_match.ioc - ) + process_name = command.rsplit("/", 1)[-1] + if not process_name or process_name == "gatekeeperd": continue - - ioc_match = self.indicators.check_process(proc_name) - if ioc_match: - self.alertstore.critical( - ioc_match.message, "", result, matched_indicator=ioc_match.ioc - ) + for checker in ( + self.indicators.check_app_id, + self.indicators.check_process, + ): + ioc_match = checker(process_name) + if ioc_match: + self.alertstore.critical( + ioc_match.message, + "", + result, + matched_indicator=ioc_match.ioc, + ) + break diff --git a/src/mvt/android/artifacts/settings.py b/src/mvt/android/artifacts/settings.py index 17c3bf2c5..e0cd5f0bc 100644 --- a/src/mvt/android/artifacts/settings.py +++ b/src/mvt/android/artifacts/settings.py @@ -3,6 +3,8 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +import re + from .artifact import AndroidArtifact ANDROID_DANGEROUS_SETTINGS = [ @@ -60,6 +62,28 @@ class Settings(AndroidArtifact): + def parse(self, content: str) -> None: + self.results: dict[str, dict[str, str]] = {} + namespace: str | None = None + for line in content.splitlines(): + heading = re.match( + r"^(CONFIG|GLOBAL|SECURE|SYSTEM) SETTINGS \(user (\d+)\)$", + line.strip(), + ) + if heading: + namespace = f"{heading.group(1).lower()}:user_{heading.group(2)}" + self.results[namespace] = {} + continue + if namespace is None or not line.startswith("_id:"): + continue + setting = re.match( + r"^_id:\S+\s+name:(.*?)\s+pkg:.*?\s+value:(.*?)" + r"(?:\s+default:.*\s+defaultSystemSet:(?:true|false))?$", + line, + ) + if setting: + self.results[namespace][setting.group(1)] = setting.group(2) + def check_indicators(self) -> None: for namespace, settings in self.results.items(): for key, value in settings.items(): diff --git a/src/mvt/android/artifacts/tombstone_crashes.py b/src/mvt/android/artifacts/tombstone_crashes.py index 1b6e115a5..c749d1ad0 100644 --- a/src/mvt/android/artifacts/tombstone_crashes.py +++ b/src/mvt/android/artifacts/tombstone_crashes.py @@ -131,6 +131,13 @@ def parse_protobuf( self, file_name: str, file_timestamp: datetime.datetime, data: bytes ) -> None: """Parse Android tombstone crash files from a protobuf object.""" + self.results.append(self.parse_protobuf_record(file_name, file_timestamp, data)) + + def parse_protobuf_record( + self, file_name: str, file_timestamp: datetime.datetime, data: bytes + ) -> dict: + if not data: + raise ValueError("empty protobuf tombstone") tombstone_pb = Tombstone().parse(data) tombstone_dict = tombstone_pb.to_dict( casing=betterproto2.Casing.SNAKE, include_default_values=True @@ -143,20 +150,31 @@ def parse_protobuf( tombstone_dict["file_name"] = file_name tombstone_dict["file_timestamp"] = convert_datetime_to_iso(file_timestamp) tombstone_dict["process_name"] = self._proccess_name_from_thread(tombstone_dict) + if isinstance(tombstone_dict.get("selinux_label"), str): + tombstone_dict["selinux_label"] = tombstone_dict["selinux_label"].rstrip( + "\x00" + ) # Confirm the tombstone is valid, and matches the output model tombstone = TombstoneCrashResult.model_validate(tombstone_dict) - self.results.append(tombstone.model_dump()) + return tombstone.model_dump() def parse( self, file_name: str, file_timestamp: datetime.datetime, content: bytes ) -> None: """Parse text Android tombstone crash files.""" + self.results.append(self.parse_text_record(file_name, file_timestamp, content)) + + def parse_text_record( + self, file_name: str, file_timestamp: datetime.datetime, content: bytes + ) -> dict: + if not content: + raise ValueError("empty plaintext tombstone") tombstone_dict = { "file_name": file_name, "file_timestamp": convert_datetime_to_iso(file_timestamp), } - lines = content.decode("utf-8").splitlines() + lines = content.decode("utf-8", errors="replace").splitlines() for line_num, line in enumerate(lines, 1): if not line.strip() or TOMBSTONE_DELIMITER in line: continue @@ -171,7 +189,7 @@ def parse( # Validate the tombstone and add it to the results tombstone = TombstoneCrashResult.model_validate(tombstone_dict) - self.results.append(tombstone.model_dump()) + return tombstone.model_dump() def _parse_tombstone_line( self, line: str, key: str, destination_key: str, tombstone: dict @@ -200,7 +218,9 @@ def _load_key_value_line( if not separator or line_key != key: return False - value_clean = value.strip().strip("'") + value_clean = value.strip() + if len(value_clean) >= 2 and value_clean[0] == value_clean[-1] == "'": + value_clean = value_clean[1:-1] if destination_key == "uid": tombstone[destination_key] = int(value_clean) elif destination_key == "process_uptime": @@ -274,9 +294,7 @@ def _load_timestamp_line(self, line: str, tombstone: dict) -> bool: @staticmethod def _parse_timestamp_string(timestamp: str) -> str: timestamp_parsed = parser.parse(timestamp) - # 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) + return convert_datetime_to_iso(timestamp_parsed) @staticmethod def _proccess_name_from_thread(tombstone_dict: dict) -> str: diff --git a/src/mvt/android/modules/androidqf/__init__.py b/src/mvt/android/modules/androidqf/__init__.py index 1d8d6195a..05e7cde94 100644 --- a/src/mvt/android/modules/androidqf/__init__.py +++ b/src/mvt/android/modules/androidqf/__init__.py @@ -3,6 +3,8 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +from mvt.common.module import MVTModule + from .aqf_files import AQFFiles from .aqf_getprop import AQFGetProp from .aqf_log_timestamps import AQFLogTimestamps @@ -12,7 +14,7 @@ from .mounts import Mounts from .root_binaries import RootBinaries -ANDROIDQF_MODULES = [ +ANDROIDQF_MODULES: list[type[MVTModule]] = [ AQFPackages, AQFProcesses, AQFGetProp, diff --git a/src/mvt/android/modules/backup/__init__.py b/src/mvt/android/modules/backup/__init__.py index 47a5e1cf9..683a38396 100644 --- a/src/mvt/android/modules/backup/__init__.py +++ b/src/mvt/android/modules/backup/__init__.py @@ -3,6 +3,8 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +from mvt.common.module import MVTModule + from .sms import SMS -BACKUP_MODULES = [SMS] +BACKUP_MODULES: list[type[MVTModule]] = [SMS] diff --git a/src/mvt/android/modules/bugreport/__init__.py b/src/mvt/android/modules/bugreport/__init__.py index 1594af9c3..e23bf5a9a 100644 --- a/src/mvt/android/modules/bugreport/__init__.py +++ b/src/mvt/android/modules/bugreport/__init__.py @@ -3,6 +3,8 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +from mvt.common.module import MVTModule + from .dumpsys_accessibility import DumpsysAccessibility from .dumpsys_activities import DumpsysActivities from .dumpsys_appops import DumpsysAppops @@ -16,8 +18,11 @@ from .dumpsys_adb_state import DumpsysADBState from .fs_timestamps import BugReportTimestamps from .tombstones import Tombstones +from .mounts import Mounts +from .processes import Processes +from .settings import Settings -BUGREPORT_MODULES = [ +BUGREPORT_MODULES: list[type[MVTModule]] = [ DumpsysAccessibility, DumpsysActivities, DumpsysAppops, @@ -31,4 +36,7 @@ DumpsysADBState, BugReportTimestamps, Tombstones, + Processes, + Settings, + Mounts, ] diff --git a/src/mvt/android/modules/bugreport/base.py b/src/mvt/android/modules/bugreport/base.py index 156e01cf2..01af8c962 100644 --- a/src/mvt/android/modules/bugreport/base.py +++ b/src/mvt/android/modules/bugreport/base.py @@ -72,7 +72,11 @@ def _get_file_content(self, file_path: str) -> bytes: if not self.extract_path: raise ValueError("extract_path is not set") joined = os.path.join(self.extract_path, file_path) - if not Path(joined).resolve().is_relative_to(Path(self.extract_path).resolve()): + if ( + not Path(joined) + .resolve() + .is_relative_to(Path(self.extract_path).resolve()) + ): raise ValueError("unsafe file_path") handle = open(joined, "rb") @@ -100,6 +104,27 @@ def _get_dumpstate_file(self) -> Optional[bytes]: return None + @staticmethod + def extract_command_section(content: str, heading: str) -> str: + """Return a bugreport command section without consuming the next one. + + Bugreport separators include timing text, so looking for a line equal to + ``------`` is not sufficient and can accidentally feed the remainder of + dumpstate to a parser. + """ + lines: list[str] = [] + in_section = False + for line in content.splitlines(): + stripped = line.strip() + if not in_section: + if stripped.startswith(heading): + in_section = True + continue + if stripped.startswith("------"): + break + lines.append(line) + return "\n".join(lines) + def _get_file_modification_time(self, file_path: str) -> datetime.datetime: if self.zip_archive: file_timetuple = self.zip_archive.getinfo(file_path).date_time diff --git a/src/mvt/android/modules/bugreport/dumpsys_accessibility.py b/src/mvt/android/modules/bugreport/dumpsys_accessibility.py index 72208c02e..02d85a308 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_accessibility.py +++ b/src/mvt/android/modules/bugreport/dumpsys_accessibility.py @@ -49,9 +49,7 @@ def run(self) -> None: self.parse(content) for result in self.results: - self.log.info( - 'Found installed accessibility service "%s"', result.get("service") - ) + self.log.info('Found accessibility service "%s"', result.get("component")) self.log.info( "Identified a total of %d accessibility services", len(self.results) diff --git a/src/mvt/android/modules/bugreport/dumpsys_getprop.py b/src/mvt/android/modules/bugreport/dumpsys_getprop.py index 198a1d2ab..b9c20eab1 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_getprop.py +++ b/src/mvt/android/modules/bugreport/dumpsys_getprop.py @@ -44,21 +44,8 @@ def run(self) -> None: ) return - lines = [] - in_getprop = False - - for line in content.decode(errors="ignore").splitlines(): - if line.strip().startswith("------ SYSTEM PROPERTIES"): - in_getprop = True - continue - - if not in_getprop: - continue - - if line.strip() == "------": - break - - lines.append(line) - - self.parse("\n".join(lines)) + section = self.extract_command_section( + content.decode(errors="ignore"), "------ SYSTEM PROPERTIES" + ) + self.parse(section) self.log.info("Extracted %d Android system properties", len(self.results)) diff --git a/src/mvt/android/modules/bugreport/dumpsys_packages.py b/src/mvt/android/modules/bugreport/dumpsys_packages.py index aebec4c26..ff944e3e0 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_packages.py +++ b/src/mvt/android/modules/bugreport/dumpsys_packages.py @@ -50,7 +50,10 @@ def run(self) -> None: for result in self.results: dangerous_permissions_count = 0 - for perm in result["permissions"]: + permissions = list(result["permissions"]) + for user in result.get("users", []): + permissions.extend(user.get("permissions", [])) + for perm in permissions: if perm["name"] in DANGEROUS_PERMISSIONS: dangerous_permissions_count += 1 diff --git a/src/mvt/android/modules/bugreport/dumpsys_platform_compat.py b/src/mvt/android/modules/bugreport/dumpsys_platform_compat.py index 968bc25ea..dd06b40b8 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_platform_compat.py +++ b/src/mvt/android/modules/bugreport/dumpsys_platform_compat.py @@ -13,7 +13,7 @@ class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule): - """This module extracts details on uninstalled apps.""" + """This module extracts raw per-package compatibility overrides.""" def __init__( self, @@ -48,4 +48,4 @@ def run(self) -> None: ) self.parse(content) - self.log.info("Found %d uninstalled apps", len(self.results)) + self.log.info("Found %d package compatibility overrides", len(self.results)) diff --git a/src/mvt/android/modules/bugreport/dumpsys_receivers.py b/src/mvt/android/modules/bugreport/dumpsys_receivers.py index 8907d5e36..a82f0213e 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_receivers.py +++ b/src/mvt/android/modules/bugreport/dumpsys_receivers.py @@ -33,7 +33,7 @@ def __init__( results=results, ) - self.results = results if results else {} + self.results = results if results else [] def run(self) -> None: content = self._get_dumpstate_file() @@ -49,4 +49,4 @@ def run(self) -> None: ) self.parse(dumpsys_section) - self.log.info("Extracted receivers for %d intents", len(self.results)) + self.log.info("Extracted %d package receivers", len(self.results)) diff --git a/src/mvt/android/modules/bugreport/fs_timestamps.py b/src/mvt/android/modules/bugreport/fs_timestamps.py index 5a2ca485f..3c7e3248b 100644 --- a/src/mvt/android/modules/bugreport/fs_timestamps.py +++ b/src/mvt/android/modules/bugreport/fs_timestamps.py @@ -4,12 +4,15 @@ # https://license.mvt.re/1.1/ import logging +import datetime from typing import Optional +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from mvt.common.utils import convert_datetime_to_iso from .base import BugReportModule from mvt.common.module_types import ModuleResults from mvt.android.artifacts.file_timestamps import FileTimestampsArtifact +from mvt.android.artifacts.getprop import GetProp class BugReportTimestamps(FileTimestampsArtifact, BugReportModule): @@ -38,15 +41,44 @@ def __init__( def run(self) -> None: filesystem_files = self._get_files_by_pattern("FS/*") + timezone_name = None + dumpstate = self._get_dumpstate_file() + if dumpstate: + section = self.extract_command_section( + dumpstate.decode("utf-8", errors="replace"), + "------ SYSTEM PROPERTIES", + ) + properties = GetProp() + properties.parse(section) + timezone_name = properties.get_device_timezone() + timezone = None + if timezone_name: + try: + timezone = ZoneInfo(timezone_name) + except ZoneInfoNotFoundError: + self.log.warning("Unknown device timezone %s", timezone_name) + self.results = [] for file in filesystem_files: # Only the modification time is available in the zip file metadata. # The timezone is the local timezone of the machine the phone. modification_time = self._get_file_modification_time(file) + utc_time = None + if timezone is not None: + utc_time = convert_datetime_to_iso( + modification_time.replace(tzinfo=timezone).astimezone( + datetime.timezone.utc + ) + ) self.results.append( { "path": file, "modified_time": convert_datetime_to_iso(modification_time), + "modified_time_utc": utc_time, + "timezone": timezone_name, + "timestamp_source": ( + "zip_metadata" if self.zip_archive else "filesystem_metadata" + ), } ) diff --git a/src/mvt/android/modules/bugreport/mounts.py b/src/mvt/android/modules/bugreport/mounts.py new file mode 100644 index 000000000..ecef40c03 --- /dev/null +++ b/src/mvt/android/modules/bugreport/mounts.py @@ -0,0 +1,41 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 The MVT Authors. + +import re + +from mvt.android.artifacts.mounts import Mounts as MountsArtifact + +from .base import BugReportModule + + +class Mounts(MountsArtifact, BugReportModule): + """Extract and deduplicate process mount namespaces from mountinfo.""" + + def run(self) -> None: + mount_files = self._get_files_by_pattern("FS/proc/*/mountinfo") + if not mount_files: + mount_files = self._get_files_by_pattern("*/proc/*/mountinfo") + unique: dict[tuple, dict] = {} + for file_path in mount_files: + pid_match = re.search(r"/proc/(\d+)/mountinfo$", file_path) + if not pid_match: + continue + entries = self.parse_mountinfo( + self._get_file_content(file_path).decode("utf-8", errors="replace"), + int(pid_match.group(1)), + ) + for entry in entries: + identity = ( + entry["major_minor"], + entry["root"], + entry["mount_point"], + entry["device"], + entry["filesystem_type"], + entry["mount_options"], + ) + if identity in unique: + unique[identity]["process_ids"].extend(entry["process_ids"]) + else: + unique[identity] = entry + self.results = list(unique.values()) + self.log.info("Extracted %d unique mount records", len(self.results)) diff --git a/src/mvt/android/modules/bugreport/processes.py b/src/mvt/android/modules/bugreport/processes.py new file mode 100644 index 000000000..4f9b33471 --- /dev/null +++ b/src/mvt/android/modules/bugreport/processes.py @@ -0,0 +1,22 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 The MVT Authors. + +from mvt.android.artifacts.processes import Processes as ProcessesArtifact + +from .base import BugReportModule + + +class Processes(ProcessesArtifact, BugReportModule): + """Extract the process and thread table from dumpstate.""" + + def run(self) -> None: + data = self._get_dumpstate_file() + if not data: + self.log.error("Unable to find dumpstate file") + return + section = self.extract_command_section( + data.decode("utf-8", errors="replace"), + "------ PROCESSES AND THREADS", + ) + self.parse(section) + self.log.info("Identified %d running process threads", len(self.results)) diff --git a/src/mvt/android/modules/bugreport/settings.py b/src/mvt/android/modules/bugreport/settings.py new file mode 100644 index 000000000..20180adfc --- /dev/null +++ b/src/mvt/android/modules/bugreport/settings.py @@ -0,0 +1,22 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 The MVT Authors. + +from mvt.android.artifacts.settings import Settings as SettingsArtifact + +from .base import BugReportModule + + +class Settings(SettingsArtifact, BugReportModule): + """Extract all SettingsProvider namespaces and users.""" + + def run(self) -> None: + data = self._get_dumpstate_file() + if not data: + self.log.error("Unable to find dumpstate file") + return + section = self.extract_dumpsys_section( + data.decode("utf-8", errors="replace"), "DUMP OF SERVICE settings:" + ) + self.parse(section) + count = sum(len(settings) for settings in self.results.values()) + self.log.info("Identified %d Android settings", count) diff --git a/src/mvt/android/modules/bugreport/tombstones.py b/src/mvt/android/modules/bugreport/tombstones.py index c4a7afb4a..4d121a140 100644 --- a/src/mvt/android/modules/bugreport/tombstones.py +++ b/src/mvt/android/modules/bugreport/tombstones.py @@ -43,21 +43,76 @@ def run(self) -> None: ) return - for tombstone_file in sorted(tombstone_files): - tombstone_filename = tombstone_file.split("/")[-1] - modification_time = self._get_file_modification_time(tombstone_file) - tombstone_data = self._get_file_content(tombstone_file) + grouped: dict[str, dict[str, str]] = {} + for file_path in tombstone_files: + file_name = file_path.rsplit("/", 1)[-1] + source = "protobuf" if file_name.endswith(".pb") else "text" + crash_id = file_name.removesuffix(".pb") + grouped.setdefault(crash_id, {})[source] = file_path - try: - if tombstone_file.endswith(".pb"): - self.parse_protobuf( - tombstone_filename, modification_time, tombstone_data + for crash_id, paths in sorted(grouped.items()): + parsed_sources: dict[str, dict] = {} + source_records: dict[str, dict] = {} + for source in ("text", "protobuf"): + file_path = paths.get(source) + if file_path is None: + continue + file_name = file_path.rsplit("/", 1)[-1] + file_timestamp = self._get_file_modification_time(file_path) + source_info = { + "file_name": file_name, + "file_timestamp": file_timestamp.isoformat(), + "parsed": False, + "error": None, + "record": None, + } + try: + data = self._get_file_content(file_path) + if source == "protobuf": + record = self.parse_protobuf_record( + file_name, file_timestamp, data + ) + else: + record = self.parse_text_record(file_name, file_timestamp, data) + source_info["parsed"] = True + source_info["record"] = record + source_records[source] = record + except Exception as exc: + source_info["error"] = str(exc) + self.log.error( + "Error parsing tombstone file %s: %s", file_path, exc ) - else: - self.parse(tombstone_filename, modification_time, tombstone_data) - except ValueError as e: - # Catch any exceptions raised during parsing or validation. - self.log.error(f"Error parsing tombstone file {tombstone_file}: {e}") + parsed_sources[source] = source_info + + if not source_records: + continue + preferred = source_records.get("protobuf") or source_records["text"] + canonical = dict(preferred) + text_record = source_records.get("text") + if text_record: + for key, value in text_record.items(): + if canonical.get(key) in (None, "", [], {}): + canonical[key] = value + + differences = {} + protobuf_record = source_records.get("protobuf") + if text_record and protobuf_record: + for key in text_record.keys() & protobuf_record.keys(): + if key in ("file_name", "file_timestamp"): + continue + if text_record[key] != protobuf_record[key]: + differences[key] = { + "text": text_record[key], + "protobuf": protobuf_record[key], + } + canonical.update( + { + "crash_id": crash_id, + "sources": parsed_sources, + "differences": differences, + } + ) + self.results.append(canonical) self.log.info( "Extracted a total of %d tombstone files", diff --git a/src/mvt/android/modules/intrusion_logs/__init__.py b/src/mvt/android/modules/intrusion_logs/__init__.py index f8be973f7..a0e7bdd7d 100644 --- a/src/mvt/android/modules/intrusion_logs/__init__.py +++ b/src/mvt/android/modules/intrusion_logs/__init__.py @@ -3,11 +3,13 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +from mvt.common.module import MVTModule + from .connect_event import ConnectEvent from .dns_event import DnsEvent from .security_event import SecurityEvent -INTRUSION_LOGS_MODULES = [ +INTRUSION_LOGS_MODULES: list[type[MVTModule]] = [ DnsEvent, ConnectEvent, SecurityEvent, diff --git a/src/mvt/ios/modules/backup/__init__.py b/src/mvt/ios/modules/backup/__init__.py index 0fb833ae4..ce8499bbf 100644 --- a/src/mvt/ios/modules/backup/__init__.py +++ b/src/mvt/ios/modules/backup/__init__.py @@ -3,9 +3,16 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +from mvt.common.module import MVTModule + from .backup_info import BackupInfo from .configuration_profiles import ConfigurationProfiles from .manifest import Manifest from .profile_events import ProfileEvents -BACKUP_MODULES = [BackupInfo, ConfigurationProfiles, Manifest, ProfileEvents] +BACKUP_MODULES: list[type[MVTModule]] = [ + BackupInfo, + ConfigurationProfiles, + Manifest, + ProfileEvents, +] diff --git a/src/mvt/ios/modules/fs/__init__.py b/src/mvt/ios/modules/fs/__init__.py index 06dba246f..b00ee35f4 100644 --- a/src/mvt/ios/modules/fs/__init__.py +++ b/src/mvt/ios/modules/fs/__init__.py @@ -3,6 +3,8 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +from mvt.common.module import MVTModule + from .analytics import Analytics from .analytics_ios_versions import AnalyticsIOSVersions from .cache_files import CacheFiles @@ -15,7 +17,7 @@ from .webkit_localstorage import WebkitLocalStorage from .webkit_safariviewservice import WebkitSafariViewService -FS_MODULES = [ +FS_MODULES: list[type[MVTModule]] = [ CacheFiles, Filesystem, Netusage, diff --git a/src/mvt/ios/modules/mixed/__init__.py b/src/mvt/ios/modules/mixed/__init__.py index a846bf2e3..027dce098 100644 --- a/src/mvt/ios/modules/mixed/__init__.py +++ b/src/mvt/ios/modules/mixed/__init__.py @@ -3,6 +3,8 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +from mvt.common.module import MVTModule + from .applications import Applications from .calendar import Calendar from .calls import Calls @@ -28,7 +30,7 @@ from .whatsapp import Whatsapp from .whatsapp_contacts import WhatsappContacts -MIXED_MODULES = [ +MIXED_MODULES: list[type[MVTModule]] = [ Calls, ChromeFavicon, ChromeHistory, diff --git a/tests/android/test_artifact_dumpsys_accessibility.py b/tests/android/test_artifact_dumpsys_accessibility.py index 1f0b23499..c727c5710 100644 --- a/tests/android/test_artifact_dumpsys_accessibility.py +++ b/tests/android/test_artifact_dumpsys_accessibility.py @@ -23,7 +23,7 @@ def test_parsing(self): assert len(da.results) == 4 assert da.results[0]["package_name"] == "com.android.settings" assert ( - da.results[0]["service"] + da.results[0]["component"] == "com.android.settings/com.samsung.android.settings.development.gpuwatch.GPUWatchInterceptor" ) @@ -37,7 +37,9 @@ def test_parsing_v14_aosp_format(self): da.parse(data) assert len(da.results) == 1 assert da.results[0]["package_name"] == "com.malware.accessibility" - assert da.results[0]["service"] == "com.malware.service.malwareservice" + assert da.results[0]["service_name"] == "com.malware.service.malwareservice" + assert da.results[0]["enabled"] is True + assert da.results[0]["installed"] is False def test_accessibility_service_alert(self): da = DumpsysAccessibilityArtifact() @@ -52,6 +54,22 @@ def test_accessibility_service_alert(self): assert da.alertstore.alerts[0].level == AlertLevel.MEDIUM assert da.alertstore.alerts[0].event == da.results[0] + def test_same_component_is_kept_for_each_user(self): + da = DumpsysAccessibilityArtifact() + da.parse( + """User state[attributes:{id=0 + installed services: { + 0 : com.example/.Service + } +User state[attributes:{id=10 + installed services: { + 0 : com.example/.Service + } +""" + ) + + assert [result["user_id"] for result in da.results] == [0, 10] + def test_ioc_check(self, indicator_file): da = DumpsysAccessibilityArtifact() file = get_artifact("android_data/dumpsys_accessibility.txt") diff --git a/tests/android/test_artifact_dumpsys_adb.py b/tests/android/test_artifact_dumpsys_adb.py index f5574a671..8d501b7dd 100644 --- a/tests/android/test_artifact_dumpsys_adb.py +++ b/tests/android/test_artifact_dumpsys_adb.py @@ -3,6 +3,8 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +import base64 + from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState from mvt.common.alerts import AlertLevel @@ -11,6 +13,17 @@ class TestDumpsysADBArtifact: + def test_parsing_binary_xml_recovers_key(self): + public_key = base64.b64encode(bytes(range(256))) + keystore = DumpsysADBArtifact().parse_binary_xml( + b"ABX\x00binary-key=" + public_key + b" user@host\x00lastConnection" + ) + + assert len(keystore) == 1 + assert keystore[0]["key"] == public_key.decode() + assert keystore[0]["user"] == "user@host" + assert keystore[0]["last_connected"] is None + def test_parsing(self): da_adb = DumpsysADBArtifact() file = get_artifact("android_data/dumpsys_adb.txt") diff --git a/tests/android/test_artifact_dumpsys_appops.py b/tests/android/test_artifact_dumpsys_appops.py index 862b87212..a86084f54 100644 --- a/tests/android/test_artifact_dumpsys_appops.py +++ b/tests/android/test_artifact_dumpsys_appops.py @@ -25,10 +25,37 @@ def test_parsing(self): assert da.results[0]["uid"] == "0" assert len(da.results[0]["permissions"]) == 1 assert da.results[0]["permissions"][0]["name"] == "MANAGE_IPSEC_TUNNELS" - assert da.results[0]["permissions"][0]["access"] == "allow" + assert da.results[0]["permissions"][0]["mode"] == "allow" assert da.results[6]["package_name"] == "com.sec.factory.camera" assert len(da.results[6]["permissions"][1]["entries"]) == 1 assert len(da.results[11]["permissions"]) == 4 + wake_lock = next( + permission + for permission in da.results[11]["permissions"] + if permission["name"] == "WAKE_LOCK" + ) + assert wake_lock["entries"][0]["duration"] == "+126ms" + + def test_running_and_attribution_are_retained(self): + da = DumpsysAppopsArtifact() + da.parse( + """ Uid 0: + state=cch + Package com.example: + CAMERA (allow): + camera=[ + Access: [fg-s] 2025-01-01 00:00:00.000 (-1s) duration=+2ms + ] + RECORD_AUDIO (allow): + Running start at: +3s +""" + ) + + camera = da.results[0]["permissions"][0]["entries"][0] + running = da.results[0]["permissions"][1]["entries"][0] + assert camera["attribution"] == "camera" + assert running["event"] == "running" + assert running["relative_time"] == "+3s" def test_ioc_check(self, indicator_file): da = DumpsysAppopsArtifact() diff --git a/tests/android/test_artifact_dumpsys_battery_daily.py b/tests/android/test_artifact_dumpsys_battery_daily.py index dbb9c36b4..5f7f997c6 100644 --- a/tests/android/test_artifact_dumpsys_battery_daily.py +++ b/tests/android/test_artifact_dumpsys_battery_daily.py @@ -57,18 +57,19 @@ def test_uninstall_and_downgrade_create_medium_alerts(self): assert uninstall_alert.message == ( "Detected uninstall of package com.example.removed (vers 0)" ) - assert uninstall_alert.event_time == "2022-08-16" + assert uninstall_alert.event_time == "2022-08-16 15:56:39" assert uninstall_alert.event["package_name"] == "com.example.removed" - assert uninstall_alert.event["vers"] == "0" + assert uninstall_alert.event["version_code"] == 0 + assert uninstall_alert.event["action"] == "uninstall" assert downgrade_alert.level == AlertLevel.MEDIUM assert downgrade_alert.message == ( "Detected downgrade of package com.example.app from vers 10 to vers 9" ) - assert downgrade_alert.event_time == "2022-08-17" + assert downgrade_alert.event_time == "2022-08-17 15:56:39" assert downgrade_alert.event["package_name"] == "com.example.app" assert downgrade_alert.event["action"] == "downgrade" - assert downgrade_alert.event["previous_vers"] == "10" + assert downgrade_alert.event["previous_version_code"] == 10 def test_newest_first_update_is_not_reported_as_downgrade(self): dba = DumpsysBatteryDailyArtifact() @@ -107,7 +108,19 @@ def test_newest_first_downgrade_creates_medium_alert(self): assert downgrade_alert.event_time == "2026-01-10" assert downgrade_alert.event["package_name"] == "com.example.app" assert downgrade_alert.event["action"] == "downgrade" - assert downgrade_alert.event["previous_vers"] == "102" + assert downgrade_alert.event["previous_version_code"] == 102 + + def test_duplicate_updates_retain_occurrence_count(self): + dba = DumpsysBatteryDailyArtifact() + dba.parse( + """ Daily from 2026-01-10-01-02-03 to 2026-01-11-04-05-06: + Update com.example.app vers=12 + Update com.example.app vers=12 +""" + ) + + assert dba.results[0]["occurrences"] == 2 + assert dba.results[0]["period_start"] == "2026-01-10 01:02:03" def test_reinstall_after_uninstall_is_not_reported_as_downgrade(self): dba = DumpsysBatteryDailyArtifact() diff --git a/tests/android/test_artifact_dumpsys_battery_history.py b/tests/android/test_artifact_dumpsys_battery_history.py index dd0b12016..091ce55cb 100644 --- a/tests/android/test_artifact_dumpsys_battery_history.py +++ b/tests/android/test_artifact_dumpsys_battery_history.py @@ -54,6 +54,7 @@ def test_parsing_absolute_timestamps(self): assert len(dba.results) == 2 assert dba.results[0] == { "time_elapsed": "07-15 20:27:39.431", + "timestamp": "1900-07-15 20:27:39.431000", "event": "start_job", "uid": "u0a123", "package_name": "com.example", @@ -61,3 +62,23 @@ def test_parsing_absolute_timestamps(self): } assert dba.results[1]["event"] == "end_job" assert dba.results[1]["uid"] == "u0a123" + + def test_wake_lock_without_component_is_retained(self): + dba = DumpsysBatteryHistoryArtifact() + dba.parse( + "Battery History:\n" + " 0 (2) 100 RESET:TIME: 2025-09-05-01-04-52-139\n" + ' +1s (2) 100 +running +wake_lock=1000:"*alarm*:TIME_TICK"\n' + ' +2s (2) 100 +running +wake_lock=u0a1:"*walarm*:com.whatsapp.MessageHandler.LOGOUT_ACTION"\n' + "\n" + ) + + assert [record["event"] for record in dba.results] == ["wake", "wake"] + assert dba.results[0]["package_name"] is None + assert dba.results[1]["package_name"] == "com.whatsapp" + + def test_decorated_sync_job_uses_component_package(self): + dba = DumpsysBatteryHistoryArtifact() + dba.parse('+1s (2) 100 +job=u0a1:"@SyncManager@gmail-ls/com.google:android"\n') + + assert dba.results[0]["package_name"] == "com.google" diff --git a/tests/android/test_artifact_dumpsys_dbinfo.py b/tests/android/test_artifact_dumpsys_dbinfo.py index 691d43c45..0dcc0bc30 100644 --- a/tests/android/test_artifact_dumpsys_dbinfo.py +++ b/tests/android/test_artifact_dumpsys_dbinfo.py @@ -53,9 +53,32 @@ def test_parsing_month_day_timestamp_without_pid(self): assert dbi.results == [ { - "isodate": "07-15 20:27:39.431", + "timestamp": "07-15 20:27:39.431", + "pid": None, "action": "executeForCursorWindow", + "duration_ms": 1, + "status": "succeeded", "sql": "SELECT 1", "path": "/data/user/0/com.example/databases/current.db", + "pool_path": "/data/user/0/com.example/databases/current.db", + "connection_number": None, + "is_primary": None, } ] + + def test_parses_operations_from_multiple_connections(self): + dbi = DumpsysDBInfoArtifact() + dbi.parse( + """Connection pool for /data/example.db: + Connection #0: + isPrimaryConnection: true + Most recently executed operations: + 0: [2025-01-01 00:00:00.000] execute took 1ms - succeeded, sql="SELECT 1", path=/data/example.db + Connection #1: + isPrimaryConnection: false + Most recently executed operations: + 0: [2025-01-01 00:00:01.000] execute took 2ms - succeeded, sql="SELECT 2", path=/data/example.db +""" + ) + + assert [record["connection_number"] for record in dbi.results] == [0, 1] diff --git a/tests/android/test_artifact_dumpsys_package_activities.py b/tests/android/test_artifact_dumpsys_package_activities.py index 5eab63d64..66ef79089 100644 --- a/tests/android/test_artifact_dumpsys_package_activities.py +++ b/tests/android/test_artifact_dumpsys_package_activities.py @@ -21,12 +21,16 @@ def test_parsing(self): assert len(dpa.results) == 0 dpa.parse(data) - assert len(dpa.results) == 4 - assert dpa.results[0]["package_name"] == "com.samsung.android.app.social" + assert len(dpa.results) == 10 + assert dpa.results[0]["package_name"] == "com.samsung.android.messaging" assert ( - dpa.results[0]["activity"] - == "com.samsung.android.app.social/.feed.FeedsActivity" + dpa.results[0]["component"] + == "com.samsung.android.messaging/.ui.RcsTransferContent" ) + assert {result["resolver_type"] for result in dpa.results} == { + "full_mime_type", + "non_data_action", + } def test_ioc_check(self, indicator_file): dpa = DumpsysPackageActivitiesArtifact() @@ -41,4 +45,4 @@ def test_ioc_check(self, indicator_file): dpa.indicators = ind assert len(dpa.alertstore.alerts) == 0 dpa.check_indicators() - assert len(dpa.alertstore.alerts) == 1 + assert len(dpa.alertstore.alerts) == 2 diff --git a/tests/android/test_artifact_dumpsys_packages.py b/tests/android/test_artifact_dumpsys_packages.py index bd5fb7533..f22cc67f2 100644 --- a/tests/android/test_artifact_dumpsys_packages.py +++ b/tests/android/test_artifact_dumpsys_packages.py @@ -25,7 +25,11 @@ def test_parsing(self): == "com.samsung.android.provider.filterprovider" ) assert dpa.results[0]["version_name"] == "5.0.07" - assert dpa.results[0]["first_install_time"] == "2008-12-31 16:00:00" + assert dpa.results[0]["version_code"] == 500700000 + assert dpa.results[0]["min_sdk"] == 28 + assert dpa.results[0]["target_sdk"] == 28 + assert dpa.results[0]["package_type"] == "active" + assert dpa.results[0]["users"][0]["user_id"] == 0 assert dpa.results[0]["system"] is True def test_parsing_system_flag(self): @@ -60,42 +64,27 @@ def test_ioc_check(self, indicator_file): dpa.check_indicators() assert len(dpa.alertstore.alerts) == 1 - def test_per_user_fields_use_primary_user(self): - details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details( - """ User 0: installed=true - firstInstallTime=2024-01-10 09:19:39 - runtime permissions: - android.permission.CAMERA: granted=true - User 95: installed=false - firstInstallTime=1970-01-01 01:00:00 - runtime permissions: - android.permission.CAMERA: granted=false - android.permission.RECORD_AUDIO: granted=false + def test_hidden_packages_and_per_user_state(self): + dpa = DumpsysPackagesArtifact() + dpa.parse( + """Packages: + Package [com.example.active]: + appId=10001 + versionCode=12 minSdk=29 targetSdk=35 + User 0: installed=true hidden=false + firstInstallTime=2025-01-01 01:02:03 + User 10: installed=false hidden=true + firstInstallTime=2025-01-02 01:02:03 +Hidden system packages: + Package [com.example.hidden]: + appId=10002 + installerPackageName=null """ ) - assert details["first_install_time"] == "2024-01-10 09:19:39" - runtime_permissions = [ - permission - for permission in details["permissions"] - if permission["type"] == "runtime" - ] - assert runtime_permissions == [ - { - "name": "android.permission.CAMERA", - "granted": True, - "type": "runtime", - } + assert [record["package_type"] for record in dpa.results] == [ + "active", + "hidden_system", ] - - def test_per_user_fields_fall_back_when_user_zero_is_missing(self): - details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details( - """ User 10: installed=true - firstInstallTime=2024-02-10 09:19:39 - runtime permissions: - android.permission.CAMERA: granted=true -""" - ) - - assert details["first_install_time"] == "2024-02-10 09:19:39" - assert details["permissions"][-1]["name"] == "android.permission.CAMERA" + assert len(dpa.results[0]["users"]) == 2 + assert dpa.results[1]["installer"] is None diff --git a/tests/android/test_artifact_dumpsys_platform_compat.py b/tests/android/test_artifact_dumpsys_platform_compat.py index c8c59b3da..c202645ae 100644 --- a/tests/android/test_artifact_dumpsys_platform_compat.py +++ b/tests/android/test_artifact_dumpsys_platform_compat.py @@ -22,6 +22,9 @@ def test_parsing(self): assert len(dbi.results) == 2 assert dbi.results[0]["package_name"] == "org.torproject.torbrowser" assert dbi.results[1]["package_name"] == "org.article19.circulo.next" + assert dbi.results[0]["change_id"] == 168419799 + assert dbi.results[0]["change_name"] == "DOWNSCALED" + assert dbi.results[0]["override_value"] is False def test_ioc_check(self, indicator_file): dbi = DumpsysPlatformCompatArtifact() diff --git a/tests/android/test_artifact_dumpsys_receivers.py b/tests/android/test_artifact_dumpsys_receivers.py index 7875a52ef..6e8b92e00 100644 --- a/tests/android/test_artifact_dumpsys_receivers.py +++ b/tests/android/test_artifact_dumpsys_receivers.py @@ -19,17 +19,14 @@ def test_parsing(self): assert len(dr.results) == 0 dr.parse(data) - assert len(dr.results) == 4 - assert ( - list(dr.results.keys())[0] - == "com.android.storagemanager.automatic.SHOW_NOTIFICATION" - ) - assert ( - dr.results["com.android.storagemanager.automatic.SHOW_NOTIFICATION"][0][ - "package_name" - ] - == "com.android.storagemanager" + assert len(dr.results) == 9 + assert dr.results[0]["resolver_type"] == "full_mime_type" + storage_manager = next( + result + for result in dr.results + if result["key"] == "com.android.storagemanager.automatic.SHOW_NOTIFICATION" ) + assert storage_manager["package_name"] == "com.android.storagemanager" def test_parsing_misindented_action(self): dr = DumpsysReceiversArtifact() @@ -44,12 +41,8 @@ def test_parsing_misindented_action(self): dr.parse(data) - assert ( - dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][ - "package_name" - ] - == "com.psycatgames.nhiegame" - ) + assert dr.results[1]["key"] == "android.intent.action.MY_PACKAGE_REPLACED" + assert dr.results[1]["package_name"] == "com.psycatgames.nhiegame" def test_parsing_misindented_first_action(self): dr = DumpsysReceiversArtifact() @@ -62,12 +55,8 @@ def test_parsing_misindented_first_action(self): dr.parse(data) - assert ( - dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][ - "package_name" - ] - == "com.psycatgames.nhiegame" - ) + assert dr.results[0]["key"] == "android.intent.action.MY_PACKAGE_REPLACED" + assert dr.results[0]["package_name"] == "com.psycatgames.nhiegame" def test_ioc_check(self, indicator_file): dr = DumpsysReceiversArtifact() diff --git a/tests/android/test_artifact_getprop.py b/tests/android/test_artifact_getprop.py index d9fec6b7f..0fbe742e5 100644 --- a/tests/android/test_artifact_getprop.py +++ b/tests/android/test_artifact_getprop.py @@ -39,3 +39,29 @@ def test_ioc_check(self, indicator_file): assert len(gp.alertstore.alerts) == 0 gp.check_indicators() assert len(gp.alertstore.alerts) == 1 + + def test_empty_values_and_invalid_lines(self): + gp = GetProp() + gp.parse("[empty]: []\n[valid]: [value]\n0\n[broken]: [value") + + assert gp.results == [ + {"name": "empty", "value": ""}, + {"name": "valid", "value": "value"}, + ] + + def test_multiline_value(self): + gp = GetProp() + gp.parse( + "[persist.sys.boot.reason.history]: [" + "reboot,ota,1697044974\n" + "reboot,watchdog,1696958574]\n" + "[ro.build.version.sdk]: [35]\n" + ) + + assert gp.results == [ + { + "name": "persist.sys.boot.reason.history", + "value": "reboot,ota,1697044974\nreboot,watchdog,1696958574", + }, + {"name": "ro.build.version.sdk", "value": "35"}, + ] diff --git a/tests/android/test_artifact_processes.py b/tests/android/test_artifact_processes.py index 2806472d6..4d510377c 100644 --- a/tests/android/test_artifact_processes.py +++ b/tests/android/test_artifact_processes.py @@ -20,7 +20,7 @@ def test_parsing(self): assert len(p.results) == 0 p.parse(data) assert len(p.results) == 17 - assert p.results[0]["proc_name"] == "init" + assert p.results[0]["command"] == "init" def test_ioc_check(self, indicator_file): p = Processes() @@ -36,3 +36,14 @@ def test_ioc_check(self, indicator_file): assert len(p.alertstore.alerts) == 0 p.check_indicators() assert len(p.alertstore.alerts) == 1 + + def test_bugreport_thread_columns(self): + p = Processes() + p.parse( + "LABEL USER PID TID PPID VSZ RSS WCHAN ADDR S PRI NI RTPRIO SCH PCY TIME CMD\n" + "u:r:init:s0 root 1 2 0 100 20 0 0 S 19 0 - 0 fg 00:00:01 init\n" + ) + + assert p.results[0]["label"] == "u:r:init:s0" + assert p.results[0]["tid"] == 2 + assert p.results[0]["command"] == "init" diff --git a/tests/android/test_artifact_tombstones.py b/tests/android/test_artifact_tombstones.py index f88a9c261..e9159bdd5 100644 --- a/tests/android/test_artifact_tombstones.py +++ b/tests/android/test_artifact_tombstones.py @@ -128,8 +128,5 @@ def validate_tombstone_result(self, tombstone_result: dict): assert tombstone_result.get("pid") == 25541 assert tombstone_result.get("process_name") == "mtk.ape.decoder" - # With Android logs we want to keep timestamps as device local time for consistency. - # We often don't know the time offset for a log entry and so can't convert everything to UTC. - # MVT should output the local time only: - # So original 2023-04-12 12:32:40.518290770+0200 -> 2023-04-12 12:32:40.000000 - assert tombstone_result.get("timestamp") == "2023-04-12 12:32:40.518290" + # Tombstones include an explicit offset, so normalize them to UTC. + assert tombstone_result.get("timestamp") == "2023-04-12 10:32:40.518290" diff --git a/tests/android_androidqf/test_mounts.py b/tests/android_androidqf/test_mounts.py index d66925d9c..5f471769e 100644 --- a/tests/android_androidqf/test_mounts.py +++ b/tests/android_androidqf/test_mounts.py @@ -13,6 +13,19 @@ class TestAndroidqfMountsArtifact: + def test_parse_proc_mountinfo(self): + from mvt.android.artifacts.mounts import Mounts as MountsArtifact + + results = MountsArtifact.parse_mountinfo( + "41 40 254:13 / / ro,relatime shared:1 - erofs /dev/block/dm-13 ro,seclabel\n", + 123, + ) + + assert results[0]["mount_id"] == 41 + assert results[0]["mount_point"] == "/" + assert results[0]["filesystem_type"] == "erofs" + assert results[0]["process_ids"] == [123] + def test_parse_mounts_token_checks(self): """ Test the artifact-level `parse` method using tolerant token checks. diff --git a/tests/android_androidqf/test_settings.py b/tests/android_androidqf/test_settings.py index ce14460ff..6edfec81f 100644 --- a/tests/android_androidqf/test_settings.py +++ b/tests/android_androidqf/test_settings.py @@ -6,12 +6,27 @@ from pathlib import Path from mvt.android.modules.androidqf.aqf_settings import AQFSettings +from mvt.android.artifacts.settings import Settings from mvt.common.module import run_module from ..utils import get_android_androidqf, list_files class TestSettingsModule: + def test_bugreport_settings_format(self): + settings = Settings() + settings.parse( + "GLOBAL SETTINGS (user 0)\n" + "_id:1 name:adb_wifi_enabled pkg:android value:0 default:0 defaultSystemSet:true\n" + "SECURE SETTINGS (user 10)\n" + "_id:2 name:accessibility_enabled pkg:android value:1\n" + ) + + assert settings.results == { + "global:user_0": {"adb_wifi_enabled": "0"}, + "secure:user_10": {"accessibility_enabled": "1"}, + } + def test_parsing(self): data_path = get_android_androidqf() m = AQFSettings(target_path=data_path) diff --git a/tests/android_bugreport/test_bugreport.py b/tests/android_bugreport/test_bugreport.py index 75df1061d..413bb1802 100644 --- a/tests/android_bugreport/test_bugreport.py +++ b/tests/android_bugreport/test_bugreport.py @@ -54,10 +54,11 @@ def test_packages_module(self): == "com.samsung.android.provider.filterprovider" ) assert m.results[1]["package_name"] == "com.instagram.android" - assert m.results[0]["installer"] == "" + assert m.results[0]["installer"] is None assert m.results[1]["installer"] == "com.android.vending" assert len(m.results[0]["permissions"]) == 4 - assert len(m.results[1]["permissions"]) == 32 + assert len(m.results[1]["permissions"]) == 20 + assert len(m.results[1]["users"][0]["permissions"]) == 19 def test_getprop_module(self): m = self.launch_bug_report_module(DumpsysGetProp) @@ -66,29 +67,34 @@ def test_getprop_module(self): def test_receivers_match_exact_package_name(self, indicators_factory): intent = "android.intent.action.PHONE_STATE" false_positive = { + "resolver_type": "non_data_action", + "key": intent, "package_name": "com.android.phone", - "receiver": ( + "component": ( "com.android.phone/" "com.android.services.telephony.sip.SipIncomingCallReceiver" ), + "filter_count": 1, } malicious_receiver = { + "resolver_type": "non_data_action", + "key": intent, "package_name": "com.android.services", - "receiver": "com.android.services/com.example.SomeReceiver", + "component": "com.android.services/com.example.SomeReceiver", + "filter_count": 1, } - module = DumpsysReceivers( - results={intent: [false_positive, malicious_receiver]} - ) + module = DumpsysReceivers(results=[false_positive, malicious_receiver]) module.indicators = indicators_factory(app_ids=["com.android.services"]) module.check_indicators() assert len(module.alertstore.alerts) == 1 alert = module.alertstore.alerts[0] - assert alert.event == {intent: malicious_receiver} + assert alert.event == malicious_receiver assert alert.matched_indicator.value == "com.android.services" def test_tombstones_modules(self): m = self.launch_bug_report_module(Tombstones) assert len(m.results) == 2 assert m.results[1]["pid"] == 3559 + assert m.results[0]["sources"]["text"]["parsed"] is True