diff --git a/src/mvt/android/cmd_check_androidqf.py b/src/mvt/android/cmd_check_androidqf.py index 319b614..6f1089c 100644 --- a/src/mvt/android/cmd_check_androidqf.py +++ b/src/mvt/android/cmd_check_androidqf.py @@ -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) diff --git a/src/mvt/android/modules/bugreport/dumpsys_adb_state.py b/src/mvt/android/modules/bugreport/dumpsys_adb_state.py index 506af30..06b29d5 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_adb_state.py +++ b/src/mvt/android/modules/bugreport/dumpsys_adb_state.py @@ -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 "" + 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, + ) diff --git a/tests/android/test_artifact_dumpsys_adb.py b/tests/android/test_artifact_dumpsys_adb.py index 47dc3df..f5574a6 100644 --- a/tests/android/test_artifact_dumpsys_adb.py +++ b/tests/android/test_artifact_dumpsys_adb.py @@ -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 == [] diff --git a/tests/test_check_android_androidqf.py b/tests/test_check_android_androidqf.py index 9b271be..a8a6a5a 100644 --- a/tests/test_check_android_androidqf.py +++ b/tests/test_check_android_androidqf.py @@ -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(