diff --git a/scripts/artifacts/hyundai_callHistory.py b/scripts/artifacts/hyundai_callHistory.py index 2dd8818..5612f60 100644 --- a/scripts/artifacts/hyundai_callHistory.py +++ b/scripts/artifacts/hyundai_callHistory.py @@ -1,35 +1,76 @@ __artifacts_v2__ = { "hyundaiCallHistory": { "name": "Hyundai - Call History", - "description": "Bluetooth call history from a Hyundai infotainment CH_*.db.", - "author": "Nixy Camacho", - "version": "0.2", + "description": "Bluetooth call history per connected phone from Hyundai/Kia infotainment CH_{mac}.db databases.", + "author": "Nixy Camacho, @pmpulkownik", + "version": "0.3", "creation_date": "2023-06-09", - "last_update_date": "2026-06-29", + "last_update_date": "2026-09-02", "requirements": "none", "category": "Hyundai Vehicles", - "notes": "Rewritten as a single query (the original ran nine separate SELECTs and crashed on " - "os.path.splittext). date / date_sort are interpreted as Unix epochs and " - "normalized to UTC.", + "notes": "Extracts call logs per device and derives device Bluetooth MAC directly from the database filename (CH_{mac}.db).", "paths": ('*/bluetooth/DB_BMS/CH_*.db*',), "output_types": "standard", "artifact_icon": "phone-call", } } +import os +import re +import sqlite3 from datetime import datetime, timezone -from scripts.ilapfuncs import artifact_processor, convert_unix_ts_to_utc, open_sqlite_db_readonly +from scripts.ilapfuncs import (artifact_processor, convert_unix_ts_to_utc, logfunc, + open_sqlite_db_readonly) + +# Ordered as queried. A column missing from a phone's database is reported empty +# rather than costing that phone every row -- see _present_columns. +_CALL_COLUMNS = ('date', 'date_sort', '_id', 'given_name', 'family_name', 'phone_number', + 'calltype', 'duration', 'numberType') + + +def _format_mac_from_filename(filename): + """ + Extracts and formats MAC address from filename like CH_AABBCCDDEEFF.db -> AA:BB:CC:DD:EE:FF + """ + base = os.path.basename(filename) + stem = base.rsplit('.', 1)[0] + raw_mac = stem[3:] if stem.startswith('CH_') else stem + + clean_hex = re.sub(r'[^A-Fa-f0-9]', '', raw_mac) + if len(clean_hex) == 12: + return ':'.join(clean_hex[i:i + 2] for i in range(0, 12, 2)).upper() + return raw_mac.upper() + + +def _present_columns(cursor, table, wanted): + try: + cursor.execute(f'PRAGMA table_info({table})') + present = {row[1] for row in cursor.fetchall()} + except sqlite3.Error: + return [] + return [column for column in wanted if column in present] + + +def _value(record, column): + """The stored value, or '' when the column is absent or NULL.""" + value = record.get(column) + return '' if value is None else value def _ts(value): - if value is None or value == '': + if value is None or value == '' or value == 0: return '' if isinstance(value, (int, float)): + if value > 1e11: + value = value / 1000.0 return convert_unix_ts_to_utc(value) text = str(value).strip() if text.isdigit(): - return convert_unix_ts_to_utc(int(text)) + val = int(text) + if val > 1e11: + val = val / 1000.0 + return convert_unix_ts_to_utc(val) try: dt = datetime.fromisoformat(text.replace('Z', '+00:00')) return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt.astimezone(timezone.utc) @@ -40,24 +81,83 @@ def _ts(value): @artifact_processor def hyundaiCallHistory(context): data_list = [] - source_path = '' + source_paths = [] + for file_found in context.get_files_found(): file_found = str(file_found) - if not file_found.endswith('.db'): + if os.path.isdir(file_found) or not file_found.endswith('.db'): continue - source_path = file_found + + device_mac = _format_mac_from_filename(file_found) + db_filename = os.path.basename(file_found) + db = open_sqlite_db_readonly(file_found) + if db is None: + continue cursor = db.cursor() - cursor.execute(''' - SELECT _id, given_name, family_name, phone_number, calltype, date, date_sort, - duration, numberType - FROM bluetooth_callhistory - ''') - for row in cursor.fetchall(): - data_list.append((row[0], row[1], row[2], row[3], row[4], _ts(row[5]), _ts(row[6]), - row[7], row[8])) + + columns = _present_columns(cursor, 'bluetooth_callhistory', _CALL_COLUMNS) + if not columns: + logfunc(f"{db_filename}: no bluetooth_callhistory table, skipped") + db.close() + continue + + statement = f'SELECT {", ".join(columns)} FROM bluetooth_callhistory' + if 'date' in columns: + statement += ' ORDER BY date DESC' + + try: + cursor.execute(statement) + rows = cursor.fetchall() + except sqlite3.Error as ex: + logfunc(f"{db_filename}: reading bluetooth_callhistory failed, skipped") + logfunc(f" - {str(ex)}") + db.close() + continue + db.close() + source_paths.append(file_found) + + missing = [column for column in _CALL_COLUMNS if column not in columns] + if missing: + logfunc(f"{db_filename}: reported without {', '.join(missing)}") + + for row in rows: + record = dict(zip(columns, row)) + given_name = _value(record, 'given_name') + family_name = _value(record, 'family_name') + + name_parts = [str(p).strip() for p in (given_name, family_name) if p] + full_name = ' '.join(name_parts) if name_parts else '' + + data_list.append(( + _ts(record.get('date')), + _ts(record.get('date_sort')), + device_mac, + _value(record, '_id'), + full_name, + given_name, + family_name, + _value(record, 'phone_number'), + _value(record, 'calltype'), + _value(record, 'duration'), + _value(record, 'numberType'), + db_filename + )) + + data_headers = ( + ('Date (UTC)', 'datetime'), + ('Date Sort (UTC)', 'datetime'), + 'Device MAC', + 'Record ID', + 'Full Name', + 'First Name', + 'Last Name', + ('Phone Number', 'phonenumber'), + 'Call Type (as stored)', + 'Duration (as stored)', + 'Number Type (as stored)', + 'Source Database' + ) - data_headers = ('id', 'given_name', 'family_name', ('phone_number', 'phonenumber'), 'calltype', - ('date', 'datetime'), ('date_sort', 'datetime'), 'duration', 'numberType') - return data_headers, data_list, context.get_relative_path(source_path) + return data_headers, data_list, '\n'.join(source_paths) diff --git a/scripts/artifacts/hyundai_contacts.py b/scripts/artifacts/hyundai_contacts.py index 5910e67..e9bd7b3 100644 --- a/scripts/artifacts/hyundai_contacts.py +++ b/scripts/artifacts/hyundai_contacts.py @@ -1,39 +1,133 @@ __artifacts_v2__ = { "hyundaiContacts": { - "name": "Hyundai - Contacts", - "description": "Bluetooth contacts from a Hyundai infotainment MC_*.db.", - "author": "Nixy Camacho", - "version": "0.2", + "name": "Hyundai - Bluetooth Contacts", + "description": "Bluetooth contacts per connected phone from Hyundai/Kia infotainment MC_{mac}.db databases.", + "author": "Nixy Camacho, @pmpulkownik", + "version": "0.3", "creation_date": "2023-06-09", - "last_update_date": "2026-06-29", + "last_update_date": "2026-09-02", "requirements": "none", "category": "Hyundai Vehicles", - "notes": "Rewritten as a single query (the original ran separate SELECTs and crashed on " - "os.path.splittext).", + "notes": "Extracts contacts per device and derives device Bluetooth MAC address directly from the database filename (MC_{mac}.db).", "paths": ('*/bluetooth/DB_BMS/MC_*.db*',), "output_types": "standard", - "artifact_icon": "user", + "artifact_icon": "users", } } -from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly +import os +import re +import sqlite3 + +from scripts.ilapfuncs import artifact_processor, logfunc, open_sqlite_db_readonly + +# Ordered as reported. A column missing from a phone's database is reported empty +# rather than costing that phone every row -- see _present_columns. +_CONTACT_COLUMNS = ('_id', 'given_name', 'family_name', 'phone_number', 'phone_type') + + +def _format_mac_from_filename(filename): + """ + Extracts and formats MAC address from filename like MC_AABBCCDDEEFF.db -> AA:BB:CC:DD:EE:FF + """ + base = os.path.basename(filename) + stem = base.rsplit('.', 1)[0] + raw_mac = stem[3:] if stem.startswith('MC_') else stem + + clean_hex = re.sub(r'[^A-Fa-f0-9]', '', raw_mac) + if len(clean_hex) == 12: + return ':'.join(clean_hex[i:i + 2] for i in range(0, 12, 2)).upper() + return raw_mac.upper() + + +def _present_columns(cursor, table, wanted): + try: + cursor.execute(f'PRAGMA table_info({table})') + present = {row[1] for row in cursor.fetchall()} + except sqlite3.Error: + return [] + return [column for column in wanted if column in present] + + +def _value(record, column): + """The stored value, or '' when the column is absent or NULL.""" + value = record.get(column) + return '' if value is None else value @artifact_processor def hyundaiContacts(context): data_list = [] - source_path = '' + source_paths = [] + for file_found in context.get_files_found(): file_found = str(file_found) - if not file_found.endswith('.db'): + if os.path.isdir(file_found) or not file_found.endswith('.db'): continue - source_path = file_found + + device_mac = _format_mac_from_filename(file_found) + db_filename = os.path.basename(file_found) + db = open_sqlite_db_readonly(file_found) + if db is None: + continue cursor = db.cursor() - cursor.execute('SELECT _id, given_name, family_name, phone_number FROM bluetooth_contacts') - for row in cursor.fetchall(): - data_list.append((row[0], row[1], row[2], row[3])) + + columns = _present_columns(cursor, 'bluetooth_contacts', _CONTACT_COLUMNS) + if not columns: + logfunc(f"{db_filename}: no bluetooth_contacts table, skipped") + db.close() + continue + + statement = f'SELECT {", ".join(columns)} FROM bluetooth_contacts' + if '_id' in columns: + statement += ' ORDER BY _id ASC' + + try: + cursor.execute(statement) + rows = cursor.fetchall() + except sqlite3.Error as ex: + logfunc(f"{db_filename}: reading bluetooth_contacts failed, skipped") + logfunc(f" - {str(ex)}") + db.close() + continue + db.close() + source_paths.append(file_found) + + missing = [column for column in _CONTACT_COLUMNS if column not in columns] + if missing: + logfunc(f"{db_filename}: reported without {', '.join(missing)}") + + for row in rows: + record = dict(zip(columns, row)) + given_name = _value(record, 'given_name') + family_name = _value(record, 'family_name') + + # Build full name cleanly if names are present + name_parts = [str(p).strip() for p in (given_name, family_name) if p] + full_name = ' '.join(name_parts) if name_parts else '' + + data_list.append(( + device_mac, + _value(record, '_id'), + full_name, + given_name, + family_name, + _value(record, 'phone_number'), + _value(record, 'phone_type'), + db_filename + )) + + data_headers = ( + 'Device MAC', + 'Record ID', + 'Full Name', + 'First Name', + 'Last Name', + ('Phone Number', 'phonenumber'), + 'Phone Type (as stored)', + 'Source Database' + ) - data_headers = ('ID', 'given_name', 'family_name', ('phone_number', 'phonenumber')) - return data_headers, data_list, context.get_relative_path(source_path) + return data_headers, data_list, '\n'.join(source_paths) diff --git a/scripts/artifacts/hyundai_devices.py b/scripts/artifacts/hyundai_devices.py index 0e30d3b..68ccdb4 100644 --- a/scripts/artifacts/hyundai_devices.py +++ b/scripts/artifacts/hyundai_devices.py @@ -1,15 +1,14 @@ __artifacts_v2__ = { "hyundaiDevices": { - "name": "Hyundai - Bluetooth Devices", - "description": "Bluetooth device MAC addresses and friendly names from a Hyundai " - "wireless_dev_list.dat.", - "author": "Nixy Camacho", - "version": "0.2", + "name": "Hyundai - Bluetooth Paired Devices", + "description": "Bluetooth device MAC addresses and friendly names from Hyundai/Kia wireless_dev_list.dat.", + "author": "Nixy Camacho, @pmpulkownik", + "version": "0.3", "creation_date": "2023-06-09", - "last_update_date": "2026-06-29", + "last_update_date": "2026-09-02", "requirements": "none", "category": "Hyundai Vehicles", - "notes": "", + "notes": "Parses binary structured records from wireless_dev_list.dat to extract paired device MAC addresses and friendly names.", "paths": ('*/wireless_dev_list.dat',), "output_types": "standard", "artifact_icon": "bluetooth", @@ -17,32 +16,41 @@ } import re - from scripts.ilapfuncs import artifact_processor -_ADDR_RE = re.compile(r"[A-Za-z0-9]+:[A-Za-z0-9]+:[A-Za-z0-9]+:[A-Za-z0-9]+:[A-Za-z0-9]+:" - r"[A-Za-z0-9]+", re.IGNORECASE) +# Matches: MAC address (17 ASCII chars + null), optional repeated MAC, and the friendly name string +_RECORD_RE = re.compile( + rb'([0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5})\x00' + rb'(?:[0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5}\x00)?' + rb'([^\x00\r\n\t]+)' +) @artifact_processor def hyundaiDevices(context): data_list = [] source_path = '' + for file_found in context.get_files_found(): file_found = str(file_found) source_path = file_found - dev_addr, dev_name = [], [] - with open(file_found, 'r', encoding='latin-1') as f: - for line in f: - text_only = _ADDR_RE.sub('~~~', line).split('~~~') - if len(text_only) == 1: - name = text_only[0].strip().strip('\x00').strip('\x01').strip('\x02') - if name and name not in dev_name: - dev_name.append(name) - addrs = _ADDR_RE.findall(line) - if len(addrs) == 1 and addrs[0] not in dev_addr: - dev_addr.append(addrs[0]) - data_list.extend(zip(dev_addr, dev_name)) + + try: + with open(file_found, 'rb') as f: + content = f.read() + + for match in _RECORD_RE.finditer(content): + mac_addr = match.group(1).decode('ascii', 'replace').upper() + raw_name = match.group(2) + try: + dev_name = raw_name.decode('utf-8').strip().strip('\x00\x01\x02\x03\x04\x05') + except UnicodeDecodeError: + dev_name = raw_name.decode('latin-1', 'replace').strip().strip('\x00\x01\x02\x03\x04\x05') + + if dev_name and (mac_addr, dev_name) not in data_list: + data_list.append((mac_addr, dev_name)) + except (OSError, IOError): + continue data_headers = ('Bluetooth MAC Address', 'Device Friendly Name') return data_headers, data_list, context.get_relative_path(source_path)