From baf6cd1690b0765301643f221733a69f0ec653f9 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 15:17:59 +0500 Subject: [PATCH 01/22] cli: add --mp_per_plugin flag (experimental) --- ileapp.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/ileapp.py b/ileapp.py index 8caf43795..410927de7 100755 --- a/ileapp.py +++ b/ileapp.py @@ -203,6 +203,9 @@ def main(): help=("Path to a keychain file captured from the device. Some apps keep " "their database key in the keychain, which is collected separately " "from the file system extraction.")) + parser.add_argument('--mp_per_plugin', required=False, action="store_true", default=False, + help=("EXPERIMENTAL: Run each plugin in its own subprocess (spawn). " + "This enables skipping a long-running plugin (planned) without stopping the whole run.")) # Check if no arguments were provided if len(sys.argv) == 1: @@ -352,6 +355,9 @@ def main(): out_params = OutputParameters(output_path, custom_output_folder) Context.set_output_params(out_params) + if args.mp_per_plugin: + logfunc("EXPERIMENTAL MODE ENABLED: --mp_per_plugin (per-plugin subprocess execution)") + initialize_lava(input_path, out_params.output_folder_base, extracttype) # Record history if enabled From d59bf349eb55fbc61b24f3ad66378309e3908d17 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 15:18:03 +0500 Subject: [PATCH 02/22] lava: make initialize idempotent; add open/close helpers --- scripts/lavafuncs.py | 33 ++++++++++++++++++++++++++++++--- 1 file changed, 30 insertions(+), 3 deletions(-) diff --git a/scripts/lavafuncs.py b/scripts/lavafuncs.py index b229978b7..6d053afc0 100644 --- a/scripts/lavafuncs.py +++ b/scripts/lavafuncs.py @@ -44,6 +44,9 @@ lava_db_name = '_lava_artifacts.db' lava_json_name = '_lava_data.lava' +def _get_lava_db_path(output_path: str) -> str: + return os.path.join(output_path, lava_db_name) + def sanitize_sql_name(name): """ @@ -137,7 +140,7 @@ def initialize_lava(input_path, output_path, input_type): } } - db_path = os.path.join(output_path, lava_db_name) + db_path = _get_lava_db_path(output_path) lava_db = sqlite3.connect(db_path) cursor = lava_db.cursor() @@ -155,7 +158,7 @@ def initialize_lava(input_path, output_path, input_type): file_path_id INTEGER NOT NULL, FOREIGN KEY (artifact_search_pattern_id) REFERENCES _artifact_search_patterns(id), FOREIGN KEY (file_path_id) REFERENCES _file_path_list(id))''') - cursor.execute('''CREATE TABLE _lava_media_items ( + cursor.execute('''CREATE TABLE IF NOT EXISTS _lava_media_items ( id TEXT PRIMARY KEY, source_path TEXT, extraction_path TEXT, @@ -164,13 +167,15 @@ def initialize_lava(input_path, output_path, input_type): created_at INTEGER, updated_at INTEGER, is_embedded INTEGER)''') - cursor.execute('''CREATE TABLE _lava_media_references ( + cursor.execute('''CREATE TABLE IF NOT EXISTS _lava_media_references ( id TEXT PRIMARY KEY, media_item_id TEXT, module_name TEXT, artifact_name TEXT, name TEXT, FOREIGN KEY (media_item_id) REFERENCES _lava_media_items(id))''') + # Make view creation idempotent (eg, safe across subprocesses) + cursor.execute('''DROP VIEW IF EXISTS _lava_media_info''') cursor.execute('''CREATE VIEW _lava_media_info AS SELECT lmr.id as 'media_ref_id', @@ -187,6 +192,28 @@ def initialize_lava(input_path, output_path, input_type): lmi.is_embedded FROM _lava_media_references as lmr LEFT JOIN _lava_media_items as lmi ON lmr.media_item_id = lmi.id''') + lava_db.commit() + + +def lava_open_existing(output_path: str) -> None: + """ + Open an existing `_lava_artifacts.db` inside a child process. + + The parent process is expected to have created the database file already. + """ + global lava_db + db_path = _get_lava_db_path(output_path) + lava_db = sqlite3.connect(db_path) + + +def lava_close_db() -> None: + """Close the global LAVA sqlite connection if open.""" + global lava_db + if lava_db is not None: + try: + lava_db.close() + finally: + lava_db = None def lava_process_artifact( From 428d9266e4f6dc5e5cf3f2ec4493e0a2a9ab8e72 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 15:20:45 +0500 Subject: [PATCH 03/22] lava: add metadata delta + merge helpers for subprocess mode --- scripts/lavafuncs.py | 195 +++++++++++++++++++++++++++++++++++-------- 1 file changed, 161 insertions(+), 34 deletions(-) diff --git a/scripts/lavafuncs.py b/scripts/lavafuncs.py index 6d053afc0..9017bfe63 100644 --- a/scripts/lavafuncs.py +++ b/scripts/lavafuncs.py @@ -34,6 +34,7 @@ from collections import OrderedDict import re import datetime +import typing from scripts.version_info import leapp_name, leapp_version from scripts.context import Context @@ -216,6 +217,163 @@ def lava_close_db() -> None: lava_db = None +def _normalize_data_views_for_lava(data_views: typing.Optional[dict], data_headers) -> typing.Optional[dict]: + """ + Normalize & sanitize data views for storing in LAVA metadata. + + - Upgrades legacy 'chat' view to 'conversation' + - Sanitizes values that reference column names so they match table schema sanitization + """ + if not data_views: + return None + + # Deep copy (avoid mutating caller/module globals) + try: + normalized = json.loads(json.dumps(data_views)) + except Exception: + # If non-JSONable, fall back to shallow copy + normalized = dict(data_views) + + # Backward compatibility for chat view. Remove 'chat' once modules are updated. + if "chat" in normalized: + view_params = normalized.pop("chat") + normalized["conversation"] = view_params + + view_params = normalized.get("conversation") + if not view_params: + return normalized + + # Get original column names for dynamic sanitization check + column_names = [item[0] if isinstance(item, tuple) else item for item in data_headers] + + # Conversion map for backward compatibility. Remove once modules are updated. + convert_map = { + "threadDiscriminatorColumn": "conversationDiscriminatorColumn", + "threadLabelColumn": "conversationLabelColumn", + } + + sanitized_params = {} + for key, value in view_params.items(): + final_key = convert_map.get(key, key) + + # Sanitize value if it's a column name, otherwise pass through + if value in column_names: + sanitized_params[final_key] = sanitize_sql_name(value) + else: + sanitized_params[final_key] = value + + normalized["conversation"] = sanitized_params + return normalized + + +def lava_build_artifact_meta_delta( + *, + category: str, + module_name: str, + module_filename: str, + artifact_name: str, + artifact_info: dict, + table_name: str, + column_map: dict, + object_columns: typing.Optional[dict] = None, + record_count: typing.Optional[int] = None, + artifact_icon: typing.Optional[str] = None, + source_path: typing.Optional[str] = None, + data_headers=None, + data_views: typing.Optional[dict] = None, +) -> dict: + """ + Build a pure-data delta for the LAVA metadata structure (picklable). + + Intended to be produced in a subprocess and merged into the parent's in-memory + lava_data, so the parent can write a correct `_lava_data.lava` at the end. + """ + artifact_meta = { + "artifact_key": table_name, + "tablename": table_name, + "name": artifact_name, + "description": artifact_info.get('description', ''), + "author": artifact_info.get('author', ''), + "created_date": artifact_info.get('creation_date', ''), + "last_updated_date": artifact_info.get('last_update_date', ''), + "notes": artifact_info.get('notes', ''), + "category": category + } + + artifact = { + "name": artifact_name, + "tablename": table_name, + "module": module_name, + "column_map": column_map + } + + if artifact_icon: + artifact['artifact_icon'] = artifact_icon + + if record_count is not None: + artifact["record_count"] = record_count + + if source_path: + artifact['source_path'] = source_path + + if object_columns: + artifact["object_columns"] = [{"name": name, "type": type_} for name, type_ in object_columns.items()] + + if data_views: + normalized = _normalize_data_views_for_lava(data_views, data_headers) if data_headers else data_views + if normalized: + artifact['data_views'] = normalized + + return { + "meta_modules": [{ + "module_name": module_name, + "module_filename": module_filename, + "artifacts": [artifact_meta] + }], + "artifacts": { + category: [artifact] + } + } + + +def lava_merge_meta_delta(lava_data_obj: dict, delta: dict) -> None: + """ + Merge a delta created by lava_build_artifact_meta_delta() into lava_data_obj. + """ + if not delta: + return + + lava_data_obj.setdefault("artifacts", OrderedDict()) + lava_data_obj.setdefault("meta", {}).setdefault("modules", []) + + # Merge artifacts by category + for category, artifacts in (delta.get("artifacts") or {}).items(): + if category not in lava_data_obj["artifacts"]: + lava_data_obj["artifacts"][category] = [] + existing = {a.get("tablename") for a in lava_data_obj["artifacts"][category]} + for artifact in artifacts: + if artifact.get("tablename") not in existing: + lava_data_obj["artifacts"][category].append(artifact) + + # Merge meta modules + artifact meta + for mod in delta.get("meta_modules") or []: + module_name = mod.get("module_name") + module_filename = mod.get("module_filename") + module_info = next((m for m in lava_data_obj["meta"]["modules"] if m.get("module_name") == module_name), None) + if not module_info: + module_info = { + "module_name": module_name, + "module_filename": module_filename, + "artifacts": [] + } + lava_data_obj["meta"]["modules"].append(module_info) + + existing_keys = {a.get("artifact_key") for a in (module_info.get("artifacts") or [])} + for artifact_meta in mod.get("artifacts") or []: + if artifact_meta.get("artifact_key") not in existing_keys: + module_info["artifacts"].append(artifact_meta) + + def lava_process_artifact( category, module_name, @@ -294,40 +452,9 @@ def lava_process_artifact( artifact["object_columns"] = [{"name": name, "type": type_} for name, type_ in object_columns.items()] if data_views: - view_params = None - - # Backward compatibility for chat view. Remove 'chat' once modules are updated. - if "chat" in data_views: - view_params = data_views.pop("chat") - data_views["conversation"] = view_params # Upgrade to conversation - elif "conversation" in data_views: - view_params = data_views.get("conversation") - - if view_params: - sanitized_params = {} - - # Get original column names for dynamic sanitization check - column_names = [item[0] if isinstance(item, tuple) else item for item in data] - - # Conversion map for backward compatibility. Remove once modules are updated. - convert_map = { - "threadDiscriminatorColumn": "conversationDiscriminatorColumn", - "threadLabelColumn": "conversationLabelColumn" - } - - for key, value in view_params.items(): - # Remap old keys to new keys - final_key = convert_map.get(key, key) - - # Sanitize value if it's a column name, otherwise pass through - if value in column_names: - sanitized_params[final_key] = sanitize_sql_name(value) - else: - sanitized_params[final_key] = value - - data_views["conversation"] = sanitized_params - - artifact['data_views'] = data_views + normalized = _normalize_data_views_for_lava(data_views, data) + if normalized: + artifact['data_views'] = normalized lava_data["artifacts"][category].append(artifact) From d3be2bb343b6e6ab650ac400f2be686af24d7adb Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 15:24:27 +0500 Subject: [PATCH 04/22] mp: add output params snapshot + seeker proxy scaffolding --- scripts/ilapfuncs.py | 82 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/scripts/ilapfuncs.py b/scripts/ilapfuncs.py index 849768480..f129a6a72 100644 --- a/scripts/ilapfuncs.py +++ b/scripts/ilapfuncs.py @@ -2,6 +2,7 @@ import codecs # pylint: disable=unused-import # re-exported import contextlib import csv +import dataclasses import hashlib import inspect import io @@ -111,6 +112,87 @@ def __init__(self, output_folder, custom_folder_name=None): os.makedirs(self.media_folder, exist_ok=True) os.makedirs(self.html_media_folder, exist_ok=True) + +@dataclasses.dataclass(frozen=True) +class OutputParametersExisting: + """ + Lightweight, picklable output parameters for subprocess use. + + Unlike OutputParameters.__init__, this does NOT create a new timestamped report folder. + It is meant to point at an already-initialized output folder tree created by the parent. + """ + output_folder_base: str + data_folder: str + media_folder: str + html_media_folder: str + + screen_output_file_path: str + screen_output_file_path_devinfo: str + screen_output_file_path_lava_only: str + + +def output_params_from_existing_output_folder_base( + output_folder_base: str, + *, + ensure_dirs: bool = False) -> OutputParametersExisting: + """ + Build OutputParametersExisting from an existing output folder base. + + If ensure_dirs=True, ensures the expected log/data/media directories exist (exist_ok=True). + """ + data_folder = os.path.join(output_folder_base, 'data') + media_folder = os.path.join(output_folder_base, 'media') + html_media_folder = os.path.join(output_folder_base, '_HTML', 'media') + + screen_output_file_path = os.path.join( + output_folder_base, '_HTML', '_Script_Logs', 'Screen_Output.html') + screen_output_file_path_devinfo = os.path.join( + output_folder_base, '_HTML', '_Script_Logs', 'DeviceInfo.html') + screen_output_file_path_lava_only = os.path.join( + output_folder_base, '_HTML', '_Script_Logs', 'Lava_only_artifacts_log.html') + + if ensure_dirs: + os.makedirs(os.path.join(output_folder_base, '_HTML', '_Script_Logs'), exist_ok=True) + os.makedirs(data_folder, exist_ok=True) + os.makedirs(media_folder, exist_ok=True) + os.makedirs(html_media_folder, exist_ok=True) + + # logfunc() writes to these static paths, so we must set them in subprocesses too. + OutputParameters.screen_output_file_path = screen_output_file_path + OutputParameters.screen_output_file_path_devinfo = screen_output_file_path_devinfo + OutputParameters.screen_output_file_path_lava_only = screen_output_file_path_lava_only + + return OutputParametersExisting( + output_folder_base=output_folder_base, + data_folder=data_folder, + media_folder=media_folder, + html_media_folder=html_media_folder, + screen_output_file_path=screen_output_file_path, + screen_output_file_path_devinfo=screen_output_file_path_devinfo, + screen_output_file_path_lava_only=screen_output_file_path_lava_only, + ) + + +@dataclasses.dataclass(frozen=True) +class FileInfoSnapshot: + """ + Picklable snapshot of scripts.search_files.FileInfo. + """ + source_path: str + creation_date: float + modification_date: float + + +class SeekerProxy: + """ + Minimal seeker stand-in for subprocesses. + + Currently only provides .file_infos, which is used by media helpers to map extracted + paths back to original source paths and timestamps. + """ + def __init__(self, file_infos: dict[str, FileInfoSnapshot]): + self.file_infos = file_infos + class GuiWindow: '''This only exists to hold window handle if script is run from GUI''' window_handle = None # static variable From ad344ee6b57e439dd1e989ffb1c784ea6e03866c Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 15:30:00 +0500 Subject: [PATCH 05/22] mp: add per-plugin subprocess runner --- scripts/mp_plugin_runner.py | 190 ++++++++++++++++++++++++++++++++++++ 1 file changed, 190 insertions(+) create mode 100644 scripts/mp_plugin_runner.py diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py new file mode 100644 index 000000000..a6528bbcc --- /dev/null +++ b/scripts/mp_plugin_runner.py @@ -0,0 +1,190 @@ +""" +Multiprocessing helpers for iLEAPP. + +This module is designed to be importable in a spawned subprocess. +It provides a per-plugin runner that rehydrates minimal Context state and +returns small, picklable deltas back to the parent process (icons + LAVA metadata). +""" + +from __future__ import annotations + +import os +import traceback +import typing +from collections import OrderedDict + +import scripts.plugin_loader as plugin_loader +from scripts.context import Context +from scripts.ilapfuncs import ( + FileInfoSnapshot, + SeekerProxy, + check_output_types, + output_params_from_existing_output_folder_base, +) +import scripts.lavafuncs as lavafuncs + + +def _build_file_infos_snapshot(file_infos_subset: dict[str, tuple[str, float, float]]) -> dict[str, FileInfoSnapshot]: + """ + Convert a picklable subset dict from parent into FileInfoSnapshot objects. + + Input format: { extracted_path: (source_path, ctime, mtime) }. + """ + snapshot: dict[str, FileInfoSnapshot] = {} + for extracted_path, info_tuple in (file_infos_subset or {}).items(): + try: + source_path, ctime, mtime = info_tuple + snapshot[extracted_path] = FileInfoSnapshot( + source_path=str(source_path), + creation_date=float(ctime), + modification_date=float(mtime), + ) + except Exception: + # Best-effort: ignore malformed entries + continue + return snapshot + + +def _init_lava_metadata_only(input_path: str, output_path: str, input_type: str) -> None: + """ + Initialize lavafuncs.lava_data without creating/initializing tables. + + The parent process should have already created the DB file & base schema. + """ + lavafuncs.lava_data = { + "param_input": input_path, + "param_output": output_path, + "param_type": input_type, + "processing_status": "In Progress", + "lava_db_name": lavafuncs.lava_db_name, + "modules": [], + "artifacts": OrderedDict(), + "meta": {"modules": []}, + } + + +def run_one_plugin(payload: dict, result_queue) -> None: + """ + Run a single plugin in a subprocess. + + Expected payload keys: + - plugin_key: str + - files_found: list[str] + - category_folder: str + - wrap_text: bool + - time_offset: str + - output_folder_base: str + - input_path: str + - extracttype: str + - file_infos_subset: dict[str, tuple[str, float, float]] + """ + try: + plugin_key: str = payload["plugin_key"] + files_found: list[str] = payload.get("files_found") or [] + category_folder: str = payload["category_folder"] + wrap_text: bool = bool(payload.get("wrap_text", True)) + time_offset: str = payload.get("time_offset") or "UTC" + output_folder_base: str = payload["output_folder_base"] + input_path: str = payload.get("input_path") or "" + extracttype: str = payload.get("extracttype") or "" + file_infos_subset: dict[str, tuple[str, float, float]] = payload.get("file_infos_subset") or {} + + # Rehydrate output params + log paths for logfunc() + out_params_existing = output_params_from_existing_output_folder_base( + output_folder_base, ensure_dirs=True + ) + Context.set_output_params(out_params_existing) + + # Minimal seeker proxy (media helpers depend on seeker.file_infos) + seeker_proxy = SeekerProxy(_build_file_infos_snapshot(file_infos_subset)) + + # Load plugin inside child to avoid pickling callables + loader = plugin_loader.PluginLoader() + plugin_spec = loader[plugin_key] + + artifact_info = plugin_spec.artifact_info or {} + output_types = artifact_info.get("output_types", ["html", "tsv", "timeline", "lava", "kml"]) + + # Setup LAVA connection only if needed by this plugin + wants_lava = check_output_types("lava", output_types) or check_output_types("lava_only", output_types) + if wants_lava: + _init_lava_metadata_only(input_path, output_folder_base, extracttype) + lavafuncs.lava_open_existing(output_folder_base) + + # Execute plugin + data_headers, data_list, source_path = plugin_spec.method( + files_found, category_folder, seeker_proxy, wrap_text, time_offset + ) + + # Build deltas to send back + category = artifact_info.get("category", "") + artifact_name = artifact_info.get("name", plugin_key) + artifact_icon = artifact_info.get("artifact_icon", "") + + icons_delta: dict[str, dict[str, str]] = {} + if data_list and category: + icons_delta = {category: {artifact_name: artifact_icon}} + + lava_meta_delta: dict | None = None + lava_only_delta: list[dict] = [] + + if wants_lava and data_headers: + # Derive table name + maps consistently with lava_create_sqlite_table() + func_name = plugin_spec.name + table_name, column_map, object_columns = lavafuncs.lava_create_sqlite_table(func_name, data_headers) + + module_filename = "" + try: + module_filename = os.path.basename(Context.get_module_file_path()) + except Exception: + module_filename = f"{plugin_spec.module_name}.py" + + lava_meta_delta = lavafuncs.lava_build_artifact_meta_delta( + category=category, + module_name=plugin_spec.module_name, + module_filename=module_filename, + artifact_name=artifact_name, + artifact_info=artifact_info, + table_name=table_name, + column_map=column_map, + object_columns=object_columns, + record_count=len(data_list) if data_list else 0, + artifact_icon=artifact_icon, + source_path=source_path, + data_headers=data_headers, + data_views=artifact_info.get("data_views"), + ) + + # If lava_only, record for the parent's lava-only log + if check_output_types("lava_only", output_types): + lava_only_delta.append({ + "category": category, + "artifact_name": artifact_name, + "table_name": table_name, + "records": len(data_list) if data_list else 0, + }) + + result_queue.put({ + "ok": True, + "plugin_key": plugin_key, + "record_count": len(data_list) if data_list else 0, + "icons_delta": icons_delta, + "lava_meta_delta": lava_meta_delta, + "lava_only_delta": lava_only_delta, + }) + + except Exception as ex: + result_queue.put({ + "ok": False, + "plugin_key": payload.get("plugin_key"), + "error": str(ex), + "traceback": traceback.format_exc(), + }) + finally: + # Ensure DB connection is closed in the child + try: + lavafuncs.lava_close_db() + except Exception: + pass + + From ccba74c4fd62b93c6aa5d64aa4d2762d944ab458 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 15:31:06 +0500 Subject: [PATCH 06/22] mp: execute plugins in per-plugin subprocess mode --- ileapp.py | 118 ++++++++++++++++++++++++++++++++++++++++--- scripts/lavafuncs.py | 5 +- 2 files changed, 115 insertions(+), 8 deletions(-) diff --git a/ileapp.py b/ileapp.py index 410927de7..882ee2920 100755 --- a/ileapp.py +++ b/ileapp.py @@ -10,6 +10,7 @@ import scripts.report as report import traceback import sys +import multiprocessing import scripts.plugin_loader as plugin_loader import leapp_functions.app.history as history @@ -25,7 +26,7 @@ from scripts.context import Context from scripts.ios_keychain import report_supplied_keychain from scripts.lavafuncs import lava_json_name - +from scripts.mp_plugin_runner import run_one_plugin def validate_args(args): if args.artifact_paths or args.create_profile_casedata: @@ -359,19 +360,23 @@ def main(): logfunc("EXPERIMENTAL MODE ENABLED: --mp_per_plugin (per-plugin subprocess execution)") initialize_lava(input_path, out_params.output_folder_base, extracttype) + if args.mp_per_plugin: + # Parent does not need an open connection while children write to the DB. + lava_close_db() # Record history if enabled history.record_input_path(input_path) history.record_output_path(output_path) crunch_artifacts(selected_plugins, extracttype, input_path, out_params, wrap_text, loader, casedata, time_offset, - profile_filename, itunes_backup_password) + profile_filename, itunes_backup_password, decryption_keys=None, mp_per_plugin=args.mp_per_plugin) lava_finalize_output(out_params.output_folder_base) def crunch_artifacts( plugins: typing.Sequence[plugin_loader.PluginSpec], extracttype, input_path, out_params, wrap_text, - loader: plugin_loader.PluginLoader, casedata, time_offset, profile_filename, itunes_backup_password=None, decryption_keys=None): + loader: plugin_loader.PluginLoader, casedata, time_offset, profile_filename, + itunes_backup_password=None, decryption_keys=None, mp_per_plugin: bool = False): start = process_time() start_wall = perf_counter() @@ -544,11 +549,85 @@ def crunch_artifacts( logfunc('Error was {}'.format(str(ex))) continue # cannot do work try: - plugin.method(files_found, category_folder, seeker, wrap_text, time_offset) + if not mp_per_plugin: + plugin.method(files_found, category_folder, seeker, wrap_text, time_offset) + else: + ctx = multiprocessing.get_context("spawn") + q = ctx.Queue() + file_infos_subset = {} + try: + for pth in files_found: + fi = getattr(seeker, "file_infos", {}).get(pth) + if fi: + file_infos_subset[pth] = (fi.source_path, fi.creation_date, fi.modification_date) + except Exception: + file_infos_subset = {} + + payload = { + "plugin_key": plugin.name, + "files_found": files_found, + "category_folder": category_folder, + "wrap_text": wrap_text, + "time_offset": time_offset, + "output_folder_base": out_params.output_folder_base, + "input_path": input_path, + "extracttype": extracttype, + "file_infos_subset": file_infos_subset, + } + proc = ctx.Process(target=run_one_plugin, args=(payload, q)) + proc.start() + proc.join() + + result = None + if not q.empty(): + result = q.get() + if not result or not result.get("ok"): + err = (result or {}).get("error") if result else "Child process failed without result" + tb = (result or {}).get("traceback") if result else "" + raise RuntimeError(f"Subprocess plugin run failed for {plugin.name}: {err}\n{tb}") + + # Merge deltas into parent process globals (icons + LAVA meta + lava_only artifacts) + icons_delta = result.get("icons_delta") or {} + for cat, icon_map in icons_delta.items(): + icons.setdefault(cat, {}).update(icon_map) + + lava_meta_delta = result.get("lava_meta_delta") + if lava_meta_delta: + lava_merge_meta_delta(lava_data, lava_meta_delta) + + for item in (result.get("lava_only_delta") or []): + try: + lava_only_info(item["category"], item["artifact_name"], item["table_name"], item["records"]) + except Exception: + pass + if plugin.name == 'logarchive': lava_db_path = os.path.join(out_params.output_folder_base, '_lava_artifacts.db') if does_table_exist_in_db(lava_db_path, 'logarchive'): - loader["logarchive_artifacts"].method([lava_db_path], category_folder, seeker, wrap_text, time_offset) + if not mp_per_plugin: + loader["logarchive_artifacts"].method([lava_db_path], category_folder, seeker, wrap_text, time_offset) + else: + # run this follow-on artifact in its own subprocess too + ctx = multiprocessing.get_context("spawn") + q = ctx.Queue() + payload = { + "plugin_key": "logarchive_artifacts", + "files_found": [lava_db_path], + "category_folder": category_folder, + "wrap_text": wrap_text, + "time_offset": time_offset, + "output_folder_base": out_params.output_folder_base, + "input_path": input_path, + "extracttype": extracttype, + "file_infos_subset": {}, + } + proc = ctx.Process(target=run_one_plugin, args=(payload, q)) + proc.start() + proc.join() + if not q.empty(): + result = q.get() + if result and result.get("ok") and result.get("lava_meta_delta"): + lava_merge_meta_delta(lava_data, result["lava_meta_delta"]) if does_table_exist_in_db(lava_db_path, 'logarchive_artifacts'): unifed_logs_artifacts = [] unifed_logs_artifacts = [plugin.name for plugin in loader.plugins @@ -556,7 +635,34 @@ def crunch_artifacts( and plugin.name != 'logarchive' and plugin.name != 'logarchive_artifacts'] for unifed_log_artifact in unifed_logs_artifacts: - loader[unifed_log_artifact].method([lava_db_path], category_folder, seeker, wrap_text, time_offset) + if not mp_per_plugin: + loader[unifed_log_artifact].method([lava_db_path], category_folder, seeker, wrap_text, time_offset) + else: + ctx = multiprocessing.get_context("spawn") + q = ctx.Queue() + payload = { + "plugin_key": unifed_log_artifact, + "files_found": [lava_db_path], + "category_folder": category_folder, + "wrap_text": wrap_text, + "time_offset": time_offset, + "output_folder_base": out_params.output_folder_base, + "input_path": input_path, + "extracttype": extracttype, + "file_infos_subset": {}, + } + proc = ctx.Process(target=run_one_plugin, args=(payload, q)) + proc.start() + proc.join() + if not q.empty(): + result = q.get() + if result and result.get("ok"): + icons_delta = result.get("icons_delta") or {} + for cat, icon_map in icons_delta.items(): + icons.setdefault(cat, {}).update(icon_map) + lava_meta_delta = result.get("lava_meta_delta") + if lava_meta_delta: + lava_merge_meta_delta(lava_data, lava_meta_delta) except Exception as ex: # pylint: disable=broad-exception-caught logfunc('Reading {} artifact had errors!'.format(plugin.name)) logfunc('Error was {}'.format(str(ex))) diff --git a/scripts/lavafuncs.py b/scripts/lavafuncs.py index 9017bfe63..9db8d0849 100644 --- a/scripts/lavafuncs.py +++ b/scripts/lavafuncs.py @@ -852,5 +852,6 @@ def lava_finalize_output(output_path): with open(os.path.join(output_path, lava_json_name), 'w', encoding='utf-8') as f: json.dump(lava_data, f, indent=4) - # Close the SQLite database - lava_db.close() + # Close the SQLite database (may be None in per-plugin multiprocessing mode) + if lava_db is not None: + lava_db.close() From 532c4037a9680c8b1ee03038526cb95820630644 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 15:34:24 +0500 Subject: [PATCH 07/22] mp: run itunes Info.plist artifacts in subprocess mode --- ileapp.py | 143 +++++++++++++++++++++--------------------------------- 1 file changed, 55 insertions(+), 88 deletions(-) diff --git a/ileapp.py b/ileapp.py index 882ee2920..82c09a434 100755 --- a/ileapp.py +++ b/ileapp.py @@ -453,6 +453,50 @@ def crunch_artifacts( log.write(f'Extraction/Path selected: {input_path}

') log.write(f'Timezone selected: {time_offset}

') + ctx_mp = multiprocessing.get_context("spawn") if mp_per_plugin else None + + def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: str, file_infos_subset: dict | None = None): + """Run a plugin in a spawned subprocess and merge returned deltas into parent globals.""" + q = ctx_mp.Queue() + payload = { + "plugin_key": plugin_key, + "files_found": files_found, + "category_folder": category_folder, + "wrap_text": wrap_text, + "time_offset": time_offset, + "output_folder_base": out_params.output_folder_base, + "input_path": input_path, + "extracttype": extracttype, + "file_infos_subset": file_infos_subset or {}, + } + proc = ctx_mp.Process(target=run_one_plugin, args=(payload, q)) + proc.start() + proc.join() + + result = None + if not q.empty(): + result = q.get() + if not result or not result.get("ok"): + err = (result or {}).get("error") if result else "Child process failed without result" + tb = (result or {}).get("traceback") if result else "" + raise RuntimeError(f"Subprocess plugin run failed for {plugin_key}: {err}\n{tb}") + + icons_delta = result.get("icons_delta") or {} + for cat, icon_map in icons_delta.items(): + icons.setdefault(cat, {}).update(icon_map) + + lava_meta_delta = result.get("lava_meta_delta") + if lava_meta_delta: + lava_merge_meta_delta(lava_data, lava_meta_delta) + + for item in (result.get("lava_only_delta") or []): + try: + lava_only_info(item["category"], item["artifact_name"], item["table_name"], item["records"]) + except Exception: + pass + + return result + # Special processing for iTunesBackup Info.plist as it is a seperate entity, not part of the Manifest.db. Seeker won't find it if extracttype == 'itunes': info_plist_path = os.path.join(input_path, 'Info.plist') @@ -466,7 +510,10 @@ def crunch_artifacts( except (FileExistsError, FileNotFoundError) as ex: logfunc('Error creating report directory at path {}'.format(report_folder)) logfunc('Error was {}'.format(str(ex))) - loader["itunes_backup_info"].method([info_plist_path], report_folder, seeker, wrap_text, time_offset) + if not mp_per_plugin: + loader["itunes_backup_info"].method([info_plist_path], report_folder, seeker, wrap_text, time_offset) + else: + _run_plugin_subprocess("itunes_backup_info", [info_plist_path], report_folder) report_folder = os.path.join(out_params.output_folder_base, '_HTML', 'Installed Apps') if not os.path.exists(report_folder): try: @@ -474,7 +521,10 @@ def crunch_artifacts( except (FileExistsError, FileNotFoundError) as ex: logfunc('Error creating report directory at path {}'.format(report_folder)) logfunc('Error was {}'.format(str(ex))) - loader["itunes_backup_installed_applications"].method([info_plist_path], report_folder, seeker, wrap_text, time_offset) + if not mp_per_plugin: + loader["itunes_backup_installed_applications"].method([info_plist_path], report_folder, seeker, wrap_text, time_offset) + else: + _run_plugin_subprocess("itunes_backup_installed_applications", [info_plist_path], report_folder) #del search_list['last_build'] # removing last_build as this takes its place print([info_plist_path]) # Future: remove special consideration for itunes? Merge into main search else: @@ -552,8 +602,6 @@ def crunch_artifacts( if not mp_per_plugin: plugin.method(files_found, category_folder, seeker, wrap_text, time_offset) else: - ctx = multiprocessing.get_context("spawn") - q = ctx.Queue() file_infos_subset = {} try: for pth in files_found: @@ -562,44 +610,7 @@ def crunch_artifacts( file_infos_subset[pth] = (fi.source_path, fi.creation_date, fi.modification_date) except Exception: file_infos_subset = {} - - payload = { - "plugin_key": plugin.name, - "files_found": files_found, - "category_folder": category_folder, - "wrap_text": wrap_text, - "time_offset": time_offset, - "output_folder_base": out_params.output_folder_base, - "input_path": input_path, - "extracttype": extracttype, - "file_infos_subset": file_infos_subset, - } - proc = ctx.Process(target=run_one_plugin, args=(payload, q)) - proc.start() - proc.join() - - result = None - if not q.empty(): - result = q.get() - if not result or not result.get("ok"): - err = (result or {}).get("error") if result else "Child process failed without result" - tb = (result or {}).get("traceback") if result else "" - raise RuntimeError(f"Subprocess plugin run failed for {plugin.name}: {err}\n{tb}") - - # Merge deltas into parent process globals (icons + LAVA meta + lava_only artifacts) - icons_delta = result.get("icons_delta") or {} - for cat, icon_map in icons_delta.items(): - icons.setdefault(cat, {}).update(icon_map) - - lava_meta_delta = result.get("lava_meta_delta") - if lava_meta_delta: - lava_merge_meta_delta(lava_data, lava_meta_delta) - - for item in (result.get("lava_only_delta") or []): - try: - lava_only_info(item["category"], item["artifact_name"], item["table_name"], item["records"]) - except Exception: - pass + _run_plugin_subprocess(plugin.name, files_found, category_folder, file_infos_subset) if plugin.name == 'logarchive': lava_db_path = os.path.join(out_params.output_folder_base, '_lava_artifacts.db') @@ -607,27 +618,7 @@ def crunch_artifacts( if not mp_per_plugin: loader["logarchive_artifacts"].method([lava_db_path], category_folder, seeker, wrap_text, time_offset) else: - # run this follow-on artifact in its own subprocess too - ctx = multiprocessing.get_context("spawn") - q = ctx.Queue() - payload = { - "plugin_key": "logarchive_artifacts", - "files_found": [lava_db_path], - "category_folder": category_folder, - "wrap_text": wrap_text, - "time_offset": time_offset, - "output_folder_base": out_params.output_folder_base, - "input_path": input_path, - "extracttype": extracttype, - "file_infos_subset": {}, - } - proc = ctx.Process(target=run_one_plugin, args=(payload, q)) - proc.start() - proc.join() - if not q.empty(): - result = q.get() - if result and result.get("ok") and result.get("lava_meta_delta"): - lava_merge_meta_delta(lava_data, result["lava_meta_delta"]) + _run_plugin_subprocess("logarchive_artifacts", [lava_db_path], category_folder, {}) if does_table_exist_in_db(lava_db_path, 'logarchive_artifacts'): unifed_logs_artifacts = [] unifed_logs_artifacts = [plugin.name for plugin in loader.plugins @@ -638,31 +629,7 @@ def crunch_artifacts( if not mp_per_plugin: loader[unifed_log_artifact].method([lava_db_path], category_folder, seeker, wrap_text, time_offset) else: - ctx = multiprocessing.get_context("spawn") - q = ctx.Queue() - payload = { - "plugin_key": unifed_log_artifact, - "files_found": [lava_db_path], - "category_folder": category_folder, - "wrap_text": wrap_text, - "time_offset": time_offset, - "output_folder_base": out_params.output_folder_base, - "input_path": input_path, - "extracttype": extracttype, - "file_infos_subset": {}, - } - proc = ctx.Process(target=run_one_plugin, args=(payload, q)) - proc.start() - proc.join() - if not q.empty(): - result = q.get() - if result and result.get("ok"): - icons_delta = result.get("icons_delta") or {} - for cat, icon_map in icons_delta.items(): - icons.setdefault(cat, {}).update(icon_map) - lava_meta_delta = result.get("lava_meta_delta") - if lava_meta_delta: - lava_merge_meta_delta(lava_data, lava_meta_delta) + _run_plugin_subprocess(unifed_log_artifact, [lava_db_path], category_folder, {}) except Exception as ex: # pylint: disable=broad-exception-caught logfunc('Reading {} artifact had errors!'.format(plugin.name)) logfunc('Error was {}'.format(str(ex))) From f3616c71810037babd9f85702eacae5354a87a68 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 15:35:53 +0500 Subject: [PATCH 08/22] mp: fix lava_data merge to use module global --- ileapp.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/ileapp.py b/ileapp.py index 82c09a434..5941ef3f2 100755 --- a/ileapp.py +++ b/ileapp.py @@ -14,6 +14,7 @@ import scripts.plugin_loader as plugin_loader import leapp_functions.app.history as history +import scripts.lavafuncs as lavafuncs from shutil import copy2 from getpass import getpass @@ -487,7 +488,8 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: lava_meta_delta = result.get("lava_meta_delta") if lava_meta_delta: - lava_merge_meta_delta(lava_data, lava_meta_delta) + # IMPORTANT: use module global, not the `lava_data` name imported into this module. + lavafuncs.lava_merge_meta_delta(lavafuncs.lava_data, lava_meta_delta) for item in (result.get("lava_only_delta") or []): try: From dc7717a5980b3e11fd1bb3767518833e9a740bde Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 18:11:56 +0500 Subject: [PATCH 09/22] mp: propagate installed iOS version across plugin subprocesses --- ileapp.py | 13 +++++++++++++ scripts/mp_plugin_runner.py | 19 +++++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/ileapp.py b/ileapp.py index 5941ef3f2..961ddd89f 100755 --- a/ileapp.py +++ b/ileapp.py @@ -455,9 +455,11 @@ def crunch_artifacts( log.write(f'Timezone selected: {time_offset}

') ctx_mp = multiprocessing.get_context("spawn") if mp_per_plugin else None + installed_os_version = '' def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: str, file_infos_subset: dict | None = None): """Run a plugin in a spawned subprocess and merge returned deltas into parent globals.""" + nonlocal installed_os_version q = ctx_mp.Queue() payload = { "plugin_key": plugin_key, @@ -469,6 +471,7 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: "input_path": input_path, "extracttype": extracttype, "file_infos_subset": file_infos_subset or {}, + "installed_os_version": installed_os_version, } proc = ctx_mp.Process(target=run_one_plugin, args=(payload, q)) proc.start() @@ -497,6 +500,16 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: except Exception: pass + # Keep installed OS version in parent so we can pass it to future subprocesses + discovered_os_version = result.get("installed_os_version") or '' + if discovered_os_version: + installed_os_version = discovered_os_version + try: + iOS.set_version(discovered_os_version) + Context.set_installed_os_version(discovered_os_version) + except Exception: + pass + return result # Special processing for iTunesBackup Info.plist as it is a seperate entity, not part of the Manifest.db. Seeker won't find it diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py index a6528bbcc..d16a5b392 100644 --- a/scripts/mp_plugin_runner.py +++ b/scripts/mp_plugin_runner.py @@ -19,6 +19,7 @@ FileInfoSnapshot, SeekerProxy, check_output_types, + iOS, output_params_from_existing_output_folder_base, ) import scripts.lavafuncs as lavafuncs @@ -77,6 +78,7 @@ def run_one_plugin(payload: dict, result_queue) -> None: - input_path: str - extracttype: str - file_infos_subset: dict[str, tuple[str, float, float]] + - installed_os_version: str (optional) """ try: plugin_key: str = payload["plugin_key"] @@ -88,6 +90,7 @@ def run_one_plugin(payload: dict, result_queue) -> None: input_path: str = payload.get("input_path") or "" extracttype: str = payload.get("extracttype") or "" file_infos_subset: dict[str, tuple[str, float, float]] = payload.get("file_infos_subset") or {} + installed_os_version: str = payload.get("installed_os_version") or "" # Rehydrate output params + log paths for logfunc() out_params_existing = output_params_from_existing_output_folder_base( @@ -95,6 +98,14 @@ def run_one_plugin(payload: dict, result_queue) -> None: ) Context.set_output_params(out_params_existing) + # Propagate installed iOS version into the subprocess (many plugins depend on it) + if installed_os_version: + try: + iOS.set_version(installed_os_version) + Context.set_installed_os_version(installed_os_version) + except Exception: + pass + # Minimal seeker proxy (media helpers depend on seeker.file_infos) seeker_proxy = SeekerProxy(_build_file_infos_snapshot(file_infos_subset)) @@ -116,6 +127,13 @@ def run_one_plugin(payload: dict, result_queue) -> None: files_found, category_folder, seeker_proxy, wrap_text, time_offset ) + # Capture installed OS version if this plugin discovered it + discovered_os_version = "" + try: + discovered_os_version = Context.get_installed_os_version() or "" + except Exception: + discovered_os_version = "" + # Build deltas to send back category = artifact_info.get("category", "") artifact_name = artifact_info.get("name", plugin_key) @@ -171,6 +189,7 @@ def run_one_plugin(payload: dict, result_queue) -> None: "icons_delta": icons_delta, "lava_meta_delta": lava_meta_delta, "lava_only_delta": lava_only_delta, + "installed_os_version": discovered_os_version, }) except Exception as ex: From bfc910ed464ba98cec302a214d3b9292ad493787 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 18:45:06 +0500 Subject: [PATCH 10/22] mp: handle Ctrl+C as skip via terminate; use SimpleQueue --- ileapp.py | 60 +++++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 56 insertions(+), 4 deletions(-) diff --git a/ileapp.py b/ileapp.py index 961ddd89f..0854cc27d 100755 --- a/ileapp.py +++ b/ileapp.py @@ -11,6 +11,8 @@ import traceback import sys import multiprocessing +import signal +import time import scripts.plugin_loader as plugin_loader import leapp_functions.app.history as history @@ -456,11 +458,45 @@ def crunch_artifacts( ctx_mp = multiprocessing.get_context("spawn") if mp_per_plugin else None installed_os_version = '' + current_proc = None + last_interrupt_ts = 0.0 + + def _terminate_current_plugin_proc(reason: str): + nonlocal current_proc + if current_proc is not None and current_proc.is_alive(): + logfunc(f"Skip requested ({reason}). Terminating current plugin subprocess (pid={current_proc.pid}) ...") + try: + current_proc.terminate() + except Exception: + pass + + def _interrupt_handler(signum, frame): + """ + Ctrl+C / Ctrl+Break handling for mp mode: + - first press: terminate current plugin process and continue + - second press within 2 seconds: abort run + """ + nonlocal last_interrupt_ts + now = time.time() + if now - last_interrupt_ts < 2.0: + logfunc("Second interrupt received. Aborting run.") + raise KeyboardInterrupt + last_interrupt_ts = now + _terminate_current_plugin_proc("SIGINT/SIGBREAK") + + if mp_per_plugin: + # Register interrupt handler (cross-platform): + # - SIGINT is Ctrl+C everywhere + # - SIGBREAK is Ctrl+Break on Windows (if present) + signal.signal(signal.SIGINT, _interrupt_handler) + if hasattr(signal, "SIGBREAK"): + signal.signal(signal.SIGBREAK, _interrupt_handler) def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: str, file_infos_subset: dict | None = None): """Run a plugin in a spawned subprocess and merge returned deltas into parent globals.""" nonlocal installed_os_version - q = ctx_mp.Queue() + nonlocal current_proc + q = ctx_mp.SimpleQueue() payload = { "plugin_key": plugin_key, "files_found": files_found, @@ -474,13 +510,29 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: "installed_os_version": installed_os_version, } proc = ctx_mp.Process(target=run_one_plugin, args=(payload, q)) + current_proc = proc proc.start() - proc.join() + + # Join in a loop so we can react to Ctrl+C and treat it as "skip current plugin" + while proc.is_alive(): + proc.join(timeout=0.25) result = None - if not q.empty(): - result = q.get() + try: + # SimpleQueue doesn't reliably support .empty() cross-platform; just try get_nowait + if hasattr(q, "get_nowait"): + result = q.get_nowait() + else: + # Fallback: try blocking very briefly + result = q.get(timeout=0.01) + except Exception: + result = None if not result or not result.get("ok"): + # If we killed the process due to interrupt/skip, treat it as a skip and keep going. + if proc.exitcode is not None and proc.exitcode < 0: + logfunc(f"Plugin {plugin_key} was interrupted (exitcode={proc.exitcode}). Skipping.") + current_proc = None + return {"ok": True, "plugin_key": plugin_key, "skipped": True} err = (result or {}).get("error") if result else "Child process failed without result" tb = (result or {}).get("traceback") if result else "" raise RuntimeError(f"Subprocess plugin run failed for {plugin_key}: {err}\n{tb}") From 85dc33a9b21de638d42c11a76e2fc3eebf453bed Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 18:49:35 +0500 Subject: [PATCH 11/22] mp: read SimpleQueue result without unsupported timeout --- ileapp.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/ileapp.py b/ileapp.py index 0854cc27d..04c42753a 100755 --- a/ileapp.py +++ b/ileapp.py @@ -519,12 +519,9 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: result = None try: - # SimpleQueue doesn't reliably support .empty() cross-platform; just try get_nowait - if hasattr(q, "get_nowait"): - result = q.get_nowait() - else: - # Fallback: try blocking very briefly - result = q.get(timeout=0.01) + # multiprocessing.SimpleQueue.get() has no timeout; guard with empty() to avoid blocking. + if not q.empty(): + result = q.get() except Exception: result = None if not result or not result.get("ok"): From 04abe07eaba642dd643948fa2859f7ddadfacbd5 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 19:09:04 +0500 Subject: [PATCH 12/22] mp: fix SIGINT handler time shadowing; add SeekerProxy.search --- ileapp.py | 16 +++++++++++++-- scripts/ilapfuncs.py | 41 ++++++++++++++++++++++++++++++++++++- scripts/mp_plugin_runner.py | 4 +++- 3 files changed, 57 insertions(+), 4 deletions(-) diff --git a/ileapp.py b/ileapp.py index 04c42753a..430b9ca9e 100755 --- a/ileapp.py +++ b/ileapp.py @@ -12,7 +12,7 @@ import sys import multiprocessing import signal -import time +import time as time_module import scripts.plugin_loader as plugin_loader import leapp_functions.app.history as history @@ -460,6 +460,16 @@ def crunch_artifacts( installed_os_version = '' current_proc = None last_interrupt_ts = 0.0 + seeker_all_files = [] + try: + # For iTunes backups, seeker._all_files is a dict keyed by "virtual paths" in backup + # (eg. private/var/mobile/Library/...). + if hasattr(seeker, "_all_files") and isinstance(seeker._all_files, dict): + seeker_all_files = list(seeker._all_files.keys()) + elif hasattr(seeker, "_all_files") and isinstance(seeker._all_files, list): + seeker_all_files = list(seeker._all_files) + except Exception: + seeker_all_files = [] def _terminate_current_plugin_proc(reason: str): nonlocal current_proc @@ -477,7 +487,8 @@ def _interrupt_handler(signum, frame): - second press within 2 seconds: abort run """ nonlocal last_interrupt_ts - now = time.time() + # IMPORTANT: use time module explicitly; `time` can be shadowed by datetime.time via star-imports. + now = time_module.time() if now - last_interrupt_ts < 2.0: logfunc("Second interrupt received. Aborting run.") raise KeyboardInterrupt @@ -508,6 +519,7 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: "extracttype": extracttype, "file_infos_subset": file_infos_subset or {}, "installed_os_version": installed_os_version, + "seeker_all_files": seeker_all_files, } proc = ctx_mp.Process(target=run_one_plugin, args=(payload, q)) current_proc = proc diff --git a/scripts/ilapfuncs.py b/scripts/ilapfuncs.py index f129a6a72..5606992eb 100644 --- a/scripts/ilapfuncs.py +++ b/scripts/ilapfuncs.py @@ -16,6 +16,7 @@ import shutil import sqlite3 import sys +import typing import xml from datetime import datetime, timezone, timedelta @@ -190,8 +191,46 @@ class SeekerProxy: Currently only provides .file_infos, which is used by media helpers to map extracted paths back to original source paths and timestamps. """ - def __init__(self, file_infos: dict[str, FileInfoSnapshot]): + def __init__(self, file_infos: dict[str, FileInfoSnapshot], all_files: typing.Optional[list[str]] = None): self.file_infos = file_infos + # Optional lightweight search index: list of "virtual paths" (eg. iTunes full paths like + # private/var/mobile/Library/...). + self._all_files = all_files or [] + self.searched = {} + + def search(self, filepattern, return_on_first_hit: bool = False, force: bool = False): + """ + Lightweight seeker.search() for subprocess mode. + + This mimics FileSeekerItunes.search() behavior for matching (fnmatch against virtual paths), + but returns *already-extracted* paths only. It does NOT perform extraction/copying. + It exists so plugins that call seeker.search() for auxiliary files don't crash in mp mode. + """ + try: + if filepattern in self.searched and not force: + pathlist = self.searched[filepattern] + return pathlist[0] if return_on_first_hit and pathlist else pathlist + + import fnmatch as _fnmatch # local import to keep module import side-effects minimal + + matches = _fnmatch.filter(self._all_files, filepattern) if self._all_files else [] + + # Map virtual paths to extracted paths via known file_infos (only those copied/extracted so far) + found = [] + if matches: + for extracted_path, finfo in (self.file_infos or {}).items(): + try: + if finfo and finfo.source_path in matches: + found.append(extracted_path) + if return_on_first_hit: + break + except Exception: + continue + + self.searched[filepattern] = found + return found[0] if return_on_first_hit and found else found + except Exception: + return "" if return_on_first_hit else [] class GuiWindow: '''This only exists to hold window handle if script is run from GUI''' diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py index d16a5b392..925639488 100644 --- a/scripts/mp_plugin_runner.py +++ b/scripts/mp_plugin_runner.py @@ -79,6 +79,7 @@ def run_one_plugin(payload: dict, result_queue) -> None: - extracttype: str - file_infos_subset: dict[str, tuple[str, float, float]] - installed_os_version: str (optional) + - seeker_all_files: list[str] (optional) """ try: plugin_key: str = payload["plugin_key"] @@ -91,6 +92,7 @@ def run_one_plugin(payload: dict, result_queue) -> None: extracttype: str = payload.get("extracttype") or "" file_infos_subset: dict[str, tuple[str, float, float]] = payload.get("file_infos_subset") or {} installed_os_version: str = payload.get("installed_os_version") or "" + seeker_all_files: list[str] = payload.get("seeker_all_files") or [] # Rehydrate output params + log paths for logfunc() out_params_existing = output_params_from_existing_output_folder_base( @@ -107,7 +109,7 @@ def run_one_plugin(payload: dict, result_queue) -> None: pass # Minimal seeker proxy (media helpers depend on seeker.file_infos) - seeker_proxy = SeekerProxy(_build_file_infos_snapshot(file_infos_subset)) + seeker_proxy = SeekerProxy(_build_file_infos_snapshot(file_infos_subset), seeker_all_files) # Load plugin inside child to avoid pickling callables loader = plugin_loader.PluginLoader() From dba64c0a582dc512ae62af5d96f60b45dc8a442c Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 19:33:18 +0500 Subject: [PATCH 13/22] debug: instrument mp hang around join/queue and SIGINT --- ileapp.py | 96 ++++++++++++++++++++++++++++++++++++- scripts/mp_plugin_runner.py | 44 +++++++++++++++++ 2 files changed, 139 insertions(+), 1 deletion(-) diff --git a/ileapp.py b/ileapp.py index 430b9ca9e..3d3e8307e 100755 --- a/ileapp.py +++ b/ileapp.py @@ -13,6 +13,7 @@ import multiprocessing import signal import time as time_module +import json as _agent_json import scripts.plugin_loader as plugin_loader import leapp_functions.app.history as history @@ -460,6 +461,25 @@ def crunch_artifacts( installed_os_version = '' current_proc = None last_interrupt_ts = 0.0 + subprocess_phase = "idle" + + #region agent log + def _agent_log(hypothesis_id: str, location: str, message: str, data: dict): + try: + payload = { + "sessionId": "debug-session", + "runId": "hang1", + "hypothesisId": hypothesis_id, + "location": location, + "message": message, + "data": data, + "timestamp": int(time_module.time() * 1000), + } + with open("/Users/pl-2134/Development/iLEAPP/.cursor/debug.log", "a", encoding="utf-8") as f: + f.write(_agent_json.dumps(payload, ensure_ascii=False) + "\n") + except Exception: + pass + #endregion agent log seeker_all_files = [] try: # For iTunes backups, seeker._all_files is a dict keyed by "virtual paths" in backup @@ -487,8 +507,24 @@ def _interrupt_handler(signum, frame): - second press within 2 seconds: abort run """ nonlocal last_interrupt_ts + nonlocal subprocess_phase # IMPORTANT: use time module explicitly; `time` can be shadowed by datetime.time via star-imports. now = time_module.time() + #region agent log + _agent_log( + "A", + "ileapp.py:_interrupt_handler", + "signal received", + { + "signum": int(signum), + "phase": subprocess_phase, + "has_proc": bool(current_proc is not None), + "proc_pid": getattr(current_proc, "pid", None), + "proc_alive": bool(current_proc.is_alive()) if current_proc is not None else None, + "since_last_s": round(now - last_interrupt_ts, 3), + }, + ) + #endregion agent log if now - last_interrupt_ts < 2.0: logfunc("Second interrupt received. Aborting run.") raise KeyboardInterrupt @@ -507,6 +543,7 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: """Run a plugin in a spawned subprocess and merge returned deltas into parent globals.""" nonlocal installed_os_version nonlocal current_proc + nonlocal subprocess_phase q = ctx_mp.SimpleQueue() payload = { "plugin_key": plugin_key, @@ -521,21 +558,78 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: "installed_os_version": installed_os_version, "seeker_all_files": seeker_all_files, } + #region agent log + _agent_log( + "A", + "ileapp.py:_run_plugin_subprocess", + "spawn child", + { + "plugin_key": plugin_key, + "ctx_start_method": getattr(ctx_mp, "get_start_method", lambda: None)(), + "global_start_method": multiprocessing.get_start_method(allow_none=True), + "files_found_len": len(files_found or []), + "file_infos_subset_len": len(file_infos_subset or {}), + "seeker_all_files_len": len(seeker_all_files or []), + }, + ) + #endregion agent log proc = ctx_mp.Process(target=run_one_plugin, args=(payload, q)) current_proc = proc proc.start() + #region agent log + _agent_log( + "B", + "ileapp.py:_run_plugin_subprocess", + "child started", + {"plugin_key": plugin_key, "pid": proc.pid}, + ) + #endregion agent log # Join in a loop so we can react to Ctrl+C and treat it as "skip current plugin" + subprocess_phase = "join_loop" while proc.is_alive(): proc.join(timeout=0.25) + subprocess_phase = "post_join" + #region agent log + _agent_log( + "B", + "ileapp.py:_run_plugin_subprocess", + "child exited", + {"plugin_key": plugin_key, "pid": proc.pid, "exitcode": proc.exitcode}, + ) + #endregion agent log result = None try: # multiprocessing.SimpleQueue.get() has no timeout; guard with empty() to avoid blocking. - if not q.empty(): + subprocess_phase = "queue_check" + empty_val = None + try: + empty_val = bool(q.empty()) + except Exception: + empty_val = None + #region agent log + _agent_log( + "A", + "ileapp.py:_run_plugin_subprocess", + "queue check", + {"plugin_key": plugin_key, "queue_empty": empty_val}, + ) + #endregion agent log + if empty_val is False: + subprocess_phase = "queue_get" result = q.get() + subprocess_phase = "post_queue_get" except Exception: result = None + #region agent log + _agent_log( + "A", + "ileapp.py:_run_plugin_subprocess", + "queue result", + {"plugin_key": plugin_key, "got_result": bool(result is not None), "result_ok": (result or {}).get("ok") if result else None}, + ) + #endregion agent log if not result or not result.get("ok"): # If we killed the process due to interrupt/skip, treat it as a skip and keep going. if proc.exitcode is not None and proc.exitcode < 0: diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py index 925639488..ecd3f5abb 100644 --- a/scripts/mp_plugin_runner.py +++ b/scripts/mp_plugin_runner.py @@ -12,6 +12,8 @@ import traceback import typing from collections import OrderedDict +import json as _agent_json +import time as _agent_time import scripts.plugin_loader as plugin_loader from scripts.context import Context @@ -81,6 +83,24 @@ def run_one_plugin(payload: dict, result_queue) -> None: - installed_os_version: str (optional) - seeker_all_files: list[str] (optional) """ + #region agent log + def _agent_log(hypothesis_id: str, location: str, message: str, data: dict): + try: + payload2 = { + "sessionId": "debug-session", + "runId": "hang1", + "hypothesisId": hypothesis_id, + "location": location, + "message": message, + "data": data, + "timestamp": int(_agent_time.time() * 1000), + } + with open("/Users/pl-2134/Development/iLEAPP/.cursor/debug.log", "a", encoding="utf-8") as f: + f.write(_agent_json.dumps(payload2, ensure_ascii=False) + "\n") + except Exception: + pass + #endregion agent log + try: plugin_key: str = payload["plugin_key"] files_found: list[str] = payload.get("files_found") or [] @@ -93,6 +113,14 @@ def run_one_plugin(payload: dict, result_queue) -> None: file_infos_subset: dict[str, tuple[str, float, float]] = payload.get("file_infos_subset") or {} installed_os_version: str = payload.get("installed_os_version") or "" seeker_all_files: list[str] = payload.get("seeker_all_files") or [] + #region agent log + _agent_log( + "C", + "mp_plugin_runner.py:run_one_plugin", + "child start", + {"plugin_key": plugin_key, "files_found_len": len(files_found or []), "seeker_all_files_len": len(seeker_all_files or [])}, + ) + #endregion agent log # Rehydrate output params + log paths for logfunc() out_params_existing = output_params_from_existing_output_folder_base( @@ -184,6 +212,14 @@ def run_one_plugin(payload: dict, result_queue) -> None: "records": len(data_list) if data_list else 0, }) + #region agent log + _agent_log( + "A", + "mp_plugin_runner.py:run_one_plugin", + "child putting result", + {"plugin_key": plugin_key, "ok": True, "record_count": len(data_list) if data_list else 0}, + ) + #endregion agent log result_queue.put({ "ok": True, "plugin_key": plugin_key, @@ -195,6 +231,14 @@ def run_one_plugin(payload: dict, result_queue) -> None: }) except Exception as ex: + #region agent log + _agent_log( + "A", + "mp_plugin_runner.py:run_one_plugin", + "child exception", + {"plugin_key": payload.get("plugin_key"), "exc_type": type(ex).__name__, "exc": str(ex)}, + ) + #endregion agent log result_queue.put({ "ok": False, "plugin_key": payload.get("plugin_key"), From 6f2a2d25b945e06e9468e8e77142e699c45fa909 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 19:43:50 +0500 Subject: [PATCH 14/22] debug: avoid mp hang by draining queue while child alive; skip before queue read --- ileapp.py | 52 +++++++++++++++++++++++++++++++------ scripts/mp_plugin_runner.py | 8 ++++++ 2 files changed, 52 insertions(+), 8 deletions(-) diff --git a/ileapp.py b/ileapp.py index 3d3e8307e..342c5db45 100755 --- a/ileapp.py +++ b/ileapp.py @@ -545,6 +545,7 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: nonlocal current_proc nonlocal subprocess_phase q = ctx_mp.SimpleQueue() + q_reader = getattr(q, "_reader", None) payload = { "plugin_key": plugin_key, "files_found": files_found, @@ -585,9 +586,34 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: ) #endregion agent log - # Join in a loop so we can react to Ctrl+C and treat it as "skip current plugin" + # Join in a loop so we can react to Ctrl+C and treat it as "skip current plugin". + # Also proactively drain the queue while the child is alive to avoid the child blocking on put(). subprocess_phase = "join_loop" + result = None while proc.is_alive(): + # Try to drain once per tick if data is available, but never block. + try: + if result is None and q_reader is not None and hasattr(q_reader, "poll") and q_reader.poll(0): + subprocess_phase = "queue_get" + result = q.get() + subprocess_phase = "join_loop" + #region agent log + _agent_log( + "A", + "ileapp.py:_run_plugin_subprocess", + "drained result while child alive", + {"plugin_key": plugin_key, "result_ok": (result or {}).get("ok") if result else None}, + ) + #endregion agent log + except Exception as ex: + #region agent log + _agent_log( + "A", + "ileapp.py:_run_plugin_subprocess", + "queue drain error while child alive", + {"plugin_key": plugin_key, "exc_type": type(ex).__name__, "exc": str(ex)}, + ) + #endregion agent log proc.join(timeout=0.25) subprocess_phase = "post_join" #region agent log @@ -599,24 +625,34 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: ) #endregion agent log - result = None + # If we killed the process due to interrupt/skip, treat it as a skip and keep going. + # IMPORTANT: do this BEFORE attempting any queue reads, because queue state can be corrupted + # if the child was terminated mid-put. + if proc.exitcode is not None and proc.exitcode < 0: + logfunc(f"Plugin {plugin_key} was interrupted (exitcode={proc.exitcode}). Skipping.") + current_proc = None + return {"ok": True, "plugin_key": plugin_key, "skipped": True} + + if result is None: + result = None try: - # multiprocessing.SimpleQueue.get() has no timeout; guard with empty() to avoid blocking. + # multiprocessing.SimpleQueue.get() has no timeout; use underlying reader.poll() to avoid blocking. subprocess_phase = "queue_check" - empty_val = None + has_data = None try: - empty_val = bool(q.empty()) + if q_reader is not None and hasattr(q_reader, "poll"): + has_data = bool(q_reader.poll(0)) except Exception: - empty_val = None + has_data = None #region agent log _agent_log( "A", "ileapp.py:_run_plugin_subprocess", "queue check", - {"plugin_key": plugin_key, "queue_empty": empty_val}, + {"plugin_key": plugin_key, "has_data": has_data, "already_have_result": bool(result is not None)}, ) #endregion agent log - if empty_val is False: + if result is None and has_data: subprocess_phase = "queue_get" result = q.get() subprocess_phase = "post_queue_get" diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py index ecd3f5abb..282e08794 100644 --- a/scripts/mp_plugin_runner.py +++ b/scripts/mp_plugin_runner.py @@ -229,6 +229,14 @@ def _agent_log(hypothesis_id: str, location: str, message: str, data: dict): "lava_only_delta": lava_only_delta, "installed_os_version": discovered_os_version, }) + #region agent log + _agent_log( + "A", + "mp_plugin_runner.py:run_one_plugin", + "child put complete", + {"plugin_key": plugin_key}, + ) + #endregion agent log except Exception as ex: #region agent log From 180e4e9cbfc92d82398999b554cfc932db42fd47 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Fri, 2 Jan 2026 19:52:37 +0500 Subject: [PATCH 15/22] chore: remove debug instrumentation --- ileapp.py | 117 ++---------------------------------- scripts/mp_plugin_runner.py | 52 ---------------- 2 files changed, 4 insertions(+), 165 deletions(-) diff --git a/ileapp.py b/ileapp.py index 342c5db45..d9c3b4fe8 100755 --- a/ileapp.py +++ b/ileapp.py @@ -13,7 +13,7 @@ import multiprocessing import signal import time as time_module -import json as _agent_json + import scripts.plugin_loader as plugin_loader import leapp_functions.app.history as history @@ -461,25 +461,7 @@ def crunch_artifacts( installed_os_version = '' current_proc = None last_interrupt_ts = 0.0 - subprocess_phase = "idle" - - #region agent log - def _agent_log(hypothesis_id: str, location: str, message: str, data: dict): - try: - payload = { - "sessionId": "debug-session", - "runId": "hang1", - "hypothesisId": hypothesis_id, - "location": location, - "message": message, - "data": data, - "timestamp": int(time_module.time() * 1000), - } - with open("/Users/pl-2134/Development/iLEAPP/.cursor/debug.log", "a", encoding="utf-8") as f: - f.write(_agent_json.dumps(payload, ensure_ascii=False) + "\n") - except Exception: - pass - #endregion agent log + # Tracks subprocess state for Ctrl+C handling; kept lightweight (no debug logging). seeker_all_files = [] try: # For iTunes backups, seeker._all_files is a dict keyed by "virtual paths" in backup @@ -507,24 +489,8 @@ def _interrupt_handler(signum, frame): - second press within 2 seconds: abort run """ nonlocal last_interrupt_ts - nonlocal subprocess_phase # IMPORTANT: use time module explicitly; `time` can be shadowed by datetime.time via star-imports. now = time_module.time() - #region agent log - _agent_log( - "A", - "ileapp.py:_interrupt_handler", - "signal received", - { - "signum": int(signum), - "phase": subprocess_phase, - "has_proc": bool(current_proc is not None), - "proc_pid": getattr(current_proc, "pid", None), - "proc_alive": bool(current_proc.is_alive()) if current_proc is not None else None, - "since_last_s": round(now - last_interrupt_ts, 3), - }, - ) - #endregion agent log if now - last_interrupt_ts < 2.0: logfunc("Second interrupt received. Aborting run.") raise KeyboardInterrupt @@ -543,7 +509,6 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: """Run a plugin in a spawned subprocess and merge returned deltas into parent globals.""" nonlocal installed_os_version nonlocal current_proc - nonlocal subprocess_phase q = ctx_mp.SimpleQueue() q_reader = getattr(q, "_reader", None) payload = { @@ -559,71 +524,21 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: "installed_os_version": installed_os_version, "seeker_all_files": seeker_all_files, } - #region agent log - _agent_log( - "A", - "ileapp.py:_run_plugin_subprocess", - "spawn child", - { - "plugin_key": plugin_key, - "ctx_start_method": getattr(ctx_mp, "get_start_method", lambda: None)(), - "global_start_method": multiprocessing.get_start_method(allow_none=True), - "files_found_len": len(files_found or []), - "file_infos_subset_len": len(file_infos_subset or {}), - "seeker_all_files_len": len(seeker_all_files or []), - }, - ) - #endregion agent log proc = ctx_mp.Process(target=run_one_plugin, args=(payload, q)) current_proc = proc proc.start() - #region agent log - _agent_log( - "B", - "ileapp.py:_run_plugin_subprocess", - "child started", - {"plugin_key": plugin_key, "pid": proc.pid}, - ) - #endregion agent log # Join in a loop so we can react to Ctrl+C and treat it as "skip current plugin". # Also proactively drain the queue while the child is alive to avoid the child blocking on put(). - subprocess_phase = "join_loop" result = None while proc.is_alive(): # Try to drain once per tick if data is available, but never block. try: if result is None and q_reader is not None and hasattr(q_reader, "poll") and q_reader.poll(0): - subprocess_phase = "queue_get" result = q.get() - subprocess_phase = "join_loop" - #region agent log - _agent_log( - "A", - "ileapp.py:_run_plugin_subprocess", - "drained result while child alive", - {"plugin_key": plugin_key, "result_ok": (result or {}).get("ok") if result else None}, - ) - #endregion agent log - except Exception as ex: - #region agent log - _agent_log( - "A", - "ileapp.py:_run_plugin_subprocess", - "queue drain error while child alive", - {"plugin_key": plugin_key, "exc_type": type(ex).__name__, "exc": str(ex)}, - ) - #endregion agent log + except Exception: + pass proc.join(timeout=0.25) - subprocess_phase = "post_join" - #region agent log - _agent_log( - "B", - "ileapp.py:_run_plugin_subprocess", - "child exited", - {"plugin_key": plugin_key, "pid": proc.pid, "exitcode": proc.exitcode}, - ) - #endregion agent log # If we killed the process due to interrupt/skip, treat it as a skip and keep going. # IMPORTANT: do this BEFORE attempting any queue reads, because queue state can be corrupted @@ -637,41 +552,17 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: result = None try: # multiprocessing.SimpleQueue.get() has no timeout; use underlying reader.poll() to avoid blocking. - subprocess_phase = "queue_check" has_data = None try: if q_reader is not None and hasattr(q_reader, "poll"): has_data = bool(q_reader.poll(0)) except Exception: has_data = None - #region agent log - _agent_log( - "A", - "ileapp.py:_run_plugin_subprocess", - "queue check", - {"plugin_key": plugin_key, "has_data": has_data, "already_have_result": bool(result is not None)}, - ) - #endregion agent log if result is None and has_data: - subprocess_phase = "queue_get" result = q.get() - subprocess_phase = "post_queue_get" except Exception: result = None - #region agent log - _agent_log( - "A", - "ileapp.py:_run_plugin_subprocess", - "queue result", - {"plugin_key": plugin_key, "got_result": bool(result is not None), "result_ok": (result or {}).get("ok") if result else None}, - ) - #endregion agent log if not result or not result.get("ok"): - # If we killed the process due to interrupt/skip, treat it as a skip and keep going. - if proc.exitcode is not None and proc.exitcode < 0: - logfunc(f"Plugin {plugin_key} was interrupted (exitcode={proc.exitcode}). Skipping.") - current_proc = None - return {"ok": True, "plugin_key": plugin_key, "skipped": True} err = (result or {}).get("error") if result else "Child process failed without result" tb = (result or {}).get("traceback") if result else "" raise RuntimeError(f"Subprocess plugin run failed for {plugin_key}: {err}\n{tb}") diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py index 282e08794..925639488 100644 --- a/scripts/mp_plugin_runner.py +++ b/scripts/mp_plugin_runner.py @@ -12,8 +12,6 @@ import traceback import typing from collections import OrderedDict -import json as _agent_json -import time as _agent_time import scripts.plugin_loader as plugin_loader from scripts.context import Context @@ -83,24 +81,6 @@ def run_one_plugin(payload: dict, result_queue) -> None: - installed_os_version: str (optional) - seeker_all_files: list[str] (optional) """ - #region agent log - def _agent_log(hypothesis_id: str, location: str, message: str, data: dict): - try: - payload2 = { - "sessionId": "debug-session", - "runId": "hang1", - "hypothesisId": hypothesis_id, - "location": location, - "message": message, - "data": data, - "timestamp": int(_agent_time.time() * 1000), - } - with open("/Users/pl-2134/Development/iLEAPP/.cursor/debug.log", "a", encoding="utf-8") as f: - f.write(_agent_json.dumps(payload2, ensure_ascii=False) + "\n") - except Exception: - pass - #endregion agent log - try: plugin_key: str = payload["plugin_key"] files_found: list[str] = payload.get("files_found") or [] @@ -113,14 +93,6 @@ def _agent_log(hypothesis_id: str, location: str, message: str, data: dict): file_infos_subset: dict[str, tuple[str, float, float]] = payload.get("file_infos_subset") or {} installed_os_version: str = payload.get("installed_os_version") or "" seeker_all_files: list[str] = payload.get("seeker_all_files") or [] - #region agent log - _agent_log( - "C", - "mp_plugin_runner.py:run_one_plugin", - "child start", - {"plugin_key": plugin_key, "files_found_len": len(files_found or []), "seeker_all_files_len": len(seeker_all_files or [])}, - ) - #endregion agent log # Rehydrate output params + log paths for logfunc() out_params_existing = output_params_from_existing_output_folder_base( @@ -212,14 +184,6 @@ def _agent_log(hypothesis_id: str, location: str, message: str, data: dict): "records": len(data_list) if data_list else 0, }) - #region agent log - _agent_log( - "A", - "mp_plugin_runner.py:run_one_plugin", - "child putting result", - {"plugin_key": plugin_key, "ok": True, "record_count": len(data_list) if data_list else 0}, - ) - #endregion agent log result_queue.put({ "ok": True, "plugin_key": plugin_key, @@ -229,24 +193,8 @@ def _agent_log(hypothesis_id: str, location: str, message: str, data: dict): "lava_only_delta": lava_only_delta, "installed_os_version": discovered_os_version, }) - #region agent log - _agent_log( - "A", - "mp_plugin_runner.py:run_one_plugin", - "child put complete", - {"plugin_key": plugin_key}, - ) - #endregion agent log except Exception as ex: - #region agent log - _agent_log( - "A", - "mp_plugin_runner.py:run_one_plugin", - "child exception", - {"plugin_key": payload.get("plugin_key"), "exc_type": type(ex).__name__, "exc": str(ex)}, - ) - #endregion agent log result_queue.put({ "ok": False, "plugin_key": payload.get("plugin_key"), From 1e7f4f5a3c5b7dde7556a64ace9cd36994eeb0e4 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Mon, 5 Jan 2026 13:56:57 +0500 Subject: [PATCH 16/22] fix: only create LAVA tables when data exists, matching single-process behavior --- scripts/mp_plugin_runner.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py index 925639488..421eccdcd 100644 --- a/scripts/mp_plugin_runner.py +++ b/scripts/mp_plugin_runner.py @@ -148,7 +148,7 @@ def run_one_plugin(payload: dict, result_queue) -> None: lava_meta_delta: dict | None = None lava_only_delta: list[dict] = [] - if wants_lava and data_headers: + if wants_lava and data_headers and data_list: # Derive table name + maps consistently with lava_create_sqlite_table() func_name = plugin_spec.name table_name, column_map, object_columns = lavafuncs.lava_create_sqlite_table(func_name, data_headers) From 9a4466d5ba6f4616c205889ce5180778244f90a0 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Mon, 5 Jan 2026 14:17:13 +0500 Subject: [PATCH 17/22] fix: suppress pkg_resources deprecation warnings in child processes --- scripts/mp_plugin_runner.py | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py index 421eccdcd..f828ac6e3 100644 --- a/scripts/mp_plugin_runner.py +++ b/scripts/mp_plugin_runner.py @@ -11,6 +11,7 @@ import os import traceback import typing +import warnings from collections import OrderedDict import scripts.plugin_loader as plugin_loader @@ -21,6 +22,7 @@ check_output_types, iOS, output_params_from_existing_output_folder_base, + logfunc, ) import scripts.lavafuncs as lavafuncs @@ -81,6 +83,10 @@ def run_one_plugin(payload: dict, result_queue) -> None: - installed_os_version: str (optional) - seeker_all_files: list[str] (optional) """ + # Suppress common deprecation warnings that appear in every subprocess + warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources") + warnings.filterwarnings("ignore", message=".*pkg_resources is deprecated.*") + try: plugin_key: str = payload["plugin_key"] files_found: list[str] = payload.get("files_found") or [] @@ -125,9 +131,18 @@ def run_one_plugin(payload: dict, result_queue) -> None: lavafuncs.lava_open_existing(output_folder_base) # Execute plugin - data_headers, data_list, source_path = plugin_spec.method( - files_found, category_folder, seeker_proxy, wrap_text, time_offset - ) + try: + data_headers, data_list, source_path = plugin_spec.method( + files_found, category_folder, seeker_proxy, wrap_text, time_offset + ) + except TypeError as ex: + logfunc(f"TypeError: {ex}") + logfunc(f"Traceback: {traceback.format_exc()}") + raise ex + except Exception as ex: + logfunc(f"Exception: {ex}") + logfunc(f"Traceback: {traceback.format_exc()}") + raise ex # Capture installed OS version if this plugin discovered it discovered_os_version = "" From 2756453729a7c0bbe7c5e26e0ec4cc2702a44ae0 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Mon, 5 Jan 2026 14:31:10 +0500 Subject: [PATCH 18/22] refactor: remove redundant exception handling in plugin execution --- scripts/mp_plugin_runner.py | 17 ++++------------- 1 file changed, 4 insertions(+), 13 deletions(-) diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py index f828ac6e3..5a06e0b65 100644 --- a/scripts/mp_plugin_runner.py +++ b/scripts/mp_plugin_runner.py @@ -83,7 +83,7 @@ def run_one_plugin(payload: dict, result_queue) -> None: - installed_os_version: str (optional) - seeker_all_files: list[str] (optional) """ - # Suppress common deprecation warnings that appear in every subprocess + # Suppress common deprecation warnings that appear in every subprocess, we need to do this because the plugin loader is using pkg_resources. warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources") warnings.filterwarnings("ignore", message=".*pkg_resources is deprecated.*") @@ -131,18 +131,9 @@ def run_one_plugin(payload: dict, result_queue) -> None: lavafuncs.lava_open_existing(output_folder_base) # Execute plugin - try: - data_headers, data_list, source_path = plugin_spec.method( - files_found, category_folder, seeker_proxy, wrap_text, time_offset - ) - except TypeError as ex: - logfunc(f"TypeError: {ex}") - logfunc(f"Traceback: {traceback.format_exc()}") - raise ex - except Exception as ex: - logfunc(f"Exception: {ex}") - logfunc(f"Traceback: {traceback.format_exc()}") - raise ex + data_headers, data_list, source_path = plugin_spec.method( + files_found, category_folder, seeker_proxy, wrap_text, time_offset + ) # Capture installed OS version if this plugin discovered it discovered_os_version = "" From 0d834f70a7832523d083a23cc92fed4c2fe6cf08 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Thu, 8 Jan 2026 19:49:42 +0500 Subject: [PATCH 19/22] feat: add SIGUSR1 and SIGUSR2 support for skip functionality - Add dedicated _skip_handler for SIGUSR1/SIGUSR2 signals - These signals always skip current plugin without abort risk - Can be sent externally via: kill -USR1 or kill -USR2 - Only registered on Unix-like systems (where available) --- ileapp.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/ileapp.py b/ileapp.py index d9c3b4fe8..68850582d 100755 --- a/ileapp.py +++ b/ileapp.py @@ -497,6 +497,15 @@ def _interrupt_handler(signum, frame): last_interrupt_ts = now _terminate_current_plugin_proc("SIGINT/SIGBREAK") + def _skip_handler(signum, frame): + """ + SIGUSR1/SIGUSR2 handling for mp mode: + - Always skip current plugin (no abort risk) + - Can be sent externally via: kill -USR1 or kill -USR2 + """ + signal_name = "SIGUSR1" if signum == signal.SIGUSR1 else "SIGUSR2" + _terminate_current_plugin_proc(signal_name) + if mp_per_plugin: # Register interrupt handler (cross-platform): # - SIGINT is Ctrl+C everywhere @@ -504,6 +513,13 @@ def _interrupt_handler(signum, frame): signal.signal(signal.SIGINT, _interrupt_handler) if hasattr(signal, "SIGBREAK"): signal.signal(signal.SIGBREAK, _interrupt_handler) + # Register skip handlers (Unix-like systems only): + # - SIGUSR1: dedicated skip signal (no abort risk) + # - SIGUSR2: dedicated skip signal (no abort risk) + if hasattr(signal, "SIGUSR1"): + signal.signal(signal.SIGUSR1, _skip_handler) + if hasattr(signal, "SIGUSR2"): + signal.signal(signal.SIGUSR2, _skip_handler) def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: str, file_infos_subset: dict | None = None): """Run a plugin in a spawned subprocess and merge returned deltas into parent globals.""" From ad7b95eda0f91c38ff8e9ba96136371d17e9c31e Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Tue, 13 Jan 2026 14:26:40 +0500 Subject: [PATCH 20/22] Handle PyInstaller multiprocessing args in CLI to keep frozen binary working --- ileapp.py | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/ileapp.py b/ileapp.py index 68850582d..222928ce7 100755 --- a/ileapp.py +++ b/ileapp.py @@ -217,7 +217,23 @@ def main(): parser.print_help(sys.stderr) sys.exit() - args = parser.parse_args() + # Filter out multiprocessing spawn arguments that argparse doesn't recognize + # When using multiprocessing with 'spawn', Python may inject internal flags that + # our CLI parser doesn't know about (and would otherwise treat as fatal errors). + filtered_argv = [] + for arg in sys.argv[1:]: + # Skip multiprocessing / interpreter-injected arguments + if ( + arg.startswith('--multiprocessing-') + or arg.startswith('tracker_fd=') + or arg.startswith('pipe_handle=') + or arg in ['-B', '-S', '-I'] # Python optimization / startup flags + ): + continue + filtered_argv.append(arg) + + # Use filtered arguments for parsing so frozen mp children don't break argparse + args = parser.parse_args(filtered_argv) available_plugins = [] loader_paths = [plugin_loader.PLUGINPATH] @@ -792,4 +808,6 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: return True if __name__ == '__main__': + # Support for multiprocessing with PyInstaller frozen binaries + multiprocessing.freeze_support() main() From 00e2bc13aa1118dbab54922ec16307f5a90e56d9 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Wed, 22 Jul 2026 12:42:17 +0500 Subject: [PATCH 21/22] fix: close LAVA database connection when using per-plugin multiprocessing --- ileapp.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/ileapp.py b/ileapp.py index 222928ce7..e029fd9b8 100755 --- a/ileapp.py +++ b/ileapp.py @@ -380,9 +380,6 @@ def main(): logfunc("EXPERIMENTAL MODE ENABLED: --mp_per_plugin (per-plugin subprocess execution)") initialize_lava(input_path, out_params.output_folder_base, extracttype) - if args.mp_per_plugin: - # Parent does not need an open connection while children write to the DB. - lava_close_db() # Record history if enabled history.record_input_path(input_path) @@ -474,6 +471,11 @@ def crunch_artifacts( log.write(f'Timezone selected: {time_offset}

') ctx_mp = multiprocessing.get_context("spawn") if mp_per_plugin else None + if mp_per_plugin: + # Parent does not need an open connection while children write to the DB. + # Closed here (not right after initialize_lava) so any parent-side LAVA + # writes during setup above still have a valid connection. + lava_close_db() installed_os_version = '' current_proc = None last_interrupt_ts = 0.0 From b1727ed9f3054f3adb7d11333059891eaf764872 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Thu, 30 Jul 2026 14:51:26 +0500 Subject: [PATCH 22/22] fix: close LAVA database conneciton when using per-plugin multiprocessing. --- ileapp.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/ileapp.py b/ileapp.py index e029fd9b8..4e4442ac2 100755 --- a/ileapp.py +++ b/ileapp.py @@ -471,11 +471,6 @@ def crunch_artifacts( log.write(f'Timezone selected: {time_offset}

') ctx_mp = multiprocessing.get_context("spawn") if mp_per_plugin else None - if mp_per_plugin: - # Parent does not need an open connection while children write to the DB. - # Closed here (not right after initialize_lava) so any parent-side LAVA - # writes during setup above still have a valid connection. - lava_close_db() installed_os_version = '' current_proc = None last_interrupt_ts = 0.0 @@ -558,6 +553,11 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: "installed_os_version": installed_os_version, "seeker_all_files": seeker_all_files, } + # Parent doesn't hold a write lock while the child writes to the same DB file + # (lavafuncs uses plain sqlite3.connect with no WAL/busy_timeout, so a lingering + # parent connection risks "database is locked"). Reopened right after the child + # finishes, since the loop needs it again for the next plugin's bookkeeping. + lava_close_db() proc = ctx_mp.Process(target=run_one_plugin, args=(payload, q)) current_proc = proc proc.start() @@ -574,6 +574,10 @@ def _run_plugin_subprocess(plugin_key: str, files_found: list, category_folder: pass proc.join(timeout=0.25) + # Child is done writing; safe to reopen the parent's connection for the + # rest of this iteration's bookkeeping and the next plugin's. + lava_open_existing(out_params.output_folder_base) + # If we killed the process due to interrupt/skip, treat it as a skip and keep going. # IMPORTANT: do this BEFORE attempting any queue reads, because queue state can be corrupted # if the child was terminated mid-put.