diff --git a/src/skillspector/cli.py b/src/skillspector/cli.py index 35718800..1feae2f0 100644 --- a/src/skillspector/cli.py +++ b/src/skillspector/cli.py @@ -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) @@ -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 @@ -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" ) @@ -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: @@ -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)}) @@ -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 @@ -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: diff --git a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py index 63a6b55e..ece8391d 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py +++ b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py @@ -28,6 +28,7 @@ from __future__ import annotations +import json import os import re import sys @@ -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): @@ -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 diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 438d5827..b6d6c566 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -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 @@ -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 @@ -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}" diff --git a/tests/unit/test_patterns_new.py b/tests/unit/test_patterns_new.py index 029553df..6b12122e 100644 --- a/tests/unit/test_patterns_new.py +++ b/tests/unit/test_patterns_new.py @@ -1906,6 +1906,54 @@ def test_extract_packages_package_json_caret_is_not_a_pin(self) -> None: assert versions["semver"] is None assert versions["glob"] is None + def test_package_json_on_a_single_line_is_not_invisible(self) -> None: + # Regression: the line-oriented scan never entered the dependency section, so a valid + # one-line manifest yielded no dependencies at all — silently. + content = '{"name":"x","dependencies":{"express":"^4.18.0","lodash":"4.17.21"}}' + names = {p[0] for p in sc_mod._extract_packages_from_package_json(content)} + assert names == {"express", "lodash"} + + def test_package_json_compact_keeps_versions(self) -> None: + # Version resolution is not this PR's subject: it stays whatever the shared predicate + # decides (#319). Only the parsing of the manifest changes, and a compact manifest must + # resolve exactly like the indented one. + content = '{"dependencies":{"lodash":"4.17.21","semver":"^7.5.0"}}' + versions = {p[0]: p[1] for p in sc_mod._extract_packages_from_package_json(content)} + assert versions["lodash"] == "4.17.21" + assert versions["semver"] is None + + def test_package_json_line_numbers_survive_parsing(self) -> None: + content = '{\n "name": "x",\n "dependencies": {\n "express": "4.18.0"\n }\n}\n' + lines = {p[0]: p[2] for p in sc_mod._extract_packages_from_package_json(content)} + assert lines["express"] == 4 + + def test_package_json_line_prefers_the_dependency_over_a_script(self) -> None: + # A name that also appears in "scripts" must not steal the line number. + content = ( + "{\n" + ' "scripts": { "express": "node server.js" },\n' + ' "dependencies": {\n' + ' "express": "4.18.0"\n' + " }\n" + "}\n" + ) + lines = {p[0]: p[2] for p in sc_mod._extract_packages_from_package_json(content)} + assert lines["express"] == 4 + + def test_package_json_invalid_falls_back_to_the_scan(self) -> None: + # A manifest that does not parse keeps the previous behaviour instead of going blind. + content = '{\n "dependencies": {\n "express": "4.18.0",\n' # truncated + names = {p[0] for p in sc_mod._extract_packages_from_package_json(content)} + assert "express" in names + + def test_package_json_non_object_is_empty(self) -> None: + assert sc_mod._extract_packages_from_package_json("[1, 2, 3]") == [] + + def test_package_json_ignores_non_string_specs(self) -> None: + content = '{"dependencies":{"ok":"1.0.0","broken":{"version":"1.0.0"},"n":42}}' + names = {p[0] for p in sc_mod._extract_packages_from_package_json(content)} + assert names == {"ok"} + def test_extract_packages_package_json(self) -> None: content = ( '{\n "dependencies": {\n "express": "^4.18.0",\n "lodash": "4.17.21"\n }\n}'