Merge branch 'main' into fix/ios-parser-field-coverage

Stop persisting environment variables to config.yaml

src/mvt/android/artifacts/settings.py

@@ -4,6 +4,11 @@
 #   https://license.mvt.re/1.1/
 
 import re
+from datetime import datetime
+from typing import Optional, Sequence
+
+from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
+from mvt.common.utils import convert_datetime_to_iso
 
 from .artifact import AndroidArtifact
 
@@ -60,45 +65,240 @@ ANDROID_DANGEROUS_SETTINGS = [
     },
 ]
 
+# dumpsys prints the fields of a setting record, and of a change history entry,
+# always in this order and separated by a single space. After the value come
+# `default:` and `defaultSystemSet:` when a default is recorded, then `tag:`;
+# some vendor builds add whether the value survives a restore, either as
+# `isValuePreservedInRestore:` or as a bare `notPreservedInRestore` token.
+SETTING_FIELDS = (
+    "_id",
+    "name",
+    "pkg",
+    "value",
+    "default",
+    "defaultSystemSet",
+    "tag",
+    "isValuePreservedInRestore",
+)
+HISTORY_FIELDS = ("time", "mode", "oldValue", "newValue", "package")
+
+NAMESPACE_PATTERN = re.compile(
+    r"^(CONFIG|GLOBAL|SECURE|SYSTEM) SETTINGS \(user (\d+)\)$"
+)
+SECTION_END_PATTERN = re.compile(r"ending at: (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})")
+
 
 class Settings(AndroidArtifact):
-    def parse(self, content: str) -> None:
+    """Parser for the `dumpsys settings` output.
-        self.results: dict[str, dict[str, str]] = {}
+
-        namespace: str | None = None
+    Every row of the settings provider becomes one result, keeping the fields
-        for line in content.splitlines():
+    dumpsys prints alongside the value: the row id, the package which recorded
-            heading = re.match(
+    the setting, the default, the tag, and the change history. A setting name
-                r"^(CONFIG|GLOBAL|SECURE|SYSTEM) SETTINGS \(user (\d+)\)$",
+    can appear more than once within a namespace, so results are a list rather
-                line.strip(),
+    than a mapping.
-            )
+    """
-            if heading:
+
-                namespace = f"{heading.group(1).lower()}:user_{heading.group(2)}"
+    def serialize(self, result: ModuleAtomicResult) -> ModuleSerializedResult:
-                self.results[namespace] = {}
+        records = []
+        for entry in result.get("history", []):
+            if not entry.get("timestamp"):
                 continue
-            if namespace is None or not line.startswith("_id:"):
+
-                continue
+            records.append(
-            setting = re.match(
+                {
-                r"^_id:\S+\s+name:(.*?)\s+pkg:.*?\s+value:(.*?)"
+                    "timestamp": entry["timestamp"],
-                r"(?:\s+default:.*\s+defaultSystemSet:(?:true|false))?$",
+                    "module": self.__class__.__name__,
-                line,
+                    "event": "settings_change",
+                    "data": (
+                        f"{result.get('namespace')} setting "
+                        f'"{result.get("name")}" changed from '
+                        f'"{entry.get("oldValue")}" to "{entry.get("newValue")}" '
+                        f"by {entry.get('pkg')}"
+                    ),
+                }
             )
-            if setting:
+
-                self.results[namespace][setting.group(1)] = setting.group(2)
+        return records
 
     def check_indicators(self) -> None:
-        for namespace, settings in self.results.items():
+        for result in self.results:
-            for key, value in settings.items():
+            name = result.get("name")
-                for danger in ANDROID_DANGEROUS_SETTINGS:
+            value = result.get("value")
-                    # Check if one of the dangerous settings is using an unsafe
+            for danger in ANDROID_DANGEROUS_SETTINGS:
-                    # value (different than the one specified).
+                # Check if one of the dangerous settings is using an unsafe
-                    if danger["key"] == key and danger["safe_value"] != value:
+                # value (different than the one specified).
-                        self.alertstore.medium(
+                if danger["key"] != name or danger["safe_value"] == value:
-                            f'Found suspicious "{namespace}" setting "{key} = {value}" ({danger["description"]})',
+                    continue
-                            "",
+
-                            {
+                history = result.get("history") or []
-                                "namespace": namespace,
+                self.alertstore.medium(
-                                "key": key,
+                    f'Found suspicious "{result.get("namespace")}" setting '
-                                "value": value,
+                    f'"{name} = {value}" ({danger["description"]})',
-                                "description": danger["description"],
+                    history[-1]["timestamp"] if history else "",
-                            },
+                    result,
-                        )
+                )
-                        break
+                break
+
+    def parse(self, content: str) -> None:
+        self.results: list[ModuleAtomicResult] = []
+        section_end = self._parse_section_end(content)
+        namespace: Optional[str] = None
+        user: Optional[str] = None
+        record_lines: list[str] = []
+        history_lines: list[str] = []
+        in_history = False
+
+        def flush() -> None:
+            nonlocal record_lines, history_lines, in_history
+            if record_lines:
+                self.results.append(
+                    self._build_record(
+                        namespace, user, record_lines, history_lines, section_end
+                    )
+                )
+            record_lines = []
+            history_lines = []
+            in_history = False
+
+        for line in content.splitlines():
+            heading = NAMESPACE_PATTERN.match(line.strip())
+            if heading:
+                flush()
+                namespace = heading.group(1).lower()
+                user = heading.group(2)
+                continue
+
+            if line.startswith("--------- "):
+                # dumpsys closes every section with a duration trailer.
+                flush()
+                namespace = None
+                continue
+
+            if namespace is None:
+                continue
+
+            if not line.strip():
+                # dumpsys prints a blank line after every namespace block and
+                # after a change history, and other dumps such as the
+                # generation registry follow the last block, so a blank line
+                # closes the record being read.
+                flush()
+                continue
+
+            if line.startswith("_id:"):
+                flush()
+                record_lines = [line]
+                continue
+
+            if not record_lines:
+                continue
+
+            stripped = line.strip()
+            if stripped.startswith("History ("):
+                in_history = True
+                continue
+
+            if in_history:
+                if stripped.startswith("time:"):
+                    history_lines.append(stripped)
+                elif stripped and history_lines:
+                    # A history entry can be wrapped over several lines.
+                    history_lines[-1] += " " + stripped
+                continue
+
+            # Anything else continues the value of the record being read.
+            record_lines.append(line)
+
+        flush()
+
+    @staticmethod
+    def _split_fields(text: str, keys: Sequence[str]) -> dict[str, str]:
+        """Split the `key:value` fields of one record.
+
+        Values are free-form and may contain spaces and newlines, so a field
+        runs up to the start of the next key which is actually present. Keys
+        dumpsys did not print are skipped.
+        """
+        fields: dict[str, str] = {}
+        key = keys[0]
+        if not text.startswith(f"{key}:"):
+            return fields
+
+        remainder = text[len(key) + 1 :]
+        for next_key in keys[1:]:
+            value, separator, rest = remainder.partition(f" {next_key}:")
+            if separator:
+                fields[key] = value
+                key, remainder = next_key, rest
+
+        fields[key] = remainder
+        return fields
+
+    @staticmethod
+    def _parse_section_end(content: str) -> Optional[datetime]:
+        """Return the time the settings section was dumped, if reported."""
+        match = SECTION_END_PATTERN.search(content)
+        if not match:
+            return None
+
+        try:
+            return datetime.strptime(match.group(1), "%Y-%m-%d %H:%M:%S")
+        except ValueError:
+            return None
+
+    @staticmethod
+    def _resolve_timestamp(
+        value: str, section_end: Optional[datetime]
+    ) -> Optional[str]:
+        """Add the missing year to a `MM-DD HH:MM:SS.mmm` history timestamp.
+
+        dumpsys prints the change history without a year, so it is resolved
+        against the time the section was dumped: the most recent matching date
+        at or before that time.
+        """
+        if section_end is None:
+            return None
+
+        try:
+            partial = datetime.strptime(value, "%m-%d %H:%M:%S.%f")
+            timestamp = partial.replace(year=section_end.year)
+            if timestamp > section_end:
+                timestamp = partial.replace(year=section_end.year - 1)
+        except ValueError:
+            return None
+
+        return convert_datetime_to_iso(timestamp)
+
+    def _parse_history(
+        self, line: str, section_end: Optional[datetime]
+    ) -> ModuleAtomicResult:
+        fields = self._split_fields(line, HISTORY_FIELDS)
+        return {
+            "timestamp": self._resolve_timestamp(fields.get("time", ""), section_end),
+            "oldValue": fields.get("oldValue"),
+            "newValue": fields.get("newValue"),
+            "pkg": fields.get("package"),
+        }
+
+    def _build_record(
+        self,
+        namespace: Optional[str],
+        user: Optional[str],
+        record_lines: list[str],
+        history_lines: list[str],
+        section_end: Optional[datetime],
+    ) -> ModuleAtomicResult:
+        text = "\n".join(record_lines).rstrip()
+        # The bare `notPreservedInRestore` token has no `key:` shape and is
+        # printed last, so peel it off before splitting the fields.
+        head = text.removesuffix(" notPreservedInRestore")
+
+        record: ModuleAtomicResult = {"namespace": namespace, "user": user}
+        record.update(self._split_fields(head, SETTING_FIELDS))
+        if head != text:
+            record["isValuePreservedInRestore"] = "false"
+
+        record["history"] = [
+            self._parse_history(entry, section_end) for entry in history_lines
+        ]
+        return record

src/mvt/android/modules/androidqf/aqf_settings.py

@@ -3,11 +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 logging
-from typing import Optional
-
 from mvt.android.artifacts.settings import Settings as SettingsArtifact
-from mvt.common.module_types import ModuleResults
 
 from .base import AndroidQFModule
 
@@ -15,43 +11,23 @@ from .base import AndroidQFModule
 class AQFSettings(SettingsArtifact, AndroidQFModule):
     """This module analyse setting files"""
 
-    def __init__(
-        self,
-        file_path: Optional[str] = None,
-        target_path: Optional[str] = None,
-        results_path: Optional[str] = None,
-        module_options: Optional[dict] = None,
-        log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[ModuleResults] = None,
-    ) -> None:
-        super().__init__(
-            file_path=file_path,
-            target_path=target_path,
-            results_path=results_path,
-            module_options=module_options,
-            log=log,
-            results=results,
-        )
-        self.results: dict = results if results is not None else {}
-
     def run(self) -> None:
         for setting_file in self._get_files_by_pattern("*/settings_*.txt"):
             namespace = setting_file[setting_file.rfind("_") + 1 : -4]
 
-            self.results[namespace] = {}
             data = self._get_file_content(setting_file)
             for line in data.decode("utf-8").splitlines():
-                line = line.strip()
+                name, separator, value = line.strip().partition("=")
-                try:
+                if not separator:
-                    key, value = line.split("=", 1)
-                except ValueError:
                     continue
 
-                try:
+                self.results.append(
-                    self.results[namespace][key] = value
+                    {
-                except IndexError:
+                        "namespace": namespace,
-                    continue
+                        "user": None,
+                        "name": name,
+                        "value": value,
+                    }
+                )
 
-        self.log.info(
+        self.log.info("Identified %d settings", len(self.results))
-            "Identified %d settings", sum([len(val) for val in self.results.values()])
-        )

src/mvt/android/modules/bugreport/settings.py

@@ -18,5 +18,4 @@ class Settings(SettingsArtifact, BugReportModule):
             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", len(self.results))
-        self.log.info("Identified %d Android settings", count)

src/mvt/common/config.py

@@ -59,13 +59,16 @@ class MVTSettings(BaseSettings):
         dotenv_settings: PydanticBaseSettingsSource,
         file_secret_settings: PydanticBaseSettingsSource,
     ) -> Tuple[PydanticBaseSettingsSource, ...]:
-        yaml_source = YamlConfigSettingsSource(settings_cls, MVT_CONFIG_PATH)
         sources: Tuple[PydanticBaseSettingsSource, ...] = (
-            yaml_source,
+            YamlConfigSettingsSource(settings_cls, MVT_CONFIG_PATH),
             init_settings,
         )
-        # Always load env variables by default
+        # Load env variables only when asked to. initialise() constructs the
-        sources = (env_settings,) + sources
+        # settings once without them so that what gets written back to
+        # config.yaml never includes values taken from the environment.
+        # init_settings() returns the keyword arguments passed to the constructor.
+        if init_settings().get("load_env", True):
+            sources = (env_settings,) + sources
         return sources
 
     def save_settings(
@@ -92,7 +95,7 @@ class MVTSettings(BaseSettings):
 
         Afterwards we load the settings again, this time including the env variables.
         """
-        # Set invalid env prefix to avoid loading env variables.
+        # Construct the settings without env variables so they are not persisted.
         settings = cls(load_env=False)
         settings.save_settings()
 

src/mvt/common/utils.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 base64
 import cProfile
 import datetime
 import hashlib
@@ -184,6 +185,29 @@ def keys_bytes_to_string(obj: Any) -> Any:
     return new_obj
 
 
+def sanitize_json_data(obj: Any) -> Any:
+    """Recursively convert values to JSON-compatible representations.
+
+    SQLite and plist records can contain binary and date values nested inside
+    dictionaries. Preserve binary data as base64 rather than discarding it or
+    relying on a lossy text decoding during JSON serialization.
+    """
+    if isinstance(obj, datetime.datetime):
+        return convert_datetime_to_iso(obj)
+    if isinstance(obj, datetime.date):
+        return obj.isoformat()
+    if isinstance(obj, bytes):
+        return base64.b64encode(obj).decode("ascii")
+    if isinstance(obj, dict):
+        return {
+            sanitize_json_data(key): sanitize_json_data(value)
+            for key, value in obj.items()
+        }
+    if isinstance(obj, (tuple, list, set)):
+        return [sanitize_json_data(value) for value in obj]
+    return obj
+
+
 def get_sha256_from_file_path(file_path: str) -> str:
     """Calculate the SHA256 hash of a file from a file path.
 

src/mvt/ios/modules/mixed/calendar.py

@@ -11,7 +11,7 @@ from mvt.common.module_types import (
     ModuleResults,
     ModuleSerializedResult,
 )
-from mvt.common.utils import convert_mactime_to_iso
+from mvt.common.utils import convert_mactime_to_iso, sanitize_json_data
 
 from ..base import IOSExtraction
 
@@ -90,54 +90,63 @@ class Calendar(IOSExtraction):
         """
         conn = self._open_sqlite_db(self.file_path)
         cur = conn.cursor()
-
+        try:
-        cur.execute(
+            cur.execute("PRAGMA table_info(CalendarItem);")
-            """
+            calendar_columns = [row[1] for row in cur]
-        SELECT
+            cur.execute("PRAGMA table_info(Participant);")
-            CalendarItem.ROWID as "id",
+            participant_columns = [row[1] for row in cur]
-            CalendarItem.summary as "summary",
+            selected = [
-            CalendarItem.description as "description",
+                f'c."{column}" AS "calendar_{column}"' for column in calendar_columns
-            CalendarItem.start_date as "start_date",
+            ] + [
-            CalendarItem.end_date as "end_date",
+                f'p."{column}" AS "participant_{column}"'
-            CalendarItem.all_day as "all_day",
+                for column in participant_columns
-            CalendarItem.calendar_id as "calendar_id",
+            ]
-            CalendarItem.organizer_id as "organizer_id",
+            cur.execute(
-            CalendarItem.url as "url",
+                f"SELECT {', '.join(selected)} FROM CalendarItem c "
-            CalendarItem.last_modified as "last_modified",
+                "LEFT JOIN Participant p ON p.ROWID = c.organizer_id;"
-            CalendarItem.external_id as "external_id",
+            )
-            CalendarItem.external_mod_tag as "external_mod_tag",
+            names = [description[0] for description in cur.description]
-            CalendarItem.unique_identifier as "unique_identifier",
+            for row in cur:
-            CalendarItem.hidden as "hidden",
+                raw = sanitize_json_data(dict(zip(names, row)))
-            CalendarItem.UUID as "uuid",
+                entry = {
-            CalendarItem.creation_date as "creation_date",
+                    "id": raw.get("calendar_ROWID"),
-            CalendarItem.action as "action",
+                    "summary": raw.get("calendar_summary"),
-            CalendarItem.created_by_id as "created_by_id",
+                    "description": raw.get("calendar_description"),
-            Participant.UUID as "participant_uuid",
+                    "all_day": raw.get("calendar_all_day"),
-            Participant.email as "participant_email",
+                    "calendar_id": raw.get("calendar_calendar_id"),
-            Participant.phone_number as "participant_phone",
+                    "organizer_id": raw.get("calendar_organizer_id"),
-            Participant.comment as "participant_comment",
+                    "url": raw.get("calendar_url"),
-            Participant.last_modified as "participant_last_modified"
+                    "external_id": raw.get("calendar_external_id"),
-        FROM CalendarItem
+                    "external_mod_tag": raw.get("calendar_external_mod_tag"),
-            LEFT JOIN Participant ON Participant.ROWID = CalendarItem.organizer_id;
+                    "unique_identifier": raw.get("calendar_unique_identifier"),
-        """
+                    "hidden": raw.get("calendar_hidden"),
-        )
+                    "uuid": raw.get("calendar_UUID"),
-
+                    "action": raw.get("calendar_action"),
-        names = [description[0] for description in cur.description]
+                    "created_by_id": raw.get("calendar_created_by_id"),
-        for item in cur:
+                    "participant_uuid": raw.get("participant_UUID"),
-            entry = {}
+                    "participant_email": raw.get("participant_email"),
-            for index, value in enumerate(item):
+                    "participant_phone": raw.get("participant_phone_number"),
-                if names[index] in self.timestamps:
+                    "participant_comment": raw.get("participant_comment"),
-                    if value is None or isinstance(value, str):
+                    "record": raw,
-                        entry[names[index]] = value
+                }
-                    else:
+                timestamp_sources = {
-                        entry[names[index]] = convert_mactime_to_iso(value)
+                    "start_date": "calendar_start_date",
-                else:
+                    "end_date": "calendar_end_date",
-                    entry[names[index]] = value
+                    "last_modified": "calendar_last_modified",
-
+                    "creation_date": "calendar_creation_date",
-            self.results.append(entry)
+                    "participant_last_modified": "participant_last_modified",
-
+                }
-        cur.close()
+                for output, source in timestamp_sources.items():
-        conn.close()
+                    value = raw.get(source)
+                    entry[output] = (
+                        convert_mactime_to_iso(value)
+                        if value is not None and not isinstance(value, str)
+                        else value
+                    )
+                self.results.append(entry)
+        finally:
+            cur.close()
+            conn.close()
 
     def run(self) -> None:
         self._find_ios_database(

src/mvt/ios/modules/mixed/calls.py

@@ -7,7 +7,7 @@ import logging
 from typing import Optional
 
 from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
-from mvt.common.utils import convert_mactime_to_iso
+from mvt.common.utils import convert_mactime_to_iso, sanitize_json_data
 
 from ..base import IOSExtraction
 
@@ -57,31 +57,30 @@ class Calls(IOSExtraction):
             return
         conn = self._open_sqlite_db(self.file_path)
         cur = conn.cursor()
-        cur.execute(
+        try:
-            """
+            cur.execute("SELECT * FROM ZCALLRECORD;")
-            SELECT
+            names = [description[0] for description in cur.description]
-                ZDATE, ZDURATION, ZLOCATION, ZADDRESS, ZSERVICE_PROVIDER
+            for row in cur:
-            FROM ZCALLRECORD;
+                original = dict(zip(names, row))
-        """
+                address = original.get("ZADDRESS")
-        )
+                if isinstance(address, bytes):
-        # names = [description[0] for description in cur.description]
+                    address = address.decode("utf-8", errors="replace")
-
+                raw = sanitize_json_data(original)
-        for row in cur:
+                date = raw.get("ZDATE")
-            self.results.append(
+                self.results.append(
-                {
+                    {
-                    "isodate": convert_mactime_to_iso(row[0]),
+                        "isodate": (
-                    "duration": row[1],
+                            convert_mactime_to_iso(date) if date is not None else ""
-                    "location": row[2],
+                        ),
-                    "number": (
+                        "duration": raw.get("ZDURATION"),
-                        row[3].decode("utf-8")
+                        "location": raw.get("ZLOCATION"),
-                        if isinstance(row[3], bytes)
+                        "number": address,
-                        else row[3]
+                        "provider": raw.get("ZSERVICE_PROVIDER"),
-                    ),
+                        "call": raw,
-                    "provider": row[4],
+                    }
-                }
+                )
-            )
+        finally:
-
+            cur.close()
-        cur.close()
+            conn.close()
-        conn.close()
 
         self.log.info("Extracted a total of %d calls", len(self.results))

src/mvt/ios/modules/mixed/contacts.py

@@ -8,6 +8,7 @@ import sqlite3
 from typing import Optional
 
 from mvt.common.module_types import ModuleResults
+from mvt.common.utils import sanitize_json_data
 
 from ..base import IOSExtraction
 
@@ -51,28 +52,45 @@ class Contacts(IOSExtraction):
         conn = self._open_sqlite_db(self.file_path)
         cur = conn.cursor()
         try:
-            try:
+            cur.execute("PRAGMA table_info(ABPerson);")
-                cur.execute(
+            person_columns = [row[1] for row in cur]
-                    """
+            cur.execute("PRAGMA table_info(ABMultiValue);")
-                    SELECT
+            multivalue_columns = [row[1] for row in cur]
-                        multi.value, person.first, person.middle, person.last,
+            if not person_columns or not multivalue_columns:
-                        person.organization
+                return
-                    FROM ABPerson person, ABMultiValue multi
+
-                    WHERE person.rowid = multi.record_id and multi.value not null
+            select_columns = [
-                    ORDER by person.rowid ASC;
+                f'p."{column}" AS "person_{column}"' for column in person_columns
-                """
+            ] + [
-                )
+                f'm."{column}" AS "multivalue_{column}"'
-            except sqlite3.OperationalError as e:
+                for column in multivalue_columns
-                self.log.info("Error while reading the contact table: %s", e)
+            ]
-                return None
+            cur.execute(
+                f"SELECT {', '.join(select_columns)} "
+                "FROM ABPerson p LEFT JOIN ABMultiValue m "
+                "ON p.ROWID = m.record_id ORDER BY p.ROWID, m.ROWID;"
+            )
             names = [description[0] for description in cur.description]
-
             for row in cur:
-                new_contact = {}
+                raw = sanitize_json_data(dict(zip(names, row)))
-                for index, value in enumerate(row):
+                self.results.append(
-                    new_contact[names[index]] = value
+                    {
-
+                        "value": raw.get("multivalue_value"),
-                self.results.append(new_contact)
+                        "first": raw.get("person_First"),
+                        "middle": raw.get("person_Middle"),
+                        "last": raw.get("person_Last"),
+                        "organization": raw.get("person_Organization"),
+                        # Keep the capitalization returned by the previous
+                        # query for consumers of existing JSON output.
+                        "First": raw.get("person_First"),
+                        "Middle": raw.get("person_Middle"),
+                        "Last": raw.get("person_Last"),
+                        "Organization": raw.get("person_Organization"),
+                        "contact": raw,
+                    }
+                )
+        except sqlite3.OperationalError as exc:
+            self.log.info("Error while reading the contact table: %s", exc)
         finally:
             cur.close()
             conn.close()

src/mvt/ios/modules/mixed/interactionc.py

@@ -14,7 +14,7 @@ from mvt.common.module_types import (
     ModuleResults,
     ModuleSerializedResult,
 )
-from mvt.common.utils import convert_mactime_to_iso
+from mvt.common.utils import convert_mactime_to_iso, sanitize_json_data
 
 from ..base import IOSExtraction
 from .whatsapp_contacts import WhatsappContacts
@@ -547,4 +547,22 @@ class InteractionC(IOSExtraction):
 
         self._postprocess_results()
 
+        # Preserve every column from the underlying interaction row. The
+        # joined compatibility queries above intentionally expose a stable
+        # result schema but otherwise discard useful forensic fields.
+        conn = self._open_sqlite_db(self.file_path)
+        cur = conn.cursor()
+        try:
+            cur.execute("SELECT * FROM ZINTERACTIONS;")
+            names = [description[0] for description in cur.description]
+            interactions = {}
+            for row in cur:
+                raw = sanitize_json_data(dict(zip(names, row)))
+                interactions[raw.get("Z_PK")] = raw
+        finally:
+            cur.close()
+            conn.close()
+        for result in self.results:
+            result["interaction"] = interactions.get(result.get("table_id"), {})
+
         self.log.info("Extracted a total of %d InteractionC events", len(self.results))

src/mvt/ios/modules/mixed/net_datausage.py

@@ -7,6 +7,8 @@ import logging
 from typing import Optional
 
 from mvt.common.module_types import ModuleResults
+from mvt.common.utils import convert_mactime_to_iso, sanitize_json_data
+
 from ..net_base import NetBase
 
 DATAUSAGE_BACKUP_IDS = [
@@ -42,6 +44,77 @@ class Datausage(NetBase):
             results=results,
         )
 
+    def _extract_net_data(self):
+        assert self.file_path is not None
+        conn = self._open_sqlite_db(self.file_path)
+        cur = conn.cursor()
+        try:
+            cur.execute("PRAGMA table_info(ZPROCESS);")
+            process_columns = [row[1] for row in cur]
+            cur.execute("PRAGMA table_info(ZLIVEUSAGE);")
+            live_columns = [row[1] for row in cur]
+            selected = [
+                f'p."{column}" AS "process_{column}"' for column in process_columns
+            ] + [f'l."{column}" AS "live_{column}"' for column in live_columns]
+            cur.execute(
+                f"SELECT {', '.join(selected)} FROM ZLIVEUSAGE l "
+                "LEFT JOIN ZPROCESS p ON l.ZHASPROCESS = p.Z_PK;"
+            )
+            names = [description[0] for description in cur.description]
+            rows = [dict(zip(names, row)) for row in cur]
+
+            process_only = [
+                f'p."{column}" AS "process_{column}"' for column in process_columns
+            ] + [f'NULL AS "live_{column}"' for column in live_columns]
+            cur.execute(
+                f"SELECT {', '.join(process_only)} FROM ZPROCESS p "
+                "WHERE NOT EXISTS ("
+                "SELECT 1 FROM ZLIVEUSAGE l WHERE l.ZHASPROCESS = p.Z_PK);"
+            )
+            names = [description[0] for description in cur.description]
+            rows.extend(dict(zip(names, row)) for row in cur)
+        finally:
+            cur.close()
+            conn.close()
+
+        for raw_row in rows:
+            first_timestamp = raw_row.get("process_ZFIRSTTIMESTAMP")
+            process_timestamp = raw_row.get("process_ZTIMESTAMP")
+            live_timestamp = raw_row.get("live_ZTIMESTAMP")
+            first_isodate = (
+                convert_mactime_to_iso(first_timestamp)
+                if first_timestamp
+                else first_timestamp
+            )
+            isodate = (
+                convert_mactime_to_iso(process_timestamp)
+                if process_timestamp
+                else process_timestamp
+            )
+            live_isodate = (
+                convert_mactime_to_iso(live_timestamp)
+                if live_timestamp
+                else first_isodate
+            )
+            self.results.append(
+                {
+                    "first_isodate": first_isodate,
+                    "isodate": isodate,
+                    "proc_name": raw_row.get("process_ZPROCNAME"),
+                    "bundle_id": raw_row.get("process_ZBUNDLENAME"),
+                    "proc_id": raw_row.get("process_Z_PK"),
+                    "wifi_in": raw_row.get("live_ZWIFIIN"),
+                    "wifi_out": raw_row.get("live_ZWIFIOUT"),
+                    "wwan_in": raw_row.get("live_ZWWANIN"),
+                    "wwan_out": raw_row.get("live_ZWWANOUT"),
+                    "live_id": raw_row.get("live_Z_PK"),
+                    "live_proc_id": raw_row.get("live_ZHASPROCESS"),
+                    "live_isodate": live_isodate,
+                    "record": sanitize_json_data(raw_row),
+                }
+            )
+        self.log.info("Extracted information on %d processes", len(self.results))
+
     def run(self) -> None:
         self._find_ios_database(
             backup_ids=DATAUSAGE_BACKUP_IDS, root_paths=DATAUSAGE_ROOT_PATHS

src/mvt/ios/modules/mixed/safari_browserstate.py

@@ -15,7 +15,11 @@ from mvt.common.module_types import (
     ModuleResults,
     ModuleSerializedResult,
 )
-from mvt.common.utils import convert_mactime_to_iso, keys_bytes_to_string
+from mvt.common.utils import (
+    convert_mactime_to_iso,
+    keys_bytes_to_string,
+    sanitize_json_data,
+)
 
 from ..base import IOSExtraction
 
@@ -94,85 +98,76 @@ class SafariBrowserState(IOSExtraction):
     def _process_browser_state_db(self, db_path):
         self._recover_sqlite_db_if_needed(db_path)
         conn = self._open_sqlite_db(db_path)
-
         cur = conn.cursor()
         try:
             try:
                 cur.execute(
                     """
-                    SELECT
+                    SELECT tabs.*, tab_sessions.session_data AS session_data
-                        tabs.title,
-                        tabs.url,
-                        tabs.user_visible_url,
-                        tabs.last_viewed_time,
-                        tab_sessions.session_data
                     FROM tabs
-                    JOIN tab_sessions ON tabs.uuid = tab_sessions.tab_uuid
+                    LEFT JOIN tab_sessions ON tabs.uuid = tab_sessions.tab_uuid
                     ORDER BY tabs.last_viewed_time;
-                """
+                    """
                 )
+                names = [description[0] for description in cur.description]
             except sqlite3.OperationalError:
-                # Old version iOS <12 likely
+                # Older databases store session_data directly in tabs.
                 try:
-                    cur.execute(
+                    cur.execute("SELECT * FROM tabs ORDER BY last_viewed_time;")
-                        """
+                    names = [description[0] for description in cur.description]
-                        SELECT
+                except sqlite3.OperationalError as exc:
-                            title, url, user_visible_url, last_viewed_time, session_data
+                    self.log.error("Error executing query: %s", exc)
-                    FROM tabs
-                    ORDER BY last_viewed_time;
-                """
-                    )
-                except sqlite3.OperationalError as e:
-                    self.log.error(f"Error executing query: {e}")
                     return
 
             for row in cur:
+                raw = dict(zip(names, row))
                 session_entries = []
-
+                session_blob = raw.get("session_data")
-                if row[4]:
+                if session_blob:
                     # Skip a 4 byte header before the plist content.
-                    session_plist = row[4][4:]
+                    session_plist = session_blob[4:]
                     session_data = {}
                     try:
                         session_data = plistlib.load(io.BytesIO(session_plist))
                         session_data = keys_bytes_to_string(session_data)
-                    except plistlib.InvalidFileException:
+                    except (plistlib.InvalidFileException, ValueError, TypeError):
                         pass
 
-                    if "SessionHistoryEntries" in session_data.get("SessionHistory", {}):
+                    for session_entry in session_data.get("SessionHistory", {}).get(
-                        for session_entry in session_data["SessionHistory"].get(
+                        "SessionHistoryEntries", []
-                            "SessionHistoryEntries"
+                    ):
-                        ):
+                        self._session_history_count += 1
-                            self._session_history_count += 1
+                        entry_data = session_entry.get("SessionHistoryEntryData")
-
+                        session_entries.append(
-                            data_length = 0
+                            {
-                            if "SessionHistoryEntryData" in session_entry:
+                                "entry_title": session_entry.get(
-                                data_length = len(
+                                    "SessionHistoryEntryOriginalURL"
-                                    session_entry.get("SessionHistoryEntryData")
+                                ),
-                                )
+                                "entry_url": session_entry.get(
-
+                                    "SessionHistoryEntryURL"
-                            session_entries.append(
+                                ),
-                                {
+                                "data_length": (
-                                    "entry_title": session_entry.get(
+                                    len(entry_data) if entry_data is not None else 0
-                                        "SessionHistoryEntryOriginalURL"
+                                ),
-                                    ),
+                            }
-                                    "entry_url": session_entry.get(
+                        )
-                                        "SessionHistoryEntryURL"
-                                    ),
-                                    "data_length": data_length,
-                                }
-                            )
 
+                last_viewed = raw.get("last_viewed_time")
                 self.results.append(
                     {
-                        "tab_title": row[0],
+                        "tab_title": raw.get("title"),
-                        "tab_url": row[1],
+                        "tab_url": raw.get("url"),
-                        "tab_visible_url": row[2],
+                        "tab_visible_url": raw.get("user_visible_url"),
-                        "last_viewed_timestamp": convert_mactime_to_iso(row[3]),
+                        "last_viewed_timestamp": (
+                            convert_mactime_to_iso(last_viewed)
+                            if last_viewed is not None
+                            else ""
+                        ),
                         "session_data": session_entries,
                         "safari_browser_state_db": os.path.relpath(
                             db_path, self.target_path
                         ),
+                        "tab": sanitize_json_data(raw),
                     }
                 )
         finally:

src/mvt/ios/modules/mixed/safari_history.py

@@ -13,7 +13,11 @@ from mvt.common.module_types import (
     ModuleSerializedResult,
 )
 from mvt.common.url import URL
-from mvt.common.utils import convert_mactime_to_datetime, convert_mactime_to_iso
+from mvt.common.utils import (
+    convert_mactime_to_datetime,
+    convert_mactime_to_iso,
+    sanitize_json_data,
+)
 
 from ..base import IOSExtraction
 
@@ -133,39 +137,41 @@ class SafariHistory(IOSExtraction):
         self._recover_sqlite_db_if_needed(history_path)
         conn = self._open_sqlite_db(history_path)
         cur = conn.cursor()
-        cur.execute(
+        try:
-            """
+            cur.execute("PRAGMA table_info(history_items);")
-            SELECT
+            item_columns = [row[1] for row in cur]
-                history_items.id,
+            cur.execute("PRAGMA table_info(history_visits);")
-                history_items.url,
+            visit_columns = [row[1] for row in cur]
-                history_visits.id,
+            selected = [
-                history_visits.visit_time,
+                f'i."{column}" AS "item_{column}"' for column in item_columns
-                history_visits.redirect_source,
+            ] + [f'v."{column}" AS "visit_{column}"' for column in visit_columns]
-                history_visits.redirect_destination
+            cur.execute(
-            FROM history_items
+                f"SELECT {', '.join(selected)} FROM history_items i "
-            JOIN history_visits ON history_visits.history_item = history_items.id
+                "JOIN history_visits v ON v.history_item = i.id "
-            ORDER BY history_visits.visit_time;
+                "ORDER BY v.visit_time;"
-        """
-        )
-
-        for row in cur:
-            self.results.append(
-                {
-                    "id": row[0],
-                    "url": row[1],
-                    "visit_id": row[2],
-                    "timestamp": row[3],
-                    "isodate": convert_mactime_to_iso(row[3]),
-                    "redirect_source": row[4],
-                    "redirect_destination": row[5],
-                    "safari_history_db": os.path.relpath(
-                        history_path, self.target_path
-                    ),
-                }
             )
-
+            names = [description[0] for description in cur.description]
-        cur.close()
+            for row in cur:
-        conn.close()
+                raw = sanitize_json_data(dict(zip(names, row)))
+                timestamp = raw.get("visit_visit_time")
+                self.results.append(
+                    {
+                        "id": raw.get("item_id"),
+                        "url": raw.get("item_url"),
+                        "visit_id": raw.get("visit_id"),
+                        "timestamp": timestamp,
+                        "isodate": convert_mactime_to_iso(timestamp),
+                        "redirect_source": raw.get("visit_redirect_source"),
+                        "redirect_destination": raw.get("visit_redirect_destination"),
+                        "safari_history_db": os.path.relpath(
+                            history_path, self.target_path
+                        ),
+                        "record": raw,
+                    }
+                )
+        finally:
+            cur.close()
+            conn.close()
 
     def run(self) -> None:
         if self.is_backup:

src/mvt/ios/modules/mixed/tcc.py

@@ -12,7 +12,7 @@ from mvt.common.module_types import (
     ModuleResults,
     ModuleSerializedResult,
 )
-from mvt.common.utils import convert_unix_to_iso
+from mvt.common.utils import convert_unix_to_iso, sanitize_json_data
 
 from ..base import IOSExtraction
 
@@ -103,126 +103,49 @@ class TCC(IOSExtraction):
     def process_db(self, file_path):
         conn = self._open_sqlite_db(file_path)
         cur = conn.cursor()
-        db_version = "v3"
         try:
-            cur.execute(
-                """SELECT
-                service, client, client_type, auth_value,
-                auth_reason, last_modified
-            FROM access;"""
-            )
-        except sqlite3.OperationalError:
-            # v2 version
             try:
-                cur.execute(
+                cur.execute("SELECT * FROM access;")
-                    """SELECT
+            except sqlite3.OperationalError as exc:
-                    service, client, client_type, allowed,
+                self.log.error("Error parsing TCC database: %s", exc)
-                    prompt_count, last_modified
+                return
-                    FROM access;"""
+
+            names = [description[0] for description in cur.description]
+            for row in cur:
+                record = sanitize_json_data(dict(zip(names, row)))
+                client_type = record.get("client_type")
+                record["client_type_value"] = client_type
+                record["client_type"] = (
+                    "bundle_id" if client_type == 0 else "absolute_path"
                 )
-                db_version = "v2"
-            except sqlite3.OperationalError:
-                try:
-                    cur.execute(
-                        """SELECT
-                        service, client, client_type, allowed,
-                        prompt_count
-                        FROM access;"""
-                    )
-                    db_version = "v1"
-                except sqlite3.OperationalError as e:
-                    self.log.error(f"Error parsing TCC database: {e}")
 
-        for row in cur:
+                if "auth_value" in record:
-            service = row[0]
+                    auth_value = record["auth_value"]
-            client = row[1]
+                    record["auth_value_value"] = auth_value
-            client_type = row[2]
+                    record["auth_value"] = AUTH_VALUES.get(auth_value, "unknown")
-            client_type_desc = "bundle_id" if client_type == 0 else "absolute_path"
+                    auth_reason = record.get("auth_reason")
-            if db_version == "v3":
+                    record["auth_reason_desc"] = AUTH_REASONS.get(
-                auth_value = row[3]
+                        auth_reason, "unknown"
-                auth_value_desc = AUTH_VALUES.get(auth_value, "")
-                auth_reason = row[4]
-                auth_reason_desc = AUTH_REASONS.get(auth_reason, "unknown")
-                last_modified = convert_unix_to_iso(row[5])
-
-                if service in ["kTCCServiceMicrophone", "kTCCServiceCamera"]:
-                    device = (
-                        "microphone" if service == "kTCCServiceMicrophone" else "camera"
                     )
-                    self.log.info(
+                elif "allowed" in record:
-                        'Found client "%s" with access %s to %s on %s by %s',
+                    allowed = record["allowed"]
-                        client,
+                    record["allowed_value"] = AUTH_VALUE_OLD.get(allowed, "unknown")
-                        auth_value_desc,
+
-                        device,
+                if record.get("last_modified") is not None:
-                        last_modified,
+                    record["last_modified_value"] = record["last_modified"]
-                        auth_reason_desc,
+                    record["last_modified"] = convert_unix_to_iso(
+                        record["last_modified"]
+                    )
+                if record.get("last_reminded") is not None:
+                    record["last_reminded_value"] = record["last_reminded"]
+                    record["last_reminded"] = convert_unix_to_iso(
+                        record["last_reminded"]
                     )
 
-                self.results.append(
+                self.results.append(record)
-                    {
+        finally:
-                        "service": service,
+            cur.close()
-                        "client": client,
+            conn.close()
-                        "client_type": client_type_desc,
-                        "auth_value": auth_value_desc,
-                        "auth_reason_desc": auth_reason_desc,
-                        "last_modified": last_modified,
-                    }
-                )
-            else:
-                allowed_value = row[3]
-                allowed_desc = AUTH_VALUE_OLD.get(allowed_value, "")
-                prompt_count = row[4]
-
-                if db_version == "v2":
-                    last_modified = convert_unix_to_iso(row[5])
-                    if service in ["kTCCServiceMicrophone", "kTCCServiceCamera"]:
-                        device = "camera"
-                        if service == "kTCCServiceMicrophone":
-                            device = "microphone"
-
-                        self.log.info(
-                            'Found client "%s" with access %s to %s at %s',
-                            client,
-                            allowed_desc,
-                            device,
-                            last_modified,
-                        )
-
-                    self.results.append(
-                        {
-                            "service": service,
-                            "client": client,
-                            "client_type": client_type_desc,
-                            "allowed_value": allowed_desc,
-                            "prompt_count": prompt_count,
-                            "last_modified": last_modified,
-                        }
-                    )
-                else:
-                    if service in ["kTCCServiceMicrophone", "kTCCServiceCamera"]:
-                        device = "camera"
-                        if service == "kTCCServiceMicrophone":
-                            device = "microphone"
-
-                        self.log.info(
-                            'Found client "%s" with access %s to %s',
-                            client,
-                            allowed_desc,
-                            device,
-                        )
-
-                    self.results.append(
-                        {
-                            "service": service,
-                            "client": client,
-                            "client_type": client_type_desc,
-                            "allowed_value": allowed_desc,
-                            "prompt_count": prompt_count,
-                        }
-                    )
-
-        cur.close()
-        conn.close()
 
     def run(self) -> None:
         self._find_ios_database(backup_ids=TCC_BACKUP_IDS, root_paths=TCC_ROOT_PATHS)

src/mvt/ios/modules/mixed/webkit_resource_load_statistics.py

@@ -13,7 +13,7 @@ from mvt.common.module_types import (
     ModuleResults,
     ModuleSerializedResult,
 )
-from mvt.common.utils import convert_unix_to_iso
+from mvt.common.utils import convert_unix_to_iso, sanitize_json_data
 
 from ..base import IOSExtraction
 
@@ -85,52 +85,45 @@ class WebkitResourceLoadStatistics(IOSExtraction):
 
         try:
             try:
-                cur.execute("PRAGMA table_info(ObservedDomains);")
+                cur.execute("SELECT * FROM ObservedDomains;")
-                available_columns = {row[1] for row in cur}
-                required_columns = [
-                    "domainID",
-                    "registrableDomain",
-                    "lastSeen",
-                    "hadUserInteraction",
-                ]
-                if not set(required_columns).issubset(available_columns):
-                    return
-
-                optional_columns = [
-                    column
-                    for column in [
-                        "mostRecentUserInteractionTime",
-                        "mostRecentWebPushInteractionTime",
-                    ]
-                    if column in available_columns
-                ]
-                selected_columns = required_columns + optional_columns
-                cur.execute(
-                    f"SELECT {', '.join(selected_columns)} FROM ObservedDomains;"
-                )
             except sqlite3.OperationalError:
                 return
 
+            names = [description[0] for description in cur.description]
+            required_columns = {
+                "domainID",
+                "registrableDomain",
+                "lastSeen",
+                "hadUserInteraction",
+            }
+            if not required_columns.issubset(names):
+                return
             for row in cur:
+                raw = sanitize_json_data(dict(zip(names, row)))
+                last_seen = raw.get("lastSeen")
                 result = {
-                    "domain_id": row[0],
+                    "domain_id": raw.get("domainID"),
-                    "registrable_domain": row[1],
+                    "registrable_domain": raw.get("registrableDomain"),
-                    "last_seen": row[2],
+                    "last_seen": last_seen,
-                    "had_user_interaction": bool(row[3]),
+                    "had_user_interaction": bool(raw.get("hadUserInteraction")),
-                    "last_seen_isodate": convert_unix_to_iso(row[2]),
+                    "last_seen_isodate": convert_unix_to_iso(last_seen),
                     "domain": domain,
                     "path": path,
+                    "record": raw,
                 }
-                for index, column in enumerate(optional_columns, start=4):
+                for column, field in (
-                    field = {
+                    (
-                        "mostRecentUserInteractionTime": (
+                        "mostRecentUserInteractionTime",
-                            "most_recent_user_interaction_time"
+                        "most_recent_user_interaction_time",
-                        ),
+                    ),
-                        "mostRecentWebPushInteractionTime": (
+                    (
-                            "most_recent_web_push_interaction_time"
+                        "mostRecentWebPushInteractionTime",
-                        ),
+                        "most_recent_web_push_interaction_time",
-                    }[column]
+                    ),
-                    timestamp = row[index]
+                ):
+                    if column not in raw:
+                        continue
+                    timestamp = raw[column]
                     result[field] = timestamp
                     if timestamp is not None and timestamp >= 0:
                         result[f"{field}_isodate"] = convert_unix_to_iso(timestamp)

tests/android/test_artifact_settings.py

@@ -0,0 +1,170 @@
+# 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/
+
+from mvt.android.artifacts.settings import Settings
+
+from ..utils import get_artifact
+
+
+def parse_bugreport_settings() -> Settings:
+    settings = Settings()
+    with open(get_artifact("android_data/bugreport/dumpstate.txt")) as handle:
+        data = handle.read()
+
+    settings.parse(settings.extract_dumpsys_section(data, "DUMP OF SERVICE settings:"))
+    return settings
+
+
+def find(settings: Settings, name: str) -> list:
+    return [result for result in settings.results if result["name"] == name]
+
+
+class TestSettingsArtifact:
+    def test_parsing(self):
+        settings = parse_bugreport_settings()
+
+        assert len(settings.results) == 12
+        assert {result["namespace"] for result in settings.results} == {
+            "config",
+            "global",
+            "secure",
+        }
+        assert settings.results[0] == {
+            "namespace": "config",
+            "user": "0",
+            "_id": "682",
+            "name": "namespace_one/blocked_components",
+            "pkg": "com.example.services",
+            "value": (
+                "com.android.settings,com.android.vending,\n"
+                "com.example.dialer,\n"
+                "com.example.camera"
+            ),
+            "default": (
+                "com.android.settings,\n"
+                "        com.android.vending,\n"
+                "        com.example.dialer"
+            ),
+            "defaultSystemSet": "false",
+            "history": [],
+        }
+
+    def test_multiline_values_are_kept_whole(self):
+        settings = parse_bugreport_settings()
+
+        assert find(settings, "namespace_one/allowed_packages")[0]["value"] == (
+            "com.example.messaging,\ncom.example.chat"
+        )
+        assert find(settings, "widget_instance_data")[0]["value"] == (
+            '{\n  "version": 1,\n  "data": [\n    {\n      "number": 10000,\n'
+            '      "package_name": "com.example.widget"\n    }\n  ]\n}'
+        )
+
+    def test_trailing_default_is_not_part_of_the_value(self):
+        settings = parse_bugreport_settings()
+
+        record = find(settings, "namespace_one/streaming_blocked_components")[0]
+        assert record["value"] == "com.example.dialer,com.example.camera"
+        assert record["default"] == "com.android.settings,\n        com.android.vending"
+
+    def test_trailing_metadata_is_not_part_of_the_value(self):
+        settings = parse_bugreport_settings()
+
+        record = find(settings, "lock_screen_show_notifications")[0]
+        assert record["value"] == "1"
+        assert record["defaultSystemSet"] == "true"
+        assert record["isValuePreservedInRestore"] == "true"
+
+        # Without a default, the tag or the restore token follows the value.
+        record = find(settings, "accessibility_enabled")[1]
+        assert record["value"] == "0"
+        assert record["tag"] == "null"
+        assert "default" not in record
+
+        record = find(settings, "send_action_app_error")[0]
+        assert record["value"] == "1"
+        assert record["isValuePreservedInRestore"] == "false"
+
+    def test_dumps_after_the_last_block_are_not_part_of_the_last_row(self):
+        settings = parse_bugreport_settings()
+
+        assert settings.results[-1] == {
+            "namespace": "secure",
+            "user": "10",
+            "_id": "311",
+            "name": "accessibility_enabled",
+            "pkg": "android",
+            "value": "0",
+            "tag": "null",
+            "history": [],
+        }
+
+    def test_repeated_names_are_kept_as_separate_records(self):
+        settings = parse_bugreport_settings()
+
+        widgets = find(settings, "widget_instance_data")
+        assert [record["_id"] for record in widgets] == ["771", "41654"]
+
+        accessibility = find(settings, "accessibility_enabled")
+        assert [(record["user"], record["value"]) for record in accessibility] == [
+            ("0", "1"),
+            ("10", "0"),
+        ]
+
+    def test_setting_without_recording_package(self):
+        settings = parse_bugreport_settings()
+
+        record = find(settings, "hidden_api_blacklist_exemptions")[0]
+        assert "pkg" not in record
+        assert record["value"] == "{null}"
+
+    def test_history_timestamps_resolved_against_section_end(self):
+        settings = parse_bugreport_settings()
+
+        # The section was dumped on 2022-03-29, so an 11-02 entry belongs to
+        # the previous year and an 03-14 entry to the same year.
+        assert find(settings, "development_settings_enabled")[0]["history"] == [
+            {
+                "timestamp": "2021-11-02 11:21:22.212000",
+                "oldValue": "null",
+                "newValue": "1",
+                "pkg": "com.android.settings",
+            },
+            {
+                "timestamp": "2022-03-14 09:02:11.100000",
+                "oldValue": "1",
+                "newValue": "0",
+                "pkg": "com.example.updater",
+            },
+        ]
+
+    def test_history_without_a_section_end_has_no_timestamp(self):
+        settings = Settings()
+        settings.parse(
+            "SECURE SETTINGS (user 0)\n"
+            "_id:240 name:accessibility_enabled pkg:android value:1\n"
+            "\tHistory (accessibility_enabled)\n"
+            "\t\ttime:03-28 22:41:07.980 mode:update oldValue:0 newValue:1 "
+            "package:com.example.helper\n"
+        )
+
+        assert settings.results[0]["history"] == [
+            {
+                "timestamp": None,
+                "oldValue": "0",
+                "newValue": "1",
+                "pkg": "com.example.helper",
+            }
+        ]
+
+    def test_dangerous_setting_is_detected_with_the_changing_package(self):
+        settings = parse_bugreport_settings()
+        settings.check_indicators()
+
+        assert len(settings.alertstore.alerts) == 1
+        alert = settings.alertstore.alerts[0]
+        assert "accessibility_enabled = 1" in alert.message
+        assert alert.event_time == "2022-03-28 22:41:07.980000"
+        assert alert.event["history"][0]["pkg"] == "com.example.helper"

tests/android_androidqf/test_settings.py

@@ -6,27 +6,12 @@
 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)
@@ -34,7 +19,13 @@ class TestSettingsModule:
         parent_path = Path(data_path).absolute().parent.as_posix()
         m.from_dir(parent_path, files)
         run_module(m)
-        assert len(m.results) == 1
+        assert len(m.results) == 9
-        assert "random" in m.results.keys()
+        assert {result["namespace"] for result in m.results} == {"random"}
+        assert m.results[0] == {
+            "namespace": "random",
+            "user": None,
+            "name": "samsung_errorlog_agree",
+            "value": "0",
+        }
         assert len(m.alertstore.alerts) == 1
         assert "samsung_errorlog_agree" in m.alertstore.alerts[0].message

tests/android_bugreport/test_bugreport.py

@@ -10,6 +10,7 @@ from mvt.android.modules.bugreport.dumpsys_appops import DumpsysAppops
 from mvt.android.modules.bugreport.dumpsys_getprop import DumpsysGetProp
 from mvt.android.modules.bugreport.dumpsys_packages import DumpsysPackages
 from mvt.android.modules.bugreport.dumpsys_receivers import DumpsysReceivers
+from mvt.android.modules.bugreport.settings import Settings
 from mvt.android.modules.bugreport.tombstones import Tombstones
 from mvt.common.module import run_module
 
@@ -93,6 +94,25 @@ class TestBugreportAnalysis:
         assert alert.event == malicious_receiver
         assert alert.matched_indicator.value == "com.android.services"
 
+    def test_settings_module(self):
+        m = self.launch_bug_report_module(Settings)
+        assert len(m.results) == 12
+
+        assert len(m.alertstore.alerts) == 1
+        assert "accessibility_enabled = 1" in m.alertstore.alerts[0].message
+
+        assert len(m.timeline) == 3
+        change = [
+            entry
+            for entry in m.timeline
+            if entry["timestamp"] == "2022-03-28 22:41:07.980000"
+        ][0]
+        assert change["event"] == "settings_change"
+        assert change["data"] == (
+            'secure setting "accessibility_enabled" changed from "0" to "1" '
+            "by com.example.helper"
+        )
+
     def test_tombstones_modules(self):
         m = self.launch_bug_report_module(Tombstones)
         assert len(m.results) == 2

tests/artifacts/android_data/bugreport/dumpstate.txt

@@ -264,5 +264,53 @@ ChangeId(143539591; name=SELINUX_LATEST_CHANGES; disabled)
 ChangeId(247079863; name=DISALLOW_INVALID_GROUP_REFERENCE; enableSinceTargetSdk=34)
 ChangeId(174227820; name=FORCE_DISABLE_HEVC_SUPPORT; disabled)
 ChangeId(168419799; name=DOWNSCALED; disabled; packageOverrides={com.google.android.apps.tachyon=false, org.torproject.torbrowser=false}; rawOverrides={org.torproject.torbrowser=false, org.article19.circulo.next=false}; overridable)
+-------------------------------------------------------------------------------
+DUMP OF SERVICE settings:
+CONFIG SETTINGS (user 0)
+_id:682 name:namespace_one/blocked_components pkg:com.example.services value:com.android.settings,com.android.vending,
+com.example.dialer,
+com.example.camera default:com.android.settings,
+        com.android.vending,
+        com.example.dialer defaultSystemSet:false
+_id:684 name:namespace_one/streaming_blocked_components pkg:com.example.services value:com.example.dialer,com.example.camera default:com.android.settings,
+        com.android.vending defaultSystemSet:false
+_id:680 name:namespace_one/allowed_packages pkg:com.example.services value:com.example.messaging,
+com.example.chat
 
+GLOBAL SETTINGS (user 0)
+_id:2070 name:adb_wifi_enabled pkg:android value:0 default:0 defaultSystemSet:true
+_id:778 name:hidden_api_blacklist_exemptions value:{null}
+_id:9640 name:send_action_app_error pkg:android value:1 notPreservedInRestore
+_id:9631 name:development_settings_enabled pkg:com.android.settings value:1 default:1 defaultSystemSet:true
+	History (development_settings_enabled)
+		time:11-02 11:21:22.212 mode:update oldValue:null newValue:1 package:com.android.settings
+		time:03-14 09:02:11.100 mode:update oldValue:1 newValue:0 package:com.example.updater
+_id:771 name:widget_instance_data pkg:com.android.systemui value:{
+  "version": 1,
+  "data": [
+    {
+      "number": 10000,
+      "package_name": "com.example.widget"
+    }
+  ]
+} defaultSystemSet:true
+_id:41654 name:widget_instance_data pkg:com.android.systemui value:{
+  "version": 3,
+  "data": []
+} defaultSystemSet:true
 
+SECURE SETTINGS (user 0)
+_id:907 name:lock_screen_show_notifications pkg:com.android.settings value:1 default:1 defaultSystemSet:true isValuePreservedInRestore:true
+_id:240 name:accessibility_enabled pkg:android value:1 default:0 defaultSystemSet:true
+	History (accessibility_enabled)
+		time:03-28 22:41:07.980 mode:update oldValue:0 newValue:1 package:com.example.helper
+
+SECURE SETTINGS (user 10)
+_id:311 name:accessibility_enabled pkg:android value:0 tag:null
+
+GENERATION REGISTRY
+Maximum number of backing stores:8
+Number of backing stores:1
+_Backing store for type:SETTINGS_SECURE user:10 size:1024 cachedEntries:1
+
+--------- 0.019s was the duration of dumpsys settings, ending at: 2022-03-29 23:14:28

tests/common/test_config.py

@@ -0,0 +1,30 @@
+# 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 yaml
+
+from mvt.common import config
+from mvt.common.config import MVTSettings
+
+
+def test_env_variables_are_not_persisted_to_config_file(tmp_path, monkeypatch):
+    config_path = tmp_path / "config.yaml"
+    monkeypatch.setattr(config, "MVT_CONFIG_FOLDER", str(tmp_path))
+    monkeypatch.setattr(config, "MVT_CONFIG_PATH", str(config_path))
+    monkeypatch.setenv("MVT_NETWORK_ACCESS_ALLOWED", "false")
+    monkeypatch.setenv("MVT_IOS_BACKUP_PASSWORD", "env-only-password")
+
+    settings = MVTSettings.initialise()
+
+    assert os.path.isfile(config_path)
+    saved = yaml.safe_load(config_path.read_text()) or {}
+    assert "NETWORK_ACCESS_ALLOWED" not in saved
+    assert "IOS_BACKUP_PASSWORD" not in saved
+
+    # The environment must still apply to the settings in use.
+    assert settings.NETWORK_ACCESS_ALLOWED is False
+    assert settings.IOS_BACKUP_PASSWORD == "env-only-password"

tests/common/test_utils.py

@@ -18,6 +18,7 @@ from mvt.common.utils import (
     generate_hashes_from_path,
     get_sha256_from_file_path,
     init_logging,
+    sanitize_json_data,
     set_verbose_logging,
 )
 
@@ -108,6 +109,18 @@ class TestCustomJSONEncoder:
         )
 
 
+def test_sanitize_json_data_preserves_nested_binary_and_dates():
+    value = {
+        "blob": b"\x00\xff",
+        "nested": [datetime(2023, 11, 13, 12, 21, 49, 727467)],
+    }
+
+    assert sanitize_json_data(value) == {
+        "blob": "AP8=",
+        "nested": ["2023-11-13 12:21:49.727467"],
+    }
+
+
 class TestInitLogging:
     def test__init_logging_is_idempotent(self):
         # Loaded module packages may import an MVT CLI module, which calls

tests/ios_backup/test_calendar.py

@@ -21,6 +21,8 @@ class TestCalendarModule:
         assert len(m.timeline) == 4
         assert len(m.alertstore.alerts) == 0
         assert m.results[0]["summary"] == "Super interesting meeting"
+        assert m.results[0]["record"]["calendar_app_link"] is None
+        assert "participant_proposed_start_date" in m.results[0]["record"]
 
     def test_calendar_with_explicit_file_path(self):
         backup_path = get_ios_backup_folder()

tests/ios_backup/test_calls.py

@@ -0,0 +1,38 @@
+# 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 sqlite3
+
+from mvt.common.module import run_module
+from mvt.ios.modules.mixed.calls import Calls
+
+
+def test_calls_preserves_complete_source_record(tmp_path):
+    db_path = tmp_path / "CallHistory.storedata"
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        """
+        CREATE TABLE ZCALLRECORD (
+            ZDATE REAL, ZDURATION REAL, ZLOCATION TEXT, ZADDRESS BLOB,
+            ZSERVICE_PROVIDER TEXT, ZORIGINATED INTEGER, ZANSWERED INTEGER,
+            ZMETADATA BLOB
+        );
+        """
+    )
+    conn.execute(
+        "INSERT INTO ZCALLRECORD VALUES (?, ?, ?, ?, ?, ?, ?, ?);",
+        (700000000, 42, "Berlin", b"+491234", "carrier", 1, 0, b"\x00\xff"),
+    )
+    conn.commit()
+    conn.close()
+
+    module = Calls(file_path=str(db_path))
+    run_module(module)
+
+    assert len(module.results) == 1
+    assert module.results[0]["number"] == "+491234"
+    assert module.results[0]["call"]["ZORIGINATED"] == 1
+    assert module.results[0]["call"]["ZANSWERED"] == 0
+    assert module.results[0]["call"]["ZMETADATA"] == "AP8="

tests/ios_backup/test_contacts.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/
+
+import sqlite3
+
+from mvt.common.module import run_module
+from mvt.ios.modules.mixed.contacts import Contacts
+
+
+def test_contacts_preserves_complete_rows_and_people_without_values(tmp_path):
+    db_path = tmp_path / "AddressBook.sqlitedb"
+    conn = sqlite3.connect(db_path)
+    conn.executescript(
+        """
+        CREATE TABLE ABPerson (
+            ROWID INTEGER PRIMARY KEY, First TEXT, Middle TEXT, Last TEXT,
+            Organization TEXT, Note TEXT
+        );
+        CREATE TABLE ABMultiValue (
+            ROWID INTEGER PRIMARY KEY, record_id INTEGER, value TEXT,
+            label INTEGER
+        );
+        INSERT INTO ABPerson VALUES (1, 'Alice', NULL, 'Example', NULL, 'note');
+        INSERT INTO ABPerson VALUES (2, 'Bob', NULL, 'NoValue', NULL, 'retained');
+        INSERT INTO ABMultiValue VALUES (10, 1, '+491234', 3);
+        """
+    )
+    conn.close()
+
+    module = Contacts(file_path=str(db_path))
+    run_module(module)
+
+    assert len(module.results) == 2
+    alice = next(result for result in module.results if result["first"] == "Alice")
+    bob = next(result for result in module.results if result["first"] == "Bob")
+    assert alice["contact"]["person_Note"] == "note"
+    assert alice["contact"]["multivalue_label"] == 3
+    assert alice["First"] == "Alice"
+    assert bob["value"] is None
+    assert bob["contact"]["person_Note"] == "retained"

tests/ios_backup/test_datausage.py

@@ -20,6 +20,8 @@ class TestDatausageModule:
         assert m.results[0]["isodate"][0:19] == "2019-08-27 15:08:09"
         assert len(m.results) == 42
         assert len(m.timeline) == 60
+        assert "live_ZBILLCYCLEEND" in m.results[0]["record"]
+        assert "process_Z_ENT" in m.results[0]["record"]
         assert (
             len(m.alertstore.alerts) == 1
         )  # We now have a detection for missing processes.

tests/ios_backup/test_interactionc.py

@@ -20,6 +20,10 @@ class TestInteractionCModule:
         run_module(m)
 
         assert len(m.results) == 3
+        assert all(
+            result["interaction"]["Z_PK"] == result["table_id"] for result in m.results
+        )
+        assert all("ZMECHANISM" in result["interaction"] for result in m.results)
 
         incoming = next(
             r for r in m.results if r["sender_identifier"] == "100000000000001@lid"

tests/ios_backup/test_safari_browserstate.py

@@ -5,6 +5,7 @@
 
 import logging
 import shutil
+import sqlite3
 
 import pytest
 
@@ -52,6 +53,7 @@ class TestSafariBrowserStateModule:
         assert len(m.results) == 1
         assert len(m.timeline) == 1
         assert len(m.alertstore.alerts) == 0
+        assert "tab" in m.results[0]
 
     def test_parsing_backup_with_profile(self, backup_with_safari_profile):
         m = SafariBrowserState(target_path=backup_with_safari_profile)
@@ -74,3 +76,30 @@ class TestSafariBrowserStateModule:
         assert len(m.alertstore.alerts) == 1
         assert len(m.results) == 1
         assert m.results[0]["tab_url"] == "https://en.wikipedia.org/wiki/NSO_Group"
+
+    def test_tabs_without_session_rows_are_not_dropped(self, tmp_path):
+        db_path = tmp_path / "BrowserState.db"
+        conn = sqlite3.connect(db_path)
+        conn.executescript(
+            """
+            CREATE TABLE tabs (
+                uuid TEXT, title TEXT, url TEXT, user_visible_url TEXT,
+                last_viewed_time REAL, private_browsing INTEGER
+            );
+            CREATE TABLE tab_sessions (tab_uuid TEXT, session_data BLOB);
+            INSERT INTO tabs VALUES (
+                'tab-1', 'Example', 'https://example.test',
+                'https://example.test', 700000000, 1
+            );
+            """
+        )
+        conn.close()
+        module = SafariBrowserState(target_path=str(tmp_path))
+
+        module._process_browser_state_db(str(db_path))
+
+        assert len(module.results) == 1
+        assert module.results[0]["tab_url"] == "https://example.test"
+        assert module.results[0]["session_data"] == []
+        assert module.results[0]["tab"]["private_browsing"] == 1
+        assert module.results[0]["tab"]["session_data"] is None

tests/ios_backup/test_safari_history.py

@@ -105,6 +105,7 @@ class TestSafariHistoryModule:
         assert len(m.results) == 2
         assert {result["url"] for result in m.results} == {DEFAULT_URL, PROFILE_URL}
         assert len({result["safari_history_db"] for result in m.results}) == 2
+        assert all("visit_history_item" in result["record"] for result in m.results)
 
     def test_parsing_fs_dump_with_profile(self, fs_dump_with_safari_profile):
         m = SafariHistory(target_path=fs_dump_with_safari_profile)

tests/ios_backup/test_tcc.py

@@ -22,6 +22,9 @@ class TestTCCModule:
         assert m.results[0]["service"] == "kTCCServiceUbiquity"
         assert m.results[0]["client"] == "com.apple.Preferences"
         assert m.results[0]["auth_value"] == "allowed"
+        assert m.results[0]["auth_value_value"] == 2
+        assert "policy_id" in m.results[0]
+        assert "csreq" in m.results[0]
 
     def test_tcc_detection(self, indicator_file):
         m = TCC(target_path=get_ios_backup_folder())

tests/ios_backup/test_webkit_resource_load_statistics.py

@@ -34,6 +34,7 @@ class TestWebkitResourceLoadStatisticsModule:
             "most_recent_web_push_interaction_time" not in result
             for result in m.results
         )
+        assert all("record" in result for result in m.results)
 
     def test_webkit_full_timestamp_schema(self, tmp_path):
         db_path = tmp_path / "observations.db"
@@ -68,3 +69,4 @@ class TestWebkitResourceLoadStatisticsModule:
         assert "most_recent_user_interaction_time_isodate" in result
         assert result["most_recent_web_push_interaction_time"] == -1.0
         assert "most_recent_web_push_interaction_time_isodate" not in result
+        assert result["record"]["domainID"] == 1