fix: resolve runtime errors in 7 artifact parsers

scripts/artifacts/FCMQueuedMessagesSkype.py

@@ -21,7 +21,6 @@
 """
 
 import pathlib
-import sys
 import datetime
 import json
 import typing
@@ -201,7 +200,7 @@ def process_content(content: str):
         return html.escape(html.unescape(content), quote=False)
 
 
-def get_fcm_skype(files_found, report_folder, seeker, wrap_text, mode):
+def get_fcm_skype(files_found, report_folder, _seeker, _wrap_text, mode):
     if mode == "s":
         app_name = "Skype"
         app_id = "com.skype.raider"
@@ -283,7 +282,10 @@ def get_fcm_skype(files_found, report_folder, seeker, wrap_text, mode):
                         conversation_id = rec.key_values["conversationId"]
                         # trim the number at the start for the lookup, so it can be used consistently with calls
                         conversation_id = STARTS_WITH_NUMBER.sub("", conversation_id, 1)
-                        payload = json.loads(rec.key_values["rawPayload"])
+                        try:
+                            payload = json.loads(rec.key_values["rawPayload"])
+                        except json.JSONDecodeError:
+                            continue
                         metadata = make_metadata_field(
                             payload,
                             {

scripts/artifacts/airGuard.py

@@ -13,19 +13,16 @@
     }
 }
 
-import sqlite3
-import textwrap
-
 from scripts.artifact_report import ArtifactHtmlReport
-from scripts.ilapfuncs import logfunc, tsv, timeline, is_platform_windows, open_sqlite_db_readonly, kmlgen, does_table_exist_in_db, convert_ts_human_to_utc, convert_utc_human_to_timezone
+from scripts.ilapfuncs import logfunc, tsv, timeline, open_sqlite_db_readonly, kmlgen, does_table_exist_in_db, convert_ts_human_to_utc, convert_utc_human_to_timezone
+
+def get_airGuard(files_found, report_folder, _seeker, _wrap_text):
 
-def get_airGuard(files_found, report_folder, seeker, wrap_text):
-
     data_list_scans = []
     data_list_tracker = []
-
+    file_found = ''
+
     for file_found in files_found:
-        file_name = str(file_found)
 
         if file_found.endswith('attd_db'):
             db = open_sqlite_db_readonly(file_found)
@@ -118,12 +115,16 @@ def get_airGuard(files_found, report_folder, seeker, wrap_text):
                     if start_scan_ts is None or start_scan_ts == 'None':
                         pass
                     else:
+                        if len(start_scan_ts.split(':')) == 2:
+                            start_scan_ts += ':00'
                         start_scan_ts = convert_utc_human_to_timezone(convert_ts_human_to_utc(start_scan_ts),'UTC')
-                    
+
                     end_scan_ts = str(row[1]).replace("T", " ")
                     if end_scan_ts is None or end_scan_ts == 'None':
                         pass
                     else:
+                        if len(end_scan_ts.split(':')) == 2:
+                            end_scan_ts += ':00'
                         end_scan_ts = convert_utc_human_to_timezone(convert_ts_human_to_utc(end_scan_ts),'UTC')
 
                     data_list_scans.append((start_scan_ts,end_scan_ts,row[2],row[3],row[4],row[5],file_found))
@@ -142,10 +143,10 @@ def get_airGuard(files_found, report_folder, seeker, wrap_text):
         report.write_artifact_data_table(data_headers, data_list_tracker, file_found)
         report.end_artifact_report()
 
-        tsvname = f'AirGuard AirTag Tracker'
+        tsvname = 'AirGuard AirTag Tracker'
         tsv(report_folder, data_headers, data_list_tracker, tsvname)
-        
-        tlactivity = f'AirGuard AirTag Tracker'
+
+        tlactivity = 'AirGuard AirTag Tracker'
         timeline(report_folder, tlactivity, data_list_tracker, data_headers)
 
         kmlactivity = 'AirGuard AirTag Tracker'
@@ -164,10 +165,10 @@ def get_airGuard(files_found, report_folder, seeker, wrap_text):
         report.write_artifact_data_table(data_headers, data_list_scans, file_found)
         report.end_artifact_report()
 
-        tsvname = f'AirGuard AirTag Scans'
+        tsvname = 'AirGuard AirTag Scans'
         tsv(report_folder, data_headers, data_list_scans, tsvname)
-        
-        tlactivity = f'AirGuard AirTag Scans'
+
+        tlactivity = 'AirGuard AirTag Scans'
         timeline(report_folder, tlactivity, data_list_scans, data_headers)
 
     else:

scripts/artifacts/deviceHealthServices_Battery.py

@@ -29,16 +29,12 @@
     }
 }
 
-import sqlite3
-import textwrap
 import os
 
-from packaging import version
-from scripts.artifact_report import ArtifactHtmlReport
 from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, convert_ts_human_to_utc, convert_utc_human_to_timezone
 
 @artifact_processor
-def Turbo_Battery(files_found, report_folder, seeker, wrap_text):
+def Turbo_Battery(files_found, _report_folder, _seeker, _wrap_text):
     source_file_turbo = ''
     turbo_db = ''
     data_list = []
@@ -80,7 +76,7 @@ def Turbo_Battery(files_found, report_folder, seeker, wrap_text):
                     if timestamp is None:
                         pass
                     else:
-                        timestamp = convert_utc_human_to_timezone(convert_ts_human_to_utc(timestamp),time_offset)
+                        timestamp = convert_utc_human_to_timezone(convert_ts_human_to_utc(timestamp),row[4])
                     data_list.append((timestamp,row[1],row[2],row[3],row[4],file_found))
 
             db.close()
@@ -90,40 +86,41 @@ def Turbo_Battery(files_found, report_folder, seeker, wrap_text):
     return data_headers, data_list, source_file_turbo
 
 @artifact_processor
-def Turbo_Bluetooth(files_found, report_folder, seeker, wrap_text):     
+def Turbo_Bluetooth(files_found, _report_folder, _seeker, _wrap_text):
     source_file_bluetooth = ''
-    turbo_db = ''
     data_list = []
 
-    if file_found.lower().endswith('bluetooth.db'):
-        bluetooth_db = str(file_found)
-        source_file_bluetooth = file_found.replace(seeker.directory, '')
-
-        db = open_sqlite_db_readonly(bluetooth_db)
-        cursor = db.cursor()
-        cursor.execute('''
-        select
-        datetime(timestamp_millis/1000,'unixepoch'),
-        bd_addr,
-        device_identifier,
-        battery_level,
-        volume_level,
-        time_zone
-        from battery_event
-        join device_address on battery_event.device_idx = device_address.device_idx
-        ''')
+    for file_found in files_found:
+        file_found = str(file_found)
+        if file_found.lower().endswith('bluetooth.db'):
+            bluetooth_db = str(file_found)
+            source_file_bluetooth = os.path.basename(file_found)
+
+            db = open_sqlite_db_readonly(bluetooth_db)
+            cursor = db.cursor()
+            cursor.execute('''
+            select
+            datetime(timestamp_millis/1000,'unixepoch'),
+            bd_addr,
+            device_identifier,
+            battery_level,
+            volume_level,
+            time_zone
+            from battery_event
+            join device_address on battery_event.device_idx = device_address.device_idx
+            ''')
 
-        all_rows = cursor.fetchall()
-        usageentries = len(all_rows)
-        if usageentries > 0:
-            for row in all_rows:
-                timestamp = row[0]
-                if timestamp is None:
-                    pass
-                else:
-                    timestamp = convert_utc_human_to_timezone(convert_ts_human_to_utc(timestamp),time_offset)
-                data_list.append((timestamp,row[1],row[2],row[3],row[4],row[5],file_found))
-        db.close()
+            all_rows = cursor.fetchall()
+            usageentries = len(all_rows)
+            if usageentries > 0:
+                for row in all_rows:
+                    timestamp = row[0]
+                    if timestamp is None:
+                        pass
+                    else:
+                        timestamp = convert_utc_human_to_timezone(convert_ts_human_to_utc(timestamp),row[5])
+                    data_list.append((timestamp,row[1],row[2],row[3],row[4],row[5],file_found))
+            db.close()
 
     data_headers = (('Timestamp','datetime'),'BT Device MAC Address','BT Device ID','Battery Level','Volume Level','Timezone','Source')
 

scripts/artifacts/googleInitiatedNav.py

@@ -1,50 +1,43 @@
 import blackboxprotobuf
-from datetime import *
+from datetime import datetime, timezone
 from scripts.artifact_report import ArtifactHtmlReport
-from scripts.ilapfuncs import logfunc, tsv, is_platform_windows, convert_utc_human_to_timezone, kmlgen, timeline
+from scripts.ilapfuncs import logfunc, tsv, convert_utc_human_to_timezone, timeline
 
-def get_googleInitiatedNav(files_found, report_folder, seeker, wrap_text):
+def get_googleInitiatedNav(files_found, report_folder, _seeker, _wrap_text):
     data_list = []
     for file_found in files_found:
         with open(file_found, 'rb') as f:
             data = f.read()
-
-            arreglo = (data)
-            pb = arreglo[8:]
-            values, types = blackboxprotobuf.decode_message(pb)
+
+            pb = data[8:]
+            values, _ = blackboxprotobuf.decode_message(pb)
 
-        if isinstance(values, dict):
-            timestamp = values['1']['2']
+        entries = values.get('1', [])
+        if isinstance(entries, dict):
+            entries = [entries]
+        for entry in entries:
+            timestamp = entry['2']
             timestamp = datetime.fromtimestamp(timestamp/1000000, tz=timezone.utc)
             timestamp = convert_utc_human_to_timezone(timestamp, 'UTC')
-            intendeddest = values['1']['4']['1'].decode()
-
+            intendeddest = entry['4']['1'].decode()
             data_list.append((timestamp, intendeddest))
-        else:
-            for data in values['1']:
-                timestamp = data['2']
-                timestamp = datetime.fromtimestamp(timestamp/1000000, tz=timezone.utc)
-                timestamp = convert_utc_human_to_timezone(timestamp, 'UTC')
-                intendeddest = data['4']['1'].decode()
-
-                data_list.append((timestamp, intendeddest))
 
         if len(data_list) > 0:
             report = ArtifactHtmlReport('Google Initiated Navigation')
-            report.start_artifact_report(report_folder, f'Google Initiated Navigation')
+            report.start_artifact_report(report_folder, 'Google Initiated Navigation')
             report.add_script()
             data_headers = ('Timestamp', 'Initiated Navigation Destination')
             report.write_artifact_data_table(data_headers, data_list, file_found)
             report.end_artifact_report()
-            
-            tsvname = f'Google Initiated Navigation'
+
+            tsvname = 'Google Initiated Navigation'
             tsv(report_folder, data_headers, data_list, tsvname)
-            
-            tlactivity = f'Google Initiated Navigation'
+
+            tlactivity = 'Google Initiated Navigation'
             timeline(report_folder, tlactivity, data_list, data_headers)
-            
+
         else:
-            logfunc(f'No Google Initiated Navigation available')
+            logfunc('No Google Initiated Navigation available')
 
 __artifacts__ = {
         "googleInitiatedNav": (

scripts/artifacts/googleVoice.py

@@ -61,11 +61,10 @@
 import os
 import time
 import struct
-import inspect
 from scripts.ilapfuncs import artifact_processor, get_binary_file_content, open_sqlite_db_readonly, does_table_exist_in_db, check_in_media
 
 @artifact_processor
-def googlevoice_accounts(files_found, report_folder, seeker, wrap_text):
+def googlevoice_accounts(files_found, _report_folder, _seeker, _wrap_text):
     data_headers = ('Account Number', 'Full Name', 'Email Address', 'Linked Phone Number', 'Current Google Voice Number')
     data_list = []
     source_path = ""
@@ -144,7 +143,7 @@ def googlevoice_accounts(files_found, report_folder, seeker, wrap_text):
     return data_headers, data_list, source_path
 
 @artifact_processor
-def googlevoice_calls(files_found, report_folder, seeker, wrap_text):
+def googlevoice_calls(files_found, _report_folder, _seeker, _wrap_text):
     data_headers = (('Timestamp', 'datetime'), 'Account Number', 'Direction', 'Caller', 'Recipient', 'Call Status', 'Voicemail Left', 'Duration', ('Call Recording', 'media'))
     data_list = []
     source_path = ""
@@ -244,7 +243,6 @@ def googlevoice_calls(files_found, report_folder, seeker, wrap_text):
                             # 23 has values if an incoming call was recorded
                             recording = ""
                             if '23' in message[0]:
-                                artifact_info = inspect.stack()[0]
                                 message_id = message[0]['1'].decode('utf-8')
                                 recording = ""
 
@@ -262,7 +260,7 @@ def googlevoice_calls(files_found, report_folder, seeker, wrap_text):
     return data_headers, data_list, source_path
 
 @artifact_processor
-def googlevoice_voicemails(files_found, report_folder, seeker, wrap_text):
+def googlevoice_voicemails(files_found, _report_folder, _seeker, _wrap_text):
     data_headers = (('Timestamp', 'datetime'), 'Account Number', 'Caller', 'Recipient', 'Duration', 'Read Status', 'Transcript', ('Audio File', 'media'))
     data_list = []
     source_path = ""
@@ -359,7 +357,6 @@ def googlevoice_voicemails(files_found, report_folder, seeker, wrap_text):
 
                             # Audio File
                             audio = ""
-                            artifact_info = inspect.stack()[0]
                             message_id = message[0]['1'].decode('utf-8')
                             # get the voicemail audio file
                             for audio_file in files_found:
@@ -372,7 +369,7 @@ def googlevoice_voicemails(files_found, report_folder, seeker, wrap_text):
     return data_headers, data_list, source_path
 
 @artifact_processor
-def googlevoice_messages(files_found, report_folder, seeker, wrap_text):
+def googlevoice_messages(files_found, _report_folder, _seeker, _wrap_text):
     data_headers = (('Timestamp', 'datetime'), 'Account Number', 'Conversation ID', 'Direction', 'Sender', 'Recipient(s)', 'Read Status', 'Message', ('Image', 'media'))
     data_list = []
     source_path = ""
@@ -456,11 +453,14 @@ def googlevoice_messages(files_found, report_folder, seeker, wrap_text):
                             # Message
                             message_content = ""
                             if '10' in message[0]:
-                                message_content = message[0]['10'].decode('utf-8')
+                                raw = message[0]['10']
+                                if isinstance(raw, bytes):
+                                    message_content = raw.decode('utf-8')
+                                elif isinstance(raw, dict):
+                                    message_content = next((v.decode('utf-8') for v in raw.values() if isinstance(v, bytes)), str(raw))
 
                             # Image
                             if "MMS" in message_content:
-                                artifact_info = inspect.stack()[0]
                                 message_id = message[0]['1'].decode('utf-8')
 
                                 # get image file
@@ -534,7 +534,6 @@ def googlevoice_messages(files_found, report_folder, seeker, wrap_text):
 
                             # Image
                             if "MMS" in message_content:
-                                artifact_info = inspect.stack()[0]
                                 message_id = message[0]['1'].decode('utf-8')
 
                                 # get the image file

scripts/artifacts/keepNotes.py

@@ -14,31 +14,34 @@
 }
 
 import sqlite3
-import datetime
 import os
 
 from scripts.artifact_report import ArtifactHtmlReport
 from scripts.ilapfuncs import logfunc, tsv, timeline, open_sqlite_db_readonly
 
-def get_keepNotes(files_found, report_folder, seeker, wrap_text):
+def get_keepNotes(files_found, report_folder, _seeker, _wrap_text):
     for file_found in files_found:
         file_found = str(file_found)
         filename = os.path.basename(file_found)
 
         if filename.endswith('keep.db'):
             db = open_sqlite_db_readonly(file_found)
             cursor = db.cursor()
-            cursor.execute('''
-            SELECT 
-                datetime(tree_entity.time_created/1000, 'unixepoch') AS "Time Created",
-                datetime(tree_entity.time_last_updated/1000, 'unixepoch') AS "Time Last Updated",
-                datetime(tree_entity.user_edited_timestamp/1000, 'unixepoch') AS "User Edited Timestamp",
-                tree_entity.title AS Title,
-                text_search_note_content_content.c0text AS "Text",
-                tree_entity.last_modifier_email AS "Last Modifier Email"
-            FROM text_search_note_content_content
-            INNER JOIN tree_entity ON text_search_note_content_content.docid = tree_entity._id
-            ''')
+            try:
+                cursor.execute('''
+                SELECT
+                    datetime(tree_entity.time_created/1000, 'unixepoch') AS "Time Created",
+                    datetime(tree_entity.time_last_updated/1000, 'unixepoch') AS "Time Last Updated",
+                    datetime(tree_entity.user_edited_timestamp/1000, 'unixepoch') AS "User Edited Timestamp",
+                    tree_entity.title AS Title,
+                    text_search_note_content_content.c0text AS "Text",
+                    tree_entity.last_modifier_email AS "Last Modifier Email"
+                FROM text_search_note_content_content
+                INNER JOIN tree_entity ON text_search_note_content_content.docid = tree_entity._id
+                ''')
+            except sqlite3.OperationalError as e:
+                logfunc(f'Google Keep Notes query failed: {e}')
+                continue
 
             all_rows = cursor.fetchall()
             usageentries = len(all_rows)