diff --git a/src/mvt/common/utils.py b/src/mvt/common/utils.py index 2a3e39a..7b9632e 100644 --- a/src/mvt/common/utils.py +++ b/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. diff --git a/src/mvt/ios/modules/mixed/calendar.py b/src/mvt/ios/modules/mixed/calendar.py index 40f8ad9..6566afc 100644 --- a/src/mvt/ios/modules/mixed/calendar.py +++ b/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() - - cur.execute( - """ - SELECT - CalendarItem.ROWID as "id", - CalendarItem.summary as "summary", - CalendarItem.description as "description", - CalendarItem.start_date as "start_date", - CalendarItem.end_date as "end_date", - CalendarItem.all_day as "all_day", - CalendarItem.calendar_id as "calendar_id", - CalendarItem.organizer_id as "organizer_id", - CalendarItem.url as "url", - CalendarItem.last_modified as "last_modified", - CalendarItem.external_id as "external_id", - CalendarItem.external_mod_tag as "external_mod_tag", - CalendarItem.unique_identifier as "unique_identifier", - CalendarItem.hidden as "hidden", - CalendarItem.UUID as "uuid", - CalendarItem.creation_date as "creation_date", - CalendarItem.action as "action", - CalendarItem.created_by_id as "created_by_id", - Participant.UUID as "participant_uuid", - Participant.email as "participant_email", - Participant.phone_number as "participant_phone", - Participant.comment as "participant_comment", - Participant.last_modified as "participant_last_modified" - FROM CalendarItem - LEFT JOIN Participant ON Participant.ROWID = CalendarItem.organizer_id; - """ - ) - - names = [description[0] for description in cur.description] - for item in cur: - entry = {} - for index, value in enumerate(item): - if names[index] in self.timestamps: - if value is None or isinstance(value, str): - entry[names[index]] = value - else: - entry[names[index]] = convert_mactime_to_iso(value) - else: - entry[names[index]] = value - - self.results.append(entry) - - cur.close() - conn.close() + try: + cur.execute("PRAGMA table_info(CalendarItem);") + calendar_columns = [row[1] for row in cur] + cur.execute("PRAGMA table_info(Participant);") + participant_columns = [row[1] for row in cur] + selected = [ + f'c."{column}" AS "calendar_{column}"' for column in calendar_columns + ] + [ + f'p."{column}" AS "participant_{column}"' + for column in participant_columns + ] + cur.execute( + f"SELECT {', '.join(selected)} FROM CalendarItem c " + "LEFT JOIN Participant p ON p.ROWID = c.organizer_id;" + ) + names = [description[0] for description in cur.description] + for row in cur: + raw = sanitize_json_data(dict(zip(names, row))) + entry = { + "id": raw.get("calendar_ROWID"), + "summary": raw.get("calendar_summary"), + "description": raw.get("calendar_description"), + "all_day": raw.get("calendar_all_day"), + "calendar_id": raw.get("calendar_calendar_id"), + "organizer_id": raw.get("calendar_organizer_id"), + "url": raw.get("calendar_url"), + "external_id": raw.get("calendar_external_id"), + "external_mod_tag": raw.get("calendar_external_mod_tag"), + "unique_identifier": raw.get("calendar_unique_identifier"), + "hidden": raw.get("calendar_hidden"), + "uuid": raw.get("calendar_UUID"), + "action": raw.get("calendar_action"), + "created_by_id": raw.get("calendar_created_by_id"), + "participant_uuid": raw.get("participant_UUID"), + "participant_email": raw.get("participant_email"), + "participant_phone": raw.get("participant_phone_number"), + "participant_comment": raw.get("participant_comment"), + "record": raw, + } + timestamp_sources = { + "start_date": "calendar_start_date", + "end_date": "calendar_end_date", + "last_modified": "calendar_last_modified", + "creation_date": "calendar_creation_date", + "participant_last_modified": "participant_last_modified", + } + for output, source in timestamp_sources.items(): + 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( diff --git a/src/mvt/ios/modules/mixed/calls.py b/src/mvt/ios/modules/mixed/calls.py index cee5862..c8628c9 100644 --- a/src/mvt/ios/modules/mixed/calls.py +++ b/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( - """ - SELECT - ZDATE, ZDURATION, ZLOCATION, ZADDRESS, ZSERVICE_PROVIDER - FROM ZCALLRECORD; - """ - ) - # names = [description[0] for description in cur.description] - - for row in cur: - self.results.append( - { - "isodate": convert_mactime_to_iso(row[0]), - "duration": row[1], - "location": row[2], - "number": ( - row[3].decode("utf-8") - if isinstance(row[3], bytes) - else row[3] - ), - "provider": row[4], - } - ) - - cur.close() - conn.close() + try: + cur.execute("SELECT * FROM ZCALLRECORD;") + names = [description[0] for description in cur.description] + for row in cur: + original = dict(zip(names, row)) + address = original.get("ZADDRESS") + if isinstance(address, bytes): + address = address.decode("utf-8", errors="replace") + raw = sanitize_json_data(original) + date = raw.get("ZDATE") + self.results.append( + { + "isodate": ( + convert_mactime_to_iso(date) if date is not None else "" + ), + "duration": raw.get("ZDURATION"), + "location": raw.get("ZLOCATION"), + "number": address, + "provider": raw.get("ZSERVICE_PROVIDER"), + "call": raw, + } + ) + finally: + cur.close() + conn.close() self.log.info("Extracted a total of %d calls", len(self.results)) diff --git a/src/mvt/ios/modules/mixed/contacts.py b/src/mvt/ios/modules/mixed/contacts.py index ca0470c..64fda03 100644 --- a/src/mvt/ios/modules/mixed/contacts.py +++ b/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( - """ - SELECT - multi.value, person.first, person.middle, person.last, - person.organization - FROM ABPerson person, ABMultiValue multi - WHERE person.rowid = multi.record_id and multi.value not null - ORDER by person.rowid ASC; - """ - ) - except sqlite3.OperationalError as e: - self.log.info("Error while reading the contact table: %s", e) - return None + cur.execute("PRAGMA table_info(ABPerson);") + person_columns = [row[1] for row in cur] + cur.execute("PRAGMA table_info(ABMultiValue);") + multivalue_columns = [row[1] for row in cur] + if not person_columns or not multivalue_columns: + return + + select_columns = [ + f'p."{column}" AS "person_{column}"' for column in person_columns + ] + [ + f'm."{column}" AS "multivalue_{column}"' + for column in multivalue_columns + ] + 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 = {} - for index, value in enumerate(row): - new_contact[names[index]] = value - - self.results.append(new_contact) + raw = sanitize_json_data(dict(zip(names, row))) + self.results.append( + { + "value": raw.get("multivalue_value"), + "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() diff --git a/src/mvt/ios/modules/mixed/interactionc.py b/src/mvt/ios/modules/mixed/interactionc.py index 4d21382..34c1cf7 100644 --- a/src/mvt/ios/modules/mixed/interactionc.py +++ b/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)) diff --git a/src/mvt/ios/modules/mixed/net_datausage.py b/src/mvt/ios/modules/mixed/net_datausage.py index 61b0eef..a883816 100644 --- a/src/mvt/ios/modules/mixed/net_datausage.py +++ b/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 diff --git a/src/mvt/ios/modules/mixed/safari_browserstate.py b/src/mvt/ios/modules/mixed/safari_browserstate.py index e67b7d3..7ea555f 100644 --- a/src/mvt/ios/modules/mixed/safari_browserstate.py +++ b/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 - tabs.title, - tabs.url, - tabs.user_visible_url, - tabs.last_viewed_time, - tab_sessions.session_data + SELECT tabs.*, tab_sessions.session_data AS 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( - """ - SELECT - title, url, user_visible_url, last_viewed_time, session_data - FROM tabs - ORDER BY last_viewed_time; - """ - ) - except sqlite3.OperationalError as e: - self.log.error(f"Error executing query: {e}") + cur.execute("SELECT * FROM tabs ORDER BY last_viewed_time;") + names = [description[0] for description in cur.description] + except sqlite3.OperationalError as exc: + self.log.error("Error executing query: %s", exc) return for row in cur: + raw = dict(zip(names, row)) session_entries = [] - - if row[4]: + session_blob = raw.get("session_data") + 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["SessionHistory"].get( - "SessionHistoryEntries" - ): - self._session_history_count += 1 - - data_length = 0 - if "SessionHistoryEntryData" in session_entry: - data_length = len( - session_entry.get("SessionHistoryEntryData") - ) - - session_entries.append( - { - "entry_title": session_entry.get( - "SessionHistoryEntryOriginalURL" - ), - "entry_url": session_entry.get( - "SessionHistoryEntryURL" - ), - "data_length": data_length, - } - ) + for session_entry in session_data.get("SessionHistory", {}).get( + "SessionHistoryEntries", [] + ): + self._session_history_count += 1 + entry_data = session_entry.get("SessionHistoryEntryData") + session_entries.append( + { + "entry_title": session_entry.get( + "SessionHistoryEntryOriginalURL" + ), + "entry_url": session_entry.get( + "SessionHistoryEntryURL" + ), + "data_length": ( + len(entry_data) if entry_data is not None else 0 + ), + } + ) + last_viewed = raw.get("last_viewed_time") self.results.append( { - "tab_title": row[0], - "tab_url": row[1], - "tab_visible_url": row[2], - "last_viewed_timestamp": convert_mactime_to_iso(row[3]), + "tab_title": raw.get("title"), + "tab_url": raw.get("url"), + "tab_visible_url": raw.get("user_visible_url"), + "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: diff --git a/src/mvt/ios/modules/mixed/safari_history.py b/src/mvt/ios/modules/mixed/safari_history.py index 4a50f02..1eba5b7 100644 --- a/src/mvt/ios/modules/mixed/safari_history.py +++ b/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( - """ - SELECT - history_items.id, - history_items.url, - history_visits.id, - history_visits.visit_time, - history_visits.redirect_source, - history_visits.redirect_destination - FROM history_items - JOIN history_visits ON history_visits.history_item = history_items.id - ORDER BY history_visits.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 - ), - } + try: + cur.execute("PRAGMA table_info(history_items);") + item_columns = [row[1] for row in cur] + cur.execute("PRAGMA table_info(history_visits);") + visit_columns = [row[1] for row in cur] + selected = [ + f'i."{column}" AS "item_{column}"' for column in item_columns + ] + [f'v."{column}" AS "visit_{column}"' for column in visit_columns] + cur.execute( + f"SELECT {', '.join(selected)} FROM history_items i " + "JOIN history_visits v ON v.history_item = i.id " + "ORDER BY v.visit_time;" ) - - cur.close() - conn.close() + names = [description[0] for description in cur.description] + for row in cur: + 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: diff --git a/src/mvt/ios/modules/mixed/tcc.py b/src/mvt/ios/modules/mixed/tcc.py index ed878c5..a5152f9 100644 --- a/src/mvt/ios/modules/mixed/tcc.py +++ b/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( - """SELECT - service, client, client_type, allowed, - prompt_count, last_modified - FROM access;""" + cur.execute("SELECT * FROM access;") + except sqlite3.OperationalError as exc: + self.log.error("Error parsing TCC database: %s", exc) + return + + 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: - service = row[0] - client = row[1] - client_type = row[2] - client_type_desc = "bundle_id" if client_type == 0 else "absolute_path" - if db_version == "v3": - auth_value = row[3] - 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" + if "auth_value" in record: + auth_value = record["auth_value"] + record["auth_value_value"] = auth_value + record["auth_value"] = AUTH_VALUES.get(auth_value, "unknown") + auth_reason = record.get("auth_reason") + record["auth_reason_desc"] = AUTH_REASONS.get( + auth_reason, "unknown" ) - self.log.info( - 'Found client "%s" with access %s to %s on %s by %s', - client, - auth_value_desc, - device, - last_modified, - auth_reason_desc, + elif "allowed" in record: + allowed = record["allowed"] + record["allowed_value"] = AUTH_VALUE_OLD.get(allowed, "unknown") + + if record.get("last_modified") is not None: + record["last_modified_value"] = record["last_modified"] + 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( - { - "service": service, - "client": client, - "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() + self.results.append(record) + finally: + cur.close() + conn.close() def run(self) -> None: self._find_ios_database(backup_ids=TCC_BACKUP_IDS, root_paths=TCC_ROOT_PATHS) diff --git a/src/mvt/ios/modules/mixed/webkit_resource_load_statistics.py b/src/mvt/ios/modules/mixed/webkit_resource_load_statistics.py index ef7b58f..a8970de 100644 --- a/src/mvt/ios/modules/mixed/webkit_resource_load_statistics.py +++ b/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);") - 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;" - ) + cur.execute("SELECT * 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], - "registrable_domain": row[1], - "last_seen": row[2], - "had_user_interaction": bool(row[3]), - "last_seen_isodate": convert_unix_to_iso(row[2]), + "domain_id": raw.get("domainID"), + "registrable_domain": raw.get("registrableDomain"), + "last_seen": last_seen, + "had_user_interaction": bool(raw.get("hadUserInteraction")), + "last_seen_isodate": convert_unix_to_iso(last_seen), "domain": domain, "path": path, + "record": raw, } - for index, column in enumerate(optional_columns, start=4): - field = { - "mostRecentUserInteractionTime": ( - "most_recent_user_interaction_time" - ), - "mostRecentWebPushInteractionTime": ( - "most_recent_web_push_interaction_time" - ), - }[column] - timestamp = row[index] + for column, field in ( + ( + "mostRecentUserInteractionTime", + "most_recent_user_interaction_time", + ), + ( + "mostRecentWebPushInteractionTime", + "most_recent_web_push_interaction_time", + ), + ): + 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) diff --git a/tests/common/test_utils.py b/tests/common/test_utils.py index 6bef147..7e07dd3 100644 --- a/tests/common/test_utils.py +++ b/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 diff --git a/tests/ios_backup/test_calendar.py b/tests/ios_backup/test_calendar.py index 5f1035d..ab632b7 100644 --- a/tests/ios_backup/test_calendar.py +++ b/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() diff --git a/tests/ios_backup/test_calls.py b/tests/ios_backup/test_calls.py new file mode 100644 index 0000000..1b2511b --- /dev/null +++ b/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=" diff --git a/tests/ios_backup/test_contacts.py b/tests/ios_backup/test_contacts.py new file mode 100644 index 0000000..8fdf414 --- /dev/null +++ b/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" diff --git a/tests/ios_backup/test_datausage.py b/tests/ios_backup/test_datausage.py index 17b0762..4b6e945 100644 --- a/tests/ios_backup/test_datausage.py +++ b/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. diff --git a/tests/ios_backup/test_interactionc.py b/tests/ios_backup/test_interactionc.py index eec2656..f2f7a80 100644 --- a/tests/ios_backup/test_interactionc.py +++ b/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" diff --git a/tests/ios_backup/test_safari_browserstate.py b/tests/ios_backup/test_safari_browserstate.py index fa2149d..af6741a 100644 --- a/tests/ios_backup/test_safari_browserstate.py +++ b/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 diff --git a/tests/ios_backup/test_safari_history.py b/tests/ios_backup/test_safari_history.py index 258b352..6cfff99 100644 --- a/tests/ios_backup/test_safari_history.py +++ b/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) diff --git a/tests/ios_backup/test_tcc.py b/tests/ios_backup/test_tcc.py index d1bf073..ae08187 100644 --- a/tests/ios_backup/test_tcc.py +++ b/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()) diff --git a/tests/ios_backup/test_webkit_resource_load_statistics.py b/tests/ios_backup/test_webkit_resource_load_statistics.py index 5e30cfe..74997da 100644 --- a/tests/ios_backup/test_webkit_resource_load_statistics.py +++ b/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