Skip to content
Open
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
7 changes: 5 additions & 2 deletions src/specify_cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -214,10 +214,12 @@ def _install_shared_infra_or_exit(
raise typer.Exit(1)


def ensure_executable_scripts(project_path: Path, tracker: StepTracker | None = None) -> None:
def ensure_executable_scripts(
project_path: Path, tracker: StepTracker | None = None
) -> list[str]:
"""Ensure POSIX .sh scripts under .specify/scripts and .specify/extensions (recursively) have execute bits (no-op on Windows)."""
if os.name == "nt":
return # Windows: skip silently
return [] # Windows: skip silently
scan_roots = [
project_path / ".specify" / "scripts",
project_path / ".specify" / "extensions",
Expand Down Expand Up @@ -265,6 +267,7 @@ def ensure_executable_scripts(project_path: Path, tracker: StepTracker | None =
console.print("[yellow]Some scripts could not be updated:[/yellow]")
for f in failures:
console.print(f" - {f}")
return failures

# ---------------------------------------------------------------------------
# Skills directory helpers
Expand Down
35 changes: 35 additions & 0 deletions src/specify_cli/_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,41 @@
DOCKER_AGENT_CHECK_TIMEOUT = 5


def windows_path_is_junction(path: Path) -> bool:
"""Detect a Windows junction using Python 3.11-compatible reparse data."""
try:
path_stat = path.lstat()
except FileNotFoundError:
return False
except OSError as exc:
raise ValueError(
f"Cannot determine whether path is a junction: {path}: {exc}"
) from exc

reparse_tag = getattr(path_stat, "st_reparse_tag", None)
if reparse_tag is None:
raise ValueError(f"Cannot determine whether path is a junction: {path}")
mount_point_tag = getattr(stat, "IO_REPARSE_TAG_MOUNT_POINT", 0xA0000003)
return reparse_tag == mount_point_tag


def path_is_junction(path: Path) -> bool:
"""Return whether *path* is a junction on supported Python versions."""
checker = getattr(path, "is_junction", None)
if callable(checker):
try:
return checker()
except AttributeError:
pass
except OSError as exc:
raise ValueError(
f"Cannot determine whether path is a junction: {path}: {exc}"
) from exc
if os.name == "nt":
return windows_path_is_junction(path)
return False


def docker_agent_command(executable: str | None = None) -> list[str] | None:
"""Return a runnable Docker Agent command, or ``None`` if unavailable.

Expand Down
135 changes: 106 additions & 29 deletions src/specify_cli/agents.py
Original file line number Diff line number Diff line change
Expand Up @@ -557,6 +557,34 @@ def _compute_output_name(

return f"speckit-{short_name}"

def validate_integration_output_paths(
self,
agent_name: str,
command_names: Iterable[str],
project_root: Path,
) -> None:
"""Run the integration-owned path guard before any output is written."""
self._ensure_configs()
agent_config = self.AGENT_CONFIGS.get(agent_name)
if not agent_config:
return

from specify_cli.integrations import get_integration # noqa: PLC0415

integration = get_integration(agent_name)
if integration is None:
return
output_root = self._resolve_agent_dir(
agent_name, agent_config, project_root
)
integration.validate_output_path(output_root, project_root)
for command_name in command_names:
output_name = self._compute_output_name(
agent_name, command_name, agent_config
)
output_path = output_root / f"{output_name}{agent_config['extension']}"
integration.validate_output_path(output_path, project_root)

@staticmethod
def _ensure_inside(candidate: Path, base: Path) -> None:
"""Validate that a write target stays within the expected base directory.
Expand Down Expand Up @@ -651,6 +679,11 @@ def register_commands(
commands_dir = _resolved_dir or self._resolve_agent_dir(
agent_name, agent_config, project_root,
)
from specify_cli.integrations import get_integration # noqa: PLC0415

_integration = get_integration(agent_name)
if _integration is not None:
_integration.validate_output_path(commands_dir, project_root)
commands_dir.mkdir(parents=True, exist_ok=True)

registered = []
Expand All @@ -677,14 +710,8 @@ def register_commands(
# ``.bob/commands``.
_sep = agent_config.get("invoke_separator", ".")
registrar_writes_skills = agent_config.get("extension") == "/SKILL.md"
try:
from specify_cli.integrations import get_integration # noqa: PLC0415

_integ = get_integration(agent_name)
if _integ is not None:
_sep = _integ.invoke_separator_for_mode(registrar_writes_skills)
except (ImportError, ValueError, KeyError):
pass
if _integration is not None:
_sep = _integration.invoke_separator_for_mode(registrar_writes_skills)
_prefix = get_invocation_prefix(agent_name, registrar_writes_skills)

for cmd_info in commands:
Expand Down Expand Up @@ -835,18 +862,14 @@ def register_commands(
raise ValueError(f"Unsupported format: {agent_config['format']}")

# -- Post-process for non-skills agents -----------------------
_integration = None
if agent_config["extension"] != "/SKILL.md":
from specify_cli.integrations import ( # noqa: PLC0415
get_integration,
)

_integration = get_integration(agent_name)
if _integration is not None:
output = _integration.post_process_command_content(output)

dest_file = commands_dir / f"{output_name}{agent_config['extension']}"
self._ensure_inside(dest_file, commands_dir)
if _integration is not None:
_integration.validate_output_path(dest_file, project_root)
dest_file.parent.mkdir(parents=True, exist_ok=True)
self._write_registered_output(
dest_file,
Expand Down Expand Up @@ -927,6 +950,8 @@ def register_commands(
commands_dir / f"{alias_output_name}{agent_config['extension']}"
)
self._ensure_inside(alias_file, commands_dir)
if _integration is not None:
_integration.validate_output_path(alias_file, project_root)
alias_file.parent.mkdir(parents=True, exist_ok=True)
self._write_registered_output(
alias_file,
Expand Down Expand Up @@ -1084,6 +1109,9 @@ def register_commands_for_all_agents(
Dictionary mapping agent names to list of registered commands
"""
results = {}
from specify_cli.integrations.base import ( # noqa: PLC0415
IntegrationOutputPathError,
)

self._ensure_configs()
active_skills_agent = (
Expand Down Expand Up @@ -1191,6 +1219,8 @@ def register_commands_for_all_agents(
active_created_skills_dir = (
recovered_active_skills_dir or agent_dir
)
except IntegrationOutputPathError:
raise
except ValueError:
continue
except OSError:
Expand Down Expand Up @@ -1241,6 +1271,9 @@ def register_commands_for_non_skill_agents(
Dictionary mapping agent names to list of registered commands
"""
results = {}
from specify_cli.integrations.base import ( # noqa: PLC0415
IntegrationOutputPathError,
)
self._ensure_configs()
extra_agents_set = frozenset(extra_agents) if extra_agents else frozenset()
for agent_name, agent_config in self.AGENT_CONFIGS.items():
Expand Down Expand Up @@ -1275,6 +1308,8 @@ def register_commands_for_non_skill_agents(
)
if registered:
results[agent_name] = registered
except IntegrationOutputPathError:
raise
except ValueError:
continue
return results
Expand All @@ -1294,6 +1329,16 @@ def unregister_commands(
project_root: Path to project root
"""
self._ensure_configs()
from specify_cli.integrations import get_integration # noqa: PLC0415
from specify_cli.integrations.base import ( # noqa: PLC0415
IntegrationOutputPathError,
)
from specify_cli.shared_infra import ( # noqa: PLC0415
_ensure_safe_shared_destination,
)

cleanup_files: list[tuple[Path, Path]] = []
copilot_prompts: set[Path] = set()
for agent_name, cmd_names in registered_commands.items():
if agent_name not in self.AGENT_CONFIGS:
continue
Expand All @@ -1302,6 +1347,9 @@ def unregister_commands(
commands_dir = self._resolve_agent_dir(
agent_name, agent_config, project_root,
)
integration = get_integration(agent_name)
if integration is not None:
integration.validate_output_path(commands_dir, project_root)

# Collect all directories to clean: canonical (or resolved
# legacy) plus the legacy dir if it exists separately.
Expand All @@ -1313,6 +1361,10 @@ def unregister_commands(
dirs_to_clean.append(legacy_dir)

for cmd_name in cmd_names:
if not self._is_safe_command_name(cmd_name):
raise IntegrationOutputPathError(
f"Unsafe registered command name: {cmd_name!r}"
)
output_name = self._compute_output_name(
agent_name, cmd_name, agent_config
)
Expand All @@ -1330,25 +1382,50 @@ def unregister_commands(
self._ensure_inside(cmd_file, target_dir)
except ValueError:
continue
if cmd_file.exists() or cmd_file.is_symlink():
cmd_file.unlink()
# For SKILL.md agents each command lives in its own
# subdirectory (e.g. .agents/skills/speckit-ext-cmd/
# SKILL.md). Remove the parent dir when it becomes
# empty to avoid orphaned directories.
parent = cmd_file.parent
if parent != target_dir and parent.exists():
try:
parent.rmdir()
except OSError:
pass
if integration is not None:
integration.validate_output_path(
cmd_file, project_root
)
cleanup_files.append((cmd_file, target_dir))

if agent_name == "copilot":
prompt_file = (
project_root / ".github" / "prompts" / f"{cmd_name}.prompt.md"
project_root
/ ".github"
/ "prompts"
/ f"{cmd_name}.prompt.md"
)
if prompt_file.exists():
prompt_file.unlink()
try:
_ensure_safe_shared_destination(
project_root,
prompt_file,
parent_must_exist=False,
)
except ValueError as exc:
raise IntegrationOutputPathError(str(exc)) from exc
copilot_prompts.add(prompt_file)

seen: set[Path] = set()
for cmd_file, target_dir in cleanup_files:
if cmd_file in seen:
continue
seen.add(cmd_file)
if cmd_file.exists() or cmd_file.is_symlink():
cmd_file.unlink()
# For SKILL.md agents each command lives in its own
# subdirectory (e.g. .agents/skills/speckit-ext-cmd/
# SKILL.md). Remove the parent dir when it becomes
# empty to avoid orphaned directories.
parent = cmd_file.parent
if parent != target_dir and parent.exists():
try:
parent.rmdir()
except OSError:
pass

for prompt_file in copilot_prompts:
if prompt_file.exists():
prompt_file.unlink()


# Populate AGENT_CONFIGS after class definition.
Expand Down
2 changes: 2 additions & 0 deletions src/specify_cli/commands/bundle/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,8 @@ def _run_init(integration: str, *, script_type: str, offline: bool = False) -> N
integration_options=None,
extensions=None,
trust_extension_urls=False,
dry_run=False,
json_output=False,
)
except typer.Exit as exc:
if exc.exit_code:
Expand Down
Loading