Repository navigation
Expand file tree
/
Copy pathpath_utils.py
More file actions
230 lines (195 loc) · 8.46 KB
/
Copy pathpath_utils.py
File metadata and controls
230 lines (195 loc) · 8.46 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
import hashlib
import re
from collections.abc import Iterable, Mapping
from functools import lru_cache
from pathlib import Path
from pathspec import PathSpec
from .. import constants as cs
_PROJECT_NAME_INVALID_CHARS = re.compile(r"[^A-Za-z0-9_-]+")
_PROJECT_NAME_DIGEST_LEN = 8
_PROJECT_NAME_FALLBACK_BASE = "repo"
def derive_project_name(repo_path: Path) -> str:
resolved = repo_path.resolve()
digest = hashlib.sha256(str(resolved).encode("utf-8")).hexdigest()[
:_PROJECT_NAME_DIGEST_LEN
]
base = _PROJECT_NAME_INVALID_CHARS.sub("_", resolved.name).strip("_")
if not base:
base = _PROJECT_NAME_FALLBACK_BASE
return f"{base}__{digest}"
def resolve_repo_path(repo_path: str | None, target_default: str) -> Path:
if repo_path:
return Path(repo_path).resolve()
if target_default and target_default != ".":
return Path(target_default).resolve()
return Path.cwd().resolve()
@lru_cache(maxsize=4096)
def cached_relative_path(file_path: Path, repo_path: Path) -> Path:
return file_path.relative_to(repo_path)
@lru_cache(maxsize=4096)
def cached_resolve_posix(file_path: Path) -> str:
return file_path.resolve().as_posix()
@lru_cache(maxsize=4096)
def cached_file_identity_posix(file_path: Path) -> str:
"""Absolute POSIX identity that survives the file's own deletion.
``resolve()`` dereferences a leaf symlink to its target, but once the link
is deleted that target can no longer be recovered, so a node keyed on it can
never be matched for deletion and leaks (GHSA-85gg aside, issue #1154).
Resolving only the parent (which outlives the leaf) and keeping the leaf
name yields a key that ingestion and deletion agree on, and one that stays
inside the repository for a link whose target points outside it.
"""
return (file_path.parent.resolve() / file_path.name).as_posix()
# #495: .cgrignore lines and --exclude values are interpreted with
# .gitignore (gitwildmatch) semantics: bare names match at any depth (as
# before), and globs / anchoring / dir-only trailing slash now work. The
# spec is compiled once per pattern set (frozensets are hashable).
@lru_cache(maxsize=64)
def compiled_ignore_spec(patterns: frozenset[str]) -> PathSpec:
return PathSpec.from_lines(cs.GITWILDMATCH_STYLE, sorted(patterns))
def matches_ignore_patterns(rel_path_str: str, patterns: frozenset[str]) -> bool:
return compiled_ignore_spec(patterns).match_file(rel_path_str)
_GLOB_MAGIC = re.compile(r"[*?\[]")
def unignore_could_match_within(pattern: str, rel_dir: str) -> bool:
# Dir-pruning guard: keep a pruned-by-default directory when an
# unignore pattern could match it or anything beneath it.
if "/" not in pattern.rstrip("/"):
# slash-free patterns are unanchored: they can match at any depth.
return True
head, *glob_rest = _GLOB_MAGIC.split(pattern, 1)
if glob_rest:
# the glob may complete the trailing segment; keep whole segments.
head = head.rsplit("/", 1)[0]
head = head.strip("/")
return (
not head
or head == rel_dir
or head.startswith(f"{rel_dir}/")
or rel_dir.startswith(f"{head}/")
)
def should_keep_dir(
dirname: str,
dir_prefix: str,
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
) -> bool:
"""Whether a repository walk descends into this directory.
The one predicate every whole-repo walk prunes with, so a sweep that
reads files the indexer skipped (or skips files the indexer holds)
cannot happen by construction (issue #1088).
"""
rel_dir = f"{dir_prefix}{dirname}"
# an explicit exclude can never be rescued by unignore (excludes win
# at the file level too), so prune the subtree outright.
if exclude_paths and matches_ignore_patterns(f"{rel_dir}/", exclude_paths):
return False
if dirname not in cs.IGNORE_PATTERNS:
return True
# Cargo's src/bin/ holds first-party binaries, not build output;
# mirrors has_ignored_dir_part.
if (
dirname == cs.DIR_BIN
and dir_prefix.rstrip(cs.SEPARATOR_SLASH).rsplit(cs.SEPARATOR_SLASH, 1)[-1]
== cs.DIR_SRC
):
return True
return bool(
unignore_paths
and any(unignore_could_match_within(u, rel_dir) for u in unignore_paths)
)
def should_skip_path(
path: Path,
repo_path: Path,
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
is_file: bool | None = None,
) -> bool:
_is_file = path.is_file() if is_file is None else is_file
if _is_file and path.suffix in cs.IGNORE_SUFFIXES:
return True
# Containment below is lexical, so a symlink whose target escapes the root
# would pass it and let a repo-scoped sweep read or overwrite outside files
# (GHSA-85gg-2gfq-q95m). Resolve first, mirroring validate_project_path
# (decorators.py) and absolute_path_within_project_root (this module).
try:
path.resolve().relative_to(repo_path.resolve())
except ValueError:
return True
rel_path = cached_relative_path(path, repo_path)
rel_path_str = rel_path.as_posix()
# a trailing slash marks the path as a directory for dir-only patterns.
match_path = rel_path_str if _is_file else f"{rel_path_str}/"
if exclude_paths and matches_ignore_patterns(match_path, exclude_paths):
return True
# unignore rescues only built-in ignores, never explicit user excludes.
if unignore_paths and matches_ignore_patterns(match_path, unignore_paths):
return False
if (
not _is_file
and unignore_paths
and any(unignore_could_match_within(u, rel_path_str) for u in unignore_paths)
):
# structure traversal must descend into a built-in-ignored dir when
# an unignore pattern can match beneath it (mirrors _should_keep_dir),
# or rescued files get no Folder/Package ancestry in the graph.
return False
dir_parts = rel_path.parent.parts if _is_file else rel_path.parts
return has_ignored_dir_part(dir_parts)
def has_ignored_dir_part(dir_parts: tuple[str, ...]) -> bool:
# `bin` is a build-output ignore (dotnet's <proj>/bin, repo-root bin/)
# EXCEPT directly under src/: Cargo's multi-binary layout puts
# first-party binaries in src/bin/, where build systems never emit.
for index, part in enumerate(dir_parts):
if part not in cs.IGNORE_PATTERNS:
continue
if part == cs.DIR_BIN and index > 0 and dir_parts[index - 1] == cs.DIR_SRC:
continue
return True
return False
def should_skip_rel_file(
rel_path_str: str,
dir_parts: tuple[str, ...],
suffix: str,
exclude_paths: frozenset[str] | None = None,
unignore_paths: frozenset[str] | None = None,
) -> bool:
if suffix in cs.IGNORE_SUFFIXES:
return True
if exclude_paths and matches_ignore_patterns(rel_path_str, exclude_paths):
return True
# unignore rescues only built-in ignores, never explicit user excludes.
if unignore_paths and matches_ignore_patterns(rel_path_str, unignore_paths):
return False
return has_ignored_dir_part(dir_parts)
def project_roots_from_rows(
rows: Iterable[Mapping[str, object]],
) -> dict[str, str | None]:
"""Build {project_name: root_path} from CYPHER_LIST_PROJECTS rows."""
roots: dict[str, str | None] = {}
for row in rows:
name = row.get("name")
if not isinstance(name, str):
continue
root = row.get("root_path")
roots[name] = root if isinstance(root, str) else None
return roots
def absolute_path_within_project_root(
qualified_name: str, absolute_path: str, roots: dict[str, str | None]
) -> bool:
"""A stored absolute path may only be read from inside its own project's
indexed root; projects with no recorded root (legacy graphs) stay
readable (issue #425). Project names may contain dots, so the owning
project is the longest known name prefixing the qualified name. The
resolve() calls are load-bearing: containment is checked lexically, so
an unresolved ``..`` segment or symlink would escape the root."""
matches = [name for name in roots if qualified_name.startswith(name + ".")]
if not matches:
return True
owner = matches[0]
for name in matches[1:]:
if len(name) > len(owner):
owner = name
root = roots[owner]
if root is None:
return True
return Path(absolute_path).resolve().is_relative_to(Path(root).resolve())