From d02b9676c85636b54e241f628c97e7cfb8b75388 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Tue, 18 Aug 2026 21:34:15 +0200 Subject: [PATCH] Fix InteractionC contact resolution and resolve WhatsApp LIDs to contacts The two primary InteractionC queries contained a SQL syntax error in their direction CASE expression (a double column alias), so they always failed and the module silently fell back to a reduced query without the recipient join. As a result outgoing messages were serialized with no counterpart at all ("from None (None)"). Fix the syntax so recipient names and identifiers are extracted again, and normalize the raw 0/1 direction values from the fallback queries to INCOMING/OUTGOING. WhatsApp identifies chat peers in interactionC.db by LID and stores the peer LID in the domain identifier, which InteractionC could not map to a person. Declare a dependency on the WhatsappContacts module and resolve sender, recipient and domain identifiers (LID, JID or phone number) against the WhatsApp contacts database, adding resolved phone number and name fields to WhatsApp records. Rewrite the timeline serialization to use the resolved values, fall back to the chat peer from the domain identifier when no recipient was recorded, label the local user instead of printing None, and include the message direction and group name. --- src/mvt/ios/modules/mixed/interactionc.py | 141 +++++++++++++++++- .../1f5a521220a3ad80ebfdc196978df8e7a2e49dee | Bin 0 -> 24576 bytes tests/ios_backup/test_interactionc.py | 74 +++++++++ tests/ios_fs/test_filesystem.py | 8 +- 4 files changed, 211 insertions(+), 12 deletions(-) create mode 100644 tests/artifacts/ios_backup/1f/1f5a521220a3ad80ebfdc196978df8e7a2e49dee create mode 100644 tests/ios_backup/test_interactionc.py diff --git a/src/mvt/ios/modules/mixed/interactionc.py b/src/mvt/ios/modules/mixed/interactionc.py index 81a67e2..73d8002 100644 --- a/src/mvt/ios/modules/mixed/interactionc.py +++ b/src/mvt/ios/modules/mixed/interactionc.py @@ -4,8 +4,9 @@ # https://license.mvt.re/1.1/ import logging +import re import sqlite3 -from typing import Optional +from typing import Optional, Tuple from mvt.common.module_types import ( ModuleAtomicResult, @@ -15,6 +16,7 @@ from mvt.common.module_types import ( from mvt.common.utils import convert_mactime_to_iso from ..base import IOSExtraction +from .whatsapp_contacts import WhatsappContacts INTERACTIONC_BACKUP_IDS = [ "1f5a521220a3ad80ebfdc196978df8e7a2e49dee", @@ -34,7 +36,7 @@ QUERIES = [ CASE ZINTERACTIONS.ZDIRECTION WHEN '0' THEN 'INCOMING' WHEN '1' THEN 'OUTGOING' - END 'DIRECTION' AS "direction", + END AS "direction", ZCONTACTS.ZDISPLAYNAME AS "sender_display_name", ZCONTACTS.ZIDENTIFIER AS "sender_identifier", ZCONTACTS.ZPERSONID AS "sender_personid", @@ -89,7 +91,7 @@ QUERIES = [ CASE ZINTERACTIONS.ZDIRECTION WHEN '0' THEN 'INCOMING' WHEN '1' THEN 'OUTGOING' - END 'DIRECTION' AS "direction", + END AS "direction", ZCONTACTS.ZDISPLAYNAME AS "sender_display_name", ZCONTACTS.ZIDENTIFIER AS "sender_identifier", ZCONTACTS.ZPERSONID AS "sender_personid", @@ -117,7 +119,7 @@ QUERIES = [ CASE ZCONTACTS.ZLASTINCOMINGRECIPIENTDATE WHEN '0' THEN '0' ELSE ZCONTACTS.ZLASTINCOMINGRECIPIENTDATE - END 'LAST INCOMING RECIPIENT DATE' AS "last_incoming_recipient_date", + END AS "last_incoming_recipient_date", ZCONTACTS.ZLASTOUTGOINGRECIPIENTDATE AS "last_outgoing_recipient_date", ZCONTACTS.ZCUSTOMIDENTIFIER AS "custom_id", ZINTERACTIONS.ZCONTENTURL AS "interaction_content_url", @@ -218,9 +220,16 @@ QUERIES = [ ] +WHATSAPP_BUNDLE_ID = "net.whatsapp.WhatsApp" + + class InteractionC(IOSExtraction): """This module extracts data from InteractionC db.""" + # WhatsApp identifies chat peers by LID in interactionC.db, which only the + # WhatsApp contacts database can map back to a phone number and name. + dependencies = [WhatsappContacts] + def __init__( self, file_path: Optional[str] = None, @@ -252,7 +261,48 @@ class InteractionC(IOSExtraction): "last_outgoing_recipient_date", ] + @staticmethod + def _describe_party(record: ModuleAtomicResult, prefix: str) -> Optional[str]: + name = record.get(f"{prefix}_display_name") or record.get( + f"{prefix}_resolved_name" + ) + identifier = record.get(f"{prefix}_resolved_phone_number") or record.get( + f"{prefix}_identifier" + ) + if name and identifier: + # A display name that is just a formatted copy of the phone + # number adds no information. + name_digits = re.sub(r"\D", "", name) + if name_digits and name_digits == re.sub(r"\D", "", identifier): + return identifier + return f"{name} ({identifier})" + return name or identifier or None + def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: + sender = self._describe_party(record, "sender") + # The chat peer from the domain identifier stands in when the + # recipient was not recorded (or the recipient join is unavailable). + recipient = self._describe_party(record, "recipient") or self._describe_party( + record, "domain" + ) + direction = record.get("direction") + if not sender and direction == "OUTGOING": + sender = "local user" + if not recipient and direction == "INCOMING": + recipient = "local user" + + header = f"[{record['bundle_id']}]" + if record.get("account"): + header += f" {record['account']}" + if direction: + header += f" {direction}" + + data = f"{header} from {sender or 'unknown'} to {recipient or 'unknown'}" + if record.get("group_name"): + data += f" (group: {record['group_name']})" + if record.get("content"): + data += f": {record['content']}" + records = [] processed = [] for timestamp in self.timestamps: @@ -269,16 +319,89 @@ class InteractionC(IOSExtraction): "timestamp": record[timestamp], "module": self.__class__.__name__, "event": timestamp, - "data": f"[{record['bundle_id']}] {record['account']} - " - f"from {record['sender_display_name']} ({record['sender_identifier']}) " - f"to {record.get('recipient_display_name', '')} ({record.get('recipient_identifier', '')}):" - f" {record.get('content', '')}", + "data": data, } ) processed.append(record[timestamp]) return records + def _whatsapp_contact_maps(self) -> Tuple[dict, dict]: + """Build LID and phone-digit lookup maps from the WhatsappContacts + module results, when available.""" + by_lid: dict = {} + by_phone: dict = {} + contacts_module = self.dependency_modules.get(WhatsappContacts) + if not contacts_module: + return by_lid, by_phone + + for contact in contacts_module.results: + name = contact.get("full_name") or contact.get("given_name") + phone = contact.get("phone_number") + entry = (phone, name) + if contact.get("lid"): + by_lid[contact["lid"]] = entry + if phone: + by_phone[re.sub(r"\D", "", phone)] = entry + whatsapp_id = contact.get("whatsapp_id") + if whatsapp_id and "@" in whatsapp_id: + by_phone.setdefault(whatsapp_id.split("@")[0], entry) + + return by_lid, by_phone + + @staticmethod + def _resolve_whatsapp_identifier( + value, by_lid: dict, by_phone: dict + ) -> Tuple[Optional[str], Optional[str]]: + """Resolve a WhatsApp identifier (LID, JID or phone number) to a + (phone_number, contact_name) tuple.""" + if not value: + return None, None + + value = str(value) + if value.endswith("@lid"): + return by_lid.get(value, (None, None)) + if value.endswith("@g.us"): + return None, None + if value.endswith("@s.whatsapp.net"): + digits = value.split("@")[0] + phone, name = by_phone.get(digits, (None, None)) + return phone or f"+{digits}", name + if value.startswith("+"): + _, name = by_phone.get(re.sub(r"\D", "", value), (None, None)) + return None, name + + return None, None + + def _postprocess_results(self) -> None: + by_lid, by_phone = self._whatsapp_contact_maps() + + for entry in self.results: + # The fallback queries return ZDIRECTION raw instead of labelled. + if entry.get("direction") in (0, "0"): + entry["direction"] = "INCOMING" + elif entry.get("direction") in (1, "1"): + entry["direction"] = "OUTGOING" + + if entry.get("bundle_id") != WHATSAPP_BUNDLE_ID: + continue + + candidates = { + "sender": [entry.get("sender_identifier"), entry.get("custom_id")], + "recipient": [entry.get("recipient_identifier")], + "domain": [entry.get("domain_identifier")], + } + for prefix, values in candidates.items(): + phone = name = None + for value in values: + phone, name = self._resolve_whatsapp_identifier( + value, by_lid, by_phone + ) + if phone or name: + break + entry[f"{prefix}_resolved_phone_number"] = phone + entry[f"{prefix}_resolved_name"] = name + def run(self) -> None: self._find_ios_database( backup_ids=INTERACTIONC_BACKUP_IDS, root_paths=INTERACTIONC_ROOT_PATHS @@ -325,4 +448,6 @@ class InteractionC(IOSExtraction): cur.close() conn.close() + self._postprocess_results() + self.log.info("Extracted a total of %d InteractionC events", len(self.results)) diff --git a/tests/artifacts/ios_backup/1f/1f5a521220a3ad80ebfdc196978df8e7a2e49dee b/tests/artifacts/ios_backup/1f/1f5a521220a3ad80ebfdc196978df8e7a2e49dee new file mode 100644 index 0000000000000000000000000000000000000000..0bcf761caf9731b78afa3a185ca2e2cbe340335d GIT binary patch literal 24576 zcmeI2O>fgM7{{HytSbm*8bZAwEz*Pr8j4oR%Yn<%#Oo~bQep>KFVH~|lQveM5*#UT z;0qw}2{`a6IB-DX%!lEIvoxL4B@8&B>F+L$>*ukbzkhygDed@ecV{#(9`8Tj>yL~U zSs&8?VwcUdu39OZ<M@rzM^dw0;ge}(yihL$Z@;D^*^T}FLT&^%{{)DqMXR@%=^;y@i5URKPlyx^(hf#0REUa2)!SG9385!_Vz@ZSB*$Lfs0j4>mu zX$fK3o1UEPvC5_niJ~_v^oxJ9=V zLUtgb`9jcp!ie7#QzY1I)5JRp-B}A~J<9y{mY~zgC_{BO0};4^ZMn>m(bxu~;p_;C zrR@f7rMK+{k)xPnoe8OTVX4+wTGHN6k}cLJGmZDsN;yj@ZT>q(4&znKRqNj<^WC;(pAxrP=RH^V$v~U!<$31V|k@^&J`(c?FT^1b;Hz0Vuge zop}GrUFa&IHdx3-8gDA>SkFXRq1zE@d7iFy7`b>n)$6zVV#ngRlcc(+)fy`++Iv0O zewpJy{ZPs#mumXE1WP8Cv&LLlp~%A`*F3NJcZ=ALe8;6n6VkGi+?mO28M#A6Q74B< z%<+kyBYL_3MLFpNp2g(WC_D*xNO?E#In5cp_y~&6msnC19i3z+45Dt{29{fg-Jwpr zcJf>*l;lx+)AAYjQbkFs#LwnsJs|pb{jI#AfB*=900@8p2!H?xfB*=900@8p2pmt~ zN_mlJl}e>W$b6m5A9N_2&pwg#dbq!5_J_mWf$8mU@9Yk^$2Urg@`v?GBww+;HLzbC&WSZ!Wy zUXx|5xz;>iS|B9js94voYtq`1uW8Boh9IXYe*TyL^p64pAOHd&00JNY0w4eaAOHd& z00JOzf(hXH{{$~CrUn8a00JNY0w4eaAOHd&00JNY0uuu9^S`G5B=Qdh1V8`;KmY_l a00ck)1V8`;KmY_l;G_|#YGq9;m3{&GM*U0x literal 0 HcmV?d00001 diff --git a/tests/ios_backup/test_interactionc.py b/tests/ios_backup/test_interactionc.py new file mode 100644 index 0000000..69a9957 --- /dev/null +++ b/tests/ios_backup/test_interactionc.py @@ -0,0 +1,74 @@ +# 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/ + +from mvt.common.module import run_module +from mvt.ios.modules.mixed.interactionc import InteractionC +from mvt.ios.modules.mixed.whatsapp_contacts import WhatsappContacts + +from ..utils import get_ios_backup_folder + + +class TestInteractionCModule: + def test_extraction_with_whatsapp_contacts(self): + contacts = WhatsappContacts(target_path=get_ios_backup_folder()) + run_module(contacts) + + m = InteractionC(target_path=get_ios_backup_folder()) + m.dependency_modules = {WhatsappContacts: contacts} + run_module(m) + + assert len(m.results) == 3 + + incoming = next( + r for r in m.results if r["sender_identifier"] == "100000000000001@lid" + ) + assert incoming["direction"] == "INCOMING" + assert incoming["sender_resolved_phone_number"] == "+14155550100" + assert incoming["sender_resolved_name"] == "Alice Example" + + outgoing = next( + r for r in m.results if r["direction"] == "OUTGOING" + ) + assert outgoing["recipient_identifier"] == "+14155550100" + assert outgoing["recipient_resolved_name"] == "Alice Example" + assert outgoing["domain_resolved_phone_number"] == "+14155550100" + assert outgoing["domain_resolved_name"] == "Alice Example" + + sms = next( + r for r in m.results if r["bundle_id"] == "com.apple.MobileSMS" + ) + assert sms.get("sender_resolved_name") is None + assert sms["sender_display_name"] == "Bob Example" + + events = [entry["data"] for entry in m.timeline] + assert ( + "[net.whatsapp.WhatsApp] INCOMING from " + "Alice Example (+14155550100) to local user" in events + ) + assert ( + "[net.whatsapp.WhatsApp] OUTGOING from local user to " + "Alice Example (+14155550100)" in events + ) + assert ( + "[com.apple.MobileSMS] INCOMING from " + "Bob Example (+14155550101) to local user" in events + ) + + def test_extraction_without_whatsapp_contacts(self): + # Without the WhatsappContacts dependency the module still runs, and + # unresolvable LIDs are shown as-is. + m = InteractionC(target_path=get_ios_backup_folder()) + run_module(m) + + assert len(m.results) == 3 + events = [entry["data"] for entry in m.timeline] + assert ( + "[net.whatsapp.WhatsApp] INCOMING from " + "100000000000001@lid to local user" in events + ) + assert ( + "[net.whatsapp.WhatsApp] OUTGOING from local user to " + "+14155550100" in events + ) diff --git a/tests/ios_fs/test_filesystem.py b/tests/ios_fs/test_filesystem.py index 10e647e..22d36d1 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) == 17 - assert len(m.timeline) == 17 + assert len(m.results) == 19 + assert len(m.timeline) == 19 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) == 17 - assert len(m.timeline) == 17 + assert len(m.results) == 19 + assert len(m.timeline) == 19 assert len(m.alertstore.alerts) == 1