From 3b6513104ffadf6c5aa7f8996ab56001d748813e Mon Sep 17 00:00:00 2001 From: Brigs Date: Sun, 30 Aug 2026 22:18:05 -0400 Subject: [PATCH 1/3] Add native Berla iVe .iVa input -t iva takes an iVe export directly. A .iVa is a ZIP holding another ZIP, and the seekers do not descend into nested archives, so before this the export had to be unwrapped by hand and a direct run produced an empty report. FileSeekerIva reaches through the nesting itself. When the export carries a raw disk image, that image is read with the vendored qnxprobe, which is the more complete route: on the tested export it reaches a fourth QNX6 volume the vendor's own extracted file set does not include. When no raw image is present the extracted file set is used as it stands. Either way Vehicle.json is staged at the root, so the export's acquisition record is reported beside the vehicle data. Everything intermediate lands in a temporary directory and cleanup() removes it; the unwrap script remains the way to keep the intermediate zip for cheap re-runs on a large export. The qnxprobe runner moves out of FileSeekerRaw into a module helper both seekers share. The GUI needs only a file picker entry: it derives the input type from the suffix, and .iVa lowercases to the new type on its own. Measured against the evidence .iVa: all nine vehicle artifacts return the same row counts as the raw image run, plus the export record's four rows, with no database errors, and -t zip and -t raw runs are unchanged after the refactor. Co-Authored-By: Claude Opus 5 --- scripts/search_files.py | 184 +++++++++++++++++++++++++++++++--------- vleapp.py | 8 +- vleappGUI.py | 5 +- 3 files changed, 153 insertions(+), 44 deletions(-) diff --git a/scripts/search_files.py b/scripts/search_files.py index 5b76ef1..556cac5 100755 --- a/scripts/search_files.py +++ b/scripts/search_files.py @@ -32,7 +32,7 @@ from pathlib import Path from scripts.ilapfuncs import * -from shutil import copy2 +from shutil import copy2, copyfileobj from zipfile import ZipFile from fnmatch import _compile_pattern from functools import lru_cache @@ -566,6 +566,49 @@ def update(self, event): self.volumes += 1 +def _extract_image_volumes(probe, image_path, staged_zip, exclude=None): + """Run the vendored reader over a raw image, streaming its output to the log. + + -u so the reader's stdout is unbuffered and its report arrives while it + works rather than in one block at the end. --progress puts machine readable + progress on stderr; stderr is merged into stdout here and the two are told + apart by the leading brace, which no report line has, so one stream is read + and there is no second pipe to deadlock on. The report names the partition + table, every filesystem confirmed and what was extracted; that belongs in + the run log, because it is the record of which volumes the rows came from. + """ + cmd = [sys.executable, '-u', probe, '--progress', '--extract', staged_zip] + for text in (exclude or ()): + cmd += ['--exclude', text] + cmd.append(image_path) + + logfunc(f'Reading volumes out of {os.path.basename(image_path)} with the ' + f'vendored qnxprobe. This is the slow part of a raw image run.') + progress = _RawExtractProgress() + tail = [] + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, text=True, bufsize=1) + for line in proc.stdout: + line = line.rstrip() + if not line: + continue + if line.lstrip().startswith('{'): + try: + progress.update(json.loads(line)) + continue + except ValueError: + pass # not ours after all, fall through and log it + logfunc(line) + tail.append(line) + del tail[:-20] + proc.wait() + if proc.returncode != 0 or not os.path.isfile(staged_zip): + detail = '\n'.join(tail) or 'no output' + raise RuntimeError( + f'qnxprobe could not extract any filesystem from {image_path}. ' + f'Exit {proc.returncode}. Last output:\n{detail}') + + class FileSeekerRaw(FileSeekerZip): """Read a raw disk image by extracting its volumes to a zip first. @@ -601,45 +644,7 @@ def __init__(self, image_path, data_folder, exclude=None): f'the vendored reader is missing: {probe}. Raw image input needs ' 'scripts/vendor/qnxprobe.py.') - # -u so the reader's stdout is unbuffered and its report arrives while it - # works rather than in one block at the end. --progress puts machine - # readable progress on stderr; stderr is merged into stdout here and the - # two are told apart by the leading brace, which no report line has, so - # one stream is read and there is no second pipe to deadlock on. - cmd = [sys.executable, '-u', probe, '--progress', '--extract', staged_zip] - for text in (exclude or ()): - cmd += ['--exclude', text] - cmd.append(image_path) - - logfunc(f'Reading volumes out of {os.path.basename(image_path)} with the ' - f'vendored qnxprobe. This is the slow part of a raw image run.') - progress = _RawExtractProgress() - tail = [] - proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, text=True, bufsize=1) - for line in proc.stdout: - line = line.rstrip() - if not line: - continue - if line.lstrip().startswith('{'): - try: - progress.update(json.loads(line)) - continue - except ValueError: - pass # not ours after all, fall through and log it - # The report names the partition table, every filesystem confirmed and - # what was extracted. That belongs in the run log: it is the record of - # which volumes the rows below came from. - logfunc(line) - tail.append(line) - del tail[:-20] - proc.wait() - if proc.returncode != 0 or not os.path.isfile(staged_zip): - detail = '\n'.join(tail) or 'no output' - raise RuntimeError( - f'qnxprobe could not extract any filesystem from {image_path}. ' - f'Exit {proc.returncode}. Last output:\n{detail}') - + _extract_image_volumes(probe, image_path, staged_zip, exclude) FileSeekerZip.__init__(self, staged_zip, data_folder) def cleanup(self): @@ -647,6 +652,105 @@ def cleanup(self): shutil.rmtree(getattr(self, '_stage_dir', ''), ignore_errors=True) + +class FileSeekerIva(FileSeekerZip): + """Read a Berla iVe .iVa export directly. + + An .iVa is a ZIP holding another ZIP, which holds the vehicle's source + files: usually a raw disk image of the head unit plus the file set iVe + extracted from it, beside Vehicle.json and iVe's own encrypted database. + The seekers do not descend into nested archives, so before this a .iVa had + to be unwrapped by hand (admin/scripts/unwrap_berla_iva.py) and the report + of a direct run was empty. + + This reaches through the nesting itself. When the export carries a raw + image, that image is read with the vendored qnxprobe, which is the more + complete route: on the tested export it reaches a volume the vendor's own + extracted file set does not include. When no raw image is present, the + extracted file set is used as it stands. Either way Vehicle.json is staged + at the root so the export's acquisition record is reported alongside the + vehicle data. + + Everything intermediate lands in a temporary directory and cleanup() + removes it. The unwrap script remains the way to KEEP the intermediate + zip, which makes re-runs cheap on a large export. + """ + + SOURCE_FILES_MEMBER = 'DCASourceFilesUpload.zip' + + def __init__(self, iva_path, data_folder, exclude=None): + self._stage_dir = tempfile.mkdtemp(prefix='vleapp_iva_') + probe = os.path.join(os.path.dirname(os.path.abspath(__file__)), + 'vendor', 'qnxprobe.py') + if not os.path.isfile(probe): + raise FileNotFoundError( + f'the vendored reader is missing: {probe}. .iVa input needs ' + 'scripts/vendor/qnxprobe.py.') + + vehicle_json = None + with ZipFile(iva_path) as outer: + names = outer.namelist() + if 'Vehicle.json' in names: + vehicle_json = outer.read('Vehicle.json') + else: + logfunc('This .iVa carries no Vehicle.json, so no acquisition ' + 'record will be reported for it.') + inner_names = [n for n in names if n.lower().endswith('.zip')] + if not inner_names: + raise RuntimeError( + f'{os.path.basename(iva_path)} holds no inner zip, so it ' + 'does not look like an iVe export.') + inner_path = os.path.join(self._stage_dir, 'inner.zip') + logfunc(f'Unpacking {os.path.basename(iva_path)}: ' + f'{inner_names[0]} ...') + with outer.open(inner_names[0]) as src, open(inner_path, 'wb') as dst: + copyfileobj(src, dst, 16 << 20) + + with ZipFile(inner_path) as inner: + if self.SOURCE_FILES_MEMBER not in inner.namelist(): + raise RuntimeError( + f'{inner_names[0]} carries no {self.SOURCE_FILES_MEMBER}; ' + 'this export does not include the vehicle source files.') + source_zip = os.path.join(self._stage_dir, self.SOURCE_FILES_MEMBER) + logfunc(f'Unpacking {self.SOURCE_FILES_MEMBER} ...') + with inner.open(self.SOURCE_FILES_MEMBER) as src, \ + open(source_zip, 'wb') as dst: + copyfileobj(src, dst, 16 << 20) + os.remove(inner_path) + + with ZipFile(source_zip) as source: + images = [n for n in source.namelist() + if n.startswith('DiskImages/') + and n.lower().endswith(('.img', '.bin', '.dd', '.raw'))] + image_path = None + if images: + image_path = os.path.join(self._stage_dir, + os.path.basename(images[0])) + logfunc(f'The export carries a raw image, {images[0]}; reading ' + 'the vehicle data from the image itself.') + with source.open(images[0]) as src, open(image_path, 'wb') as dst: + copyfileobj(src, dst, 16 << 20) + + staged_zip = os.path.join(self._stage_dir, 'iva_volumes.zip') + if image_path is not None: + _extract_image_volumes(probe, image_path, staged_zip, exclude) + os.remove(image_path) + os.remove(source_zip) + else: + logfunc('The export carries no raw image; using the file set iVe ' + 'extracted.') + staged_zip = source_zip + + if vehicle_json is not None: + with ZipFile(staged_zip, 'a') as add: + add.writestr('Vehicle.json', vehicle_json) + + FileSeekerZip.__init__(self, staged_zip, data_folder) + + def cleanup(self): + FileSeekerZip.cleanup(self) + shutil.rmtree(getattr(self, '_stage_dir', ''), ignore_errors=True) + class FileSeekerFile(FileSeekerBase): """ This is a class that extends FileSeekerBase to facilitate searching for and copying a specific file diff --git a/vleapp.py b/vleapp.py index 568a720..5e1472d 100755 --- a/vleapp.py +++ b/vleapp.py @@ -149,12 +149,13 @@ def create_casedata(path): def main(): parser = argparse.ArgumentParser(description='VLEAPP: Vehicle Logs, Events, and Protobuf Parser.') - parser.add_argument('-t', choices=['fs', 'tar', 'zip', 'gz', 'file', 'raw'], required=False, action="store", + parser.add_argument('-t', choices=['fs', 'tar', 'zip', 'gz', 'file', 'raw', 'iva'], required=False, action="store", help=("Specify the input type. " "'fs' for a folder containing extracted files with normal paths and names, " "'tar', 'zip', or 'gz' for compressed packages containing files with normal names, " "'file' for a single file input, " - "'raw' for a raw disk image whose QNX6 or ext volumes are read without mounting.")) + "'raw' for a raw disk image whose QNX6 or ext volumes are read without mounting, " + "'iva' for a Berla iVe .iVa export.")) parser.add_argument('-o', '--output_path', required=False, action="store", help='Path to base output folder (this must exist)') parser.add_argument('-i', '--input_path', required=False, action="store", help='Path to input file/folder') @@ -346,6 +347,9 @@ def crunch_artifacts( elif extracttype == 'raw': seeker = FileSeekerRaw(input_path, out_params.data_folder) + elif extracttype == 'iva': + seeker = FileSeekerIva(input_path, out_params.data_folder) + else: logfunc('Error on argument -o (input type)') return False diff --git a/vleappGUI.py b/vleappGUI.py index 7c78953..05f69e9 100755 --- a/vleappGUI.py +++ b/vleappGUI.py @@ -575,10 +575,11 @@ def select_input(button_type): input_filename = tk_filedialog.askopenfilename(parent=main_window, title='Select a file', filetypes=(('All supported files', - '*.tar *.zip *.gz *.img *.bin *.dd *.raw'), + '*.tar *.zip *.gz *.img *.bin *.dd *.raw *.iVa'), ('tar file', '*.tar'), ('zip file', '*.zip'), ('gz file', '*.gz'), - ('raw disk image', '*.img *.bin *.dd *.raw'))) + ('raw disk image', '*.img *.bin *.dd *.raw'), + ('Berla iVe export', '*.iVa'))) else: input_filename = tk_filedialog.askdirectory(parent=main_window, title='Select a folder') input_entry.delete(0, 'end') From 321f8c9fe351036057bc19741c157f50e4700c34 Mon Sep 17 00:00:00 2001 From: Brigs Date: Sun, 30 Aug 2026 22:18:05 -0400 Subject: [PATCH 2/3] Validator: map .iVa by extension, take raw only by declaration input_type_for learns .iva, which is unambiguous, and honours an entry-level input_type field, which wins outright. A raw disk image is never guessed from its extension: the registry holds .bin files that are readable images and .bin files that are chip-level LUN dumps no walker opens, so raw is opt-in per entry, and an unknown declared type maps to nothing rather than something. Co-Authored-By: Claude Opus 5 --- admin/scripts/validate_sample_data.py | 26 +++++++++++++++++++++----- 1 file changed, 21 insertions(+), 5 deletions(-) diff --git a/admin/scripts/validate_sample_data.py b/admin/scripts/validate_sample_data.py index 6e34311..ceb1c52 100644 --- a/admin/scripts/validate_sample_data.py +++ b/admin/scripts/validate_sample_data.py @@ -261,14 +261,25 @@ def check_registry(artifacts, registry, registry_path, verify_hashes, report): f"artifacts cite {len(cited)}") -def input_type_for(source): +RUNNABLE_TYPES = ("fs", "tar", "zip", "gz", "file", "raw", "iva") + + +def input_type_for(source, declared=None): """Map an evidence source to the tool's -t input type. The registry's match key is spelled "zip" for historical reasons but may point at any container the tool accepts, and --emit may also hand this an extraction directory. Returns None when nothing names a known type, so the caller reports it instead of guessing. + + declared is an entry's optional "input_type" field and wins outright. It + exists because an extension cannot decide the raw case: the registry holds + .bin files that are readable disk images and .bin files that are chip-level + LUN dumps no walker opens, so raw is opt-in per entry rather than guessed. + .iVa is mapped by extension because it is unambiguous. """ + if declared: + return declared if declared in RUNNABLE_TYPES else None if source.is_dir(): return "fs" name = source.name.lower() @@ -278,6 +289,8 @@ def input_type_for(source): return "tar" if name.endswith(".zip"): return "zip" + if name.endswith(".iva"): + return "iva" return None @@ -307,7 +320,7 @@ def lava_output_predicate(report): return None -def execute_tool(source, label, log_name, report, keep=False): +def execute_tool(source, label, log_name, report, keep=False, declared_type=None): """Parse one evidence source and return its per-artifact LAVA row counts. Returns (produced, output_root, log_path); produced is None when the run @@ -318,10 +331,12 @@ def execute_tool(source, label, log_name, report, keep=False): deleted unless keep is set; the run log is written beside it and survives, because an artifact that produced nothing has usually logged why. """ - input_type = input_type_for(source) + input_type = input_type_for(source, declared_type) if input_type is None: report.error(f"{label}: cannot pick a -t input type for {source.name}; " - "known inputs are a directory, .zip, .tar, .tar.gz, .tgz") + "known inputs are a directory, .zip, .tar, .tar.gz, .tgz, " + ".iVa, or an entry-level \"input_type\" field (a raw disk " + "image is never guessed from its extension)") return None, None, None output_root = tempfile.mkdtemp(prefix=RUN_PREFIX) @@ -412,7 +427,8 @@ def run_corpus(registry, registry_path, corpus, report, keep=False): report.error(f"--run '{corpus}' points at a missing file: {source}") return - produced, _, _ = execute_tool(source, f"--run '{corpus}'", corpus, report, keep=keep) + produced, _, _ = execute_tool(source, f"--run '{corpus}'", corpus, report, keep=keep, + declared_type=entry.get("input_type")) if produced is None: return From e6ad2bc64ad3ffe40e0b948938febbe8dffb9785 Mon Sep 17 00:00:00 2001 From: Brigs Date: Sun, 30 Aug 2026 22:18:05 -0400 Subject: [PATCH 3/3] Follow the native .iVa route in the prose and the citations The unwrap script's docstring and both artifacts' notes described unwrapping as the only way to reach a .iVa's vehicle data, which -t iva makes untrue. The script remains the way to keep the intermediate zip for cheap re-runs, and the notes now say so instead. The Bluetooth artifacts' sample_data moves from the derived corpus entry to the evidence .iVa itself, re-measured through -t iva at the same 507, 126 and 2 rows, so the derived copy of the export's inner zip no longer needs to exist in the corpus. Co-Authored-By: Claude Opus 5 --- admin/scripts/unwrap_berla_iva.py | 10 ++++++---- scripts/artifacts/berla_ive_export.py | 22 +++++++++++----------- scripts/artifacts/ford_sync_bluetooth.py | 24 +++++++++++------------- 3 files changed, 28 insertions(+), 28 deletions(-) diff --git a/admin/scripts/unwrap_berla_iva.py b/admin/scripts/unwrap_berla_iva.py index c014cd7..b5d493f 100644 --- a/admin/scripts/unwrap_berla_iva.py +++ b/admin/scripts/unwrap_berla_iva.py @@ -12,10 +12,12 @@ 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. +VLEAPP reads a .iVa directly with -t iva, so this script is no longer the only +route. It remains the way to KEEP the intermediate zip: -t iva unpacks to a +temporary directory and removes it after the run, so on a large export the unwrap +cost is paid again on every re-run, while this script pays it once. It also +verifies the lifted zip against the SHA-256 iVe recorded for it, which the seeker +route leaves to the zip member CRCs. python3 admin/scripts/unwrap_berla_iva.py CASE.iVa -o outdir python3 vleapp.py -t zip -i outdir/DCASourceFilesUpload.zip -o reports diff --git a/scripts/artifacts/berla_ive_export.py b/scripts/artifacts/berla_ive_export.py index 2f908ee..d93562a 100644 --- a/scripts/artifacts/berla_ive_export.py +++ b/scripts/artifacts/berla_ive_export.py @@ -7,7 +7,7 @@ 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. +reaches it: -t iva, or admin/scripts/unwrap_berla_iva.py to keep the intermediate zip. """ import json @@ -29,11 +29,11 @@ "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 " + "archives, so with any input type other than iva a .iVa reaches only " + "this file and the vehicle's own data is not seen. Run the export " + "with -t iva, which reaches through to the raw image inside; " + "admin/scripts/unwrap_berla_iva.py remains the way to keep the " + "intermediate zip for cheap re-runs. 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 " @@ -49,11 +49,11 @@ "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.", + "reachable with the vendored qnxprobe, which is what -t iva uses: it " + "reaches through the export to the raw image and reads its QNX6 " + "volumes directly. On the tested export that route produced the same " + "rows as the vendor's own extracted file set and also surfaced a " + "fourth QNX6 volume the export carried no extracted files for.", "paths": ('*/Vehicle.json',), "sample_data": { "adams_ford_syncgen3_iva": "Berla iVe export, Ford Sync Gen3 | 4 rows", diff --git a/scripts/artifacts/ford_sync_bluetooth.py b/scripts/artifacts/ford_sync_bluetooth.py index 3791c06..47736ec 100644 --- a/scripts/artifacts/ford_sync_bluetooth.py +++ b/scripts/artifacts/ford_sync_bluetooth.py @@ -30,19 +30,17 @@ "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.", + "write-ahead log or journal on the tested unit. A Berla iVe export run " + "with -t iva reads these rows from the raw image the export carries, " + "through the head unit's own QNX6 volumes, so the values are the " + "unit's rather than iVe's parse of them. The same rows were also " + "reproduced from the file set iVe itself extracted: both routes 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", + "adams_ford_syncgen3_iva": "Ford Sync Gen3, via -t iva | 507 rows", "ford_syncg4_logical": "Ford Sync G4 | 0 rows, BT/btpbk not present", }, "output_types": "standard", @@ -75,7 +73,7 @@ "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", + "adams_ford_syncgen3_iva": "Ford Sync Gen3, via -t iva | 126 rows", "ford_syncg4_logical": "Ford Sync G4 | 0 rows, BT/btpbk not present", }, "output_types": "standard", @@ -102,7 +100,7 @@ "of device, vendor id and product id are reported as stored.", "paths": ('*/BT/btpersist*',), "sample_data": { - "adams_ford_syncgen3": "Ford Sync Gen3 | 2 rows", + "adams_ford_syncgen3_iva": "Ford Sync Gen3, via -t iva | 2 rows", "ford_syncg4_logical": "Ford Sync G4 | 0 rows, BT/btpersist not present", }, "output_types": "standard",