From f3bb98d9e08f73996842379f6fa9dd0b8e2c7ff4 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Mon, 27 Jul 2026 12:55:29 +0500 Subject: [PATCH 1/3] feat: run each plugin in its own subprocess with skip/abort on Ctrl+C Adds --mp_per_plugin/--mp flag. Each plugin runs in a spawned subprocess; first Ctrl+C terminates just that plugin and continues, a second within 5s aborts the run. SIGUSR1/SIGUSR2 act as alternate skip triggers. Parent merges icons/LAVA-artifact deltas back after each subprocess completes. Rebased onto upstream's _HTML/_Script_Logs output layout and lava_db_name-based db path. Co-Authored-By: Claude Opus 5 --- aleapp.py | 162 ++++++++++++++++++++++++++++++++++-- scripts/ilapfuncs.py | 15 +++- scripts/lavafuncs.py | 23 +++++ scripts/mp_plugin_runner.py | 157 ++++++++++++++++++++++++++++++++++ 4 files changed, 347 insertions(+), 10 deletions(-) create mode 100644 scripts/mp_plugin_runner.py diff --git a/aleapp.py b/aleapp.py index 5c0092868..5dd7b9836 100755 --- a/aleapp.py +++ b/aleapp.py @@ -19,6 +19,13 @@ from scripts.lavafuncs import * # pylint: disable=wildcard-import,unused-wildcard-import from scripts.context import Context +import multiprocessing +import queue as _queue +import signal + +import scripts.mp_plugin_runner as mp_plugin_runner +import scripts.lavafuncs as _lavafuncs + def validate_args(args): if args.artifact_paths or args.create_profile_casedata: return # Skip further validation if --artifact_paths is used @@ -165,6 +172,13 @@ def main(): "This argument is meant to be used alone, without any other arguments.")) parser.add_argument('--custom_output_folder', required=False, action="store", help="Custom name for the output folder") parser.add_argument('--custom_artifacts_path', required=False, action="store", help="Additional path to load artifacts from (e.g., scripts/alternate_artifacts)") + parser.add_argument( + '--mp_per_plugin', '--mp', + action='store_true', + default=False, + dest='mp_per_plugin', + help='Run each plugin in a separate subprocess (enables skip with Ctrl+C).', + ) profile_filename = None casedata = {} @@ -316,13 +330,13 @@ def main(): 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, profile_filename) + crunch_artifacts(selected_plugins, extracttype, input_path, out_params, wrap_text, loader, casedata, profile_filename, 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, profile_filename): + loader: plugin_loader.PluginLoader, casedata, profile_filename, mp_per_plugin: bool = False): start = process_time() start_wall = perf_counter() @@ -357,6 +371,109 @@ def crunch_artifacts( temp_file.close() return False + # --- subprocess mode setup --- + ctx_mp = None + _current_proc = [None] # list for mutability in closure + _last_sigint_ts = [0.0] # timestamp of last handled SIGINT + _abort_requested = [False] + old_sigint = None + + if mp_per_plugin: + import time as _time_module + ctx_mp = multiprocessing.get_context('spawn') + + def _terminate_current(reason: str): + proc = _current_proc[0] + if proc is not None and proc.is_alive(): + logfunc(f'Skip requested ({reason}). Terminating plugin subprocess...') + try: + proc.terminate() + except Exception: + pass + + def _handle_sigint(signum, frame): + now = _time_module.time() + elapsed = now - _last_sigint_ts[0] + # Ignore signals arriving within 500ms of the last one. + # sudo sends SIGINT to the entire process group, so the parent can + # receive it twice within microseconds from a single Ctrl+C press. + if elapsed < 0.5: + return + if _last_sigint_ts[0] > 0.0 and elapsed < 5.0: + # Second real Ctrl+C within 5 seconds → abort + _abort_requested[0] = True + print('\nAborting run.', flush=True) + else: + print('\nSkipping current plugin (Ctrl+C again within 5s to abort).', flush=True) + _last_sigint_ts[0] = now + _terminate_current('SIGINT') + + def _handle_sigusr(signum, frame): + _terminate_current('SIGUSR') + + old_sigint = signal.signal(signal.SIGINT, _handle_sigint) + if hasattr(signal, 'SIGUSR1'): + signal.signal(signal.SIGUSR1, _handle_sigusr) + if hasattr(signal, 'SIGUSR2'): + signal.signal(signal.SIGUSR2, _handle_sigusr) + + def _run_plugin_subprocess(plugin, files_found, category_folder): + """Spawn one subprocess for plugin; returns result dict or None if skipped.""" + file_infos_subset = { + path: (info.source_path, info.creation_date, info.modification_date) + for path, info in seeker.file_infos.items() + } + payload = { + 'plugin_key': plugin.name, + 'files_found': files_found, + 'category_folder': category_folder, + 'wrap_text': wrap_text, + 'output_folder_base': out_params.report_folder_base, + 'input_path': input_path, + 'extracttype': extracttype, + 'file_infos_subset': file_infos_subset, + 'seeker_all_files': list(seeker.file_infos.keys()), + } + + result_q = ctx_mp.Queue() + proc = ctx_mp.Process( + target=mp_plugin_runner.run_one_plugin, + args=(payload, result_q), + ) + _current_proc[0] = proc + proc.start() + + # Poll until the process finishes. The signal handler calls proc.terminate() + # directly, so we just need to detect when it exits. + while proc.is_alive(): + try: + result = result_q.get(timeout=0.25) + proc.join() + _current_proc[0] = None + return result + except _queue.Empty: + pass + + _current_proc[0] = None + + # Negative exit code means the process was killed by a signal (skip). + if proc.exitcode is not None and proc.exitcode < 0: + logfunc(f' {plugin.name} subprocess terminated (exitcode={proc.exitcode}).') + return None + + # Process exited cleanly — drain the queue one last time. + try: + result = result_q.get_nowait() + return result + except _queue.Empty: + return { + 'ok': False, + 'plugin_key': plugin.name, + 'error': 'Subprocess exited without putting a result on the queue.', + 'traceback': '', + } + # --- end subprocess mode setup --- + # Now ready to run logfunc(f'Info: {len(loader) - 1} modules loaded.') # excluding usagestatsVersion if profile_filename: @@ -376,6 +493,8 @@ def crunch_artifacts( # Search for the files per the arguments for plugin_number, plugin in enumerate(plugins, start=1): logfunc() + if mp_per_plugin and _abort_requested[0]: + break logfunc('[{}/{}] {} [{}] artifact started'.format(plugin_number, len(plugins), plugin.name, plugin.module_name)) if isinstance(plugin.search, list) or isinstance(plugin.search, tuple): @@ -417,13 +536,27 @@ def crunch_artifacts( logfunc('Error creating {} report directory at path {}'.format(plugin.name, category_folder)) logfunc('Error was {}'.format(str(ex))) continue # cannot do work - try: - plugin.method(files_found, category_folder, seeker, wrap_text) - except Exception as ex: # pylint: disable=broad-exception-caught - logfunc('Reading {} artifact had errors!'.format(plugin.name)) - logfunc('Error was {}'.format(str(ex))) - logfunc('Exception Traceback: {}'.format(traceback.format_exc())) - continue # nope + if mp_per_plugin: + result = _run_plugin_subprocess(plugin, files_found, category_folder) + if result is None: + logfunc(f'{plugin.name} [{plugin.module_name}] skipped by user.') + elif result.get('ok'): + for cat, arts in result.get('icons_delta', {}).items(): + icons.setdefault(cat, {}).update(arts) + for cat, arts in result.get('lava_artifacts_delta', {}).items(): + _lavafuncs.lava_data['artifacts'].setdefault(cat, []).extend(arts) + else: + logfunc('Error in {} [{}]: {}'.format(plugin.name, plugin.module_name, result.get('error'))) + if result.get('traceback'): + logfunc(result['traceback']) + else: + try: + plugin.method(files_found, category_folder, seeker, wrap_text) + except Exception as ex: # pylint: disable=broad-exception-caught + logfunc('Reading {} artifact had errors!'.format(plugin.name)) + logfunc('Error was {}'.format(str(ex))) + logfunc('Exception Traceback: {}'.format(traceback.format_exc())) + continue # nope else: logfunc("No file found") logfunc('{} [{}] artifact completed'.format(plugin.name, plugin.module_name)) @@ -432,6 +565,16 @@ def crunch_artifacts( log.flush() log.close() + if mp_per_plugin: + if _abort_requested[0]: + logfunc('Processing aborted by user after interrupt.') + if old_sigint is not None: + signal.signal(signal.SIGINT, old_sigint) + if hasattr(signal, 'SIGUSR1'): + signal.signal(signal.SIGUSR1, signal.SIG_DFL) + if hasattr(signal, 'SIGUSR2'): + signal.signal(signal.SIGUSR2, signal.SIG_DFL) + write_device_info() logfunc('') logfunc('Processes completed.') @@ -466,5 +609,6 @@ def crunch_artifacts( return True if __name__ == '__main__': + multiprocessing.freeze_support() main() diff --git a/scripts/ilapfuncs.py b/scripts/ilapfuncs.py index a7596a8d0..6af6f0bbd 100755 --- a/scripts/ilapfuncs.py +++ b/scripts/ilapfuncs.py @@ -76,7 +76,20 @@ def __init__(self, output_folder, custom_folder_name=None): os.makedirs(self.data_folder) os.makedirs(self.media_folder, exist_ok=True) os.makedirs(self.html_media_folder, exist_ok=True) - + +def output_params_from_existing_output_folder_base(output_folder_base: str) -> None: + """Set OutputParameters class-level paths from an already-created report folder. + + Used in subprocesses so logfunc() writes to the correct log file. + The folder must already exist — this function never creates directories. + """ + OutputParameters.screen_output_file_path = os.path.join( + output_folder_base, '_HTML', '_Script_Logs', 'Screen_Output.html' + ) + OutputParameters.screen_output_file_path_devinfo = os.path.join( + output_folder_base, '_HTML', '_Script_Logs', 'DeviceInfo.html' + ) + class GuiWindow: '''This only exists to hold window handle if script is run from GUI''' window_handle = None # static variable diff --git a/scripts/lavafuncs.py b/scripts/lavafuncs.py index cc4921a21..3ea849844 100644 --- a/scripts/lavafuncs.py +++ b/scripts/lavafuncs.py @@ -666,3 +666,26 @@ def lava_finalize_output(output_path): # Close the SQLite database lava_db.close() + +def lava_open_existing(output_path: str) -> None: + """Connect to the existing lava db created by the parent process. + + Called in a subprocess — never creates tables, only opens a connection. + """ + global lava_data, lava_db + lava_data = { + "artifacts": OrderedDict(), + "modules": [], + } + db_path = os.path.join(output_path, lava_db_name) + lava_db = sqlite3.connect(db_path) + +def lava_close_db() -> None: + """Close the SQLite connection. Safe to call even if already closed.""" + global lava_db + if lava_db is not None: + try: + lava_db.close() + except Exception: + pass + lava_db = None diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py new file mode 100644 index 000000000..63513b633 --- /dev/null +++ b/scripts/mp_plugin_runner.py @@ -0,0 +1,157 @@ +""" +Multiprocessing helpers for ALEAPP. + +Importable in a spawned subprocess — no side effects at import time. +Rehydrates minimal per-plugin state and returns picklable deltas +(icons + LAVA artifact metadata) back to the parent. +""" +from __future__ import annotations + +import os +import traceback +import warnings + +import scripts.lavafuncs as lavafuncs +import scripts.plugin_loader as plugin_loader +from scripts.ilapfuncs import ( + check_output_types, + icons, + output_params_from_existing_output_folder_base, +) +from scripts.search_files import FileInfo, FileSeekerBase + + +class SeekerProxy(FileSeekerBase): + """Minimal seeker for subprocess use. + + The parent resolves files_found before spawning — the child never + needs to search. This proxy exposes file_infos so media helpers + (check_in_media) can look up FileInfo by extracted path. + + ``data_folder`` mirrors the real FileSeekerDir/Tar/Zip attribute that + plugins use to strip the extraction-root prefix from copied file paths + (e.g. ``file_found.replace(seeker.data_folder, '')``). It equals the + ``data/`` sub-directory under the report folder base — the same value + the parent sets when constructing the real seeker. + """ + + def __init__( + self, + file_infos_subset: dict[str, tuple[str, float, float]], + all_files: list[str], + data_folder: str = "", + ): + self.file_infos: dict[str, FileInfo] = { + path: FileInfo(src, ctime, mtime) + for path, (src, ctime, mtime) in file_infos_subset.items() + } + self._all_files = all_files + self.data_folder: str = data_folder + + def search(self, filepattern, return_on_first_hit=False): + return [] + + def cleanup(self): + pass + + +def run_one_plugin(payload: dict, result_queue) -> None: + """Run a single plugin inside a subprocess. + + Expected payload keys: + plugin_key str + files_found list[str] + category_folder str + wrap_text bool + output_folder_base str + input_path str + extracttype str + file_infos_subset dict[str, tuple[str, float, float]] + seeker_all_files list[str] + """ + # Ignore SIGINT in the child — Ctrl+C sends SIGINT to the whole process group, + # which would otherwise kill the child AND trigger the parent's handler twice. + # The parent manages child termination explicitly via proc.terminate(). + import signal as _signal + _signal.signal(_signal.SIGINT, _signal.SIG_IGN) + + # Scope the suppression to pkg_resources only — process-global suppression + # would mask real deprecation warnings from other modules. + warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources") + + 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)) + 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 = payload.get("file_infos_subset") or {} + seeker_all_files: list[str] = payload.get("seeker_all_files") or [] + + # Let logfunc() write to the correct file + output_params_from_existing_output_folder_base(output_folder_base) + + # Reload PluginLoader — avoids pickling LazyLoader callables + loader = plugin_loader.PluginLoader() + plugin_spec = loader[plugin_key] + + seeker = SeekerProxy( + file_infos_subset, + seeker_all_files, + data_folder=os.path.join(output_folder_base, "data"), + ) + + artifact_info = plugin_spec.artifact_info or {} + output_types = artifact_info.get( + "output_types", ["html", "tsv", "timeline", "lava", "kml"] + ) + wants_lava = check_output_types("lava", output_types) + + if wants_lava: + lavafuncs.lava_open_existing(output_folder_base) + + # ALEAPP plugin signature — no time_offset (Android, not iOS) + plugin_spec.method(files_found, category_folder, seeker, wrap_text) + + # icons is the module-level dict in ilapfuncs; artifact_processor mutates it. + # Spawn-context guarantee: each subprocess starts a fresh Python interpreter, + # so ilapfuncs.icons begins as {} — it cannot contain entries from prior + # plugins run in other subprocesses. dict(icons) therefore captures only + # the icons registered by this plugin. + icons_delta: dict[str, dict[str, str]] = dict(icons) + + # Collect LAVA artifact entries written by this plugin. + # Picklability: each artifact dict contains only plain Python primitives. + # column_map → {str: str} (sanitized_name → original_name) + # object_columns → list of {"name": str, "type": str} dicts + # data_views → {str: str/bool} after sanitize_sql_name processing + # All other fields (name, tablename, module, record_count) are str/int. + # No tuple keys, no non-primitive values — safe to pass through a Queue. + lava_artifacts_delta: dict = {} + if wants_lava and lavafuncs.lava_data: + lava_artifacts_delta = { + cat: list(arts) + for cat, arts in lavafuncs.lava_data.get("artifacts", {}).items() + } + + result_queue.put({ + "ok": True, + "plugin_key": plugin_key, + "icons_delta": icons_delta, + "lava_artifacts_delta": lava_artifacts_delta, + }) + + except Exception as ex: + result_queue.put({ + "ok": False, + "plugin_key": payload.get("plugin_key"), + "error": str(ex), + "traceback": traceback.format_exc(), + }) + finally: + try: + lavafuncs.lava_close_db() + except Exception: + pass From 6e1c0a45bf2564fcf70b7c2f059749fc2a701e1d Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Mon, 27 Jul 2026 13:06:40 +0500 Subject: [PATCH 2/3] fix: use OutputParameters.output_folder_base in mp subprocess payload Upstream renamed report_folder_base -> output_folder_base; the mp_per_plugin subprocess payload still referenced the old attribute, crashing on the first plugin. Co-Authored-By: Claude Opus 5 --- aleapp.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aleapp.py b/aleapp.py index 5dd7b9836..423f4175c 100755 --- a/aleapp.py +++ b/aleapp.py @@ -428,7 +428,7 @@ def _run_plugin_subprocess(plugin, files_found, category_folder): 'files_found': files_found, 'category_folder': category_folder, 'wrap_text': wrap_text, - 'output_folder_base': out_params.report_folder_base, + 'output_folder_base': out_params.output_folder_base, 'input_path': input_path, 'extracttype': extracttype, 'file_infos_subset': file_infos_subset, From 105358a0fa2873205dcc023d1b3f5cde012ab572 Mon Sep 17 00:00:00 2001 From: Abu-Huraira21 Date: Mon, 27 Jul 2026 13:19:36 +0500 Subject: [PATCH 3/3] fix: propagate Context and LAVA meta.modules into mp subprocesses MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two mp_per_plugin-only bugs found via log diff between an --mp run and a plain run: - lava_open_existing() built lava_data without the 'meta' key, so lava_process_artifact()'s lava_data['meta']['modules'] access raised KeyError: 'meta' for every plugin. - Context.set_output_params() only ran once in the parent at startup; a spawned subprocess is a fresh interpreter, so any plugin calling Context.get_output_params() (mister_skinnylegs plugins, media linking helpers) raised "Context not set. OutputParameters not available." Now set per-subprocess from the known output folder layout. Also wire meta_modules_delta through the result queue and merge it in the parent, mirroring the existing icons/lava_artifacts delta pattern — otherwise the final _lava_data.lava's meta.modules would end up empty under --mp, since every plugin runs in an isolated subprocess. Co-Authored-By: Claude Opus 5 --- aleapp.py | 7 +++++++ scripts/lavafuncs.py | 3 +++ scripts/mp_plugin_runner.py | 21 +++++++++++++++++++++ 3 files changed, 31 insertions(+) diff --git a/aleapp.py b/aleapp.py index 423f4175c..c763a1069 100755 --- a/aleapp.py +++ b/aleapp.py @@ -545,6 +545,13 @@ def _run_plugin_subprocess(plugin, files_found, category_folder): icons.setdefault(cat, {}).update(arts) for cat, arts in result.get('lava_artifacts_delta', {}).items(): _lavafuncs.lava_data['artifacts'].setdefault(cat, []).extend(arts) + for module_info in result.get('meta_modules_delta', []): + existing = next((m for m in _lavafuncs.lava_data['meta']['modules'] + if m['module_name'] == module_info['module_name']), None) + if existing: + existing['artifacts'].extend(module_info['artifacts']) + else: + _lavafuncs.lava_data['meta']['modules'].append(module_info) else: logfunc('Error in {} [{}]: {}'.format(plugin.name, plugin.module_name, result.get('error'))) if result.get('traceback'): diff --git a/scripts/lavafuncs.py b/scripts/lavafuncs.py index 3ea849844..158771d55 100644 --- a/scripts/lavafuncs.py +++ b/scripts/lavafuncs.py @@ -676,6 +676,9 @@ def lava_open_existing(output_path: str) -> None: lava_data = { "artifacts": OrderedDict(), "modules": [], + "meta": { + "modules": [] + } } db_path = os.path.join(output_path, lava_db_name) lava_db = sqlite3.connect(db_path) diff --git a/scripts/mp_plugin_runner.py b/scripts/mp_plugin_runner.py index 63513b633..7698039bd 100644 --- a/scripts/mp_plugin_runner.py +++ b/scripts/mp_plugin_runner.py @@ -10,9 +10,11 @@ import os import traceback import warnings +from types import SimpleNamespace import scripts.lavafuncs as lavafuncs import scripts.plugin_loader as plugin_loader +from scripts.context import Context from scripts.ilapfuncs import ( check_output_types, icons, @@ -93,6 +95,19 @@ def run_one_plugin(payload: dict, result_queue) -> None: # Let logfunc() write to the correct file output_params_from_existing_output_folder_base(output_folder_base) + # Context.set_output_params() only ran in the parent process at startup — + # a spawned subprocess is a fresh interpreter, so Context._output_params + # is None here unless we set it too. Plugins that call + # Context.get_output_params() (e.g. mister_skinnylegs plugins, media + # linking helpers) would otherwise raise "Context not set". + output_params = SimpleNamespace( + output_folder_base=output_folder_base, + 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"), + ) + Context.set_output_params(output_params) + # Reload PluginLoader — avoids pickling LazyLoader callables loader = plugin_loader.PluginLoader() plugin_spec = loader[plugin_key] @@ -130,17 +145,23 @@ def run_one_plugin(payload: dict, result_queue) -> None: # All other fields (name, tablename, module, record_count) are str/int. # No tuple keys, no non-primitive values — safe to pass through a Queue. lava_artifacts_delta: dict = {} + meta_modules_delta: list = [] if wants_lava and lavafuncs.lava_data: lava_artifacts_delta = { cat: list(arts) for cat, arts in lavafuncs.lava_data.get("artifacts", {}).items() } + # Per-module artifact metadata, written into the final _lava_data.lava. + # Each subprocess starts with an empty lava_data['meta']['modules'], + # so this list holds only the module entry(ies) this plugin added. + meta_modules_delta = list(lavafuncs.lava_data.get("meta", {}).get("modules", [])) result_queue.put({ "ok": True, "plugin_key": plugin_key, "icons_delta": icons_delta, "lava_artifacts_delta": lava_artifacts_delta, + "meta_modules_delta": meta_modules_delta, }) except Exception as ex: