diff --git a/docs/repo-sentinel-baseline-review.md b/docs/repo-sentinel-baseline-review.md index c9e080d..5586a1e 100644 --- a/docs/repo-sentinel-baseline-review.md +++ b/docs/repo-sentinel-baseline-review.md @@ -213,14 +213,23 @@ variables and global/system Git config are excluded; lazy fetching is disabled. Archive attributes, textconv, checkout filters, and executable file modes do not transform or execute the returned bytes. -Initial admission is intentionally narrow: regular files with mode `100644` -or `100755`, ASCII path components containing letters, digits, spaces, dots, -underscores or hyphens, at most 255 bytes per component. Dot components, `.git`, -Windows device names (including spaces before an extension), trailing -dots/spaces, and case-folding collisions are -refused. Symlinks, gitlinks and other modes refuse the entire snapshot. This -portable subset excludes legitimate names, including non-ASCII names; refusal -must remain visible rather than silently dropping files. +Path admission now represents logical Git-tree identity rather than host +filesystem portability. Each raw component must decode as strict UTF-8, +re-encode to the original bytes, and remain within the 255 raw-byte component +bound. Empty components, raw `/`, exact `.` or `..`, C0/C1 controls, DEL, +malformed framing, exact duplicate paths, and exact file/directory namespace +conflicts refuse the complete snapshot. Symlinks, gitlinks and other modes +remain unsupported. + +The strict UTF-8 rule is a deliberate compatibility boundary for this reader, +not a claim that arbitrary non-NUL, non-slash Git path bytes are invalid Git. +Decoded names are not normalized, case-folded, transliterated, or repaired. +Logical identity is exact Python `str` equality, so case variants and NFC/NFD +variants remain distinct and exact-string sorting stays deterministic. Host-only +concerns such as `.git`, backslash, colon, Windows device names, wildcards, and +trailing dots or spaces are intentionally left to the materializer. Ordinary +Unicode and inert punctuation such as `&`, apostrophes, `!`, `~`, and `$` are +admitted without filename-specific exceptions. Defaults bound the complete read to 4,096 files, 8,192 object reads, 16 MiB of cumulative raw object bodies, 2 MiB per blob, 32 directory levels and 30 seconds. @@ -234,12 +243,13 @@ Real bare-repository tests exercise both object formats, raw binary content, archive attributes, replacement refs, type/mode mismatch, unsafe paths, collisions, malformed trees and file/object/byte/depth budgets. A stalled child checks the deadline, and injected transport corruption checks independent -identity validation. Run `python -m unittest tests.test_repo_sentinel_reader`. +identity validation. Run +`python -m unittest discover -s tests -p 'test_repo_sentinel_reader.py'`. -This API is not wired into the scanner or workflow yet. PR-ref acquisition, -filesystem materialization, producer/head binding and enforcement remain -separate work. Existing scan and baseline behavior is unchanged. Rollback is -removal of the reader and its tests; there is no persisted state or migration. +These APIs are not wired into the scanner or workflow yet. PR-ref acquisition, +producer/head binding, and enforcement remain separate work. Existing scan and +baseline behavior is unchanged. Rollback restores the prior reader/materializer +contract; there is no persisted state or migration. ### Temporary Snapshot Materialization @@ -250,6 +260,34 @@ directory below an existing, caller-owned scratch root. The scratch root itself must be a real directory, not a symlink or Windows reparse point; its ancestors and stability are caller trust requirements. +The empty private staging container is created first so root-dependent path +limits can be evaluated. Before any snapshot subdirectory or file is populated, +the materializer revalidates every `SnapshotFile.path`, derives all implicit +directories, and completes indexed collision and type-prefix checks across the +whole tree. This independent preflight also protects against caller-constructed +`Snapshot` values that did not originate in the reader. + +Canonical materialization rejects Windows separators, drive/UNC/device forms, +ADS colons, reserved punctuation and device names, including the Windows-defined +superscript-digit `COM`/`LPT` aliases, trailing dots/spaces, and +ASCII-case-insensitive `.git` components on every supported host. Portable-v1 +case aliases use an explicit ASCII-only `A-Z` mapping independent of the runner +OS. NFC and NFD keys detect Unicode portability aliases separately without +changing logical identity. This deterministic policy does not claim to emulate +every evolving filesystem-specific Unicode casing rule. Full Unicode lowercasing +or case folding is not used, so names such as `ẞ`/`ß`, `ß`/`SS`, and `İ`/`i` +remain distinct. Logical names can therefore be reader-valid but +materializer-invalid by design. + +The preflight retains the reader's 255-byte component and depth bounds. It also +checks UTF-16 component and conservative `MAX_PATH` length against the actual +private staging root on Windows, and runtime `pathconf` component/full-path +limits on POSIX. Unavailable or indeterminate POSIX limits refuse the operation; +the absolute limit is consequently a documented runtime-root precondition +rather than an emulation of a foreign filesystem. Validated component tuples +are joined with `root.joinpath(*components)`; unvalidated logical path strings +never reach host path joining. + Files are created exclusively and read back with exact byte comparison before the context yields `MaterializedSnapshot(snapshot, root)`. Existing files and symlinks are never overwritten. Directory entries are checked before use; @@ -269,16 +307,19 @@ The caller must prevent concurrent mutation by other processes or the consumer. These checks do not defend against a hostile local actor changing ancestors or files during use. Reader byte/count limits bound the selected data, but its deadline does not bound filesystem I/O, cleanup or consumer execution. Host -path-length or disk failures refuse materialization rather than shortening paths -or returning fewer files. Empty Git directories are not part of the reader's -file snapshot and are not reconstructed. - -Run `python -m unittest tests.test_repo_sentinel_materialize` for real-object -fixtures covering bytes, executable-mode metadata, empty files/trees, lifecycle, -reader refusal, collisions, symlinks, same-length corruption, partial setup failures -and cleanup failures. Scanner execution, PR-ref acquisition and workflow -activation remain unwired. Rollback removes this additive helper/tests/docs; -there is no persisted output format or repository-setting migration. +path-length or disk failures refuse materialization rather than shortening +paths or returning fewer files. Empty Git directories are not part of the +reader's file snapshot and are not reconstructed. + +Run +`python -m unittest discover -s tests -p 'test_repo_sentinel_materialize.py'` +for real-object and direct-`Snapshot` fixtures covering bytes, executable-mode +metadata, empty files/trees, lifecycle, layered path refusal, aliases, +normalization and type collisions, path limits, preflight atomicity, symlinks, +same-length corruption, partial setup failures, and cleanup failures. Scanner +execution, PR-ref acquisition, and workflow activation remain unwired. Rollback +restores the prior reader/materializer contract; there is no persisted output +format or repository-setting migration. ## Relationship To Issue #5 diff --git a/scripts/repo_sentinel_materialize.py b/scripts/repo_sentinel_materialize.py index dac8c32..54da570 100644 --- a/scripts/repo_sentinel_materialize.py +++ b/scripts/repo_sentinel_materialize.py @@ -2,15 +2,17 @@ from __future__ import annotations +import ntpath import os import stat +import unicodedata from collections.abc import Iterator from contextlib import contextmanager from dataclasses import dataclass, field from pathlib import Path from tempfile import TemporaryDirectory -from repo_sentinel_reader import ReaderLimits, Snapshot, read_snapshot +from repo_sentinel_reader import ReaderLimits, Snapshot, SnapshotFile, read_snapshot class MaterializationRefused(ValueError): @@ -23,6 +25,141 @@ class MaterializedSnapshot: root: Path = field(repr=False) +@dataclass(frozen=True) +class _MaterializationPlan: + files: tuple[tuple[SnapshotFile, tuple[str, ...]], ...] + directories: tuple[tuple[str, ...], ...] + + +_WINDOWS_FORBIDDEN = frozenset('<>:"\\|?*') +_WINDOWS_DEVICES = frozenset({"CON", "PRN", "AUX", "NUL", "CONIN$", "CONOUT$"}) +_WINDOWS_PORT_DEVICE_SUFFIXES = frozenset("123456789¹²³") +_ASCII_CASE_TRANSLATION = { + codepoint: codepoint + (ord("a") - ord("A")) + for codepoint in range(ord("A"), ord("Z") + 1) +} + + +def _has_control_characters(value: str) -> bool: + return any(ord(character) < 0x20 or 0x7f <= ord(character) <= 0x9f + for character in value) + + +def _logical_components(path: str, limits: ReaderLimits) -> tuple[str, ...]: + if type(path) is not str or ntpath.splitdrive(path)[0] or ntpath.isabs(path): + raise MaterializationRefused("unsupported_host_path") + components = tuple(path.split("/")) + if (not components or len(components) > limits.max_depth + 1 + or any(not component or component in (".", "..") + or _has_control_characters(component) for component in components)): + raise MaterializationRefused("unsupported_host_path") + for component in components: + try: + encoded = component.encode("utf-8", errors="strict") + except UnicodeEncodeError: + raise MaterializationRefused("unsupported_host_path") from None + if len(encoded) > 255: + raise MaterializationRefused("path_limit") + return components + + +def _validate_portable_component(component: str) -> None: + stem = component.split(".", 1)[0].rstrip(" ").upper() + if (any(character in _WINDOWS_FORBIDDEN for character in component) + or component.endswith((".", " ")) + or (component.isascii() and component.lower() == ".git") + or stem in _WINDOWS_DEVICES + or (len(stem) == 4 and stem[:3] in ("COM", "LPT") + and stem[3] in _WINDOWS_PORT_DEVICE_SUFFIXES)): + raise MaterializationRefused("unsupported_host_path") + + +def _path_key( + components: tuple[str, ...], normalization: str | None +) -> tuple[str, ...]: + if normalization is not None: + return tuple( + unicodedata.normalize(normalization, component) for component in components + ) + return tuple(component.translate(_ASCII_CASE_TRANSLATION) for component in components) + + +def _pathconf(root: Path, name: str) -> int | None: + try: + value = os.pathconf(root, name) + except (AttributeError, OSError, ValueError): + return None + return value if type(value) is int and value > 0 else None + + +def _validate_host_lengths(root: Path, paths: tuple[tuple[str, ...], ...]) -> None: + if os.name == "nt": + for components in paths: + for component in components: + if len(component.encode("utf-16-le")) // 2 > 255: + raise MaterializationRefused("path_limit") + target = root.joinpath(*components) + if len(str(target).encode("utf-16-le")) // 2 >= 260: + raise MaterializationRefused("path_limit") + return + if os.name != "posix": + raise MaterializationRefused("unsupported_host_path") + name_limit = _pathconf(root, "PC_NAME_MAX") + path_limit = _pathconf(root, "PC_PATH_MAX") + if name_limit is None or path_limit is None: + raise MaterializationRefused("unsupported_host_path") + for components in paths: + try: + if name_limit is not None and any( + len(os.fsencode(component)) > name_limit for component in components + ): + raise MaterializationRefused("path_limit") + target = root.joinpath(*components) + if path_limit is not None and len(os.fsencode(target)) >= path_limit: + raise MaterializationRefused("path_limit") + except UnicodeEncodeError: + raise MaterializationRefused("unsupported_host_path") from None + + +def _preflight_snapshot( + snapshot: Snapshot, root: Path, limits: ReaderLimits +) -> _MaterializationPlan: + files: list[tuple[SnapshotFile, tuple[str, ...]]] = [] + logical_nodes: dict[tuple[str, ...], str] = {} + for item in snapshot.files: + components = _logical_components(item.path, limits) + for component in components: + _validate_portable_component(component) + for depth in range(1, len(components)): + directory = components[:depth] + if logical_nodes.get(directory) == "file": + raise MaterializationRefused("path_collision") + logical_nodes.setdefault(directory, "directory") + if components in logical_nodes: + raise MaterializationRefused("path_collision") + logical_nodes[components] = "file" + files.append((item, components)) + + for normalization in (None, "NFC", "NFD"): + aliases: dict[tuple[str, ...], tuple[str, ...]] = {} + for logical_path in logical_nodes: + key = _path_key(logical_path, normalization) + previous = aliases.get(key) + if previous is not None and previous != logical_path: + raise MaterializationRefused("host_path_collision") + aliases[key] = logical_path + + all_paths = tuple(logical_nodes) + _validate_host_lengths(root, all_paths) + directories = tuple( + sorted( + (path for path, kind in logical_nodes.items() if kind == "directory"), + key=lambda path: (len(path), path), + ) + ) + return _MaterializationPlan(tuple(files), directories) + + def _regular_path(path: Path, *, directory: bool = False) -> None: info = path.lstat() reparse = getattr(info, "st_file_attributes", 0) & getattr( @@ -76,17 +213,15 @@ def materialized_snapshot( dir=scratch_root.resolve()) root = Path(temporary.name) / "data" root.mkdir(mode=0o700) - directories = {root} - for item in snapshot.files: - directory = root - for component in item.path.split("/")[:-1]: - directory = directory / component - if directory not in directories: - directory.mkdir(mode=0o700) - directories.add(directory) - _regular_path(directory, directory=True) - _regular_path(root, directory=True) - _write_verified(root / item.path, item.data) + _regular_path(root, directory=True) + plan = _preflight_snapshot(snapshot, root, limits) + for components in plan.directories: + directory = root.joinpath(*components) + directory.mkdir(mode=0o700) + _regular_path(directory, directory=True) + _regular_path(root, directory=True) + for item, components in plan.files: + _write_verified(root.joinpath(*components), item.data) except OSError: raise MaterializationRefused("filesystem_io_failed") from None yield MaterializedSnapshot(snapshot, root) diff --git a/scripts/repo_sentinel_reader.py b/scripts/repo_sentinel_reader.py index 8351de2..6f3a914 100644 --- a/scripts/repo_sentinel_reader.py +++ b/scripts/repo_sentinel_reader.py @@ -90,16 +90,18 @@ def _git(repository: Path, arguments: list[str], cap: int, deadline: float) -> b def _component(raw: bytes) -> str: - # A deliberately portable subset, before any future filesystem writes. - if not re.fullmatch(rb"[A-Za-z0-9._ -]{1,255}", raw): - raise ReaderRefused("unsupported_path") - name = raw.decode("ascii") - stem = name.split(".", 1)[0].rstrip(" ").upper() - if (name in (".", "..") or name.lower() == ".git" - or name.endswith((".", " ")) - or stem in {"CON", "PRN", "AUX", "NUL"} - or re.fullmatch(r"(?:COM|LPT)[1-9]", stem)): - raise ReaderRefused("unsupported_path") + if len(raw) > 255: + raise ReaderRefused("path_limit") + try: + name = raw.decode("utf-8", errors="strict") + except UnicodeDecodeError: + raise ReaderRefused("unsupported_path_encoding") from None + if name.encode("utf-8") != raw: + raise ReaderRefused("unsupported_path_encoding") + if (not name or b"/" in raw or name in (".", "..") + or any(ord(character) < 0x20 or 0x7f <= ord(character) <= 0x9f + for character in name)): + raise ReaderRefused("invalid_logical_path") return name @@ -161,9 +163,9 @@ def walk(oid: str, parent: str, depth: int) -> None: child = tree[nul + 1:nul + 1 + oid_bytes].hex() offset = nul + 1 + oid_bytes path = parent + name - if path.casefold() in paths: + if path in paths: raise ReaderRefused("path_collision") - paths.add(path.casefold()) + paths.add(path) if mode == b"40000": walk(child, path + "/", depth + 1) elif mode in (b"100644", b"100755"): diff --git a/tests/test_repo_sentinel_materialize.py b/tests/test_repo_sentinel_materialize.py index 346c4ca..d0cdee6 100644 --- a/tests/test_repo_sentinel_materialize.py +++ b/tests/test_repo_sentinel_materialize.py @@ -12,7 +12,13 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import repo_sentinel_materialize as materializer # noqa: E402 -from repo_sentinel_reader import ReaderLimits, ReaderRefused # noqa: E402 +from repo_sentinel_reader import ( # noqa: E402 + ReaderLimits, + ReaderRefused, + Snapshot, + SnapshotFile, + read_snapshot, +) class MaterializationTests(unittest.TestCase): @@ -49,6 +55,38 @@ def commit(self, tree: str) -> str: def materialize(self): return materializer.materialized_snapshot(self.repository, self.head, self.scratch) + def synthetic_snapshot(self, *paths: str) -> Snapshot: + files = tuple( + SnapshotFile(path, "100644", f"{index:040x}", f"payload-{index}".encode()) + for index, path in enumerate(paths, start=1) + ) + return Snapshot("0" * 40, "1" * 40, files) + + def assert_preflight_refused(self, snapshot: Snapshot, code: str) -> None: + captured: list[TemporaryDirectory[str]] = [] + + def capture(**kwargs: object) -> TemporaryDirectory[str]: + temporary = TemporaryDirectory(**kwargs) + cleanup = temporary.cleanup + self.addCleanup(cleanup) + temporary.cleanup = lambda: None + captured.append(temporary) + return temporary + + with ( + patch.object(materializer, "read_snapshot", return_value=snapshot), + patch.object(materializer, "TemporaryDirectory", side_effect=capture), + patch.object(materializer, "_write_verified") as writer, + self.assertRaisesRegex(materializer.MaterializationRefused, f"^{code}$"), + self.materialize(), + ): + self.fail("refused snapshot reached consumer") + self.assertEqual(writer.call_count, 0) + self.assertEqual(len(captured), 1) + output = Path(captured[0].name) / "data" + self.assertTrue(output.is_dir()) + self.assertEqual(list(output.iterdir()), []) + def test_exact_files_are_yielded_as_data_then_removed(self) -> None: sentinel = self.scratch / "existing.txt" sentinel.write_bytes(b"keep unrelated scratch content") @@ -64,6 +102,197 @@ def test_exact_files_are_yielded_as_data_then_removed(self) -> None: self.assertFalse(output.exists()) self.assertEqual(list(self.scratch.iterdir()), [sentinel]) + def test_portable_positive_matrix_preserves_logical_identity(self) -> None: + paths = ( + "README.md", + "space name.txt", + "a&b.md", + "it's.md", + "bang!.md", + "a–b.md", + "curly’apostrophe.md", + "café.md", + "notes & café/it's–ready!.md", + "portable/cash$-review¬es!.md", + "portable/tilde~name.md", + "portable/confusable∕separator.md", + "portable/ß.md", + "portable/SS.md", + "portable/İ.md", + "portable/i.md", + "portable/COM⁴.txt", + "portable/LPT⁰.log", + ) + snapshot = self.synthetic_snapshot(*paths) + with ( + patch.object(materializer, "read_snapshot", return_value=snapshot), + self.materialize() as result, + ): + self.assertEqual(result.snapshot, snapshot) + for item in snapshot.files: + target = result.root.joinpath(*item.path.split("/")) + self.assertEqual(target.read_bytes(), item.data) + + def test_host_specific_paths_are_rejected_before_any_write(self) -> None: + paths = ( + ".git/config", + ".GiT/config", + "a\\b", + "CON", + "con.txt", + "PRN", + "AUX", + "NUL.txt", + "COM1", + "COM1.txt", + "LPT9", + "COM¹", + "COM²", + "COM³", + "LPT¹", + "LPT²", + "LPT³", + "COM¹.txt", + "LPT³.log", + "NUL .txt", + "COM1 .log", + "foo.", + "foo ", + "a:b", + "star*.txt", + "x?y", + 'a"b', + "ab", + "a|b", + "C:", + r"C:\x", + r"\\server\share", + r"\\?\C:\x", + r"\\.\NUL", + r"\rooted", + "/absolute", + ".", + "..", + "safe//file", + "safe/", + "../escape", + "safe/../escape", + "safe/a\0b", + "safe/a\x7fb", + "safe/a\u0080b", + ) + for path in paths: + with self.subTest(path=path): + self.assert_preflight_refused( + self.synthetic_snapshot(path), "unsupported_host_path" + ) + + def test_reader_accepts_a_host_specific_name_that_materializer_refuses(self) -> None: + blob = self.object("blob", b"logical data") + tree = self.object("tree", b"100644 a:b\0" + bytes.fromhex(blob)) + head = self.commit(tree) + snapshot = read_snapshot(self.repository, head) + self.assertEqual([item.path for item in snapshot.files], ["a:b"]) + with ( + self.assertRaisesRegex( + materializer.MaterializationRefused, "^unsupported_host_path$" + ), + materializer.materialized_snapshot(self.repository, head, self.scratch), + ): + self.fail("host-specific path reached consumer") + self.assertEqual(list(self.scratch.iterdir()), []) + + def test_arbitrary_snapshot_path_representation_is_revalidated(self) -> None: + invalid_paths = (None, Path("path-object"), "bad\udcff") + for path in invalid_paths: + with self.subTest(kind=type(path).__name__): + item = SnapshotFile(path, "100644", "2" * 40, b"payload") + snapshot = Snapshot("0" * 40, "1" * 40, (item,)) + self.assert_preflight_refused(snapshot, "unsupported_host_path") + + def test_host_aliases_and_type_conflicts_are_rejected_in_preflight(self) -> None: + cases = ( + ("A.txt", "a.txt"), + ("é.txt", "e\u0301.txt"), + ("Dir/left", "dir/right"), + ("dir", "DIR/child"), + ) + for paths in cases: + with self.subTest(paths=paths): + self.assert_preflight_refused( + self.synthetic_snapshot(*paths), "host_path_collision" + ) + + def test_portable_case_policy_keeps_non_ascii_names_distinct(self) -> None: + paths = ("ẞ.txt", "ß.txt", "SS.txt", "İ.txt", "i.txt") + snapshot = self.synthetic_snapshot(*paths) + with ( + patch.object(materializer, "read_snapshot", return_value=snapshot), + self.materialize() as result, + ): + self.assertEqual( + {path.name for path in result.root.iterdir()}, + set(paths), + ) + + def test_exact_duplicate_is_rejected_in_preflight(self) -> None: + for paths in (("same", "same"), ("dir", "dir/child")): + with self.subTest(paths=paths): + self.assert_preflight_refused( + self.synthetic_snapshot(*paths), "path_collision" + ) + + def test_component_and_full_path_limits_are_preflighted(self) -> None: + long_path = "/".join(["a" * 220] * 20 + ["file"]) + for path in ("a" * 256, long_path): + with self.subTest(length=len(path)): + self.assert_preflight_refused( + self.synthetic_snapshot(path), "path_limit" + ) + + def test_unknown_posix_path_limits_fail_closed(self) -> None: + with ( + patch.object(materializer.os, "name", "posix"), + patch.object(materializer, "_pathconf", return_value=None), + self.assertRaisesRegex( + materializer.MaterializationRefused, "^unsupported_host_path$" + ), + ): + materializer._validate_host_lengths(self.scratch, (("safe",),)) + + def test_late_invalid_path_cannot_leave_partial_snapshot_content(self) -> None: + self.assert_preflight_refused( + self.synthetic_snapshot("safe/ok.txt", "later/NUL.txt"), + "unsupported_host_path", + ) + + def test_late_superscript_device_cannot_leave_partial_snapshot_content(self) -> None: + self.assert_preflight_refused( + self.synthetic_snapshot("safe.txt", "COM¹.txt"), + "unsupported_host_path", + ) + + def test_only_validated_components_reach_host_path_joining(self) -> None: + snapshot = self.synthetic_snapshot("nested/file.txt") + path_type = type(self.scratch) + original = path_type.__truediv__ + + def reject_raw_path(left: Path, right: object) -> Path: + if isinstance(right, str) and "/" in right: + raise AssertionError("raw logical path reached host joining") + return original(left, right) + + with ( + patch.object(materializer, "read_snapshot", return_value=snapshot), + patch.object(path_type, "__truediv__", reject_raw_path), + self.materialize() as result, + ): + self.assertEqual( + result.root.joinpath("nested", "file.txt").read_bytes(), + snapshot.files[0].data, + ) + def test_consumer_exception_is_preserved_and_directory_is_removed(self) -> None: with self.assertRaisesRegex(RuntimeError, "consumer failed"): with self.materialize() as result: diff --git a/tests/test_repo_sentinel_reader.py b/tests/test_repo_sentinel_reader.py index aa8959b..15c48fc 100644 --- a/tests/test_repo_sentinel_reader.py +++ b/tests/test_repo_sentinel_reader.py @@ -84,20 +84,90 @@ def test_nonregular_modes_and_type_mismatch_are_refused(self) -> None: with self.subTest(mode=mode), self.assertRaises(ReaderRefused): read_snapshot(root, self.commit(root, tree)) - def test_unsafe_paths_and_case_collisions_are_refused(self) -> None: + def test_strict_utf8_logical_paths_admit_repository_punctuation(self) -> None: root, _, blob = self.fixture() - for name in (b"..", b".GiT", b"a/b", b"a\\b", b"NUL.txt", b"LPT1", - b"NUL .txt", b"COM1 .log", b"x.", b"x ", b"a:b", - b"a\nb", b"a\tb", b"\xff"): + names = ( + "README.md", + "a b.txt", + "a&b.md", + "it's.md", + "bang!.md", + "a–b.md", + "curly’apostrophe.md", + "café.md", + ) + for name in names: + with self.subTest(name=name): + tree = self.tree(root, [(b"100644", name.encode("utf-8"), blob)]) + result = read_snapshot(root, self.commit(root, tree)) + self.assertEqual([item.path for item in result.files], [name]) + child = self.tree(root, [(b"100644", "it's–ready!.md".encode(), blob)]) + tree = self.tree(root, [(b"40000", "notes & café".encode(), child)]) + result = read_snapshot(root, self.commit(root, tree)) + self.assertEqual( + [item.path for item in result.files], + ["notes & café/it's–ready!.md"], + ) + + def test_host_specific_names_remain_valid_logical_identity(self) -> None: + root, _, blob = self.fixture() + names = ( + "a\\b", "a:b", "foo.", "foo ", "CON", "con.txt", "COM1.txt", + "LPT9", "star*.txt", "x?y", "C:", "C:\\x", "\\\\server\\share", + ".git", ".GiT", "cash$-review¬es!.md", "COM¹", "COM²", "COM³", + "LPT¹", "LPT²", "LPT³", "COM¹.txt", "LPT³.log", + ) + for name in names: + with self.subTest(name=name): + tree = self.tree(root, [(b"100644", name.encode("utf-8"), blob)]) + result = read_snapshot(root, self.commit(root, tree)) + self.assertEqual([item.path for item in result.files], [name]) + + def test_invalid_logical_paths_are_refused_with_stable_codes(self) -> None: + root, _, blob = self.fixture() + cases = ( + (b"", "invalid_logical_path"), + (b".", "invalid_logical_path"), + (b"..", "invalid_logical_path"), + (b"a/b", "invalid_logical_path"), + (b"a\nb", "invalid_logical_path"), + (b"a\tb", "invalid_logical_path"), + (b"a\x7fb", "invalid_logical_path"), + (b"a\xc2\x80b", "invalid_logical_path"), + (b"\xff", "unsupported_path_encoding"), + (b"a" * 256, "path_limit"), + ) + for name, code in cases: with self.subTest(name=name): tree = self.tree(root, [(b"100644", name, blob)]) - with self.assertRaisesRegex(ReaderRefused, "^unsupported_path$"): + with self.assertRaisesRegex(ReaderRefused, f"^{code}$"): read_snapshot(root, self.commit(root, tree)) + + def test_case_and_normalization_variants_remain_distinct(self) -> None: + root, _, blob = self.fixture() + names = ("A", "a", "é", "e\u0301", "ß", "SS", "İ", "i") + tree = self.tree( + root, + [(b"100644", name.encode("utf-8"), blob) for name in names], + ) + result = read_snapshot(root, self.commit(root, tree)) + self.assertEqual({item.path for item in result.files}, set(names)) + + def test_exact_duplicate_and_file_directory_conflicts_are_refused(self) -> None: + root, _, blob = self.fixture() child = self.tree(root, [(b"100644", b"child", blob)]) - tree = self.tree(root, [(b"40000", b"Dir", child), - (b"100644", b"dir", blob)]) - with self.assertRaisesRegex(ReaderRefused, "^path_collision$"): - read_snapshot(root, self.commit(root, tree)) + trees = ( + self.tree(root, [(b"100644", b"same", blob), + (b"100644", b"same", blob)]), + self.tree(root, [(b"40000", b"same", child), + (b"100644", b"same", blob)]), + ) + for tree in trees: + with ( + self.subTest(tree=tree), + self.assertRaisesRegex(ReaderRefused, "^path_collision$"), + ): + read_snapshot(root, self.commit(root, tree)) def test_limits_refuse_instead_of_returning_a_partial_snapshot(self) -> None: root, head, blob = self.fixture()