Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 30 additions & 2 deletions diffgraph/git_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -335,7 +335,9 @@ def _resolve_untracked(
for raw_path in output.split(b"\0"):
if not raw_path:
continue
path = os.fsdecode(raw_path)
path = _decode_git_path(raw_path, warnings)
if path is None:
continue
full_path = os.path.join(root, path)
try:
file_stat = os.lstat(full_path)
Expand Down Expand Up @@ -519,9 +521,19 @@ def _parse_raw(data: bytes, warnings: List[ResolutionWarning]) -> List[_RawEntry
path_count = 2 if status in ("R", "C") else 1
if not status or index + path_count > len(fields):
raise ValueError("malformed raw-diff path fields")
paths = [os.fsdecode(value) for value in fields[index : index + path_count]]
paths = [
_decode_git_path(value, warnings)
for value in fields[index : index + path_count]
]
index += path_count

# Git permits arbitrary bytes in a pathname, but artifact paths
# must be valid Unicode strings. Skip only this record while
# retaining its NUL-delimited boundaries, so a single invalid
# filename cannot poison the rest of the snapshot.
if any(path is None for path in paths):
continue

if status in ("R", "C"):
old_path, new_path = paths
elif status == "A":
Expand Down Expand Up @@ -552,6 +564,22 @@ def _parse_raw(data: bytes, warnings: List[ResolutionWarning]) -> List[_RawEntry
return parsed


def _decode_git_path(
value: bytes, warnings: List[ResolutionWarning]
) -> Optional[str]:
"""Decode a Git pathname without leaking surrogate escapes into artifacts."""

try:
return value.decode("utf-8")
except UnicodeDecodeError:
warnings.append(ResolutionWarning(
"undecodable_path",
"Git reported a path that is not valid UTF-8; the entry was skipped",
value.decode("utf-8", "backslashreplace"),
))
return None


def _mode(value: bytes) -> Optional[str]:
text = value.decode("ascii")
return None if not text or set(text) == {"0"} else text
Expand Down
1 change: 1 addition & 0 deletions diffgraph/schema/diffgraph-v2.schema.json
Original file line number Diff line number Diff line change
Expand Up @@ -396,6 +396,7 @@
"not_a_git_repository",
"git_diff_failed",
"malformed_git_output",
"undecodable_path",
"missing_object_id",
"unsupported_worktree_entry",
"worktree_read_failed",
Expand Down
1 change: 1 addition & 0 deletions diffgraph/structural.py
Original file line number Diff line number Diff line change
Expand Up @@ -359,6 +359,7 @@ def _resolution_warning(item: ResolutionWarning) -> Dict[str, str]:
"not_a_git_repository",
"git_diff_failed",
"malformed_git_output",
"undecodable_path",
"missing_object_id",
"unsupported_worktree_entry",
"worktree_read_failed",
Expand Down
16 changes: 16 additions & 0 deletions tests/test_git_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,22 @@ def test_nul_parsing_preserves_tabs_and_newlines_in_paths(tmp_path):
assert entry.old_oid == entry.new_oid == oid(repo, "HEAD:" + old_name)


def test_undecodable_untracked_path_is_a_scoped_warning(tmp_path):
"""One non-UTF-8 path must not leak surrogate escapes or hide valid files."""
repo = make_repo(tmp_path)
write(repo, "good.py", b"value = 1\n")
raw_path = os.fsencode(repo) + b"/bad-\xff.py"
with open(raw_path, "wb") as handle:
handle.write(b"value = 2\n")

result = resolve_unstaged(str(repo))

assert [entry.new_path for entry in result.entries] == ["good.py"]
assert [(warning.code, warning.path) for warning in result.warnings] == [
("undecodable_path", "bad-\\xff.py")
]


def test_repeated_resolution_is_deterministic(tmp_path):
repo = make_repo(tmp_path)
for name in ("z.txt", "a.txt", "middle.txt"):
Expand Down
19 changes: 19 additions & 0 deletions tests/test_structural.py
Original file line number Diff line number Diff line change
Expand Up @@ -873,6 +873,25 @@ def test_resolution_warning_preserves_machine_readable_code(monkeypatch, tmp_pat
]


def test_undecodable_path_warning_is_schema_valid(monkeypatch, tmp_path):
root = repo(tmp_path)
warning = ResolutionWarning(
"undecodable_path", "Git reported a non-UTF-8 filename", None
)
monkeypatch.setattr(
"diffgraph.structural.resolve_unstaged",
lambda repository, pathspecs: SnapshotResolution((), (warning,)),
)

artifact = analyze_local_diff(str(root))

assert_valid(artifact)
assert artifact["metadata"]["warnings"] == [{
"code": "undecodable_path",
"detail": "undecodable_path: Git reported a non-UTF-8 filename",
}]


def test_unmerged_index_warning_preserves_machine_readable_code(monkeypatch, tmp_path):
root = repo(tmp_path)
warning = ResolutionWarning("unmerged_index_entry", "conflict remains", "app.py")
Expand Down
Loading