Add Ford Sync Gen3 Bluetooth artifacts and Berla iVe export handling

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:
+
+    <case>.iVa                        ZIP
+      Summary.json                    SHA-256 of every inner file
+      Vehicle.json                    vehicle and acquisition record
+      <case>.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 <report folder>")
+    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()

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)