diff --git a/scripts/artifacts/L360circlesettings.py b/scripts/artifacts/L360circlesettings.py new file mode 100644 index 00000000..19c66851 --- /dev/null +++ b/scripts/artifacts/L360circlesettings.py @@ -0,0 +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 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 new file mode 100644 index 00000000..496efbe6 --- /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 (DriveBladeDB)', + '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 new file mode 100644 index 00000000..48abcd04 --- /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 (EventStore)', + '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 new file mode 100644 index 00000000..347835db --- /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: # 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 new file mode 100644 index 00000000..d93b09d9 --- /dev/null +++ b/scripts/artifacts/L360noshowalerts.py @@ -0,0 +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 + +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 new file mode 100644 index 00000000..6c93c310 --- /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: # 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 new file mode 100644 index 00000000..61753da2 --- /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 + + +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..2059a1f5 --- /dev/null +++ b/scripts/artifacts/L360user.py @@ -0,0 +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 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 diff --git a/scripts/artifacts/mewe.py b/scripts/artifacts/mewe.py index fc9f9400..ee3ac969 100644 --- a/scripts/artifacts/mewe.py +++ b/scripts/artifacts/mewe.py @@ -1,146 +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_NAME = 'app_database' -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_NAME) 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/shared_prefs/SGSession.xml'), - get_mewe) -} \ No newline at end of file +# 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