{{ message }}
-
Notifications
You must be signed in to change notification settings - Fork 12k
Expand file tree
/
Copy path_utils.py
More file actions
377 lines (322 loc) · 14.6 KB
/
Copy path_utils.py
File metadata and controls
377 lines (322 loc) · 14.6 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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
"""System utilities: subprocess, tool detection, file operations."""
from __future__ import annotations
import json
import json5
import os
import shutil
import stat
import subprocess
import tempfile
import yaml
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Any
from ._console import console
from ._download_security import normalize_zip_member_name
CLAUDE_LOCAL_PATH = Path.home() / ".claude" / "local" / "claude"
CLAUDE_NPM_LOCAL_PATH = Path.home() / ".claude" / "local" / "node_modules" / ".bin" / "claude"
DOCKER_AGENT_CHECK_TIMEOUT = 5
def docker_agent_command(executable: str | None = None) -> list[str] | None:
"""Return a runnable Docker Agent command, or ``None`` if unavailable.
Docker Agent is distributed either as the standalone ``docker-agent``
executable or as the ``docker agent`` Docker CLI plugin. The plugin form
is verified with a bounded, read-only version probe so a plain Docker CLI
is not mistaken for an installed Docker Agent.
"""
resolved_from_path = executable is None
if executable is None:
if shutil.which("docker-agent"):
return ["docker-agent", "run"]
executable = shutil.which("docker")
if executable is None:
return None
executable_name = Path(executable).name.lower()
if executable_name in {"docker", "docker.exe"}:
command = [executable, "agent", "version"]
run_command = [executable, "agent", "run"]
else:
# An explicit non-Docker executable is an operator override. Preserve
# the existing override contract without probing a custom binary.
return [executable, "run"]
try:
result = subprocess.run(
command,
capture_output=True,
check=False,
timeout=DOCKER_AGENT_CHECK_TIMEOUT,
)
except (OSError, subprocess.TimeoutExpired):
return None
if result.returncode != 0:
return None
return ["docker", "agent", "run"] if resolved_from_path else run_command
def relative_extension_path_violation(value: Any) -> str | None:
"""Return why ``value`` is unsafe as an extension-relative ``file`` path.
Single source of truth for the path-safety policy shared by
``ExtensionManifest._validate()`` (manifest-load validation) and
``CommandRegistrar.register_commands()`` (runtime guard), so the two cannot
drift. Returns a human-readable reason string when ``value`` is unsafe, or
``None`` when it is an acceptable relative path within the extension
directory.
Policy: the value must be a non-empty, portable file path with no
leading/trailing whitespace, absolute/anchored form, ``..`` traversal,
platform-reserved component, or directory-only suffix. The value is
evaluated under both POSIX and Windows path semantics because a native
``Path`` is OS-dependent (a ``PurePosixPath`` on POSIX does not interpret
Windows drive/UNC forms, and ``C:foo`` is anchored but not
``is_absolute()`` yet resolves against the CWD on its drive). Rejecting any
non-empty anchor covers POSIX-absolute (``/abs``), Windows drive-relative
(``C:foo``), Windows absolute (``C:\\foo``), and UNC/rooted forms.
"""
if not isinstance(value, str) or not value:
return "must be a non-empty string"
if value.strip() != value:
return "must not have leading or trailing whitespace"
if "\\" in value:
return "must use forward slashes as path separators"
posix_path = PurePosixPath(value)
win_path = PureWindowsPath(value)
if (
posix_path.anchor
or win_path.anchor
or ".." in posix_path.parts
or ".." in win_path.parts
):
return (
"must be a relative path within the extension directory "
"(no absolute paths, drive letters, or '..' segments)"
)
if value.endswith(("/", "\\")):
return "must name a file or command, not a directory"
try:
normalize_zip_member_name(value)
except ValueError:
return (
"must use portable path components "
"(no reserved names or platform-invalid characters)"
)
return None
def dump_frontmatter(data: dict[str, Any]) -> str:
"""Serialize skill/command frontmatter to a YAML string.
Centralizes the dump options used for SKILL.md frontmatter: ``allow_unicode``
preserves Unicode descriptions and ``sort_keys=False`` keeps key order, so no
call site can silently drop either.
"""
return yaml.safe_dump(data, sort_keys=False, allow_unicode=True).strip()
def run_command(
cmd: list[str],
check_return: bool = True,
capture: bool = False,
) -> str | None:
"""Run a command without invoking a shell and optionally capture output.
Commands are always executed with ``shell=False`` and must be passed as an
argv ``list[str]``. There is deliberately no ``shell`` parameter: the
argv-list contract makes shell interpolation impossible by construction, so
the shell-injection surface cannot be re-enabled at a call site.
"""
try:
if capture:
result = subprocess.run(cmd, check=check_return, capture_output=True, text=True)
return result.stdout.strip()
else:
subprocess.run(cmd, check=check_return)
return None
except subprocess.CalledProcessError as e:
if check_return:
console.print(f"[red]Error running command:[/red] {' '.join(cmd)}")
console.print(f"[red]Exit code:[/red] {e.returncode}")
if hasattr(e, 'stderr') and e.stderr:
console.print(f"[red]Error output:[/red] {e.stderr}")
raise
return None
def check_tool(tool: str, tracker=None) -> bool:
"""Check if a tool is installed. Optionally update tracker.
Args:
tool: Name of the tool to check
tracker: StepTracker | None to update with results
Returns:
True if tool is found, False otherwise
"""
# Special handling for Claude CLI local installs
# See: https://github.com/github/spec-kit/issues/123
# See: https://github.com/github/spec-kit/issues/550
# Claude Code can be installed in two local paths:
# 1. ~/.claude/local/claude (after `claude migrate-installer`)
# 2. ~/.claude/local/node_modules/.bin/claude (npm-local install, e.g. via nvm)
# Neither path may be on the system PATH, so we check them explicitly.
if tool == "claude":
if CLAUDE_LOCAL_PATH.is_file() or CLAUDE_NPM_LOCAL_PATH.is_file():
if tracker:
tracker.complete(tool, "available")
return True
# Per-integration executable resolution.
if tool == "kiro-cli":
# Kiro currently supports both executable names. Prefer kiro-cli and
# accept kiro as a compatibility fallback.
found = shutil.which("kiro-cli") is not None or shutil.which("kiro") is not None
elif tool == "rovodev":
found = shutil.which("acli") is not None
elif tool == "docker-agent":
found = docker_agent_command() is not None
else:
found = shutil.which(tool) is not None
if tracker:
if found:
tracker.complete(tool, "available")
else:
tracker.error(tool, "not found")
return found
def handle_vscode_settings(sub_item, dest_file, rel_path, verbose=False, tracker=None) -> None:
"""Handle merging or copying of .vscode/settings.json files.
Note: when merge produces changes, rewritten output is normalized JSON and
existing JSONC comments/trailing commas are not preserved.
"""
def log(message, color="green"):
if verbose and not tracker:
console.print(f"[{color}]{message}[/] {rel_path}")
def atomic_write_json(target_file: Path, payload: dict[str, Any]) -> None:
"""Atomically write JSON while preserving existing mode bits when possible."""
temp_path: Path | None = None
try:
with tempfile.NamedTemporaryFile(
mode='w',
encoding='utf-8',
dir=target_file.parent,
prefix=f"{target_file.name}.",
suffix=".tmp",
delete=False,
) as f:
temp_path = Path(f.name)
json.dump(payload, f, indent=4)
f.write('\n')
if target_file.exists():
try:
existing_stat = target_file.stat()
os.chmod(temp_path, stat.S_IMODE(existing_stat.st_mode))
if hasattr(os, "chown"):
try:
os.chown(temp_path, existing_stat.st_uid, existing_stat.st_gid)
except PermissionError:
# Best-effort owner/group preservation without requiring elevated privileges.
pass
except OSError:
# Best-effort metadata preservation; data safety is prioritized.
pass
os.replace(temp_path, target_file)
except Exception:
if temp_path:
temp_path.unlink(missing_ok=True)
raise
try:
with open(sub_item, 'r', encoding='utf-8') as f:
# json5 natively supports comments and trailing commas (JSONC)
new_settings = json5.load(f)
if dest_file.exists():
merged = merge_json_files(dest_file, new_settings, verbose=verbose and not tracker)
if merged is not None:
atomic_write_json(dest_file, merged)
log("Merged:", "green")
log("Note: comments/trailing commas are normalized when rewritten", "yellow")
else:
log("Skipped merge (preserved existing settings)", "yellow")
else:
shutil.copy2(sub_item, dest_file)
log("Copied (no existing settings.json):", "blue")
except (OSError, ValueError, KeyError) as e:
log(f"Warning: Could not merge settings: {e}", "yellow")
if not dest_file.exists():
shutil.copy2(sub_item, dest_file)
def merge_json_files(existing_path: Path, new_content: Any, verbose: bool = False) -> dict[str, Any] | None:
"""Merge new JSON content into existing JSON file.
Performs a polite deep merge where:
- New keys are added
- Existing keys are preserved (not overwritten) unless both values are dictionaries
- Nested dictionaries are merged recursively only when both sides are dictionaries
- Lists and other values are preserved from base if they exist
Args:
existing_path: Path to existing JSON file
new_content: New JSON content to merge in
verbose: Whether to print merge details
Returns:
Merged JSON content as dict, or None if the existing file should be left untouched.
"""
# Load existing content first to have a safe fallback
existing_content = None
exists = existing_path.exists()
if exists:
try:
with open(existing_path, 'r', encoding='utf-8') as f:
# Handle comments (JSONC) natively with json5
# Note: json5 handles BOM automatically
existing_content = json5.load(f)
except FileNotFoundError:
# Handle race condition where file is deleted after exists() check
exists = False
except (OSError, ValueError) as e:
if verbose:
console.print(f"[yellow]Warning: Could not read or parse existing JSON in {existing_path.name} ({e}).[/yellow]")
# Skip merge to preserve existing file if unparseable or inaccessible (e.g. PermissionError)
return None
# Validate template content
if not isinstance(new_content, dict):
if verbose:
console.print(f"[yellow]Warning: Template content for {existing_path.name} is not a dictionary. Preserving existing settings.[/yellow]")
return None
if not exists:
return new_content
# If existing content parsed but is not a dict, skip merge to avoid data loss
if not isinstance(existing_content, dict):
if verbose:
console.print(f"[yellow]Warning: Existing JSON in {existing_path.name} is not an object. Skipping merge to avoid data loss.[/yellow]")
return None
def deep_merge_polite(base: dict[str, Any], update: dict[str, Any]) -> dict[str, Any]:
"""Recursively merge update dict into base dict, preserving base values."""
result = base.copy()
for key, value in update.items():
if key not in result:
# Add new key
result[key] = value
elif isinstance(result[key], dict) and isinstance(value, dict):
# Recursively merge nested dictionaries
result[key] = deep_merge_polite(result[key], value)
else:
# Key already exists and values are not both dicts; preserve existing value.
# This ensures user settings aren't overwritten by template defaults.
pass
return result
merged = deep_merge_polite(existing_content, new_content)
# Detect if anything actually changed. If not, return None so the caller
# can skip rewriting the file (preserving user's comments/formatting).
if merged == existing_content:
return None
if verbose:
console.print(f"[cyan]Merged JSON file:[/cyan] {existing_path.name}")
return merged
def _display_project_path(project_root: Path, path: str | Path) -> str:
"""Return a stable POSIX-style display path for paths under a project."""
path_obj = Path(path)
try:
rel_path = path_obj.relative_to(project_root) if path_obj.is_absolute() else path_obj
except ValueError:
try:
rel_path = path_obj.resolve().relative_to(project_root.resolve())
except (OSError, ValueError):
return path_obj.as_posix()
return rel_path.as_posix()
def version_satisfies(current: str, required: str) -> bool:
"""Check if current version satisfies required version specifier.
Evaluates the version against the specifier using the project's
prerelease policy (prereleases are allowed).
Args:
current: Current version (e.g., "0.1.5")
required: Required version specifier (e.g., ">=0.1.0,<2.0.0")
Returns:
True if version satisfies requirement
"""
from packaging import version as pkg_version
from packaging.specifiers import InvalidSpecifier, SpecifierSet
try:
current_ver = pkg_version.Version(current)
specifier = SpecifierSet(required)
return specifier.contains(current_ver, prereleases=True)
except (pkg_version.InvalidVersion, InvalidSpecifier):
return False
You can’t perform that action at this time.