From 19915d278fff8796f136fceeba31c7a5f4584d85 Mon Sep 17 00:00:00 2001 From: charpy4n6 <112824930+charpy4n6@users.noreply.github.com> Date: Mon, 17 Nov 2025 13:35:02 -0500 Subject: [PATCH 1/8] Add files via upload Parses external.db located at com.android.providers.media --- scripts/artifacts/externalDB.py | 90 +++++++++++++++++++++++++++++++++ 1 file changed, 90 insertions(+) create mode 100644 scripts/artifacts/externalDB.py diff --git a/scripts/artifacts/externalDB.py b/scripts/artifacts/externalDB.py new file mode 100644 index 00000000..43045349 --- /dev/null +++ b/scripts/artifacts/externalDB.py @@ -0,0 +1,90 @@ +__artifacts_v2__ = { + "externalDB": { + "name": "External Database", + "description": "Parses External Database", + "author": "Heather Charpentier", + "creation_date": "2025-11-17", + "last_updated_date": "2025-11-17", + "requirements": "none", + "category": "ExternalDB", + "notes": "", + "paths": ('*/data/com.android.providers.media/databases/external.db*'), + "output_types": "standard", + "artifact_icon": "download", + } +} + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly + +@artifact_processor +def externalDB(files_found, report_folder, seeker, wrap_text): + + data_list = [] + source_path = '' + + candidate_dbs = [] + + for file_found in files_found: + file_str = str(file_found).lower() + + if file_str.endswith('.db') and ( + 'media' in file_str or + 'external' in file_str or + 'internal' in file_str or + 'files' in file_str + ): + candidate_dbs.append(str(file_found)) + + if not candidate_dbs: + return ( + ('Path','Size',('Date Taken','datetime'),('Date Added','datetime'),('Date Modified','datetime'),'Album Name','Display Name','Directory','Duration'), + data_list, + source_path + ) + + for db_path in candidate_dbs: + + try: + db = open_sqlite_db_readonly(db_path) + source_path = db_path + + cursor = db.cursor() + + cursor.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='files';" + ) + table_check = cursor.fetchone() + + if not table_check: + db.close() + continue + + cursor.execute(''' + SELECT + datetime(datetaken/1000, 'unixepoch') AS "Date Taken", + datetime(date_added, 'unixepoch') AS "Date Added", + datetime(date_modified, 'unixepoch') AS "Date Modified", + album AS "Album Name", + _display_name AS "Display Name", + _data AS "Path", + duration AS "Duration", + _size AS "Size" + FROM files + ''') + + rows = cursor.fetchall() + + for row in rows: + data_list.append(( + row[0], row[1], row[2], row[3], row[4],row[5], row[6], row[7] + )) + + db.close() + + except Exception as e: + continue + + data_headers = (('Date Taken','datetime'),('Date Added','datetime'),('Date Modified','datetime'),'Album Name','Display Name','Path','Duration','Size') + + return data_headers, data_list, source_path + From b408f0716724b74e3fb54ee44eb07aa41c59ba35 Mon Sep 17 00:00:00 2001 From: charpy4n6 <112824930+charpy4n6@users.noreply.github.com> Date: Thu, 11 Dec 2025 13:23:02 -0500 Subject: [PATCH 2/8] MeWe The application updated and the database related to MeWe messages is now called app_v3.db. It is in the same location and its tables and columns still match the original mewe.py. --- scripts/artifacts/mewe.py | 25 ++++++++++++++----------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/scripts/artifacts/mewe.py b/scripts/artifacts/mewe.py index fc9f9400..866f1721 100644 --- a/scripts/artifacts/mewe.py +++ b/scripts/artifacts/mewe.py @@ -4,8 +4,9 @@ from scripts.ilapfuncs import logfunc, tsv, timeline, open_sqlite_db_readonly APP_NAME = 'MeWe' -DB_NAME = 'app_database' +DB_NAMES = ('app_database', 'app_v3.db') # Added support for app_v3.db SGSESSION_FILE = 'SGSession.xml' + CHAT_MESSAGES_QUERY = ''' SELECT DATETIME(createdAt, 'unixepoch'), @@ -31,6 +32,7 @@ JOIN CHAT_THREAD ON threadId = CHAT_THREAD.id ''' + def _perform_query(cursor, query): try: cursor.execute(query) @@ -71,11 +73,11 @@ def _parse_xml(xml_file, xml_file_name, report_folder, title, report_name): value = node.attrib['value'] except: value = node.text - + data_list.append((node.attrib['name'], value)) tl_bool = False - + _make_reports(f'{APP_NAME} - {report_name}', data_headers, data_list, report_folder, xml_file_name, tl_bool) @@ -116,12 +118,12 @@ def get_mewe(files_found, report_folder, seeker, wrap_text): db_file_name = None xml_file = None xml_file_name = None - + app_database_processed = False sgsession_processed = False for ff in files_found: - if ff.endswith(DB_NAME) and not app_database_processed: + if ff.endswith(DB_NAMES) and not app_database_processed: db_file = ff db_file_name = ff.replace(seeker.data_folder, '') _parse_app_database(db_file, db_file_name, report_folder) @@ -131,16 +133,17 @@ def get_mewe(files_found, report_folder, seeker, wrap_text): xml_file_name = ff.replace(seeker.data_folder, '') _parse_xml(xml_file, xml_file_name, report_folder, SGSESSION_FILE, 'SGSession') sgsession_processed = True - + artifacts = [ app_database_processed, sgsession_processed ] if not (True in artifacts): logfunc(f'{APP_NAME} data not found') + __artifacts__ = { - "mewe": ( - "MeWe", - ('*/com.mewe/databases/app_database', '*/com.mewe/shared_prefs/SGSession.xml'), - get_mewe) -} \ No newline at end of file + "mewe": ( + "MeWe", + ('*/com.mewe/databases/app_database', '*/com.mewe/databases/app_v3.db', '*/com.mewe/shared_prefs/SGSession.xml'), + get_mewe) +} From 52aebbd217b650628920466caf82e5544799ae7d Mon Sep 17 00:00:00 2001 From: charpy4n6 <112824930+charpy4n6@users.noreply.github.com> Date: Wed, 1 Jul 2026 13:08:06 -0400 Subject: [PATCH 3/8] Updated Support for Life 360 Addresses changes with the databases and the application capabilities itself. Tested on my Life360 test data. --- scripts/artifacts/L360circlesettings.py | 68 +++++++ scripts/artifacts/L360driveblade.py | 208 +++++++++++++++++++ scripts/artifacts/L360eventstore.py | 249 +++++++++++++++++++++++ scripts/artifacts/L360memberscircles.py | 72 +++++++ scripts/artifacts/L360noshowalerts.py | 258 ++++++++++++++++++++++++ scripts/artifacts/L360petprofile.py | 73 +++++++ scripts/artifacts/L360places.py | 151 ++++++++++++++ scripts/artifacts/L360user.py | 54 +++++ 8 files changed, 1133 insertions(+) create mode 100644 scripts/artifacts/L360circlesettings.py create mode 100644 scripts/artifacts/L360driveblade.py create mode 100644 scripts/artifacts/L360eventstore.py create mode 100644 scripts/artifacts/L360memberscircles.py create mode 100644 scripts/artifacts/L360noshowalerts.py create mode 100644 scripts/artifacts/L360petprofile.py create mode 100644 scripts/artifacts/L360places.py create mode 100644 scripts/artifacts/L360user.py diff --git a/scripts/artifacts/L360circlesettings.py b/scripts/artifacts/L360circlesettings.py new file mode 100644 index 00000000..f7c81d4f --- /dev/null +++ b/scripts/artifacts/L360circlesettings.py @@ -0,0 +1,68 @@ +__artifacts_v2__ = { + 'Life360_CircleSettings': { + 'name': 'Life360 CircleSettings', + 'description': 'Parses Life360 Circle Settings', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-11', + 'last_update_date': '2026-06-12', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360LocalStoreRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'circle' + } +} + +from datetime import datetime, timezone +from scripts.ilapfuncs import ( + artifact_processor, + get_file_path, + get_sqlite_db_records, + logfunc +) + +@artifact_processor +def Life360_CircleSettings(context): + + data_list = [] + + files_found = context.get_files_found() + source_path = get_file_path(files_found, 'L360LocalStoreRoomDatabase') + + query = ''' + SELECT + je.key AS "Circle ID", + json_extract(je.value, '$.featureSetId') AS "App Features", + json_extract(je.value, '$.featureSetRefId') AS "App Features ID", + json_extract(je.value, '$.features.collision_alerts_push') AS "Collision Alerts Push", + json_extract(je.value, '$.features.collision_alerts_sms') AS "Collision Alerts SMS", + json_extract(je.value, '$.features.customer_support') AS "Customer Support", + json_extract(je.value, '$.features.data_breach_detection') AS "Data Breach Detection", + json_extract(je.value, '$.features.driver_reports') AS "Driver Reports", + json_extract(je.value, '$.features.location_history') AS "Location History", + json_extract(je.value, '$.features.place_alerts') AS "Place Alerts", + json_extract(je.value, '$.features.plan') AS "Plan", + json_extract(je.value, '$.features.sos_alerts_push') AS "SOS Alerts Push", + json_extract(je.value, '$.features.sos_alerts_sms') AS "SOS Alerts SMS", + json_extract(je.value, '$.features.tilegps_activation') AS "TileGPS Activation", + json_extract(je.value, '$.features.uber_one') AS "Uber One" + FROM Premium p + CROSS JOIN json_each(p.circleFeatureSetInfo) je; + ''' + + try: + db_records = get_sqlite_db_records(source_path, query) + + logfunc(f'Life360_CircleSettings: Records found = {len(db_records)}') + + for record in db_records: + + data_list.append((record[0], record[1], record[2], record[3], record[4], record[5], record[6], record[7], record[8], record[9], record[10], record[11], record[12], record[13], record[14])) + + except Exception as e: + logfunc(f'Error processing Life360 CircleSettings: {e}') + + data_headers = ('Circle ID', 'App Features', 'App Features ID', 'Collision Alerts Push', 'Collision Alerts SMS', 'Customer Support', 'Data Breach Detection', 'Driver Reports', 'Location History', 'Place Alerts', 'Plan', 'SOS Alerts Push', 'SOS Alerts SMS', 'TileGPS Activation', 'Uber One') + + return data_headers, data_list, source_path \ No newline at end of file diff --git a/scripts/artifacts/L360driveblade.py b/scripts/artifacts/L360driveblade.py new file mode 100644 index 00000000..e5ae18db --- /dev/null +++ b/scripts/artifacts/L360driveblade.py @@ -0,0 +1,208 @@ +# pylint: disable=W0613 +__artifacts_v2__ = { + 'Life360_Drives': { + 'name': 'Life360 Drives', + 'description': 'Parses Life360 Drives', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/DriveBladeDB*',), + 'output_types': 'standard', + 'artifact_icon': 'map-pin' + }, + 'Life360_DriveEvents': { + 'name': 'Life360 Drive Events', + 'description': 'Parses Life360 Drive Events', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/DriveBladeDB*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'map-pin' + }, + 'Life360_DriveWaypoints': { + 'name': 'Life360 Drive Waypoints', + 'description': 'Parses Life360 Drive Waypoints', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/DriveBladeDB*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'map-pin' + } +} + +import datetime +import sqlite3 + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc + + +def _ms_to_utc(value): + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _find(context, suffix): + for file_found in context.get_files_found(): + file_found = str(file_found) + if file_found.endswith(suffix): + return file_found + return '' + + +def _q(cursor, sql): + try: + cursor.execute(sql) + return cursor.fetchall() + except sqlite3.Error: + return [] + + +@artifact_processor +def Life360_Drives(context): + source = _find(context, 'DriveBladeDB') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + driveId, + userId, + startTime, + endTime, + topSpeed, + topSpeed * 2.23694, + averageSpeed, + averageSpeed * 2.23694, + distance, + duration, + speedingCount, + hardBrakingCount, + rapidAccelerationCount, + distractedCount, + crashCount, + score, + updatedAt + FROM Drives + ORDER BY startTime + '''): + data_list.append(( + row[0], row[1], + _ms_to_utc(row[2]), _ms_to_utc(row[3]), + row[4], row[5], row[6], row[7], + row[8], row[9], + row[10], row[11], row[12], row[13], row[14], + row[15], _ms_to_utc(row[16]) + )) + db.close() + + data_headers = ( + 'Drive ID', 'User ID', + ('Start Time', 'datetime'), ('End Time', 'datetime'), + 'Top Speed MPS', 'Top Speed MPH', + 'Average Speed MPS', 'Average Speed MPH', + 'Distance', 'Duration', + 'Speeding Events', 'Hard Braking Events', + 'Rapid Acceleration Events', 'Distracted Events', 'Crash Events', + 'Score', ('Updated At', 'datetime') + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_DriveEvents(context): + source = _find(context, 'DriveBladeDB') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + DriveEvents.eventId, + DriveEvents.driveId, + Drives.userId, + DriveEvents.eventTime, + DriveEvents.eventType, + DriveEvents.lat, + DriveEvents.lon, + DriveEvents.speed, + DriveEvents.speed * 2.23694, + DriveEvents.accuracy + FROM DriveEvents + LEFT JOIN Drives ON Drives.driveId = DriveEvents.driveId + ORDER BY DriveEvents.eventTime + '''): + data_list.append(( + row[0], row[1], row[2], + _ms_to_utc(row[3]), + row[4], row[5], row[6], + row[7], row[8], row[9] + )) + db.close() + + data_headers = ( + 'Event ID', 'Drive ID', 'User ID', + ('Event Time', 'datetime'), + 'Event Type', 'Latitude', 'Longitude', + 'Speed MPS', 'Speed MPH', 'Accuracy' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_DriveWaypoints(context): + source = _find(context, 'DriveBladeDB') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + DriveWaypoints.driveId, + Drives.userId, + DriveWaypoints.timestamp, + DriveWaypoints.lat, + DriveWaypoints.lon, + DriveWaypoints.speed, + DriveWaypoints.speed * 2.23694, + DriveWaypoints.accuracy + FROM DriveWaypoints + LEFT JOIN Drives ON Drives.driveId = DriveWaypoints.driveId + ORDER BY DriveWaypoints.timestamp + '''): + data_list.append(( + row[0], row[1], + _ms_to_utc(row[2]), + row[3], row[4], + row[5], row[6], row[7] + )) + db.close() + + data_headers = ( + 'Drive ID', 'User ID', + ('Timestamp', 'datetime'), + 'Latitude', 'Longitude', + 'Speed MPS', 'Speed MPH', 'Accuracy' + ) + return data_headers, data_list, source \ No newline at end of file diff --git a/scripts/artifacts/L360eventstore.py b/scripts/artifacts/L360eventstore.py new file mode 100644 index 00000000..e59d1506 --- /dev/null +++ b/scripts/artifacts/L360eventstore.py @@ -0,0 +1,249 @@ +__artifacts_v2__ = { + 'Life360_Locations': { + 'name': 'Life360 Locations', + 'description': 'Parses Life360 Location Events', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'map-pin' + }, + 'Life360_Waypoints': { + 'name': 'Life360 Waypoints', + 'description': 'Parses Life360 Waypoints', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'map-pin' + }, + 'Life360_BatteryLevel': { + 'name': 'Life360 Battery Level', + 'description': 'Parses Life360 Battery Level Events', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), + 'output_types': 'standard', + 'artifact_icon': 'battery-charging' + }, + 'Life360_Drive_Events': { + 'name': 'Life360 Drive Events', + 'description': 'Parses Life360 Drive Events', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'truck' + } +} + +import datetime +import sqlite3 + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc + + +def _ms_to_utc(value): + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _find(context, suffix): + for file_found in context.get_files_found(): + file_found = str(file_found) + if file_found.endswith(suffix): + return file_found + return '' + + +def _q(cursor, sql): + try: + cursor.execute(sql) + return cursor.fetchall() + except sqlite3.Error: + return [] + + +@artifact_processor +def Life360_Locations(context): + source = _find(context, 'L360EventStore_service.db') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + json_extract(event.data, '$.locationData.time') AS timestamp, + json_extract(event.data, '$.locationData.latitude') AS latitude, + json_extract(event.data, '$.locationData.longitude') AS longitude, + json_extract(event.data, '$.locationData.horizontalAccuracy') AS horizontal_accuracy, + json_extract(event.data, '$.locationData.speed') AS speed_mps, + json_extract(event.data, '$.locationData.speed') * 2.23694 AS speed_mph, + json_extract(event.data, '$.locationData.speedAccuracy') AS speed_accuracy + FROM event + WHERE eventVersion = 6 + ORDER BY timestamp + '''): + data_list.append(( + _ms_to_utc(row[0]), + row[1], row[2], row[3], + row[4], row[5], row[6] + )) + db.close() + + data_headers = ( + ('Timestamp', 'datetime'), 'Latitude', 'Longitude', + 'Horizontal Accuracy', 'Speed MPS', 'Speed MPH', 'Speed Accuracy' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_Waypoints(context): + source = _find(context, 'L360EventStore_service.db') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + json_extract(event.data, '$.tripId') AS trip_id, + json_extract(json_each.value, '$.timestamp') AS timestamp, + json_extract(json_each.value, '$.latitude') AS latitude, + json_extract(json_each.value, '$.longitude') AS longitude, + json_extract(json_each.value, '$.accuracy') AS accuracy, + json_extract(json_each.value, '$.speed') AS speed_mps, + json_extract(json_each.value, '$.speed') * 2.23694 AS speed_mph + FROM event, + json_each(event.data, '$.type.waypoints') + WHERE event.eventVersion IN (1, 3, 4, 5) + ORDER BY timestamp + '''): + data_list.append(( + row[0], + _ms_to_utc(row[1]), + row[2], row[3], row[4], + row[5], row[6] + )) + db.close() + + data_headers = ( + 'Trip ID', ('Timestamp', 'datetime'), + 'Latitude', 'Longitude', 'Accuracy', + 'Speed MPS', 'Speed MPH' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_BatteryLevel(context): + source = _find(context, 'L360EventStore_service.db') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT DISTINCT + timestamp, + CASE + WHEN eventVersion IN (1, 2) + THEN json_extract(event.data, '$.batteryLevel') + WHEN eventVersion = 6 + THEN json_extract(event.data, '$.metaData.battery') + END AS battery_level, + CASE + WHEN eventVersion IN (1, 2) THEN + CASE + WHEN json_extract(event.data, '$.chargingState') = 1 THEN 'True' + WHEN json_extract(event.data, '$.chargingState') = 0 THEN 'False' + ELSE NULL + END + WHEN eventVersion = 6 THEN + CASE + WHEN json_extract(event.data, '$.metaData.chargingState') = 1 THEN 'True' + WHEN json_extract(event.data, '$.metaData.chargingState') = 0 THEN 'False' + ELSE NULL + END + END AS charging_state, + CASE + WHEN eventVersion = 2 + THEN json_extract(event.data, '$.powerMode') + ELSE NULL + END AS power_mode + FROM event + WHERE eventVersion IN (1, 2, 6) + AND (data LIKE '%batteryLevel%' OR data LIKE '%metaData%') + ORDER BY timestamp + '''): + data_list.append(( + _ms_to_utc(row[0]), + row[1], row[2], row[3] + )) + db.close() + + data_headers = ( + ('Timestamp', 'datetime'), + 'Battery Level', 'Charging State', 'Power Mode' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_Drive_Events(context): + source = _find(context, 'L360EventStore_service.db') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + json_extract(event.data, '$.tripId') AS trip_id, + json_extract(json_each.value, '$.timestamp') AS timestamp, + json_extract(json_each.value, '$.latitude') AS latitude, + json_extract(json_each.value, '$.longitude') AS longitude, + json_extract(json_each.value, '$.type') AS type + FROM event, + json_each(event.data, '$.type.driveEvents') + WHERE event.eventVersion IN (3, 4, 5) + ORDER BY timestamp + '''): + data_list.append(( + row[0], + _ms_to_utc(row[1]), + row[2], row[3], row[4] + )) + db.close() + + data_headers = ( + 'Trip ID', ('Timestamp', 'datetime'), + 'Latitude', 'Longitude', 'Type' + ) + return data_headers, data_list, source \ No newline at end of file diff --git a/scripts/artifacts/L360memberscircles.py b/scripts/artifacts/L360memberscircles.py new file mode 100644 index 00000000..3fc91bb2 --- /dev/null +++ b/scripts/artifacts/L360memberscircles.py @@ -0,0 +1,72 @@ +__artifacts_v2__ = { + 'Life360_MemberCircles': { + 'name': 'Life360 Members and Circles', + 'description': 'Parses Life360 Members and Circles', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/MembersEngineRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'user' + } +} + +from datetime import datetime, timezone +from scripts.ilapfuncs import ( + artifact_processor, + get_file_path, + get_sqlite_db_records, + logfunc +) + +@artifact_processor +def Life360_MemberCircles(context): + + data_list = [] + + files_found = context.get_files_found() + source_path = get_file_path(files_found, 'MembersEngineRoomDatabase') + + query = ''' + SELECT + members.created_at AS "Created Timestamp", + members.last_updated AS "Last Updated Timestamp", + members.id AS "Member ID", + members.first_name AS "First Name", + members.last_name AS "Last Name", + members.login_email AS "Email", + members.login_phone AS "Phone Number", + members.avatar AS "Avatar", + members.is_admin AS "Admin", + members.role AS "Role", + circles.created_at AS "Circle Created Timestamp", + circles.last_updated AS "Circle Last Updated Timestamp", + circles.name AS "Circle Name" + FROM members + LEFT JOIN circles + ON members.circle_id = circles.id + ''' + + try: + db_records = get_sqlite_db_records(source_path, query) + + logfunc(f'Life360_MemberCircles: Records found = {len(db_records)}') + + for record in db_records: + + created_timestamp = datetime.fromtimestamp(int(record[0]), tz=timezone.utc) + updated_timestamp = datetime.fromtimestamp(int(record[1]) / 1000, tz=timezone.utc) + second_created_timestamp = datetime.fromtimestamp(int(record[10]), tz=timezone.utc) + second_updated_timestamp = datetime.fromtimestamp(int(record[11]) / 1000, tz=timezone.utc) + + data_list.append((created_timestamp, updated_timestamp, record[2], record[3], record[4], record[5], record[6], record[7], record[8], record[9], second_created_timestamp,second_updated_timestamp, record[12])) + + except Exception as e: + logfunc(f'Error processing Life360 MemberCircles: {e}') + + data_headers = (('Created Timestamp', 'datetime'), ('Updated Timestamp', 'datetime'), 'Member ID', 'First Name', 'Last Name', 'Email', 'Phone Number', 'Avatar', 'Admin', 'Role', ('Circle Created Timestamp', 'datetime'), ('Circle Updated Timestamp', 'datetime'), 'Circle Name') + + return data_headers, data_list, source_path \ No newline at end of file diff --git a/scripts/artifacts/L360noshowalerts.py b/scripts/artifacts/L360noshowalerts.py new file mode 100644 index 00000000..519c450e --- /dev/null +++ b/scripts/artifacts/L360noshowalerts.py @@ -0,0 +1,258 @@ +# pylint: disable=W0613 +__artifacts_v2__ = { + 'Life360_NoShowAlerts': { + 'name': 'Life360 No Show Alerts', + 'description': 'Parses Life360 No Show Alerts including records recovered from WAL', + 'author': 'Heather Charpentier', + 'creation_date': '2026-07-01', + 'last_update_date': '2026-07-01', + 'requirements': 'none', + 'category': 'Life360', + 'notes': 'WAL file recovery', + 'paths': ('*/com.life360.android.safetymapd/databases/NoShowAlertRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'alert-triangle' + } +} + +import datetime +import struct +import sqlite3 + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc + + +COL_NAMES = [ + 'id', 'type', 'last_updated', 'trigger_condition', + 'critical_alert', 'run_at', 'place_id', + 'observed_user_id', 'creator_id', 'circle_id', 'daily' +] + + +def _ms_to_utc(value): + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _ns_to_utc(value): + """run_at is stored in nanoseconds.""" + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1_000_000_000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _find(context, suffix): + for file_found in context.get_files_found(): + if str(file_found).endswith(suffix): + return str(file_found) + return '' + + +def _read_varint(data, pos): + result = 0 + for i in range(9): + byte = data[pos + i] + result = (result << 7) | (byte & 0x7f) + if not (byte & 0x80): + return result, pos + i + 1 + return result, pos + 9 + + +def _decode_value(cell, pos, stype): + """Decode a SQLite serial type value from cell data.""" + if stype == 0: + return None, pos + elif stype == 1: + return int.from_bytes(cell[pos:pos+1], 'big', signed=True), pos + 1 + elif stype == 2: + return int.from_bytes(cell[pos:pos+2], 'big', signed=True), pos + 2 + elif stype == 3: + return int.from_bytes(cell[pos:pos+3], 'big', signed=True), pos + 3 + elif stype == 4: + return int.from_bytes(cell[pos:pos+4], 'big', signed=True), pos + 4 + elif stype == 5: + return int.from_bytes(cell[pos:pos+6], 'big', signed=True), pos + 6 + elif stype == 6: + return int.from_bytes(cell[pos:pos+8], 'big', signed=True), pos + 8 + elif stype == 7: + return struct.unpack('>d', cell[pos:pos+8])[0], pos + 8 + elif stype == 8: + return 0, pos # integer 0, no bytes + elif stype == 9: + return 1, pos # integer 1, no bytes + elif stype >= 13 and stype % 2 == 1: + length = (stype - 13) // 2 + return cell[pos:pos+length].decode('utf-8', errors='replace'), pos + length + elif stype >= 12 and stype % 2 == 0: + length = (stype - 12) // 2 + return bytes(cell[pos:pos+length]), pos + length + return None, pos + + +def _parse_leaf_page(page_data, page_offset=0): + """Parse a SQLite B-tree leaf table page and return list of record dicts.""" + records = [] + if not page_data or page_data[0] != 13: + return records # not a leaf table page + + num_cells = struct.unpack('>H', page_data[3:5])[0] + if num_cells == 0: + return records + + for i in range(num_cells): + try: + ptr = struct.unpack('>H', page_data[8 + i * 2: 10 + i * 2])[0] + cell = page_data[ptr:] + + payload_len, pos = _read_varint(cell, 0) + rowid, pos = _read_varint(cell, pos) + + header_start = pos + header_size, pos = _read_varint(cell, pos) + header_end = header_start + header_size + + col_types = [] + while pos < header_end: + stype, pos = _read_varint(cell, pos) + col_types.append(stype) + + record = {} + data_pos = pos + for j, stype in enumerate(col_types): + col_name = COL_NAMES[j] if j < len(COL_NAMES) else f'col{j}' + val, data_pos = _decode_value(cell, data_pos, stype) + record[col_name] = val + + if record.get('id'): + record['_cell_offset'] = page_offset + ptr + records.append(record) + except Exception: + continue + + return records + + +def _recover_from_wal(wal_path, data_page_num=4): + """ + Parse all WAL frames and collect every record from every version of + the data page, including frames before deletion. Each frame version + represents a separate update with a different last_updated timestamp, + so all are returned as individual rows rather than deduplicated. + """ + recovered = [] + try: + with open(wal_path, 'rb') as f: + wal_data = f.read() + + if len(wal_data) < 32: + return recovered + + magic = struct.unpack('>I', wal_data[:4])[0] + if magic not in (0x377f0682, 0x377f0683): + return recovered + + page_size = struct.unpack('>I', wal_data[8:12])[0] + frame_size = 24 + page_size + num_frames = (len(wal_data) - 32) // frame_size + + for i in range(num_frames): + frame_offset = 32 + i * frame_size + page_num = struct.unpack('>I', wal_data[frame_offset:frame_offset + 4])[0] + if page_num != data_page_num: + continue + + page_data = wal_data[frame_offset + 24: frame_offset + 24 + page_size] + page_offset = frame_offset + 24 + for record in _parse_leaf_page(page_data, page_offset): + if record.get('id'): + record['_wal_frame'] = i + 1 + record['_wal_offset'] = record.get('_cell_offset', '') + recovered.append(record) + + except Exception as e: + logfunc(f'Life360_NoShowAlerts WAL parse error: {e}') + + return recovered + + +def _find_wal(context): + """Find the WAL file — it may have a timestamp prefix.""" + for file_found in context.get_files_found(): + path = str(file_found) + if path.endswith('NoShowAlertRoomDatabase-wal'): + return path + return '' + + +@artifact_processor +def Life360_NoShowAlerts(context): + source = _find(context, 'NoShowAlertRoomDatabase') + wal_path = _find_wal(context) + data_list = [] + + # --- Live records from the main database --- + live_ids = set() + if source: + try: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + cursor.execute(''' + SELECT + last_updated, + run_at, + trigger_condition, + type, + place_id, + observed_user_id, + creator_id, + id + FROM no_show_alerts + ''') + for row in cursor.fetchall(): + live_ids.add(row[7]) + data_list.append(( + _ms_to_utc(row[0]), + _ns_to_utc(row[1]), + row[2], row[3], row[4], row[5], row[6], + 'Live', '', '' + )) + db.close() + except Exception as e: + logfunc(f'Life360_NoShowAlerts DB error: {e}') + + # --- Deleted records recovered from WAL --- + if wal_path: + recovered = _recover_from_wal(wal_path) + wal_count = 0 + for rec in recovered: + data_list.append(( + _ms_to_utc(rec.get('last_updated')), + _ns_to_utc(rec.get('run_at')), + rec.get('trigger_condition', ''), + rec.get('type', ''), + rec.get('place_id', ''), + rec.get('observed_user_id', ''), + rec.get('creator_id', ''), + 'Recovered from WAL', + f'WAL Frame {rec.get("_wal_frame", "")}', + rec.get('_wal_offset', '') + )) + wal_count += 1 + logfunc(f'Life360_NoShowAlerts: {wal_count} record(s) recovered from WAL') + + logfunc(f'Life360_NoShowAlerts: Total records = {len(data_list)}') + + data_headers = ( + ('Last Updated', 'datetime'), ('Run At', 'datetime'), + 'Trigger Condition', 'Type', + 'Place ID', 'Observed User ID', 'Creator ID', + 'Source', 'WAL Location', 'WAL Offset' + ) + return data_headers, data_list, source \ No newline at end of file diff --git a/scripts/artifacts/L360petprofile.py b/scripts/artifacts/L360petprofile.py new file mode 100644 index 00000000..dc8efb44 --- /dev/null +++ b/scripts/artifacts/L360petprofile.py @@ -0,0 +1,73 @@ +__artifacts_v2__ = { + 'Life360_PetProfile': { + 'name': 'Life360 PetProfile', + 'description': 'Parses Life360 PetProfile', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/PetProfileRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'github' + } +} + +from datetime import datetime, timezone, timedelta +from scripts.ilapfuncs import ( + artifact_processor, + get_file_path, + get_sqlite_db_records, + logfunc +) + +@artifact_processor +def Life360_PetProfile(context): + + data_list = [] + + files_found = context.get_files_found() + source_path = get_file_path(files_found, 'PetProfileRoomDatabase') + + query = ''' + SELECT + createdAt AS "Created Timestamp", + lastUpdated AS "Updated Timestamp", + type AS "Type", + petType AS "Pet Type", + breed AS "Breed", + color AS "Color", + weightInKg AS "Weight kg", + gender AS "Gender", + birthdate AS "Birthdate", + name AS "Name", + trackerId AS "Tracker ID", + primaryCircleId AS "Circle ID", + avatarBaseUrl AS "Avatar" + FROM pet_profile + ''' + + try: + db_records = get_sqlite_db_records(source_path, query) + + logfunc(f'Life360_PetProfile: Records found = {len(db_records)}') + + epoch = datetime(1970, 1, 1, tzinfo=timezone.utc) + + for record in db_records: + + created_timestamp = datetime.fromtimestamp(int(record[0]) / 1000, tz=timezone.utc) + updated_timestamp = datetime.fromtimestamp(int(record[1]) / 1000, tz=timezone.utc) + birthdate = None + if record[8] is not None: + birthdate = epoch + timedelta(days=int(record[8])) + + data_list.append((created_timestamp, updated_timestamp, record[2], record[3], record[4], record[5], record[6], record[7], birthdate, record[9], record[10], record[11], record[12])) + + except Exception as e: + logfunc(f'Error processing Life360 PetProfile: {e}') + + data_headers = (('Created Timestamp', 'datetime'), ('Updated Timestamp', 'datetime'), 'Type', 'Pet Type', 'Breed', 'Color', 'Weight kg', 'Gender', ('Birthdate', 'datetime'), 'Name', 'Tracker ID', 'Circle ID', 'Avatar') + + return data_headers, data_list, source_path \ No newline at end of file diff --git a/scripts/artifacts/L360places.py b/scripts/artifacts/L360places.py new file mode 100644 index 00000000..993a7e9f --- /dev/null +++ b/scripts/artifacts/L360places.py @@ -0,0 +1,151 @@ +# pylint: disable=W0613 +__artifacts_v2__ = { + 'Life360_Places': { + 'name': 'Life360 Places', + 'description': 'Parses Life360 Places', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-11', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360LocalStoreRoomDatabase*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'home' + }, + 'Life360_PlaceAlerts': { + 'name': 'Life360 Place Alerts', + 'description': 'Parses Life360 Place Alerts', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-11', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360LocalStoreRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'home' + }, + +} + +import datetime +import sqlite3 + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc + + +def _ms_to_utc(value): + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _bool(value): + if value is None: + return '' + return 'Yes' if int(value) == 1 else 'No' + + +def _find(context, suffix): + for file_found in context.get_files_found(): + file_found = str(file_found) + if file_found.endswith(suffix): + return file_found + return '' + + +def _q(cursor, sql): + try: + cursor.execute(sql) + return cursor.fetchall() + except sqlite3.Error: + return [] + + +@artifact_processor +def Life360_Places(context): + source = _find(context, 'L360LocalStoreRoomDatabase') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + place_id, + circle_id, + name, + latitude, + longitude, + radius, + source, + source_id, + owner_id, + has_alerts, + website, + types, + category + FROM places + ORDER BY name + '''): + data_list.append(( + row[0], row[1], row[2], + row[3], row[4], row[5], + row[6], row[7], row[8], + _bool(row[9]), + row[10], row[11], row[12] + )) + db.close() + + data_headers = ( + 'Place ID', 'Circle ID', 'Name', + 'Latitude', 'Longitude', 'Radius', + 'Source', 'Source ID', 'Owner ID', + 'Has Alerts', 'Website', 'Types', 'Category' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_PlaceAlerts(context): + source = _find(context, 'L360LocalStoreRoomDatabase') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + place_alerts.place_id, + place_alerts.circle_id, + place_alerts.member_id, + place_alerts.name, + place_alerts.arrive, + place_alerts.leave, + places.latitude, + places.longitude + FROM place_alerts + LEFT JOIN places ON places.place_id = place_alerts.place_id + ORDER BY place_alerts.name + '''): + data_list.append(( + row[0], row[1], row[2], + row[3], + _bool(row[4]), _bool(row[5]), + row[6], row[7] + )) + db.close() + + data_headers = ( + 'Place ID', 'Circle ID', 'Member ID', + 'Place Name', + 'Arrive Alert', 'Leave Alert', + 'Latitude', 'Longitude' + ) + return data_headers, data_list, source \ No newline at end of file diff --git a/scripts/artifacts/L360user.py b/scripts/artifacts/L360user.py new file mode 100644 index 00000000..c0b6a6c5 --- /dev/null +++ b/scripts/artifacts/L360user.py @@ -0,0 +1,54 @@ +__artifacts_v2__ = { + 'Life360_User': { + 'name': 'Life360 User', + 'description': 'Parses Life360 User Info', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-11', + 'last_update_date': '2026-06-12', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/com.amplitude.api*',), + 'output_types': 'standard', + 'artifact_icon': 'user' + } +} + +from datetime import datetime, timezone +from scripts.ilapfuncs import ( + artifact_processor, + get_file_path, + get_sqlite_db_records, + logfunc +) + +@artifact_processor +def Life360_User(context): + + data_list = [] + + files_found = context.get_files_found() + source_path = get_file_path(files_found, 'com.amplitude.api') + + query = ''' + SELECT + key AS "Key", + value AS "Value" + FROM store + ''' + + try: + db_records = get_sqlite_db_records(source_path, query) + + logfunc(f'Life360_User: Records found = {len(db_records)}') + + for record in db_records: + + data_list.append((record[0], record[1])) + + except Exception as e: + logfunc(f'Error processing Life360 User: {e}') + + data_headers = ('Key', 'Value') + + return data_headers, data_list, source_path \ No newline at end of file From 11dd3ff3f9bfb72d7e632de231092ed68ab4332b Mon Sep 17 00:00:00 2001 From: Brigs Date: Wed, 1 Jul 2026 14:27:00 -0400 Subject: [PATCH 4/8] Remove externalDB.py and revert mewe.py to main externalDB.py is unrelated to Life360 and had framework issues; mewe.py in this branch reverted the modern __artifacts_v2__/LAVA version on main back to legacy v1 with no LAVA support, so restore main's version. --- scripts/artifacts/externalDB.py | 90 ----------- scripts/artifacts/mewe.py | 276 +++++++++++++++----------------- 2 files changed, 127 insertions(+), 239 deletions(-) delete mode 100644 scripts/artifacts/externalDB.py diff --git a/scripts/artifacts/externalDB.py b/scripts/artifacts/externalDB.py deleted file mode 100644 index 43045349..00000000 --- a/scripts/artifacts/externalDB.py +++ /dev/null @@ -1,90 +0,0 @@ -__artifacts_v2__ = { - "externalDB": { - "name": "External Database", - "description": "Parses External Database", - "author": "Heather Charpentier", - "creation_date": "2025-11-17", - "last_updated_date": "2025-11-17", - "requirements": "none", - "category": "ExternalDB", - "notes": "", - "paths": ('*/data/com.android.providers.media/databases/external.db*'), - "output_types": "standard", - "artifact_icon": "download", - } -} - -from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly - -@artifact_processor -def externalDB(files_found, report_folder, seeker, wrap_text): - - data_list = [] - source_path = '' - - candidate_dbs = [] - - for file_found in files_found: - file_str = str(file_found).lower() - - if file_str.endswith('.db') and ( - 'media' in file_str or - 'external' in file_str or - 'internal' in file_str or - 'files' in file_str - ): - candidate_dbs.append(str(file_found)) - - if not candidate_dbs: - return ( - ('Path','Size',('Date Taken','datetime'),('Date Added','datetime'),('Date Modified','datetime'),'Album Name','Display Name','Directory','Duration'), - data_list, - source_path - ) - - for db_path in candidate_dbs: - - try: - db = open_sqlite_db_readonly(db_path) - source_path = db_path - - cursor = db.cursor() - - cursor.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name='files';" - ) - table_check = cursor.fetchone() - - if not table_check: - db.close() - continue - - cursor.execute(''' - SELECT - datetime(datetaken/1000, 'unixepoch') AS "Date Taken", - datetime(date_added, 'unixepoch') AS "Date Added", - datetime(date_modified, 'unixepoch') AS "Date Modified", - album AS "Album Name", - _display_name AS "Display Name", - _data AS "Path", - duration AS "Duration", - _size AS "Size" - FROM files - ''') - - rows = cursor.fetchall() - - for row in rows: - data_list.append(( - row[0], row[1], row[2], row[3], row[4],row[5], row[6], row[7] - )) - - db.close() - - except Exception as e: - continue - - data_headers = (('Date Taken','datetime'),('Date Added','datetime'),('Date Modified','datetime'),'Album Name','Display Name','Path','Duration','Size') - - return data_headers, data_list, source_path - diff --git a/scripts/artifacts/mewe.py b/scripts/artifacts/mewe.py index 866f1721..ee3ac969 100644 --- a/scripts/artifacts/mewe.py +++ b/scripts/artifacts/mewe.py @@ -1,149 +1,127 @@ -import xml.etree.ElementTree as ET - -from scripts.artifact_report import ArtifactHtmlReport -from scripts.ilapfuncs import logfunc, tsv, timeline, open_sqlite_db_readonly - -APP_NAME = 'MeWe' -DB_NAMES = ('app_database', 'app_v3.db') # Added support for app_v3.db -SGSESSION_FILE = 'SGSession.xml' - -CHAT_MESSAGES_QUERY = ''' - SELECT - DATETIME(createdAt, 'unixepoch'), - threadId, - groupId, - ownerId, - ownerName, - textPlain, - CASE currentUserMessage - WHEN 1 THEN "Sent" - ELSE "Received" - END currentUserMessage, - CASE attachmentType - WHEN "UNSUPPORTED" THEN '' - ELSE attachmentType - END attachmentType, - attachmentName, - CASE deleted - WHEN 1 THEN "YES" - ELSE "NO" - END deleted - FROM CHAT_MESSAGE - JOIN CHAT_THREAD ON threadId = CHAT_THREAD.id -''' - - -def _perform_query(cursor, query): - try: - cursor.execute(query) - rows = cursor.fetchall() - return len(rows), rows - except: - return 0, None - - -def _make_reports(title, data_headers, data_list, report_folder, db_file_name, tl_bool): - report = ArtifactHtmlReport(title) - report.start_artifact_report(report_folder, title) - report.add_script() - report.write_artifact_data_table(data_headers, data_list, db_file_name) - report.end_artifact_report() - - tsv(report_folder, data_headers, data_list, title, db_file_name) - - if tl_bool == True: - timeline(report_folder, title, data_list, data_headers) - - -def _parse_xml(xml_file, xml_file_name, report_folder, title, report_name): - logfunc(f'{title} found') - - tree = ET.parse(xml_file) - data_headers = ('Key', 'Value') - data_list = [] - - root = tree.getroot() - for node in root: - # skip not relevant keys - if '.' in node.attrib['name']: - continue - - value = None - try: - value = node.attrib['value'] - except: - value = node.text - - data_list.append((node.attrib['name'], value)) - - tl_bool = False - - _make_reports(f'{APP_NAME} - {report_name}', data_headers, data_list, report_folder, xml_file_name, tl_bool) - - -def _parse_chat_messages(messages_count, rows, report_folder, db_file_name): - logfunc(f'{messages_count} messages found') - - data_headers = ( - 'Timestamp', 'Thread Id', 'Thread Name', 'User Id', 'User Name', - 'Message Text', 'Message Direction', 'Message Type', - 'Attachment Name', 'Deleted' - ) - data_list = [( - row[0], row[1], row[2], row[3], row[4], row[5], - row[6], row[7], row[8] if row[8] else '', row[9] - ) for row in rows] - - tl_bool = True - - _make_reports(f'{APP_NAME} - Chat', data_headers, data_list, report_folder, db_file_name, tl_bool) - - -def _parse_app_database(db_file, db_file_name, report_folder): - db = open_sqlite_db_readonly(db_file) - cursor = db.cursor() - - messages_count, rows = _perform_query(cursor, CHAT_MESSAGES_QUERY) - if messages_count > 0 and rows: - _parse_chat_messages(messages_count, rows, report_folder, db_file_name) - else: - logfunc(f'No {APP_NAME} chat data found') - - cursor.close() - db.close() - - -def get_mewe(files_found, report_folder, seeker, wrap_text): - db_file = None - db_file_name = None - xml_file = None - xml_file_name = None - - app_database_processed = False - sgsession_processed = False - - for ff in files_found: - if ff.endswith(DB_NAMES) and not app_database_processed: - db_file = ff - db_file_name = ff.replace(seeker.data_folder, '') - _parse_app_database(db_file, db_file_name, report_folder) - app_database_processed = True - if ff.endswith(SGSESSION_FILE) and not sgsession_processed: - xml_file = ff - xml_file_name = ff.replace(seeker.data_folder, '') - _parse_xml(xml_file, xml_file_name, report_folder, SGSESSION_FILE, 'SGSession') - sgsession_processed = True - - artifacts = [ - app_database_processed, sgsession_processed - ] - if not (True in artifacts): - logfunc(f'{APP_NAME} data not found') - - -__artifacts__ = { - "mewe": ( - "MeWe", - ('*/com.mewe/databases/app_database', '*/com.mewe/databases/app_v3.db', '*/com.mewe/shared_prefs/SGSession.xml'), - get_mewe) -} +# pylint: disable=W0613,W0702 +__artifacts_v2__ = { + "get_mewe_chat": { + "name": "MeWe - Chat", + "description": "", + "author": "", + "creation_date": "2021-11-10", + "last_update_date": "2021-11-10", + "requirements": "none", + "category": "MeWe", + "notes": "", + "paths": ('*/com.mewe/databases/app_database',), + "output_types": "standard", + "artifact_icon": "message-square", + }, + "get_mewe_session": { + "name": "MeWe - SGSession", + "description": "", + "author": "", + "creation_date": "2021-11-10", + "last_update_date": "2021-11-10", + "requirements": "none", + "category": "MeWe", + "notes": "", + "paths": ('*/com.mewe/shared_prefs/SGSession.xml',), + "output_types": ['html', 'tsv', 'lava'], + "artifact_icon": "key", + } +} + +import datetime +import re +import xml.etree.ElementTree as ET + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc + +# Module-level constants (kept for backwards-compatibility; snapchat.py imports APP_NAME) +APP_NAME = 'MeWe' +DB_NAME = 'app_database' +SGSESSION_FILE = 'SGSession.xml' + +CHAT_MESSAGES_QUERY = ''' + SELECT + createdAt, + threadId, + groupId, + ownerId, + ownerName, + textPlain, + CASE currentUserMessage WHEN 1 THEN "Sent" ELSE "Received" END currentUserMessage, + CASE attachmentType WHEN "UNSUPPORTED" THEN '' ELSE attachmentType END attachmentType, + attachmentName, + CASE deleted WHEN 1 THEN "YES" ELSE "NO" END deleted + FROM CHAT_MESSAGE + JOIN CHAT_THREAD ON threadId = CHAT_THREAD.id +''' + + +INVALID_XML_CHARS = re.compile(r'[\x00-\x08\x0b\x0c\x0e-\x1f]') +BARE_AMPERSAND = re.compile(r'&(?!(?:amp|lt|gt|quot|apos|#\d+|#x[0-9A-Fa-f]+);)') + + +def _parse_xml(file_found): + """Parse XML, recovering from invalid tokens / unescaped ampersands; empty element if unparseable.""" + try: + return ET.parse(file_found).getroot() + except ET.ParseError: + with open(file_found, encoding='utf-8', errors='replace') as f: + xml = BARE_AMPERSAND.sub('&', INVALID_XML_CHARS.sub('', f.read())) + try: + return ET.fromstring(xml) + except ET.ParseError as ex: + logfunc(f'Skipping unparseable XML {file_found}: {ex}') + return ET.Element('empty') + + +@artifact_processor +def get_mewe_chat(files_found, report_folder, seeker, wrap_text): + data_list = [] + source_path = '' + for file_found in files_found: + file_found = str(file_found) + if not file_found.endswith(DB_NAME): + continue + + source_path = file_found + db = open_sqlite_db_readonly(file_found) + cursor = db.cursor() + try: + cursor.execute(CHAT_MESSAGES_QUERY) + rows = cursor.fetchall() + except: + rows = [] + db.close() + + for row in rows: + timestamp = datetime.datetime.fromtimestamp(int(row[0]), datetime.timezone.utc) if row[0] else '' + data_list.append((timestamp, row[1], row[2], row[3], row[4], row[5], row[6], row[7], + row[8] if row[8] else '', row[9])) + + data_headers = (('Timestamp', 'datetime'), 'Thread Id', 'Thread Name', 'User Id', 'User Name', + 'Message Text', 'Message Direction', 'Message Type', 'Attachment Name', 'Deleted') + return data_headers, data_list, source_path + + +@artifact_processor +def get_mewe_session(files_found, report_folder, seeker, wrap_text): + data_list = [] + source_path = '' + for file_found in files_found: + file_found = str(file_found) + if not file_found.endswith(SGSESSION_FILE): + continue + + source_path = file_found + root = _parse_xml(file_found) + for node in root: + if '.' in node.attrib['name']: + continue # skip not relevant keys + try: + value = node.attrib['value'] + except: + value = node.text + data_list.append((node.attrib['name'], value)) + + data_headers = ('Key', 'Value') + return data_headers, data_list, source_path From 00a56c4691c2d65355b92df5a0d93b15d4e88c45 Mon Sep 17 00:00:00 2001 From: charpy4n6 <112824930+charpy4n6@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:53:47 -0400 Subject: [PATCH 5/8] Update L360driveblade.py --- scripts/artifacts/L360driveblade.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/artifacts/L360driveblade.py b/scripts/artifacts/L360driveblade.py index e5ae18db..c5f67a6d 100644 --- a/scripts/artifacts/L360driveblade.py +++ b/scripts/artifacts/L360driveblade.py @@ -15,7 +15,7 @@ }, 'Life360_DriveEvents': { 'name': 'Life360 Drive Events', - 'description': 'Parses Life360 Drive Events', + 'description': 'Parses Life360 Drive Events (DriveBladeDB)', 'author': 'Heather Charpentier', 'creation_date': '2026-06-10', 'last_update_date': '2026-06-10', @@ -205,4 +205,4 @@ def Life360_DriveWaypoints(context): 'Latitude', 'Longitude', 'Speed MPS', 'Speed MPH', 'Accuracy' ) - return data_headers, data_list, source \ No newline at end of file + return data_headers, data_list, source From dc26a73f32dfef3210eda2d8ec9e10f1a1dc4a49 Mon Sep 17 00:00:00 2001 From: charpy4n6 <112824930+charpy4n6@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:54:23 -0400 Subject: [PATCH 6/8] Update L360eventstore.py --- scripts/artifacts/L360eventstore.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/artifacts/L360eventstore.py b/scripts/artifacts/L360eventstore.py index e59d1506..8a1cb6e5 100644 --- a/scripts/artifacts/L360eventstore.py +++ b/scripts/artifacts/L360eventstore.py @@ -40,7 +40,7 @@ }, 'Life360_Drive_Events': { 'name': 'Life360 Drive Events', - 'description': 'Parses Life360 Drive Events', + 'description': 'Parses Life360 Drive Events (EventStore)', 'author': 'Heather Charpentier', 'creation_date': '2026-06-10', 'last_update_date': '2026-06-30', @@ -246,4 +246,4 @@ def Life360_Drive_Events(context): 'Trip ID', ('Timestamp', 'datetime'), 'Latitude', 'Longitude', 'Type' ) - return data_headers, data_list, source \ No newline at end of file + return data_headers, data_list, source From 2843f4aa9a37d4133a1de51246e6cb2df16a6dbb Mon Sep 17 00:00:00 2001 From: Brigs Date: Wed, 1 Jul 2026 15:07:36 -0400 Subject: [PATCH 7/8] Lint cleanup: Life360 artifacts to 10.00/10 Remove unused imports (logfunc, sqlite3, datetime/timezone), drop two throwaway varint vars in the WAL parser, fix indentation in L360petprofile, and mark the defensive broad excepts with an inline pylint disable. No logic changes; all 14 Life360 artifacts load and lint clean. --- scripts/artifacts/L360circlesettings.py | 133 +++--- scripts/artifacts/L360driveblade.py | 416 +++++++++---------- scripts/artifacts/L360eventstore.py | 498 +++++++++++------------ scripts/artifacts/L360memberscircles.py | 142 +++---- scripts/artifacts/L360noshowalerts.py | 513 ++++++++++++------------ scripts/artifacts/L360petprofile.py | 144 +++---- scripts/artifacts/L360places.py | 300 +++++++------- scripts/artifacts/L360user.py | 105 +++-- 8 files changed, 1124 insertions(+), 1127 deletions(-) diff --git a/scripts/artifacts/L360circlesettings.py b/scripts/artifacts/L360circlesettings.py index f7c81d4f..19c66851 100644 --- a/scripts/artifacts/L360circlesettings.py +++ b/scripts/artifacts/L360circlesettings.py @@ -1,68 +1,67 @@ -__artifacts_v2__ = { - 'Life360_CircleSettings': { - 'name': 'Life360 CircleSettings', - 'description': 'Parses Life360 Circle Settings', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-11', - 'last_update_date': '2026-06-12', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/L360LocalStoreRoomDatabase*',), - 'output_types': 'standard', - 'artifact_icon': 'circle' - } -} - -from datetime import datetime, timezone -from scripts.ilapfuncs import ( - artifact_processor, - get_file_path, - get_sqlite_db_records, - logfunc -) - -@artifact_processor -def Life360_CircleSettings(context): - - data_list = [] - - files_found = context.get_files_found() - source_path = get_file_path(files_found, 'L360LocalStoreRoomDatabase') - - query = ''' - SELECT - je.key AS "Circle ID", - json_extract(je.value, '$.featureSetId') AS "App Features", - json_extract(je.value, '$.featureSetRefId') AS "App Features ID", - json_extract(je.value, '$.features.collision_alerts_push') AS "Collision Alerts Push", - json_extract(je.value, '$.features.collision_alerts_sms') AS "Collision Alerts SMS", - json_extract(je.value, '$.features.customer_support') AS "Customer Support", - json_extract(je.value, '$.features.data_breach_detection') AS "Data Breach Detection", - json_extract(je.value, '$.features.driver_reports') AS "Driver Reports", - json_extract(je.value, '$.features.location_history') AS "Location History", - json_extract(je.value, '$.features.place_alerts') AS "Place Alerts", - json_extract(je.value, '$.features.plan') AS "Plan", - json_extract(je.value, '$.features.sos_alerts_push') AS "SOS Alerts Push", - json_extract(je.value, '$.features.sos_alerts_sms') AS "SOS Alerts SMS", - json_extract(je.value, '$.features.tilegps_activation') AS "TileGPS Activation", - json_extract(je.value, '$.features.uber_one') AS "Uber One" - FROM Premium p - CROSS JOIN json_each(p.circleFeatureSetInfo) je; - ''' - - try: - db_records = get_sqlite_db_records(source_path, query) - - logfunc(f'Life360_CircleSettings: Records found = {len(db_records)}') - - for record in db_records: - - data_list.append((record[0], record[1], record[2], record[3], record[4], record[5], record[6], record[7], record[8], record[9], record[10], record[11], record[12], record[13], record[14])) - - except Exception as e: - logfunc(f'Error processing Life360 CircleSettings: {e}') - - data_headers = ('Circle ID', 'App Features', 'App Features ID', 'Collision Alerts Push', 'Collision Alerts SMS', 'Customer Support', 'Data Breach Detection', 'Driver Reports', 'Location History', 'Place Alerts', 'Plan', 'SOS Alerts Push', 'SOS Alerts SMS', 'TileGPS Activation', 'Uber One') - +__artifacts_v2__ = { + 'Life360_CircleSettings': { + 'name': 'Life360 CircleSettings', + 'description': 'Parses Life360 Circle Settings', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-11', + 'last_update_date': '2026-06-12', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360LocalStoreRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'circle' + } +} + +from scripts.ilapfuncs import ( + artifact_processor, + get_file_path, + get_sqlite_db_records, + logfunc +) + +@artifact_processor +def Life360_CircleSettings(context): + + data_list = [] + + files_found = context.get_files_found() + source_path = get_file_path(files_found, 'L360LocalStoreRoomDatabase') + + query = ''' + SELECT + je.key AS "Circle ID", + json_extract(je.value, '$.featureSetId') AS "App Features", + json_extract(je.value, '$.featureSetRefId') AS "App Features ID", + json_extract(je.value, '$.features.collision_alerts_push') AS "Collision Alerts Push", + json_extract(je.value, '$.features.collision_alerts_sms') AS "Collision Alerts SMS", + json_extract(je.value, '$.features.customer_support') AS "Customer Support", + json_extract(je.value, '$.features.data_breach_detection') AS "Data Breach Detection", + json_extract(je.value, '$.features.driver_reports') AS "Driver Reports", + json_extract(je.value, '$.features.location_history') AS "Location History", + json_extract(je.value, '$.features.place_alerts') AS "Place Alerts", + json_extract(je.value, '$.features.plan') AS "Plan", + json_extract(je.value, '$.features.sos_alerts_push') AS "SOS Alerts Push", + json_extract(je.value, '$.features.sos_alerts_sms') AS "SOS Alerts SMS", + json_extract(je.value, '$.features.tilegps_activation') AS "TileGPS Activation", + json_extract(je.value, '$.features.uber_one') AS "Uber One" + FROM Premium p + CROSS JOIN json_each(p.circleFeatureSetInfo) je; + ''' + + try: + db_records = get_sqlite_db_records(source_path, query) + + logfunc(f'Life360_CircleSettings: Records found = {len(db_records)}') + + for record in db_records: + + data_list.append((record[0], record[1], record[2], record[3], record[4], record[5], record[6], record[7], record[8], record[9], record[10], record[11], record[12], record[13], record[14])) + + except Exception as e: # pylint: disable=broad-exception-caught + logfunc(f'Error processing Life360 CircleSettings: {e}') + + data_headers = ('Circle ID', 'App Features', 'App Features ID', 'Collision Alerts Push', 'Collision Alerts SMS', 'Customer Support', 'Data Breach Detection', 'Driver Reports', 'Location History', 'Place Alerts', 'Plan', 'SOS Alerts Push', 'SOS Alerts SMS', 'TileGPS Activation', 'Uber One') + return data_headers, data_list, source_path \ No newline at end of file diff --git a/scripts/artifacts/L360driveblade.py b/scripts/artifacts/L360driveblade.py index c5f67a6d..6588c77f 100644 --- a/scripts/artifacts/L360driveblade.py +++ b/scripts/artifacts/L360driveblade.py @@ -1,208 +1,208 @@ -# pylint: disable=W0613 -__artifacts_v2__ = { - 'Life360_Drives': { - 'name': 'Life360 Drives', - 'description': 'Parses Life360 Drives', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-10', - 'last_update_date': '2026-06-10', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/DriveBladeDB*',), - 'output_types': 'standard', - 'artifact_icon': 'map-pin' - }, - 'Life360_DriveEvents': { - 'name': 'Life360 Drive Events', - 'description': 'Parses Life360 Drive Events (DriveBladeDB)', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-10', - 'last_update_date': '2026-06-10', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/DriveBladeDB*',), - 'output_types': ['html', 'tsv', 'lava', 'kml'], - 'artifact_icon': 'map-pin' - }, - 'Life360_DriveWaypoints': { - 'name': 'Life360 Drive Waypoints', - 'description': 'Parses Life360 Drive Waypoints', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-10', - 'last_update_date': '2026-06-10', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/DriveBladeDB*',), - 'output_types': ['html', 'tsv', 'lava', 'kml'], - 'artifact_icon': 'map-pin' - } -} - -import datetime -import sqlite3 - -from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc - - -def _ms_to_utc(value): - if not value: - return '' - try: - return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) - except (ValueError, OverflowError, OSError, TypeError): - return '' - - -def _find(context, suffix): - for file_found in context.get_files_found(): - file_found = str(file_found) - if file_found.endswith(suffix): - return file_found - return '' - - -def _q(cursor, sql): - try: - cursor.execute(sql) - return cursor.fetchall() - except sqlite3.Error: - return [] - - -@artifact_processor -def Life360_Drives(context): - source = _find(context, 'DriveBladeDB') - data_list = [] - - if source: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - - for row in _q(cursor, ''' - SELECT - driveId, - userId, - startTime, - endTime, - topSpeed, - topSpeed * 2.23694, - averageSpeed, - averageSpeed * 2.23694, - distance, - duration, - speedingCount, - hardBrakingCount, - rapidAccelerationCount, - distractedCount, - crashCount, - score, - updatedAt - FROM Drives - ORDER BY startTime - '''): - data_list.append(( - row[0], row[1], - _ms_to_utc(row[2]), _ms_to_utc(row[3]), - row[4], row[5], row[6], row[7], - row[8], row[9], - row[10], row[11], row[12], row[13], row[14], - row[15], _ms_to_utc(row[16]) - )) - db.close() - - data_headers = ( - 'Drive ID', 'User ID', - ('Start Time', 'datetime'), ('End Time', 'datetime'), - 'Top Speed MPS', 'Top Speed MPH', - 'Average Speed MPS', 'Average Speed MPH', - 'Distance', 'Duration', - 'Speeding Events', 'Hard Braking Events', - 'Rapid Acceleration Events', 'Distracted Events', 'Crash Events', - 'Score', ('Updated At', 'datetime') - ) - return data_headers, data_list, source - - -@artifact_processor -def Life360_DriveEvents(context): - source = _find(context, 'DriveBladeDB') - data_list = [] - - if source: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - - for row in _q(cursor, ''' - SELECT - DriveEvents.eventId, - DriveEvents.driveId, - Drives.userId, - DriveEvents.eventTime, - DriveEvents.eventType, - DriveEvents.lat, - DriveEvents.lon, - DriveEvents.speed, - DriveEvents.speed * 2.23694, - DriveEvents.accuracy - FROM DriveEvents - LEFT JOIN Drives ON Drives.driveId = DriveEvents.driveId - ORDER BY DriveEvents.eventTime - '''): - data_list.append(( - row[0], row[1], row[2], - _ms_to_utc(row[3]), - row[4], row[5], row[6], - row[7], row[8], row[9] - )) - db.close() - - data_headers = ( - 'Event ID', 'Drive ID', 'User ID', - ('Event Time', 'datetime'), - 'Event Type', 'Latitude', 'Longitude', - 'Speed MPS', 'Speed MPH', 'Accuracy' - ) - return data_headers, data_list, source - - -@artifact_processor -def Life360_DriveWaypoints(context): - source = _find(context, 'DriveBladeDB') - data_list = [] - - if source: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - - for row in _q(cursor, ''' - SELECT - DriveWaypoints.driveId, - Drives.userId, - DriveWaypoints.timestamp, - DriveWaypoints.lat, - DriveWaypoints.lon, - DriveWaypoints.speed, - DriveWaypoints.speed * 2.23694, - DriveWaypoints.accuracy - FROM DriveWaypoints - LEFT JOIN Drives ON Drives.driveId = DriveWaypoints.driveId - ORDER BY DriveWaypoints.timestamp - '''): - data_list.append(( - row[0], row[1], - _ms_to_utc(row[2]), - row[3], row[4], - row[5], row[6], row[7] - )) - db.close() - - data_headers = ( - 'Drive ID', 'User ID', - ('Timestamp', 'datetime'), - 'Latitude', 'Longitude', - 'Speed MPS', 'Speed MPH', 'Accuracy' - ) - return data_headers, data_list, source +# pylint: disable=W0613 +__artifacts_v2__ = { + 'Life360_Drives': { + 'name': 'Life360 Drives', + 'description': 'Parses Life360 Drives', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/DriveBladeDB*',), + 'output_types': 'standard', + 'artifact_icon': 'map-pin' + }, + 'Life360_DriveEvents': { + 'name': 'Life360 Drive Events', + 'description': 'Parses Life360 Drive Events (DriveBladeDB)', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/DriveBladeDB*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'map-pin' + }, + 'Life360_DriveWaypoints': { + 'name': 'Life360 Drive Waypoints', + 'description': 'Parses Life360 Drive Waypoints', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/DriveBladeDB*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'map-pin' + } +} + +import datetime +import sqlite3 + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly + + +def _ms_to_utc(value): + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _find(context, suffix): + for file_found in context.get_files_found(): + file_found = str(file_found) + if file_found.endswith(suffix): + return file_found + return '' + + +def _q(cursor, sql): + try: + cursor.execute(sql) + return cursor.fetchall() + except sqlite3.Error: + return [] + + +@artifact_processor +def Life360_Drives(context): + source = _find(context, 'DriveBladeDB') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + driveId, + userId, + startTime, + endTime, + topSpeed, + topSpeed * 2.23694, + averageSpeed, + averageSpeed * 2.23694, + distance, + duration, + speedingCount, + hardBrakingCount, + rapidAccelerationCount, + distractedCount, + crashCount, + score, + updatedAt + FROM Drives + ORDER BY startTime + '''): + data_list.append(( + row[0], row[1], + _ms_to_utc(row[2]), _ms_to_utc(row[3]), + row[4], row[5], row[6], row[7], + row[8], row[9], + row[10], row[11], row[12], row[13], row[14], + row[15], _ms_to_utc(row[16]) + )) + db.close() + + data_headers = ( + 'Drive ID', 'User ID', + ('Start Time', 'datetime'), ('End Time', 'datetime'), + 'Top Speed MPS', 'Top Speed MPH', + 'Average Speed MPS', 'Average Speed MPH', + 'Distance', 'Duration', + 'Speeding Events', 'Hard Braking Events', + 'Rapid Acceleration Events', 'Distracted Events', 'Crash Events', + 'Score', ('Updated At', 'datetime') + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_DriveEvents(context): + source = _find(context, 'DriveBladeDB') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + DriveEvents.eventId, + DriveEvents.driveId, + Drives.userId, + DriveEvents.eventTime, + DriveEvents.eventType, + DriveEvents.lat, + DriveEvents.lon, + DriveEvents.speed, + DriveEvents.speed * 2.23694, + DriveEvents.accuracy + FROM DriveEvents + LEFT JOIN Drives ON Drives.driveId = DriveEvents.driveId + ORDER BY DriveEvents.eventTime + '''): + data_list.append(( + row[0], row[1], row[2], + _ms_to_utc(row[3]), + row[4], row[5], row[6], + row[7], row[8], row[9] + )) + db.close() + + data_headers = ( + 'Event ID', 'Drive ID', 'User ID', + ('Event Time', 'datetime'), + 'Event Type', 'Latitude', 'Longitude', + 'Speed MPS', 'Speed MPH', 'Accuracy' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_DriveWaypoints(context): + source = _find(context, 'DriveBladeDB') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + DriveWaypoints.driveId, + Drives.userId, + DriveWaypoints.timestamp, + DriveWaypoints.lat, + DriveWaypoints.lon, + DriveWaypoints.speed, + DriveWaypoints.speed * 2.23694, + DriveWaypoints.accuracy + FROM DriveWaypoints + LEFT JOIN Drives ON Drives.driveId = DriveWaypoints.driveId + ORDER BY DriveWaypoints.timestamp + '''): + data_list.append(( + row[0], row[1], + _ms_to_utc(row[2]), + row[3], row[4], + row[5], row[6], row[7] + )) + db.close() + + data_headers = ( + 'Drive ID', 'User ID', + ('Timestamp', 'datetime'), + 'Latitude', 'Longitude', + 'Speed MPS', 'Speed MPH', 'Accuracy' + ) + return data_headers, data_list, source diff --git a/scripts/artifacts/L360eventstore.py b/scripts/artifacts/L360eventstore.py index 8a1cb6e5..a67555ed 100644 --- a/scripts/artifacts/L360eventstore.py +++ b/scripts/artifacts/L360eventstore.py @@ -1,249 +1,249 @@ -__artifacts_v2__ = { - 'Life360_Locations': { - 'name': 'Life360 Locations', - 'description': 'Parses Life360 Location Events', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-10', - 'last_update_date': '2026-06-30', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), - 'output_types': ['html', 'tsv', 'lava', 'kml'], - 'artifact_icon': 'map-pin' - }, - 'Life360_Waypoints': { - 'name': 'Life360 Waypoints', - 'description': 'Parses Life360 Waypoints', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-10', - 'last_update_date': '2026-06-30', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), - 'output_types': ['html', 'tsv', 'lava', 'kml'], - 'artifact_icon': 'map-pin' - }, - 'Life360_BatteryLevel': { - 'name': 'Life360 Battery Level', - 'description': 'Parses Life360 Battery Level Events', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-10', - 'last_update_date': '2026-06-30', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), - 'output_types': 'standard', - 'artifact_icon': 'battery-charging' - }, - 'Life360_Drive_Events': { - 'name': 'Life360 Drive Events', - 'description': 'Parses Life360 Drive Events (EventStore)', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-10', - 'last_update_date': '2026-06-30', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), - 'output_types': ['html', 'tsv', 'lava', 'kml'], - 'artifact_icon': 'truck' - } -} - -import datetime -import sqlite3 - -from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc - - -def _ms_to_utc(value): - if not value: - return '' - try: - return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) - except (ValueError, OverflowError, OSError, TypeError): - return '' - - -def _find(context, suffix): - for file_found in context.get_files_found(): - file_found = str(file_found) - if file_found.endswith(suffix): - return file_found - return '' - - -def _q(cursor, sql): - try: - cursor.execute(sql) - return cursor.fetchall() - except sqlite3.Error: - return [] - - -@artifact_processor -def Life360_Locations(context): - source = _find(context, 'L360EventStore_service.db') - data_list = [] - - if source: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - - for row in _q(cursor, ''' - SELECT - json_extract(event.data, '$.locationData.time') AS timestamp, - json_extract(event.data, '$.locationData.latitude') AS latitude, - json_extract(event.data, '$.locationData.longitude') AS longitude, - json_extract(event.data, '$.locationData.horizontalAccuracy') AS horizontal_accuracy, - json_extract(event.data, '$.locationData.speed') AS speed_mps, - json_extract(event.data, '$.locationData.speed') * 2.23694 AS speed_mph, - json_extract(event.data, '$.locationData.speedAccuracy') AS speed_accuracy - FROM event - WHERE eventVersion = 6 - ORDER BY timestamp - '''): - data_list.append(( - _ms_to_utc(row[0]), - row[1], row[2], row[3], - row[4], row[5], row[6] - )) - db.close() - - data_headers = ( - ('Timestamp', 'datetime'), 'Latitude', 'Longitude', - 'Horizontal Accuracy', 'Speed MPS', 'Speed MPH', 'Speed Accuracy' - ) - return data_headers, data_list, source - - -@artifact_processor -def Life360_Waypoints(context): - source = _find(context, 'L360EventStore_service.db') - data_list = [] - - if source: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - - for row in _q(cursor, ''' - SELECT - json_extract(event.data, '$.tripId') AS trip_id, - json_extract(json_each.value, '$.timestamp') AS timestamp, - json_extract(json_each.value, '$.latitude') AS latitude, - json_extract(json_each.value, '$.longitude') AS longitude, - json_extract(json_each.value, '$.accuracy') AS accuracy, - json_extract(json_each.value, '$.speed') AS speed_mps, - json_extract(json_each.value, '$.speed') * 2.23694 AS speed_mph - FROM event, - json_each(event.data, '$.type.waypoints') - WHERE event.eventVersion IN (1, 3, 4, 5) - ORDER BY timestamp - '''): - data_list.append(( - row[0], - _ms_to_utc(row[1]), - row[2], row[3], row[4], - row[5], row[6] - )) - db.close() - - data_headers = ( - 'Trip ID', ('Timestamp', 'datetime'), - 'Latitude', 'Longitude', 'Accuracy', - 'Speed MPS', 'Speed MPH' - ) - return data_headers, data_list, source - - -@artifact_processor -def Life360_BatteryLevel(context): - source = _find(context, 'L360EventStore_service.db') - data_list = [] - - if source: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - - for row in _q(cursor, ''' - SELECT DISTINCT - timestamp, - CASE - WHEN eventVersion IN (1, 2) - THEN json_extract(event.data, '$.batteryLevel') - WHEN eventVersion = 6 - THEN json_extract(event.data, '$.metaData.battery') - END AS battery_level, - CASE - WHEN eventVersion IN (1, 2) THEN - CASE - WHEN json_extract(event.data, '$.chargingState') = 1 THEN 'True' - WHEN json_extract(event.data, '$.chargingState') = 0 THEN 'False' - ELSE NULL - END - WHEN eventVersion = 6 THEN - CASE - WHEN json_extract(event.data, '$.metaData.chargingState') = 1 THEN 'True' - WHEN json_extract(event.data, '$.metaData.chargingState') = 0 THEN 'False' - ELSE NULL - END - END AS charging_state, - CASE - WHEN eventVersion = 2 - THEN json_extract(event.data, '$.powerMode') - ELSE NULL - END AS power_mode - FROM event - WHERE eventVersion IN (1, 2, 6) - AND (data LIKE '%batteryLevel%' OR data LIKE '%metaData%') - ORDER BY timestamp - '''): - data_list.append(( - _ms_to_utc(row[0]), - row[1], row[2], row[3] - )) - db.close() - - data_headers = ( - ('Timestamp', 'datetime'), - 'Battery Level', 'Charging State', 'Power Mode' - ) - return data_headers, data_list, source - - -@artifact_processor -def Life360_Drive_Events(context): - source = _find(context, 'L360EventStore_service.db') - data_list = [] - - if source: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - - for row in _q(cursor, ''' - SELECT - json_extract(event.data, '$.tripId') AS trip_id, - json_extract(json_each.value, '$.timestamp') AS timestamp, - json_extract(json_each.value, '$.latitude') AS latitude, - json_extract(json_each.value, '$.longitude') AS longitude, - json_extract(json_each.value, '$.type') AS type - FROM event, - json_each(event.data, '$.type.driveEvents') - WHERE event.eventVersion IN (3, 4, 5) - ORDER BY timestamp - '''): - data_list.append(( - row[0], - _ms_to_utc(row[1]), - row[2], row[3], row[4] - )) - db.close() - - data_headers = ( - 'Trip ID', ('Timestamp', 'datetime'), - 'Latitude', 'Longitude', 'Type' - ) - return data_headers, data_list, source +__artifacts_v2__ = { + 'Life360_Locations': { + 'name': 'Life360 Locations', + 'description': 'Parses Life360 Location Events', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'map-pin' + }, + 'Life360_Waypoints': { + 'name': 'Life360 Waypoints', + 'description': 'Parses Life360 Waypoints', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'map-pin' + }, + 'Life360_BatteryLevel': { + 'name': 'Life360 Battery Level', + 'description': 'Parses Life360 Battery Level Events', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), + 'output_types': 'standard', + 'artifact_icon': 'battery-charging' + }, + 'Life360_Drive_Events': { + 'name': 'Life360 Drive Events', + 'description': 'Parses Life360 Drive Events (EventStore)', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360EventStore_service.db*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'truck' + } +} + +import datetime +import sqlite3 + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly + + +def _ms_to_utc(value): + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _find(context, suffix): + for file_found in context.get_files_found(): + file_found = str(file_found) + if file_found.endswith(suffix): + return file_found + return '' + + +def _q(cursor, sql): + try: + cursor.execute(sql) + return cursor.fetchall() + except sqlite3.Error: + return [] + + +@artifact_processor +def Life360_Locations(context): + source = _find(context, 'L360EventStore_service.db') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + json_extract(event.data, '$.locationData.time') AS timestamp, + json_extract(event.data, '$.locationData.latitude') AS latitude, + json_extract(event.data, '$.locationData.longitude') AS longitude, + json_extract(event.data, '$.locationData.horizontalAccuracy') AS horizontal_accuracy, + json_extract(event.data, '$.locationData.speed') AS speed_mps, + json_extract(event.data, '$.locationData.speed') * 2.23694 AS speed_mph, + json_extract(event.data, '$.locationData.speedAccuracy') AS speed_accuracy + FROM event + WHERE eventVersion = 6 + ORDER BY timestamp + '''): + data_list.append(( + _ms_to_utc(row[0]), + row[1], row[2], row[3], + row[4], row[5], row[6] + )) + db.close() + + data_headers = ( + ('Timestamp', 'datetime'), 'Latitude', 'Longitude', + 'Horizontal Accuracy', 'Speed MPS', 'Speed MPH', 'Speed Accuracy' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_Waypoints(context): + source = _find(context, 'L360EventStore_service.db') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + json_extract(event.data, '$.tripId') AS trip_id, + json_extract(json_each.value, '$.timestamp') AS timestamp, + json_extract(json_each.value, '$.latitude') AS latitude, + json_extract(json_each.value, '$.longitude') AS longitude, + json_extract(json_each.value, '$.accuracy') AS accuracy, + json_extract(json_each.value, '$.speed') AS speed_mps, + json_extract(json_each.value, '$.speed') * 2.23694 AS speed_mph + FROM event, + json_each(event.data, '$.type.waypoints') + WHERE event.eventVersion IN (1, 3, 4, 5) + ORDER BY timestamp + '''): + data_list.append(( + row[0], + _ms_to_utc(row[1]), + row[2], row[3], row[4], + row[5], row[6] + )) + db.close() + + data_headers = ( + 'Trip ID', ('Timestamp', 'datetime'), + 'Latitude', 'Longitude', 'Accuracy', + 'Speed MPS', 'Speed MPH' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_BatteryLevel(context): + source = _find(context, 'L360EventStore_service.db') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT DISTINCT + timestamp, + CASE + WHEN eventVersion IN (1, 2) + THEN json_extract(event.data, '$.batteryLevel') + WHEN eventVersion = 6 + THEN json_extract(event.data, '$.metaData.battery') + END AS battery_level, + CASE + WHEN eventVersion IN (1, 2) THEN + CASE + WHEN json_extract(event.data, '$.chargingState') = 1 THEN 'True' + WHEN json_extract(event.data, '$.chargingState') = 0 THEN 'False' + ELSE NULL + END + WHEN eventVersion = 6 THEN + CASE + WHEN json_extract(event.data, '$.metaData.chargingState') = 1 THEN 'True' + WHEN json_extract(event.data, '$.metaData.chargingState') = 0 THEN 'False' + ELSE NULL + END + END AS charging_state, + CASE + WHEN eventVersion = 2 + THEN json_extract(event.data, '$.powerMode') + ELSE NULL + END AS power_mode + FROM event + WHERE eventVersion IN (1, 2, 6) + AND (data LIKE '%batteryLevel%' OR data LIKE '%metaData%') + ORDER BY timestamp + '''): + data_list.append(( + _ms_to_utc(row[0]), + row[1], row[2], row[3] + )) + db.close() + + data_headers = ( + ('Timestamp', 'datetime'), + 'Battery Level', 'Charging State', 'Power Mode' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_Drive_Events(context): + source = _find(context, 'L360EventStore_service.db') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + json_extract(event.data, '$.tripId') AS trip_id, + json_extract(json_each.value, '$.timestamp') AS timestamp, + json_extract(json_each.value, '$.latitude') AS latitude, + json_extract(json_each.value, '$.longitude') AS longitude, + json_extract(json_each.value, '$.type') AS type + FROM event, + json_each(event.data, '$.type.driveEvents') + WHERE event.eventVersion IN (3, 4, 5) + ORDER BY timestamp + '''): + data_list.append(( + row[0], + _ms_to_utc(row[1]), + row[2], row[3], row[4] + )) + db.close() + + data_headers = ( + 'Trip ID', ('Timestamp', 'datetime'), + 'Latitude', 'Longitude', 'Type' + ) + return data_headers, data_list, source diff --git a/scripts/artifacts/L360memberscircles.py b/scripts/artifacts/L360memberscircles.py index 3fc91bb2..347835db 100644 --- a/scripts/artifacts/L360memberscircles.py +++ b/scripts/artifacts/L360memberscircles.py @@ -1,72 +1,72 @@ -__artifacts_v2__ = { - 'Life360_MemberCircles': { - 'name': 'Life360 Members and Circles', - 'description': 'Parses Life360 Members and Circles', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-10', - 'last_update_date': '2026-06-10', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/MembersEngineRoomDatabase*',), - 'output_types': 'standard', - 'artifact_icon': 'user' - } -} - -from datetime import datetime, timezone -from scripts.ilapfuncs import ( - artifact_processor, - get_file_path, - get_sqlite_db_records, - logfunc -) - -@artifact_processor -def Life360_MemberCircles(context): - - data_list = [] - - files_found = context.get_files_found() - source_path = get_file_path(files_found, 'MembersEngineRoomDatabase') - - query = ''' - SELECT - members.created_at AS "Created Timestamp", - members.last_updated AS "Last Updated Timestamp", - members.id AS "Member ID", - members.first_name AS "First Name", - members.last_name AS "Last Name", - members.login_email AS "Email", - members.login_phone AS "Phone Number", - members.avatar AS "Avatar", - members.is_admin AS "Admin", - members.role AS "Role", - circles.created_at AS "Circle Created Timestamp", - circles.last_updated AS "Circle Last Updated Timestamp", - circles.name AS "Circle Name" - FROM members - LEFT JOIN circles - ON members.circle_id = circles.id - ''' - - try: - db_records = get_sqlite_db_records(source_path, query) - - logfunc(f'Life360_MemberCircles: Records found = {len(db_records)}') - - for record in db_records: - - created_timestamp = datetime.fromtimestamp(int(record[0]), tz=timezone.utc) - updated_timestamp = datetime.fromtimestamp(int(record[1]) / 1000, tz=timezone.utc) - second_created_timestamp = datetime.fromtimestamp(int(record[10]), tz=timezone.utc) - second_updated_timestamp = datetime.fromtimestamp(int(record[11]) / 1000, tz=timezone.utc) - - data_list.append((created_timestamp, updated_timestamp, record[2], record[3], record[4], record[5], record[6], record[7], record[8], record[9], second_created_timestamp,second_updated_timestamp, record[12])) - - except Exception as e: - logfunc(f'Error processing Life360 MemberCircles: {e}') - - data_headers = (('Created Timestamp', 'datetime'), ('Updated Timestamp', 'datetime'), 'Member ID', 'First Name', 'Last Name', 'Email', 'Phone Number', 'Avatar', 'Admin', 'Role', ('Circle Created Timestamp', 'datetime'), ('Circle Updated Timestamp', 'datetime'), 'Circle Name') - +__artifacts_v2__ = { + 'Life360_MemberCircles': { + 'name': 'Life360 Members and Circles', + 'description': 'Parses Life360 Members and Circles', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/MembersEngineRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'user' + } +} + +from datetime import datetime, timezone +from scripts.ilapfuncs import ( + artifact_processor, + get_file_path, + get_sqlite_db_records, + logfunc +) + +@artifact_processor +def Life360_MemberCircles(context): + + data_list = [] + + files_found = context.get_files_found() + source_path = get_file_path(files_found, 'MembersEngineRoomDatabase') + + query = ''' + SELECT + members.created_at AS "Created Timestamp", + members.last_updated AS "Last Updated Timestamp", + members.id AS "Member ID", + members.first_name AS "First Name", + members.last_name AS "Last Name", + members.login_email AS "Email", + members.login_phone AS "Phone Number", + members.avatar AS "Avatar", + members.is_admin AS "Admin", + members.role AS "Role", + circles.created_at AS "Circle Created Timestamp", + circles.last_updated AS "Circle Last Updated Timestamp", + circles.name AS "Circle Name" + FROM members + LEFT JOIN circles + ON members.circle_id = circles.id + ''' + + try: + db_records = get_sqlite_db_records(source_path, query) + + logfunc(f'Life360_MemberCircles: Records found = {len(db_records)}') + + for record in db_records: + + created_timestamp = datetime.fromtimestamp(int(record[0]), tz=timezone.utc) + updated_timestamp = datetime.fromtimestamp(int(record[1]) / 1000, tz=timezone.utc) + second_created_timestamp = datetime.fromtimestamp(int(record[10]), tz=timezone.utc) + second_updated_timestamp = datetime.fromtimestamp(int(record[11]) / 1000, tz=timezone.utc) + + data_list.append((created_timestamp, updated_timestamp, record[2], record[3], record[4], record[5], record[6], record[7], record[8], record[9], second_created_timestamp,second_updated_timestamp, record[12])) + + except Exception as e: # pylint: disable=broad-exception-caught + logfunc(f'Error processing Life360 MemberCircles: {e}') + + data_headers = (('Created Timestamp', 'datetime'), ('Updated Timestamp', 'datetime'), 'Member ID', 'First Name', 'Last Name', 'Email', 'Phone Number', 'Avatar', 'Admin', 'Role', ('Circle Created Timestamp', 'datetime'), ('Circle Updated Timestamp', 'datetime'), 'Circle Name') + return data_headers, data_list, source_path \ No newline at end of file diff --git a/scripts/artifacts/L360noshowalerts.py b/scripts/artifacts/L360noshowalerts.py index 519c450e..d93b09d9 100644 --- a/scripts/artifacts/L360noshowalerts.py +++ b/scripts/artifacts/L360noshowalerts.py @@ -1,258 +1,257 @@ -# pylint: disable=W0613 -__artifacts_v2__ = { - 'Life360_NoShowAlerts': { - 'name': 'Life360 No Show Alerts', - 'description': 'Parses Life360 No Show Alerts including records recovered from WAL', - 'author': 'Heather Charpentier', - 'creation_date': '2026-07-01', - 'last_update_date': '2026-07-01', - 'requirements': 'none', - 'category': 'Life360', - 'notes': 'WAL file recovery', - 'paths': ('*/com.life360.android.safetymapd/databases/NoShowAlertRoomDatabase*',), - 'output_types': 'standard', - 'artifact_icon': 'alert-triangle' - } -} - -import datetime -import struct -import sqlite3 - -from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc - - -COL_NAMES = [ - 'id', 'type', 'last_updated', 'trigger_condition', - 'critical_alert', 'run_at', 'place_id', - 'observed_user_id', 'creator_id', 'circle_id', 'daily' -] - - -def _ms_to_utc(value): - if not value: - return '' - try: - return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) - except (ValueError, OverflowError, OSError, TypeError): - return '' - - -def _ns_to_utc(value): - """run_at is stored in nanoseconds.""" - if not value: - return '' - try: - return datetime.datetime.fromtimestamp(int(value) / 1_000_000_000, datetime.timezone.utc) - except (ValueError, OverflowError, OSError, TypeError): - return '' - - -def _find(context, suffix): - for file_found in context.get_files_found(): - if str(file_found).endswith(suffix): - return str(file_found) - return '' - - -def _read_varint(data, pos): - result = 0 - for i in range(9): - byte = data[pos + i] - result = (result << 7) | (byte & 0x7f) - if not (byte & 0x80): - return result, pos + i + 1 - return result, pos + 9 - - -def _decode_value(cell, pos, stype): - """Decode a SQLite serial type value from cell data.""" - if stype == 0: - return None, pos - elif stype == 1: - return int.from_bytes(cell[pos:pos+1], 'big', signed=True), pos + 1 - elif stype == 2: - return int.from_bytes(cell[pos:pos+2], 'big', signed=True), pos + 2 - elif stype == 3: - return int.from_bytes(cell[pos:pos+3], 'big', signed=True), pos + 3 - elif stype == 4: - return int.from_bytes(cell[pos:pos+4], 'big', signed=True), pos + 4 - elif stype == 5: - return int.from_bytes(cell[pos:pos+6], 'big', signed=True), pos + 6 - elif stype == 6: - return int.from_bytes(cell[pos:pos+8], 'big', signed=True), pos + 8 - elif stype == 7: - return struct.unpack('>d', cell[pos:pos+8])[0], pos + 8 - elif stype == 8: - return 0, pos # integer 0, no bytes - elif stype == 9: - return 1, pos # integer 1, no bytes - elif stype >= 13 and stype % 2 == 1: - length = (stype - 13) // 2 - return cell[pos:pos+length].decode('utf-8', errors='replace'), pos + length - elif stype >= 12 and stype % 2 == 0: - length = (stype - 12) // 2 - return bytes(cell[pos:pos+length]), pos + length - return None, pos - - -def _parse_leaf_page(page_data, page_offset=0): - """Parse a SQLite B-tree leaf table page and return list of record dicts.""" - records = [] - if not page_data or page_data[0] != 13: - return records # not a leaf table page - - num_cells = struct.unpack('>H', page_data[3:5])[0] - if num_cells == 0: - return records - - for i in range(num_cells): - try: - ptr = struct.unpack('>H', page_data[8 + i * 2: 10 + i * 2])[0] - cell = page_data[ptr:] - - payload_len, pos = _read_varint(cell, 0) - rowid, pos = _read_varint(cell, pos) - - header_start = pos - header_size, pos = _read_varint(cell, pos) - header_end = header_start + header_size - - col_types = [] - while pos < header_end: - stype, pos = _read_varint(cell, pos) - col_types.append(stype) - - record = {} - data_pos = pos - for j, stype in enumerate(col_types): - col_name = COL_NAMES[j] if j < len(COL_NAMES) else f'col{j}' - val, data_pos = _decode_value(cell, data_pos, stype) - record[col_name] = val - - if record.get('id'): - record['_cell_offset'] = page_offset + ptr - records.append(record) - except Exception: - continue - - return records - - -def _recover_from_wal(wal_path, data_page_num=4): - """ - Parse all WAL frames and collect every record from every version of - the data page, including frames before deletion. Each frame version - represents a separate update with a different last_updated timestamp, - so all are returned as individual rows rather than deduplicated. - """ - recovered = [] - try: - with open(wal_path, 'rb') as f: - wal_data = f.read() - - if len(wal_data) < 32: - return recovered - - magic = struct.unpack('>I', wal_data[:4])[0] - if magic not in (0x377f0682, 0x377f0683): - return recovered - - page_size = struct.unpack('>I', wal_data[8:12])[0] - frame_size = 24 + page_size - num_frames = (len(wal_data) - 32) // frame_size - - for i in range(num_frames): - frame_offset = 32 + i * frame_size - page_num = struct.unpack('>I', wal_data[frame_offset:frame_offset + 4])[0] - if page_num != data_page_num: - continue - - page_data = wal_data[frame_offset + 24: frame_offset + 24 + page_size] - page_offset = frame_offset + 24 - for record in _parse_leaf_page(page_data, page_offset): - if record.get('id'): - record['_wal_frame'] = i + 1 - record['_wal_offset'] = record.get('_cell_offset', '') - recovered.append(record) - - except Exception as e: - logfunc(f'Life360_NoShowAlerts WAL parse error: {e}') - - return recovered - - -def _find_wal(context): - """Find the WAL file — it may have a timestamp prefix.""" - for file_found in context.get_files_found(): - path = str(file_found) - if path.endswith('NoShowAlertRoomDatabase-wal'): - return path - return '' - - -@artifact_processor -def Life360_NoShowAlerts(context): - source = _find(context, 'NoShowAlertRoomDatabase') - wal_path = _find_wal(context) - data_list = [] - - # --- Live records from the main database --- - live_ids = set() - if source: - try: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - cursor.execute(''' - SELECT - last_updated, - run_at, - trigger_condition, - type, - place_id, - observed_user_id, - creator_id, - id - FROM no_show_alerts - ''') - for row in cursor.fetchall(): - live_ids.add(row[7]) - data_list.append(( - _ms_to_utc(row[0]), - _ns_to_utc(row[1]), - row[2], row[3], row[4], row[5], row[6], - 'Live', '', '' - )) - db.close() - except Exception as e: - logfunc(f'Life360_NoShowAlerts DB error: {e}') - - # --- Deleted records recovered from WAL --- - if wal_path: - recovered = _recover_from_wal(wal_path) - wal_count = 0 - for rec in recovered: - data_list.append(( - _ms_to_utc(rec.get('last_updated')), - _ns_to_utc(rec.get('run_at')), - rec.get('trigger_condition', ''), - rec.get('type', ''), - rec.get('place_id', ''), - rec.get('observed_user_id', ''), - rec.get('creator_id', ''), - 'Recovered from WAL', - f'WAL Frame {rec.get("_wal_frame", "")}', - rec.get('_wal_offset', '') - )) - wal_count += 1 - logfunc(f'Life360_NoShowAlerts: {wal_count} record(s) recovered from WAL') - - logfunc(f'Life360_NoShowAlerts: Total records = {len(data_list)}') - - data_headers = ( - ('Last Updated', 'datetime'), ('Run At', 'datetime'), - 'Trigger Condition', 'Type', - 'Place ID', 'Observed User ID', 'Creator ID', - 'Source', 'WAL Location', 'WAL Offset' - ) +# pylint: disable=W0613 +__artifacts_v2__ = { + 'Life360_NoShowAlerts': { + 'name': 'Life360 No Show Alerts', + 'description': 'Parses Life360 No Show Alerts including records recovered from WAL', + 'author': 'Heather Charpentier', + 'creation_date': '2026-07-01', + 'last_update_date': '2026-07-01', + 'requirements': 'none', + 'category': 'Life360', + 'notes': 'WAL file recovery', + 'paths': ('*/com.life360.android.safetymapd/databases/NoShowAlertRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'alert-triangle' + } +} + +import datetime +import struct + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc + + +COL_NAMES = [ + 'id', 'type', 'last_updated', 'trigger_condition', + 'critical_alert', 'run_at', 'place_id', + 'observed_user_id', 'creator_id', 'circle_id', 'daily' +] + + +def _ms_to_utc(value): + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _ns_to_utc(value): + """run_at is stored in nanoseconds.""" + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1_000_000_000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _find(context, suffix): + for file_found in context.get_files_found(): + if str(file_found).endswith(suffix): + return str(file_found) + return '' + + +def _read_varint(data, pos): + result = 0 + for i in range(9): + byte = data[pos + i] + result = (result << 7) | (byte & 0x7f) + if not (byte & 0x80): + return result, pos + i + 1 + return result, pos + 9 + + +def _decode_value(cell, pos, stype): + """Decode a SQLite serial type value from cell data.""" + if stype == 0: + return None, pos + elif stype == 1: + return int.from_bytes(cell[pos:pos+1], 'big', signed=True), pos + 1 + elif stype == 2: + return int.from_bytes(cell[pos:pos+2], 'big', signed=True), pos + 2 + elif stype == 3: + return int.from_bytes(cell[pos:pos+3], 'big', signed=True), pos + 3 + elif stype == 4: + return int.from_bytes(cell[pos:pos+4], 'big', signed=True), pos + 4 + elif stype == 5: + return int.from_bytes(cell[pos:pos+6], 'big', signed=True), pos + 6 + elif stype == 6: + return int.from_bytes(cell[pos:pos+8], 'big', signed=True), pos + 8 + elif stype == 7: + return struct.unpack('>d', cell[pos:pos+8])[0], pos + 8 + elif stype == 8: + return 0, pos # integer 0, no bytes + elif stype == 9: + return 1, pos # integer 1, no bytes + elif stype >= 13 and stype % 2 == 1: + length = (stype - 13) // 2 + return cell[pos:pos+length].decode('utf-8', errors='replace'), pos + length + elif stype >= 12 and stype % 2 == 0: + length = (stype - 12) // 2 + return bytes(cell[pos:pos+length]), pos + length + return None, pos + + +def _parse_leaf_page(page_data, page_offset=0): + """Parse a SQLite B-tree leaf table page and return list of record dicts.""" + records = [] + if not page_data or page_data[0] != 13: + return records # not a leaf table page + + num_cells = struct.unpack('>H', page_data[3:5])[0] + if num_cells == 0: + return records + + for i in range(num_cells): + try: + ptr = struct.unpack('>H', page_data[8 + i * 2: 10 + i * 2])[0] + cell = page_data[ptr:] + + _, pos = _read_varint(cell, 0) + _, pos = _read_varint(cell, pos) + + header_start = pos + header_size, pos = _read_varint(cell, pos) + header_end = header_start + header_size + + col_types = [] + while pos < header_end: + stype, pos = _read_varint(cell, pos) + col_types.append(stype) + + record = {} + data_pos = pos + for j, stype in enumerate(col_types): + col_name = COL_NAMES[j] if j < len(COL_NAMES) else f'col{j}' + val, data_pos = _decode_value(cell, data_pos, stype) + record[col_name] = val + + if record.get('id'): + record['_cell_offset'] = page_offset + ptr + records.append(record) + except Exception: # pylint: disable=broad-exception-caught + continue + + return records + + +def _recover_from_wal(wal_path, data_page_num=4): + """ + Parse all WAL frames and collect every record from every version of + the data page, including frames before deletion. Each frame version + represents a separate update with a different last_updated timestamp, + so all are returned as individual rows rather than deduplicated. + """ + recovered = [] + try: + with open(wal_path, 'rb') as f: + wal_data = f.read() + + if len(wal_data) < 32: + return recovered + + magic = struct.unpack('>I', wal_data[:4])[0] + if magic not in (0x377f0682, 0x377f0683): + return recovered + + page_size = struct.unpack('>I', wal_data[8:12])[0] + frame_size = 24 + page_size + num_frames = (len(wal_data) - 32) // frame_size + + for i in range(num_frames): + frame_offset = 32 + i * frame_size + page_num = struct.unpack('>I', wal_data[frame_offset:frame_offset + 4])[0] + if page_num != data_page_num: + continue + + page_data = wal_data[frame_offset + 24: frame_offset + 24 + page_size] + page_offset = frame_offset + 24 + for record in _parse_leaf_page(page_data, page_offset): + if record.get('id'): + record['_wal_frame'] = i + 1 + record['_wal_offset'] = record.get('_cell_offset', '') + recovered.append(record) + + except Exception as e: # pylint: disable=broad-exception-caught + logfunc(f'Life360_NoShowAlerts WAL parse error: {e}') + + return recovered + + +def _find_wal(context): + """Find the WAL file — it may have a timestamp prefix.""" + for file_found in context.get_files_found(): + path = str(file_found) + if path.endswith('NoShowAlertRoomDatabase-wal'): + return path + return '' + + +@artifact_processor +def Life360_NoShowAlerts(context): + source = _find(context, 'NoShowAlertRoomDatabase') + wal_path = _find_wal(context) + data_list = [] + + # --- Live records from the main database --- + live_ids = set() + if source: + try: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + cursor.execute(''' + SELECT + last_updated, + run_at, + trigger_condition, + type, + place_id, + observed_user_id, + creator_id, + id + FROM no_show_alerts + ''') + for row in cursor.fetchall(): + live_ids.add(row[7]) + data_list.append(( + _ms_to_utc(row[0]), + _ns_to_utc(row[1]), + row[2], row[3], row[4], row[5], row[6], + 'Live', '', '' + )) + db.close() + except Exception as e: # pylint: disable=broad-exception-caught + logfunc(f'Life360_NoShowAlerts DB error: {e}') + + # --- Deleted records recovered from WAL --- + if wal_path: + recovered = _recover_from_wal(wal_path) + wal_count = 0 + for rec in recovered: + data_list.append(( + _ms_to_utc(rec.get('last_updated')), + _ns_to_utc(rec.get('run_at')), + rec.get('trigger_condition', ''), + rec.get('type', ''), + rec.get('place_id', ''), + rec.get('observed_user_id', ''), + rec.get('creator_id', ''), + 'Recovered from WAL', + f'WAL Frame {rec.get("_wal_frame", "")}', + rec.get('_wal_offset', '') + )) + wal_count += 1 + logfunc(f'Life360_NoShowAlerts: {wal_count} record(s) recovered from WAL') + + logfunc(f'Life360_NoShowAlerts: Total records = {len(data_list)}') + + data_headers = ( + ('Last Updated', 'datetime'), ('Run At', 'datetime'), + 'Trigger Condition', 'Type', + 'Place ID', 'Observed User ID', 'Creator ID', + 'Source', 'WAL Location', 'WAL Offset' + ) return data_headers, data_list, source \ No newline at end of file diff --git a/scripts/artifacts/L360petprofile.py b/scripts/artifacts/L360petprofile.py index dc8efb44..6c93c310 100644 --- a/scripts/artifacts/L360petprofile.py +++ b/scripts/artifacts/L360petprofile.py @@ -1,73 +1,73 @@ -__artifacts_v2__ = { - 'Life360_PetProfile': { - 'name': 'Life360 PetProfile', - 'description': 'Parses Life360 PetProfile', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-10', - 'last_update_date': '2026-06-10', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/PetProfileRoomDatabase*',), - 'output_types': 'standard', - 'artifact_icon': 'github' - } -} - -from datetime import datetime, timezone, timedelta -from scripts.ilapfuncs import ( - artifact_processor, - get_file_path, - get_sqlite_db_records, - logfunc -) - -@artifact_processor -def Life360_PetProfile(context): - - data_list = [] - - files_found = context.get_files_found() - source_path = get_file_path(files_found, 'PetProfileRoomDatabase') - - query = ''' - SELECT - createdAt AS "Created Timestamp", - lastUpdated AS "Updated Timestamp", - type AS "Type", - petType AS "Pet Type", - breed AS "Breed", - color AS "Color", - weightInKg AS "Weight kg", - gender AS "Gender", - birthdate AS "Birthdate", - name AS "Name", - trackerId AS "Tracker ID", - primaryCircleId AS "Circle ID", - avatarBaseUrl AS "Avatar" - FROM pet_profile - ''' - - try: - db_records = get_sqlite_db_records(source_path, query) - - logfunc(f'Life360_PetProfile: Records found = {len(db_records)}') - - epoch = datetime(1970, 1, 1, tzinfo=timezone.utc) - - for record in db_records: - - created_timestamp = datetime.fromtimestamp(int(record[0]) / 1000, tz=timezone.utc) - updated_timestamp = datetime.fromtimestamp(int(record[1]) / 1000, tz=timezone.utc) - birthdate = None - if record[8] is not None: - birthdate = epoch + timedelta(days=int(record[8])) - - data_list.append((created_timestamp, updated_timestamp, record[2], record[3], record[4], record[5], record[6], record[7], birthdate, record[9], record[10], record[11], record[12])) - - except Exception as e: - logfunc(f'Error processing Life360 PetProfile: {e}') - - data_headers = (('Created Timestamp', 'datetime'), ('Updated Timestamp', 'datetime'), 'Type', 'Pet Type', 'Breed', 'Color', 'Weight kg', 'Gender', ('Birthdate', 'datetime'), 'Name', 'Tracker ID', 'Circle ID', 'Avatar') - +__artifacts_v2__ = { + 'Life360_PetProfile': { + 'name': 'Life360 PetProfile', + 'description': 'Parses Life360 PetProfile', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-10', + 'last_update_date': '2026-06-10', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/PetProfileRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'github' + } +} + +from datetime import datetime, timezone, timedelta +from scripts.ilapfuncs import ( + artifact_processor, + get_file_path, + get_sqlite_db_records, + logfunc +) + +@artifact_processor +def Life360_PetProfile(context): + + data_list = [] + + files_found = context.get_files_found() + source_path = get_file_path(files_found, 'PetProfileRoomDatabase') + + query = ''' + SELECT + createdAt AS "Created Timestamp", + lastUpdated AS "Updated Timestamp", + type AS "Type", + petType AS "Pet Type", + breed AS "Breed", + color AS "Color", + weightInKg AS "Weight kg", + gender AS "Gender", + birthdate AS "Birthdate", + name AS "Name", + trackerId AS "Tracker ID", + primaryCircleId AS "Circle ID", + avatarBaseUrl AS "Avatar" + FROM pet_profile + ''' + + try: + db_records = get_sqlite_db_records(source_path, query) + + logfunc(f'Life360_PetProfile: Records found = {len(db_records)}') + + epoch = datetime(1970, 1, 1, tzinfo=timezone.utc) + + for record in db_records: + + created_timestamp = datetime.fromtimestamp(int(record[0]) / 1000, tz=timezone.utc) + updated_timestamp = datetime.fromtimestamp(int(record[1]) / 1000, tz=timezone.utc) + birthdate = None + if record[8] is not None: + birthdate = epoch + timedelta(days=int(record[8])) + + data_list.append((created_timestamp, updated_timestamp, record[2], record[3], record[4], record[5], record[6], record[7], birthdate, record[9], record[10], record[11], record[12])) + + except Exception as e: # pylint: disable=broad-exception-caught + logfunc(f'Error processing Life360 PetProfile: {e}') + + data_headers = (('Created Timestamp', 'datetime'), ('Updated Timestamp', 'datetime'), 'Type', 'Pet Type', 'Breed', 'Color', 'Weight kg', 'Gender', ('Birthdate', 'datetime'), 'Name', 'Tracker ID', 'Circle ID', 'Avatar') + return data_headers, data_list, source_path \ No newline at end of file diff --git a/scripts/artifacts/L360places.py b/scripts/artifacts/L360places.py index 993a7e9f..61753da2 100644 --- a/scripts/artifacts/L360places.py +++ b/scripts/artifacts/L360places.py @@ -1,151 +1,151 @@ -# pylint: disable=W0613 -__artifacts_v2__ = { - 'Life360_Places': { - 'name': 'Life360 Places', - 'description': 'Parses Life360 Places', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-11', - 'last_update_date': '2026-06-30', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/L360LocalStoreRoomDatabase*',), - 'output_types': ['html', 'tsv', 'lava', 'kml'], - 'artifact_icon': 'home' - }, - 'Life360_PlaceAlerts': { - 'name': 'Life360 Place Alerts', - 'description': 'Parses Life360 Place Alerts', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-11', - 'last_update_date': '2026-06-30', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/L360LocalStoreRoomDatabase*',), - 'output_types': 'standard', - 'artifact_icon': 'home' - }, - -} - -import datetime -import sqlite3 - -from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly, logfunc - - -def _ms_to_utc(value): - if not value: - return '' - try: - return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) - except (ValueError, OverflowError, OSError, TypeError): - return '' - - -def _bool(value): - if value is None: - return '' - return 'Yes' if int(value) == 1 else 'No' - - -def _find(context, suffix): - for file_found in context.get_files_found(): - file_found = str(file_found) - if file_found.endswith(suffix): - return file_found - return '' - - -def _q(cursor, sql): - try: - cursor.execute(sql) - return cursor.fetchall() - except sqlite3.Error: - return [] - - -@artifact_processor -def Life360_Places(context): - source = _find(context, 'L360LocalStoreRoomDatabase') - data_list = [] - - if source: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - - for row in _q(cursor, ''' - SELECT - place_id, - circle_id, - name, - latitude, - longitude, - radius, - source, - source_id, - owner_id, - has_alerts, - website, - types, - category - FROM places - ORDER BY name - '''): - data_list.append(( - row[0], row[1], row[2], - row[3], row[4], row[5], - row[6], row[7], row[8], - _bool(row[9]), - row[10], row[11], row[12] - )) - db.close() - - data_headers = ( - 'Place ID', 'Circle ID', 'Name', - 'Latitude', 'Longitude', 'Radius', - 'Source', 'Source ID', 'Owner ID', - 'Has Alerts', 'Website', 'Types', 'Category' - ) - return data_headers, data_list, source - - -@artifact_processor -def Life360_PlaceAlerts(context): - source = _find(context, 'L360LocalStoreRoomDatabase') - data_list = [] - - if source: - db = open_sqlite_db_readonly(source) - cursor = db.cursor() - - for row in _q(cursor, ''' - SELECT - place_alerts.place_id, - place_alerts.circle_id, - place_alerts.member_id, - place_alerts.name, - place_alerts.arrive, - place_alerts.leave, - places.latitude, - places.longitude - FROM place_alerts - LEFT JOIN places ON places.place_id = place_alerts.place_id - ORDER BY place_alerts.name - '''): - data_list.append(( - row[0], row[1], row[2], - row[3], - _bool(row[4]), _bool(row[5]), - row[6], row[7] - )) - db.close() - - data_headers = ( - 'Place ID', 'Circle ID', 'Member ID', - 'Place Name', - 'Arrive Alert', 'Leave Alert', - 'Latitude', 'Longitude' - ) +# pylint: disable=W0613 +__artifacts_v2__ = { + 'Life360_Places': { + 'name': 'Life360 Places', + 'description': 'Parses Life360 Places', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-11', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360LocalStoreRoomDatabase*',), + 'output_types': ['html', 'tsv', 'lava', 'kml'], + 'artifact_icon': 'home' + }, + 'Life360_PlaceAlerts': { + 'name': 'Life360 Place Alerts', + 'description': 'Parses Life360 Place Alerts', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-11', + 'last_update_date': '2026-06-30', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/L360LocalStoreRoomDatabase*',), + 'output_types': 'standard', + 'artifact_icon': 'home' + }, + +} + +import datetime +import sqlite3 + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly + + +def _ms_to_utc(value): + if not value: + return '' + try: + return datetime.datetime.fromtimestamp(int(value) / 1000, datetime.timezone.utc) + except (ValueError, OverflowError, OSError, TypeError): + return '' + + +def _bool(value): + if value is None: + return '' + return 'Yes' if int(value) == 1 else 'No' + + +def _find(context, suffix): + for file_found in context.get_files_found(): + file_found = str(file_found) + if file_found.endswith(suffix): + return file_found + return '' + + +def _q(cursor, sql): + try: + cursor.execute(sql) + return cursor.fetchall() + except sqlite3.Error: + return [] + + +@artifact_processor +def Life360_Places(context): + source = _find(context, 'L360LocalStoreRoomDatabase') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + place_id, + circle_id, + name, + latitude, + longitude, + radius, + source, + source_id, + owner_id, + has_alerts, + website, + types, + category + FROM places + ORDER BY name + '''): + data_list.append(( + row[0], row[1], row[2], + row[3], row[4], row[5], + row[6], row[7], row[8], + _bool(row[9]), + row[10], row[11], row[12] + )) + db.close() + + data_headers = ( + 'Place ID', 'Circle ID', 'Name', + 'Latitude', 'Longitude', 'Radius', + 'Source', 'Source ID', 'Owner ID', + 'Has Alerts', 'Website', 'Types', 'Category' + ) + return data_headers, data_list, source + + +@artifact_processor +def Life360_PlaceAlerts(context): + source = _find(context, 'L360LocalStoreRoomDatabase') + data_list = [] + + if source: + db = open_sqlite_db_readonly(source) + cursor = db.cursor() + + for row in _q(cursor, ''' + SELECT + place_alerts.place_id, + place_alerts.circle_id, + place_alerts.member_id, + place_alerts.name, + place_alerts.arrive, + place_alerts.leave, + places.latitude, + places.longitude + FROM place_alerts + LEFT JOIN places ON places.place_id = place_alerts.place_id + ORDER BY place_alerts.name + '''): + data_list.append(( + row[0], row[1], row[2], + row[3], + _bool(row[4]), _bool(row[5]), + row[6], row[7] + )) + db.close() + + data_headers = ( + 'Place ID', 'Circle ID', 'Member ID', + 'Place Name', + 'Arrive Alert', 'Leave Alert', + 'Latitude', 'Longitude' + ) return data_headers, data_list, source \ No newline at end of file diff --git a/scripts/artifacts/L360user.py b/scripts/artifacts/L360user.py index c0b6a6c5..2059a1f5 100644 --- a/scripts/artifacts/L360user.py +++ b/scripts/artifacts/L360user.py @@ -1,54 +1,53 @@ -__artifacts_v2__ = { - 'Life360_User': { - 'name': 'Life360 User', - 'description': 'Parses Life360 User Info', - 'author': 'Heather Charpentier', - 'creation_date': '2026-06-11', - 'last_update_date': '2026-06-12', - 'requirements': 'none', - 'category': 'Life360', - 'notes': '', - 'paths': ('*/com.life360.android.safetymapd/databases/com.amplitude.api*',), - 'output_types': 'standard', - 'artifact_icon': 'user' - } -} - -from datetime import datetime, timezone -from scripts.ilapfuncs import ( - artifact_processor, - get_file_path, - get_sqlite_db_records, - logfunc -) - -@artifact_processor -def Life360_User(context): - - data_list = [] - - files_found = context.get_files_found() - source_path = get_file_path(files_found, 'com.amplitude.api') - - query = ''' - SELECT - key AS "Key", - value AS "Value" - FROM store - ''' - - try: - db_records = get_sqlite_db_records(source_path, query) - - logfunc(f'Life360_User: Records found = {len(db_records)}') - - for record in db_records: - - data_list.append((record[0], record[1])) - - except Exception as e: - logfunc(f'Error processing Life360 User: {e}') - - data_headers = ('Key', 'Value') - +__artifacts_v2__ = { + 'Life360_User': { + 'name': 'Life360 User', + 'description': 'Parses Life360 User Info', + 'author': 'Heather Charpentier', + 'creation_date': '2026-06-11', + 'last_update_date': '2026-06-12', + 'requirements': 'none', + 'category': 'Life360', + 'notes': '', + 'paths': ('*/com.life360.android.safetymapd/databases/com.amplitude.api*',), + 'output_types': 'standard', + 'artifact_icon': 'user' + } +} + +from scripts.ilapfuncs import ( + artifact_processor, + get_file_path, + get_sqlite_db_records, + logfunc +) + +@artifact_processor +def Life360_User(context): + + data_list = [] + + files_found = context.get_files_found() + source_path = get_file_path(files_found, 'com.amplitude.api') + + query = ''' + SELECT + key AS "Key", + value AS "Value" + FROM store + ''' + + try: + db_records = get_sqlite_db_records(source_path, query) + + logfunc(f'Life360_User: Records found = {len(db_records)}') + + for record in db_records: + + data_list.append((record[0], record[1])) + + except Exception as e: # pylint: disable=broad-exception-caught + logfunc(f'Error processing Life360 User: {e}') + + data_headers = ('Key', 'Value') + return data_headers, data_list, source_path \ No newline at end of file From fe674ac3b815fd40c634d7f6d59adfec3ac877f1 Mon Sep 17 00:00:00 2001 From: Brigs Date: Wed, 1 Jul 2026 15:10:06 -0400 Subject: [PATCH 8/8] Disambiguate duplicate 'Life360 Drive Events' artifact names The DriveBladeDB and EventStore variants shared the identical display name, which collides on the HTML report filename/title and LAVA table. Rename to 'Life360 Drive Events (DriveBladeDB)' and 'Life360 Drive Events (EventStore)' to mirror their descriptions. --- scripts/artifacts/L360driveblade.py | 2 +- scripts/artifacts/L360eventstore.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/artifacts/L360driveblade.py b/scripts/artifacts/L360driveblade.py index 6588c77f..496efbe6 100644 --- a/scripts/artifacts/L360driveblade.py +++ b/scripts/artifacts/L360driveblade.py @@ -14,7 +14,7 @@ 'artifact_icon': 'map-pin' }, 'Life360_DriveEvents': { - 'name': 'Life360 Drive Events', + 'name': 'Life360 Drive Events (DriveBladeDB)', 'description': 'Parses Life360 Drive Events (DriveBladeDB)', 'author': 'Heather Charpentier', 'creation_date': '2026-06-10', diff --git a/scripts/artifacts/L360eventstore.py b/scripts/artifacts/L360eventstore.py index a67555ed..48abcd04 100644 --- a/scripts/artifacts/L360eventstore.py +++ b/scripts/artifacts/L360eventstore.py @@ -39,7 +39,7 @@ 'artifact_icon': 'battery-charging' }, 'Life360_Drive_Events': { - 'name': 'Life360 Drive Events', + 'name': 'Life360 Drive Events (EventStore)', 'description': 'Parses Life360 Drive Events (EventStore)', 'author': 'Heather Charpentier', 'creation_date': '2026-06-10',