Merge pull request #1029 from Johann-PLW/main

Update appConduit.py

scripts/artifacts/appConduit.py

@@ -1,79 +1,69 @@
 __artifacts_v2__ = {
-    "get_appConduit": {  # This should match the function name exactly
+    "appConduit": {  # This should match the function name exactly
         "name": "App Conduit",
         "description": "The AppConduit log file stores information about interactions between iPhone and other iOS devices, i.e. Apple Watch",
         "author": "@ydkhatri",
-        "version": "1.0",
-        "date": "2020-08-05",
+        "creation_date": "2020-08-05",
+        "last_update_date": "2025-01-22",
         "requirements": "none",
         "category": "App Conduit",
         "notes": "",
-        "paths": ('*/AppConduit.log.*',),
-        "function": "get_appConduit",
-        "output_types": "all"  # or ["html", "tsv", "timeline", "lava"]
+        "paths": ('*/mobile/Library/Logs/AppConduit/AppConduit.log.*',),
+        "output_types": "standard",  # or ["html", "tsv", "timeline", "lava"]
+        "artifact_icon": "activity"
     }   
 }
 
-import datetime
-import glob
-import os
-import sys
-import stat
 import pathlib
-import string
-import json
 import re
-import textwrap
+import scripts.builds_ids as builds_ids
 
-from html import escape
+from scripts.ilapfuncs import artifact_processor, get_txt_file_content, convert_log_ts_to_utc
 
-from scripts.artifact_report import ArtifactHtmlReport
-from scripts.ilapfuncs import logfunc, tsv, is_platform_windows, convert_time_obj_to_utc, convert_time_obj_to_utc, convert_utc_human_to_timezone
-from scripts.lavafuncs import lava_process_artifact, lava_insert_sqlite_data
-
-def get_appConduit(files_found, report_folder, seeker, wrap_text, timezone_offset):
+@artifact_processor
+def appConduit(files_found, report_folder, seeker, wrap_text, timezone_offset):
+    source_paths = set()
     data_list = []
-    device_type_and_info = []
+    device_type_and_info = {}
 
     info = ''
     reg_filter = (r'(([A-Za-z]+[\s]+([a-zA-Z]+[\s]+[0-9]+)[\s]+([0-9]+\:[0-9]+\:[0-9]+)[\s]+([0-9]{4}))([\s]+[\[\d\]]+[\s]+[\<a-z\>]+[\s]+[\(\w\)]+)[\s\-]+(((.*)(device+\:([\w]+\-[\w]+\-[\w]+\-[\w]+\-[\w]+))(.*)$)))')        
     date_filter = re.compile(reg_filter)
 
-    source_files = []
     for file_found in files_found:
-        file_found = str(file_found)
         if file_found.startswith('\\\\?\\'):
             file_name = pathlib.Path(file_found[4:]).name
-            source_files.append(file_found[4:])
+            file_location = pathlib.Path(file_found[4:]).parent
         else:
             file_name = pathlib.Path(file_found).name
-            source_files.append(file_found)
-
-        file = open(file_found, "r", encoding="utf8")
-        linecount = 0
+            file_location = pathlib.Path(file_found).parent
+        source_paths.add(str(file_location))
 
+        file = get_txt_file_content(file_found)
         for line in file:
-            linecount = linecount + 1
             line_match = re.match(date_filter, line)
-
 
             if line_match:
                 date_time = line_match.group(3, 5, 4)
                 conv_time = ' '.join(date_time)
-                dtime_obj = datetime.datetime.strptime(conv_time, '%b %d %Y %H:%M:%S')
-                dtime_obj = convert_time_obj_to_utc(dtime_obj)
-                dtime_obj = convert_utc_human_to_timezone(dtime_obj, timezone_offset)
+                dtime_obj = convert_log_ts_to_utc(conv_time)
 
                 values  = line_match.group(9)
                 device_id = line_match.group(11)
 
                 if 'devicesAreNowConnected' in values:
-                    device_type_and_info.append((device_id,line_match.group(12).split(" ")[4],line_match.group(12).split(" ")[5]))
+                    pairing_id = line_match.group(12).split(" ")[3][:-1]
+                    device_type = line_match.group(12).split(" ")[4]
+                    device_info = builds_ids.device_id.get(device_type, device_type)
+                    os_build = line_match.group(12).split(" ")[7].strip("()")
+                    os_info = builds_ids.OS_build.get(os_build, os_build)
+                    device_type_and_info.setdefault(pairing_id, f"{device_info} - {os_info}")
                     info = 'Connected'
-                    data_list.append((dtime_obj,info,device_id,file_name))
+                    data_list.append((dtime_obj, info, device_id, device_type_and_info.get(pairing_id, ''), file_name))
                 if 'devicesAreNoLongerConnected' in values:
+                    pairing_id = line_match.group(12).split(" ")[3][:-1]
                     info = 'Disconnected'
-                    data_list.append((dtime_obj,info,device_id,file_name))
+                    data_list.append((dtime_obj, info, device_id, device_type_and_info.get(pairing_id, ''), file_name))
                 # if 'Resuming because' in values:
                 #     info = 'Resumed'
                 #     data_list.append((dtime_obj,info,device_id,device_type_tmp,file_name))
@@ -84,37 +74,7 @@ def get_appConduit(files_found, report_folder, seeker, wrap_text, timezone_offse
                 #     info = 'Reachable again after reunion sync'
                 #     data_list.append((dtime_obj,info,device_id,device_type_tmp,file_name))
 
-    device_type_and_info = list(set(device_type_and_info))
-
-    data_headers_device_info = ('Device ID', 'Device type and version', 'Device extra information')
-    data_headers = ('Time', 'Device interaction', 'Device ID', 'Log File Name')   
+    data_headers = (('Timestamp', 'datetime'), 'Device interaction', 'Device ID', 'Device type and OS version', 'Log File Name')
+    source_path = ', '.join(source_paths)
 
-    report = ArtifactHtmlReport('App Conduit')
-    report.start_artifact_report(report_folder, 'App Conduit', 'The AppConduit log file stores information about interactions between iPhone and other iOS devices, i.e. Apple Watch')
-
-    report.add_script()
-    source_files_found = ', '.join(source_files)
-
-    report.write_artifact_data_table(data_headers_device_info, device_type_and_info, source_files_found, cols_repeated_at_bottom=False)
-    report.write_artifact_data_table(data_headers, data_list, file_found, True, False)
-    report.end_artifact_report()
-
-    tsvname = 'App Conduit'
-    tsv(report_folder, data_headers, data_list, tsvname)
-
-    #LAVA section 
-    data_headers_device_info_l = ['Device ID', 'Device type and version', 'Device extra information']
-    data_headers_l = ['Time', 'Device interaction', 'Device ID', 'Log File Name']
-
-    category = "App Conduit"
-    module_name = "get_appConduit"
-
-    data_headers_l[0] = (data_headers_l[0], 'datetime')
-
-    table_name1, object_columns1, column_map1 = lava_process_artifact(category, module_name, 'App Conduit Device Info', data_headers_device_info_l, len(device_type_and_info))
-    lava_insert_sqlite_data(table_name1, device_type_and_info, object_columns1, data_headers_device_info_l, column_map1)
-
-    # Process second artifact for LAVA
-    table_name2, object_columns2, column_map2 = lava_process_artifact(category, module_name, 'App Conduit Device Interaction', data_headers_l, len(data_list))
-    lava_insert_sqlite_data(table_name2, data_list, object_columns2, data_headers_l, column_map2)
-
+    return data_headers, data_list, source_path

scripts/ilapfuncs.py

@@ -147,6 +147,45 @@ def __init__(self, output_folder, custom_folder_name=None):
         os.makedirs(os.path.join(self.report_folder_base, 'Script Logs'))
         os.makedirs(self.temp_folder)
 
+### New timestamp conversion functions
+def convert_unix_ts_in_seconds(ts):
+    digits = int(math.log10(ts))+1
+    if digits > 10:
+        extra_digits = digits - 10
+        ts = ts // 10**extra_digits
+    return int(ts)
+
+def convert_unix_ts_to_utc(ts):
+    if ts:
+        ts = convert_unix_ts_in_seconds(ts)
+        return datetime.fromtimestamp(ts, tz=timezone.utc)
+    else:
+        return ts
+
+def convert_unix_ts_to_str(ts):
+    if ts:
+        ts = convert_unix_ts_in_seconds(ts)
+        return datetime.fromtimestamp(ts, UTC).strftime('%Y-%m-%d %H:%M:%S')
+    else:
+        return ts
+
+def convert_cocoa_core_data_ts_to_utc(cocoa_core_data_ts):
+    if cocoa_core_data_ts:
+        unix_timestamp = cocoa_core_data_ts + 978307200
+        return convert_unix_ts_to_utc(unix_timestamp)
+    else:
+        return cocoa_core_data_ts
+
+def convert_log_ts_to_utc(str_dt):
+    if str_dt:
+        try:
+            return datetime.strptime(str_dt, '%b %d %Y %H:%M:%S').replace(tzinfo=timezone.utc)
+        except:
+            return str_dt
+    else:
+        return str_dt
+
+### Legacy timestamp conversion functions
 def convert_local_to_utc(local_timestamp_str):
     # Parse the timestamp string with timezone offset, ex. 2023-10-27 18:18:29-0400
     local_timestamp = datetime.strptime(local_timestamp_str, "%Y-%m-%d %H:%M:%S%z")
@@ -184,34 +223,6 @@ def convert_ts_int_to_timezone(time, time_offset):
     #return the converted value
     return timezone_time
 
-def convert_unix_ts_in_seconds(ts):
-    digits = int(math.log10(ts))+1
-    if digits > 10:
-        extra_digits = digits - 10
-        ts = ts // 10**extra_digits
-    return int(ts)
-
-def convert_unix_ts_to_utc(ts):
-    if ts:
-        ts = convert_unix_ts_in_seconds(ts)
-        return datetime.fromtimestamp(ts, tz=timezone.utc)
-    else:
-        return ts
-
-def convert_unix_ts_to_str(ts):
-    if ts:
-        ts = convert_unix_ts_in_seconds(ts)
-        return datetime.fromtimestamp(ts, UTC).strftime('%Y-%m-%d %H:%M:%S')
-    else:
-        return ts
-
-def convert_cocoa_core_data_ts_to_utc(cocoa_core_data_ts):
-    if cocoa_core_data_ts:
-        unix_timestamp = cocoa_core_data_ts + 978307200
-        return convert_unix_ts_to_utc(unix_timestamp)
-    else:
-        return cocoa_core_data_ts
-
 def webkit_timestampsconv(webkittime):
     unix_timestamp = webkittime + 978307200
     finaltime = datetime.fromtimestamp(unix_timestamp, tz=timezone.utc)
@@ -370,6 +381,19 @@ def get_next_unused_name(path):
         num += 1
     return os.path.join(folder, new_name)
 
+def get_txt_file_content(file_path):
+    try:
+        with open(file_path, "r", encoding="utf-8") as file:
+            file_content = file.readlines()
+            return file_content
+    except FileNotFoundError:
+        logfunc(f"Error: File not found at {file_path}")
+    except PermissionError:
+        logfunc(f"Error: Permission denied when trying to read {file_path}")
+    except Exception as e:
+        logfunc(f"Unexpected error reading file {file_path}: {str(e)}")
+    return []
+
 def get_plist_content(data):
     try:
         plist_content = plistlib.loads(data)