mirror of
https://github.com/mvt-project/mvt.git
synced 2026-07-17 09:57:24 +02:00
Refactor DumpsysBatteryDaily module and add related artifact
This commit is contained in:
@@ -0,0 +1,78 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-2023 Claudio Guarnieri.
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
from typing import Union
|
||||
|
||||
from .artifact import AndroidArtifact
|
||||
|
||||
|
||||
class DumpsysBatteryDailyArtifact(AndroidArtifact):
|
||||
"""
|
||||
Parser for dumpsys dattery daily updates.
|
||||
"""
|
||||
|
||||
def serialize(self, record: dict) -> Union[dict, list]:
|
||||
return {
|
||||
"timestamp": record["from"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "battery_daily",
|
||||
"data": f"Recorded update of package {record['package_name']} "
|
||||
f"with vers {record['vers']}",
|
||||
}
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
if not self.indicators:
|
||||
return
|
||||
|
||||
for result in self.results:
|
||||
ioc = self.indicators.check_app_id(result["package_name"])
|
||||
if ioc:
|
||||
result["matched_indicator"] = ioc
|
||||
self.detected.append(result)
|
||||
continue
|
||||
|
||||
def parse(self, output: str) -> None:
|
||||
daily = None
|
||||
daily_updates = []
|
||||
for line in output.splitlines():
|
||||
if line.startswith(" Daily from "):
|
||||
if len(daily_updates) > 0:
|
||||
self.results.extend(daily_updates)
|
||||
daily_updates = []
|
||||
|
||||
timeframe = line[13:].strip()
|
||||
date_from, date_to = timeframe.strip(":").split(" to ", 1)
|
||||
daily = {"from": date_from[0:10], "to": date_to[0:10]}
|
||||
continue
|
||||
|
||||
if not daily:
|
||||
continue
|
||||
|
||||
if not line.strip().startswith("Update "):
|
||||
continue
|
||||
|
||||
line = line.strip().replace("Update ", "")
|
||||
package_name, vers = line.split(" ", 1)
|
||||
vers_nr = vers.split("=", 1)[1]
|
||||
|
||||
already_seen = False
|
||||
for update in daily_updates:
|
||||
if package_name == update["package_name"] and vers_nr == update["vers"]:
|
||||
already_seen = True
|
||||
break
|
||||
|
||||
if not already_seen:
|
||||
daily_updates.append(
|
||||
{
|
||||
"action": "update",
|
||||
"from": daily["from"],
|
||||
"to": daily["to"],
|
||||
"package_name": package_name,
|
||||
"vers": vers_nr,
|
||||
}
|
||||
)
|
||||
|
||||
if len(daily_updates) > 0:
|
||||
self.results.extend(daily_updates)
|
||||
@@ -4,14 +4,14 @@
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import logging
|
||||
from typing import Optional, Union
|
||||
from typing import Optional
|
||||
|
||||
from mvt.android.parsers import parse_dumpsys_battery_daily
|
||||
from mvt.android.artifacts.dumpsys_battery_daily import DumpsysBatteryDailyArtifact
|
||||
|
||||
from .base import AndroidExtraction
|
||||
|
||||
|
||||
class DumpsysBatteryDaily(AndroidExtraction):
|
||||
class DumpsysBatteryDaily(DumpsysBatteryDailyArtifact, AndroidExtraction):
|
||||
"""This module extracts records from battery daily updates."""
|
||||
|
||||
def __init__(
|
||||
@@ -32,32 +32,12 @@ class DumpsysBatteryDaily(AndroidExtraction):
|
||||
results=results,
|
||||
)
|
||||
|
||||
def serialize(self, record: dict) -> Union[dict, list]:
|
||||
return {
|
||||
"timestamp": record["from"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "battery_daily",
|
||||
"data": f"Recorded update of package {record['package_name']} "
|
||||
f"with vers {record['vers']}",
|
||||
}
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
if not self.indicators:
|
||||
return
|
||||
|
||||
for result in self.results:
|
||||
ioc = self.indicators.check_app_id(result["package_name"])
|
||||
if ioc:
|
||||
result["matched_indicator"] = ioc
|
||||
self.detected.append(result)
|
||||
continue
|
||||
|
||||
def run(self) -> None:
|
||||
self._adb_connect()
|
||||
output = self._adb_command("dumpsys batterystats --daily")
|
||||
self._adb_disconnect()
|
||||
|
||||
self.results = parse_dumpsys_battery_daily(output)
|
||||
self.parse(output)
|
||||
|
||||
self.log.info(
|
||||
"Extracted %d records from battery daily stats", len(self.results)
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-2023 Claudio Guarnieri.
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from mvt.android.artifacts.dumpsys_battery_daily import DumpsysBatteryDailyArtifact
|
||||
|
||||
from .base import AndroidQFModule
|
||||
|
||||
|
||||
class DumpsysBatteryDaily(DumpsysBatteryDailyArtifact, AndroidQFModule):
|
||||
def __init__(
|
||||
self,
|
||||
file_path: Optional[str] = None,
|
||||
target_path: Optional[str] = None,
|
||||
results_path: Optional[str] = None,
|
||||
module_options: Optional[dict] = None,
|
||||
log: logging.Logger = logging.getLogger(__name__),
|
||||
results: Optional[list] = None,
|
||||
) -> None:
|
||||
super().__init__(
|
||||
file_path=file_path,
|
||||
target_path=target_path,
|
||||
results_path=results_path,
|
||||
module_options=module_options,
|
||||
log=log,
|
||||
results=results,
|
||||
)
|
||||
|
||||
def run(self) -> None:
|
||||
dumpsys_file = self._get_files_by_pattern("*/dumpsys.txt")
|
||||
if not dumpsys_file:
|
||||
return
|
||||
|
||||
# Extract section
|
||||
data = self._get_file_content(dumpsys_file[0])
|
||||
section = self.extract_dumpsys_section(
|
||||
data.decode("utf-8", errors="replace"), "DUMP OF SERVICE batterystats:"
|
||||
)
|
||||
|
||||
# Parse it
|
||||
self.parse(section)
|
||||
self.log.info("Extracted a total of %d battery daily stats", len(self.results))
|
||||
@@ -4,14 +4,14 @@
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import logging
|
||||
from typing import Optional, Union
|
||||
from typing import Optional
|
||||
|
||||
from mvt.android.parsers import parse_dumpsys_battery_daily
|
||||
from mvt.android.artifacts.dumpsys_battery_daily import DumpsysBatteryDailyArtifact
|
||||
|
||||
from .base import BugReportModule
|
||||
|
||||
|
||||
class BatteryDaily(BugReportModule):
|
||||
class BatteryDaily(DumpsysBatteryDailyArtifact, BugReportModule):
|
||||
"""This module extracts records from battery daily updates."""
|
||||
|
||||
def __init__(
|
||||
@@ -32,26 +32,6 @@ class BatteryDaily(BugReportModule):
|
||||
results=results,
|
||||
)
|
||||
|
||||
def serialize(self, record: dict) -> Union[dict, list]:
|
||||
return {
|
||||
"timestamp": record["from"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "battery_daily",
|
||||
"data": f"Recorded update of package {record['package_name']} "
|
||||
f"with vers {record['vers']}",
|
||||
}
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
if not self.indicators:
|
||||
return
|
||||
|
||||
for result in self.results:
|
||||
ioc = self.indicators.check_app_id(result["package_name"])
|
||||
if ioc:
|
||||
result["matched_indicator"] = ioc
|
||||
self.detected.append(result)
|
||||
continue
|
||||
|
||||
def run(self) -> None:
|
||||
content = self._get_dumpstate_file()
|
||||
if not content:
|
||||
@@ -61,30 +41,9 @@ class BatteryDaily(BugReportModule):
|
||||
)
|
||||
return
|
||||
|
||||
lines = []
|
||||
in_batterystats = False
|
||||
in_daily = False
|
||||
for line in content.decode(errors="ignore").splitlines():
|
||||
if line.strip() == "DUMP OF SERVICE batterystats:":
|
||||
in_batterystats = True
|
||||
continue
|
||||
|
||||
if not in_batterystats:
|
||||
continue
|
||||
|
||||
if line.strip() == "Daily stats:":
|
||||
lines.append(line)
|
||||
in_daily = True
|
||||
continue
|
||||
|
||||
if not in_daily:
|
||||
continue
|
||||
|
||||
if line.strip() == "":
|
||||
break
|
||||
|
||||
lines.append(line)
|
||||
|
||||
self.results = parse_dumpsys_battery_daily("\n".join(lines))
|
||||
dumpsys_section = self.extract_dumpsys_section(
|
||||
content.decode("utf-8", errors="replace"), "DUMP OF SERVICE batterystats:"
|
||||
)
|
||||
self.parse(dumpsys_section)
|
||||
|
||||
self.log.info("Extracted a total of %d battery daily stats", len(self.results))
|
||||
|
||||
@@ -4,7 +4,6 @@
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
from .dumpsys import (
|
||||
parse_dumpsys_battery_daily,
|
||||
parse_dumpsys_battery_history,
|
||||
parse_dumpsys_receiver_resolver_table,
|
||||
)
|
||||
|
||||
@@ -7,54 +7,6 @@ import re
|
||||
from typing import Any, Dict, List
|
||||
|
||||
|
||||
def parse_dumpsys_battery_daily(output: str) -> list:
|
||||
results = []
|
||||
daily = None
|
||||
daily_updates = []
|
||||
for line in output.splitlines():
|
||||
if line.startswith(" Daily from "):
|
||||
if len(daily_updates) > 0:
|
||||
results.extend(daily_updates)
|
||||
daily_updates = []
|
||||
|
||||
timeframe = line[13:].strip()
|
||||
date_from, date_to = timeframe.strip(":").split(" to ", 1)
|
||||
daily = {"from": date_from[0:10], "to": date_to[0:10]}
|
||||
continue
|
||||
|
||||
if not daily:
|
||||
continue
|
||||
|
||||
if not line.strip().startswith("Update "):
|
||||
continue
|
||||
|
||||
line = line.strip().replace("Update ", "")
|
||||
package_name, vers = line.split(" ", 1)
|
||||
vers_nr = vers.split("=", 1)[1]
|
||||
|
||||
already_seen = False
|
||||
for update in daily_updates:
|
||||
if package_name == update["package_name"] and vers_nr == update["vers"]:
|
||||
already_seen = True
|
||||
break
|
||||
|
||||
if not already_seen:
|
||||
daily_updates.append(
|
||||
{
|
||||
"action": "update",
|
||||
"from": daily["from"],
|
||||
"to": daily["to"],
|
||||
"package_name": package_name,
|
||||
"vers": vers_nr,
|
||||
}
|
||||
)
|
||||
|
||||
if len(daily_updates) > 0:
|
||||
results.extend(daily_updates)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def parse_dumpsys_battery_history(output: str) -> List[Dict[str, Any]]:
|
||||
results = []
|
||||
|
||||
|
||||
Reference in New Issue
Block a user