From 82bdef8c5dca1f7e550dd7243551c69acd295378 Mon Sep 17 00:00:00 2001 From: keshavp <32313895+keshprad@users.noreply.github.com> Date: Wed, 12 Aug 2026 23:03:09 -0700 Subject: [PATCH] release: SkillSpector 2.9.4 Signed-off-by: keshavp <32313895+keshprad@users.noreply.github.com> --- CHANGELOG.md | 12 + docs/release/skillspector-2.9.4.md | 54 ++++ pyproject.toml | 2 +- src/skillspector/cli.py | 9 +- src/skillspector/input_handler.py | 368 +++++++++++++++++++++--- src/skillspector/mcp_server.py | 6 +- src/skillspector/multi_skill.py | 32 ++- src/skillspector/nodes/build_context.py | 114 +++++++- src/skillspector/nodes/resolve_input.py | 4 +- tests/integration/test_graph_scanner.py | 16 +- tests/nodes/test_build_context.py | 156 +++++++++- tests/nodes/test_resolve_input.py | 17 ++ tests/test_multi_skill.py | 35 +++ tests/unit/test_cli.py | 26 ++ tests/unit/test_input_handler.py | 273 +++++++++++++++++- uv.lock | 2 +- 16 files changed, 1041 insertions(+), 85 deletions(-) create mode 100644 docs/release/skillspector-2.9.4.md diff --git a/CHANGELOG.md b/CHANGELOG.md index f65e2fd38..447d15585 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,15 @@ +### 2.9.4 (Wednesday, August 12, 2026) +### Features/Bug Fixes +* fix(mcp): reject local targets over HTTP transport (#196) +* Add Skill Inspector companion skill (#253) +* fix(lp3): remediation and docs name allowed-tools for SKILL.md (#316) +* chore(openssf-scorecard): Add badge (#351) +* Detect whitespace padding used to hide prompt-injection instructions (P9) (#24) +* fix(analyzers): HIGH SC8 when skill ships __pycache__ or .pyc (#357) +* Revert "Scope the locality guard to the namespace" +* Scope the locality guard to the namespace +* fix(security): reject symlinks in skill walk + disable git symlinks on clone +--- ### 2.9.3 (Tuesday, August 11, 2026) ### Features/Bug Fixes * fix(llm): surface invalid responses as degraded (skipped, non-fatal, incomplete) diff --git a/docs/release/skillspector-2.9.4.md b/docs/release/skillspector-2.9.4.md new file mode 100644 index 000000000..df8c66893 --- /dev/null +++ b/docs/release/skillspector-2.9.4.md @@ -0,0 +1,54 @@ +# SkillSpector v2.9.4 + +Released: 2026-08-12 + +## Summary + +This patch strengthens SkillSpector’s safe handling of MCP requests and untrusted skill content, while adding broader prompt-injection and supply-chain detection coverage. It also improves permission guidance, ships a companion Skill Inspector guide, and refreshes project documentation. + +## Highlights + +- HTTP-exposed MCP servers now reject caller-controlled local scan targets and local YARA-rule directories while preserving local scanning for trusted stdio use. +- Detect whitespace-padding prompt-injection attempts and shipped Python bytecode, with improved minimum risk scoring for high-impact findings. + +## Added + +- Add detection for whitespace padding used to hide prompt-injection instructions. +- Add a HIGH SC8 finding when a skill ships Python bytecode or `__pycache__` content. +- Add the Skill Inspector companion skill guide. + +## Changed + +- Treat `allowed-tools` as valid least-privilege permission guidance in remediations and documentation. +- Add an OpenSSF Scorecard badge to the project documentation. + +## Fixed + +- Reject local filesystem scan targets and local YARA-rule directories for HTTP MCP transport, preventing remote callers from selecting scanner-host paths. +- Reject symlinked skill content during discovery and disable Git symlink materialization when cloning input repositories. +- Ensure high-impact findings receive an appropriate minimum risk score. + +## Security + +- Harden HTTP MCP transport against local-path access and strengthen skill-content handling against symlink traversal. + +## Breaking Changes and Migration + +- HTTP MCP clients can no longer scan local filesystem paths or provide local YARA-rule directories. Use a remote repository or URL for HTTP requests; use trusted stdio transport for local scans. + +## Deprecations + +- None. + +## Validation + +- Internal GitLab merge-request CI passed lint, unit, integration, Docker smoke, and Sonar analysis for the six imported public changes. +- `uv run --locked --extra dev pytest -q tests/unit/test_mcp_server.py` — 26 passed for the HTTP MCP transport remediation. + +## Known Limitations + +- HTTP MCP transport intentionally rejects local filesystem inputs; this is a security boundary rather than an unsupported scanner capability. + +## References + +- `CHANGELOG.md` diff --git a/pyproject.toml b/pyproject.toml index 6522d4731..780ee3677 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "skillspector" -version = "2.9.3" +version = "2.9.4" description = "SkillSpector: Security scanner for AI agent skills (Claude Code, Cursor, and similar). Scans skills for vulnerabilities, malicious patterns, and security risks before installation. Supports Git repos, URLs, zips, and local directories; runs static pattern checks and optional LLM semantic analysis; outputs terminal, JSON, and Markdown reports with risk scoring." readme = "README.md" license = "Apache-2.0" diff --git a/src/skillspector/cli.py b/src/skillspector/cli.py index aa1ed6581..7afabd494 100644 --- a/src/skillspector/cli.py +++ b/src/skillspector/cli.py @@ -36,6 +36,7 @@ from skillspector.cleanup import cleanup_result from skillspector.constants import RISK_THRESHOLD from skillspector.graph import graph +from skillspector.input_handler import validate_local_input_path from skillspector.logging_config import get_logger, set_level from skillspector.mcp_registry import scan_registry from skillspector.multi_skill import MultiSkillDetectionResult, detect_skills @@ -323,7 +324,13 @@ def scan( if verbose: set_level("DEBUG") - resolved_path = Path(input_path).resolve() + resolved_path = Path(input_path) + if not input_path.startswith(("http://", "https://", "git@")): + try: + resolved_path = validate_local_input_path(resolved_path) + except ValueError as e: + console.print(f"[red]Error:[/red] {e}") + raise typer.Exit(code=2) from e if recursive and resolved_path.is_dir(): detection = detect_skills(resolved_path) if detection.is_multi_skill: diff --git a/src/skillspector/input_handler.py b/src/skillspector/input_handler.py index 125e6d922..e2bded1d7 100644 --- a/src/skillspector/input_handler.py +++ b/src/skillspector/input_handler.py @@ -36,13 +36,17 @@ from __future__ import annotations import ipaddress +import os import re import shutil import socket import subprocess import tempfile import zipfile +from errno import ELOOP, ENOENT, ENOTDIR from pathlib import Path +from stat import S_ISLNK, S_ISREG +from typing import BinaryIO, cast from urllib.parse import urlparse import httpx @@ -51,6 +55,9 @@ logger = get_logger(__name__) +_HAS_SECURE_DIR_FD = os.open in os.supports_dir_fd and hasattr(os, "O_NOFOLLOW") +_IS_WINDOWS = os.name == "nt" + ALLOWED_GIT_HOSTS = frozenset( { "github.com", @@ -112,6 +119,272 @@ def _is_private_ip(host: str) -> bool: return False +def _root_owned_root_alias(path: Path) -> Path | None: + """Return a root-owned symlink directly below ``/``, if *path* is one.""" + absolute_path = Path(os.path.abspath(path)) + if absolute_path.anchor != os.path.sep or len(absolute_path.parts) != 2: + return None + try: + path_stat = absolute_path.lstat() + except OSError: + return None + if S_ISLNK(path_stat.st_mode) and path_stat.st_uid == 0: + return absolute_path + return None + + +def _normalize_root_owned_alias(path: Path) -> Path: + """Resolve a trusted root-level system alias while retaining child path components.""" + absolute_path = Path(os.path.abspath(path)) + if absolute_path.anchor != os.path.sep or len(absolute_path.parts) < 3: + return absolute_path + root_alias = _root_owned_root_alias(Path(absolute_path.anchor, absolute_path.parts[1])) + if root_alias is None: + return absolute_path + try: + return root_alias.resolve(strict=True).joinpath(*absolute_path.parts[2:]) + except OSError: + return absolute_path + + +def _has_symlinked_parent(path: Path) -> bool: + """Return whether any parent of *path* is a symlink or Windows junction.""" + current = Path(path.anchor) + for part in path.parts[1:-1]: + current /= part + try: + if current.is_symlink() or current.is_junction(): + return True + except OSError: + return True + return False + + +class _UnsafeFileError(ValueError): + """Raised when a path cannot be safely treated as a regular file.""" + + +class _FileOpenError(ValueError): + """Raised when an otherwise safe file cannot be opened for operational reasons.""" + + def __init__(self, file_path: Path, cause: OSError) -> None: + super().__init__(f"Could not safely open file: {file_path}") + self.error_class = type(cause).__name__ + + +def validate_local_input_path(path: Path) -> Path: + """Normalize a local input path after rejecting symlinks and their ancestors.""" + if path.is_symlink() and _root_owned_root_alias(path) is None: + raise ValueError(f"Refusing to resolve a symlinked input: {path}") + if path.is_junction(): + raise ValueError(f"Refusing to resolve a junctioned input: {path}") + normalized_path = _normalize_root_owned_alias(path) + if _has_symlinked_parent(normalized_path): + raise ValueError(f"Refusing to resolve input with a symlinked parent: {path}") + return normalized_path + + +def _open_regular_file_no_follow(file_path: Path) -> BinaryIO: + """Open a regular file without following symlinks. + + Descriptor-relative opens protect every path component against replacement + races. Windows uses a reparse-point handle and validates the opened handle's + canonical path before exposing its contents. + """ + absolute_path = _normalize_root_owned_alias(file_path) + if _has_symlinked_parent(absolute_path): + raise _UnsafeFileError(f"Could not safely open file: {file_path}") + if _HAS_SECURE_DIR_FD: + return _open_regular_file_from_trusted_directory(absolute_path) + if _IS_WINDOWS: + return _open_regular_file_from_windows_handle(absolute_path) + raise _UnsafeFileError( + f"Secure no-follow file opens are unavailable on this platform: {file_path}" + ) + + +def _open_regular_file_from_trusted_directory(file_path: Path) -> BinaryIO: + """Open *file_path* one non-symlinked component at a time.""" + directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW + directory_fd: int | None = None + try: + directory_fd = os.open(file_path.anchor, directory_flags) + for part in file_path.parts[1:-1]: + next_directory_fd = os.open(part, directory_flags, dir_fd=directory_fd) + _close_fd_safely(directory_fd) + directory_fd = next_directory_fd + source_fd = os.open( + file_path.name, + os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_NONBLOCK", 0), + dir_fd=directory_fd, + ) + except FileNotFoundError: + raise FileNotFoundError(f"File not found: {file_path}") from None + except OSError as exc: + if exc.errno in {ELOOP, ENOTDIR}: + raise _UnsafeFileError(f"Could not safely open file: {file_path}") from exc + raise _FileOpenError(file_path, exc) from exc + finally: + if directory_fd is not None: + _close_fd_safely(directory_fd) + + return _fdopen_regular_file(source_fd, file_path) + + +def _open_regular_file_from_windows_handle(file_path: Path) -> BinaryIO: + """Open a regular Windows file without traversing a reparse point. + + ``FILE_FLAG_OPEN_REPARSE_POINT`` prevents a final symlink or junction from + being dereferenced. ``GetFinalPathNameByHandleW`` then detects an ancestor + that changed into a reparse point after the initial parent check, before the + opened handle can be used to read content. + """ + import ctypes + import msvcrt + from ctypes import wintypes + + class _ByHandleFileInformation(ctypes.Structure): + _fields_ = [ + ("dwFileAttributes", wintypes.DWORD), + ("ftCreationTime", wintypes.FILETIME), + ("ftLastAccessTime", wintypes.FILETIME), + ("ftLastWriteTime", wintypes.FILETIME), + ("dwVolumeSerialNumber", wintypes.DWORD), + ("nFileSizeHigh", wintypes.DWORD), + ("nFileSizeLow", wintypes.DWORD), + ("nNumberOfLinks", wintypes.DWORD), + ("nFileIndexHigh", wintypes.DWORD), + ("nFileIndexLow", wintypes.DWORD), + ] + + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) # type: ignore[attr-defined] + create_file = kernel32.CreateFileW + create_file.argtypes = [ + wintypes.LPCWSTR, + wintypes.DWORD, + wintypes.DWORD, + ctypes.c_void_p, + wintypes.DWORD, + wintypes.DWORD, + wintypes.HANDLE, + ] + create_file.restype = wintypes.HANDLE + get_file_information = kernel32.GetFileInformationByHandle + get_file_information.argtypes = [wintypes.HANDLE, ctypes.POINTER(_ByHandleFileInformation)] + get_file_information.restype = wintypes.BOOL + get_final_path_name = kernel32.GetFinalPathNameByHandleW + get_final_path_name.argtypes = [ + wintypes.HANDLE, + wintypes.LPWSTR, + wintypes.DWORD, + wintypes.DWORD, + ] + get_final_path_name.restype = wintypes.DWORD + close_handle = kernel32.CloseHandle + close_handle.argtypes = [wintypes.HANDLE] + close_handle.restype = wintypes.BOOL + + generic_read = 0x80000000 + file_share_all = 0x00000001 | 0x00000002 | 0x00000004 + open_existing = 3 + file_attribute_reparse_point = 0x00000400 + file_flag_open_reparse_point = 0x00200000 + invalid_handle_value = ctypes.c_void_p(-1).value + + handle = create_file( + os.fspath(file_path), + generic_read, + file_share_all, + None, + open_existing, + file_flag_open_reparse_point, + None, + ) + if handle == invalid_handle_value: + error = _windows_last_error() + if error.errno == ENOENT: + raise FileNotFoundError(f"File not found: {file_path}") from None + raise _FileOpenError(file_path, error) + + try: + information = _ByHandleFileInformation() + if not get_file_information(handle, ctypes.byref(information)): + raise _FileOpenError(file_path, _windows_last_error()) + if information.dwFileAttributes & file_attribute_reparse_point: + raise _UnsafeFileError(f"Could not safely open file: {file_path}") + + opened_path = _windows_final_path_name(get_final_path_name, handle, file_path) + if _windows_normalized_path(opened_path) != _windows_normalized_path(os.fspath(file_path)): + raise _UnsafeFileError(f"Could not safely open file: {file_path}") + + source_fd = msvcrt.open_osfhandle(handle, os.O_RDONLY | os.O_BINARY) # type: ignore[attr-defined] + except BaseException: + close_handle(handle) + raise + + return _fdopen_regular_file(source_fd, file_path) + + +def _windows_final_path_name(get_final_path_name: object, handle: int, file_path: Path) -> str: + """Return the canonical DOS path for an already-open Windows handle.""" + import ctypes + + buffer_size = 260 + while True: + buffer = ctypes.create_unicode_buffer(buffer_size) + result = cast(int, get_final_path_name(handle, buffer, buffer_size, 0)) # type: ignore[operator] + if result == 0: + raise _FileOpenError(file_path, _windows_last_error()) + if result < buffer_size: + return buffer.value + buffer_size = result + 1 + + +def _windows_last_error() -> OSError: + """Return the current Windows error as an ``OSError`` instance.""" + import ctypes + + return cast(OSError, ctypes.WinError(ctypes.get_last_error())) # type: ignore[attr-defined] + + +def _windows_normalized_path(path: str) -> str: + """Normalize a Windows DOS path for an exact opened-handle comparison.""" + long_path_prefix = "\\\\?\\" + long_unc_prefix = "\\\\?\\UNC\\" + if path.startswith(long_unc_prefix): + path = "\\\\" + path[len(long_unc_prefix) :] + elif path.startswith(long_path_prefix): + path = path[len(long_path_prefix) :] + return os.path.normcase(os.path.normpath(os.path.abspath(path))) + + +def _close_fd_safely(fd: int) -> None: + """Close a descriptor without masking the operation that owns it.""" + try: + os.close(fd) + except OSError: + pass + + +def _fdopen_regular_file(source_fd: int, file_path: Path) -> BinaryIO: + """Transfer an opened descriptor to a validated binary file object.""" + try: + source = os.fdopen(source_fd, "rb") + except OSError as exc: + _close_fd_safely(source_fd) + raise _FileOpenError(file_path, exc) from exc + try: + if not S_ISREG(os.fstat(source.fileno()).st_mode): + raise _UnsafeFileError(f"Refusing to open a symlinked or non-regular file: {file_path}") + except OSError as exc: + source.close() + raise _FileOpenError(file_path, exc) from exc + except BaseException: + source.close() + raise + return source + + class InputHandler: """ Handles input resolution for different source types. @@ -145,14 +418,15 @@ def resolve(self, input_path: str) -> tuple[Path, str]: return self._clone_git(input_path), "git" if self._is_file_url(input_path): return self._download_file(input_path), "url" + normalized_local_path = validate_local_input_path(Path(input_path)) if input_path.endswith(".zip"): - return self._extract_zip(Path(input_path)), "zip" + return self._extract_zip(normalized_local_path), "zip" if input_path.endswith(".md"): - return self._wrap_single_file(Path(input_path)), "file" - if Path(input_path).is_dir(): - return Path(input_path).resolve(), "directory" - if Path(input_path).is_file(): - return self._wrap_single_file(Path(input_path)), "file" + return self._wrap_single_file(normalized_local_path), "file" + if normalized_local_path.is_dir(): + return normalized_local_path, "directory" + if normalized_local_path.is_file(): + return self._wrap_single_file(normalized_local_path), "file" raise ValueError( f"Cannot determine input type for: {input_path}\n" "Supported formats: Git URL, file URL, .zip file, .md file, or directory" @@ -230,7 +504,16 @@ def _clone_git(self, url: str) -> Path: clone_dir = temp_dir / "repo" try: subprocess.run( - ["git", "clone", "--depth", "1", url, str(clone_dir)], + [ + "git", + "-c", + "core.symlinks=false", + "clone", + "--depth", + "1", + url, + str(clone_dir), + ], check=True, capture_output=True, timeout=60, @@ -356,38 +639,37 @@ def _extract_zip(self, zip_path: Path) -> Path: member name is applied before extraction to reject entries whose resolved path escapes the extraction directory. """ - if not zip_path.exists(): - raise FileNotFoundError(f"Zip file not found: {zip_path}") from None - temp_dir = self._get_temp_dir() - extract_dir = temp_dir / "extracted" - extract_dir.mkdir(exist_ok=True) - try: - with zipfile.ZipFile(zip_path, "r") as zf: - infos = zf.infolist() - if len(infos) > INGEST_MAX_ZIP_MEMBERS: - raise IngestLimitExceededError( - f"Zip exceeded ingest cap: {len(infos)} members > " - f"INGEST_MAX_ZIP_MEMBERS ({INGEST_MAX_ZIP_MEMBERS})" - ) - total_uncompressed = sum(info.file_size for info in infos) - if total_uncompressed > INGEST_MAX_BYTES: - raise IngestLimitExceededError( - f"Zip exceeded ingest cap: uncompressed " - f"{total_uncompressed} bytes > INGEST_MAX_BYTES " - f"({INGEST_MAX_BYTES})" - ) - extract_root = extract_dir.resolve() - for member in zf.namelist(): - member_path = (extract_dir / member).resolve() - if not str(member_path).startswith(str(extract_root)): - raise ValueError( - f"Zip entry '{member}' would escape extraction directory (zip-slip). " - "Archive is potentially malicious." + with _open_regular_file_no_follow(zip_path) as archive_file: + temp_dir = self._get_temp_dir() + extract_dir = temp_dir / "extracted" + extract_dir.mkdir(exist_ok=True) + try: + with zipfile.ZipFile(archive_file, "r") as zf: + infos = zf.infolist() + if len(infos) > INGEST_MAX_ZIP_MEMBERS: + raise IngestLimitExceededError( + f"Zip exceeded ingest cap: {len(infos)} members > " + f"INGEST_MAX_ZIP_MEMBERS ({INGEST_MAX_ZIP_MEMBERS})" + ) + total_uncompressed = sum(info.file_size for info in infos) + if total_uncompressed > INGEST_MAX_BYTES: + raise IngestLimitExceededError( + f"Zip exceeded ingest cap: uncompressed " + f"{total_uncompressed} bytes > INGEST_MAX_BYTES " + f"({INGEST_MAX_BYTES})" ) - zf.extractall(extract_dir) - except zipfile.BadZipFile: - logger.warning("Invalid zip or extract failed: %s", zip_path) - raise ValueError(f"Invalid zip file: {zip_path}") from None + extract_root = extract_dir.resolve() + for member in zf.namelist(): + member_path = (extract_dir / member).resolve() + if not str(member_path).startswith(str(extract_root)): + raise ValueError( + f"Zip entry '{member}' would escape extraction directory (zip-slip). " + "Archive is potentially malicious." + ) + zf.extractall(extract_dir) + except zipfile.BadZipFile: + logger.warning("Invalid zip or extract failed: %s", zip_path) + raise ValueError(f"Invalid zip file: {zip_path}") from None contents = list(extract_dir.iterdir()) if len(contents) == 1 and contents[0].is_dir(): return contents[0] @@ -395,11 +677,11 @@ def _extract_zip(self, zip_path: Path) -> Path: def _wrap_single_file(self, file_path: Path) -> Path: """Wrap a single file in a temporary directory for consistent handling.""" - if not file_path.exists(): - raise FileNotFoundError(f"File not found: {file_path}") from None - temp_dir = self._get_temp_dir() - dest = temp_dir / file_path.name - shutil.copy2(file_path, dest) + with _open_regular_file_no_follow(file_path) as source: + temp_dir = self._get_temp_dir() + dest = temp_dir / file_path.name + with dest.open("wb") as target: + shutil.copyfileobj(source, target) return temp_dir diff --git a/src/skillspector/mcp_server.py b/src/skillspector/mcp_server.py index e377d972a..4fb3157e6 100644 --- a/src/skillspector/mcp_server.py +++ b/src/skillspector/mcp_server.py @@ -50,10 +50,12 @@ def _is_local_target(target: str) -> bool: stripped = target.strip() if stripped.startswith("file://"): return True - if stripped.startswith(("http://", "https://", "git@", "ssh://", "git+ssh://")): + if stripped.startswith("git@"): return False if stripped.startswith(("\\\\", "//")): return True + if "://" in stripped: + return False try: candidate = Path(stripped).expanduser() @@ -61,8 +63,6 @@ def _is_local_target(target: str) -> bool: return True if candidate.is_absolute() or candidate.drive: return True - if "://" in stripped: - return False return candidate.exists() diff --git a/src/skillspector/multi_skill.py b/src/skillspector/multi_skill.py index be4c7ebab..ff0c59f75 100644 --- a/src/skillspector/multi_skill.py +++ b/src/skillspector/multi_skill.py @@ -25,6 +25,12 @@ from dataclasses import dataclass, field from pathlib import Path +from skillspector.input_handler import ( + _FileOpenError, + _open_regular_file_no_follow, + _UnsafeFileError, + validate_local_input_path, +) from skillspector.logging_config import get_logger logger = get_logger(__name__) @@ -60,6 +66,10 @@ def detect_skills(directory: Path) -> MultiSkillDetectionResult: Returns a MultiSkillDetectionResult with detected skills. """ + try: + directory = validate_local_input_path(directory) + except ValueError: + return MultiSkillDetectionResult(is_multi_skill=False) if not directory.is_dir(): return MultiSkillDetectionResult(is_multi_skill=False) @@ -69,7 +79,7 @@ def detect_skills(directory: Path) -> MultiSkillDetectionResult: skills: list[SkillDirectory] = [] for child in sorted(directory.iterdir()): - if not child.is_dir(): + if _is_link_or_junction(child) or not child.is_dir(): continue if child.name.startswith("."): continue @@ -93,7 +103,18 @@ def detect_skills(directory: Path) -> MultiSkillDetectionResult: def _has_skill_md(directory: Path) -> bool: """Check if directory contains a SKILL.md or skill.md at root level.""" - return (directory / "SKILL.md").is_file() or (directory / "skill.md").is_file() + return any( + not _is_link_or_junction(path) and path.is_file() + for path in (directory / "SKILL.md", directory / "skill.md") + ) + + +def _is_link_or_junction(path: Path) -> bool: + """Return True for links or uninspectable paths that must not be followed.""" + try: + return path.is_symlink() or path.is_junction() + except OSError: + return True def _extract_skill_name(skill_dir: Path) -> str: @@ -104,11 +125,12 @@ def _extract_skill_name(skill_dir: Path) -> str: for name in ("SKILL.md", "skill.md"): path = skill_dir / name - if not path.is_file(): + if _is_link_or_junction(path) or not path.is_file(): continue try: - content = path.read_text(encoding="utf-8", errors="replace") - except OSError: + with _open_regular_file_no_follow(path) as source: + content = source.read().decode("utf-8", errors="replace") + except (OSError, _FileOpenError, _UnsafeFileError): continue if not content.startswith("---"): break diff --git a/src/skillspector/nodes/build_context.py b/src/skillspector/nodes/build_context.py index e07149005..0caa441f5 100644 --- a/src/skillspector/nodes/build_context.py +++ b/src/skillspector/nodes/build_context.py @@ -32,6 +32,12 @@ import yaml from skillspector.constants import MAX_FILE_BYTES, build_model_config +from skillspector.input_handler import ( + _FileOpenError, + _open_regular_file_no_follow, + _UnsafeFileError, + validate_local_input_path, +) from skillspector.inspection_ledger import ( InspectionLedgerEvent, LedgerOutcome, @@ -86,7 +92,7 @@ def _resolve_skill_dir(state: SkillspectorState) -> Path: if not skill_path or not isinstance(skill_path, str) or not skill_path.strip(): raise ValueError("skill_path is required; provide input_path or skill_path to scan") try: - resolved = Path(skill_path).resolve() + resolved = validate_local_input_path(Path(skill_path)) except (OSError, RuntimeError) as e: raise ValueError(f"Invalid skill_path: {skill_path}") from e if not resolved.is_dir(): @@ -130,23 +136,50 @@ def _selected_baseline_component( return None +def _is_symlink(path: Path) -> bool: + """Return whether *path* is a link or junction without masking later stat errors.""" + try: + return path.is_symlink() or path.is_junction() + except OSError: + return False + + +def _resolves_outside(path: Path, root: Path) -> bool: + """Return whether *path* resolves outside an already-resolved *root*.""" + try: + return not path.resolve(strict=False).is_relative_to(root) + except OSError: + return False + + +def _read_text_no_follow(path: Path) -> str: + """Read a regular file without following symlinks at open time.""" + with _open_regular_file_no_follow(path) as source: + return source.read().decode("utf-8", errors="replace") + + def _walk_skill_files( skill_dir: Path, ) -> tuple[list[str], list[InspectionLedgerEvent]]: """Walk skill files and record scan-scope exclusions. - Skips _SKIP_DIRS and hidden files except those starting with .claude. + Skips _SKIP_DIRS, hidden files except those starting with .claude, and + symlinks, which must never supply content to remote LLM analyzers. """ paths: list[str] = [] exclusions: list[InspectionLedgerEvent] = [] - for root, dirnames, filenames in os.walk(skill_dir): + skill_root = skill_dir.resolve(strict=False) + for root, dirnames, filenames in os.walk(skill_dir, followlinks=False): root_path = Path(root) dirnames.sort() filenames.sort() relative_root = root_path.relative_to(skill_dir) skipped_dirnames = [name for name in dirnames if name in _SKIP_DIRS] - dirnames[:] = [name for name in dirnames if name not in _SKIP_DIRS] + symlinked_dirnames = [name for name in dirnames if _is_symlink(root_path / name)] + dirnames[:] = [ + name for name in dirnames if name not in _SKIP_DIRS and name not in symlinked_dirnames + ] for dirname in skipped_dirnames: boundary = (relative_root / dirname).as_posix() exclusions.append( @@ -158,6 +191,17 @@ def _walk_skill_files( reason=LedgerReason.EXCLUDED_DIRECTORY, ) ) + for dirname in symlinked_dirnames: + boundary = (relative_root / dirname).as_posix() + exclusions.append( + ledger_event( + outcome=LedgerOutcome.OUT_OF_SCOPE, + record_type=LedgerRecordType.SCOPE_BOUNDARY, + phase="discovery", + path=f"{boundary}/", + reason=LedgerReason.NOT_REGULAR_FILE, + ) + ) for filename in filenames: relative_path = (relative_root / filename).as_posix() @@ -174,10 +218,21 @@ def _walk_skill_files( continue # Use forward slashes on every OS: these relative paths are dict keys - # and SARIF/URI locations, so they must be portable. Do not filter - # on ``is_file()`` here: it follows symlinks and silently discards - # dangling or non-regular entries before the cache phase can record - # their terminal ledger evidence. + # and SARIF/URI locations, so they must be portable. Other + # non-regular entries remain inventoried for cache-phase evidence; + # symlinks are excluded before they can be read. + full = root_path / filename + if _is_symlink(full) or _resolves_outside(full, skill_root): + exclusions.append( + ledger_event( + outcome=LedgerOutcome.OUT_OF_SCOPE, + record_type=LedgerRecordType.SCOPE_BOUNDARY, + phase="discovery", + path=relative_path, + reason=LedgerReason.NOT_REGULAR_FILE, + ) + ) + continue paths.append(relative_path) paths.sort() return paths, exclusions @@ -310,8 +365,20 @@ def _read_file_cache( """Build readable file content and terminal events for cache failures.""" file_cache: dict[str, str] = {} ledger_events: list[InspectionLedgerEvent] = [] + skill_root = skill_dir.resolve(strict=False) for path in components: full = skill_dir / path + if _is_symlink(full) or _resolves_outside(full, skill_root): + ledger_events.append( + ledger_event( + outcome=LedgerOutcome.OUT_OF_SCOPE, + record_type=LedgerRecordType.SCOPE_BOUNDARY, + phase="cache", + path=path, + reason=LedgerReason.NOT_REGULAR_FILE, + ) + ) + continue try: file_stat = full.stat() except FileNotFoundError as exc: @@ -350,7 +417,7 @@ def _read_file_cache( ) continue try: - content = full.read_text(encoding="utf-8", errors="replace") + content = _read_text_no_follow(full) file_cache[path] = content except FileNotFoundError as exc: ledger_events.append( @@ -363,6 +430,28 @@ def _read_file_cache( error_class=type(exc).__name__, ) ) + except _UnsafeFileError: + ledger_events.append( + ledger_event( + outcome=LedgerOutcome.OUT_OF_SCOPE, + record_type=LedgerRecordType.SCOPE_BOUNDARY, + phase="cache", + path=path, + reason=LedgerReason.NOT_REGULAR_FILE, + ) + ) + except _FileOpenError as exc: + logger.debug("Could not read file: %s", path) + ledger_events.append( + ledger_event( + outcome=LedgerOutcome.FAILED, + record_type=LedgerRecordType.SYSTEM, + phase="cache", + path=path, + reason=LedgerReason.READ_ERROR, + error_class=exc.error_class, + ) + ) except OSError as exc: logger.debug("Could not read file: %s", path) ledger_events.append( @@ -384,13 +473,14 @@ def _parse_manifest(skill_dir: Path) -> dict[str, object]: Returns dict with name, description, triggers (list), permissions (list), allowed-tools (list), parameters (list). Returns {} if no file or parse fails. """ + skill_root = skill_dir.resolve(strict=False) for name in ("SKILL.md", "skill.md"): path = skill_dir / name - if not path.is_file(): + if _is_symlink(path) or _resolves_outside(path, skill_root) or not path.is_file(): continue try: - content = path.read_text(encoding="utf-8", errors="replace") - except OSError: + content = _read_text_no_follow(path) + except (OSError, _FileOpenError, _UnsafeFileError): logger.debug("Could not read manifest file: %s", name) return {} if not content.startswith("---"): diff --git a/src/skillspector/nodes/resolve_input.py b/src/skillspector/nodes/resolve_input.py index 7324d847e..95f856889 100644 --- a/src/skillspector/nodes/resolve_input.py +++ b/src/skillspector/nodes/resolve_input.py @@ -24,7 +24,7 @@ from pathlib import Path -from skillspector.input_handler import InputHandler +from skillspector.input_handler import InputHandler, validate_local_input_path from skillspector.logging_config import get_logger from skillspector.state import SkillspectorState @@ -59,7 +59,7 @@ def resolve_input(state: SkillspectorState) -> dict[str, object]: if skill_path and isinstance(skill_path, str) and skill_path.strip(): try: - resolved = Path(skill_path).resolve() + resolved = validate_local_input_path(Path(skill_path)) return { "skill_path": str(resolved), "temp_dir_for_cleanup": None, diff --git a/tests/integration/test_graph_scanner.py b/tests/integration/test_graph_scanner.py index 0aed2a5d7..f05614e88 100644 --- a/tests/integration/test_graph_scanner.py +++ b/tests/integration/test_graph_scanner.py @@ -13,7 +13,11 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Graph-invoke tests with safe/malicious skill dirs.""" +"""Graph-invoke tests with safe/malicious skill dirs. + +These tests exercise the deterministic scanner path. Live provider behavior is +covered separately in ``test_agent_cli_live.py``. +""" from pathlib import Path @@ -25,7 +29,7 @@ class TestGraphScanSafeSkill: def test_scan_safe_skill(self, safe_skill_dir: Path) -> None: """Scanning a safe skill returns low risk and has components.""" - result = graph.invoke({"skill_path": str(safe_skill_dir)}) + result = graph.invoke({"skill_path": str(safe_skill_dir), "use_llm": False}) assert "findings" in result assert "sarif_report" in result @@ -50,7 +54,7 @@ def test_scan_single_file(self, tmp_path: Path) -> None: encoding="utf-8", ) - result = graph.invoke({"skill_path": str(tmp_path)}) + result = graph.invoke({"skill_path": str(tmp_path), "use_llm": False}) assert result.get("manifest", {}).get("name") == "test-skill" assert result["risk_score"] == 0 @@ -76,7 +80,7 @@ def test_scan_extracts_metadata(self, tmp_path: Path) -> None: encoding="utf-8", ) - result = graph.invoke({"skill_path": str(tmp_path)}) + result = graph.invoke({"skill_path": str(tmp_path), "use_llm": False}) manifest = result.get("manifest", {}) assert manifest.get("name") == "my-skill" @@ -90,7 +94,7 @@ class TestGraphScanMaliciousSkill: def test_scan_malicious_skill(self, malicious_skill_dir: Path) -> None: """Scanning a malicious skill returns findings and high risk when implemented.""" - result = graph.invoke({"skill_path": str(malicious_skill_dir)}) + result = graph.invoke({"skill_path": str(malicious_skill_dir), "use_llm": False}) assert "findings" in result assert "filtered_findings" in result @@ -127,7 +131,7 @@ def test_critical_issue_high_severity_finding(self, tmp_path: Path) -> None: encoding="utf-8", ) - result = graph.invoke({"skill_path": str(tmp_path)}) + result = graph.invoke({"skill_path": str(tmp_path), "use_llm": False}) assert len(result["findings"]) >= 1 severities = [getattr(f, "severity", None) for f in result["findings"]] diff --git a/tests/nodes/test_build_context.py b/tests/nodes/test_build_context.py index bfc544ca4..19c4ca655 100644 --- a/tests/nodes/test_build_context.py +++ b/tests/nodes/test_build_context.py @@ -24,6 +24,7 @@ import json import os from pathlib import Path +from typing import BinaryIO import pytest from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer @@ -138,6 +139,29 @@ def test_build_context_ast_cache_handle_is_checkpoint_serializable(tmp_path: Pat assert serializer.dumps_typed(result) +def test_build_context_reads_directory_with_windows_secure_open( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Windows' handle-based fallback keeps normal directory scans usable.""" + _make_skill_spec_dir(tmp_path) + + def open_with_windows_handle(path: Path) -> BinaryIO: + return path.open("rb") + + monkeypatch.setattr("skillspector.input_handler._HAS_SECURE_DIR_FD", False) + monkeypatch.setattr("skillspector.input_handler._IS_WINDOWS", True) + monkeypatch.setattr( + "skillspector.input_handler._open_regular_file_from_windows_handle", + open_with_windows_handle, + ) + + result = build_context({"skill_path": str(tmp_path)}) + + assert result["file_cache"]["SKILL.md"].startswith("---") + assert result["file_cache"]["scripts/run.py"] == "print(1)\n" + assert result["manifest"]["name"] == "test-skill" + + def test_build_context_missing_skill_path() -> None: """Missing skill_path raises instead of producing a clean empty scan.""" state: SkillspectorState = {} @@ -446,14 +470,11 @@ def test_build_context_reports_read_error_without_fake_empty_content( """Unreadable files remain inventoried but are absent from the content cache.""" target = tmp_path / "broken.py" target.write_text("print(1)\n", encoding="utf-8") - original = Path.read_text - def fail_target(path: Path, *args: object, **kwargs: object) -> str: - if path == target: - raise PermissionError("sensitive operating-system detail") - return original(path, *args, **kwargs) + def deny_open(*args: object, **kwargs: object) -> int: + raise PermissionError("sensitive operating-system detail") - monkeypatch.setattr(Path, "read_text", fail_target) + monkeypatch.setattr("skillspector.input_handler.os.open", deny_open) result = build_context({"skill_path": str(tmp_path)}) assert "broken.py" in result["components"] @@ -479,8 +500,8 @@ def test_build_context_records_non_regular_files_in_the_ledger(tmp_path: Path) - assert event["reason_code"] == "not_regular_file" -def test_build_context_records_dangling_symlink_in_the_ledger(tmp_path: Path) -> None: - """Dangling entries are not silently omitted during discovery.""" +def test_build_context_excludes_dangling_symlink_from_scan_scope(tmp_path: Path) -> None: + """Symlinks are excluded rather than read as files from an unknown target.""" dangling = tmp_path / "missing.py" try: dangling.symlink_to("no-longer-present.py") @@ -489,10 +510,10 @@ def test_build_context_records_dangling_symlink_in_the_ledger(tmp_path: Path) -> result = build_context({"skill_path": str(tmp_path)}) - assert "missing.py" in result["components"] + assert "missing.py" not in result["components"] assert "missing.py" not in result["file_cache"] event = next(entry for entry in result["inspection_ledger"] if entry["path"] == "missing.py") - assert event["reason_code"] == "file_disappeared" + assert event["reason_code"] == "not_regular_file" def test_build_context_records_stat_errors_in_the_ledger( @@ -531,3 +552,118 @@ def test_build_context_records_non_regular_entries_in_the_ledger(tmp_path: Path) entry for entry in result["inspection_ledger"] if entry["path"] == "inspection.pipe" ) assert event["reason_code"] == "not_regular_file" + + +def test_build_context_rejects_symlink_to_external_file(tmp_path: Path) -> None: + """A symlinked file outside skill_dir must not enter the component cache.""" + secret = tmp_path.parent / "external_secret.txt" + secret.write_text("AWS_SECRET=hunter2", encoding="utf-8") + + skill_dir = tmp_path / "skill" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text("---\nname: s\ndescription: d\n---\n", encoding="utf-8") + (skill_dir / "creds.md").symlink_to(secret) + + result = build_context({"skill_path": str(skill_dir)}) + + assert "creds.md" not in result["components"] + assert "creds.md" not in result["file_cache"] + assert all("hunter2" not in content for content in result["file_cache"].values()) + + +def test_build_context_rejects_symlinked_directory(tmp_path: Path) -> None: + """A symlinked subdirectory outside skill_dir must not be traversed.""" + external = tmp_path.parent / "external_dir" + external.mkdir(exist_ok=True) + (external / "leak.md").write_text("PRIVATE_KEY=xyz", encoding="utf-8") + + skill_dir = tmp_path / "skill" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text("---\nname: s\ndescription: d\n---\n", encoding="utf-8") + (skill_dir / "linked").symlink_to(external, target_is_directory=True) + + result = build_context({"skill_path": str(skill_dir)}) + + assert not any(path.startswith("linked/") for path in result["components"]) + assert all("PRIVATE_KEY" not in content for content in result["file_cache"].values()) + + +def test_build_context_rejects_junctioned_directory( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Windows junctions must be excluded before os.walk can traverse them.""" + linked = tmp_path / "linked" + linked.mkdir() + (linked / "leak.md").write_text("PRIVATE_KEY=xyz", encoding="utf-8") + original_is_junction = Path.is_junction + + def is_junction(path: Path) -> bool: + return path == linked or original_is_junction(path) + + monkeypatch.setattr(Path, "is_junction", is_junction) + result = build_context({"skill_path": str(tmp_path)}) + + assert not any(path.startswith("linked/") for path in result["components"]) + assert all("PRIVATE_KEY" not in content for content in result["file_cache"].values()) + event = next(entry for entry in result["inspection_ledger"] if entry["path"] == "linked/") + assert event["reason_code"] == "not_regular_file" + + +def test_build_context_rejects_in_tree_symlink(tmp_path: Path) -> None: + """Even an in-tree symlink is skipped rather than read through.""" + skill_dir = tmp_path / "skill" + skill_dir.mkdir() + (skill_dir / "real.md").write_text("real content", encoding="utf-8") + (skill_dir / "SKILL.md").write_text("---\nname: s\ndescription: d\n---\n", encoding="utf-8") + (skill_dir / "alias.md").symlink_to(skill_dir / "real.md") + + result = build_context({"skill_path": str(skill_dir)}) + + assert "real.md" in result["components"] + assert "alias.md" not in result["components"] + + +def test_build_context_rejects_file_swapped_to_symlink_before_read( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A path replaced after stat must not leak its new symlink target.""" + from skillspector.nodes.build_context import _open_regular_file_no_follow + + secret = tmp_path.parent / "external_secret.txt" + secret.write_text("AWS_SECRET=hunter2", encoding="utf-8") + target = tmp_path / "payload.md" + target.write_text("safe", encoding="utf-8") + + def replace_target(path: Path) -> BinaryIO: + if path.name == target.name: + path.unlink() + path.symlink_to(secret) + return _open_regular_file_no_follow(path) + + monkeypatch.setattr( + "skillspector.nodes.build_context._open_regular_file_no_follow", replace_target + ) + result = build_context({"skill_path": str(tmp_path)}) + + assert "payload.md" in result["components"] + assert "payload.md" not in result["file_cache"] + assert all("hunter2" not in content for content in result["file_cache"].values()) + event = next(entry for entry in result["inspection_ledger"] if entry["path"] == "payload.md") + assert event["reason_code"] == "not_regular_file" + + +def test_build_context_rejects_symlinked_manifest(tmp_path: Path) -> None: + """Manifest parsing cannot bypass symlink rejection applied to the cache.""" + external = tmp_path.parent / "external_manifest.md" + external.write_text( + "---\nname: private-name\ndescription: private-description\n---\n", encoding="utf-8" + ) + skill_dir = tmp_path / "skill" + skill_dir.mkdir() + (skill_dir / "SKILL.md").symlink_to(external) + + result = build_context({"skill_path": str(skill_dir)}) + + assert result["manifest"] == {} + assert "SKILL.md" not in result["components"] + assert "SKILL.md" not in result["file_cache"] diff --git a/tests/nodes/test_resolve_input.py b/tests/nodes/test_resolve_input.py index 107ad75f8..c577ec20e 100644 --- a/tests/nodes/test_resolve_input.py +++ b/tests/nodes/test_resolve_input.py @@ -17,6 +17,8 @@ from pathlib import Path +import pytest + from skillspector.nodes.resolve_input import resolve_input @@ -38,6 +40,21 @@ def test_resolve_input_with_skill_path_only(tmp_path: Path) -> None: assert update.get("temp_dir_for_cleanup") is None +def test_resolve_input_rejects_skill_path_with_symlinked_parent(tmp_path: Path) -> None: + """The skill_path-only route must enforce the same symlink policy as input_path.""" + external_skill = tmp_path / "external" / "skill" + external_skill.mkdir(parents=True) + (external_skill / "SKILL.md").write_text("# External skill", encoding="utf-8") + symlinked_parent = tmp_path / "linked" + try: + symlinked_parent.symlink_to(external_skill.parent, target_is_directory=True) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + + with pytest.raises(ValueError, match="symlinked parent"): + resolve_input({"skill_path": str(symlinked_parent / external_skill.name)}) + + def test_resolve_input_prefers_input_path_over_skill_path(tmp_path: Path) -> None: """When both are set, input_path wins.""" (tmp_path / "SKILL.md").write_text("# Test", encoding="utf-8") diff --git a/tests/test_multi_skill.py b/tests/test_multi_skill.py index 3c1b634ab..4ae682133 100644 --- a/tests/test_multi_skill.py +++ b/tests/test_multi_skill.py @@ -124,6 +124,41 @@ def test_hidden_directories_skipped(self, tmp_path: Path) -> None: names = {s.name for s in result.skills} assert "hidden" not in names + def test_symlinked_skill_directory_is_skipped(self, tmp_path: Path) -> None: + """Detection must not read a skill manifest through a directory symlink.""" + for name in ("skill-a", "skill-b"): + sub = tmp_path / name + sub.mkdir() + (sub / "SKILL.md").write_text(f"---\nname: {name}\n---\n", encoding="utf-8") + external = tmp_path.parent / "external-skill" + external.mkdir() + (external / "SKILL.md").write_text("---\nname: private\n---\n", encoding="utf-8") + try: + (tmp_path / "linked-skill").symlink_to(external, target_is_directory=True) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + + result = detect_skills(tmp_path) + + assert result.is_multi_skill is True + assert {skill.name for skill in result.skills} == {"skill-a", "skill-b"} + + def test_symlinked_root_is_not_detected(self, tmp_path: Path) -> None: + """Direct callers cannot use detection to inspect a symlinked root.""" + external = tmp_path / "external" + external.mkdir() + (external / "SKILL.md").write_text("---\nname: private\n---\n", encoding="utf-8") + symlink = tmp_path / "linked-root" + try: + symlink.symlink_to(external, target_is_directory=True) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + + result = detect_skills(symlink) + + assert result.is_multi_skill is False + assert result.has_root_skill is False + def test_nonexistent_path(self, tmp_path: Path) -> None: """Non-existent path returns not multi-skill.""" result = detect_skills(tmp_path / "does-not-exist") diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index fb7061f6c..f6bd964ed 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -49,6 +49,32 @@ def test_cli_scan_local_directory(tmp_path: Path) -> None: assert "scan-test" in result.output or "skill" in result.output +def test_cli_rejects_symlinked_parent_before_preflight( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Recursive preflight must not inspect a directory behind a symlinked parent.""" + external_skill = tmp_path / "external" / "skill" + external_skill.mkdir(parents=True) + (external_skill / "SKILL.md").write_text("---\nname: private\n---\n", encoding="utf-8") + symlinked_parent = tmp_path / "linked" + try: + symlinked_parent.symlink_to(external_skill.parent, target_is_directory=True) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + + def fail_if_called(_: Path) -> MultiSkillDetectionResult: + raise AssertionError("preflight must not inspect an unsafe input path") + + monkeypatch.setattr("skillspector.cli.detect_skills", fail_if_called) + result = runner.invoke( + app, + ["scan", str(symlinked_parent / external_skill.name), "--recursive", "--no-llm"], + ) + + assert result.exit_code == 2 + assert "symlinked parent" in result.output + + def test_cli_scan_output_to_file(tmp_path: Path) -> None: """scan with --output writes report to file.""" skill_dir = tmp_path / "skill" diff --git a/tests/unit/test_input_handler.py b/tests/unit/test_input_handler.py index e3c7301fd..4a0567a5c 100644 --- a/tests/unit/test_input_handler.py +++ b/tests/unit/test_input_handler.py @@ -15,12 +15,52 @@ """Tests for skillspector input_handler (resolve directory, zip, single file).""" +import ctypes +import os +import sys +from errno import ENOENT from pathlib import Path +from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest -from skillspector.input_handler import ALLOWED_GIT_HOSTS, InputHandler +from skillspector.input_handler import ( + ALLOWED_GIT_HOSTS, + InputHandler, + _open_regular_file_from_windows_handle, +) + + +def _mock_windows_secure_open( + monkeypatch: pytest.MonkeyPatch, + source: Path, + *, + handle: int = 1, + attributes: int = 0, + final_path: str | None = None, +) -> None: + """Install a handle-level Windows open simulation on any platform.""" + + def get_file_information(_handle: int, information: object) -> bool: + information._obj.dwFileAttributes = attributes # type: ignore[attr-defined] + return True + + def get_final_path(_handle: int, buffer: object, _size: int, _flags: int) -> int: + opened_path = final_path or str(source) + buffer.value = opened_path # type: ignore[attr-defined] + return len(opened_path) + + kernel32 = SimpleNamespace( + CreateFileW=lambda *_args: handle, + GetFileInformationByHandle=get_file_information, + GetFinalPathNameByHandleW=get_final_path, + CloseHandle=lambda _handle: True, + ) + msvcrt = SimpleNamespace(open_osfhandle=lambda _handle, _flags: os.open(source, os.O_RDONLY)) + monkeypatch.setattr(ctypes, "WinDLL", lambda *_args, **_kwargs: kernel32, raising=False) + monkeypatch.setattr(os, "O_BINARY", 0, raising=False) + monkeypatch.setitem(sys.modules, "msvcrt", msvcrt) def test_resolve_directory(tmp_path: Path) -> None: @@ -50,6 +90,216 @@ def test_resolve_single_md_file(tmp_path: Path) -> None: handler.cleanup() +def test_resolve_single_symlinked_file_raises(tmp_path: Path) -> None: + """Standalone file inputs must not dereference symlinks before scanning.""" + secret = tmp_path / "external_secret.md" + secret.write_text("AWS_SECRET=hunter2", encoding="utf-8") + symlink = tmp_path / "SKILL.md" + try: + symlink.symlink_to(secret) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + + handler = InputHandler() + try: + with pytest.raises(ValueError, match="symlinked input"): + handler.resolve(str(symlink)) + assert handler.temp_dir_for_cleanup() is None + finally: + handler.cleanup() + + +def test_resolve_file_with_symlinked_parent_raises(tmp_path: Path) -> None: + """Standalone file inputs must not traverse a symlinked parent directory.""" + external_dir = tmp_path / "external" + external_dir.mkdir() + (external_dir / "secret.md").write_text("AWS_SECRET=hunter2", encoding="utf-8") + symlinked_parent = tmp_path / "linked" + try: + symlinked_parent.symlink_to(external_dir, target_is_directory=True) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + + handler = InputHandler() + try: + with pytest.raises(ValueError, match="symlinked parent"): + handler.resolve(str(symlinked_parent / "secret.md")) + assert handler.temp_dir_for_cleanup() is None + finally: + handler.cleanup() + + +def test_resolve_directory_with_symlinked_parent_raises(tmp_path: Path) -> None: + """Directory inputs must not escape through a symlinked ancestor.""" + external_skill = tmp_path / "external" / "skill" + external_skill.mkdir(parents=True) + (external_skill / "SKILL.md").write_text("# External skill", encoding="utf-8") + symlinked_parent = tmp_path / "linked" + try: + symlinked_parent.symlink_to(external_skill.parent, target_is_directory=True) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + + handler = InputHandler() + try: + with pytest.raises(ValueError, match="symlinked parent"): + handler.resolve(str(symlinked_parent / external_skill.name)) + assert handler.temp_dir_for_cleanup() is None + finally: + handler.cleanup() + + +def test_resolve_junctioned_directory_raises(tmp_path: Path) -> None: + """Directory inputs must reject terminal Windows junctions too.""" + skill_dir = tmp_path / "skill" + skill_dir.mkdir() + handler = InputHandler() + try: + with patch.object(Path, "is_junction", autospec=True) as is_junction: + is_junction.side_effect = lambda path: path == skill_dir + with pytest.raises(ValueError, match="junctioned input"): + handler.resolve(str(skill_dir)) + finally: + handler.cleanup() + + +def test_resolve_file_with_junction_parent_raises(tmp_path: Path) -> None: + """Standalone file inputs must not traverse Windows junctions.""" + source = tmp_path / "linked" / "SKILL.md" + source.parent.mkdir() + source.write_text("# Skill", encoding="utf-8") + handler = InputHandler() + try: + with patch.object(Path, "is_junction", autospec=True) as is_junction: + is_junction.side_effect = lambda path: path == source.parent + with pytest.raises(ValueError, match="symlinked parent"): + handler.resolve(str(source)) + assert handler.temp_dir_for_cleanup() is None + finally: + handler.cleanup() + + +def test_resolve_file_through_root_owned_system_alias(tmp_path: Path) -> None: + """Root-owned system aliases do not prevent scanning ordinary local files.""" + root_alias = Path("/var") + try: + relative_path = tmp_path.relative_to("/private/var") + except ValueError: + pytest.skip("temporary directory is not below the macOS /var alias") + if not root_alias.is_symlink(): + pytest.skip("/var is not a system alias on this platform") + + source = tmp_path / "SKILL.md" + source.write_text("# Skill", encoding="utf-8") + handler = InputHandler() + try: + resolved, source_type = handler.resolve(str(root_alias / relative_path / source.name)) + assert (resolved / source.name).read_text(encoding="utf-8") == "# Skill" + assert source_type == "file" + finally: + handler.cleanup() + + +def test_resolve_symlinked_zip_raises(tmp_path: Path) -> None: + """Local archives must be rejected before their symlink target is opened.""" + archive = tmp_path / "external_archive.zip" + archive.write_bytes(b"not opened") + symlink = tmp_path / "skill.zip" + try: + symlink.symlink_to(archive) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + + handler = InputHandler() + try: + with pytest.raises(ValueError, match="symlinked input"): + handler.resolve(str(symlink)) + assert handler.temp_dir_for_cleanup() is None + finally: + handler.cleanup() + + +def test_resolve_file_open_failure_does_not_create_temp_dir(tmp_path: Path) -> None: + """Failed secure opens leave no handler-owned temporary directory behind.""" + source = tmp_path / "SKILL.md" + source.write_text("# Skill", encoding="utf-8") + handler = InputHandler() + try: + with patch("skillspector.input_handler.os.open", side_effect=OSError("denied")): + with pytest.raises(ValueError, match="Could not safely open"): + handler.resolve(str(source)) + assert handler.temp_dir_for_cleanup() is None + finally: + handler.cleanup() + + +def test_resolve_file_rejects_platform_without_safe_open_support( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Scanning must fail closed when neither secure-open implementation is available.""" + source = tmp_path / "SKILL.md" + source.write_text("# Skill", encoding="utf-8") + handler = InputHandler() + try: + monkeypatch.setattr("skillspector.input_handler._HAS_SECURE_DIR_FD", False) + monkeypatch.setattr("skillspector.input_handler._IS_WINDOWS", False) + with pytest.raises(ValueError, match="Secure no-follow file opens are unavailable"): + handler.resolve(str(source)) + assert handler.temp_dir_for_cleanup() is None + finally: + handler.cleanup() + + +def test_windows_no_follow_open_reads_verified_regular_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Windows secure-open accepts a verified regular file handle.""" + source = tmp_path / "SKILL.md" + source.write_text("# Skill", encoding="utf-8") + _mock_windows_secure_open(monkeypatch, source) + + with _open_regular_file_from_windows_handle(source) as opened: + assert opened.read() == b"# Skill" + + +def test_windows_no_follow_open_rejects_missing_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Windows secure-open reports a missing file without exposing a handle.""" + source = tmp_path / "missing.md" + _mock_windows_secure_open(monkeypatch, source, handle=ctypes.c_void_p(-1).value) + monkeypatch.setattr( + "skillspector.input_handler._windows_last_error", lambda: OSError(ENOENT, "missing") + ) + + with pytest.raises(FileNotFoundError, match="File not found"): + _open_regular_file_from_windows_handle(source) + + +def test_windows_no_follow_open_rejects_reparse_point( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Windows secure-open rejects a reparse-point handle before reading it.""" + source = tmp_path / "SKILL.md" + source.write_text("# Skill", encoding="utf-8") + _mock_windows_secure_open(monkeypatch, source, attributes=0x00000400) + + with pytest.raises(ValueError, match="Could not safely open"): + _open_regular_file_from_windows_handle(source) + + +def test_windows_no_follow_open_rejects_canonical_path_mismatch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Windows secure-open rejects a handle whose resolved path changed.""" + source = tmp_path / "SKILL.md" + source.write_text("# Skill", encoding="utf-8") + _mock_windows_secure_open(monkeypatch, source, final_path=str(tmp_path / "outside.md")) + + with pytest.raises(ValueError, match="Could not safely open"): + _open_regular_file_from_windows_handle(source) + + def test_resolve_zip_file(tmp_path: Path) -> None: """Resolving a .zip file extracts and returns the extract dir.""" import zipfile @@ -97,6 +347,27 @@ def test_cleanup_idempotent(tmp_path: Path) -> None: handler.cleanup() +def test_clone_git_disables_symlinks() -> None: + """git clone must prevent symlinked entries from materializing as links.""" + handler = InputHandler() + try: + with ( + patch.object(handler, "_validate_url_host", return_value="github.com"), + patch( + "skillspector.input_handler.subprocess.run", + return_value=MagicMock(returncode=0), + ) as mock_run, + ): + handler._clone_git("https://github.com/example/repo.git") + cmd = mock_run.call_args.args[0] + + assert cmd[:2] == ["git", "-c"] + assert "core.symlinks=false" in cmd + assert cmd.index("-c") < cmd.index("clone") + finally: + handler.cleanup() + + def test_scp_url_is_git_url() -> None: """scp-style SSH URL is recognised as a Git URL.""" assert InputHandler()._is_git_url("git@github.com:org/repo.git") is True diff --git a/uv.lock b/uv.lock index c4f6b6a88..36be8df7c 100644 --- a/uv.lock +++ b/uv.lock @@ -2675,7 +2675,7 @@ wheels = [ [[package]] name = "skillspector" -version = "2.9.3" +version = "2.9.4" source = { editable = "." } dependencies = [ { name = "boto3" },