if args.artifact_paths or args.create_profile_casedata:

return # Skip further validation if --artifact_paths is used

# Ensure other arguments are provided

mandatory_args = ['input_path', 'output_path', 't']

for arg in mandatory_args:

value = getattr(args, arg)

if value is None:

raise argparse.ArgumentError(None, f'No {arg.upper()} provided. Run the program again.')

# Check existence of paths

if not os.path.exists(args.input_path):

raise argparse.ArgumentError(None, f'INPUT path \'{args.input_path}\' does not exist! Run the program again.')

# Several images are read together only as a raw image set (the members of a RAID set)

more_paths = getattr(args, 'input_paths', [args.input_path])[1:]

if more_paths and args.t != 'raw':

raise argparse.ArgumentError(None, 'Several INPUT paths are read together only with -t raw, as '

'the member images of a RAID set. Run the program again.')

for more_path in more_paths:

if not os.path.exists(more_path):

raise argparse.ArgumentError(None, f'INPUT path \'{more_path}\' does not exist! Run the program again.')

logical = logical_among(getattr(args, 'input_paths', [args.input_path]))

if logical is not None:

raise argparse.ArgumentError(None, logical_refusal(logical))

if not os.path.exists(args.output_path):

raise argparse.ArgumentError(None, f'OUTPUT path \'{args.output_path}\' does not exist! Run the program again.')

if not os.path.isdir(os.path.abspath(args.output_path)):

raise argparse.ArgumentError(None, f'OUTPUT path \'{args.output_path}\' must be a directory! Run the program again.')

# Validate new folder name for output path

output_folder_valid, output_folder_error = validate_output_folder_available(

os.path.abspath(args.output_path), args.custom_output_folder)

if not output_folder_valid:

raise argparse.ArgumentError(None, output_folder_error)

# Validate input_path based on type

abs_input_path = os.path.abspath(args.input_path)

if args.t == 'fs': # Filesystem input type

# Check if input path is a directory

if not os.path.isdir(abs_input_path):

raise argparse.ArgumentError(None, f'INPUT path \'{args.input_path}\' is not a directory. Type "fs" requires '

f'a directory input. Run the program again.')

# Check if directory is empty

if not os.listdir(abs_input_path):

raise argparse.ArgumentError(None, f'Input directory \'{args.input_path}\' is empty. Run the program again.')

elif args.t == 'file': # Single file input type

if not os.path.isfile(abs_input_path):

raise argparse.ArgumentError(None, f'INPUT path \'{args.input_path}\' is not a file. Type "file" requires a '

f'single file input. Run the program again.')

if args.load_case_data and not os.path.exists(args.load_case_data):

raise argparse.ArgumentError(None, 'LEAPP Case Data file not found! Run the program again.')

if args.load_profile and not os.path.exists(args.load_profile):

raise argparse.ArgumentError(None, 'iLEAPP Profile file not found! Run the program again.')

if args.keychain and not os.path.isfile(args.keychain):

raise argparse.ArgumentError(None, 'Keychain file not found! Run the program again.')

try:

pytz.timezone(args.timezone)

except pytz.UnknownTimeZoneError as ex:

raise argparse.ArgumentError(

available_modules = [(module_data.category, module_data.name) for module_data in plugins]

available_modules.sort()

modules_in_profile = {}

user_choice = ''

print('--- iLEAPP Profile file creation ---\n')

instructions = 'You can type:\n'

instructions += ' - \'a\' to add or remove modules in the profile file\n'

instructions += ' - \'l\' to display the list of all available modules with their number\n'

instructions += ' - \'p\' to display the modules added into the profile file\n'

instructions += ' - \'q\' to quit and save\n'

while not user_choice:

print(instructions)

user_choice = input('Please enter your choice: ').lower()

print()

if user_choice == "l":

print('Available modules:')

for number, available_module in enumerate(available_modules):

print(number + 1, available_module)

print()

user_choice = ''

elif user_choice == "p":

if modules_in_profile:

for number, module in modules_in_profile.items():

print(number, module)

print()

else:

print('No module added to the profile file\n')

user_choice = ''

elif user_choice == 'a':

modules_numbers = input('Enter the numbers of modules, seperated by a comma, to add or remove in the profile file: ')

modules_numbers = modules_numbers.split(',')

modules_numbers = [module_number.strip() for module_number in modules_numbers]

for module_number in modules_numbers:

if module_number.isdigit():

module_number = int(module_number)

if module_number > 0 and module_number <= len(available_modules):

if module_number not in modules_in_profile:

module_to_add = available_modules[module_number - 1]

modules_in_profile[module_number] = module_to_add

print(f'module number {module_number} {module_to_add} was added')

else:

module_to_remove = modules_in_profile[module_number]

print(f'module number {module_number} {module_to_remove} was removed')

del modules_in_profile[module_number]

else:

print('Please enter the number of a module!!!\n')

print()

user_choice = ''

elif user_choice == "q":

if modules_in_profile:

modules = [module_info[1] for module_info in modules_in_profile.values()]

profile_filename = ''

while not profile_filename:

profile_filename = input('Enter the name of the profile: ')

profile_filename += '.ilprofile'

filename = os.path.join(path, profile_filename)

with open(filename, "wt", encoding="utf-8") as profile_file:

json.dump({"leapp": "ileapp", "format_version": 1, "plugins": modules}, profile_file)

print('\nProfile saved:', filename)

print()

else:

print('No module added. The profile file was not created.\n')

print()

return

else:

print('Please enter a valid choice!!!\n')

case_data_values = {}

print('--- LEAPP Case Data file creation ---\n')

print('Enter the following information:')

case_data_values['Case Number'] = input("Case Number: ")

case_data_values['Agency'] = input("Agency: ")

case_data_values['Examiner'] = input("Examiner : ")

print()

case_data_filename = ''

while not case_data_filename:

case_data_filename = input('Enter the name of the Case Data file: ')

case_data_filename += '.lcasedata'

filename = os.path.join(path, case_data_filename)

with open(filename, "wt", encoding="utf-8") as case_data_file:

json.dump({"leapp": "case_data", "case_data_values": case_data_values}, case_data_file)

print('\nCase Data file saved:', filename)

print()

check_runtime_dependencies()

parser = argparse.ArgumentParser(description=f'iLEAPP v{leapp_version}: iOS Logs, Events, And Plists Parser.')

parser.add_argument('-t', choices=['fs', 'tar', 'zip', 'gz', 'itunes', 'file', 'raw'], required=False, action="store",

help=("Specify the input type. "

"'fs' for a folder containing extracted files with normal paths and names, "

"'tar', 'zip', or 'gz' for compressed packages containing files with normal names, "

"'itunes' for a folder containing a raw iTunes backup with hashed paths and names, "

"'raw' for a disk image (.img, .dd, .bin, or any numbered .001 segment of a split "

"set) or an acquisition (EnCase/EWF .E01, SMART .s01, EWF2 .Ex01, AFF .aff, AFM "

".afm, or any .aff in an AFD folder, AFF4 .aff4, an Apple .dmg, with any "

".dmgpart files beside it, .sparseimage, or .sparsebundle folder, or a virtual "

"machine disk, .vhd, .vhdx, .vmdk or .qcow2), read in "

"place without mounting (logical evidence, an EnCase .L01 or FTK Imager "

".ad1, is read as the files it holds): its "

"NTFS, FAT32, FAT16, FAT12, exFAT, ext2/3/4, F2FS, HFS+, APFS, QNX6, QNX4, "

"ETFS, EFS, SquashFS, JFFS2, UBI/UBIFS, YAFFS and QNX IFS volumes "

"are searched directly, "

"'file' for a single file input."))

parser.add_argument('-o', '--output_path', required=False, action="store",

help='Path to base output folder (this must exist)')

parser.add_argument('-i', '--input_path', required=False, action="store", nargs='+',

help='Path to input file/folder. With -t raw, several disk images can be given, '

'the member images of a RAID set (Linux md, or LVM2 across disks), and they '

'are read together')

parser.add_argument('--no-find-members', dest='no_find_members', required=False, action='store_true',

help='With -t raw and one image that holds a RAID member: read only that image. '

'Without this, the other members are looked for in its folder, by what each '

'file records about itself, listed with what was left out and why, and read '

'with it')

parser.add_argument('--image_password_file', required=False, action="store",

help='For an encrypted image (-t raw; an Apple disk image, an encrypted '

'AFF or an FTK Imager AD-encrypted set), or a BitLocker, encrypted '

'APFS or LUKS volume in an image: read its password (for BitLocker, a '

'password or recovery password; for APFS, a password or personal '

'recovery key; for LUKS, a passphrase) from the first line of this '

'file')

parser.add_argument('--image_password_env', required=False, action="store",

help='For an encrypted image, or a BitLocker, encrypted APFS or LUKS '

'volume in one (-t raw): take its password from this environment '

'variable. Without either, it is asked for at a terminal')

parser.add_argument('--image_private_key', required=False, action="store",

help='For an image sealed to a certificate (-t raw; an Apple disk image, '

'an AFF or an FTK Imager AD-encrypted set): the certificate\'s RSA '

'private key file, unencrypted, as PEM or DER')

parser.add_argument('--bitlocker_key', required=False, action="append", default=[],

help='For a BitLocker volume in an image (-t raw): its startup key '

'(.BEK) file. Repeatable')

parser.add_argument('-tz', '--timezone', required=False, action="store", default='UTC', type=str, help="Timezone name (e.g., 'America/New_York')")

parser.add_argument('-w', '--wrap_text', required=False, action="store_false", default=True,

help='Do not wrap text for output of data files')

parser.add_argument('-m', '--load_profile', required=False, action="store", help="Path to iLEAPP Profile file (.ilprofile).")

parser.add_argument('-d', '--load_case_data', required=False, action="store", help="Path to LEAPP Case Data file (.lcasedata).")

parser.add_argument('-c', '--create_profile_casedata', required=False, action="store",

help=("Generate an iLEAPP Profile file (.ilprofile) or LEAPP Case Data file (.lcasedata) into the specified path. "

"This argument is meant to be used alone, without any other arguments."))

parser.add_argument('-p', '--artifact_paths', required=False, action="store_true",

help=("Generate a text file list of artifact paths. "

"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('--html_row_limit', required=False, action="store", type=int,

default=artifact_report.HTML_TABLE_ROW_LIMIT,

help="Rows above which an artifact's table is left off its HTML page, which then points at "

"the LAVA database and the TSV export instead (default %(default)s). 0 writes every table.")

parser.add_argument('--hash', required=False, action='store_true',

help="Hash each staged file (SHA-256) after it is staged and before any "

"artifact opens it, and record the times its archive or image records "

"for it (UTC; a zip member's DOS date and time and a FAT or exFAT "

"entry's times are kept as text with no zone, and exFAT's UTC offsets "

"as stored, not applied) and where they came from, into LAVA's "

"_file_path_list and "

"'_TSV Exports/_Staged files SHA-256.tsv'. A folder or single-file input "

"records no times. Off by default; links and folders are recorded by "

"kind without a digest.")

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('--itunes_password', required=False, action="store", help="Password used for encrypted iTunes backup")

parser.add_argument('--keychain', required=False, action="store",

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('--version', action='version', version=f'{leapp_name} {leapp_version}')

# Check if no arguments were provided

if len(sys.argv) == 1:

parser.print_help(sys.stderr)

sys.exit()

args = parser.parse_args()

# -i takes one path, or with -t raw the several images of a RAID set

args.input_paths = list(args.input_path or [])

args.input_path = args.input_paths[0] if args.input_paths else None

if args.html_row_limit < 0:

parser.error('--html_row_limit must be 0 or a positive number of rows')

artifact_report.set_html_row_limit(args.html_row_limit)

available_plugins = []

loader_paths = [plugin_loader.PLUGINPATH]

if args.custom_artifacts_path:

loader_paths.append(pathlib.Path(args.custom_artifacts_path))

loader = plugin_loader.PluginLoader(plugin_paths=loader_paths)

for plugin in sorted(loader.plugins, key=lambda p: p.category):

if (plugin.module_name == 'iTunesBackupInfo'

or plugin.name == 'last_build'

or plugin.module_name == 'logarchive' and plugin.name != 'logarchive'):

continue

else:

available_plugins.append(plugin)

selected_plugins = available_plugins.copy()

profile_filename = None

casedata = {}

extracttype = args.t

try:

validate_args(args)

except argparse.ArgumentError as e:

parser.error(str(e))

if args.artifact_paths:

print('Artifact path list generation started.')

print('')

path_list = set()

for plugin in loader.plugins:

if plugin.module_name == 'logarchive':

continue

if isinstance(plugin.search, tuple):

for x in plugin.search:

path_list.add(x)

elif isinstance(plugin.search, str):

path_list.add(plugin.search)

else:

continue

with open('path_list.txt', 'w', encoding='utf-8') as paths:

for path in sorted(path_list):

paths.write(f'{path}\n')

print(path)

print('')

print('Artifact path list generation completed')

return

if args.create_profile_casedata:

if os.path.isdir(args.create_profile_casedata):

create_choice = ''

print('-' * 55)

print('Welcome to iLEAPP Profile or Case Data file creation\n')

instructions = 'You can type:\n'

instructions += ' - \'1\' to create an iLEAPP Profile file (.ilprofile)\n'

instructions += ' - \'2\' to create a LEAPP Case Data file (.lcasedata)\n'

instructions += ' - \'q\' to quit\n'

while not create_choice:

print(instructions)

create_choice = input('Please enter your choice: ').lower()

print()

if create_choice == '1':

create_profile(available_plugins, args.create_profile_casedata)

create_choice = ''

elif create_choice == '2':

create_casedata(args.create_profile_casedata)

create_choice = ''

elif create_choice == 'q':

return

else:

print('Please enter a valid choice!!!\n')

create_choice = ''

else:

print('OUTPUT folder for storing iLEAPP Profile file does not exist!\nRun the program again.')

return

if args.load_case_data:

case_data_filename = args.load_case_data

case_data_load_error = None

with open(case_data_filename, "rt", encoding="utf-8") as case_data_file:

try:

case_data = json.load(case_data_file)

except (ValueError, TypeError):

case_data_load_error = "File was not a valid case data file: invalid format"

print(case_data_load_error)

return

if not case_data_load_error:

if isinstance(case_data, dict):

if case_data.get("leapp") != "case_data":

case_data_load_error = "File was not a valid case data file"

print(case_data_load_error)

return

else:

print(f'Case Data loaded: {case_data_filename}')

casedata = case_data.get('case_data_values', {})

else:

case_data_load_error = "File was not a valid case data file: invalid format"

print(case_data_load_error)

return

if args.load_profile:

profile_filename = args.load_profile

profile_load_error = None

with open(profile_filename, "rt", encoding="utf-8") as profile_file:

try:

profile = json.load(profile_file)

except (ValueError, TypeError):

profile_load_error = "File was not a valid case data file: invalid format"

print(profile_load_error)

return

if not profile_load_error:

if isinstance(profile, dict):

if profile.get("leapp") != "ileapp" or profile.get("format_version") != 1:

profile_load_error = "File was not a valid profile file: incorrect LEAPP or version"

print(profile_load_error)

return

else:

profile_plugins = set(profile.get("plugins", []))

selected_plugins = [selected_plugin for selected_plugin in available_plugins

if selected_plugin.name in profile_plugins]

else:

profile_load_error = "File was not a valid profile file: invalid format"

print(profile_load_error)

return

input_path = args.input_path

wrap_text = args.wrap_text

output_path = os.path.abspath(args.output_path)

time_offset = args.timezone

custom_output_folder = args.custom_output_folder

itunes_backup_password = args.itunes_password

Context.set_keychain_path(args.keychain)

# ios file system extractions contain paths > 260 char, which causes problems

# This fixes the problem by prefixing \\?\ on each windows path.

if is_platform_windows():

if input_path[1] == ':' and extracttype =='fs': input_path = '\\\\?\\' + input_path.replace('/', '\\')

if output_path[1] == ':': output_path = '\\\\?\\' + output_path.replace('/', '\\')

# An encrypted image opens only with what locked it, and a BitLocker, APFS or LUKS

# volume in an image with its own key: from a file, an environment variable or the options

# naming key files, or asked for at a terminal, and checked before the run.

# One image that holds a RAID member has the other members looked for beside it,

# unless --no-find-members, and what was found and left out is printed first.

image_password = None

found_members = None

if extracttype == 'raw':

try:

raw_input = cli_raw_input(args.input_paths, args.image_password_file,

args.image_password_env, args.image_private_key,

args.bitlocker_key, look_beside=not args.no_find_members)

except ValueError as exc:

print(exc)

return

image_password = raw_input.keys

input_path = raw_input.input_path

# what was found beside the image given, and any image given twice

found_members = raw_input

out_params = OutputParameters(output_path, custom_output_folder)

Context.set_output_params(out_params)

initialize_lava(input_text(input_path, found_members),

out_params.output_folder_base, extracttype, profile_filename)

# Record history if enabled

if isinstance(input_path, list):

history.record_input_set(input_path, getattr(found_members, 'found', found_members))

else:

history.record_input_path(input_path)

history.record_output_path(output_path)

crunch_successful = crunch_artifacts(selected_plugins, extracttype, input_path, out_params, wrap_text, loader, casedata, time_offset,

profile_filename, itunes_backup_password, image_password=image_password, found_members=found_members, hash_files=args.hash)

lava_finalize_output(out_params.output_folder_base)

if crunch_successful:

# _lava_data.lava exists only after finalize; recording earlier is a no-op.

lava_project_path = os.path.join(out_params.output_folder_base, lava_json_name)

"""Return the plugins with the version readers among them moved to the front.

readers = [plugin for name in VERSION_READERS for plugin in plugins if plugin.name == name]

"""(SHA-256 or None, kind) for one staged path, as _file_path_list records it.

base = getattr(file_info, 'kind', 'file')

kind = base

if staged in getattr(seeker, 'padded_files', {}):

kind += "; read with zeros where the acquisition's data ran out"

if not hash_files or base not in HASHED_KINDS:

return None, kind

if getattr(file_info, 'sha256', None) is None:

digest = hashlib.sha256()

try:

with open(staged, 'rb') as handle:

for chunk in iter(lambda: handle.read(1 << 20), b''):

digest.update(chunk)

except OSError as ex:

logfunc(f'Could not hash {file_info.source_path}: {ex}')

return None, kind

file_info.sha256 = digest.hexdigest()

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,

image_password=None,

found_members=None, hash_files=False):

# input_path is one path, or for -t raw the list of images read together; then

# found_members says what was found beside the image given (a RawInput, a MemberSearch, or paths)

start = process_time()

start_wall = perf_counter()

logfunc('Processing started. Please wait. This may take a few minutes...')

logfunc('\n--------------------------------------------------------------------------------------')

logfunc(f'iLEAPP v{leapp_version}: iOS Logs, Events, And Plists Parser')

logfunc('Objective: Triage iOS Full File System and iTunes Backup Extractions.')

logfunc('By: Alexis Brignoni | @AlexisBrignoni | abrignoni.com')

logfunc('By: Yogesh Khatri | @SwiftForensics | swiftforensics.com\n')

report_supplied_keychain()

seeker = None

password = itunes_backup_password

try:

if extracttype == 'fs':

seeker = FileSeekerDir(input_path, out_params.data_folder)

elif extracttype == 'file':

seeker = FileSeekerFile(input_path, out_params.data_folder)

elif extracttype in ('tar', 'gz'):

seeker = FileSeekerTar(input_path, out_params.data_folder)

elif extracttype == 'zip':

seeker = FileSeekerZip(input_path, out_params.data_folder)

elif extracttype == 'raw':

seeker = FileSeekerRaw(input_path, out_params.data_folder, password=image_password,

found=found_members)

elif extracttype == 'itunes':

itunes_backup_type = get_itunes_backup_type(input_path)

if itunes_backup_type:

supported, encrypted, message = check_itunes_backup_status(

input_path, itunes_backup_type)

if not supported:

logfunc(message)

return False

else:

if encrypted:

while not decryption_keys:

if not password:

password = getpass("iTunes Backup password: ")

decryption_keys, _ = decrypt_itunes_backup(input_path, password)

if not decryption_keys:

return False

else:

logfunc('Input folder is not a valid iTunes backup!')

return False

seeker = FileSeekerItunes(input_path, out_params.data_folder,

itunes_backup_type, decryption_keys)

else:

logfunc('Error on argument -o (input type)')

return False

except Exception: # pylint: disable=broad-exception-caught

logfunc('Had an exception in Seeker - see details below. Terminating Program!')

temp_file = io.StringIO()

traceback.print_exc(file=temp_file)

logfunc(temp_file.getvalue())

temp_file.close()

return False

try:

# Now ready to run

plugins[:] = version_readers_first(plugins)

# add last_build at the start except for iTunes backups

if extracttype != 'itunes':

plugins.insert(0, loader["last_build"])

logfunc(f'Info: {len(loader) - 2} modules loaded.') # excluding last_build and iTunesBackupInfo

if profile_filename:

logfunc(f'Loaded profile: {profile_filename}')

logfunc(f'Artifact to parse: {len(plugins)}')

input_path = input_text(input_path, found_members)

logfunc(f'File/Directory selected: {input_path}')

hashes = None

if hash_files:

# opt in (--hash): each staged file is hashed once, after it is staged and

# before any artifact opens it, so the digest is of the bytes staged

logfunc("Staged files are hashed (SHA-256) as they are staged, and their recorded "

"times kept: in LAVA's _file_path_list and _TSV Exports/_Staged files SHA-256.tsv")

tsv_folder = os.path.join(out_params.output_folder_base, '_TSV Exports')

os.makedirs(tsv_folder, exist_ok=True)

hashes = open(os.path.join(tsv_folder, '_Staged files SHA-256.tsv'), 'w',

encoding='utf8', newline='')

hashes.write(STAGED_FILES_TSV_HEADER)

lava_add_file_path_columns(STAGED_TIME_SQL_COLUMNS)

hash_counts = {}

logfunc('\n--------------------------------------------------------------------------------------')

log = open(os.path.join(out_params.output_folder_base, '_HTML', '_Script_Logs', 'ProcessedFilesLog.html'), 'w+', encoding='utf8')

log.write(f'Extraction/Path selected: {input_path}<br><br>')

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

# 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')

if os.path.exists(info_plist_path):

# process_artifact([info_plist_path], 'iTunesBackupInfo', 'Device Info', seeker, out_params.output_folder_base)

#plugin.method([info_plist_path], out_params.output_folder_base, seeker, wrap_text)

report_folder = os.path.join(out_params.output_folder_base, '_HTML', 'iTunes Backup')

if not os.path.exists(report_folder):

try:

os.makedirs(report_folder)

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)

report_folder = os.path.join(out_params.output_folder_base, '_HTML', 'Installed Apps')

if not os.path.exists(report_folder):

try:

os.makedirs(report_folder)

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)

#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:

logfunc('Info.plist not found for iTunes Backup!')

log.write('Info.plist not found for iTunes Backup!')

# Search for the files per the arguments

parsed_modules = 0

plugin_run_times = []

lava_only = False

artifact_search_pattern_id = 0

file_path_ids = set()

for plugin_number, plugin in enumerate(plugins, start=1):

# Timed from here, before the file search, because the search runs after the

# "artifact started" line is printed and is itself a place a run can sit.

plugin_start = perf_counter()

logfunc()

logfunc('[{}/{}] {} [{}] artifact started at {} UTC'.format(plugin_number, len(plugins),

plugin.name, plugin.module_name,

strftime('%H:%M:%S', gmtime())))

output_types = plugin.artifact_info.get('output_types', '')

if isinstance(plugin.search, list) or isinstance(plugin.search, tuple):

search_regexes = plugin.search

elif plugin.search is None:

search_regexes = plugin.search

else:

search_regexes = [plugin.search]

files_found = []

log.write(f'<b>For {plugin.name} artifact</b>')

if search_regexes is None:

log.write(f'<ul><li>No search regexes provided for {plugin.name} artifact.')

log.write("<ul><li><i>'_lava_artifacts.db'</i> used as source file.</li></ul></li></ul>")

files_found = [os.path.join(out_params.output_folder_base, '_lava_artifacts.db')]

else:

for artifact_search_regex in search_regexes:

artifact_search_pattern_id += 1

lava_insert_sqlite_artifact_search_pattern(

artifact_search_pattern_id, plugin.module_name, plugin.name, artifact_search_regex)

pattern_already_searched = artifact_search_regex in seeker.searched

found = seeker.search(artifact_search_regex)

if not found:

if plugin.name == 'logarchive' and extracttype != 'fs' and extracttype != 'file':

# beside the image given (the first of several read together)

src = os.path.join(os.path.dirname(getattr(seeker, 'image_path', input_path)),

"logarchive.json")

dst = os.path.join(out_params.data_folder, "logarchive.json")

# The artifact declares several search patterns, so this branch is

# reached once per pattern that misses; only pick the export up once.

if os.path.exists(src) and dst not in files_found:

copy2(src, dst)

files_found.append(dst)

log.write(f'<ul><li>No file found for regex <i>{artifact_search_regex}</i></li></ul>')

else:

log.write(f'<ul><li>{len(found)} {"files" if len(found) > 1 else "file"} for regex <i>{artifact_search_regex}</i> located at:')

for pathh in found:

# Strip \\?\ only for log display; file_infos is keyed with the

# original long-path form on Windows.

display_path = pathh[4:] if pathh.startswith('\\\\?\\') else pathh

log.write(f'<ul><li>{Context.get_relative_path(display_path)}</li></ul>')

if seeker.file_infos.get(pathh):

file_path_id = id(seeker.file_infos.get(pathh))

if not pattern_already_searched and file_path_id not in file_path_ids:

file_info = seeker.file_infos.get(pathh)

digest, kind = staged_file_record(seeker, pathh, file_info,

hash_files)

lava_insert_sqlite_file_path(file_path_id, file_info.source_path,

digest, kind,

staged_time_values(file_info) if hash_files else None)

if hashes is not None:

hashes.write(staged_files_tsv_line(file_info, kind, digest))

hash_counts[digest is not None] = \

hash_counts.get(digest is not None, 0) + 1

file_path_ids.add(file_path_id)

lava_insert_sqlite_artifact_link_pattern_to_file(artifact_search_pattern_id, file_path_id)

log.write('</li></ul>')

files_found.extend(found)

if files_found:

if not lava_only and 'lava_only' in output_types:

lava_only = True

category_folder = os.path.join(out_params.output_folder_base, '_HTML',

sanitize_report_name(plugin.category, 'category'))

if not os.path.exists(category_folder):

try:

os.makedirs(category_folder)

except (FileExistsError, FileNotFoundError) as ex:

logfunc('Error creating {} report directory at path {}'.format(plugin.name, category_folder))

logfunc('Error was {}'.format(str(ex)))

lava_add_module(plugin.module_name, "Error", len(files_found), plugin.name)

continue # cannot do work

try:

plugin.method(files_found, category_folder, seeker, wrap_text, time_offset)

lava_commit()

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 does_table_exist_in_db(lava_db_path, 'logarchive_artifacts'):

unifed_logs_artifacts = []

unifed_logs_artifacts = [plugin.name for plugin in loader.plugins

if plugin.module_name=='logarchive'

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)

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()))

plugin_elapsed = perf_counter() - plugin_start

plugin_run_times.append((plugin.name, plugin_elapsed))

logfunc('{} [{}] artifact failed after {:.1f}s'.format(

plugin.name, plugin.module_name, plugin_elapsed))

lava_add_module(plugin.module_name, "Error", len(files_found), plugin.name)

lava_commit()

continue # nope

lava_add_module(plugin.module_name, "Complete", len(files_found), plugin.name)

else:

lava_add_module(plugin.module_name, "No files found", 0, plugin.name)

logfunc("No file found")

plugin_elapsed = perf_counter() - plugin_start

plugin_run_times.append((plugin.name, plugin_elapsed))

logfunc('{} [{}] artifact completed in {:.1f}s'.format(

plugin.name, plugin.module_name, plugin_elapsed))

parsed_modules += 1

GuiWindow.SetProgressBar(parsed_modules, len(plugins))

log.flush()

log.close()

if hashes is not None:

hashes.close()

logfunc(f'Staged files hashed: {hash_counts.get(True, 0):,}; recorded without a '

f'digest (folders, links): {hash_counts.get(False, 0):,}')

write_device_info()

if lava_only:

write_lava_only_log()

logfunc('')

logfunc('Processes completed.')

end = process_time()

end_wall = perf_counter()

run_time_secs = end - start

run_time_HMS = strftime('%H:%M:%S', gmtime(run_time_secs))

logfunc("Processing time (CPU) = {}".format(run_time_HMS))

run_time_secs = end_wall - start_wall

run_time_HMS = strftime('%H:%M:%S', gmtime(run_time_secs))

logfunc("Run time (wall clock) = {}".format(run_time_HMS))

# Filtered on the rounded value so the list holds exactly what the per-artifact

# lines showed as 1.0s or more, with no artifact visible above the cut but missing here.

slowest = sorted((entry for entry in plugin_run_times if round(entry[1], 1) >= 1.0),

key=lambda item: item[1], reverse=True)[:SLOWEST_ARTIFACTS_TO_REPORT]

if slowest:

logfunc('')

logfunc('Slowest artifacts:')

for artifact_name, elapsed in slowest:

logfunc(' {:>8.1f}s {}'.format(elapsed, artifact_name))

logfunc('')

logfunc('Report generation started.')

# remove the \\?\ prefix we added to input and output paths, so it does not reflect in report

if is_platform_windows():

if out_params.output_folder_base.startswith('\\\\?\\'):

out_params.output_folder_base = out_params.output_folder_base[4:]

if input_path.startswith('\\\\?\\'):

input_path = input_path[4:]

report.generate_report(out_params.output_folder_base, run_time_secs, run_time_HMS, extracttype, input_path, casedata, profile_filename, icons, lava_only)

logfunc('Report generation Completed.')

logfunc('')

logfunc(f'Report location: {out_params.output_folder_base}')

return True

finally:

# A raw image seeker holds the image open and the other seekers hold an

# archive handle; cleanup() releases whatever the seeker holds. Nothing

# else does, so run it on every exit path, not just the success return.