Add SQLite parser for iOS Tiktok Contacts

plaso/data/formatters/ios.yaml

@@ -187,6 +187,17 @@ short_source: 'LOG'
 source: 'iOS sysdiag log'
 ---
 type: 'conditional'
+data_type: 'ios:tiktok:contact'
+message:
+- 'Nickname: {nickname}'
+- 'URL: {url}'
+short_message:
+- 'Nickname: {nickname}'
+- 'URL: {url}'
+short_source: 'SQLITE'
+source: 'iOS TikTok contact database'
+---
+type: 'conditional'
 data_type: 'ios:twitter:contact'
 enumeration_helpers:
 - input_attribute: 'following'

plaso/data/timeliner.yaml

@@ -565,6 +565,12 @@ attribute_mappings:
   description: 'Content Modification Time'
 place_holder_event: true
 ---
+data_type: 'ios:tiktok:contact'
+attribute_mappings:
+- name: 'chat_timestamp'
+  description: 'Latest Chat Time'
+place_holder_event: true
+---
 data_type: 'ios:twitter:contact'
 attribute_mappings:
 - name: 'creation_time'

plaso/parsers/sqlite_plugins/__init__.py

@@ -26,6 +26,7 @@
 from plaso.parsers.sqlite_plugins import ios_netusage
 from plaso.parsers.sqlite_plugins import ios_powerlog
 from plaso.parsers.sqlite_plugins import ios_screentime
+from plaso.parsers.sqlite_plugins import ios_tiktok_contacts
 from plaso.parsers.sqlite_plugins import ios_twitter
 from plaso.parsers.sqlite_plugins import kodi
 from plaso.parsers.sqlite_plugins import ls_quarantine

plaso/parsers/sqlite_plugins/ios_tiktok_contacts.py

@@ -0,0 +1,103 @@
+# -*- coding:utf-8 -*-
+"""SQLite parser plugin for TikTok contacts database on iOS."""
+
+from dfdatetime import posix_time as dfdatetime_posix_time
+
+from plaso.containers import events
+from plaso.parsers import sqlite
+from plaso.parsers.sqlite_plugins import interface
+
+
+class IOSTikTokContactsEventData(events.EventData):
+  """iOS TikTok contacts event data.
+
+  Attributes:
+    chat_timestamp (dfdatetime.DateTimeValues): latest chat timestamp.
+    nickname (str): nickname of the contact.
+    url (str): url of the contact.
+  """
+
+  DATA_TYPE = 'ios:tiktok:contact'
+
+  def __init__(self):
+    """Initializes event data."""
+    super(IOSTikTokContactsEventData, self).__init__(data_type=self.DATA_TYPE)
+    self.chat_timestamp = None
+    self.nickname = None
+    self.url = None
+
+
+class IOSTikTokContactsPlugin(interface.SQLitePlugin):
+  """SQLite parser plugin for TikTok contacts database on iOS.
+
+  The TikTok contacts are stored in a SQLite database file named AwemeIM.db.
+  """
+
+  NAME = 'ios_tiktok_contacts'
+  DATA_FORMAT = 'iOS TikTok contacts SQLite database file AwemeIM.db'
+
+  REQUIRED_STRUCTURE = {
+    'AwemeContactsV5': frozenset([
+        'latestChatTimestamp', 'nickname', 'url1'])}
+
+  QUERIES = [((
+      'SELECT latestChatTimestamp, nickname, url1 FROM AwemeContactsV5'),
+      'ParseContactRow')]
+
+  SCHEMAS = {
+    'AwemeContactsV5': (
+        'CREATE TABLE AwemeContactsV5 (uid TEXT PRIMARY KEY, '
+        'accountType INTEGER, alias TEXT, aliasPinYin TEXT, '
+        'commerceUserLevel BLOB, contactName TEXT, contactNamePinYin TEXT, '
+        'customID TEXT, customVerifyInfo TEXT, enterpriseVerifyInfo TEXT, '
+        'followStatus INTEGER, followerCount BLOB, followingCount BLOB, '
+        'hasSetWelcomeMessage INTEGER, latestChatTimestamp INTEGER, '
+        'mentionAccessModel BLOB, nickname TEXT, nicknamePinYin TEXT, '
+        'recType INTEGER, secUserID TEXT, shortID TEXT, signature TEXT, '
+        'updatedTriggeredByContactModule INTEGER, url1 TEXT, url2 TEXT, '
+        'url3 TEXT, verificationType INTEGER)')}
+
+  REQUIRE_SCHEMA_MATCH = False
+
+  def _GetDateTimeRowValue(self, query_hash, row, value_name):
+    """Retrieves a date and time value from the row.
+
+    Args:
+    query_hash (int): hash of the query, that uniquely identifies the query
+        that produced the row.
+    row (sqlite3.Row): row.
+    value_name (str): name of the value.
+
+    Returns:
+    dfdatetime.CocoaTime: date and time value or None if not available.
+    """
+    timestamp = self._GetRowValue(query_hash, row, value_name)
+    if timestamp is None:
+      return None
+
+    # Convert the floating point value to an integer.
+    timestamp = int(timestamp)
+    return dfdatetime_posix_time.PosixTime(timestamp=timestamp)
+
+  # pylint: disable=unused-argument
+  def ParseContactRow(self, parser_mediator, query, row, **unused_kwargs):
+    """Parses a contact row.
+
+    Args:
+    parser_mediator (ParserMediator): mediates interactions between parsers
+        and other components, such as storage and dfVFS.
+    query (str): query that created the row.
+    row (sqlite3.Row): row.
+    """
+    query_hash = hash(query)
+
+    event_data = IOSTikTokContactsEventData()
+    event_data.chat_timestamp = self._GetDateTimeRowValue(
+        query_hash, row, 'latestChatTimestamp')
+    event_data.nickname = self._GetRowValue(query_hash, row, 'nickname')
+    event_data.url = self._GetRowValue(query_hash, row, 'url1')
+
+    parser_mediator.ProduceEventData(event_data)
+
+
+sqlite.SQLiteParser.RegisterPlugin(IOSTikTokContactsPlugin)

tests/parsers/sqlite_plugins/ios_tiktok_contacts.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Tests for TikTok on iOS SQLite database plugin."""
+
+import unittest
+
+from plaso.parsers.sqlite_plugins import ios_tiktok_contacts
+
+from tests.parsers.sqlite_plugins import test_lib
+
+
+class IOSTikTokContactsTest(test_lib.SQLitePluginTestCase):
+  """Tests for TikTok on iOS SQLite database plugin."""
+
+  def testProcess(self):
+    """Test the Process function."""
+    plugin = ios_tiktok_contacts.IOSTikTokContactsPlugin()
+    storage_writer = self._ParseDatabaseFileWithPlugin(
+        ['AwemeIM.db'], plugin)
+
+    number_of_event_data = storage_writer.GetNumberOfAttributeContainers(
+        'event_data')
+    self.assertGreater(number_of_event_data, 0)
+
+    number_of_warnings = storage_writer.GetNumberOfAttributeContainers(
+        'extraction_warning')
+    self.assertEqual(number_of_warnings, 0)
+
+    number_of_recovery_warnings = (
+        storage_writer.GetNumberOfAttributeContainers('recovery_warning'))
+    self.assertEqual(number_of_recovery_warnings, 0)
+
+        # Test a TikTok contact database entry.
+    expected_event_values = {
+        'chat_timestamp': '2021-12-03T00:00:00+00:00',
+        'data_type': 'ios:tiktok:contact',
+        'nickname': 'sample_user',
+        'url': 'https://www.tiktok.com/@sample_user'
+    }
+
+    event_data = storage_writer.GetAttributeContainerByIndex(
+        'event_data', 0)
+    self.CheckEventData(event_data, expected_event_values)
+
+
+if __name__ == '__main__':
+  unittest.main()