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
159 changes: 125 additions & 34 deletions src/specify_cli/_version.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,20 +17,22 @@
import re
import shlex
import shutil
import stat
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from pathlib import Path, PurePath
from typing import Literal, overload

import typer
from packaging.version import InvalidVersion, Version
from ._download_security import MAX_JSON_METADATA_BYTES, read_response_limited
from ._console import console
from .file_helper import _strip_extended_length_prefix

GITHUB_API_LATEST = "https://api.github.com/repos/github/spec-kit/releases/latest"
_RESOLUTION_FAILURE_OFFLINE = "offline or timeout"
Expand All @@ -51,6 +53,7 @@
)
_TIER3_REGISTRY_TIMEOUT_SECS = 5
_VERIFY_TIMEOUT_SECS = 10
_WINDOWS_MOUNT_POINT_TAG = 0xA0000003 # IO_REPARSE_TAG_MOUNT_POINT


def _get_installed_version() -> str:
Expand Down Expand Up @@ -367,17 +370,13 @@ def _expand_prefix(prefix: str) -> Path | None:
return None


def _path_is_within_prefix(path: Path, prefix: Path) -> bool:
def _path_is_within_prefix(path: PurePath, prefix: PurePath) -> bool:
"""Return whether absolute `path` is under absolute `prefix`."""
if not path.is_absolute() or not prefix.is_absolute():
return False
try:
common = os.path.commonpath(
[os.path.normcase(str(path)), os.path.normcase(str(prefix))]
)
except ValueError:
return False
return common == os.path.normcase(str(prefix))
return _strip_extended_length_prefix(path).is_relative_to(
_strip_extended_length_prefix(prefix)
)


def _resolve_path_or_original(path: Path) -> Path:
Expand Down Expand Up @@ -490,37 +489,129 @@ def _editable_marker_seen() -> bool:


def _find_installer(name: str) -> str | None:
"""Resolve an installer without trusting working-directory executables."""
try:
resolved = shutil.which(name)
if resolved is None:
return None
installer = Path(resolved)
# Windows lookup may prepend cwd even when PATH is absolute. Check
# both the lookup entry and symlink target using filesystem identity.
if not installer.is_absolute():
raise _InstallerRejected(
"""Find an installer outside the working directory and current Git tree."""
context: tuple[os.stat_result, os.stat_result | None] | None = None
rejections: list[str] = []

def rejected_lookup(resolved: str) -> str:
if not Path(resolved).is_absolute():
return (
f"Installer {name} rejected: lookup returned a relative path {resolved!r}. "
"Use an absolute PATH entry for the installed tool. "
"Run from a directory without a matching installer executable "
"to avoid implicit current-directory matches on Windows, and retry."
)
cwd = Path.cwd()
if (
installer.parent.samefile(cwd)
or installer.resolve(strict=True).parent.samefile(cwd)
):
raise _InstallerRejected(
f"Installer {name} rejected: lookup or target is in the current "
f"working directory ({resolved!r}). Run from another directory "
"and ensure PATH points to the installed tool."
return (
f"Installer {name} rejected: lookup or target is not trusted ({resolved!r}). "
"Run from another directory and ensure PATH points to the installed tool."
)

def eligible(path: Path, *, executable: bool) -> bool:
nonlocal context
if context is None:
cwd = Path.cwd().resolve()
project_root = _git_ancestor(cwd)
context = (
cwd.stat(),
project_root.stat() if project_root is not None else None,
)
cwd_status, project_status = context
# Check that traversal terminates before inspecting intermediate links.
path.resolve(strict=True)
if not executable and os.path.samestat(path.stat(), cwd_status):
return False
pending = [(path, executable)]
seen: set[tuple[Path, bool]] = set()
while pending:
current, leaf_executable = pending.pop()
if leaf_executable and os.path.samestat(current.parent.stat(), cwd_status):
return False
for ancestor in (current, *current.parents):
is_executable = leaf_executable and ancestor == current
key = (ancestor, is_executable)
if key in seen:
continue
seen.add(key)
entry_status = ancestor.lstat()
is_link = stat.S_ISLNK(entry_status.st_mode)
is_reparse = bool(
getattr(entry_status, "st_file_attributes", 0)
& stat.FILE_ATTRIBUTE_REPARSE_POINT
)
is_junction = (
is_reparse
and stat.S_ISDIR(entry_status.st_mode)
and getattr(entry_status, "st_reparse_tag", 0)
== _WINDOWS_MOUNT_POINT_TAG
)
if is_reparse and not (is_link or is_junction):
return False
Comment thread
Copilot marked this conversation as resolved.
if project_status is not None and os.path.samestat(
entry_status, project_status
):
return False
if is_link or is_junction:
if os.path.samestat(ancestor.parent.stat(), cwd_status):
return False
Comment on lines +554 to +555
target = ancestor.readlink()
pending.append(
(
target
if target.is_absolute()
else ancestor.parent / target,
is_executable,
)
)
return True

suffixes = [""]
if sys.platform == "win32":
# Python 3.11's which() does not expand PATHEXT for absolute commands.
suffixes = [
suffix
for suffix in (os.environ.get("PATHEXT") or ".COM;.EXE;.BAT;.CMD").split(
";"
)
except (OSError, RuntimeError) as exc:
raise _InstallerRejected(
f"Installer {name} rejected: could not validate its path ({exc}). "
"Run from an existing, accessible working directory and check PATH."
) from exc
return resolved
if suffix
]

for entry in os.get_exec_path():
if sys.platform == "win32":
entry = entry.strip('"')
directory = Path(entry)
if not directory.is_absolute():
if entry:
for suffix in suffixes:
candidate = str(directory / f"{name}{suffix}")
if os.path.lexists(candidate):
rejections.append(rejected_lookup(candidate))
Comment on lines +584 to +587
continue
try:
for suffix in suffixes:
found = shutil.which(str(directory / f"{name}{suffix}"))
Comment on lines +590 to +591
if found is not None:
executable = Path(found)
if (
executable.is_absolute()
and eligible(directory, executable=False)
and eligible(executable, executable=True)
):
return str(executable)
rejections.append(rejected_lookup(found))
except (OSError, RuntimeError) as exc:
rejections.append(
f"Installer {name} rejected: could not validate its path ({exc}). "
"Run from an existing, accessible working directory and check PATH."
)
console.print(
f"Skipping installer PATH entry {directory} while finding {name}: {exc}",
style="yellow",
markup=False,
soft_wrap=True,
)
if rejections:
raise _InstallerRejected(rejections[0])
return None


@overload
Expand Down
13 changes: 13 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -176,6 +176,19 @@ def clean_environ(monkeypatch):
monkeypatch.delenv("GITHUB_TOKEN", raising=False)


@pytest.fixture
def installer_paths(tmp_path):
bin_dir = tmp_path / "installer-bin"
bin_dir.mkdir()
paths = {}
for name in ("uv", "pipx"):
executable = bin_dir / (f"{name}.exe" if os.name == "nt" else name)
executable.write_bytes(b"installer fixture; never executed\n")
executable.chmod(0o755)
paths[name] = str(executable)
return paths


def _fake_self_upgrade_argv0(monkeypatch, tmp_path, env_name, path_parts):
"""Create a fake executable under tmp_path and point sys.argv[0] at it."""
monkeypatch.setenv(env_name, str(tmp_path))
Expand Down
13 changes: 13 additions & 0 deletions tests/specify_cli/self_upgrade_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@

from specify_cli._version import (
_InstallMethod,
_DetectionSignals,
_UpgradePlan,
_assemble_installer_argv,
_detect_install_method,
Expand All @@ -28,6 +29,7 @@
"_assemble_installer_argv",
"_completed_process",
"_detect_install_method",
"_detection_result",
"_verify_upgrade",
"mock_urlopen_response",
"requires_posix",
Expand All @@ -53,6 +55,17 @@
SENTINEL_GITHUB_TOKEN = "SENTINEL-GITHUB-TOKEN-VALUE"


def _detection_result(method: _InstallMethod) -> tuple[_InstallMethod, _DetectionSignals]:
return method, _DetectionSignals(
sys_argv0="",
matched_tier=None,
matched_prefix=None,
editable_marker_seen=False,
installer_registries_consulted=(),
resolved_method=method,
)


def _completed_process(
returncode: int, stdout: str = "", stderr: str = ""
) -> subprocess.CompletedProcess:
Expand Down
Loading
Loading