Fixed Hyundai Device Processing

[pmpulkownik]

What this changes

The current Hyundai/Kia scripts did not parse all of the databases if the infotainment system had multiple paired phones. Rewrote scripts to parsed all call history and contact databases. Furthermore, the device history script failed to parse the dat file due to the mac address reporting twice in the dat file for each device, changed the logic to not discard those.

For a new or changed artifact

• Ran the tool against a real extraction and confirmed the row counts, not just that it imports.
• Ran python admin/scripts/check_artifact_output.py <report folder> on that report and fixed or documented every finding. An empty or constant column is often a real result (no group chats, coarse location denied); the fix for those is to say so in the artifact's notes, which is also what stops the checker reporting them.
• Checked it against a second app data directory where the platform provides one (a second user, account, or container). --compare <multi-container report> reads the scaling for you: it should be exactly double.
• notes, description and sample_data say only what the data shows, and the numbers were re-derived from the finished run.

Anything reviewers should know

[github-actions]

Thanks for the contribution!

This PR changes artifact modules without test data for them. A small fixture with each artifact change lets reviewers run the module against real data, and the committed case keeps guarding the module after merge.

• hyundai_callHistory.py: please include a fixture with this PR.
• hyundai_contacts.py: please include a fixture with this PR.
• hyundai_devices.py: please include a fixture with this PR.

Adding a fixture

Generate it from your extraction with the helper (details in create_module_test_cases.md):

python admin/test/scripts/make_test_data.py <module> --case <case_number> --input <extraction.zip>

It writes admin/test/cases/testdata.<module>.json and one zip per artifact under admin/test/cases/data/<module>/.

Size rules:

• Under 10 MB per zip: commit the files in this PR.
• 10 to 25 MB: commit the case JSON in the PR and attach the zip to a comment here.
• Over 25 MB: say so here and a maintainer will arrange a handoff.

If your extraction cannot be shared:

• If the app appears on a public research image, generate the fixture from that instead. public_corpus_images.md lists the images and where to download them.
• Or sanitize the real file in place: keep the file the app wrote and overwrite only the personal values, which keeps the format honest.
• Or script a known session: install the app on a test device with a throwaway account, perform documented actions, and extract that.

If none of those fit, say so here and we will work it out. The PR can still be reviewed and merged with the gap recorded in the artifact's notes.

This is a request, not a gate. Nothing here blocks review.

[abrignoni]

Thanks for this — the multi-phone gap is real, and the devices script needed it most: the old code built dev_addr and dev_name as two independent lists and zip()ed them, so the MAC↔name pairing was positional and one unnamed device shifted every row after it. Pairing them inside a single record is the right fix. Deriving the MAC from the CH_/MC_ filename is a nice touch too.

Two things before it can land: the test fixtures, and four issues in the new per-database loop.

The fixtures are the actual blocker

Everything below is an afternoon's work; the fixtures aren't, because they need your extraction. needs-test-data is on the PR and the bot comment above has the generation steps. That same corpus is what lets you fill in sample_data on the three artifacts — none of them declare it right now, and the checklist box about sample_data is ticked, so I'd rather ask than assume: were the row counts derived from a finished run against a real head unit?

If the extraction can't be shared publicly, say so on the PR and we'll sort out a private route — that's normal for vehicle work.

What I confirmed works

Built synthetic CH_/MC_ databases and a wireless_dev_list.dat and ran all three artifacts through the plugin loader:

• Per-device iteration and the filename→MAC derivation both behave.
• The millisecond heuristic in _ts is correct: 1700000000 and 1700000000000 both land on 2023-11-14 22:13:20 UTC. Treating 0 as absent rather than 1970 is right too.
• Locally, python_lint's lint_changed.py reports 0 new warnings, and check_claim_language, check_html_safety, check_report_local_paths and check_source_path all pass.

Note the repo's real CI has not run on this PR — five workflows are sitting at action_required because it's your first PR here.

1. Contacts silently drops every row from a phone whose table lacks phone_type

This is the one I'd most like your read on, because it needs your corpus to settle.

The new query adds phone_type to what was a four-column SELECT. If any phone's bluetooth_contacts doesn't carry that column, execute raises OperationalError, except sqlite3.Error: pass swallows it, and that phone's contacts vanish with no error and no log line.

Reproduction: two MC_*.db files, one with phone_type and one without, four contacts total. Output was 2 rows — one phone only, silently.

That's the same failure this PR set out to fix, arriving through schema drift instead of loop logic, and it's the worst shape for it: an examiner reads an empty table as "this phone had no contacts."

Does phone_type exist on every head unit / firmware you've seen? I have no Hyundai extraction here, so I can't tell whether this is theoretical or a live bug in the field. Either way the fix is the same — select the columns the table actually has:

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]

then build the statement from the intersection and read rows back with dict(zip(columns, row)) so an absent column reports empty instead of costing you the phone. bluetooth_callhistory has the same latent exposure, so it's worth the same treatment even though this PR doesn't change its column set.

2. A database that won't open crashes the whole artifact

open_sqlite_db_readonly returns None on OperationalError (scripts/ilapfuncs.py:679), and db.cursor() on the next line then raises AttributeError. artifact_processor doesn't catch it, so vleapp.py logs the artifact as failed and it produces no output at all — every paired phone's call log gone, not just the unreadable one.

I hit it with a directory named CH_999999999999.db; the */bluetooth/DB_BMS/CH_*.db* glob matches directories too. scripts/artifacts/bmw_connected_devices.py is the pattern to copy — it guards both:

if os.path.isdir(file_found) or not file_found.endswith('.db'):
    continue
db = open_sqlite_db_readonly(file_found)
if db is None:
    continue

This was already there before your change, but it mattered less when one bad file cost one file's rows.

3. except sqlite3.Error: pass should log

It's the mechanism behind #1, and it hides schema drift generally. logfunc is already available from scripts.ilapfuncs; bmw_connected_devices.py closes the handle and continues. Silent empty output is the one failure mode worth going out of the way to avoid in this repo.

4. Return the databases parsed, not their parent directory

os.path.dirname(source_paths[0]) makes the report's "located at" line name a folder and drops which databases were read from the LAVA manifest. artifact_processor already supports a newline-joined list and relativizes each element (scripts/ilapfuncs.py:509) — seven artifacts use it, bmw_connected_devices.py among them:

return data_headers, data_list, '\n'.join(source_paths)

Worth also appending to source_paths only after the query succeeds, so a database that failed isn't credited as the source of rows it didn't contribute.

Smaller things, same round

• Control bytes reach the report. [^\x00\r\n\t]+ in hyundai_devices.py permits \x01/\x02, which the old code stripped explicitly; .strip() only takes whitespace. My fixture produced John's iPhone\x02, which goes into the HTML report and LAVA as-is.
• Duration (sec) asserts a unit nothing in the data establishes. Compare bmw_connected_devices.py, which uses Duration (as stored) for exactly this reason.
• Call Type / Number Type / Phone Type are raw stored integers, but the friendly headers read as though they're decoded — a column labelled Call Type showing 2 invites the examiner to guess. Either decode them (calltype is the most probative field in a call log, so it's worth doing) or mark them (as stored).

A patch, if it's useful

I have findings 1–4 written and verified against those fixtures — old-schema contacts now yield their rows with Phone Type empty plus a log line, the directory case is skipped instead of fatal, and source_path lists the files. _ts and the headers are untouched. It's collapsed below if you'd rather apply it than retype it; equally happy for you to take the shapes above and do it your way, since it's your PR.

Diff against 109f912 (git apply)

diff --git a/scripts/artifacts/hyundai_callHistory.py b/scripts/artifacts/hyundai_callHistory.py
index 8a0f985..76bc8fb 100644
--- a/scripts/artifacts/hyundai_callHistory.py
+++ b/scripts/artifacts/hyundai_callHistory.py
@@ -19,7 +19,14 @@ 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):
@@ -36,6 +43,28 @@ def _format_mac_from_filename(filename):
     return raw_mac.upper()
 
 
+def _present_columns(cursor, table, wanted):
+    """
+    The wanted columns this table actually carries, in the order given.
+
+    Firmware versions differ in what these tables hold, and a SELECT naming a
+    column the table lacks fails the whole statement. Selecting only what is
+    present costs the absent column rather than every call on that phone.
+    """
+    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 == '' or value == 0:
         return ''
@@ -63,66 +92,65 @@ def hyundaiCallHistory(context):
 
     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_paths.append(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()
 
+        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('''
-                SELECT 
-                    date,
-                    date_sort,
-                    _id,
-                    given_name,
-                    family_name,
-                    phone_number,
-                    calltype,
-                    duration,
-                    numberType
-                FROM bluetooth_callhistory
-                ORDER BY date DESC
-            ''')
-
-            for row in cursor.fetchall():
-                (
-                    call_date,
-                    date_sort,
-                    rec_id,
-                    given_name,
-                    family_name,
-                    phone_number,
-                    calltype,
-                    duration,
-                    number_type
-                ) = row
-
-                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(call_date),
-                    _ts(date_sort),
-                    device_mac,
-                    rec_id,
-                    full_name,
-                    given_name if given_name is not None else '',
-                    family_name if family_name is not None else '',
-                    phone_number if phone_number is not None else '',
-                    calltype if calltype is not None else '',
-                    duration if duration is not None else '',
-                    number_type if number_type is not None else '',
-                    db_filename
-                ))
-        except sqlite3.Error:
-            pass
+            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'),
@@ -139,5 +167,4 @@ def hyundaiCallHistory(context):
         'Source Database'
     )
 
-    source_repr = os.path.dirname(source_paths[0]) if source_paths else ''
-    return data_headers, data_list, context.get_relative_path(source_repr)
+    return data_headers, data_list, '\n'.join(source_paths)
diff --git a/scripts/artifacts/hyundai_contacts.py b/scripts/artifacts/hyundai_contacts.py
index c0d140c..8e8a518 100644
--- a/scripts/artifacts/hyundai_contacts.py
+++ b/scripts/artifacts/hyundai_contacts.py
@@ -18,7 +18,12 @@ __artifacts_v2__ = {
 import os
 import re
 import sqlite3
-from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly
+
+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):
@@ -35,6 +40,28 @@ def _format_mac_from_filename(filename):
     return raw_mac.upper()
 
 
+def _present_columns(cursor, table, wanted):
+    """
+    The wanted columns this table actually carries, in the order given.
+
+    Firmware versions differ in what these tables hold, and a SELECT naming a
+    column the table lacks fails the whole statement. Selecting only what is
+    present costs the absent column rather than every contact on that phone.
+    """
+    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 = []
@@ -42,49 +69,62 @@ def hyundaiContacts(context):
 
     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_paths.append(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()
 
+        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('''
-                SELECT 
-                    _id,
-                    given_name,
-                    family_name,
-                    phone_number,
-                    phone_type
-                FROM bluetooth_contacts
-                ORDER BY _id ASC
-            ''')
-
-            for row in cursor.fetchall():
-                rec_id, given_name, family_name, phone_number, phone_type = row
-
-                # 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,
-                    rec_id,
-                    full_name,
-                    given_name if given_name is not None else '',
-                    family_name if family_name is not None else '',
-                    phone_number if phone_number is not None else '',
-                    phone_type if phone_type is not None else '',
-                    db_filename
-                ))
-        except sqlite3.Error:
-            pass
+            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',
@@ -97,5 +137,4 @@ def hyundaiContacts(context):
         'Source Database'
     )
 
-    source_repr = os.path.dirname(source_paths[0]) if source_paths else ''
-    return data_headers, data_list, context.get_relative_path(source_repr)
+    return data_headers, data_list, '\n'.join(source_paths)

Findings 5–7 aren't in that patch — hyundai_devices.py is untouched, so the control-byte fix is still yours.

Thanks again for digging into this; Hyundai/Kia coverage here has needed someone with real head-unit data.

---

Generated by Claude Code

[pmpulkownik]

Thank you for the feedback. I implemented the fixes you suggested and pushed a new commit. I reran the new scripts against my case without error. Unfortunately, this is an active case so I cannot share the data publicly. I did confirm against the databases that the contact lists and call history are accurate, as well as the wireless history.

[abrignoni]

Merged — thanks @pmpulkownik. Quick turnaround on the review and you went past what was asked with the (as stored) headers. Multi-phone Hyundai/Kia coverage is a real gap closed. A small follow-up in #182 tightened the control-byte handling in the device names.

---

Generated by Claude Code