CalvinBackup/mvt
1
0
Fork 0
mirror of https://github.com/mvt-project/mvt.git synced 2026-08-17 00:20:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

Alert on AndroidQF trusted ADB keys (#860)

Browse Source
This commit is contained in:
besendorf
2026-08-05 17:36:49 +02:00
committed by GitHub
parent 53fb12aee8
commit 8617e0bf54
4 changed files with 327 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mvt/android/cmd_check_androidqf.py
+42
View File
@@ -3,6 +3,7 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import json
import logging
import os
import shutil
@@ -94,6 +95,47 @@ class CmdAndroidCheckAndroidQF(Command):
self.__zip = zipfile.ZipFile(self.target_path)
self.__files = self.__zip.namelist()
self._load_acquisition_context()
def _load_acquisition_context(self) -> None:
"""Pass AndroidQF acquisition metadata to nested commands and modules."""
context = {}
metadata_files = [
file_path
for file_path in self.__files
if file_path.replace("\\", "/").rsplit("/", 1)[-1] == "acquisition.json"
]
for file_path in metadata_files:
try:
metadata = json.loads(self._get_file_content(file_path))
if isinstance(metadata, dict):
context["started"] = metadata.get("started")
context["adb_host_public_key"] = metadata.get("adb_host_public_key")
break
except (json.JSONDecodeError, OSError, TypeError, UnicodeDecodeError):
self.log.warning(
'Unable to read AndroidQF acquisition metadata "%s"', file_path
)
if not context.get("adb_host_public_key"):
key_files = [
file_path
for file_path in self.__files
if file_path.replace("\\", "/").rsplit("/", 1)[-1] == "adb_host_key.pub"
]
for file_path in key_files:
try:
context["adb_host_public_key"] = self._get_file_content(
file_path
).decode("utf-8")
break
except (OSError, UnicodeDecodeError):
self.log.warning(
'Unable to read AndroidQF ADB host key "%s"', file_path
)
self.module_options["androidqf_acquisition"] = context
def module_init(self, module: AndroidQFModule) -> None: # type: ignore[override]
if self.__format == "zip" and self.__zip:
module.from_zip(self.__zip, self.__files)
src/mvt/android/modules/bugreport/dumpsys_adb_state.py
+123
View File
@@ -3,11 +3,15 @@
# 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
import binascii
import datetime
import logging
from typing import Optional
from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
from mvt.common.module_types import ModuleResults
from mvt.common.utils import convert_datetime_to_iso
from .base import BugReportModule
@@ -53,3 +57,122 @@ class DumpsysADBState(DumpsysADBArtifact, BugReportModule):
"Identified a total of %d trusted ADB keys",
len(self.results[0].get("user_keys", [])),
)
@staticmethod
def _key_material(public_key: object) -> str:
if isinstance(public_key, bytes):
public_key = public_key.decode("utf-8", errors="replace")
if not isinstance(public_key, str):
return ""
return public_key.strip().split(" ", 1)[0]
@staticmethod
def _is_valid_key(public_key: str) -> bool:
if not public_key:
return False
try:
return bool(base64.b64decode(public_key, validate=True))
except (binascii.Error, ValueError):
return False
@staticmethod
def _parse_acquisition_time(value: object) -> Optional[datetime.datetime]:
if not isinstance(value, str):
return None
try:
timestamp = datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return None
if not timestamp.tzinfo:
timestamp = timestamp.replace(tzinfo=datetime.timezone.utc)
return timestamp.astimezone(datetime.timezone.utc)
@staticmethod
def _parse_last_connected(value: object) -> Optional[datetime.datetime]:
try:
return datetime.datetime.fromtimestamp(
int(str(value)) / 1000,
tz=datetime.timezone.utc,
)
except (OSError, OverflowError, TypeError, ValueError):
return None
def _trusted_keys(self) -> list[dict]:
"""Return unique trusted keys, preferring keystore connection metadata."""
trusted_keys = []
seen = set()
for result in self.results:
keystore = result.get("keystore", [])
candidates = keystore if isinstance(keystore, list) else []
candidates = [*candidates, *result.get("user_keys", [])]
for candidate in candidates:
if not isinstance(candidate, dict):
continue
key = self._key_material(candidate.get("key"))
identity = key or repr(candidate)
if identity in seen:
continue
seen.add(identity)
trusted_keys.append(candidate)
return trusted_keys
def check_indicators(self) -> None:
if "androidqf_acquisition" not in self.module_options:
return super().check_indicators()
context = self.module_options.get("androidqf_acquisition")
if not isinstance(context, dict):
context = {}
acquisition_key = self._key_material(context.get("adb_host_public_key"))
if acquisition_key and not self._is_valid_key(acquisition_key):
acquisition_key = ""
acquisition_time = self._parse_acquisition_time(context.get("started"))
cutoff = (
acquisition_time - datetime.timedelta(days=1) if acquisition_time else None
)
for trusted_key in self._trusted_keys():
key = self._key_material(trusted_key.get("key"))
fingerprint = trusted_key.get("fingerprint") or "<unknown key>"
user = trusted_key.get("user") or "unknown user"
description = f"{fingerprint} ({user})"
last_connected = self._parse_last_connected(
trusted_key.get("last_connected")
)
event_time = (
convert_datetime_to_iso(last_connected) if last_connected else ""
)
if not self._is_valid_key(key):
self.alertstore.low(
f"Found an invalid trusted ADB host key: {description}",
event_time,
trusted_key,
)
continue
if not acquisition_key:
self.alertstore.low(
"Found a trusted ADB host key, but the AndroidQF acquisition "
f"does not include its host key: {description}",
event_time,
trusted_key,
)
continue
if key != acquisition_key:
self.alertstore.low(
"Found a trusted ADB host key different from the AndroidQF "
f"acquisition host: {description}",
event_time,
trusted_key,
)
continue
if cutoff and last_connected and last_connected <= cutoff:
self.alertstore.info(
"Found a trusted ADB host key last connected at least one day "
f"before the AndroidQF acquisition: {description}",
event_time,
trusted_key,
)
tests/android/test_artifact_dumpsys_adb.py
+118
View File
@@ -4,6 +4,8 @@
# https://license.mvt.re/1.1/
from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState
from mvt.common.alerts import AlertLevel
from ..utils import get_artifact
@@ -114,3 +116,119 @@ class TestDumpsysADBArtifact:
assert key_store_entry["user"] == "user@laptop"
assert key_store_entry["fingerprint"] == expected_fingerprint
assert key_store_entry["last_connected"] == "1628501829898"
class TestDumpsysADBStateAlerts:
def test_no_androidqf_context_preserves_existing_behavior(self):
module = DumpsysADBState(
results=[
{
"user_keys": [
{
"key": b"QUJDRA==",
"user": "host@example",
"fingerprint": "fingerprint",
}
]
}
]
)
module.check_indicators()
assert module.alertstore.alerts == []
def test_androidqf_trusted_keys_create_expected_alerts(self):
module = DumpsysADBState(
module_options={
"androidqf_acquisition": {
"started": "2025-06-20T18:00:00Z",
"adb_host_public_key": "QUJDRA== acquisition@host",
}
},
results=[
{
"user_keys": [
{
"key": b"QUJDRA==",
"user": "acquisition@host",
"fingerprint": "acquisition-fingerprint",
},
{
"key": b"RUZHSA==",
"user": "other@host",
"fingerprint": "other-fingerprint",
},
{
"key": b"not-base64",
"user": "invalid@host",
"fingerprint": "",
},
],
"keystore": [
{
"key": b"QUJDRA==",
"user": "acquisition@host",
"fingerprint": "acquisition-fingerprint",
"last_connected": "1750266000000",
}
],
}
],
)
module.check_indicators()
assert [alert.level for alert in module.alertstore.alerts] == [
AlertLevel.INFORMATIONAL,
AlertLevel.LOW,
AlertLevel.LOW,
]
informational, different, invalid = module.alertstore.alerts
assert "at least one day before" in informational.message
assert informational.event_time == "2025-06-18 17:00:00.000000"
assert "different from the AndroidQF acquisition host" in different.message
assert "invalid trusted ADB host key" in invalid.message
def test_missing_androidqf_host_key_creates_low_alert(self):
trusted_key = {
"key": b"QUJDRA==",
"user": "host@example",
"fingerprint": "fingerprint",
}
module = DumpsysADBState(
module_options={"androidqf_acquisition": {}},
results=[{"user_keys": [trusted_key]}],
)
module.check_indicators()
assert len(module.alertstore.alerts) == 1
assert module.alertstore.alerts[0].level == AlertLevel.LOW
assert "does not include its host key" in module.alertstore.alerts[0].message
def test_recent_acquisition_host_key_does_not_create_alert(self):
module = DumpsysADBState(
module_options={
"androidqf_acquisition": {
"started": "2025-06-20T18:00:00Z",
"adb_host_public_key": "QUJDRA== acquisition@host",
}
},
results=[
{
"keystore": [
{
"key": b"QUJDRA==",
"user": "acquisition@host",
"fingerprint": "fingerprint",
"last_connected": "1750438800000",
}
]
}
],
)
module.check_indicators()
assert module.alertstore.alerts == []
tests/test_check_android_androidqf.py
+44
View File
@@ -3,6 +3,7 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import json
import logging
import os
import shutil
@@ -32,6 +33,49 @@ class TestCheckAndroidqfCommand:
result = runner.invoke(check_androidqf, [path])
assert result.exit_code == 0
def test_acquisition_context_is_passed_to_bugreport(self, tmp_path, mocker):
data_path = tmp_path / "androidqf"
data_path.mkdir()
(data_path / "acquisition.json").write_text(
json.dumps(
{
"started": "2025-06-20T18:00:00Z",
"adb_host_public_key": "QUJDRA== acquisition@host",
}
)
)
with zipfile.ZipFile(data_path / "bugreport.zip", "w"):
pass
nested_command = mocker.patch(
"mvt.android.cmd_check_androidqf.CmdAndroidCheckBugreport"
)
nested_command.return_value.timeline = []
nested_command.return_value.alertstore.alerts = []
command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
command.init()
assert command.run_bugreport_cmd() is True
assert nested_command.call_args.kwargs["module_options"][
"androidqf_acquisition"
] == {
"started": "2025-06-20T18:00:00Z",
"adb_host_public_key": "QUJDRA== acquisition@host",
}
def test_acquisition_context_falls_back_to_public_key_file(self, tmp_path):
data_path = tmp_path / "androidqf"
data_path.mkdir()
(data_path / "adb_host_key.pub").write_text("QUJDRA== acquisition@host\n")
command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
command.init()
assert command.module_options["androidqf_acquisition"] == {
"adb_host_public_key": "QUJDRA== acquisition@host\n"
}
def test_check_encrypted_backup_prompt_valid(self, mocker):
"""Prompt for password on CLI"""
prompt_mock = mocker.patch(