From 2da3401f67b6cdaba3c8edc3f33b300557711cb7 Mon Sep 17 00:00:00 2001 From: Forest Savage Date: Sun, 23 Aug 2026 13:37:47 -0800 Subject: [PATCH 1/2] Keep case-variant evidence files when the report volume folds case Seeker dest-guard only (volume probe + ~caseN). Not an artifact. Closes abrignoni/VLEAPP#131. Signed-off-by: Forest Savage --- scripts/search_files.py | 139 ++++++++++++++++++++++++++++++++++++---- 1 file changed, 128 insertions(+), 11 deletions(-) diff --git a/scripts/search_files.py b/scripts/search_files.py index fb26ab1..fe7ed81 100755 --- a/scripts/search_files.py +++ b/scripts/search_files.py @@ -33,6 +33,86 @@ 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 _disambiguated_data_path(data_path, source_key, dest_claims, folds_case): + """Pick a dest that does not overwrite a different source.""" + 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) + n = 2 + while True: + alt = f"{root}~case{n}{ext}" + alt_key = _dest_claim_key(alt, folds_case) + if alt_key not in dest_claims and not os.path.lexists(alt): + 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 + n += 1 + + class FileInfo: """ A class to store file metadata information. @@ -61,6 +141,15 @@ def search(self, filepattern, return_on_first_hit=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): + return _disambiguated_data_path( + data_path, source_key, self._dest_claims, self._data_folder_folds_case + ) + class FileSeekerDir(FileSeekerBase): """ @@ -91,6 +180,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 +211,7 @@ 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) os.makedirs(os.path.dirname(data_path), exist_ok=True) copy2(item, data_path) self.copied[item] = data_path @@ -173,6 +264,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: @@ -190,6 +282,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) @@ -244,6 +337,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): """ @@ -290,7 +384,10 @@ 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) + 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) @@ -311,7 +408,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 @@ -319,20 +424,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() @@ -372,6 +482,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: @@ -426,7 +537,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 + ) + 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() From 99c859cafb9120d8220b2f7a63251b03b6a6145a Mon Sep 17 00:00:00 2001 From: Brigs Date: Mon, 24 Aug 2026 13:44:59 -0500 Subject: [PATCH 2/2] Derive case-collision names from the source path and skip directory members Replaces the counter suffix with name~case-.ext, where the tag is the sha256 of the evidence-relative source path (case preserved, separators normalized). The same source now maps to the same destination on every search, every run and every machine, so repeated or forced searches cannot mint additional copies. Directory members take no dest-guard: case-variant directories fold into one and their files disambiguate individually, which also stops a spurious tagged twin when an archive stores a directory entry after its contents. The dir seeker hashes the path relative to the input root rather than the mounted absolute path. Adds a regression suite covering both volume behaviours, name determinism across instances and forced re-searches, directory members, and a simulated Windows normcase. Co-Authored-By: Claude Fable 5 --- .../scripts/test_seeker_case_collision.py | 293 ++++++++++++++++++ scripts/search_files.py | 97 ++++-- 2 files changed, 368 insertions(+), 22 deletions(-) create mode 100644 admin/test/scripts/test_seeker_case_collision.py 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 fe7ed81..5d33321 100755 --- a/scripts/search_files.py +++ b/scripts/search_files.py @@ -87,8 +87,43 @@ def _dest_claim_key(data_path, folds_case): return normalized.casefold() if folds_case else normalized -def _disambiguated_data_path(data_path, source_key, dest_claims, folds_case): - """Pick a dest that does not overwrite a different source.""" +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: @@ -99,18 +134,22 @@ def _disambiguated_data_path(data_path, source_key, dest_claims, folds_case): dest_claims[key] = (source_key, data_path) return data_path root, ext = os.path.splitext(data_path) - n = 2 - while True: - alt = f"{root}~case{n}{ext}" + 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) - if alt_key not in dest_claims and not os.path.lexists(alt): - 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 - n += 1 + 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: @@ -138,6 +177,12 @@ 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''' @@ -145,9 +190,10 @@ 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): + 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 + data_path, source_key, self._dest_claims, + self._data_folder_folds_case, hash_source=hash_source ) @@ -211,7 +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) + 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 @@ -384,10 +431,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: - intended = self._intended_extract_path(member) - extracted_path = self._extract_member( - member, dest_path=self._unique_data_path(intended, 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) @@ -538,8 +591,8 @@ def search(self, filepattern, return_on_first_hit=False, force=False): if self.single_file_abs_path not in self.copied or force: try: dest_data_path = self._unique_data_path( - dest_data_path, self.single_file_abs_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,