diff --git a/docs/ios/records.md b/docs/ios/records.md index 10ddab9..909b627 100644 --- a/docs/ios/records.md +++ b/docs/ios/records.md @@ -417,7 +417,7 @@ If indicators are provided through the command-line, they are checked against th Backup: :material-check: Full filesystem dump: :material-check: -This JSON file is created by mvt-ios' `WhatsApp` module. The module extracts a list of WhatsApp messages from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ChatStorage.sqlite*. +This JSON file is created by mvt-ios' `WhatsApp` module. The module extracts a list of WhatsApp messages from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ChatStorage.sqlite*, along with one record per chat session (marked with `"record_type": "chat_session"`) containing the first and last interaction dates of each conversation. Chat sessions produce `chat_first_message` and `chat_last_message` timeline events, and group chats additionally produce a `group_created` event. A chat session's last-message date can postdate its newest stored message when the most recent messages in the chat were deleted. If indicators are provided through the command-line, they are checked against the extracted HTTP links. Any matches are stored in *whatsapp_detected.json*. diff --git a/src/mvt/ios/modules/mixed/whatsapp.py b/src/mvt/ios/modules/mixed/whatsapp.py index 0a80aad..73d8fba 100644 --- a/src/mvt/ios/modules/mixed/whatsapp.py +++ b/src/mvt/ios/modules/mixed/whatsapp.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ import logging +import sqlite3 from typing import Optional from mvt.common.module_types import ( @@ -22,9 +23,50 @@ WHATSAPP_ROOT_PATHS = [ "private/var/mobile/Containers/Shared/AppGroup/*/ChatStorage.sqlite", ] +CHAT_SESSIONS_QUERY = """ + SELECT + ZWACHATSESSION.Z_PK AS "session_pk", + ZWACHATSESSION.ZCONTACTJID AS "contact_jid", + ZWACHATSESSION.ZPARTNERNAME AS "partner_name", + ZWACHATSESSION.ZSESSIONTYPE AS "session_type", + ZWACHATSESSION.ZARCHIVED AS "archived", + ZWACHATSESSION.ZREMOVED AS "removed", + ZWACHATSESSION.ZMESSAGECOUNTER AS "message_counter", + ZWACHATSESSION.ZLASTMESSAGEDATE AS "last_message_date", + ZWAGROUPINFO.ZCREATIONDATE AS "group_creation_date", + MIN(ZWAMESSAGE.ZMESSAGEDATE) AS "first_stored_message_date", + MAX(ZWAMESSAGE.ZMESSAGEDATE) AS "last_stored_message_date", + COUNT(ZWAMESSAGE.Z_PK) AS "stored_message_count" + FROM ZWACHATSESSION + LEFT JOIN ZWAGROUPINFO + ON ZWACHATSESSION.ZGROUPINFO = ZWAGROUPINFO.Z_PK + LEFT JOIN ZWAMESSAGE + ON ZWAMESSAGE.ZCHATSESSION = ZWACHATSESSION.Z_PK + GROUP BY ZWACHATSESSION.Z_PK; +""" + +CHAT_SESSION_DATE_FIELDS = [ + "last_message_date", + "group_creation_date", + "first_stored_message_date", + "last_stored_message_date", +] + + +def _describe_chat(record: ModuleAtomicResult) -> str: + jid = record.get("contact_jid") or "unknown" + name = record.get("partner_name") + label = f"'{name}' ({jid})" if name else jid + is_group = jid.endswith("@g.us") or record.get("group_creation_date") + if is_group: + return f"WhatsApp group chat {label}" + return f"WhatsApp chat with {label}" + class Whatsapp(IOSExtraction): - """This module extracts all WhatsApp messages containing links.""" + """This module extracts all WhatsApp messages containing links, as well + as per-chat records with the first and last interaction dates of each + conversation.""" def __init__( self, @@ -45,6 +87,9 @@ class Whatsapp(IOSExtraction): ) def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: + if record.get("record_type") == "chat_session": + return self._serialize_chat_session(record) + text = record.get("ZTEXT", "").replace("\n", "\\n") links_text = "" if record.get("links"): @@ -57,6 +102,49 @@ class Whatsapp(IOSExtraction): "data": f"'{text}' from {record.get('ZFROMJID', 'Unknown')}{links_text}", } + def _serialize_chat_session( + self, record: ModuleAtomicResult + ) -> ModuleSerializedResult: + records = [] + chat = _describe_chat(record) + + if record.get("group_creation_date"): + records.append( + { + "timestamp": record["group_creation_date"], + "module": self.__class__.__name__, + "event": "group_created", + "data": f"{chat} was created", + } + ) + + if record.get("first_stored_message_date"): + records.append( + { + "timestamp": record["first_stored_message_date"], + "module": self.__class__.__name__, + "event": "chat_first_message", + "data": f"First stored message in {chat}", + } + ) + + # The chat session's own last-message date is authoritative: it can + # postdate the newest stored message if that message was deleted. + last_message_date = record.get("last_message_date") or record.get( + "last_stored_message_date" + ) + if last_message_date: + records.append( + { + "timestamp": last_message_date, + "module": self.__class__.__name__, + "event": "chat_last_message", + "data": f"Last message in {chat}", + } + ) + + return records + def check_indicators(self) -> None: if not self.indicators: return @@ -146,7 +234,45 @@ class Whatsapp(IOSExtraction): message["links"] = list(set(filtered_links)) self.results.append(message) + total_messages = len(self.results) + total_sessions = self._extract_chat_sessions(cur) + cur.close() conn.close() - self.log.info("Extracted a total of %d WhatsApp messages", len(self.results)) + self.log.info( + "Extracted a total of %d WhatsApp messages and %d chat sessions", + total_messages, + total_sessions, + ) + + def _extract_chat_sessions(self, cur: sqlite3.Cursor) -> int: + """Extract one record per chat session with the first and last + interaction dates of each conversation.""" + try: + cur.execute(CHAT_SESSIONS_QUERY) + except sqlite3.OperationalError as exc: + self.log.warning( + "Unable to extract WhatsApp chat sessions: %s", exc + ) + return 0 + + names = [description[0] for description in cur.description] + + total_sessions = 0 + for row in cur: + session = dict(zip(names, row)) + session["record_type"] = "chat_session" + + for field in CHAT_SESSION_DATE_FIELDS: + if session.get(field): + session[field] = ( + convert_mactime_to_iso(session[field]) or None + ) + else: + session[field] = None + + self.results.append(session) + total_sessions += 1 + + return total_sessions diff --git a/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d b/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d new file mode 100644 index 0000000..10a26a6 Binary files /dev/null and b/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d differ diff --git a/tests/ios_backup/test_whatsapp.py b/tests/ios_backup/test_whatsapp.py index 864fb84..0a62d0e 100644 --- a/tests/ios_backup/test_whatsapp.py +++ b/tests/ios_backup/test_whatsapp.py @@ -6,8 +6,62 @@ import logging from mvt.common.indicators import Indicators +from mvt.common.module import run_module from mvt.ios.modules.mixed.whatsapp import Whatsapp +from ..utils import get_ios_backup_folder + + +def test_extraction(): + m = Whatsapp(target_path=get_ios_backup_folder()) + run_module(m) + + messages = [r for r in m.results if "ZTEXT" in r] + sessions = [r for r in m.results if r.get("record_type") == "chat_session"] + assert len(messages) == 3 + assert len(sessions) == 2 + + linked = next(r for r in messages if r.get("links")) + assert linked["links"] == ["https://example.org/news"] + + alice = next(s for s in sessions if s["partner_name"] == "Alice Example") + assert alice["contact_jid"] == "14155550100@s.whatsapp.net" + assert alice["first_stored_message_date"] == "2025-08-27 15:06:40.000000" + assert alice["last_message_date"] == "2025-08-28 18:53:20.000000" + assert alice["group_creation_date"] is None + assert alice["stored_message_count"] == 2 + + group = next(s for s in sessions if s["partner_name"] == "Example Group") + assert group["group_creation_date"] == "2025-08-21 20:13:20.000000" + assert group["first_stored_message_date"] == "2025-08-29 22:40:00.000000" + # The last stored message predates the session's own last-message date: + # the newest message in this chat was deleted. + assert group["last_stored_message_date"] == "2025-08-29 22:40:00.000000" + assert group["last_message_date"] == "2025-08-31 02:26:40.000000" + + # 3 message events plus first/last per chat and the group creation. + assert len(m.timeline) == 8 + events = { + (entry["event"], entry["timestamp"]): entry["data"] + for entry in m.timeline + } + assert events[("chat_first_message", "2025-08-27 15:06:40.000000")] == ( + "First stored message in WhatsApp chat with " + "'Alice Example' (14155550100@s.whatsapp.net)" + ) + assert events[("chat_last_message", "2025-08-28 18:53:20.000000")] == ( + "Last message in WhatsApp chat with " + "'Alice Example' (14155550100@s.whatsapp.net)" + ) + assert events[("group_created", "2025-08-21 20:13:20.000000")] == ( + "WhatsApp group chat 'Example Group' " + "(120000000000000001@g.us) was created" + ) + assert ("chat_first_message", "2025-08-29 22:40:00.000000") in events + assert ("chat_last_message", "2025-08-31 02:26:40.000000") in events + + assert len(m.alertstore.alerts) == 0 + def test_collect_url_results_includes_expansion(): module = Whatsapp( diff --git a/tests/ios_fs/test_filesystem.py b/tests/ios_fs/test_filesystem.py index 22d36d1..3a7e0eb 100644 --- a/tests/ios_fs/test_filesystem.py +++ b/tests/ios_fs/test_filesystem.py @@ -15,8 +15,8 @@ class TestFilesystem: def test_filesystem(self): m = Filesystem(target_path=get_ios_backup_folder()) run_module(m) - assert len(m.results) == 19 - assert len(m.timeline) == 19 + assert len(m.results) == 21 + assert len(m.timeline) == 21 assert len(m.alertstore.alerts) == 0 def test_detection(self, indicator_file): @@ -29,6 +29,6 @@ class TestFilesystem: ) m.indicators = ind run_module(m) - assert len(m.results) == 19 - assert len(m.timeline) == 19 + assert len(m.results) == 21 + assert len(m.timeline) == 21 assert len(m.alertstore.alerts) == 1