diff --git a/scripts/artifacts/meta_ai.py b/scripts/artifacts/meta_ai.py new file mode 100644 index 0000000..3369208 --- /dev/null +++ b/scripts/artifacts/meta_ai.py @@ -0,0 +1,329 @@ +import json +import os +import re +import sqlite3 +import xml.etree.ElementTree as ET +from html import unescape + +from scripts.context import Context +from scripts.ilapfuncs import ( + artifact_processor, + logfunc, + open_sqlite_db_readonly, +) + +__artifacts_v2__ = { + "meta_ai_conversations": { + "name": "Meta AI - AI Conversations (Cloud)", + "description": "Conversation history from Meta AI cloud data export", + "author": "Shishir Panta", + "creation_date": "2026-04-12", + "last_update_date": "2026-06-15", + "requirements": "none", + "category": "Meta AI", + "notes": "", + "paths": ("*/meta_ai_app/*.html",), + "output_types": ["html", "lava", "tsv"], + "artifact_icon": "message-square", + }, + "meta_ai_connected_devices_cloud": { + "name": "Meta AI - Connected Devices (Cloud)", + "description": "Device connection history from Meta AI cloud data export", + "author": "Shishir Panta", + "creation_date": "2026-04-12", + "last_update_date": "2026-06-15", + "requirements": "none", + "category": "Meta AI", + "notes": "", + "paths": ("*/meta_ai_profile/*.html",), + "output_types": ["html", "lava", "tsv"], + "artifact_icon": "bluetooth", + }, + "meta_ai_cloud_media": { + "name": "Meta AI - Cloud Media Library", + "description": "Media library entries from Meta AI cloud data export", + "author": "Shishir Panta", + "creation_date": "2026-04-12", + "last_update_date": "2026-06-15", + "requirements": "none", + "category": "Meta AI", + "notes": "", + "paths": ("*/facebook_view/media/*",), + "output_types": ["html", "lava", "tsv"], + "artifact_icon": "image", + }, +} + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _read_xml_key_value(file_path): + """Reads key-value pairs from an XML SharedPreferences file.""" + try: + tree = ET.parse(file_path) + root = tree.getroot() + return {child.attrib.get('name', 'unknown'): child.text for child in root} + except (ET.ParseError, FileNotFoundError, KeyError) as e: + logfunc(f"[Meta AI] Could not parse XML file {os.path.basename(file_path)}: {e}") + return {} + + +def _parse_binary_prefs(file_path): + """Parses binary SharedPreferences files (non-XML format).""" + try: + with open(file_path, 'rb') as f: + content = f.read() + + text = content.decode('utf-8', errors='ignore') + text = text.replace('\x00', ' ').replace('\r', ' ').replace('\n', ' ') + text = ' '.join(text.split()) + + data = {} + + extractions = [ + ('device_serial', ['device_frame_color_name', 'soc_build']), + ('device_uuid', ['cloud_ota_error', 'build_flags_key']), + ('btc_address', ['device_frame_type', 'device_identifier']), + ('device_identifier', ['device_lens_color_name', 'feature_key']), + ('device_frame_type_short_name', ['device_lens_color']), + ('device_frame_color_name', ['soc_build', 'mcu_build', 'build_flavor']), + ('device_lens_color_name', ['feature_key', 'sku_code']), + ('mcu_build', ['device_type']), + ('soc_build', ['build_flavor']), + ('device_type', ['device_hardware_type']), + ('device_hardware_type', ['device_uuid']), + ] + + for key, terminators in extractions: + key_pos = text.find(key) + if key_pos == -1: + continue + value_start = key_pos + len(key) + value_end = len(text) + for terminator in terminators: + term_pos = text.find(terminator, value_start) + if term_pos != -1 and term_pos < value_end: + value_end = term_pos + value = text[value_start:value_end].strip() + if value.startswith('$'): + value = value[1:] + parts = value.split() + value = parts[0] if len(parts) == 1 else ' '.join(parts[:4]) + if value: + data[key] = value + + return data + + except Exception as e: + logfunc(f"[Meta AI] Could not parse binary prefs file {os.path.basename(file_path)}: {e}") + return {} + + +def _ms_to_unix(timestamp_ms): + """Converts a millisecond timestamp to Unix seconds (float) for LAVA datetime columns.""" + if not timestamp_ms: + return None + try: + return int(timestamp_ms) / 1000.0 + except (ValueError, TypeError): + return None + + +def _parse_device_files(): + """ + Parses app_light_prefs files using the Context filename lookup map for + O(1) access by filename rather than iterating the full file list. + Returns (paired_devices dict, meta_accounts list). + """ + lookup = Context.get_filename_lookup_map() + paired_devices = {} + meta_accounts = [] + + # connectivity_metadata.xml — direct lookup + for file_path in lookup.get('connectivity_metadata.xml', []): + data = _read_xml_key_value(file_path) + mac = data.get('DEVICE-METADATA-ID', 'Unknown') + paired_devices.setdefault(mac, {}) + paired_devices[mac]['mac'] = mac + paired_devices[mac]['serial'] = data.get('serialNumber', '') + + # device_system_info_ — variable suffix, scan map keys once + for filename, paths in lookup.items(): + if not filename.startswith('device_system_info_'): + continue + mac = filename.replace('device_system_info_', '') + paired_devices.setdefault(mac, {}) + for file_path in paths: + data = _parse_binary_prefs(file_path) + if data: + paired_devices[mac]['mac'] = data.get('device_identifier', mac) + paired_devices[mac]['btc'] = data.get('btc_address', '') + paired_devices[mac]['serial'] = data.get('device_serial', '') + paired_devices[mac]['uuid'] = data.get('device_uuid', '') + paired_devices[mac]['frame'] = data.get('device_frame_type_short_name', '') + paired_devices[mac]['frame_color'] = data.get('device_frame_color_name', '') + paired_devices[mac]['lens'] = data.get('device_lens_color_name', '') + paired_devices[mac]['mcu_build'] = data.get('mcu_build', '') + paired_devices[mac]['soc_build'] = data.get('soc_build', '') + + # meta_fx_cache — direct lookup + for file_path in lookup.get('meta_fx_cache', []): + try: + with open(file_path, 'r', encoding='utf-8') as f: + cache_data = json.load(f) + for account in cache_data.get('accounts', []): + meta_accounts.append(( + account.get('platform', 'Unknown'), + account.get('username', account.get('email', '')), + account.get('account_id', ''), + )) + except (json.JSONDecodeError, UnicodeDecodeError): + try: + with open(file_path, 'r', encoding='latin-1', errors='ignore') as f: + content = f.read() + account_ids = re.findall(r'"account_id"\s*:\s*"?(\d+)"?', content) + platforms = re.findall(r'"account_type"\s*:\s*"([^"]+)"', content) + usernames = re.findall(r'"(?:username|email)"\s*:\s*"([^"]+)"', content) + for i, account_id in enumerate(account_ids): + meta_accounts.append(( + platforms[i] if i < len(platforms) else 'Unknown', + usernames[i] if i < len(usernames) else 'N/A', + account_id, + )) + except Exception as e: + logfunc(f"[Meta AI] Could not parse meta_fx_cache: {e}") + + return paired_devices, meta_accounts + + +# --------------------------------------------------------------------------- +# Artifact processors +# --------------------------------------------------------------------------- + + + +@artifact_processor +def meta_ai_conversations(context): + data_headers = ( + 'Date', + 'Speaker', + 'Message', + ) + data_list = [] + source_path = '' + + for source_path in context.get_files_found(): + try: + with open(source_path, 'r', encoding='utf-8', errors='ignore') as f: + html_content = f.read() + + conv_dates = re.findall( + r'Conversation with Meta AI_(\d{2}-\d{2}-\d{4})_\d+\.txt', + html_content, + ) + conv_blocks = re.findall( + r'(Conversation with Meta AI.*?)', + html_content, + re.DOTALL, + ) + + for i, block in enumerate(conv_blocks): + block = unescape(block) + messages = re.findall( + r'(You|Meta AI): (.+?)(?=(?:You|Meta AI):|$)', + block, + re.DOTALL, + ) + conv_date = conv_dates[i] if i < len(conv_dates) else 'Unknown' + for speaker, message in messages: + message_clean = message.strip().replace('\n', ' ') + if message_clean and message_clean != 'Conversation with Meta AI': + data_list.append((conv_date, speaker, message_clean)) + + except Exception as e: + logfunc(f"[Meta AI] Could not parse cloud conversations from {os.path.basename(source_path)}: {e}") + + return data_headers, data_list, source_path + + +@artifact_processor +def meta_ai_connected_devices_cloud(context): + data_headers = ( + 'Serial Number', + 'Last Update', + ) + data_list = [] + source_path = '' + + for source_path in context.get_files_found(): + try: + with open(source_path, 'r', encoding='utf-8', errors='ignore') as f: + html_content = f.read() + + serial_match = re.search( + r'Serial number\s*([^<]+)', + html_content, + ) + time_match = re.search( + r'Update time\s*([^<]+)', + html_content, + ) + serial = serial_match.group(1) if serial_match else 'Unknown' + update_time = time_match.group(1) if time_match else '' + + if serial != 'Unknown': + data_list.append((serial, update_time)) + + except Exception as e: + logfunc(f"[Meta AI] Could not parse cloud devices from {os.path.basename(source_path)}: {e}") + + return data_headers, data_list, source_path + + +@artifact_processor +def meta_ai_cloud_media(context): + data_headers = ( + 'Device ID', + 'Timestamp', + 'Media File Path', + ) + data_list = [] + source_path = '' + + for source_path in context.get_files_found(): + try: + with open(source_path, 'r', encoding='utf-8', errors='ignore') as f: + html_content = f.read() + + sections = re.findall( + r'
.*?
', + html_content, + re.DOTALL, + ) + + for section in sections: + device_match = re.search( + r'Device ID\s*([^<]+)', + section, + ) + media_match = re.search( + r'href="(posts/media/your_posts/[^"]+)"', + section, + ) + time_match = re.search( + r'([A-Z][a-z]{2} \d{2}, \d{4} \d{1,2}:\d{2} [ap]m)', + section, + ) + device_id = device_match.group(1) if device_match else 'Unknown' + media_path = media_match.group(1) if media_match else '' + timestamp = time_match.group(1) if time_match else '' + + if media_path: + data_list.append((device_id, timestamp, media_path)) + + except Exception as e: + logfunc(f"[Meta AI] Could not parse cloud media from {os.path.basename(source_path)}: {e}") + + return data_headers, data_list, source_path