diff --git a/admin/test/scripts/test_seeker_case_collision.py b/admin/test/scripts/test_seeker_case_collision.py new file mode 100644 index 0000000..f0aa76a --- /dev/null +++ b/admin/test/scripts/test_seeker_case_collision.py @@ -0,0 +1,293 @@ +"""Pin the seeker's dest-guard for case-variant evidence files (issue #1948). + +An iOS extraction can hold com.apple.MobileSMS.plist and com.apple.mobileSMS.plist +in one directory as two different files. On a case-insensitive report volume both +sources fold to one destination under data/, and before the guard the second copy +silently destroyed the first, so the preserved file at a cited path could hold the +other file's bytes. + +The guard writes the colliding copy to name~case-.ext instead, where the tag +is derived from the evidence-relative source path. These tests pin three +properties: both byte streams survive, the alternate name is a pure function of +the source (stable across re-searches, seeker instances and force=True), and +directory members never mint a tagged twin. + +The claims logic folds keys only when the data folder's volume folds case, which +CI runners may not. Tests that need folding force the seeker's probed flag, which +exercises the identical code path on any filesystem. +""" +import hashlib +import os +import pathlib +import shutil +import sys +import tempfile +import unittest +import zipfile +from functools import lru_cache + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[3] +sys.path.insert(0, str(REPO_ROOT)) + +from scripts.search_files import ( # pylint: disable=wrong-import-position + FileSeekerZip, _case_variant_digest, _disambiguated_data_path, + _probe_volume_case_insensitive) +import scripts.search_files # pylint: disable=wrong-import-position + +PREF = 'private/var/mobile/Library/Preferences/' +UPPER = PREF + 'com.apple.MobileSMS.plist' +LOWER = PREF + 'com.apple.mobileSMS.plist' +UPPER_CONTENT = b'upper spelling bytes' +LOWER_CONTENT = b'lower spelling bytes' +PATTERN = '*/mobile/Library/Preferences/com.apple.[Mm]obileSMS.plist' + + +def _expected_tag(member): + return hashlib.sha256(member.encode('utf-8')).hexdigest()[:8] + + +def _tree_files(base): + found = {} + for root, _dirs, files in os.walk(base): + for name in files: + full = os.path.join(root, name) + with open(full, 'rb') as fin: + found[os.path.relpath(full, base).replace(os.sep, '/')] = fin.read() + return found + + +class TestDisambiguatorUnit(unittest.TestCase): + """The claims logic, exercised directly with both volume behaviours.""" + + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.tmpdir, ignore_errors=True) + + def _dest(self, name): + return os.path.join(self.tmpdir, name) + + def test_case_sensitive_volume_keeps_both_plain_names(self): + claims = {} + first = _disambiguated_data_path(self._dest('A.plist'), 'src/A.plist', + claims, False) + second = _disambiguated_data_path(self._dest('a.plist'), 'src/a.plist', + claims, False) + self.assertEqual(os.path.basename(first), 'A.plist') + self.assertEqual(os.path.basename(second), 'a.plist') + + def test_folding_volume_tags_the_second_source(self): + claims = {} + first = _disambiguated_data_path(self._dest('A.plist'), 'src/A.plist', + claims, True) + second = _disambiguated_data_path(self._dest('a.plist'), 'src/a.plist', + claims, True) + self.assertEqual(os.path.basename(first), 'A.plist') + self.assertEqual(os.path.basename(second), + f'a~case-{_expected_tag("src/a.plist")}.plist') + + def test_same_source_returns_the_same_path_every_time(self): + claims = {} + _disambiguated_data_path(self._dest('A.plist'), 'src/A.plist', claims, True) + paths = {_disambiguated_data_path(self._dest('a.plist'), 'src/a.plist', + claims, True) + for _ in range(4)} + self.assertEqual(len(paths), 1) + + def test_tag_is_derived_from_hash_source_when_given(self): + claims = {} + _disambiguated_data_path(self._dest('A.plist'), '/mnt/evidence/A.plist', + claims, True, hash_source='rel/A.plist') + second = _disambiguated_data_path(self._dest('a.plist'), + '/mnt/evidence/a.plist', claims, True, + hash_source='rel/a.plist') + self.assertEqual(os.path.basename(second), + f'a~case-{_expected_tag("rel/a.plist")}.plist') + + def test_separator_and_leading_slash_do_not_change_the_tag(self): + self.assertEqual(_case_variant_digest('a\\b/C.plist'), + _case_variant_digest('/a/b/C.plist')) + self.assertNotEqual(_case_variant_digest('a/b/C.plist'), + _case_variant_digest('a/b/c.plist')) + + def test_squatted_short_tag_falls_through_to_the_full_digest(self): + claims = {} + _disambiguated_data_path(self._dest('A.plist'), 'src/A.plist', claims, True) + digest = _case_variant_digest('src/a.plist') + squatter = self._dest(f'a~case-{digest[:8]}.plist') + with open(squatter, 'wb') as fout: + fout.write(b'evidence file that owns the short name') + second = _disambiguated_data_path(self._dest('a.plist'), 'src/a.plist', + claims, True) + self.assertEqual(os.path.basename(second), f'a~case-{digest}.plist') + + +class TestZipCaseCollision(unittest.TestCase): + """FileSeekerZip with the folding path forced, so it runs on any volume.""" + + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + self.zip_path = os.path.join(self.tmpdir, 'fs-full.zip') + with zipfile.ZipFile(self.zip_path, 'w') as z: + z.writestr(UPPER, UPPER_CONTENT) + z.writestr(LOWER, LOWER_CONTENT) + + def tearDown(self): + shutil.rmtree(self.tmpdir, ignore_errors=True) + + def _seeker(self, subdir): + data_folder = os.path.join(self.tmpdir, subdir) + os.makedirs(data_folder, exist_ok=True) + seeker = FileSeekerZip(self.zip_path, data_folder) + seeker._data_folder_folds_case = True # pylint: disable=protected-access + return seeker, data_folder + + def test_both_sources_survive_under_distinct_names(self): + seeker, data_folder = self._seeker('data') + try: + paths = [str(p) for p in seeker.search(PATTERN)] + self.assertEqual(len(paths), 2) + names = sorted(os.path.basename(p) for p in paths) + self.assertEqual(names, ['com.apple.MobileSMS.plist', + f'com.apple.mobileSMS~case-{_expected_tag(LOWER)}.plist']) + contents = sorted(_tree_files(data_folder).values()) + self.assertEqual(contents, sorted([UPPER_CONTENT, LOWER_CONTENT])) + finally: + seeker.cleanup() + + def test_file_infos_keep_the_true_source_for_the_tagged_copy(self): + seeker, _data_folder = self._seeker('data') + try: + paths = [str(p) for p in seeker.search(PATTERN)] + tagged = [p for p in paths if '~case-' in p] + self.assertEqual(len(tagged), 1) + self.assertEqual(seeker.file_infos[tagged[0]].source_path, LOWER) + finally: + seeker.cleanup() + + def test_forced_research_returns_stable_paths_and_mints_nothing(self): + seeker, data_folder = self._seeker('data') + try: + first = sorted(str(p) for p in seeker.search(PATTERN)) + for _ in range(3): + again = sorted(str(p) for p in seeker.search(PATTERN, force=True)) + self.assertEqual(again, first) + self.assertEqual(len(_tree_files(data_folder)), 2) + finally: + seeker.cleanup() + + def test_two_seeker_instances_produce_identical_names(self): + seeker_a, folder_a = self._seeker('data_a') + seeker_b, folder_b = self._seeker('data_b') + try: + seeker_a.search(PATTERN) + seeker_b.search(PATTERN) + names_a = sorted(_tree_files(folder_a)) + names_b = sorted(_tree_files(folder_b)) + self.assertEqual(names_a, names_b) + finally: + seeker_a.cleanup() + seeker_b.cleanup() + + def test_directory_members_never_mint_a_tagged_twin(self): + late_dir_zip = os.path.join(self.tmpdir, 'late-dir.zip') + with zipfile.ZipFile(late_dir_zip, 'w') as z: + z.writestr('a/b/file.txt', b'content') + z.writestr(zipfile.ZipInfo('a/b/'), b'') + data_folder = os.path.join(self.tmpdir, 'data_dirs') + os.makedirs(data_folder) + seeker = FileSeekerZip(late_dir_zip, data_folder) + seeker._data_folder_folds_case = True # pylint: disable=protected-access + try: + seeker.search('*/a/b*') + entries = [] + for _root, dirs, files in os.walk(data_folder): + entries.extend(dirs) + entries.extend(files) + self.assertFalse([e for e in entries if '~case-' in e], entries) + finally: + seeker.cleanup() + + +class TestZipCaseCollisionRealVolume(unittest.TestCase): + """No forcing: whatever this volume does, both byte streams must survive.""" + + def test_both_byte_streams_exist_on_disk(self): + tmpdir = tempfile.mkdtemp() + try: + zip_path = os.path.join(tmpdir, 'fs-full.zip') + with zipfile.ZipFile(zip_path, 'w') as z: + z.writestr(UPPER, UPPER_CONTENT) + z.writestr(LOWER, LOWER_CONTENT) + data_folder = os.path.join(tmpdir, 'data') + os.makedirs(data_folder) + seeker = FileSeekerZip(zip_path, data_folder) + try: + seeker.search(PATTERN) + finally: + seeker.cleanup() + contents = sorted(_tree_files(data_folder).values()) + self.assertEqual(contents, sorted([UPPER_CONTENT, LOWER_CONTENT])) + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + + +class TestWindowsPatternFold(unittest.TestCase): + """With Windows-style normcase, the bracket pattern still matches both members. + + os.path.normcase on Windows lowercases and flips separators. The seeker caches + it at import; simulate it here to pin that both case-variant archive members + match and both are preserved, which is what a Windows run does at the matching + layer. + """ + + def setUp(self): + self._saved = scripts.search_files.normcase + scripts.search_files.normcase = lru_cache(maxsize=None)( + lambda s: s.replace('/', '\\').lower()) + + def tearDown(self): + scripts.search_files.normcase = self._saved + + def test_bracket_pattern_matches_both_spellings(self): + tmpdir = tempfile.mkdtemp() + try: + zip_path = os.path.join(tmpdir, 'fs-full.zip') + with zipfile.ZipFile(zip_path, 'w') as z: + z.writestr(UPPER, UPPER_CONTENT) + z.writestr(LOWER, LOWER_CONTENT) + data_folder = os.path.join(tmpdir, 'data') + os.makedirs(data_folder) + seeker = FileSeekerZip(zip_path, data_folder) + seeker._data_folder_folds_case = True # pylint: disable=protected-access + try: + paths = [str(p) for p in seeker.search(PATTERN)] + finally: + seeker.cleanup() + self.assertEqual(len(paths), 2) + contents = sorted(_tree_files(data_folder).values()) + self.assertEqual(contents, sorted([UPPER_CONTENT, LOWER_CONTENT])) + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + + +class TestVolumeProbe(unittest.TestCase): + """The probe reports the volume truthfully and cleans up after itself.""" + + def test_probe_is_boolean_and_leaves_nothing(self): + tmpdir = tempfile.mkdtemp() + try: + result = _probe_volume_case_insensitive(tmpdir) + self.assertIsInstance(result, bool) + self.assertEqual([f for f in os.listdir(tmpdir) if 'probe' in f], []) + # A volume where aA resolves after writing only Aa folds case. + with open(os.path.join(tmpdir, 'Aa'), 'w', encoding='utf-8') as fout: + fout.write('x') + self.assertEqual(result, os.path.exists(os.path.join(tmpdir, 'aA'))) + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + + +if __name__ == '__main__': + unittest.main() diff --git a/scripts/search_files.py b/scripts/search_files.py index 8954307..75b21fc 100755 --- a/scripts/search_files.py +++ b/scripts/search_files.py @@ -33,6 +33,125 @@ from functools import lru_cache normcase = lru_cache(maxsize=None)(os.path.normcase) +def _probe_volume_case_insensitive(folder): + """True when this folder's volume folds case. + + os.path.normcase reports the platform convention, not the volume. + Probe by creating Aa then exclusively creating aA. + """ + try: + os.makedirs(folder, exist_ok=True) + except OSError: + return os.path.normcase("Aa") == os.path.normcase("aA") + probe_a = os.path.join(folder, ".leapp_case_probe_Aa") + probe_b = os.path.join(folder, ".leapp_case_probe_aA") + for leftover in (probe_a, probe_b): + try: + os.remove(leftover) + except OSError: + pass + wrote_a = False + wrote_b = False + try: + fd = os.open(probe_a, os.O_CREAT | os.O_EXCL | os.O_WRONLY) + os.write(fd, b"Aa") + os.close(fd) + wrote_a = True + try: + fd = os.open(probe_b, os.O_CREAT | os.O_EXCL | os.O_WRONLY) + os.write(fd, b"aA") + os.close(fd) + wrote_b = True + return False + except FileExistsError: + return True + except OSError: + return os.path.normcase("Aa") == os.path.normcase("aA") + except OSError: + return os.path.normcase("Aa") == os.path.normcase("aA") + finally: + if wrote_a: + try: + os.remove(probe_a) + except OSError: + pass + if wrote_b: + try: + os.remove(probe_b) + except OSError: + pass + + +def _dest_claim_key(data_path, folds_case): + normalized = os.path.normpath(data_path) + return normalized.casefold() if folds_case else normalized + + +def _case_variant_digest(hash_source): + """Hex tag for a case-variant copy, derived from the source path. + + Hash the evidence-relative spelling, case preserved, separators normalized, + so the same source maps to the same tag on every search, every run and + every machine, and each case variant gets its own tag. + """ + normalized = str(hash_source).replace('\\', '/').lstrip('/') + return hashlib.sha256(normalized.encode('utf-8', 'surrogatepass')).hexdigest() + + +def _case_variant_candidates(root, ext, digest): + """Candidate alternate names: short tag, full digest, then a counter tail. + + The later tiers only matter when a candidate name is already taken, for + example by an evidence file that legitimately carries the tagged name. The + walk over them stays finite because every blocker is a recorded claim or an + existing file, and both sets are finite. + """ + yield f"{root}~case-{digest[:8]}{ext}" + yield f"{root}~case-{digest}{ext}" + n = 2 + while True: + yield f"{root}~case-{digest}-{n}{ext}" + n += 1 + + +def _disambiguated_data_path(data_path, source_key, dest_claims, folds_case, + hash_source=None): + """Pick a dest that does not overwrite a different source. + + When the wanted destination is already claimed by a different source (two + evidence paths differing only in case fold together on a case-insensitive + report volume), the copy is written to name~case-.ext instead. The tag + names the source rather than the arrival order, so re-searches and repeat + runs land on the same path with nothing to remember. + """ + key = _dest_claim_key(data_path, folds_case) + claimed = dest_claims.get(key) + if claimed is not None: + claimed_source, claimed_path = claimed + if claimed_source == source_key: + return claimed_path + elif not os.path.lexists(data_path): + dest_claims[key] = (source_key, data_path) + return data_path + root, ext = os.path.splitext(data_path) + digest = _case_variant_digest(source_key if hash_source is None else hash_source) + for alt in _case_variant_candidates(root, ext, digest): + alt_key = _dest_claim_key(alt, folds_case) + claimed = dest_claims.get(alt_key) + if claimed is not None: + if claimed[0] == source_key: + return claimed[1] + continue + if os.path.lexists(alt): + continue + dest_claims[alt_key] = (source_key, alt) + logfunc( + f"INFO: destination {data_path} already holds a different source; " + f"writing {source_key} to {alt}" + ) + return alt + + class FileInfo: """ A class to store file metadata information. @@ -58,9 +177,25 @@ def search(self, filepattern, return_on_first_hit=False): '''Returns a list of paths for files/folders that matched''' raise NotImplementedError + def __init__(self): + # Refined by _init_dest_guard once the subclass knows its data folder; + # the defaults leave the dest-guard a pass-through. + self._dest_claims = {} + self._data_folder_folds_case = False + def cleanup(self): '''close any open handles''' + def _init_dest_guard(self, data_folder): + self._dest_claims = {} + self._data_folder_folds_case = _probe_volume_case_insensitive(data_folder) + + def _unique_data_path(self, data_path, source_key, hash_source=None): + return _disambiguated_data_path( + data_path, source_key, self._dest_claims, + self._data_folder_folds_case, hash_source=hash_source + ) + class FileSeekerDir(FileSeekerBase): """ @@ -91,6 +226,7 @@ def __init__(self, directory, data_folder): self.searched = {} self.copied = {} self.file_infos = {} + self._init_dest_guard(self.data_folder) def build_files_list(self, directory): '''Populates all paths in directory into _all_files''' @@ -121,6 +257,8 @@ def search(self, filepattern, return_on_first_hit=False, force=False): if os.path.isdir(item): pass elif os.path.isfile(item): + data_path = self._unique_data_path( + data_path, item, hash_source=item_rel_path) os.makedirs(os.path.dirname(data_path), exist_ok=True) copy2(item, data_path) self.copied[item] = data_path @@ -174,6 +312,7 @@ def __init__(self, tar_file_path, data_folder): self.searched = {} self.copied = {} self.file_infos = {} + self._init_dest_guard(self.data_folder) def search(self, filepattern, return_on_first_hit=False, force=False): if filepattern in self.searched and not force: @@ -191,6 +330,7 @@ def search(self, filepattern, return_on_first_hit=False, force=False): if member.isdir(): os.makedirs(full_path, exist_ok=True) else: + full_path = self._unique_data_path(str(full_path), member.name) parent_dir = os.path.dirname(full_path) if not os.path.exists(parent_dir): os.makedirs(parent_dir) @@ -245,6 +385,7 @@ def __init__(self, zip_file_path, data_folder): self.searched = {} self.copied = {} self.file_infos = {} + self._init_dest_guard(self.data_folder) def decode_extended_timestamp(self, extra_data): """ @@ -291,7 +432,16 @@ def search(self, filepattern, return_on_first_hit=False, force=False): if pat(root + normcase(member)) is not None: if member not in self.copied or force: try: - extracted_path = self._extract_member(member) + if member.endswith('/'): + # Case-variant directories fold into one on a + # case-insensitive volume; their files disambiguate + # individually, so directory members take no guard. + extracted_path = self._extract_member(member) + else: + intended = self._intended_extract_path(member) + extracted_path = self._extract_member( + member, + dest_path=self._unique_data_path(intended, member)) f = self.zip_file.getinfo(member) creation_date, modification_date = self.decode_extended_timestamp(f.extra) file_info = FileInfo(member, creation_date, modification_date) @@ -312,7 +462,15 @@ def search(self, filepattern, return_on_first_hit=False, force=False): self.searched[filepattern] = pathlist return pathlist - def _extract_member(self, member): + def _intended_extract_path(self, member): + clean_member = sanitize_file_path(member) + parts = [part for part in clean_member.replace('\\', '/').split('/') + if part not in ('', '.', '..')] + if not parts: + return self.data_folder + return os.path.join(self.data_folder, *parts) + + def _extract_member(self, member, dest_path=None): """Extract one member, sanitizing names ZipFile.extract() cannot write. ZipFile.extract() only replaces a fixed set of printable characters @@ -320,20 +478,25 @@ def _extract_member(self, member): name (present in real iOS extractions, e.g. chronod icon files) reach the OS untouched and Windows rejects them with EINVAL. Members whose names need sanitizing are written out manually to a cleaned path. + + dest_path, when given, is the already-disambiguated destination so a + later case-variant member cannot overwrite an earlier one. """ + intended = self._intended_extract_path(member) + if dest_path is None: + dest_path = intended clean_member = sanitize_file_path(member) - if clean_member == member: + if dest_path == intended and clean_member == member: return self.zip_file.extract(member, path=self.data_folder) - parts = [part for part in clean_member.split('/') - if part not in ('', '.', '..')] - extracted_path = os.path.join(self.data_folder, *parts) if member.endswith('/'): - os.makedirs(extracted_path, exist_ok=True) + os.makedirs(dest_path, exist_ok=True) else: - os.makedirs(os.path.dirname(extracted_path), exist_ok=True) - with self.zip_file.open(member) as fin, open(extracted_path, 'wb') as fout: + parent = os.path.dirname(dest_path) + if parent: + os.makedirs(parent, exist_ok=True) + with self.zip_file.open(member) as fin, open(dest_path, 'wb') as fout: fout.write(fin.read()) - return extracted_path + return dest_path def cleanup(self): self.zip_file.close() @@ -373,6 +536,7 @@ def __init__(self, file_path, data_folder): self.searched = {} self.copied = {} self.file_infos = {} + self._init_dest_guard(self.data_folder) def search(self, filepattern, return_on_first_hit=False, force=False): if not self.single_file_basename: @@ -427,7 +591,13 @@ def search(self, filepattern, return_on_first_hit=False, force=False): if self.single_file_abs_path not in self.copied or force: try: - os.makedirs(self.data_folder, exist_ok=True) + dest_data_path = self._unique_data_path( + dest_data_path, self.single_file_abs_path, + hash_source=self.single_file_basename) + os.makedirs( + os.path.dirname(dest_data_path) or self.data_folder, + exist_ok=True, + ) copy2(self.single_file_abs_path, dest_data_path) self.copied[self.single_file_abs_path] = dest_data_path s = Path(self.single_file_abs_path).stat()