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
28 changes: 15 additions & 13 deletions src/skillspector/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -313,13 +313,15 @@ def scan(
"""
if mcp_registry:
if recursive or baseline is not None or show_suppressed or yara_rules_dir is not None:
console.print(
err_console.print(
"[red]Error:[/red] --mcp-registry cannot be combined with "
"--recursive, --baseline, --show-suppressed, or --yara-rules-dir"
)
raise typer.Exit(code=2)
if format != FormatChoice.json:
console.print("[red]Error:[/red] --mcp-registry currently supports only --format json")
err_console.print(
"[red]Error:[/red] --mcp-registry currently supports only --format json"
)
raise typer.Exit(code=2)
try:
result = scan_registry(input_path)
Expand All @@ -334,7 +336,7 @@ def scan(
except typer.Exit:
raise
except Exception as e:
console.print(f"[red]Error:[/red] {e}")
err_console.print(f"[red]Error:[/red] {e}")
raise typer.Exit(code=2) from e
return

Expand All @@ -346,13 +348,13 @@ def scan(
try:
resolved_path = validate_local_input_path(resolved_path)
except ValueError as e:
console.print(f"[red]Error:[/red] {e}")
err_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:
if baseline is not None:
console.print(
err_console.print(
"[red]Error:[/red] --baseline is not supported for recursive "
"multi-skill scans; scan each sub-skill with its own baseline"
)
Expand Down Expand Up @@ -428,13 +430,13 @@ def scan(
except typer.Exit:
raise
except (FileNotFoundError, ValueError) as e:
console.print(f"[red]Error:[/red] {e}")
err_console.print(f"[red]Error:[/red] {e}")
raise typer.Exit(code=2) from e
except Exception as e:
if verbose:
console.print_exception()
err_console.print_exception()
else:
console.print(f"[red]Error:[/red] {e}")
err_console.print(f"[red]Error:[/red] {e}")
raise typer.Exit(code=2) from e
finally:
if result is not None:
Expand Down Expand Up @@ -494,7 +496,7 @@ def _scan_multi_skill(
severity = result.get("risk_severity") or "LOW"
console.print(f" Score: {score}/100 ({severity})\n")
except Exception as e:
console.print(f" [red]Error:[/red] {e}\n")
err_console.print(f" [red]Error:[/red] {e}\n")
execution_failed = True
results.append({"skill_name": skill.name, "error": str(e)})

Expand Down Expand Up @@ -608,7 +610,7 @@ def mcp(

run_mcp(transport=transport.value, host=host, port=port)
except ModuleNotFoundError as e:
console.print(f"[red]Error:[/red] {e}")
err_console.print(f"[red]Error:[/red] {e}")
raise typer.Exit(code=2) from e


Expand Down Expand Up @@ -682,13 +684,13 @@ def baseline(
except typer.Exit:
raise
except (FileNotFoundError, ValueError) as e:
console.print(f"[red]Error:[/red] {e}")
err_console.print(f"[red]Error:[/red] {e}")
raise typer.Exit(code=2) from e
except Exception as e:
if verbose:
console.print_exception()
err_console.print_exception()
else:
console.print(f"[red]Error:[/red] {e}")
err_console.print(f"[red]Error:[/red] {e}")
raise typer.Exit(code=2) from e
finally:
if result is not None:
Expand Down
51 changes: 46 additions & 5 deletions src/skillspector/nodes/analyzers/static_patterns_supply_chain.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@

from __future__ import annotations

import json
import os
import re
import sys
Expand Down Expand Up @@ -547,8 +548,23 @@ def _extract_packages_from_requirements(content: str) -> list[tuple[str, str | N
return results


def _extract_packages_from_package_json(content: str) -> list[tuple[str, str | None, int]]:
"""Extract (package_name, version_or_None, line_number) from package.json content."""
_NPM_DEPENDENCY_SECTIONS = ("dependencies", "devDependencies", "peerDependencies")


def _package_json_line(content: str, section: str, name: str) -> int:
"""Best-effort line for a dependency entry, so findings keep pointing somewhere useful.

Parsing JSON loses positions, and the search starts at the section header so a name that
also appears in ``scripts`` does not win.
"""
header = re.search(rf'"{re.escape(section)}"\s*:', content)
start = header.end() if header else 0
entry = re.compile(rf'"{re.escape(name)}"\s*:').search(content, start)
return get_line_number(content, entry.start()) if entry else 1


def _extract_packages_from_package_json_scan(content: str) -> list[tuple[str, str | None, int]]:
"""Line-oriented fallback, used only when the manifest is not valid JSON."""
results: list[tuple[str, str | None, int]] = []
in_deps = False
for i, line in enumerate(content.splitlines(), 1):
Expand All @@ -562,9 +578,34 @@ def _extract_packages_from_package_json(content: str) -> list[tuple[str, str | N
if in_deps:
m = re.match(r'"([^"]+)"\s*:\s*"([^"]*)"', stripped)
if m:
name = m.group(1)
version = _pinned_npm_version(m.group(2))
results.append((name, version, i))
results.append((m.group(1), _pinned_npm_version(m.group(2)), i))
return results


def _extract_packages_from_package_json(content: str) -> list[tuple[str, str | None, int]]:
"""Extract (package_name, version_or_None, line_number) from package.json content.

package.json is JSON, so it is parsed as JSON. Scanning it line by line made the result
depend on formatting: a manifest written on a single line — which is valid, and what many
generators emit — never entered the dependency section at all and yielded *no* dependencies,
silently. The line-oriented scan remains as a fallback for manifests that do not parse.
"""
try:
data = json.loads(content)
except (ValueError, TypeError):
return _extract_packages_from_package_json_scan(content)
if not isinstance(data, dict):
return []
results: list[tuple[str, str | None, int]] = []
for section in _NPM_DEPENDENCY_SECTIONS:
deps = data.get(section)
if not isinstance(deps, dict):
continue
for name, spec in deps.items():
if not isinstance(name, str) or not isinstance(spec, str):
continue
line = _package_json_line(content, section, name)
results.append((name, _pinned_npm_version(spec), line))
return results


Expand Down
189 changes: 189 additions & 0 deletions tests/unit/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,11 @@

"""Tests for skillspector CLI (skillspector scan, --version)."""

import ast
import json
import sys
from collections.abc import Callable, Iterator
from contextlib import AbstractContextManager, ExitStack, contextmanager, nullcontext
from pathlib import Path
from types import SimpleNamespace
from typing import Any
Expand All @@ -27,6 +31,7 @@
from typer.testing import CliRunner

from skillspector import __version__
from skillspector import cli as cli_module
from skillspector.cli import FormatChoice, _scan_multi_skill, app
from skillspector.multi_skill import MultiSkillDetectionResult, SkillDirectory

Expand Down Expand Up @@ -1133,3 +1138,187 @@ def fake_invoke(state: dict[str, Any], config: Any = None) -> dict[str, Any]:
assert payload["issues"] == [{"id": "X-1", "severity": "low"}]
assert payload["suppressed_count"] == 0
assert payload["suppressed"] == []


# --- Fatal diagnostics belong on stderr ---------------------------------------------------
#
# Anything driving the CLI from a script separates the two streams and parses stdout. A
# diagnostic printed there is both lost as a diagnostic and corrupting as output. The cases
# below enumerate every path that prints and then exits, so a new one cannot be added on the
# wrong stream without a test turning red.

FatalPath = tuple[list[str], AbstractContextManager[object]]


@contextmanager
def _all_of(*managers: AbstractContextManager[object]) -> Iterator[None]:
"""Enter several patches as one context, so a case can state more than one."""
with ExitStack() as stack:
for manager in managers:
stack.enter_context(manager)
yield


def _registry_payload(directory: Path) -> Path:
"""A registry input that parses, so an argument check is what fails."""
payload = directory / "registry.json"
payload.write_text('{"servers": []}', encoding="utf-8")
return payload


def _skill_dir(directory: Path) -> Path:
"""A minimal skill directory the CLI accepts as an input path."""
skill = directory / "skill"
skill.mkdir(exist_ok=True)
(skill / "SKILL.md").write_text("# Skill\n", encoding="utf-8")
return skill


def _multi_skill(directory: Path) -> AbstractContextManager[object]:
"""Take the multi-skill branch without building two real skill trees."""
return patch(
"skillspector.cli.detect_skills",
return_value=MultiSkillDetectionResult(
is_multi_skill=True,
skills=[
SkillDirectory(path=directory / "one", name="one", relative_path="one"),
SkillDirectory(path=directory / "two", name="two", relative_path="two"),
],
has_root_skill=False,
),
)


def _scan_raises(exc: BaseException) -> AbstractContextManager[object]:
"""Make the graph blow up, which is how the generic handlers are reached."""
return patch("skillspector.cli.graph.invoke", side_effect=exc)


def _registry_flag_conflict(d: Path) -> FatalPath:
args = ["scan", str(_registry_payload(d)), "--mcp-registry", "--recursive"]
return args, nullcontext()


def _registry_wrong_format(d: Path) -> FatalPath:
args = ["scan", str(_registry_payload(d)), "--mcp-registry", "--format", "markdown"]
return args, nullcontext()


def _registry_scan_fails(d: Path) -> FatalPath:
args = ["scan", str(_registry_payload(d)), "--mcp-registry", "--format", "json"]
return args, patch(
"skillspector.cli.scan_registry", side_effect=RuntimeError("registry unreachable")
)


def _symlinked_input(d: Path) -> FatalPath:
link = d / "linked-skill"
try:
link.symlink_to(_skill_dir(d), target_is_directory=True)
except OSError:
pytest.skip("symlinks are not supported on this filesystem")
return ["scan", str(link), "--no-llm"], nullcontext()


def _recursive_multi_skill_with_baseline(d: Path) -> FatalPath:
args = ["scan", str(_skill_dir(d)), "--recursive", "--baseline", str(d / "b.yaml"), "--no-llm"]
return args, _multi_skill(d)


def _multi_skill_child_crashes(d: Path) -> FatalPath:
args = ["scan", str(_skill_dir(d)), "--recursive", "--no-llm"]
return args, _all_of(_multi_skill(d), _scan_raises(RuntimeError("child scan crashed")))


def _scan_input_missing(d: Path) -> FatalPath:
args = ["scan", str(_skill_dir(d)), "--no-llm"]
return args, _scan_raises(FileNotFoundError("skill vanished"))


def _scan_crashes(d: Path) -> FatalPath:
args = ["scan", str(_skill_dir(d)), "--no-llm"]
return args, _scan_raises(RuntimeError("scan crashed"))


def _scan_crashes_verbose(d: Path) -> FatalPath:
args = ["scan", str(_skill_dir(d)), "--no-llm", "--verbose"]
return args, _scan_raises(RuntimeError("scan crashed"))


def _baseline_input_missing(d: Path) -> FatalPath:
args = ["baseline", str(_skill_dir(d)), "--no-llm", "-o", str(d / "b.yaml")]
return args, _scan_raises(FileNotFoundError("baseline input missing"))


def _baseline_crashes(d: Path) -> FatalPath:
args = ["baseline", str(_skill_dir(d)), "--no-llm", "-o", str(d / "b.yaml")]
return args, _scan_raises(RuntimeError("baseline crashed"))


def _baseline_crashes_verbose(d: Path) -> FatalPath:
args = ["baseline", str(_skill_dir(d)), "--no-llm", "-o", str(d / "b.yaml"), "--verbose"]
return args, _scan_raises(RuntimeError("baseline crashed"))


def _mcp_module_missing(d: Path) -> FatalPath:
return ["mcp"], patch.dict(sys.modules, {"skillspector.mcp_server": None})


@pytest.mark.parametrize(
("build", "needle"),
[
pytest.param(_registry_flag_conflict, "cannot be combined", id="registry-flag-conflict"),
pytest.param(_registry_wrong_format, "supports only --format json", id="registry-format"),
pytest.param(_registry_scan_fails, "registry unreachable", id="registry-scan-fails"),
pytest.param(_symlinked_input, "Refusing to resolve", id="symlinked-input"),
pytest.param(
_recursive_multi_skill_with_baseline,
"not supported for recursive",
id="recursive-baseline",
),
pytest.param(_multi_skill_child_crashes, "child scan crashed", id="multi-skill-child"),
pytest.param(_scan_input_missing, "skill vanished", id="scan-input-missing"),
pytest.param(_scan_crashes, "scan crashed", id="scan-crashes"),
pytest.param(_scan_crashes_verbose, "RuntimeError", id="scan-crashes-verbose"),
pytest.param(_baseline_input_missing, "baseline input missing", id="baseline-missing"),
pytest.param(_baseline_crashes, "baseline crashed", id="baseline-crashes"),
pytest.param(_baseline_crashes_verbose, "RuntimeError", id="baseline-verbose"),
pytest.param(_mcp_module_missing, "skillspector.mcp_server", id="mcp-module-missing"),
],
)
def test_fatal_diagnostics_never_reach_stdout(
tmp_path: Path, build: Callable[[Path], FatalPath], needle: str
) -> None:
"""A path that prints and exits writes to stderr, leaving stdout machine-readable."""
args, ctx = build(tmp_path)

with ctx:
result = runner.invoke(app, args)

assert result.exit_code == 2
assert needle in result.stderr
assert needle not in result.stdout


def test_cli_writes_no_error_styled_output_to_stdout() -> None:
"""Guards new code: the invariant above regressed twice because nothing enforced it."""
tree = ast.parse(Path(cli_module.__file__).read_text(encoding="utf-8"))
offenders: list[tuple[int, str]] = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute):
continue
target = node.func.value
if not isinstance(target, ast.Name) or target.id != "console":
continue
if node.func.attr == "print_exception":
offenders.append((node.lineno, "print_exception()"))
elif node.func.attr == "print":
text = " ".join(
part.value
for part in ast.walk(node)
if isinstance(part, ast.Constant) and isinstance(part.value, str)
)
if "[red]Error:" in text:
offenders.append((node.lineno, text[:60]))

assert offenders == [], f"error output must use err_console, found on stdout: {offenders}"
Loading
Loading