Compare commits

..
Author SHA1 Message Date
Janik Besendorf dd96ebd409 Add Android bugreport schema version 2026-08-22 14:22:35 +02:00
Janik Besendorf fd27fe3378 Fix module registry type annotations 2026-08-22 14:21:51 +02:00
Janik Besendorf 7ef26779bf Register added bugreport parsers 2026-08-22 14:21:51 +02:00
Janik Besendorf 3a43f6fcc7 Merge plaintext and protobuf tombstones 2026-08-22 14:21:51 +02:00
Janik Besendorf 45fdf85790 Add bugreport mountinfo parser 2026-08-22 14:21:51 +02:00
Janik Besendorf 627ab32d42 Add bugreport settings parser 2026-08-22 14:21:51 +02:00
Janik Besendorf 8b85972cc3 Add bugreport process table parser 2026-08-22 14:21:51 +02:00
Janik Besendorf f747751b09 Add timezone provenance to file timestamps 2026-08-22 14:21:51 +02:00
Janik Besendorf 832e46604d Parse all platform compatibility overrides 2026-08-22 14:21:51 +02:00
Janik Besendorf a3c8b56102 Parse typed multi-user package details 2026-08-22 14:21:51 +02:00
Janik Besendorf e9bf197ec6 Parse all database operation details 2026-08-22 14:20:59 +02:00
Janik Besendorf 288b313649 Parse complete battery history events 2026-08-22 14:20:59 +02:00
Janik Besendorf 87a27f6e37 Normalize battery daily update records 2026-08-22 14:20:59 +02:00
Janik Besendorf 1bf2f9b35a Retain AppOps UID and event details 2026-08-22 14:20:59 +02:00
Janik Besendorf b904414bb0 Parse ADB binary XML keys 2026-08-22 14:20:59 +02:00
Janik Besendorf 008844b440 Parse accessibility service states per user 2026-08-22 14:20:59 +02:00
Janik Besendorf ab879bb23b Parse all package resolver categories 2026-08-22 14:20:59 +02:00
Janik Besendorf dc0650bf76 Fix bugreport command section extraction 2026-08-22 14:20:59 +02:00
60 changed files with 1491 additions and 1826 deletions

No files matched your search

+1 -1
View File
@@ -84,7 +84,7 @@ details.
Users can also add top-level commands to `mvt-ios` and `mvt-android` from
installed Python packages or local files and folders. See the
[custom CLI command documentation](https://docs.mvt.re/en/latest/development/custom_commands/)
[custom CLI command documentation](https://docs.mvt.re/en/latest/custom_commands/)
for the plugin entry points and `--load-command` interface.
@@ -1,7 +1,7 @@
# Custom CLI Commands
MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
Custom commands are different from [custom forensic modules](index.md#custom-modules):
Custom commands are different from [custom forensic modules](development.md#custom-modules):
commands add new CLI operations, while modules add analysis steps to existing
`check-*` commands.
@@ -56,10 +56,6 @@ pipx inject mvt my-mvt-plugin
When MVT is installed in an active virtual environment, install the plugin with
`pip` in that environment.
Command packages that need their own settings, such as an API key, should store
them in a namespaced [plugin configuration file](plugin_configuration.md)
rather than in MVT's own `config.yaml`.
## Load a Command File
For local commands that are not packaged, create a Python file that exports one
@@ -35,16 +35,9 @@ class DependentModule(MVTModule):
prerequisite_results = self.get_dependency_results(PrerequisiteModule)
```
Selecting a single module also runs its transitive dependencies.
A module can only depend on modules the command it runs in also has. When a
declared dependency is not among them, the command logs a warning naming the
module and the missing dependency, skips that module and everything depending
on it, and runs the rest of the analysis. Selecting such a module with
`--module` therefore leaves nothing to run, which the warning explains.
A cycle in the dependency graph is a programming error rather than a
configuration problem: the command logs a warning and runs no modules at all.
Selecting a single module also runs its transitive dependencies. If a dependency
is unavailable or the dependency graph contains a cycle, the command logs a
warning and does not run any modules.
## Custom modules
@@ -167,10 +160,6 @@ For a `pipx` installation of MVT, inject the package into MVT's environment:
pipx inject mvt mvt-plugin-amnesty-custom
```
Module packages that need their own settings, such as an API key, should store
them in a namespaced [plugin configuration file](plugin_configuration.md)
rather than in MVT's own `config.yaml`.
### Naming module packages
Name module packages `mvt-plugin-<name>` (import package `mvt_plugin_<name>`),
-149
View File
@@ -1,149 +0,0 @@
# Plugin Configuration
Plugin packages that add [custom CLI commands](custom_commands.md) or
[modules](index.md#installed-module-packages) often need to store their
own settings, such as an API key, a server URL or the timestamp of the last
synchronization. MVT provides a namespaced settings base class so each plugin
keeps its configuration in its own file, and a data folder for anything else a
plugin needs to keep on disk.
!!! warning
Do not write plugin settings to MVT's own `config.yaml`. MVT rewrites that
file with the settings it knows about every time it starts, so any other
section is deleted.
## Where Settings Are Stored
Each plugin gets one YAML file in a `plugins` folder next to MVT's own
configuration:
```
~/.config/mvt/plugins/<plugin name>.yaml
```
The exact parent folder follows the platform convention used for MVT's
`config.yaml` (for example `~/Library/Application Support/mvt` on macOS). Use
`mvt.common.plugin_config.plugin_config_path()` instead of building the path by
hand.
Plugin names must be lowercase and may only contain letters, digits and dashes,
matching the `mvt-plugin-<name>` package naming convention. MVT creates the
`plugins` folder with `0700` permissions and writes the settings files with
`0600` permissions, because they commonly hold credentials. Files are written
through a temporary file and moved into place, so an interrupted save never
leaves a partially written settings file behind.
## Plugin Data Folder
Everything else a plugin keeps on disk, such as a cache, a downloaded artifact
or synchronization state, belongs in the folder returned by the `data_folder()`
class method of the plugin's settings class, or by
`mvt.common.plugin_config.plugin_data_folder()` called with the plugin name if
the plugin has no settings class:
```
~/.local/share/mvt/plugin-data/<plugin name>/ # Linux
~/Library/Application Support/mvt/plugin-data/<plugin name>/ # macOS
```
The folder sits beside MVT's own data, such as the downloaded indicators. It is
created if it is missing, with `0700` permissions. Asking for it again returns
the same path and leaves the contents alone, so a plugin can ask for it every
time it needs the folder. `ExamplePluginSettings` below is the settings class
defined in the next section:
```python
import os
def cache_path() -> str:
folder = ExamplePluginSettings.data_folder()
return os.path.join(folder, "virustotal_lookups_cache.json")
```
A plugin which has no settings class calls
`plugin_data_folder("example-plugin")` instead.
Do not fall back on a path of your own such as `~/.cache/example-plugin`: it
is a Linux-only convention, and MVT will not create it for you.
## Defining Plugin Settings
Subclass `MVTPluginSettings`, set `plugin_name` and declare typed fields with
defaults:
```python
from typing import Optional
from mvt.common.plugin_config import MVTPluginSettings
class ExamplePluginSettings(MVTPluginSettings):
plugin_name = "example-plugin"
API_KEY: Optional[str] = None
MAX_RESULTS: int = 25
LAST_SYNC: Optional[str] = None
```
`load()` returns the current settings and `save()` writes them back:
```python
from datetime import datetime, timezone
import click
def sync():
settings = ExamplePluginSettings.load()
if not settings.API_KEY:
raise click.ClickException(
"No API key configured. Set MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY or "
"run 'example-plugin configure'."
)
settings.LAST_SYNC = datetime.now(timezone.utc).isoformat()
settings.save()
```
A missing settings file is not an error: the plugin then runs on the field
defaults and on whatever the environment provides. `save()` only persists the
values that differ from the defaults, and it never touches MVT's `config.yaml`.
A settings file that cannot be parsed, or that does not hold a mapping of
setting names to values, raises a `PluginConfigLoadError` naming the file.
Every subclass that sets its own `plugin_name` gets its own file and its own
environment namespace. A subclass that does not redefine `plugin_name` inherits
it, and therefore shares the file and the environment variables of its parent
class.
## Environment Variables
Every field can also be set with an environment variable. The prefix is
`MVT_PLUGIN_`, followed by the plugin name upper-cased with dashes replaced by
underscores, followed by the field name. For the example above:
```bash
export MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY=...
export MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS=50
```
Settings resolve in this order, from highest to lowest priority:
1. Arguments passed to the settings class directly, such as
`ExamplePluginSettings(API_KEY="...")`
2. Environment variables
3. The plugin's YAML file
4. The field defaults declared on the settings class
!!! tip
On shared or multi-user machines, prefer passing API keys through
environment variables rather than saving them to the plugin file. `save()`
skips every value that the environment currently supplies, so a credential
provided that way is not copied into the settings file when a plugin saves
an unrelated setting.
Unknown keys in a plugin's YAML file are ignored, so a settings file written by
a newer version of a plugin does not break an older one.
+2 -3
View File
@@ -50,7 +50,6 @@ nav:
- Check Android Intrusion Logs: "android/intrusion_logs.md"
- Indicators of Compromise: "iocs.md"
- Development:
- Development Instructions: "development/index.md"
- Custom CLI Commands: "development/custom_commands.md"
- Plugin Configuration: "development/plugin_configuration.md"
- Development Instructions: "development.md"
- Custom CLI Commands: "custom_commands.md"
- License: "license.md"
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/
import re
from typing import Any
from .artifact import AndroidArtifact
@@ -20,10 +21,10 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
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 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
: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
continue
user_match = re.search(r"attributes:\{id=(\d+)", line)
if user_match:
user_id = int(user_match.group(1))
if not in_services:
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,
}
stripped = line.strip()
state_match = re.match(
r"(?i)(installed|enabled|binding|bound|crashed) services\s*:\s*\{(.*)",
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>
package_name, _, service = match.group(1).partition("/")
self.results.append(
{"package_name": package_name, "service": service}
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 state:
continue
if stripped == "}" or stripped.startswith("AccessibilityInputFilter"):
state = None
continue
component_match = re.search(
r"(?:\d+\s*:\s*)?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?",
stripped,
)
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,
}
+41 -1
View File
@@ -6,6 +6,7 @@
import base64
import binascii
import hashlib
import re
from .artifact import AndroidArtifact
@@ -98,6 +99,34 @@ class DumpsysADBArtifact(AndroidArtifact):
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"(?<![A-Za-z0-9+/])([A-Za-z0-9+/]{300,}={0,2})"
rb"(?: ([A-Za-z0-9_.@-]+))?",
data,
):
key = match.group(1)
try:
base64.b64decode(key, validate=True)
except (binascii.Error, ValueError):
continue
if key in seen:
continue
seen.add(key)
full_key = key + (b" " + match.group(2) if match.group(2) else b"")
key_info = self.calculate_key_info(full_key)
key_info["last_connected"] = None
keystore.append(key_info)
return keystore
@staticmethod
def calculate_key_info(user_key: bytes) -> dict:
if b" " in user_key:
@@ -118,7 +147,7 @@ class DumpsysADBArtifact(AndroidArtifact):
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 @@ class DumpsysADBArtifact(AndroidArtifact):
# Keystore is in XML format on some devices and we need to parse it
if keystore_data and keystore_data.startswith(b"<?xml"):
parsed["debugging_manager"]["keystore"] = self.parse_xml(keystore_data)
elif keystore_data and keystore_data.startswith(b"ABX\x00"):
parsed["debugging_manager"]["keystore"] = self.parse_binary_xml(
keystore_data
)
else:
# Keystore is not XML format
parsed["debugging_manager"]["keystore"] = keystore_data
parsed = parsed["debugging_manager"]
for key, value in list(parsed.items()):
if isinstance(value, bytes):
decoded = value.decode("utf-8", errors="replace")
parsed[key] = (
decoded == "true" if decoded in ("true", "false") else decoded
)
# Calculate key fingerprints for better readability
key_info = []
for user_key in parsed.get("user_keys", []):
+122 -81
View File
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/
from datetime import datetime
import re
from typing import Any
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
@@ -27,14 +28,14 @@ class DumpsysAppopsArtifact(AndroidArtifact):
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 @@ class DumpsysAppopsArtifact(AndroidArtifact):
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 @@ class DumpsysAppopsArtifact(AndroidArtifact):
):
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 @@ class DumpsysAppopsArtifact(AndroidArtifact):
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)
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
perm = {}
self.results.append(package)
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 "):
if entry:
perm["entries"].append(entry)
entry = {}
if package:
if perm:
package["permissions"].append(perm)
perm = {}
self.results.append(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
attribution_match = re.match(r"^\s{8,}([^=]+)=\[$", line)
if attribution_match:
attribution = attribution_match.group(1).strip()
continue
if line.strip() == "]":
attribution = None
continue
if entry:
perm["entries"].append(entry)
entry = {}
entry["access"] = access_type
entry["type"] = line[line.find("[") + 1 : line.find("]")]
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
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,
}
)
if line.strip() == "":
break
if entry:
perm["entries"].append(entry)
if perm:
package["permissions"].append(perm)
if package:
self.results.append(package)
finish_package()
@@ -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 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
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 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
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 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
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 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
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 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
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,
)
@@ -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),
)
+25 -11
View File
@@ -29,16 +29,27 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
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 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
if not line.startswith(" "):
in_operations = False
pool = None
continue
match = rxp.match(line)
@@ -60,11 +70,15 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
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)
@@ -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 @@ class DumpsysPackageActivitiesArtifact(AndroidArtifact):
)
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")
+205 -183
View File
@@ -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 mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
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": "",
"system": False,
"permissions": list(),
"requested_permissions": list(),
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()
],
}
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]]] = {}
@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": [],
"requested_permissions": [],
"users": [],
}
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))
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"(?<!^)(?=[A-Z])", "_", key).lower())
current_user[clean_key] = _value(raw)
details["users"].append(current_user)
permission_section = None
continue
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]
header = stripped.lower()
if header in {
"declared permissions:",
"install permissions:",
"requested permissions:",
"runtime permissions:",
}:
permission_section = header.split()[0]
continue
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
def parse_dumpsys_packages(self, output: str) -> List[Dict[str, Any]]:
"""
Parse the dumpsys package service data
"""
pkg_rxp = re.compile(r" Package \[(.+?)\].*")
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 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
if not package_name:
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]
lines.append(line)
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
if len(lines) > 0:
details = self.parse_dumpsys_package_for_details("\n".join(lines))
package.update(details)
results.append(package)
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] = []
return results
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 = []
def parse(self, content: str):
"""
Parse the Dumpsys Package section for activities
Adds results to self.results
:param content: content of the package section (string)
"""
self.results = []
package = []
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()
@@ -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 @@ class DumpsysPlatformCompatArtifact(AndroidArtifact):
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,
}
)
+39 -108
View File
@@ -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"],
)
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
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
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")
+4 -4
View File
@@ -28,18 +28,18 @@ INTERESTING_PROPERTIES = [
class GetProp(AndroidArtifact):
def parse(self, entry: str) -> None:
self.results: List[Dict[str, str]] = []
rxp = re.compile(r"\[(.+?)\]: \[(.+?)\]")
rxp = re.compile(r"^\[([^]]+)\]: \[(.*)\]$")
for line in entry.splitlines():
line = line.strip()
if line == "":
continue
matches = re.findall(rxp, line)
if not matches or len(matches[0]) != 2:
match = rxp.match(line)
if not match:
continue
prop_entry = {"name": matches[0][0], "value": matches[0][1]}
prop_entry = {"name": match.group(1), "value": match.group(2)}
self.results.append(prop_entry)
def get_device_timezone(self) -> str | None:
+45
View File
@@ -117,6 +117,51 @@ class Mounts(AndroidArtifact):
# 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
@@ -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<component>\S+)"
r"(?:\s+\((?P<filters>\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
+73 -54
View File
@@ -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:
command = result.get("command", "")
if not isinstance(command, str):
continue
# Skipping this process because of false positives.
if result["proc_name"] == "gatekeeperd":
process_name = command.rsplit("/", 1)[-1]
if not process_name or process_name == "gatekeeperd":
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
)
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
+24
View File
@@ -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 @@ ANDROID_DANGEROUS_SETTINGS = [
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():
+25 -7
View File
@@ -131,6 +131,13 @@ class TombstoneCrashArtifact(AndroidArtifact):
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 @@ class TombstoneCrashArtifact(AndroidArtifact):
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 @@ class TombstoneCrashArtifact(AndroidArtifact):
# 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
@@ -195,7 +213,9 @@ class TombstoneCrashArtifact(AndroidArtifact):
if line_key != key:
raise ValueError(f"Expected key {key}, got {line_key}")
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":
@@ -269,9 +289,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
@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:
+1
View File
@@ -53,6 +53,7 @@ class CmdAndroidCheckBugreport(Command):
self.platform = "android"
self.name = "check-bugreport"
self.output_schema_version = 2
self.modules = BUGREPORT_MODULES
self.__format: str = ""
@@ -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 .aqf_settings import AQFSettings
from .mounts import Mounts
from .root_binaries import RootBinaries
ANDROIDQF_MODULES = [
ANDROIDQF_MODULES: list[type[MVTModule]] = [
AQFPackages,
AQFProcesses,
AQFGetProp,
+3 -1
View File
@@ -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]
@@ -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_receivers import DumpsysReceivers
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 @@ BUGREPORT_MODULES = [
DumpsysADBState,
BugReportTimestamps,
Tombstones,
Processes,
Settings,
Mounts,
]
+26 -1
View File
@@ -72,7 +72,11 @@ class BugReportModule(MVTModule):
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 @@ class BugReportModule(MVTModule):
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
@@ -49,9 +49,7 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
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)
@@ -44,21 +44,8 @@ class DumpsysGetProp(GetPropArtifact, BugReportModule):
)
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))
@@ -50,7 +50,10 @@ class DumpsysPackages(DumpsysPackagesArtifact, BugReportModule):
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
@@ -13,7 +13,7 @@ from mvt.common.module_types import ModuleResults
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 @@ class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule):
)
self.parse(content)
self.log.info("Found %d uninstalled apps", len(self.results))
self.log.info("Found %d package compatibility overrides", len(self.results))
@@ -33,7 +33,7 @@ class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule):
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 @@ class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule):
)
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))
@@ -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 @@ class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
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"
),
}
)
@@ -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))
@@ -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))
@@ -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)
+68 -13
View File
@@ -43,21 +43,76 @@ class Tombstones(TombstoneCrashArtifact, BugReportModule):
)
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",
@@ -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,
+19 -95
View File
@@ -54,6 +54,7 @@ class Command:
) -> None:
self.name = ""
self.platform = ""
self.output_schema_version: Optional[int] = None
self.modules: list[type[MVTModule]] = []
self.custom_modules = custom_modules if custom_modules else []
@@ -186,6 +187,8 @@ class Command:
"ioc_files": [],
"hashes": [],
}
if self.output_schema_version is not None:
info["output_schema_version"] = self.output_schema_version
for coll in self.iocs.ioc_collections:
ioc_file_path = coll.get("stix2_file_path", "")
@@ -317,81 +320,6 @@ class Command:
console.print("")
console.print(panel)
def _skipped_modules(
self,
required: list[type[MVTModule]],
module_indexes: dict[type[MVTModule], int],
) -> dict[type[MVTModule], tuple[type[MVTModule], type[MVTModule]]]:
"""Return the modules to drop because a dependency is unavailable.
A module declaring a dependency this command cannot provide is unable
to run, and so is every module depending on it. Dropping only those
keeps a single wrong declaration - in a module scoped to several
commands, for example - from silencing an entire analysis.
The returned mapping gives, for each skipped module, the module which
is missing a dependency and the dependency it is missing.
"""
skipped: dict[type[MVTModule], tuple[type[MVTModule], type[MVTModule]]] = {}
# Skipping the one module a run was asked for leaves nothing to run,
# which the caller reports instead.
remainder = (
"" if self.module_name else " The rest of the analysis will still run."
)
# Skipping one module can skip the modules depending on it, which the
# pass over the module list may already have gone past, so repeat the
# pass until nothing changes.
changed = True
while changed:
changed = False
for module in required:
if module in skipped:
continue
for dependency in module.dependencies:
if dependency not in module_indexes:
skipped[module] = (module, dependency)
changed = True
self.log.warning(
"Module %s will be SKIPPED: it depends on module "
"%s, which is not available in this command.%s",
module.__name__,
dependency.__name__,
remainder,
)
break
if dependency in skipped:
root, missing = skipped[dependency]
skipped[module] = (root, missing)
changed = True
if dependency is root:
self.log.warning(
"Module %s will be SKIPPED: it depends on "
"module %s, itself skipped for depending on "
"unavailable module %s.%s",
module.__name__,
dependency.__name__,
missing.__name__,
remainder,
)
else:
self.log.warning(
"Module %s will be SKIPPED: it depends on "
"skipped module %s, in a chain starting at "
"module %s, which depends on unavailable "
"module %s.%s",
module.__name__,
dependency.__name__,
root.__name__,
missing.__name__,
remainder,
)
break
return skipped
def _ordered_modules(self) -> Optional[list[type[MVTModule]]]:
"""Return enabled modules in stable topological order."""
modules = self._available_modules()
@@ -404,34 +332,30 @@ class Command:
else:
selected = [module for module in modules if module.enabled]
required: set[type[MVTModule]] = set()
required = set(selected)
pending = list(selected)
while pending:
module = pending.pop()
if module in required:
continue
required.add(module)
for dependency in module.dependencies:
# Unavailable dependencies are reported by _skipped_modules().
if dependency in module_indexes:
if dependency not in module_indexes:
self.log.warning(
"Module %s depends on unavailable module %s. "
"No modules will be run.",
module.__name__,
dependency.__name__,
)
return None
if dependency not in required:
required.add(dependency)
pending.append(dependency)
ordered_required = sorted(required, key=lambda module: module_indexes[module])
skipped = self._skipped_modules(ordered_required, module_indexes)
runnable = [module for module in ordered_required if module not in skipped]
if skipped and not runnable:
self.log.warning(
"Every selected module was skipped for an unavailable "
"dependency. No modules will be run."
)
dependents: dict[type[MVTModule], list[type[MVTModule]]] = {
module: [] for module in runnable
module: [] for module in required
}
indegree = {module: 0 for module in runnable}
for module in runnable:
indegree = {module: 0 for module in required}
for module in required:
for dependency in module.dependencies:
if dependency not in indegree:
if dependency not in required:
continue
dependents[dependency].append(module)
indegree[module] += 1
@@ -450,7 +374,7 @@ class Command:
if indegree[dependent] == 0:
heappush(ready, (module_indexes[dependent], dependent))
if len(ordered) != len(runnable):
if len(ordered) != len(required):
cyclic_modules = sorted(
(module.__name__ for module, count in indegree.items() if count > 0)
)
-299
View File
@@ -1,299 +0,0 @@
# 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 contextlib
import json
import os
import re
import tempfile
from typing import Any, ClassVar, Dict, List, Tuple, Type, TypeVar
import yaml
from appdirs import user_config_dir, user_data_dir
from pydantic import ValidationError
from pydantic_settings import (
BaseSettings,
EnvSettingsSource,
PydanticBaseSettingsSource,
SettingsConfigDict,
YamlConfigSettingsSource,
)
PLUGIN_CONFIG_FOLDER_NAME = "plugins"
# Not "plugins": on macOS the configuration and data folders are the same
# directory, and that name already holds the settings files.
PLUGIN_DATA_FOLDER_NAME = "plugin-data"
PLUGIN_ENV_PREFIX = "MVT_PLUGIN_"
PLUGIN_NAME_PATTERN = re.compile(r"[a-z0-9][a-z0-9-]*")
PluginSettingsType = TypeVar("PluginSettingsType", bound="MVTPluginSettings")
class PluginConfigLoadError(Exception):
pass
def validate_plugin_name(plugin_name: Any) -> str:
"""
Check that a plugin name is safe to use in a file name and an env variable.
:param plugin_name: Name to validate.
:returns: The validated plugin name.
"""
if not isinstance(plugin_name, str):
raise TypeError(
f"Plugin name must be a string, not {type(plugin_name).__name__}"
)
if not PLUGIN_NAME_PATTERN.fullmatch(plugin_name):
raise ValueError(
f"Invalid plugin name {plugin_name!r}: plugin names must start with a "
"lowercase letter or a digit and may only contain lowercase letters, "
"digits and dashes"
)
return plugin_name
def plugin_config_folder() -> str:
"""
Return the folder where plugins store their configuration files.
The path is resolved on every call so it always reflects the current
environment.
"""
return os.path.join(user_config_dir("mvt"), PLUGIN_CONFIG_FOLDER_NAME)
def plugin_config_path(plugin_name: str) -> str:
"""
Return the path of the configuration file of a given plugin.
:param plugin_name: Name of the plugin.
"""
return os.path.join(
plugin_config_folder(), f"{validate_plugin_name(plugin_name)}.yaml"
)
def plugin_data_folder(plugin_name: str) -> str:
"""
Return the folder where a given plugin stores its data, creating it.
Plugins should keep whatever they persist, such as caches or downloaded
artifacts, in this folder. It is created with owner-only permissions. The
path is resolved on every call so it always reflects the current
environment. A plugin with a settings class calls
`MVTPluginSettings.data_folder()` instead, which passes `plugin_name` here.
:param plugin_name: Name of the plugin.
:returns: The path of the data folder of the plugin.
"""
# Validate the name before anything is created, so an unsafe name cannot
# leave a folder behind.
name = validate_plugin_name(plugin_name)
# makedirs() applies its mode only to the last folder of the path, so
# MVT's own data folder keeps the default permissions while the two
# plugin folders are private.
data_folder = os.path.join(user_data_dir("mvt"), PLUGIN_DATA_FOLDER_NAME)
os.makedirs(data_folder, mode=0o700, exist_ok=True)
folder = os.path.join(data_folder, name)
os.makedirs(folder, mode=0o700, exist_ok=True)
return folder
def plugin_env_prefix(plugin_name: str) -> str:
"""
Return the environment variable prefix used by a given plugin.
Dashes are replaced by underscores. Plugin names cannot contain underscores,
so two different plugin names never share an environment namespace.
:param plugin_name: Name of the plugin.
"""
name = validate_plugin_name(plugin_name).upper().replace("-", "_")
return f"{PLUGIN_ENV_PREFIX}{name}_"
def _settings_plugin_name(settings_cls: Type[BaseSettings]) -> str:
plugin_name = getattr(settings_cls, "plugin_name", None)
if plugin_name is None:
raise TypeError(
f"{settings_cls.__name__} must set a 'plugin_name' class attribute to "
"namespace its configuration file and environment variables"
)
return validate_plugin_name(plugin_name)
def _plugin_yaml_source(
settings_cls: Type[BaseSettings], config_path: str
) -> YamlConfigSettingsSource:
"""
Build the YAML settings source of a plugin, reporting unusable files.
A missing file is not an error, but a file which cannot be parsed or which
does not hold a mapping of setting names is reported with its path.
"""
try:
return YamlConfigSettingsSource(settings_cls, config_path)
except yaml.YAMLError as exc:
raise PluginConfigLoadError(
f"Invalid plugin configuration file {config_path}: {exc}"
) from exc
except (TypeError, ValueError) as exc:
raise PluginConfigLoadError(
f"Invalid plugin configuration file {config_path}: the file must "
"contain a mapping of setting names to values"
) from exc
class MVTPluginSettings(BaseSettings):
"""
Base class for plugin-namespaced settings.
Subclass with typed fields and set `plugin_name`. Values resolve from
constructor arguments, then environment variables (MVT_PLUGIN_<NAME>_*),
then the plugin's YAML file (~/.config/mvt/plugins/<name>.yaml), then field
defaults.
Plugins must not store their settings in MVT's own configuration file: MVT
rewrites it with the fields it knows about, dropping anything else.
`data_folder()` returns the folder the plugin keeps its data in.
"""
model_config = SettingsConfigDict(extra="ignore")
plugin_name: ClassVar[str]
def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs)
plugin_name = _settings_plugin_name(cls)
# Namespace the environment variables of this plugin. Each pydantic
# model gets its own configuration dictionary, so this does not leak
# into other plugins.
cls.model_config["env_prefix"] = plugin_env_prefix(plugin_name)
@classmethod
def settings_customise_sources(
cls,
settings_cls: Type[BaseSettings],
init_settings: PydanticBaseSettingsSource,
env_settings: PydanticBaseSettingsSource,
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> Tuple[PydanticBaseSettingsSource, ...]:
config_path = plugin_config_path(_settings_plugin_name(settings_cls))
yaml_source = _plugin_yaml_source(settings_cls, config_path)
# Explicit arguments take precedence over environment variables, which
# in turn take precedence over the configuration file.
return (init_settings, env_settings, yaml_source)
@classmethod
def load(cls: Type[PluginSettingsType]) -> PluginSettingsType:
"""
Load the settings of the plugin.
A missing configuration file is not an error: the settings then come
from the environment and from the field defaults.
"""
return cls()
@classmethod
def data_folder(cls) -> str:
"""
Return the data folder of the plugin, creating it.
The folder is the one plugin_data_folder() returns for `plugin_name`,
so a plugin with a settings class does not repeat its name.
"""
return plugin_data_folder(_settings_plugin_name(cls))
def _environment_values(self) -> Dict[str, Any]:
"""
Return the settings values currently supplied by the environment.
Values are validated by the model, so they can be compared with the
values held by this instance.
"""
settings_cls = type(self)
raw_values = EnvSettingsSource(settings_cls)()
names = [name for name in raw_values if name in settings_cls.model_fields]
if not names:
return {}
# Fall back on the current values for the fields the environment does
# not set, so that required fields do not fail validation here.
current_values = json.loads(self.model_dump_json())
try:
from_environment = settings_cls.model_validate(
{**current_values, **raw_values}
)
except ValidationError:
# An environment variable which the model cannot validate must not
# stop the other environment values from being recognised, or a
# credential would be written to the configuration file.
return self._environment_values_by_field(current_values, raw_values, names)
return {name: getattr(from_environment, name) for name in names}
def _environment_values_by_field(
self,
current_values: Dict[str, Any],
raw_values: Dict[str, Any],
names: List[str],
) -> Dict[str, Any]:
"""
Validate each environment value on its own, skipping unusable ones.
:param current_values: Serialized values held by this instance.
:param raw_values: Values supplied by the environment.
:param names: Names of the fields set by the environment.
"""
settings_cls = type(self)
values = {}
for name in names:
try:
from_environment = settings_cls.model_validate(
{**current_values, name: raw_values[name]}
)
except ValidationError:
continue
values[name] = getattr(from_environment, name)
return values
def save(self) -> None:
"""
Save the current settings to the configuration file of the plugin.
Only values which differ from the field defaults are persisted. Values
which come from the environment are not written to disk, so credentials
passed as environment variables stay out of the configuration file.
MVT's own configuration file is never modified.
"""
config_folder = plugin_config_folder()
if not os.path.isdir(config_folder):
os.makedirs(config_folder, mode=0o700, exist_ok=True)
values = json.loads(self.model_dump_json(exclude_defaults=True))
for name, environment_value in self._environment_values().items():
if name in values and getattr(self, name, None) == environment_value:
del values[name]
# Settings files can hold credentials, so write them through a private
# temporary file and move it in place. The file is then never partially
# written and never briefly readable by other users.
config_path = plugin_config_path(self.plugin_name)
descriptor, temporary_path = tempfile.mkstemp(
dir=config_folder, prefix=f".{self.plugin_name}-", suffix=".yaml"
)
try:
with os.fdopen(descriptor, "w") as config_file:
config_file.write(yaml.dump(values, default_flow_style=False))
os.replace(temporary_path, config_path)
except BaseException:
with contextlib.suppress(OSError):
os.unlink(temporary_path)
raise
+8 -1
View File
@@ -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,
]
+3 -1
View File
@@ -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_indexeddb import WebkitIndexedDB
from .webkit_localstorage import WebkitLocalStorage
from .webkit_safariviewservice import WebkitSafariViewService
FS_MODULES = [
FS_MODULES: list[type[MVTModule]] = [
CacheFiles,
Filesystem,
Netusage,
+3 -1
View File
@@ -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 .webkit_session_resource_log import WebkitSessionResourceLog
from .whatsapp import Whatsapp
from .whatsapp_contacts import WhatsappContacts
MIXED_MODULES = [
MIXED_MODULES: list[type[MVTModule]] = [
Calls,
ChromeFavicon,
ChromeHistory,
@@ -23,7 +23,7 @@ class TestDumpsysAccessibilityArtifact:
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 @@ class TestDumpsysAccessibilityArtifact:
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 @@ class TestDumpsysAccessibilityArtifact:
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")
@@ -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 @@ from ..utils import get_artifact
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")
+28 -1
View File
@@ -25,10 +25,37 @@ class TestDumpsysAppopsArtifact:
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()
@@ -57,18 +57,19 @@ class TestDumpsysBatteryDailyArtifact:
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 @@ class TestDumpsysBatteryDailyArtifact:
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()
@@ -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"
+24 -1
View File
@@ -53,9 +53,32 @@ Connection pool for /data/user/0/com.example/databases/current.db:
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]
@@ -21,12 +21,16 @@ class TestDumpsysPackageActivitiesArtifact:
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 @@ class TestDumpsysPackageActivitiesArtifact:
dpa.indicators = ind
assert len(dpa.alertstore.alerts) == 0
dpa.check_indicators()
assert len(dpa.alertstore.alerts) == 1
assert len(dpa.alertstore.alerts) == 2
+25 -36
View File
@@ -25,7 +25,11 @@ class TestDumpsysPackagesArtifact:
== "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 @@ class TestDumpsysPackagesArtifact:
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 [record["package_type"] for record in dpa.results] == [
"active",
"hidden_system",
]
assert runtime_permissions == [
{
"name": "android.permission.CAMERA",
"granted": True,
"type": "runtime",
}
]
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
@@ -22,6 +22,9 @@ class TestDumpsysPlatformCompatArtifact:
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()
@@ -19,17 +19,14 @@ class TestDumpsysReceiversArtifact:
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 @@ Receiver Resolver Table:
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 @@ Receiver Resolver Table:
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()
+9
View File
@@ -39,3 +39,12 @@ class TestGetPropArtifact:
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"},
]
+12 -1
View File
@@ -20,7 +20,7 @@ class TestProcessesArtifact:
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 @@ class TestProcessesArtifact:
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"
+2 -5
View File
@@ -128,8 +128,5 @@ class TestTombstoneCrashArtifact:
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"
+13
View File
@@ -13,6 +13,19 @@ from ..utils import get_android_androidqf, list_files
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.
+15
View File
@@ -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)
+14 -8
View File
@@ -54,10 +54,11 @@ class TestBugreportAnalysis:
== "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 @@ class TestBugreportAnalysis:
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
+4 -87
View File
@@ -157,7 +157,7 @@ class TestCommand:
assert not hasattr(cmd, "initialized")
assert "Circular module dependency detected" in caplog.text
def test_unavailable_dependency_only_skips_the_dependent_module(self, caplog):
def test_unavailable_dependency_warns_and_stops(self, caplog):
class UnavailableModule(RecordingModule):
pass
@@ -165,97 +165,14 @@ class TestCommand:
dependencies = (UnavailableModule,)
cmd = RecordingCommand()
cmd.modules = [DependentModule, IndependentModule, FirstModule]
with caplog.at_level(logging.WARNING):
cmd.run()
assert RecordingModule.run_order == ["IndependentModule", "FirstModule"]
assert cmd.initialized
assert "Module DependentModule will be SKIPPED" in caplog.text
assert "depends on module UnavailableModule" in caplog.text
def test_modules_depending_on_a_skipped_module_are_skipped_too(self, caplog):
class UnavailableModule(RecordingModule):
pass
class SkippedModule(RecordingModule):
dependencies = (UnavailableModule,)
class DependsOnSkippedModule(RecordingModule):
dependencies = (SkippedModule,)
class DependsOnTheChain(RecordingModule):
dependencies = (DependsOnSkippedModule,)
cmd = RecordingCommand()
cmd.modules = [
DependsOnTheChain,
DependsOnSkippedModule,
SkippedModule,
IndependentModule,
]
with caplog.at_level(logging.WARNING):
cmd.run()
assert RecordingModule.run_order == ["IndependentModule"]
skip_warnings = [
record.getMessage()
for record in caplog.records
if "will be SKIPPED" in record.getMessage()
]
assert len(skip_warnings) == 3
assert [warning.split()[1] for warning in skip_warnings] == [
"SkippedModule",
"DependsOnSkippedModule",
"DependsOnTheChain",
]
# Every warning names the root cause: the module missing a dependency
# and the dependency it is missing.
assert all("UnavailableModule" in warning for warning in skip_warnings)
assert all("module SkippedModule" in warning for warning in skip_warnings[1:])
def test_explicitly_selected_module_with_missing_dependency_runs_nothing(
self, caplog
):
class UnavailableModule(RecordingModule):
pass
class DependentModule(RecordingModule):
dependencies = (UnavailableModule,)
cmd = RecordingCommand(module_name="DependentModule")
cmd.modules = [DependentModule, IndependentModule]
cmd.modules = [DependentModule]
with caplog.at_level(logging.WARNING):
cmd.run()
assert RecordingModule.run_order == []
assert "Module DependentModule will be SKIPPED" in caplog.text
assert "No modules will be run" in caplog.text
# Nothing else was selected, so the warning must not promise that the
# analysis continues right before saying that it does not.
assert "The rest of the analysis will still run" not in caplog.text
def test_unaffected_dependency_chains_keep_their_order(self, caplog):
class UnavailableModule(RecordingModule):
pass
class SkippedModule(RecordingModule):
dependencies = (UnavailableModule,)
cmd = RecordingCommand()
cmd.modules = [ThirdModule, SkippedModule, SecondModule, FirstModule]
with caplog.at_level(logging.WARNING):
cmd.run()
assert RecordingModule.run_order == [
"FirstModule",
"SecondModule",
"ThirdModule",
]
assert not hasattr(cmd, "initialized")
assert "depends on unavailable module UnavailableModule" in caplog.text
def test_custom_modules_are_filtered_before_ordering(self):
cmd = RecordingCommand()
-391
View File
@@ -1,391 +0,0 @@
# 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 os
import stat
import sys
from typing import Optional
import pytest
import yaml
from mvt.common.plugin_config import (
MVTPluginSettings,
PluginConfigLoadError,
plugin_config_folder,
plugin_config_path,
plugin_data_folder,
plugin_env_prefix,
)
class ExamplePluginSettings(MVTPluginSettings):
plugin_name = "example-plugin"
API_KEY: Optional[str] = None
CACHE_FOLDER: str = "cache"
MAX_RESULTS: int = 25
class OtherPluginSettings(MVTPluginSettings):
plugin_name = "other-plugin"
API_KEY: Optional[str] = None
@pytest.fixture
def config_folder(tmp_path, monkeypatch):
monkeypatch.setattr(
"mvt.common.plugin_config.user_config_dir",
lambda *args, **kwargs: str(tmp_path),
)
return tmp_path
@pytest.fixture
def data_folder(tmp_path, monkeypatch):
folder = tmp_path / "data"
monkeypatch.setattr(
"mvt.common.plugin_config.user_data_dir",
lambda *args, **kwargs: str(folder),
)
return folder
def _write_plugin_file(plugin_name, values):
config_path = plugin_config_path(plugin_name)
os.makedirs(os.path.dirname(config_path), exist_ok=True)
content = values if isinstance(values, str) else yaml.dump(values)
with open(config_path, "w") as config_file:
config_file.write(content)
return config_path
def test_plugin_paths_and_prefixes_are_namespaced(config_folder):
assert plugin_config_folder() == str(config_folder / "plugins")
assert plugin_config_path("example-plugin") == str(
config_folder / "plugins" / "example-plugin.yaml"
)
assert plugin_env_prefix("example-plugin") == "MVT_PLUGIN_EXAMPLE_PLUGIN_"
assert plugin_env_prefix("other-plugin") == "MVT_PLUGIN_OTHER_PLUGIN_"
def test_defaults_are_used_without_file_or_environment(config_folder):
settings = ExamplePluginSettings.load()
assert settings.API_KEY is None
assert settings.CACHE_FOLDER == "cache"
assert settings.MAX_RESULTS == 25
assert not os.path.exists(plugin_config_path("example-plugin"))
def test_values_are_loaded_from_the_plugin_file(config_folder):
_write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
settings = ExamplePluginSettings.load()
assert settings.API_KEY == "from-file"
assert settings.MAX_RESULTS == 5
assert settings.CACHE_FOLDER == "cache"
def test_environment_overrides_the_plugin_file(config_folder, monkeypatch):
_write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
settings = ExamplePluginSettings.load()
assert settings.API_KEY == "from-environment"
assert settings.MAX_RESULTS == 5
def test_arguments_override_the_environment_and_the_plugin_file(
config_folder, monkeypatch
):
_write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "10")
settings = ExamplePluginSettings(API_KEY="from-argument")
assert settings.API_KEY == "from-argument"
assert settings.MAX_RESULTS == 10
def test_save_and_load_round_trip(config_folder):
settings = ExamplePluginSettings.load()
settings.API_KEY = "saved-key"
settings.MAX_RESULTS = 100
settings.save()
config_path = plugin_config_path("example-plugin")
assert os.path.isfile(config_path)
with open(config_path) as config_file:
assert yaml.safe_load(config_file) == {
"API_KEY": "saved-key",
"MAX_RESULTS": 100,
}
reloaded = ExamplePluginSettings.load()
assert reloaded.API_KEY == "saved-key"
assert reloaded.MAX_RESULTS == 100
assert reloaded.CACHE_FOLDER == "cache"
@pytest.mark.skipif(
sys.platform == "win32", reason="POSIX file permissions are not available"
)
def test_saved_file_is_only_readable_by_the_user(config_folder):
settings = ExamplePluginSettings.load()
settings.API_KEY = "saved-key"
settings.save()
config_path = plugin_config_path("example-plugin")
assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
folder_mode = stat.S_IMODE(os.stat(plugin_config_folder()).st_mode)
assert folder_mode & 0o077 == 0
@pytest.mark.skipif(
sys.platform == "win32", reason="POSIX file permissions are not available"
)
def test_save_restricts_the_permissions_of_an_existing_file(config_folder):
config_path = _write_plugin_file("example-plugin", {"API_KEY": "from-file"})
os.chmod(config_path, 0o644)
settings = ExamplePluginSettings.load()
settings.MAX_RESULTS = 100
settings.save()
assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
assert os.listdir(plugin_config_folder()) == ["example-plugin.yaml"]
def test_save_only_persists_non_default_values(config_folder):
settings = ExamplePluginSettings.load()
settings.CACHE_FOLDER = "another-cache"
settings.save()
with open(plugin_config_path("example-plugin")) as config_file:
assert yaml.safe_load(config_file) == {"CACHE_FOLDER": "another-cache"}
def test_save_does_not_persist_values_coming_from_the_environment(
config_folder, monkeypatch
):
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
settings = ExamplePluginSettings.load()
assert settings.API_KEY == "environment-secret"
settings.MAX_RESULTS = 100
settings.save()
with open(plugin_config_path("example-plugin")) as config_file:
assert yaml.safe_load(config_file) == {"MAX_RESULTS": 100}
def test_an_invalid_environment_variable_still_protects_the_other_values(
config_folder, monkeypatch
):
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "not-an-int")
settings = ExamplePluginSettings(MAX_RESULTS=100)
assert settings.API_KEY == "environment-secret"
settings.save()
with open(plugin_config_path("example-plugin")) as config_file:
saved_values = yaml.safe_load(config_file)
assert saved_values == {"MAX_RESULTS": 100}
assert "API_KEY" not in saved_values
def test_save_persists_values_which_differ_from_the_environment(
config_folder, monkeypatch
):
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
settings = ExamplePluginSettings.load()
settings.API_KEY = "chosen-key"
settings.save()
with open(plugin_config_path("example-plugin")) as config_file:
assert yaml.safe_load(config_file) == {"API_KEY": "chosen-key"}
def test_save_does_not_write_the_mvt_configuration_file(config_folder):
settings = ExamplePluginSettings.load()
settings.API_KEY = "saved-key"
settings.save()
assert os.listdir(config_folder) == ["plugins"]
def test_unknown_keys_in_the_plugin_file_are_ignored(config_folder):
_write_plugin_file(
"example-plugin",
{"API_KEY": "from-file", "UNKNOWN_SETTING": "ignored"},
)
settings = ExamplePluginSettings.load()
assert settings.API_KEY == "from-file"
assert not hasattr(settings, "UNKNOWN_SETTING")
def test_unparsable_plugin_file_is_reported_with_its_path(config_folder):
config_path = _write_plugin_file("example-plugin", "API_KEY: [unclosed\n")
with pytest.raises(PluginConfigLoadError) as raised:
ExamplePluginSettings.load()
assert config_path in str(raised.value)
def test_plugin_file_which_is_not_a_mapping_is_reported_with_its_path(config_folder):
config_path = _write_plugin_file("example-plugin", "- one\n- two\n")
with pytest.raises(PluginConfigLoadError) as raised:
ExamplePluginSettings.load()
assert config_path in str(raised.value)
assert "mapping of setting names" in str(raised.value)
def test_plugins_do_not_interfere_with_each_other(config_folder, monkeypatch):
_write_plugin_file("other-plugin", {"API_KEY": "other-file-key"})
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "example-environment-key")
example_settings = ExamplePluginSettings.load()
other_settings = OtherPluginSettings.load()
assert example_settings.API_KEY == "example-environment-key"
assert other_settings.API_KEY == "other-file-key"
example_settings.MAX_RESULTS = 100
example_settings.save()
assert sorted(os.listdir(plugin_config_folder())) == [
"example-plugin.yaml",
"other-plugin.yaml",
]
with open(plugin_config_path("other-plugin")) as config_file:
assert yaml.safe_load(config_file) == {"API_KEY": "other-file-key"}
def test_subclass_without_plugin_name_is_rejected():
with pytest.raises(TypeError, match="plugin_name"):
class MissingNameSettings(MVTPluginSettings):
API_KEY: Optional[str] = None
def test_subclass_with_invalid_plugin_name_is_rejected():
with pytest.raises(ValueError, match="Invalid plugin name"):
class InvalidNameSettings(MVTPluginSettings):
plugin_name = "Bad/Name"
def test_underscores_are_not_allowed_in_plugin_names():
# Underscores are replaced by dashes in the environment prefix, so allowing
# both would let two plugin names share one environment namespace.
with pytest.raises(ValueError, match="Invalid plugin name"):
class UnderscoreNameSettings(MVTPluginSettings):
plugin_name = "under_score"
with pytest.raises(ValueError, match="Invalid plugin name"):
plugin_config_path("under_score")
with pytest.raises(ValueError, match="Invalid plugin name"):
plugin_env_prefix("under_score")
@pytest.mark.parametrize(
"plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
)
def test_unsafe_plugin_names_have_no_configuration_path(plugin_name):
with pytest.raises(ValueError, match="Invalid plugin name"):
plugin_config_path(plugin_name)
def test_data_folder_is_namespaced_and_created(data_folder):
folder = plugin_data_folder("example-plugin")
assert folder == str(data_folder / "plugin-data" / "example-plugin")
assert os.path.isdir(folder)
def test_data_folder_can_be_requested_repeatedly(data_folder):
folder = plugin_data_folder("example-plugin")
with open(os.path.join(folder, "kept.json"), "w") as data_file:
data_file.write("{}")
assert plugin_data_folder("example-plugin") == folder
assert os.listdir(folder) == ["kept.json"]
@pytest.mark.skipif(
sys.platform == "win32", reason="POSIX file permissions are not available"
)
def test_data_folder_is_only_accessible_by_the_user(data_folder):
folder = plugin_data_folder("example-plugin")
assert stat.S_IMODE(os.stat(folder).st_mode) & 0o077 == 0
parent_mode = stat.S_IMODE(os.stat(os.path.dirname(folder)).st_mode)
assert parent_mode & 0o077 == 0
def test_plugins_get_their_own_data_folder(data_folder):
example_folder = plugin_data_folder("example-plugin")
other_folder = plugin_data_folder("other-plugin")
assert example_folder != other_folder
assert sorted(os.listdir(data_folder / "plugin-data")) == [
"example-plugin",
"other-plugin",
]
def test_data_folder_does_not_touch_the_configuration_folder(
config_folder, data_folder
):
plugin_data_folder("example-plugin")
assert not os.path.exists(plugin_config_folder())
@pytest.mark.parametrize(
"plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
)
def test_unsafe_plugin_names_have_no_data_folder(data_folder, plugin_name):
with pytest.raises(ValueError, match="Invalid plugin name"):
plugin_data_folder(plugin_name)
assert not os.path.exists(data_folder)
def test_settings_class_knows_its_data_folder(data_folder):
folder = ExamplePluginSettings.data_folder()
assert folder == plugin_data_folder("example-plugin")
assert os.path.isdir(folder)
assert OtherPluginSettings.data_folder() != folder
def test_settings_instance_uses_the_same_data_folder(config_folder, data_folder):
settings = ExamplePluginSettings.load()
assert settings.data_folder() == ExamplePluginSettings.data_folder()
def test_subclass_without_its_own_name_shares_the_data_folder(data_folder):
class InheritingSettings(ExamplePluginSettings):
pass
assert InheritingSettings.data_folder() == ExamplePluginSettings.data_folder()