From 288b3136498c0746ae0dffa979c71fe6c07966dc Mon Sep 17 00:00:00 2001 From: Janik Besendorf Date: Sat, 22 Aug 2026 14:16:23 +0200 Subject: [PATCH] Parse complete battery history events --- .../artifacts/dumpsys_battery_history.py | 225 ++++++++++++------ .../test_artifact_dumpsys_battery_history.py | 21 ++ 2 files changed, 174 insertions(+), 72 deletions(-) diff --git a/src/mvt/android/artifacts/dumpsys_battery_history.py b/src/mvt/android/artifacts/dumpsys_battery_history.py index 77d9d3c..6cb36f1 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 - - 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] - ): - 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() - ) - 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: + 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 - self.results.append( - { - "time_elapsed": time_elapsed, - "event": event, - "uid": uid, - "package_name": package_name, - "service": service, - } - ) + 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, + } + ) + + for sign, uid, raw_service in re.findall( + r"([+-])job=([^:\s]+):\"([^\"]+)\"", line + ): + 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], + ) + + 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/tests/android/test_artifact_dumpsys_battery_history.py b/tests/android/test_artifact_dumpsys_battery_history.py index dd0b120..091ce55 100644 --- a/tests/android/test_artifact_dumpsys_battery_history.py +++ b/tests/android/test_artifact_dumpsys_battery_history.py @@ -54,6 +54,7 @@ class TestDumpsysBatteryHistoryArtifact: 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 @@ class TestDumpsysBatteryHistoryArtifact: } 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"