diff --git a/admin/scripts/unwrap_berla_iva.py b/admin/scripts/unwrap_berla_iva.py new file mode 100644 index 0000000..c014cd7 --- /dev/null +++ b/admin/scripts/unwrap_berla_iva.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +"""Unwrap a Berla iVe .iVa export into something VLEAPP can process. + +An .iVa is a plain ZIP holding another ZIP, which in turn holds the vehicle's own +source files: + + .iVa ZIP + Summary.json SHA-256 of every inner file + Vehicle.json vehicle and acquisition record + .zip ZIP + DCASourceFilesUpload.zip ZIP <- the vehicle data VLEAPP wants + AcquireDB.ive iVe's own parsed database, encrypted + Manifest.json, ECUData.json, DLCData.json, CaseData.json, Audit.json + +VLEAPP's seekers do not descend into nested archives, so pointing the tool at a .iVa +matches nothing and produces an empty report. This script lifts DCASourceFilesUpload.zip +out, verifies it against the SHA-256 iVe recorded for it, and leaves a zip that can be +passed straight to VLEAPP with -t zip. + + python3 admin/scripts/unwrap_berla_iva.py CASE.iVa -o outdir + python3 vleapp.py -t zip -i outdir/DCASourceFilesUpload.zip -o reports + +AcquireDB.ive is not unwrapped. It is iVe's parsed output rather than the vehicle's own +data, and it is encrypted. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import sys +import zipfile + +SOURCE_FILES_MEMBER = "DCASourceFilesUpload.zip" +COPY_CHUNK = 16 << 20 + + +def _sha256_stream(fh) -> str: + digest = hashlib.sha256() + while True: + chunk = fh.read(COPY_CHUNK) + if not chunk: + break + digest.update(chunk) + return digest.hexdigest() + + +def _recorded_hashes(outer: zipfile.ZipFile) -> dict: + """Filename -> SHA-256, from the Summary.json iVe writes beside the payload.""" + try: + summary = json.loads(outer.read("Summary.json")) + except KeyError: + return {} + return {e["Filename"]: e["Hashvalue"] for e in summary + if e.get("Filename") and e.get("Hashvalue")} + + +def describe(outer: zipfile.ZipFile) -> None: + """Print what the export says it is. Vehicle.json is plain JSON in the outer zip.""" + try: + vehicle = json.loads(outer.read("Vehicle.json")) + except KeyError: + print(" (no Vehicle.json in this export)") + return + collection = vehicle.get("Collection", {}) + selected = collection.get("SelectedVehicle") or {} + print(f" vehicle : {selected.get('VehicleDisplay') or '(not recorded)'}") + if selected.get("Vin"): + print(f" VIN : {selected['Vin']}") + print(f" collected : {collection.get('CollectionDate') or '(not recorded)'}") + acquisitions = collection.get("Acquisitions") or [] + print(f" acquisitions : {len(acquisitions)}") + for acq in acquisitions: + counts = {k[3:]: v for k, v in acq.items() if k.startswith("Num") and v} + state = acq.get("ErrorMessage") or "no error reported" + print(f" {acq.get('AcqDate', '')[:19]} {acq.get('EcuName', '?')}" + f" {acq.get('AcqType', '?')} {state}") + if counts: + print(f" iVe parsed: {counts}") + + +def unwrap(iva_path: str, out_dir: str, verify: bool = True) -> str: + os.makedirs(out_dir, exist_ok=True) + with zipfile.ZipFile(iva_path) as outer: + print(f"opened {os.path.basename(iva_path)}") + describe(outer) + outer_recorded = _recorded_hashes(outer) + + inner_names = [n for n in outer.namelist() if n.lower().endswith(".zip")] + if not inner_names: + sys.exit("no inner .zip in this .iVa; nothing to unwrap") + inner_name = inner_names[0] + + inner_path = os.path.join(out_dir, os.path.basename(inner_name)) + print(f"\nlifting {inner_name} -> {inner_path}") + with outer.open(inner_name) as src, open(inner_path, "wb") as dst: + shutil.copyfileobj(src, dst, COPY_CHUNK) + + with zipfile.ZipFile(inner_path) as inner: + recorded = dict(outer_recorded) + # Manifest.json inside carries the same hashes; disagreement is worth knowing. + for name, value in _recorded_hashes_from_manifest(inner).items(): + if name in recorded and recorded[name].lower() != value.lower(): + sys.exit(f"the export disagrees with itself about {name}:\n" + f" Summary.json {recorded[name]}\n Manifest.json {value}") + recorded.setdefault(name, value) + if SOURCE_FILES_MEMBER not in inner.namelist(): + sys.exit(f"{SOURCE_FILES_MEMBER} not present in {inner_name}; " + "this export may not carry the vehicle's source files") + target = os.path.join(out_dir, SOURCE_FILES_MEMBER) + print(f"lifting {SOURCE_FILES_MEMBER} -> {target}") + with inner.open(SOURCE_FILES_MEMBER) as src, open(target, "wb") as dst: + shutil.copyfileobj(src, dst, COPY_CHUNK) + + os.remove(inner_path) + + if verify: + want = recorded.get(SOURCE_FILES_MEMBER) + if not want: + print("\nWARNING: the export records no SHA-256 for " + f"{SOURCE_FILES_MEMBER}, so the copy could not be verified") + else: + with open(target, "rb") as fh: + got = _sha256_stream(fh) + if got.lower() != want.lower(): + sys.exit(f"\nHASH MISMATCH for {SOURCE_FILES_MEMBER}\n" + f" recorded by iVe : {want}\n computed : {got}") + print(f"\nverified against the SHA-256 iVe recorded: {got}") + + size = os.path.getsize(target) + with zipfile.ZipFile(target) as src_zip: + members = len(src_zip.infolist()) + print(f"\nready: {target}") + print(f" {size:,} bytes, {members:,} members") + print(f"\nrun VLEAPP against it with:\n python3 vleapp.py -t zip -i \"{target}\" -o ") + return target + + +def _recorded_hashes_from_manifest(inner: zipfile.ZipFile) -> dict: + """Manifest.json inside the inner zip carries the same per-file hashes.""" + try: + manifest = json.loads(inner.read("Manifest.json")) + except KeyError: + return {} + return {i["filename"]: i["hash"] for i in manifest.get("items", []) + if i.get("filename") and i.get("hash")} + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Unwrap a Berla iVe .iVa export into a zip VLEAPP can process.") + parser.add_argument("iva", help="path to the .iVa export") + parser.add_argument("-o", "--out", default="iva_unwrapped", + help="output directory (default: iva_unwrapped)") + parser.add_argument("--no-verify", action="store_true", + help="skip the SHA-256 check against the export's own manifest") + args = parser.parse_args() + + if not os.path.isfile(args.iva): + sys.exit(f"not a file: {args.iva}") + unwrap(args.iva, args.out, verify=not args.no_verify) + + +if __name__ == "__main__": + main() diff --git a/scripts/artifacts/berla_ive_export.py b/scripts/artifacts/berla_ive_export.py new file mode 100644 index 0000000..2f908ee --- /dev/null +++ b/scripts/artifacts/berla_ive_export.py @@ -0,0 +1,126 @@ +"""Berla iVe .iVa export: the acquisition record, and how to reach the vehicle data. + +An .iVa is a ZIP whose payload is another ZIP, so VLEAPP's seekers cannot descend to +the vehicle's own files. Pointing the tool straight at a .iVa therefore matches nothing +and produces an empty report, which reads as though the vehicle held no data. + +Vehicle.json sits uncompressed at the top of the export, so this artifact fires on that +and reports what the export says it holds, including iVe's own per-acquisition counts. +That turns the empty run into a record of what is present and names the step that +reaches it: admin/scripts/unwrap_berla_iva.py. +""" + +import json +import os + +from scripts.ilapfuncs import artifact_processor + +__artifacts_v2__ = { + "berla_ive_export": { + "name": "Berla iVe Export Record", + "description": "The vehicle and acquisition record an iVe .iVa export carries, " + "with one row per acquisition giving the module, the acquisition " + "type, the status iVe recorded and the counts iVe reported parsing.", + "author": "@AlexisBrignoni, Claude", + "version": "0.1", + "creation_date": "2026-08-30", + "last_update_date": "2026-08-30", + "requirements": "none", + "category": "Vehicle Acquisition", + "notes": "From Vehicle.json at the top of a Berla iVe .iVa export. The .iVa is a " + "ZIP holding another ZIP, and the seekers do not descend into nested " + "archives, so running VLEAPP against a .iVa directly reaches only this " + "file and the vehicle's own data is not seen. Unwrap it first with " + "admin/scripts/unwrap_berla_iva.py, which lifts DCASourceFilesUpload.zip " + "out and verifies it against the SHA-256 the export records, then run " + "VLEAPP against that zip. The counts in these rows are what iVe reported " + "for its own parse; they are not produced by VLEAPP and this artifact does " + "not verify them. iVe's parsed database, AcquireDB.ive, is encrypted and " + "is not read. A row here records that an acquisition was attempted and " + "what the tool reported, not what the vehicle contains. Four columns " + "are uniform on a single-vehicle export and are kept because they " + "vary between exports and identify which unit the rows belong to: " + "Module, Driver and Collection Date each hold one value when a " + "collection covers one module, and VIN was empty on the tested " + "export because iVe carries the field but it was not populated " + "there, which is worth showing rather than hiding. Note what the " + "unwrapped export does and does not give you: iVe carries both the raw " + "image and the file set it extracted from the head unit's filesystems, " + "and VLEAPP reads only the extracted files. On the tested export those " + "filesystems are QNX6, which no filesystem type Sleuth Kit supports can " + "walk, so the raw image is not reachable with that tooling. It is " + "reachable with qnxprobe, which reads QNX6 superblocks directly and " + "writes the logical files to a zip; on the tested export that route " + "produced the same rows from the same bytes, and it also surfaced a " + "fourth QNX6 volume that the export did not carry extracted files " + "for.", + "paths": ('*/Vehicle.json',), + "sample_data": { + "adams_ford_syncgen3_iva": "Berla iVe export, Ford Sync Gen3 | 4 rows", + "ford_syncg4_logical": "Ford Sync G4 | 0 rows, not an iVe export", + }, + "output_types": "standard", + "artifact_icon": "car", + }, +} + + +def _is_ive_export(payload): + """True only for the shape an iVe Vehicle.json has. + + Vehicle.json is a common enough name that the glob alone would admit unrelated + files, so this fails closed on anything that does not carry iVe's own structure. + """ + if not isinstance(payload, dict): + return False + collection = payload.get("Collection") + if not isinstance(collection, dict): + return False + return ("SelectedVehicle" in collection and "Acquisitions" in collection + and isinstance(collection.get("Acquisitions"), list)) + + +@artifact_processor +def berla_ive_export(context): + data_list = [] + source_paths = [] + for file_found in context.get_files_found(): + file_found = str(file_found) + if os.path.isdir(file_found): + continue + try: + with open(file_found, 'r', encoding='utf-8', errors='replace') as handle: + payload = json.load(handle) + except (OSError, ValueError): + continue + if not _is_ive_export(payload): + continue + + source_paths.append(file_found) + collection = payload["Collection"] + vehicle = collection.get("SelectedVehicle") or {} + display = vehicle.get("VehicleDisplay") or '' + vin = vehicle.get("Vin") or '' + collected = collection.get("CollectionDate") or '' + + for acq in collection["Acquisitions"]: + if not isinstance(acq, dict): + continue + counts = {k[3:]: v for k, v in acq.items() + if k.startswith("Num") and isinstance(v, int) and v} + data_list.append(( + (acq.get("AcqDate") or '').replace('T', ' ')[:19], + display, vin, + acq.get("EcuName") or '', acq.get("AcqType") or '', + acq.get("ErrorMessage") or 'no error reported', + acq.get("PercentageComplete") if acq.get("PercentageComplete") is not None else '', + ', '.join(f'{k} {v}' for k, v in sorted(counts.items())) or 'none reported', + acq.get("DriverName") or '', acq.get("AcqUnit") or '', + collected, context.get_relative_path(file_found))) + + data_headers = (('Acquisition Date', 'datetime'), 'Vehicle', 'VIN', + 'Module', 'Acquisition Type', 'Status (as stored)', + 'Percent Complete (as stored)', 'Counts iVe Reported', + 'Driver (as stored)', 'Acquisition Unit (as stored)', + 'Collection Date', 'Source File') + return data_headers, data_list, '\n'.join(source_paths) diff --git a/scripts/artifacts/ford_sync_bluetooth.py b/scripts/artifacts/ford_sync_bluetooth.py new file mode 100644 index 0000000..3791c06 --- /dev/null +++ b/scripts/artifacts/ford_sync_bluetooth.py @@ -0,0 +1,296 @@ +"""Ford Sync Gen3 Bluetooth phonebook, call history and paired-device records. + +The head unit keeps these in two extension-less SQLite stores under BT/ on the user +data partition: btpbk (phonebook and call lists) and btpersist (paired devices and +their settings). Both are numbered per paired handset, slots 1 to 12. +""" + +import os +import sqlite3 + +from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly + + +__artifacts_v2__ = { + "ford_sync_bt_contacts": { + "name": "Bluetooth Phonebook", + "description": "Contacts the head unit downloaded from each paired handset, with " + "the names, the phone numbers the record carried, email and postal " + "address as stored. One row per contact per handset slot.", + "author": "@AlexisBrignoni, Claude", + "version": "0.1", + "creation_date": "2026-08-30", + "last_update_date": "2026-08-30", + "requirements": "none", + "category": "Ford Vehicles", + "notes": "From the PhoneBook tables of BT/btpbk, an extension-less SQLite store " + "on the user data partition. The unit numbers its tables per paired " + "handset, slots 1 to 12, and the slot is reported so contacts from " + "different handsets stay separable. A phonebook entry records what the " + "unit downloaded over Bluetooth; it does not establish that any number was " + "dialled or that the handset owner was present. TelType is not surfaced " + "because nothing available here documents its values. The store carries no " + "write-ahead log or journal on the tested unit. Where the input came " + "from a Berla iVe export, these rows are read from the file set iVe " + "extracted from the head unit's QNX6 volumes, not from the raw image " + "the export also carries. The values themselves are the unit's own " + "rather than iVe's parse of them. That route is not the only one: " + "these same rows were reproduced from the raw image directly, by " + "extracting its QNX6 volumes with qnxprobe and running this module " + "against that output. Both paths gave 507 contacts, 126 calls and 2 " + "paired devices, and the two stores were byte-identical by SHA-256, so " + "neither extraction is a bottleneck for what these artifacts report.", + "paths": ('*/BT/btpbk*',), + "sample_data": { + "adams_ford_syncgen3": "Ford Sync Gen3 | 507 rows", + "ford_syncg4_logical": "Ford Sync G4 | 0 rows, BT/btpbk not present", + }, + "output_types": "standard", + "artifact_icon": "book-open", + }, + "ford_sync_bt_calls": { + "name": "Bluetooth Call History", + "description": "Calls the head unit recorded for each paired handset, with the time " + "as stored, the direction, and the name and number the record " + "carried.", + "author": "@AlexisBrignoni, Claude", + "version": "0.1", + "creation_date": "2026-08-30", + "last_update_date": "2026-08-30", + "requirements": "none", + "category": "Ford Vehicles", + "notes": "From the Combined tables of BT/btpbk, which the unit maintains " + "alongside separate InCall, DialCall and MissCall tables. " + "Combined is read because it is the union of the three: on the tested unit " + "the two populated handsets gave 22+23+25=70 and 20+19+17=56, matching " + "their Combined row counts exactly. Direction is decoded from CallType, " + "and that mapping was derived from the data rather than assumed: filtering " + "Combined by each CallType produced a (number, date, time) row set " + "identical to the correspondingly named table, on both handsets " + "independently, giving 1 Received, 2 Dialled, 4 Missed. Any other value is " + "reported as stored. The unit writes the time as six separate text " + "components and records no timezone anywhere in the store, so the " + "timestamp is assembled as written and no conversion is applied. A call " + "record is what the handset reported to the unit over Bluetooth; it does " + "not establish who used the handset or that the vehicle was moving.", + "paths": ('*/BT/btpbk*',), + "sample_data": { + "adams_ford_syncgen3": "Ford Sync Gen3 | 126 rows", + "ford_syncg4_logical": "Ford Sync G4 | 0 rows, BT/btpbk not present", + }, + "output_types": "standard", + "artifact_icon": "phone", + }, + "ford_sync_bt_paired_devices": { + "name": "Bluetooth Paired Devices", + "description": "Handsets currently paired with the head unit, with the name, model, " + "manufacturer, network name, software version, Bluetooth address and " + "subscriber number the unit stored for each.", + "author": "@AlexisBrignoni, Claude", + "version": "0.1", + "creation_date": "2026-08-30", + "last_update_date": "2026-08-30", + "requirements": "none", + "category": "Ford Vehicles", + "notes": "From PairedDevInfo in BT/btpersist, joined to DeviceOrder and " + "HFPdeviceOrder on DeviceID for the primary-device flags, which are " + "reported as stored because nothing available here documents their values. " + "This table holds the pairings the unit currently retains, which is a " + "narrower set than the handsets it has ever seen: the devlog_*.txt files " + "in the same BT directory, which btDevices.py parses, covered five " + "handsets on the tested unit while this table held two. Read both. Class " + "of device, vendor id and product id are reported as stored.", + "paths": ('*/BT/btpersist*',), + "sample_data": { + "adams_ford_syncgen3": "Ford Sync Gen3 | 2 rows", + "ford_syncg4_logical": "Ford Sync G4 | 0 rows, BT/btpersist not present", + }, + "output_types": "standard", + "artifact_icon": "bluetooth", + }, +} + + + +# Slots the head unit numbers its per-handset tables with. +DEVICE_SLOTS = range(1, 13) + +# Proven on the tested unit, not assumed: for both populated handsets the +# (TelNum, Date, Time) row set of Combined filtered to each CallType was identical +# to the row set of the correspondingly named table the unit maintains alongside it. +CALL_TYPES = { + 1: 'Received', # matched InCall + 2: 'Dialled', # matched DialCall + 4: 'Missed', # matched MissCall +} + + +def _tables(db): + return {r[0] for r in db.execute( + "SELECT name FROM sqlite_master WHERE type='table'")} + + +def _stamp(row): + """Build 'YYYY-MM-DD HH:MM:SS' from the six text components the unit stores. + + Returns '' when any component is missing, rather than inventing a partial time. + No timezone is recorded anywhere in the store, so nothing is converted. + """ + yyyy, mm, dd, hh, mi, ss = (str(v or '').strip() for v in row) + if not (yyyy and mm and dd): + return '' + return f"{yyyy}-{mm.zfill(2)}-{dd.zfill(2)} {hh.zfill(2)}:{mi.zfill(2)}:{ss.zfill(2)}" + + +def _btpbk_files(context): + for file_found in context.get_files_found(): + file_found = str(file_found) + if os.path.isdir(file_found): + continue + if os.path.basename(file_found).lower().startswith('btpbk'): + yield file_found + + +def _btpersist_files(context): + for file_found in context.get_files_found(): + file_found = str(file_found) + if os.path.isdir(file_found): + continue + if os.path.basename(file_found).lower().startswith('btpersist'): + yield file_found + + +@artifact_processor +def ford_sync_bt_contacts(context): + data_list = [] + source_paths = [] + for file_found in _btpbk_files(context): + try: + db = open_sqlite_db_readonly(file_found) + except sqlite3.Error: + continue + present = _tables(db) + read_any = False + for slot in DEVICE_SLOTS: + table = f'PhoneBook{slot}' + if table not in present: + continue + try: + rows = db.execute( + f'SELECT RecId, FirstName, LastName, SortName, Email, ' + f'TelNum0, TelNum1, TelNum2, TelNum3, TelNum4, TelNum5, ' + f'TelNum6, TelNum7, TelNum8, StreetAdr0, Locality0, Region0, ' + f'POCode0, Country0 FROM "{table}"').fetchall() + except sqlite3.Error: + continue + read_any = True + for r in rows: + numbers = [str(n).strip() for n in r[5:14] if str(n or '').strip()] + address = ', '.join(str(v).strip() for v in r[14:19] + if str(v or '').strip()) + data_list.append(( + slot, r[2] or '', r[1] or '', r[3] or '', + numbers[0] if numbers else '', + ', '.join(numbers[1:]), len(numbers), r[4] or '', address, + context.get_relative_path(file_found))) + db.close() + if read_any: + source_paths.append(file_found) + + data_headers = ('Device Slot', 'Last Name', 'First Name', 'Sort Name', + 'Phone Number', 'Other Numbers', 'Number Count', 'Email', + 'Address', 'Source File') + return data_headers, data_list, '\n'.join(source_paths) + + +@artifact_processor +def ford_sync_bt_calls(context): + data_list = [] + source_paths = [] + for file_found in _btpbk_files(context): + try: + db = open_sqlite_db_readonly(file_found) + except sqlite3.Error: + continue + present = _tables(db) + read_any = False + for slot in DEVICE_SLOTS: + table = f'Combined{slot}' + if table not in present: + continue + try: + rows = db.execute( + f'SELECT Date_YYYY, Date_MM, Date_DD, Time_hh, Time_min, Time_sec, ' + f'CallType, SortName, TelNum, TelType, RecId ' + f'FROM "{table}"').fetchall() + except sqlite3.Error: + continue + read_any = True + for r in rows: + direction = CALL_TYPES.get(r[6], f'{r[6]} (as stored)') + data_list.append(( + _stamp(r[0:6]), direction, r[7] or '', r[8] or '', + r[9], slot, r[10], + context.get_relative_path(file_found))) + db.close() + if read_any: + source_paths.append(file_found) + + data_headers = (('Call Time', 'datetime'), 'Direction', 'Name', + 'Phone Number', 'Number Type (as stored)', 'Device Slot', + 'Record ID', 'Source File') + return data_headers, data_list, '\n'.join(source_paths) + + +@artifact_processor +def ford_sync_bt_paired_devices(context): + data_list = [] + source_paths = [] + for file_found in _btpersist_files(context): + try: + db = open_sqlite_db_readonly(file_found) + except sqlite3.Error: + continue + present = _tables(db) + if 'PairedDevInfo' not in present: + db.close() + continue + + primary = {} + if 'DeviceOrder' in present: + try: + primary = {r[0]: r[1] for r in + db.execute('SELECT DeviceID, primaryDevice FROM DeviceOrder')} + except sqlite3.Error: + primary = {} + hfp_primary = {} + if 'HFPdeviceOrder' in present: + try: + hfp_primary = {r[0]: r[1] for r in db.execute( + 'SELECT HFPdeviceID, HFPprimaryDevice FROM HFPdeviceOrder')} + except sqlite3.Error: + hfp_primary = {} + + try: + rows = db.execute( + 'SELECT DeviceID, DeviceName, DeviceModel, ManufacturerName, ' + 'NetworkName, DeviceSoftwareVersion, DeviceAddress, subscriberNum, ' + 'ClassOfDevice, VendorId, ProductId FROM PairedDevInfo').fetchall() + except sqlite3.Error: + db.close() + continue + db.close() + source_paths.append(file_found) + for r in rows: + data_list.append(( + r[0], r[1] or '', r[2] or '', r[3] or '', r[4] or '', + r[5] or '', r[6] or '', r[7] or '', r[8] or '', + primary.get(r[0], ''), hfp_primary.get(r[0], ''), + r[9], r[10], context.get_relative_path(file_found))) + + data_headers = ('Device Slot', 'Device Name', 'Model', 'Manufacturer', + 'Network Name', 'Device Software Version', 'Bluetooth Address', + 'Subscriber Number', 'Class of Device (as stored)', + 'Primary Device (as stored)', 'HFP Primary (as stored)', + 'Vendor ID (as stored)', 'Product ID (as stored)', 'Source File') + return data_headers, data_list, '\n'.join(source_paths)