diff --git a/.github/workflows/python_lint.yml b/.github/workflows/python_lint.yml index 5bdd1d0..1f87f1e 100644 --- a/.github/workflows/python_lint.yml +++ b/.github/workflows/python_lint.yml @@ -80,6 +80,16 @@ jobs: if: steps.changed-files-py.outputs.any_changed == 'true' run: python admin/scripts/check_source_path.py + # That same "located at" line and the LAVA manifest already carry the evidence + # location, so an artifact that reads one system-wide file does not also need a + # Source File column repeating the identical path on every row. The column earns + # its place only when the declared paths can match more than one file, which is + # why a per-user or per-app path keeps it: that column is what attributes a row + # to a user or to one of several stores. + - name: Guard against a Source File column that repeats one path + if: steps.changed-files-py.outputs.any_changed == 'true' + run: python admin/scripts/check_source_file_column.py + # scripts/vendor/ holds code copied verbatim from another repository, so it # drifts in two directions and both are silent: an edit here looks like a fix # until the next re-vendor reverts it, and an upstream release leaves this copy diff --git a/admin/scripts/check_source_file_column.py b/admin/scripts/check_source_file_column.py new file mode 100644 index 0000000..d87604c --- /dev/null +++ b/admin/scripts/check_source_file_column.py @@ -0,0 +1,226 @@ +"""Guard against a per-row Source File column that repeats one path on every row. + +An artifact returns `(data_headers, data_list, source_path)`. The third element +already carries the evidence location: `artifact_processor` splits it on newlines, +reduces each piece with `Context.get_relative_path`, prints it as the report's + + located at: + +line, and writes it into the LAVA manifest, where it shows in the artifact +information modal. So when an artifact reads exactly one file, a per-row +`Source File` column adds a column to every row of the report that holds the +same string the modal already shows. It costs width in a table an examiner reads +under time pressure and tells them nothing the modal did not. + +The column earns its place only when the source can differ between rows. That is +the case whenever the artifact's declared `paths` can match more than one file: + + * a wildcard in the file name (`*.evtx`, `*.lnk`, `*.pf`, `$I*`), + * a wildcard directory segment, so the pattern matches per-instance folders, + * a user-relative root (`Users/`, `Library/`, `AppData/`, ...), which repeats + once per user profile on the image, + * several patterns naming genuinely different files. + +This check flags the opposite: a `Source File` column on an artifact whose paths +are a single, system-anchored, fixed file name. + +Deliberately NOT flagged, because "one source on the images we happen to hold" is +not the same claim as "one source": + + * A per-user artifact that reads one hive on a single-user test image. Its path + is user-relative, so it is treated as multi-source and keeps its column. That + column is what attributes a row to a user on a multi-user machine, and + dropping it there would be a real loss of attribution. + +To silence a case this reads wrongly, add it to ALLOWLIST with a reason. +""" + +import ast +import os +import re +import sys + +ARTIFACTS_DIR = os.path.join('scripts', 'artifacts') + +_SOURCE_COLUMN = re.compile(r'^source\s*(file|path)s?$', re.IGNORECASE) + +# A path segment under which files repeat once per user profile. +_USER_ROOT = re.compile( + r'(^|/)\*?/?(Users|Library|AppData|Documents|Profiles|Accounts|\.config)(/|$)', + re.IGNORECASE) +# Android exposes one app directory under several storage views (data/data, +# data/user/, data_mirror/...) and once per Android user, so a pattern naming +# an app directory matches several real files even though it names one logical +# store. The reverse-DNS segment is the tell, and the same shape names an app +# container on iOS. +_PACKAGE_SEGMENT = re.compile(r'^[A-Za-z][A-Za-z0-9_]*(\.[A-Za-z0-9_-]+)+$') +_ANDROID_ROOT = re.compile( + r'(^|/)(data/data|data/user|data_mirror|shared_prefs|databases)(/|$)', + re.IGNORECASE) +# A path anchored at a directory that exists once per machine. Flagging is +# positive: unless the pattern is anchored at one of these, it is assumed it can +# match more than one file and the column is left alone. That keeps the check +# conservative, because a wrongly removed column costs per-row attribution while +# a wrongly kept one costs only a repeated cell. +_SYSTEM_ROOT = re.compile( + r'(^|/)(Windows|ProgramData|System32|PerfLogs' # Windows + r'|system|misc|efs' # Android /data/system, /data/misc, /efs + r'|private/var/db|var/db)(/|$)', # iOS and macOS + re.IGNORECASE) + +# module:artifact -> reason. Keep each entry keyed to the exact artifact and say why. +# +# The ALEAPP entries below are artifacts whose own notes state that their path +# pattern is tolerant, so a second copy of the system file elsewhere in an +# extraction would contribute its rows again and the column names the file each +# row came from. That is a deliberate design decision by the artifact's author, +# documented in examiner-facing notes, so the column stays. +_TOLERANT_SYSTEM_COPY = ('the artifact\'s notes state its path pattern is tolerant, so a ' + 'second copy of this system file in an extraction adds its rows ' + 'again and the column names the file each row came from') +ALLOWLIST = { + 'adbAuthorizations.py:adb_authorizations': _TOLERANT_SYSTEM_COPY, + 'appOpsAccesses.py:appops_accesses': _TOLERANT_SYSTEM_COPY, + 'installSessions.py:install_sessions': _TOLERANT_SYSTEM_COPY, + 'packageDexUsage.py:package_dex_usage_app_code': _TOLERANT_SYSTEM_COPY, + 'packageDexUsage.py:package_dex_usage_cross_package': _TOLERANT_SYSTEM_COPY, + 'packageDexUsage.py:package_dex_usage_secondary': _TOLERANT_SYSTEM_COPY, + 'packageDexUsageList.py:package_dex_usage_list_cross_package': _TOLERANT_SYSTEM_COPY, + 'packageDexUsageList.py:package_dex_usage_list_secondary': _TOLERANT_SYSTEM_COPY, + 'sRecoveryhist.py:get_sRecoveryhist': _TOLERANT_SYSTEM_COPY, +} + + +def _file_name(pattern): + return pattern.rstrip('/').split('/')[-1] + + +def _can_match_several_files(patterns): + """True when the declared paths can match more than one file on one image.""" + if len({_file_name(p).lower() for p in patterns}) > 1: + return True + for pattern in patterns: + if '*' in _file_name(pattern): + return True + body = pattern.lstrip('*').lstrip('/') + if any('*' in segment for segment in body.split('/')[:-1]): + return True + if _USER_ROOT.search('/' + body) and not _SYSTEM_ROOT.search('/' + body): + return True + if not _SYSTEM_ROOT.search('/' + body): + # Not anchored at a once-per-machine directory, so assume it can match + # more than one file rather than risk removing real attribution. + return True + if _ANDROID_ROOT.search('/' + body): + return True + if any(_PACKAGE_SEGMENT.match(segment) for segment in body.split('/')[:-1]): + return True + return False + + +def _artifact_paths(tree): + """artifact key -> declared paths, from a literal __artifacts_v2__.""" + for node in ast.walk(tree): + if not isinstance(node, ast.Assign): + continue + if not any(getattr(t, 'id', '') == '__artifacts_v2__' for t in node.targets): + continue + try: + block = ast.literal_eval(node.value) + except (ValueError, TypeError, SyntaxError): + return None + found = {} + for key, value in block.items(): + if isinstance(value, dict): + paths = value.get('paths') + if isinstance(paths, str): + paths = [paths] + found[key] = list(paths) if paths else [] + return found + return {} + + +def _source_columns(function): + """The source-file style column names in a function's data_headers, or None + when the headers are not statically readable.""" + headers = None + for node in ast.walk(function): + if isinstance(node, ast.Assign) and any( + getattr(t, 'id', '') == 'data_headers' for t in node.targets): + try: + headers = ast.literal_eval(node.value) + except (ValueError, TypeError, SyntaxError): + return None + if headers is None: + return None + names = [h[0] if isinstance(h, (tuple, list)) else h for h in headers] + return [n for n in names if isinstance(n, str) and _SOURCE_COLUMN.match(n.strip())] + + +def _processors(tree): + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + names = [d.id if isinstance(d, ast.Name) else getattr(d, 'attr', None) + for d in node.decorator_list] + if 'artifact_processor' in [n for n in names if n]: + yield node + + +def main(): + problems = [] + not_checked = [] + checked = 0 + + for file_name in sorted(os.listdir(ARTIFACTS_DIR)): + if not file_name.endswith('.py'): + continue + path = os.path.join(ARTIFACTS_DIR, file_name) + with open(path, 'r', encoding='utf-8') as handle: + try: + tree = ast.parse(handle.read()) + except SyntaxError as exc: + not_checked.append(f'{file_name}: could not parse ({exc})') + continue + paths_by_artifact = _artifact_paths(tree) + if paths_by_artifact is None: + not_checked.append(f'{file_name}: __artifacts_v2__ is not a literal') + continue + for function in _processors(tree): + columns = _source_columns(function) + if columns is None: + not_checked.append(f'{file_name}:{function.name}: data_headers is not literal') + continue + checked += 1 + if not columns: + continue + declared = paths_by_artifact.get(function.name) + if not declared: + continue + if _can_match_several_files(declared): + continue + if f'{file_name}:{function.name}' in ALLOWLIST: + continue + problems.append( + f' {file_name}:{function.name}: {columns[0]} repeats one path on every ' + f'row. Its paths match a single system-wide file ' + f'({declared[0]}), and the location already reaches the report and the ' + f'LAVA manifest through the third return element. Drop the column, or ' + f'allowlist it with a reason.') + + if problems: + print('A Source File column repeats one path on every row:\n') + print('\n'.join(problems)) + print(f'\nChecked {checked} artifact(s); {len(problems)} need attention.') + return 1 + + print(f'Checked {checked} artifact(s): no Source File column repeats a single path.') + if not_checked: + print(f'{len(not_checked)} artifact(s)/module(s) NOT checked:') + for entry in not_checked: + print(f' {entry}') + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/scripts/artifacts/ford_power_history.py b/scripts/artifacts/ford_power_history.py index 696e320..db73ef0 100644 --- a/scripts/artifacts/ford_power_history.py +++ b/scripts/artifacts/ford_power_history.py @@ -148,14 +148,13 @@ def ford_power_history(context): mode.group(1) if mode else '', fields.get('reset type', ''), fields.get('reset initiator', ''), fields.get('reset reason', ''), fields.get('reboot source', ''), - uptime, total_uptime, - context.get_relative_path(source_path))) + uptime, total_uptime)) data_headers = (('Powered On', 'datetime'), ('Previous Shutdown', 'datetime'), 'Boot Count', 'Wake Source (as stored)', 'Target Mode (as stored)', 'Reset Type', 'Reset Initiator', 'Reset Reason', 'Reboot Source (as stored)', 'Up Time (seconds)', - 'Total Up Time (as stored)', 'Source File') + 'Total Up Time (as stored)') return data_headers, data_list, context.get_relative_path(source_path) @@ -177,7 +176,7 @@ def _single_record(context, filename): return fields, context.get_relative_path(source_path) -def _power_row(fields, relative_path): +def _power_row(fields): """One row, shared by last-shutdown.txt and reset-reason.txt.""" try: # The unit is known to be milliseconds, so it is divided here rather @@ -187,11 +186,11 @@ def _power_row(fields, relative_path): stamp = None return (stamp, fields.get('boot count', ''), fields.get('initiator', ''), fields.get('reason', ''), fields.get('up-time', ''), - fields.get('total up-time', ''), relative_path) + fields.get('total up-time', '')) _POWER_HEADERS = (('Timestamp', 'datetime'), 'Boot Count', 'Initiator', 'Reason', - 'Up Time (as stored)', 'Total Up Time (as stored)', 'Source File') + 'Up Time (as stored)', 'Total Up Time (as stored)') @artifact_processor @@ -199,7 +198,7 @@ def ford_power_last_shutdown(context): fields, relative_path = _single_record(context, "last-shutdown.txt") if not fields: return (), [], relative_path - return _POWER_HEADERS, [_power_row(fields, relative_path)], relative_path + return _POWER_HEADERS, [_power_row(fields)], relative_path @artifact_processor @@ -207,4 +206,4 @@ def ford_power_reset_reason(context): fields, relative_path = _single_record(context, "reset-reason.txt") if not fields: return (), [], relative_path - return _POWER_HEADERS, [_power_row(fields, relative_path)], relative_path + return _POWER_HEADERS, [_power_row(fields)], relative_path