diff --git a/docs/install.md b/docs/install.md index c024338c..7c5ee6ee 100644 --- a/docs/install.md +++ b/docs/install.md @@ -1,6 +1,6 @@ # Installation -Before proceeding, please note that MVT requires Python 3.6+ to run. While it should be available on most operating systems, please make sure of that before proceeding. +Before proceeding, please note that MVT requires Python 3.10 or newer to run. While it should be available on most operating systems, please make sure of that before proceeding. ## Dependencies on Linux diff --git a/docs/ios/records.md b/docs/ios/records.md index 856d0db9..495db368 100644 --- a/docs/ios/records.md +++ b/docs/ios/records.md @@ -204,6 +204,10 @@ This JSON file is created by mvt-ios' `Manifest` module. The module extracts rec If indicators are provided through the command-line, they are checked against the original relative path in case. In some cases, there might be records of files created containing a domain name in their name, for example in the case of browser cache folders. Any matches are stored in *manifest_detected.json*. +An incomplete backup still lists in its manifest the files it failed to acquire. Those records carry a `"missing": true` field, and the module reports how many of them it found. Use it to tell a module which returned nothing because the artifact was not acquired apart from one which found nothing on a device that never had it. + +Files must be stored at their expected paths inside the backup folder. If the module cannot finish listing the backup files, it logs a warning and skips the missing-file check; the manifest metadata is still extracted. + --- ### `os_analytics_ad_daily.json` diff --git a/src/mvt/android/artifacts/dumpsys_accessibility.py b/src/mvt/android/artifacts/dumpsys_accessibility.py index da46043c..b6fe0e25 100644 --- a/src/mvt/android/artifacts/dumpsys_accessibility.py +++ b/src/mvt/android/artifacts/dumpsys_accessibility.py @@ -10,8 +10,45 @@ from .artifact import AndroidArtifact class DumpsysAccessibilityArtifact(AndroidArtifact): + # One list for both record shapes — a service record and a count-only + # record must stay the same shape. + _FIELDS = ( + "user_id", + "component", + "package_name", + "service_name", + "installed", + "enabled", + "binding", + "bound", + "crashed", + "accessibility_tool", + "installed_service_count", + "unnamed_service_count", + ) + def check_indicators(self) -> None: for result in self.results: + # A stated count the dump does not back with component names is a + # coverage statement, not a service: low, but not silent. + if not result.get("component"): + stated = result["installed_service_count"] + unnamed = result["unnamed_service_count"] + if unnamed == stated: + detail = "does not list their component names" + else: + detail = ( + f"lists the component names of only {stated - unnamed} " + f"of them ({unnamed} unnamed)" + ) + self.alertstore.low( + f"The accessibility dump states {stated} installed " + f"service(s) for user {result['user_id']} but {detail}", + "", + result, + ) + continue + if self.indicators: ioc_match = self.indicators.check_app_id(result["package_name"]) if ioc_match: @@ -20,11 +57,22 @@ class DumpsysAccessibilityArtifact(AndroidArtifact): ) continue - self.alertstore.medium( - f'Found accessibility service: "{result["component"]}"', - "", - result, + # Installed is not enabled. A service can sit installed for years + # without ever being switched on, and that is the difference this + # channel is read for — an alert that says only "found" makes every + # device look equally exposed. A service the dump says is switched + # OFF is reported LOW, so it still reaches the analyst without + # competing with one that is actually running; a dump that does not + # state the enabled state stays MEDIUM, because "not stated" is not + # "not enabled". + message = ( + f'Found accessibility service: "{result["component"]}" ' + f"({self._describe_state(result)})" ) + if result.get("enabled") is False and not result.get("bound"): + self.alertstore.low(message, "", result) + else: + self.alertstore.medium(message, "", result) def parse(self, content: str) -> None: """ @@ -36,6 +84,13 @@ class DumpsysAccessibilityArtifact(AndroidArtifact): self.results: list[dict[str, Any]] = [] services: dict[tuple[int | None, str], dict] = {} + # Which state sections the dump printed, per user: one user's printed + # `enabled services` says nothing about another user's. + seen_states: dict[int | None, set[str]] = {} + # `installedServiceCount=N` from the user's `attributes:{…}` line is on + # most builds the only statement about installed services in the dump: + # few print the `installed services: {…}` block. + installed_counts: dict[int | None, int] = {} user_id: int | None = None state: str | None = None @@ -44,6 +99,10 @@ class DumpsysAccessibilityArtifact(AndroidArtifact): if user_match: user_id = int(user_match.group(1)) + count_match = re.search(r"installedServiceCount=(\d+)", line) + if count_match: + installed_counts[user_id] = int(count_match.group(1)) + stripped = line.strip() state_match = re.match( r"(?i)(installed|enabled|binding|bound|crashed) services\s*:\s*\{(.*)", @@ -51,6 +110,7 @@ class DumpsysAccessibilityArtifact(AndroidArtifact): ) if state_match: state = state_match.group(1).lower() + seen_states.setdefault(user_id, set()).add(self._state_field(state)) inline = state_match.group(2) for component in re.findall( r"\{?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?\}?", inline @@ -79,24 +139,65 @@ class DumpsysAccessibilityArtifact(AndroidArtifact): service[self._state_field(state)] = True service["accessibility_tool"] = "(A11yTool)" in stripped + # A section that was never printed is NOT the same as one printed + # empty: the first says nothing, the second says nothing is enabled. + # Defaulting every flag to False would turn "not stated" into "not + # enabled". Flags for sections this dump never printed stay None. + named: dict[int | None, int] = {} + for (service_user, _component), service in services.items(): + printed = seen_states.get(service_user, set()) + for state in ("installed", "enabled", "binding", "bound", "crashed"): + if self._state_field(state) not in printed: + service[self._state_field(state)] = None + service["installed_service_count"] = installed_counts.get(service_user) + named[service_user] = named.get(service_user, 0) + 1 + self.results.extend(services.values()) + # A stated count the named services do not add up to would leave no + # trace of the rest: the module would log "a total of 0" about a dump + # that said five, or "a total of 1" about one that said two. + for count_user, count in installed_counts.items(): + unnamed = count - named.get(count_user, 0) + if unnamed > 0: + self.results.append(self._new_unlisted(count_user, count, unnamed)) + + @staticmethod + def _describe_state(result: dict) -> str: + if result.get("bound"): + return "enabled and bound" + if result.get("enabled"): + return "enabled" + if result.get("enabled") is None: + return "installed, enabled state not stated" + return "installed, not enabled" + @staticmethod def _state_field(state: str) -> str: return {"binding": "binding", "bound": "bound"}.get(state, state) + @staticmethod + def _new_unlisted(user_id: int | None, count: int, unnamed: int) -> dict: + """The dump's own count for a user, and how many of those services it + did not name. + + Every other field stays unknown: the dump named no service to carry it. + """ + record: dict[str, Any] = dict.fromkeys(DumpsysAccessibilityArtifact._FIELDS) + record["user_id"] = user_id + record["installed_service_count"] = count + record["unnamed_service_count"] = unnamed + return record + @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, - } + record: dict[str, Any] = dict.fromkeys( + DumpsysAccessibilityArtifact._FIELDS, False + ) + record["user_id"] = user_id + record["component"] = component + record["package_name"], record["service_name"] = component.split("/", 1) + # Filled in after parsing: the count is per user. + record["installed_service_count"] = None + record["unnamed_service_count"] = None + return record diff --git a/src/mvt/android/artifacts/dumpsys_adb.py b/src/mvt/android/artifacts/dumpsys_adb.py index cbd80e21..c57cf8d6 100644 --- a/src/mvt/android/artifacts/dumpsys_adb.py +++ b/src/mvt/android/artifacts/dumpsys_adb.py @@ -29,11 +29,21 @@ class DumpsysADBArtifact(AndroidArtifact): stack = [res] cur_indent = 0 in_multiline = False - for line in dump_data.strip(b"\n").split(b"\n"): + for line in dump_data.strip(b"\r\n").split(b"\n"): + line = line.removesuffix(b"\r") # Track the level of indentation indent = len(line) - len(line.lstrip()) if indent < cur_indent: # If the current line is less indented than the previous one, back out + if len(stack) <= 1: + # Dedenting below the outermost level means this is not the + # well-formed block the parser expects. Stop here rather than + # raise IndexError out of the module on the next line. + self.log.error( + "Unexpected indentation in ADB manager state, " + "stopping the parse of this section" + ) + break stack.pop() cur_indent = indent else: @@ -63,6 +73,12 @@ class DumpsysADBArtifact(AndroidArtifact): current_dict = stack[-1] if key == "}": + if len(stack) <= 1: + self.log.error( + "Unbalanced closing brace in ADB manager state, " + "stopping the parse of this section" + ) + break stack.pop() continue @@ -161,6 +177,26 @@ class DumpsysADBArtifact(AndroidArtifact): f"'{user_key['fingerprint']}'" ) + @staticmethod + def _find_state_end(content: bytes, open_brace: int) -> int: + """Find the unindented line closing the ADB manager state, or -1. + + Braces inside key comments or embedded keystore data are values, while + nested structural closing braces are indented. + """ + line_start = open_brace + while line_start < len(content): + line_end = content.find(b"\n", line_start) + if line_end == -1: + line_end = len(content) + line = content[line_start:line_end].removesuffix(b"\r") + if line_start > open_brace and line == b"}": + return line_start + if line.startswith((b"---------", b"DUMP OF SERVICE ")): + break + line_start = line_end + 1 + return -1 + def parse(self, content: bytes) -> None: """ Parse the Dumpsys ADB section @@ -180,12 +216,16 @@ class DumpsysADBArtifact(AndroidArtifact): self.log.error("Unable to find ADB manager state in dumpsys output") return - end_of_json = content.rfind(b"}\n") - if end_of_json == -1 or end_of_json <= start_of_json: - self.log.error("Unable to find complete ADB manager state in dumpsys output") + end_of_json = self._find_state_end(content, start_of_json + 1) + if end_of_json == -1: + self.log.error( + "Unable to find complete ADB manager state in dumpsys output" + ) return - json_content = content[start_of_json + 2 : end_of_json - 2].rstrip() + # The brace that opens the state and the one that closes it are not part + # of the indented body. + json_content = content[start_of_json + 2 : end_of_json].rstrip() parsed = self.indented_dump_parser(json_content) if parsed.get("debugging_manager") is None: diff --git a/src/mvt/android/artifacts/dumpsys_battery_daily.py b/src/mvt/android/artifacts/dumpsys_battery_daily.py index 2ccb6b70..905b4e18 100644 --- a/src/mvt/android/artifacts/dumpsys_battery_daily.py +++ b/src/mvt/android/artifacts/dumpsys_battery_daily.py @@ -73,8 +73,12 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact): continue line = line.strip().replace("Update ", "") - package_name, vers = line.split(" ", 1) - vers_raw = vers.split("=", 1)[1] + # A truncated or vendor-specific line must not abort the parse and + # lose every record after it. + package_name, _, vers = line.partition(" ") + vers_raw = vers.partition("=")[2] + if not package_name or not vers_raw: + continue try: version_code: int | str = int(vers_raw) except ValueError: diff --git a/src/mvt/android/cmd_check_bugreport.py b/src/mvt/android/cmd_check_bugreport.py index 036160d8..34dddbde 100644 --- a/src/mvt/android/cmd_check_bugreport.py +++ b/src/mvt/android/cmd_check_bugreport.py @@ -3,14 +3,19 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +import fnmatch +import io import logging import os from pathlib import Path from typing import List, Optional -from zipfile import ZipFile +from zipfile import BadZipFile, ZipFile from mvt.android.artifacts.getprop import GetProp -from mvt.android.modules.bugreport.base import BugReportModule +from mvt.android.modules.bugreport.base import ( + DUMPSTATE_ENTRY_POINTS, + BugReportModule, +) from mvt.common.command import Command from mvt.common.indicators import Indicators from mvt.common.module import MVTModule @@ -88,6 +93,48 @@ class CmdAndroidCheckBugreport(Command): for file_name in self.__zip.namelist(): self.__files.append(file_name) + if not self._has_dumpstate(self.__files): + nested = self._nested_bugreport(bugreport_zip) + if nested: + self.__zip = nested + self.__files = list(nested.namelist()) + + @staticmethod + def _has_dumpstate(file_names: List[str]) -> bool: + """Whether these members carry any of the entry points the bug report + modules read (see `BugReportModule._get_dumpstate_file`).""" + return any( + fnmatch.filter(file_names, pattern) for pattern in DUMPSTATE_ENTRY_POINTS + ) + + def _nested_bugreport(self, outer: ZipFile) -> Optional[ZipFile]: + """Descend one level into an OEM wrapper archive. + + MIUI / HyperOS hands out a zip of app logs, ANR traces and tcpdump + captures with the real `bugreport--.zip` nested + inside. Without this descent the outer archive has no entry point, + every module reports it found no files, and the command still exits 0 + with an empty result. + """ + candidates = [ + name for name in outer.namelist() if name.lower().endswith(".zip") + ] + candidates.sort( + key=lambda name: ( + "bugreport" not in name.lower() and "dumpstate" not in name.lower() + ) + ) + for name in candidates: + try: + inner = ZipFile(io.BytesIO(outer.read(name))) + except (BadZipFile, OSError): + continue + if self._has_dumpstate(inner.namelist()): + log.info("Found the bug report nested inside the archive: %s", name) + return inner + inner.close() + return None + def init(self) -> None: if self.target_path: self.log.info("Checking Android bug report at path: %s", self.target_path) diff --git a/src/mvt/android/modules/bugreport/base.py b/src/mvt/android/modules/bugreport/base.py index 25396f95..8d4f4ace 100644 --- a/src/mvt/android/modules/bugreport/base.py +++ b/src/mvt/android/modules/bugreport/base.py @@ -6,6 +6,7 @@ import datetime import fnmatch import logging import os +import re from pathlib import Path from typing import List, Optional from zipfile import ZipFile @@ -13,6 +14,15 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from mvt.common.module import ModuleResults, MVTModule +# `------ 0.101s was the duration of 'SOME SECTION' ------`, printed when that +# section finishes and not necessarily between two sections. +SECTION_DURATION = re.compile(r"^-{3,}\s*[0-9.]+s was the duration of", re.IGNORECASE) + +# The members a bug report archive can be entered through, in the order +# _get_dumpstate_file() tries them. An archive carrying none of them is not a +# bug report at this level (see CmdAndroidCheckBugreport._has_dumpstate). +DUMPSTATE_ENTRY_POINTS = ("main_entry.txt", "dumpState_*.log", "*/dumpsys.txt") + class BugReportModule(MVTModule): """This class provides a base for all Android Bug Report modules.""" @@ -87,7 +97,9 @@ class BugReportModule(MVTModule): return data def _get_dumpstate_file(self) -> Optional[bytes]: - main = self._get_files_by_pattern("main_entry.txt") + main_entry, dumpstate_log, dumpsys_txt = DUMPSTATE_ENTRY_POINTS + + main = self._get_files_by_pattern(main_entry) if main: main_content = self._get_file_content(main[0]) try: @@ -95,11 +107,11 @@ class BugReportModule(MVTModule): except KeyError: return None - dumpstate_logs = self._get_files_by_pattern("dumpState_*.log") + dumpstate_logs = self._get_files_by_pattern(dumpstate_log) if dumpstate_logs: return self._get_file_content(dumpstate_logs[0]) - dumpsys_files = self._get_files_by_pattern("*/dumpsys.txt") + dumpsys_files = self._get_files_by_pattern(dumpsys_txt) if dumpsys_files: return self._get_file_content(dumpsys_files[0]) @@ -122,6 +134,12 @@ class BugReportModule(MVTModule): in_section = True continue if stripped.startswith("------"): + # dumpstate prints a section's timing line when that section + # finishes, which can land in the middle of the one being + # written. Treating it as a boundary truncates the section at + # an arbitrary point, silently. + if SECTION_DURATION.match(stripped): + continue break lines.append(line) return "\n".join(lines) diff --git a/src/mvt/android/modules/bugreport/dumpsys_accessibility.py b/src/mvt/android/modules/bugreport/dumpsys_accessibility.py index 02d85a30..fddee84a 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_accessibility.py +++ b/src/mvt/android/modules/bugreport/dumpsys_accessibility.py @@ -48,9 +48,22 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule): ) self.parse(content) + listed = unnamed = 0 for result in self.results: - self.log.info('Found accessibility service "%s"', result.get("component")) + if result.get("component"): + listed += 1 + self.log.info( + 'Found accessibility service "%s"', result.get("component") + ) + continue + # The operator gets this per user as a LOW alert from + # check_indicators(); here it only has to survive into the summary, + # so that a stated count never reads as "a total of 0". + unnamed += result.get("unnamed_service_count") or 0 self.log.info( - "Identified a total of %d accessibility services", len(self.results) + "Identified a total of %d accessibility services (%d more stated by " + "the dump without a component name)", + listed, + unnamed, ) diff --git a/src/mvt/common/url.py b/src/mvt/common/url.py index 426d64bd..d432e3ed 100644 --- a/src/mvt/common/url.py +++ b/src/mvt/common/url.py @@ -349,7 +349,9 @@ class URL: return tld_obj if tld_obj is None: return "" - return tld_obj.parsed_url.netloc.lower().lstrip("www.") + # removeprefix, not lstrip: lstrip takes a set of characters, so it ate the + # leading "w"s and dots of any domain ("web.evil.com" -> "eb.evil.com"). + return tld_obj.parsed_url.netloc.lower().removeprefix("www.") def get_top_level(self) -> str: """Get only the top-level domain from a URL. diff --git a/src/mvt/ios/data/ios_versions.json b/src/mvt/ios/data/ios_versions.json index 8cea66c2..839aa200 100644 --- a/src/mvt/ios/data/ios_versions.json +++ b/src/mvt/ios/data/ios_versions.json @@ -1279,5 +1279,9 @@ { "version": "27.0", "build": "24A437" + }, + { + "version": "27.0.1", + "build": "24A446" } ] \ No newline at end of file diff --git a/src/mvt/ios/modules/backup/manifest.py b/src/mvt/ios/modules/backup/manifest.py index cc74cbb6..d0a6680d 100644 --- a/src/mvt/ios/modules/backup/manifest.py +++ b/src/mvt/ios/modules/backup/manifest.py @@ -147,6 +147,12 @@ class Manifest(IOSExtraction): ) names = [description[0] for description in cur.description] + # An incomplete backup still lists the files it failed to acquire in + # its manifest. Only regular files are stored in the backup folder, so + # directories and symlinks (flags 2 and 4) are not looked up. + stored_file_ids = self._get_stored_backup_file_ids() + missing_files = 0 + for file_entry in cur: file_data = {} for index, value in enumerate(file_entry): @@ -160,6 +166,22 @@ class Manifest(IOSExtraction): "created": "", } + if ( + stored_file_ids is not None + and file_data["flags"] == 1 + and file_data["fileID"] not in stored_file_ids + ): + # Without this, a module which found nothing for one of these + # files would look like a negative result rather than a gap in + # the acquisition. + cleaned_metadata["missing"] = True + missing_files += 1 + self.log.debug( + "File %s is listed in the manifest but was not found in the " + "backup folder", + cleaned_metadata["relative_path"], + ) + if file_data["file"]: try: file_plist = plistlib.load(io.BytesIO(file_data["file"])) @@ -197,3 +219,10 @@ class Manifest(IOSExtraction): conn.close() self.log.info("Extracted a total of %d file metadata items", len(self.results)) + + if missing_files: + self.log.info( + "Found %d files listed in the manifest but missing from the backup " + "folder. The backup might be incomplete.", + missing_files, + ) diff --git a/src/mvt/ios/modules/base.py b/src/mvt/ios/modules/base.py index a5c3354b..e5281fda 100644 --- a/src/mvt/ios/modules/base.py +++ b/src/mvt/ios/modules/base.py @@ -216,6 +216,55 @@ class IOSExtraction(MVTModule): return None + def _get_stored_backup_file_ids(self) -> Optional[set[str]]: + """List the IDs of the files actually stored in the backup folder. + + A backup folder stores each file under a two character subfolder named + after the first two characters of its file ID. Walking the folders once + is cheaper than a filesystem lookup per file ID, and it keeps callers + which compare a whole manifest against the folder from paying a + `resolve()` for every entry. + + :returns: The file IDs found at their expected paths within the backup + folder, or None if the inventory could not be completed. + """ + if not self.target_path: + return None + + file_ids: set[str] = set() + try: + backup_root = Path(self.target_path).resolve() + with os.scandir(self.target_path) as entries: + for entry in entries: + if len(entry.name) != 2 or any( + char not in "0123456789abcdef" for char in entry.name + ): + continue + if not entry.is_dir(): + continue + if not Path(entry.path).resolve().is_relative_to(backup_root): + continue + with os.scandir(entry.path) as sub_entries: + for sub_entry in sub_entries: + if sub_entry.name[:2] != entry.name: + continue + if sub_entry.is_symlink() and not Path( + sub_entry.path + ).resolve().is_relative_to(backup_root): + continue + if sub_entry.is_file(): + file_ids.add(sub_entry.name) + except OSError as exc: + self.log.warning( + "Unable to list the files stored in the backup folder %s: %s. " + "Skipping the missing-file check.", + self.target_path, + exc, + ) + return None + + return file_ids + def _get_fs_files_from_patterns(self, root_paths: list) -> Iterator[str]: if not self.target_path: return diff --git a/src/mvt/ios/modules/fs/filesystem.py b/src/mvt/ios/modules/fs/filesystem.py index 563d0394..b3cc1134 100644 --- a/src/mvt/ios/modules/fs/filesystem.py +++ b/src/mvt/ios/modules/fs/filesystem.py @@ -82,7 +82,9 @@ class Filesystem(IOSExtraction): try: dir_path = os.path.join(root, dir_name) result = { - "path": os.path.relpath(dir_path, self.target_path), + "path": os.path.relpath(dir_path, self.target_path).replace( + os.sep, "/" + ), "modified": convert_unix_to_iso(os.stat(dir_path).st_mtime), } except Exception: @@ -94,7 +96,9 @@ class Filesystem(IOSExtraction): try: file_path = os.path.join(root, file_name) result = { - "path": os.path.relpath(file_path, self.target_path), + "path": os.path.relpath(file_path, self.target_path).replace( + os.sep, "/" + ), "modified": convert_unix_to_iso(os.stat(file_path).st_mtime), } except Exception: diff --git a/tests/android/test_artifact_dumpsys_accessibility.py b/tests/android/test_artifact_dumpsys_accessibility.py index c727c571..6ba1a478 100644 --- a/tests/android/test_artifact_dumpsys_accessibility.py +++ b/tests/android/test_artifact_dumpsys_accessibility.py @@ -35,11 +35,16 @@ class TestDumpsysAccessibilityArtifact: assert len(da.results) == 0 da.parse(data) - assert len(da.results) == 1 + # One named service, plus one count-only record: the dump states + # `installedServiceCount=2` and names only one component. + assert len(da.results) == 2 assert da.results[0]["package_name"] == "com.malware.accessibility" assert da.results[0]["service_name"] == "com.malware.service.malwareservice" assert da.results[0]["enabled"] is True - assert da.results[0]["installed"] is False + # This fixture never prints an `installed services:` section, so the + # dump does not state the installed status. Reporting False would turn + # "not stated" into "not installed", so it reads None here. + assert da.results[0]["installed"] is None def test_accessibility_service_alert(self): da = DumpsysAccessibilityArtifact() @@ -50,9 +55,11 @@ class TestDumpsysAccessibilityArtifact: da.check_indicators() - assert len(da.alertstore.alerts) == 1 + assert len(da.alertstore.alerts) == 2 assert da.alertstore.alerts[0].level == AlertLevel.MEDIUM assert da.alertstore.alerts[0].event == da.results[0] + assert da.alertstore.alerts[1].level == AlertLevel.LOW + assert da.alertstore.alerts[1].event == da.results[1] def test_same_component_is_kept_for_each_user(self): da = DumpsysAccessibilityArtifact() @@ -84,7 +91,11 @@ User state[attributes:{id=10 assert len(da.alertstore.alerts) == 0 da.check_indicators() assert len(da.alertstore.alerts) == len(da.results) - assert da.alertstore.count(AlertLevel.MEDIUM) == 3 + # Every service in this fixture is installed and switched off + # (`enabled services:{}` is printed and empty), so the three non-IOC + # findings are LOW, not MEDIUM. The IOC match is unaffected by the + # state. + assert da.alertstore.count(AlertLevel.LOW) == 3 assert da.alertstore.count(AlertLevel.CRITICAL) == 1 critical_alert = next( alert diff --git a/tests/android/test_artifact_dumpsys_accessibility_count.py b/tests/android/test_artifact_dumpsys_accessibility_count.py new file mode 100644 index 00000000..8c191d32 --- /dev/null +++ b/tests/android/test_artifact_dumpsys_accessibility_count.py @@ -0,0 +1,149 @@ +# 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/ +"""The dump's own installed-service count must survive into the artifact. + +Most builds never print the `installed services: {…}` block; they state `installedServiceCount=N` in the user's `attributes:{…}` line and +list nothing. Dropping that number makes an artifact that says "no +accessibility services" about a dump that said there are five. +""" + +from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact +from mvt.common.alerts import AlertLevel + +from ..utils import get_artifact + +AOSP_NO_LIST = """\ +ACCESSIBILITY MANAGER (dumpsys accessibility) +User state[ + attributes:{id=0, touchExplorationEnabled=false, installedServiceCount=5} + Bound services:{} + Enabled services:{} + Binding services:{} + Crashed services:{} +""" + +ONE_UI_WITH_LIST = """\ +ACCESSIBILITY MANAGER (dumpsys accessibility) +User state[attributes:{id=0, installedServiceCount=2} + installed services: { + 0 : com.example.app/com.example.app.Service + 1 : com.other.app/.Helper + } + enabled services: { + } +""" + +TWO_USERS = """\ +ACCESSIBILITY MANAGER (dumpsys accessibility) +User state[attributes:{id=0, installedServiceCount=1} + installed services: { + 0 : com.example.app/com.example.app.Service + } +User state[attributes:{id=95, installedServiceCount=3} + Enabled services:{} +""" + +PARTIAL_TWO_USERS = """\ +ACCESSIBILITY MANAGER (dumpsys accessibility) +User state[attributes:{id=0, installedServiceCount=3} + Enabled services:{{com.example.app/com.example.app.Service}} +User state[attributes:{id=10, installedServiceCount=1} + Enabled services:{{com.other.app/.Helper}} +""" + +ZERO_COUNT = """\ +ACCESSIBILITY MANAGER (dumpsys accessibility) +User state[attributes:{id=0, installedServiceCount=0} + Enabled services:{} +""" + + +def _parse(content): + artifact = DumpsysAccessibilityArtifact() + artifact.results = [] + artifact.parse(content) + return artifact + + +class TestAccessibilityInstalledServiceCount: + def test_stated_count_without_a_list_is_kept(self): + artifact = _parse(AOSP_NO_LIST) + assert len(artifact.results) == 1 + record = artifact.results[0] + assert record["installed_service_count"] == 5 + assert record["component"] is None + # Every state flag stays unknown: the dump named no service to which a + # state could belong. + assert record["installed"] is None + assert record["enabled"] is None + + def test_a_stated_count_is_reported_as_low(self): + artifact = _parse(AOSP_NO_LIST) + artifact.check_indicators() + alerts = artifact.alertstore.alerts + assert len(alerts) == 1 + # A count is a coverage statement, not a running service: it must not + # compete with a service the dump says is enabled. + assert alerts[0].level == AlertLevel.LOW + assert "does not list their component names" in alerts[0].message + assert "5 installed" in alerts[0].message + + def test_a_listed_user_carries_the_count_on_each_service(self): + artifact = _parse(ONE_UI_WITH_LIST) + assert len(artifact.results) == 2 + assert {record["installed_service_count"] for record in artifact.results} == {2} + assert all(record["component"] for record in artifact.results) + + def test_only_the_unlisted_user_gets_a_count_record(self): + artifact = _parse(TWO_USERS) + listed = [record for record in artifact.results if record["component"]] + unlisted = [record for record in artifact.results if not record["component"]] + assert [record["user_id"] for record in listed] == [0] + assert [record["user_id"] for record in unlisted] == [95] + assert unlisted[0]["installed_service_count"] == 3 + + def test_a_zero_count_adds_nothing(self): + # "Zero installed" is a negative result the empty section already + # states; a record for it would be noise. + assert _parse(ZERO_COUNT).results == [] + + def test_a_partly_named_count_reports_the_unnamed_rest(self): + # The Android 14 fixture states `installedServiceCount=2` and names one + # enabled component. The listing is incomplete, and must not read as + # complete. + artifact = DumpsysAccessibilityArtifact() + artifact.results = [] + with open( + get_artifact("android_data/dumpsys_accessibility_v14_or_later.txt") + ) as handle: + artifact.parse(handle.read()) + unlisted = [record for record in artifact.results if not record["component"]] + assert len(unlisted) == 1 + assert unlisted[0]["installed_service_count"] == 2 + assert unlisted[0]["unnamed_service_count"] == 1 + + artifact.check_indicators() + low = [ + alert + for alert in artifact.alertstore.alerts + if alert.level == AlertLevel.LOW + ] + assert len(low) == 1 + assert "only 1 of them (1 unnamed)" in low[0].message + + def test_the_unnamed_rest_is_counted_per_user(self): + # User 0 names one of three, user 10 names its only one: the gap + # belongs to user 0 alone. + artifact = _parse(PARTIAL_TWO_USERS) + unlisted = [record for record in artifact.results if not record["component"]] + assert [ + (record["user_id"], record["unnamed_service_count"]) for record in unlisted + ] == [(0, 2)] + + def test_a_fully_named_count_adds_nothing(self): + artifact = _parse(ONE_UI_WITH_LIST) + assert all( + record["unnamed_service_count"] is None for record in artifact.results + ) diff --git a/tests/android/test_artifact_dumpsys_accessibility_state.py b/tests/android/test_artifact_dumpsys_accessibility_state.py new file mode 100644 index 00000000..d6bbe789 --- /dev/null +++ b/tests/android/test_artifact_dumpsys_accessibility_state.py @@ -0,0 +1,155 @@ +# 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/ +"""Installed is not the same as enabled, and the dump says which. + +Two things this guards: + + * a section the dump never printed must read as None ("not stated"), not as + False ("not enabled"); + * the alert must name the state, instead of firing identically on a device + where nothing is switched on and one where something is bound. +""" + +from types import SimpleNamespace + +from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact +from mvt.common.alerts import AlertLevel + +from ..utils import get_artifact + +NO_STATE_SECTIONS = """\ +ACCESSIBILITY MANAGER (dumpsys accessibility) +User state[attributes:{id=0, currentUser=true} + installed services: { + 0 : com.example.app/com.example.app.Service + } +""" + +ENABLED_BLOCK = """\ +ACCESSIBILITY MANAGER (dumpsys accessibility) +User state[attributes:{id=0, currentUser=true} + installed services: { + 0 : com.example.app/com.example.app.Service + 1 : com.other.app/.Helper + } + enabled services: { + 0 : com.other.app/.Helper + } + bound services:{ + 0 : com.other.app/.Helper + } +""" + +# User 0 prints only `installed services`, user 10 only `Enabled services`. +# Neither section speaks for the other user. +TWO_USERS_DIFFERENT_SECTIONS = """\ +ACCESSIBILITY MANAGER (dumpsys accessibility) +User state[attributes:{id=0, currentUser=true} + installed services: { + 0 : com.example.app/com.example.app.Service + } +User state[attributes:{id=10, currentUser=false} + Enabled services:{{com.other.app/.Helper}} +""" + + +class _IndicatorsMatching: + """Minimal stand-in: matches one package id, like the STIX2 loader would.""" + + def __init__(self, package_name: str) -> None: + self.package_name = package_name + + def check_app_id(self, app_id): + if app_id != self.package_name: + return None + return SimpleNamespace( + message=f"Found a known suspicious app: {app_id}", ioc={"value": app_id} + ) + + +class TestAccessibilityServiceState: + def _parse(self, content): + artifact = DumpsysAccessibilityArtifact() + artifact.results = [] + artifact.parse(content) + return {r["component"]: r for r in artifact.results} + + def test_absent_sections_leave_the_state_unknown(self): + # None, not False: a build that does not print the sections says + # nothing about what is enabled, and that must not read as "nothing". + state = self._parse(NO_STATE_SECTIONS)[ + "com.example.app/com.example.app.Service" + ] + assert state["installed"] is True + assert state["enabled"] is None + assert state["bound"] is None + + def test_enabled_and_bound_are_attributed_per_service(self): + results = self._parse(ENABLED_BLOCK) + installed_only = results["com.example.app/com.example.app.Service"] + active = results["com.other.app/.Helper"] + assert (installed_only["enabled"], installed_only["bound"]) == (False, False) + assert (active["enabled"], active["bound"]) == (True, True) + + def test_alert_message_carries_the_state(self): + artifact = DumpsysAccessibilityArtifact() + artifact.results = [] + with open(get_artifact("android_data/dumpsys_accessibility.txt")) as handle: + artifact.parse(handle.read()) + artifact.check_indicators() + assert artifact.alertstore.alerts + assert all("installed" in alert.message for alert in artifact.alertstore.alerts) + + def test_a_switched_off_service_is_low_and_a_running_one_medium(self): + # A service the dump says is OFF still reaches the analyst, but must not + # compete with one that is actually bound. "Not stated" is not "off". + artifact = DumpsysAccessibilityArtifact() + artifact.results = [] + artifact.parse(ENABLED_BLOCK) + artifact.check_indicators() + by_level = {} + for alert in artifact.alertstore.alerts: + by_level.setdefault(alert.level, []).append(alert.message) + assert artifact.alertstore.count(AlertLevel.LOW) == 1 + assert artifact.alertstore.count(AlertLevel.MEDIUM) == 1 + assert "com.example.app" in by_level[AlertLevel.LOW][0] + assert "com.other.app" in by_level[AlertLevel.MEDIUM][0] + + def test_an_unstated_enabled_state_stays_medium(self): + artifact = DumpsysAccessibilityArtifact() + artifact.results = [] + artifact.parse(NO_STATE_SECTIONS) + artifact.check_indicators() + assert artifact.alertstore.count(AlertLevel.MEDIUM) == 1 + assert artifact.alertstore.count(AlertLevel.LOW) == 0 + + def test_a_disabled_service_is_still_matched_against_indicators(self): + # The state decides the severity of an ordinary finding, never whether + # the package is compared with the IOC feeds. + artifact = DumpsysAccessibilityArtifact() + artifact.results = [] + artifact.parse(ENABLED_BLOCK) + artifact.indicators = _IndicatorsMatching("com.example.app") + artifact.check_indicators() + assert artifact.alertstore.count(AlertLevel.CRITICAL) == 1 + assert artifact.alertstore.count(AlertLevel.LOW) == 0 + + def test_printed_sections_are_tracked_per_user(self): + artifact = DumpsysAccessibilityArtifact() + artifact.results = [] + artifact.parse(TWO_USERS_DIFFERENT_SECTIONS) + by_user = {r["user_id"]: r for r in artifact.results} + # User 0's enabled state is not stated, so it is unknown, not off. + assert (by_user[0]["installed"], by_user[0]["enabled"]) == (True, None) + # User 10's installed state is not stated either. + assert (by_user[10]["installed"], by_user[10]["enabled"]) == (None, True) + + artifact.check_indicators() + assert artifact.alertstore.count(AlertLevel.LOW) == 0 + assert artifact.alertstore.count(AlertLevel.MEDIUM) == 2 + assert not any( + "installed, not enabled" in alert.message + for alert in artifact.alertstore.alerts + ) diff --git a/tests/android/test_artifact_dumpsys_adb.py b/tests/android/test_artifact_dumpsys_adb.py index 8d501b7d..0e5b59f4 100644 --- a/tests/android/test_artifact_dumpsys_adb.py +++ b/tests/android/test_artifact_dumpsys_adb.py @@ -130,6 +130,101 @@ class TestDumpsysADBArtifact: assert key_store_entry["fingerprint"] == expected_fingerprint assert key_store_entry["last_connected"] == "1628501829898" + def test_parsing_adb_xml_with_crlf_line_endings(self): + da_adb = DumpsysADBArtifact() + file = get_artifact("android_data/dumpsys_adb_xml.txt") + with open(file, "rb") as f: + data = f.read().replace(b"\r\n", b"\n").replace(b"\n", b"\r\n") + + da_adb.parse(data) + + assert len(da_adb.results) == 1 + assert da_adb.results[0]["user_keys"][0]["user"] == "user@laptop" + assert da_adb.results[0]["keystore"][0]["last_connected"] == "1628501829898" + + def test_parsing_adb_wifi_with_mixed_line_endings(self): + da_adb = DumpsysADBArtifact() + data = ( + b"ADB MANAGER STATE (dumpsys adb):\n" + b"{\n" + b" debugging_manager={\n" + b" connected_to_adb=true\n" + b" user_keys=QUJDRA== host@example\n" + b" adb_wifi={\n" + b" enabled=false\n" + b" }\n" + b" }\r\n" + b"}\n" + b"--------- duration\n" + ) + + da_adb.parse(data) + + assert len(da_adb.results) == 1 + assert da_adb.results[0]["user_keys"][0]["user"] == "host@example" + assert da_adb.results[0]["adb_wifi"]["enabled"] == b"false" + + ADB_STATE = ( + b"ADB MANAGER STATE (dumpsys adb):\n" + b"{\n" + b" debugging_manager={\n" + b" connected_to_adb=true\n" + b" user_keys=QUJDRA== host@example\n" + b" }\n" + b"}\n" + b"--------- 0.5s was the duration of 'dumpsys adb'\n" + ) + + def test_a_later_dumpsys_section_does_not_extend_the_adb_state(self): + # A bug report holds many sections. Looking for the last closing brace + # in the whole output pulled a later section into this one, which threw + # IndexError out of parse() and lost the ADB records entirely. + da_adb = DumpsysADBArtifact() + da_adb.parse( + self.ADB_STATE + b"DUMP OF SERVICE other:\n" + b" debugging_manager={\n" + b" connected_to_adb=false\n" + b" user_keys=RVZJTA== attacker@host\n" + b" }\n" + b"}\n" + ) + + assert len(da_adb.results) == 1 + assert [key["user"] for key in da_adb.results[0]["user_keys"]] == [ + "host@example" + ] + assert da_adb.results[0]["connected_to_adb"] is True + + def test_braces_in_key_comment_are_not_state_delimiters(self): + for user in (b"host{example", b"host}}example"): + da_adb = DumpsysADBArtifact() + da_adb.parse( + b"ADB MANAGER STATE (dumpsys adb):\n" + b"{\n" + b" debugging_manager={\n" + b" connected_to_adb=true\n" + b" user_keys=QUJDRA== " + user + b"\n }\n}\n" + ) + + assert len(da_adb.results) == 1 + assert da_adb.results[0]["user_keys"][0]["user"] == user.decode() + + def test_unbalanced_state_is_reported_rather_than_raising(self): + da_adb = DumpsysADBArtifact() + da_adb.parse( + b"ADB MANAGER STATE (dumpsys adb):\n" + b"{\n" + b" debugging_manager={\n" + b" connected_to_adb=true\n" + b" }\n" + b" }\n" + b"}\n" + ) + + # No exception, and whatever was read before the bad line is kept. + assert len(da_adb.results) == 1 + assert da_adb.results[0]["connected_to_adb"] is True + class TestDumpsysADBStateAlerts: def test_no_androidqf_context_preserves_existing_behavior(self): diff --git a/tests/android/test_artifact_dumpsys_battery_daily.py b/tests/android/test_artifact_dumpsys_battery_daily.py index 5f7f997c..d232d219 100644 --- a/tests/android/test_artifact_dumpsys_battery_daily.py +++ b/tests/android/test_artifact_dumpsys_battery_daily.py @@ -144,3 +144,18 @@ class TestDumpsysBatteryDailyArtifact: "Detected uninstall of package com.example.app (vers 0)" ) assert uninstall_alert.event_time == "2026-01-10" + + def test_malformed_update_line_does_not_lose_later_records(self): + # A truncated "Update" line, or one from a vendor that omits "vers=", + # used to raise out of parse() and lose every record after it. + dba = DumpsysBatteryDailyArtifact() + dba.parse( + " Daily from 2021-05-10-08-00-00 to 2021-05-11-08-00-00:\n" + " Update com.first vers=1\n" + " Update com.truncated\n" + " Update com.no.equals vers 2\n" + " Update com.last vers=3\n" + ) + + assert [r["package_name"] for r in dba.results] == ["com.first", "com.last"] + assert [r["version_code"] for r in dba.results] == [1, 3] diff --git a/tests/android/test_bugreport_section_extraction.py b/tests/android/test_bugreport_section_extraction.py new file mode 100644 index 00000000..5778f947 --- /dev/null +++ b/tests/android/test_bugreport_section_extraction.py @@ -0,0 +1,42 @@ +# 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/ +"""A section ends at the next section, not at a timing line printed inside it.""" + +from mvt.android.modules.bugreport.base import BugReportModule + +# dumpstate prints a section's duration when that section finishes, which can +# land in the middle of the section currently being written. +DUMPSTATE = """\ +------ SYSTEM PROPERTIES (getprop) ------ +[nfc.initialized]: [true] +------ 0.101s was the duration of 'DROPBOX SYSTEM SERVER CRASHES' ------ +[ro.build.version.sdk]: [30] +[ro.product.model]: [SM-A305F] +------ 0.064s was the duration of 'SYSTEM PROPERTIES' ------ +------ STORAGE INFO (df) ------ +/dev/root 2.9G +""" + + +class TestExtractCommandSection: + def test_a_foreign_timing_line_does_not_end_the_section(self): + section = BugReportModule.extract_command_section( + DUMPSTATE, "------ SYSTEM PROPERTIES" + ) + assert "[ro.product.model]: [SM-A305F]" in section + assert section.count("\n") == 2 + + def test_the_next_section_is_still_the_boundary(self): + section = BugReportModule.extract_command_section( + DUMPSTATE, "------ SYSTEM PROPERTIES" + ) + assert "STORAGE INFO" not in section + assert "/dev/root" not in section + + def test_timing_lines_are_not_returned_as_content(self): + section = BugReportModule.extract_command_section( + DUMPSTATE, "------ SYSTEM PROPERTIES" + ) + assert "was the duration of" not in section diff --git a/tests/android/test_check_bugreport_wrapper.py b/tests/android/test_check_bugreport_wrapper.py new file mode 100644 index 00000000..741641f2 --- /dev/null +++ b/tests/android/test_check_bugreport_wrapper.py @@ -0,0 +1,56 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2023 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/ +"""An OEM wrapper archive must not read as an empty bug report. + +MIUI / HyperOS hands out a zip of app logs with the real +`bugreport--.zip` nested inside. The outer archive has +none of the entry points the modules read, so every module +reported it found nothing and the command still exited 0 — an empty analysis +that looks like a finished one. +""" + +import io +import zipfile + +from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport + +DUMPSTATE = "== dumpstate: 2026-01-01 00:00:00\nDUMP OF SERVICE package:\n" + + +def _inner_zip() -> bytes: + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as inner: + inner.writestr("main_entry.txt", "bugreport-test-2026-01-01-00-00-00.txt") + inner.writestr("bugreport-test-2026-01-01-00-00-00.txt", DUMPSTATE) + return buffer.getvalue() + + +def _wrapper_zip() -> zipfile.ZipFile: + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as outer: + outer.writestr("app_logs/hilog.txt", "unrelated OEM log\n") + outer.writestr("bugreport-test-2026-01-01-00-00-00.zip", _inner_zip()) + return zipfile.ZipFile(io.BytesIO(buffer.getvalue())) + + +class TestCheckBugreportWrapper: + def test_nested_bugreport_is_used(self, tmp_path): + cmd = CmdAndroidCheckBugreport(results_path=str(tmp_path)) + cmd.from_zip(_wrapper_zip()) + module = cmd.modules[0](results_path=str(tmp_path)) + cmd.module_init(module) + assert "main_entry.txt" in module.zip_files + + def test_plain_bugreport_is_left_alone(self, tmp_path): + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + archive.writestr("main_entry.txt", "bugreport.txt") + archive.writestr("bugreport.txt", DUMPSTATE) + archive.writestr("attachments/extra.zip", _inner_zip()) + cmd = CmdAndroidCheckBugreport(results_path=str(tmp_path)) + cmd.from_zip(zipfile.ZipFile(io.BytesIO(buffer.getvalue()))) + module = cmd.modules[0](results_path=str(tmp_path)) + cmd.module_init(module) + assert "attachments/extra.zip" in module.zip_files diff --git a/tests/common/test_cli_plugins.py b/tests/common/test_cli_plugins.py index 5abb2c98..0618b8d6 100644 --- a/tests/common/test_cli_plugins.py +++ b/tests/common/test_cli_plugins.py @@ -1,3 +1,4 @@ +import shlex from types import SimpleNamespace import click @@ -110,7 +111,7 @@ def test_load_command_option_supports_folders_and_repeated_paths(tmp_path): def test_loaded_command_participates_in_shell_completion(tmp_path): command_path = _write_command(tmp_path / "hello.py", "hello") group = _make_group() - words = f"group --load-command {command_path} he" + words = f"group --load-command {shlex.quote(str(command_path))} he" result = CliRunner().invoke( group, diff --git a/tests/common/test_cmd_plugins.py b/tests/common/test_cmd_plugins.py index adf93227..3767491a 100644 --- a/tests/common/test_cmd_plugins.py +++ b/tests/common/test_cmd_plugins.py @@ -1,4 +1,5 @@ import json +import re from types import SimpleNamespace import pytest @@ -54,20 +55,23 @@ def _run(command, arguments): return CliRunner().invoke(command, arguments, env={"COLUMNS": "200"}) -def _table_rows(output): - """Return the content of the table rows, without the header and the box.""" +def _table_lines(output): + """Return the cells of each line of the table, header first.""" return [ - [cell.strip() for cell in line.strip().strip("│").split("│")] + [cell.strip() for cell in re.split("[│┃]", line.strip().strip("│┃"))] for line in output.splitlines() - if "│" in line + if "│" in line or "┃" in line ] +def _table_rows(output): + """Return the content of the table rows, without the header and the box.""" + return _table_lines(output)[1:] + + def _table_header(output): - for line in output.splitlines(): - if "┃" in line: - return [cell.strip() for cell in line.strip().strip("┃").split("┃")] - return [] + lines = _table_lines(output) + return lines[0] if lines else [] def _install(monkeypatch, distributions, entry_points): diff --git a/tests/common/test_indicators.py b/tests/common/test_indicators.py index ac50f82b..a7495606 100644 --- a/tests/common/test_indicators.py +++ b/tests/common/test_indicators.py @@ -311,3 +311,31 @@ class TestIndicators: ind = Indicators(log=logging) ind.load_indicators_files([], load_default=False) assert ind.total_ioc_count == 9 + + def test_check_url_matches_w_prefixed_subdomain(self, tmp_path): + import json + + stix_file = tmp_path / "w-domain.stix2" + stix_file.write_text( + json.dumps( + { + "objects": [ + { + "type": "indicator", + "pattern": "[domain-name:value = 'web.evil.com']", + } + ] + } + ), + encoding="utf-8", + ) + ind = Indicators(log=logging) + ind.load_indicators_files([str(stix_file)], load_default=False) + + for url in ( + "https://web.evil.com/path", + "https://www.web.evil.com/path", + ): + match = ind.check_url(url) + assert match is not None + assert match.ioc.value == "web.evil.com" diff --git a/tests/common/test_url.py b/tests/common/test_url.py index ce0fb635..5a10c414 100644 --- a/tests/common/test_url.py +++ b/tests/common/test_url.py @@ -22,3 +22,23 @@ def test_google_maps_url_is_not_shortened(url): def test_other_google_short_url_is_shortened(): assert URL("https://goo.gl/example").check_if_shortened() is True + + +@pytest.mark.parametrize( + "url, domain", + [ + ("https://www.example.com/path", "example.com"), + # Only the whole "www." prefix comes off, not any leading "w" or "." character. + ("https://web.example.com", "web.example.com"), + ("https://wow.com", "wow.com"), + ("https://wired.com", "wired.com"), + ("https://www.wow.com", "wow.com"), + ], +) +def test_get_domain_strips_only_a_whole_www_prefix(url, domain): + assert URL(url).domain == domain + + +def test_shortener_starting_with_w_is_detected(): + assert URL("https://w3t.org/example").check_if_shortened() is True + assert URL("https://www.w3t.org/example").check_if_shortened() is True diff --git a/tests/ios_backup/test_decrypt.py b/tests/ios_backup/test_decrypt.py index 25decce5..1e2ce148 100644 --- a/tests/ios_backup/test_decrypt.py +++ b/tests/ios_backup/test_decrypt.py @@ -7,6 +7,7 @@ import logging import threading from pathlib import Path +import pytest from Crypto.Cipher import AES from mvt.ios.decrypt import DecryptBackup, MVTEncryptedBackup @@ -81,7 +82,10 @@ def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path): assert output_path.read_bytes() == b"plain content" -def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_path): +@pytest.mark.parametrize("with_symlink", [False, True], ids=["file-id", "symlink"]) +def test_process_backup_rejects_unsafe_file_ids_and_destinations( + mocker, tmp_path, with_symlink +): backup_path = tmp_path / "backup" destination = tmp_path / "destination" outside = tmp_path / "outside" @@ -92,20 +96,27 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_pat safe_file_id = "ef" + "3" * 38 unsafe_file_id = "../../outside-file" symlink_file_id = "ab" + "4" * 38 - for file_id in (safe_file_id, symlink_file_id): + file_ids = [safe_file_id] + if with_symlink: + file_ids.append(symlink_file_id) + for file_id in file_ids: source_path = backup_path / file_id[:2] / file_id source_path.parent.mkdir(parents=True, exist_ok=True) source_path.write_bytes(b"encrypted") - (destination / "ab").symlink_to(outside, target_is_directory=True) + if with_symlink: + try: + (destination / "ab").symlink_to(outside, target_is_directory=True) + except OSError: + pytest.skip("creating symbolic links is not permitted on this system") cursor = mocker.MagicMock() - cursor.__iter__.return_value = iter( - [ - (safe_file_id, "Domain", "safe", b"plist"), - (unsafe_file_id, "Domain", "unsafe", b"plist"), - (symlink_file_id, "Domain", "symlink", b"plist"), - ] - ) + records = [ + (safe_file_id, "Domain", "safe", b"plist"), + (unsafe_file_id, "Domain", "unsafe", b"plist"), + ] + if with_symlink: + records.append((symlink_file_id, "Domain", "symlink", b"plist")) + cursor.__iter__.return_value = iter(records) cursor_context = mocker.MagicMock() cursor_context.__enter__.return_value = cursor @@ -124,7 +135,8 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_pat decryptor.process_backup() assert (destination / safe_file_id[:2] / safe_file_id).read_bytes() == b"decrypted" - assert not (outside / symlink_file_id).exists() + if with_symlink: + assert not (outside / symlink_file_id).exists() backup.extract_file_by_id.assert_called_once() assert backup.extract_file_by_id.call_args.kwargs["file_id"] == safe_file_id diff --git a/tests/ios_backup/test_manifest.py b/tests/ios_backup/test_manifest.py index aaa747d0..9a52ee78 100644 --- a/tests/ios_backup/test_manifest.py +++ b/tests/ios_backup/test_manifest.py @@ -5,7 +5,12 @@ import gc import logging +import os +import shutil import warnings +from pathlib import Path + +import pytest from mvt.common.indicators import Indicators from mvt.common.module import run_module @@ -14,8 +19,58 @@ from mvt.ios.modules.backup.manifest import Manifest from ..utils import get_ios_backup_folder +# fileID of HomeDomain::Library/SMS/sms.db in the test backup. It is one of the +# few files the test backup actually stores. +SMS_FILE_ID = "3d0d7e5fb2ce288813306e4d4636395e047a3d28" + + +@pytest.fixture +def backup_without_stored_files(tmp_path): + """A copy of the test backup with every stored file removed. + + The test backup only ships a handful of the files its manifest lists, so + dropping what it does store leaves a backup where every regular file the + manifest mentions is missing from the folder. + """ + backup_path = tmp_path / "backup" + shutil.copytree(get_ios_backup_folder(), backup_path) + + for file_id_folder in backup_path.iterdir(): + if not file_id_folder.is_dir(): + continue + for backup_file in file_id_folder.iterdir(): + backup_file.unlink() + + return str(backup_path) + class TestIOSExtraction: + @pytest.mark.parametrize("link_directory", [False, True], ids=["file", "directory"]) + @pytest.mark.parametrize("inside_backup", [False, True], ids=["outside", "inside"]) + def test_stored_file_inventory_matches_symlink_resolution( + self, tmp_path, link_directory, inside_backup + ): + backup_path = tmp_path / "backup" + backup_path.mkdir() + destination = (backup_path if inside_backup else tmp_path) / "contents" + destination.mkdir() + (destination / SMS_FILE_ID).write_bytes(b"backup file") + bucket = backup_path / SMS_FILE_ID[:2] + try: + if link_directory: + bucket.symlink_to(destination, target_is_directory=True) + else: + bucket.mkdir() + (bucket / SMS_FILE_ID).symlink_to(destination / SMS_FILE_ID) + except OSError: + pytest.skip("creating symbolic links is not permitted on this system") + + m = IOSExtraction(target_path=str(backup_path)) + assert bool(m._get_backup_file_from_id(SMS_FILE_ID)) is inside_backup + assert m._get_stored_backup_file_ids() == ( + {SMS_FILE_ID} if inside_backup else set() + ) + def test_get_backup_files_from_manifest_closes_connection(self): m = IOSExtraction(target_path=get_ios_backup_folder()) @@ -41,6 +96,103 @@ class TestManifestModule: assert len(m.timeline) == 5881 assert len(m.alertstore.alerts) == 0 + def test_manifest_flags_the_files_missing_from_an_incomplete_backup(self): + m = Manifest(target_path=get_ios_backup_folder()) + run_module(m) + + missing = [result for result in m.results if result.get("missing")] + # The test backup only stores a handful of the files its manifest + # lists, and every one of the rest is a regular file (flags 1). + assert len(missing) == 1079 + assert all(result["flags"] == 1 for result in missing) + + stored = [result for result in m.results if result["file_id"] == SMS_FILE_ID] + assert len(stored) == 1 + assert "missing" not in stored[0] + + def test_manifest_flags_every_stored_file_removed_from_the_backup_folder( + self, backup_without_stored_files + ): + m = Manifest(target_path=backup_without_stored_files) + run_module(m) + + missing = [result for result in m.results if result.get("missing")] + assert len(missing) == 1089 + + def test_manifest_flags_a_file_removed_from_the_backup_folder(self, tmp_path): + backup_path = tmp_path / "backup" + shutil.copytree(get_ios_backup_folder(), backup_path) + (backup_path / SMS_FILE_ID[:2] / SMS_FILE_ID).unlink() + + m = Manifest(target_path=str(backup_path)) + run_module(m) + + removed = [result for result in m.results if result["file_id"] == SMS_FILE_ID] + assert len(removed) == 1 + assert removed[0]["missing"] is True + + @pytest.mark.parametrize("failed_folder", ["", "3d"], ids=["root", "bucket"]) + def test_manifest_skips_missing_check_when_inventory_fails( + self, monkeypatch, caplog, failed_folder + ): + backup_path = Path(get_ios_backup_folder()) + failed_path = backup_path / failed_folder + scandir = os.scandir + + def fail_listing(path): + if Path(path) == failed_path: + raise PermissionError("cannot list backup folder") + return scandir(path) + + monkeypatch.setattr(os, "scandir", fail_listing) + m = Manifest(target_path=str(backup_path)) + with caplog.at_level(logging.INFO): + m.run() + + assert len(m.results) == 3721 + assert m._get_backup_file_from_id(SMS_FILE_ID) is not None + assert all("missing" not in result for result in m.results) + assert "Skipping the missing-file check" in caplog.text + assert "The backup might be incomplete" not in caplog.text + + def test_manifest_ignores_unreadable_unrelated_folders(self, tmp_path, monkeypatch): + backup_path = tmp_path / "backup" + shutil.copytree(get_ios_backup_folder(), backup_path) + unrelated = backup_path / "notes" + unrelated.mkdir() + scandir = os.scandir + + def fail_listing(path): + if Path(path) == unrelated: + raise PermissionError("cannot list unrelated folder") + return scandir(path) + + monkeypatch.setattr(os, "scandir", fail_listing) + m = Manifest(target_path=str(backup_path)) + m.run() + + assert sum(bool(result.get("missing")) for result in m.results) == 1079 + stored = next( + result for result in m.results if result["file_id"] == SMS_FILE_ID + ) + assert "missing" not in stored + + @pytest.mark.parametrize("wrong_folder", ["ab", "notes"]) + def test_manifest_flags_files_in_the_wrong_folder(self, tmp_path, wrong_folder): + backup_path = tmp_path / "backup" + shutil.copytree(get_ios_backup_folder(), backup_path) + destination = backup_path / wrong_folder + destination.mkdir(exist_ok=True) + (backup_path / SMS_FILE_ID[:2] / SMS_FILE_ID).rename(destination / SMS_FILE_ID) + + m = Manifest(target_path=str(backup_path)) + m.run() + + assert m._get_backup_file_from_id(SMS_FILE_ID) is None + moved = next(result for result in m.results if result["file_id"] == SMS_FILE_ID) + assert moved["missing"] is True + assert sum(bool(result.get("missing")) for result in m.results) == 1080 + def test_detection(self, indicator_file): m = Manifest(target_path=get_ios_backup_folder()) ind = Indicators(log=logging.getLogger()) diff --git a/tests/ios_backup/test_sqlite_handling.py b/tests/ios_backup/test_sqlite_handling.py index f7806401..2774098e 100644 --- a/tests/ios_backup/test_sqlite_handling.py +++ b/tests/ios_backup/test_sqlite_handling.py @@ -9,6 +9,8 @@ import plistlib import shutil import sqlite3 +import pytest + from mvt.ios.modules.base import IOSExtraction from mvt.ios.modules.fs.analytics import Analytics @@ -45,6 +47,9 @@ def test_open_sqlite_reads_wal_without_modifying_evidence(tmp_path): assert not os.path.exists(str(evidence_path) + "-shm") +@pytest.mark.skipif( + shutil.which("sqlite3") is None, reason="the recovery needs the sqlite3 binary" +) def test_recovery_preserves_source_database(tmp_path): database_path = tmp_path / "source.db" conn = sqlite3.connect(database_path) diff --git a/tests/ios_fs/test_filesystem.py b/tests/ios_fs/test_filesystem.py index 636c004a..fe3ecf68 100644 --- a/tests/ios_fs/test_filesystem.py +++ b/tests/ios_fs/test_filesystem.py @@ -3,15 +3,45 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ import logging +import ntpath +from types import SimpleNamespace from mvt.common.indicators import Indicators from mvt.common.module import run_module +from mvt.ios.modules.fs import filesystem from mvt.ios.modules.fs.filesystem import Filesystem from ..utils import get_ios_backup_folder class TestFilesystem: + def test_windows_paths_are_normalized_for_indicators( + self, monkeypatch, indicators_factory + ): + m = Filesystem(target_path=r"C:\dump") + m.indicators = indicators_factory( + file_paths=["matched/directory"], processes=["matched"] + ) + monkeypatch.setattr( + filesystem, + "os", + SimpleNamespace( + sep="\\", + path=ntpath, + walk=lambda _: [(r"C:\dump\matched", ["directory"], ["file.txt"])], + stat=lambda _: SimpleNamespace(st_mtime=0), + ), + ) + + m.run() + m.check_indicators() + + assert {result["path"] for result in m.results} == { + "matched/directory", + "matched/file.txt", + } + assert len(m.alertstore.alerts) == 3 + def test_filesystem(self): m = Filesystem(target_path=get_ios_backup_folder()) run_module(m) diff --git a/tests/test_check_android_androidqf.py b/tests/test_check_android_androidqf.py index 4e6ef984..65973109 100644 --- a/tests/test_check_android_androidqf.py +++ b/tests/test_check_android_androidqf.py @@ -85,7 +85,7 @@ class TestCheckAndroidqfCommand: 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") + (data_path / "adb_host_key.pub").write_bytes(b"QUJDRA== acquisition@host\n") command = CmdAndroidCheckAndroidQF(target_path=str(data_path)) command.init() diff --git a/tests/test_cmd_check_sysdiagnose.py b/tests/test_cmd_check_sysdiagnose.py index b00b648b..702e985f 100644 --- a/tests/test_cmd_check_sysdiagnose.py +++ b/tests/test_cmd_check_sysdiagnose.py @@ -116,7 +116,7 @@ def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path): module = SysdiagnoseExtraction() command.module_init(module) assert module.tar is None - assert module.parent_path == str(extracted_path.parent) + assert Path(module.parent_path) == extracted_path.parent finally: command.finish() diff --git a/tests/test_completion.py b/tests/test_completion.py index 08ad3ce3..d240e492 100644 --- a/tests/test_completion.py +++ b/tests/test_completion.py @@ -43,6 +43,7 @@ class TestCompletionCommand: def test_completion_install_updates_bashrc_once(self, tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) runner = CliRunner() result = runner.invoke(mvt_cli, ["completion", "bash", "--install"]) @@ -67,6 +68,7 @@ class TestCompletionCommand: self, tmp_path, monkeypatch ): monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) runner = CliRunner() result = runner.invoke(mvt_cli, ["completion", "fish", "--install"])