From 4ca0915dcb2a8575cb0ffbaec8ce52d9ec147a2b Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Wed, 28 Oct 2026 00:47:04 -0700 Subject: [PATCH] feat(cli): apply shipped baselines from local ZIP inputs Signed-off-by: Christopher Kevin --- docs/SUPPRESSION.md | 7 ++ src/skillspector/cli.py | 88 ++++++++++++++++-------- src/skillspector/nodes/resolve_input.py | 11 +++ src/skillspector/state.py | 2 + tests/nodes/test_resolve_input.py | 29 ++++++++ tests/unit/test_cli.py | 89 +++++++++++++++++++++++++ 6 files changed, 199 insertions(+), 27 deletions(-) diff --git a/docs/SUPPRESSION.md b/docs/SUPPRESSION.md index 1390a8086..3745e3cea 100644 --- a/docs/SUPPRESSION.md +++ b/docs/SUPPRESSION.md @@ -37,6 +37,13 @@ skillspector scan ./my-skill/ --baseline .skillspector-baseline.yaml --show-supp | `skillspector baseline [-o FILE] [--no-llm] [--reason TEXT]` | Scan and write a baseline that fingerprint-suppresses every current finding. Default output: `.skillspector-baseline.yaml`. | | `skillspector scan --baseline FILE` (`-b`) | Suppress findings matching the baseline before scoring/reporting. | | `skillspector scan --baseline FILE --show-suppressed` | Also list the suppressed findings (they still don't affect the score). | +| `skillspector scan --use-shipped-baseline` | Apply the canonical `.skillspector-baseline.yaml` at the skill root after reviewing the author's suppressions. An explicit `--baseline` takes precedence. | + +For a local ZIP, discovery uses the safely extracted skill root, including a +single wrapper directory. Without opt-in, an archived baseline is not loaded. +An opted-in malformed baseline fails the scan. The option applies only to the +requested root, not baselines supplied by transitive references, and does not +enable shipped-baseline discovery for remote URLs. A missing, malformed, or unsupported baseline file exits with code 2. Generation also exits with code 2 without creating or replacing the output when diff --git a/src/skillspector/cli.py b/src/skillspector/cli.py index 0775ed426..e5d6531b0 100644 --- a/src/skillspector/cli.py +++ b/src/skillspector/cli.py @@ -317,6 +317,7 @@ def _scan_state( show_suppressed: bool = False, source_local_only: bool = False, exclude_patterns: list[str] | None = None, + use_shipped_baseline: bool = False, ) -> dict[str, object]: """Build initial graph state from scan CLI args.""" state: dict[str, object] = { @@ -328,6 +329,9 @@ def _scan_state( } if exclude_patterns: state["exclude_patterns"] = list(exclude_patterns) + if use_shipped_baseline: + state["use_shipped_baseline"] = True + state["show_suppressed"] = show_suppressed if yara_rules_dir is not None: state["yara_rules_dir"] = yara_rules_dir if baseline is not None: @@ -338,6 +342,34 @@ def _scan_state( return state +def _shipped_baseline_notice(shipped: Path | None, input_path: Path, apply: bool) -> None: + if shipped is not None: + if apply: + err_console.print(_literal_line("Applying author-shipped baseline", "yellow", shipped)) + err_console.print( + "[dim]Suppressed findings do not count toward the risk score; " + "use --show-suppressed to list them.[/dim]" + ) + else: + err_console.print( + _literal_line("Shipped baseline detected (not applied)", "yellow", shipped) + ) + err_console.print( + "[dim]Review it, then re-run with --use-shipped-baseline to apply its " + "suppressions. Findings and risk score are unaffected until you opt in.[/dim]" + ) + elif apply: + err_console.print( + Text( + _visible_terminal_text( + f"--use-shipped-baseline: no shipped baseline found in {input_path}; " + "scanning without a baseline." + ), + style="dim", + ) + ) + + def _result_body(result: dict) -> str: report_body = result.get("report_body") or "" if not report_body and result.get("sarif_report") is not None: @@ -564,7 +596,7 @@ def scan( typer.Option( "--use-shipped-baseline", help="Apply a baseline shipped at the top level of the scanned skill " - "directory (.skillspector-baseline.yaml). Off by default: a skill " + "directory or local ZIP (.skillspector-baseline.yaml). Off by default: a skill " "author's baseline can suppress findings in your scan, so a discovered " "baseline is only reported until you opt in. Ignored when --baseline " "is given. Not supported for recursive multi-skill scans; scan each " @@ -896,40 +928,25 @@ def scan( f"this directory. Use --recursive to scan each independently." ) - shipped: Path | None = None - if baseline is None and resolved_path.is_dir(): - shipped = discover_baseline(resolved_path) - if shipped is not None: + zip_baseline = ( + use_shipped_baseline + and baseline is None + and resolved_path.is_file() + and resolved_path.suffix.lower() == ".zip" + ) + if baseline is None and not zip_baseline: + shipped = discover_baseline(resolved_path) if resolved_path.is_dir() else None + _shipped_baseline_notice(shipped, resolved_path, use_shipped_baseline) if use_shipped_baseline: baseline = shipped - err_console.print(_literal_line("Applying author-shipped baseline", "yellow", shipped)) - err_console.print( - "[dim]Suppressed findings do not count toward the risk score; " - "use --show-suppressed to list them.[/dim]" - ) - else: - err_console.print( - _literal_line("Shipped baseline detected (not applied)", "yellow", shipped) - ) - err_console.print( - "[dim]Review it, then re-run with --use-shipped-baseline to apply its " - "suppressions. Findings and risk score are unaffected until you opt in.[/dim]" - ) - elif use_shipped_baseline and baseline is None: - err_console.print( - Text( - _visible_terminal_text( - f"--use-shipped-baseline: no shipped baseline found in {resolved_path}; scanning without a baseline." - ), - style="dim", - ) - ) result = None try: scan_skill = partial(_scan_skill, exclude_patterns=exclude) if exclude else _scan_skill if inspect_dependencies: scan_skill = partial(scan_skill, inspect_dependencies=True) + if zip_baseline: + scan_skill = partial(scan_skill, use_shipped_baseline=True) result = scan_skill( input_path=input_path, format=format, @@ -1629,6 +1646,7 @@ def _run_graph_scan( stream_progress: bool = False, exclude_patterns: list[str] | None = None, inspect_dependencies: bool = False, + use_shipped_baseline: bool = False, ) -> dict[str, object]: state = _scan_state( input_path=input_path, @@ -1639,6 +1657,7 @@ def _run_graph_scan( show_suppressed=show_suppressed, source_local_only=source_local_only, exclude_patterns=exclude_patterns, + use_shipped_baseline=use_shipped_baseline, ) if inspect_dependencies: state["inspect_dependencies"] = True @@ -1744,6 +1763,7 @@ def _run_graph_scan_for_source( stream_progress: bool = False, exclude_patterns: list[str] | None = None, inspect_dependencies: bool = False, + use_shipped_baseline: bool = False, ) -> dict[str, object]: """Invoke a scan without widening the legacy call contract for public sources.""" run_graph_scan = ( @@ -1753,6 +1773,8 @@ def _run_graph_scan_for_source( run_graph_scan = partial(run_graph_scan, exclude_patterns=exclude_patterns) if inspect_dependencies: run_graph_scan = partial(run_graph_scan, inspect_dependencies=True) + if use_shipped_baseline: + run_graph_scan = partial(run_graph_scan, use_shipped_baseline=True) if initial_inspection_ledger is not None: if source_local_only: return run_graph_scan( @@ -2624,6 +2646,7 @@ def _scan_skill( source_local_only: bool = False, exclude_patterns: list[str] | None = None, inspect_dependencies: bool = False, + use_shipped_baseline: bool = False, ) -> dict[str, object]: yara_dir = str(yara_rules_dir.resolve()) if yara_rules_dir else None active_visited: set[str] = set() @@ -2645,6 +2668,8 @@ def _scan_skill( run_for_source = partial(run_for_source, inspect_dependencies=True) if exclude_patterns: run_for_source = partial(run_for_source, exclude_patterns=exclude_patterns) + if use_shipped_baseline: + run_for_source = partial(run_for_source, use_shipped_baseline=True) if pre_scan_ledger_events: result = run_for_source( input_path=input_path, @@ -2670,6 +2695,15 @@ def _scan_skill( source_local_only=source_local_only, stream_progress=stream_progress, ) + if use_shipped_baseline: + try: + selected = result.get("baseline_path") + _shipped_baseline_notice( + Path(selected) if isinstance(selected, str) else None, Path(input_path), True + ) + except BaseException: + cleanup_result(result) + raise if not transitive_enabled: return result if transitive_traversal is None: # Defensive: transitive scans initialize before root work. diff --git a/src/skillspector/nodes/resolve_input.py b/src/skillspector/nodes/resolve_input.py index eaffb4a06..60f04be9d 100644 --- a/src/skillspector/nodes/resolve_input.py +++ b/src/skillspector/nodes/resolve_input.py @@ -35,6 +35,7 @@ SkillspectorState, ensure_workflow_resource_budget, ) +from skillspector.suppression import discover_baseline, load_baseline logger = get_logger(__name__) @@ -79,6 +80,16 @@ def resolve_input(state: SkillspectorState) -> dict[str, object]: update["temp_dir_for_cleanup"] = str(temp_dir) else: update["temp_dir_for_cleanup"] = None + if ( + source_type == "zip" + and state.get("use_shipped_baseline") + and state.get("baseline") is None + and not state.get("baseline_path") + ): + shipped = discover_baseline(resolved) + if shipped is not None: + update["baseline"] = load_baseline(shipped) + update["baseline_path"] = str(shipped) return update except TransitiveIngestTruncatedError as exc: # The graph has no directory to scan. Propagate a typed, sanitized diff --git a/src/skillspector/state.py b/src/skillspector/state.py index 3472aadd4..ae9c5e4da 100644 --- a/src/skillspector/state.py +++ b/src/skillspector/state.py @@ -322,6 +322,8 @@ class SkillspectorState(TypedDict, total=False): # When it is inside the scan target, build_context excludes only that file # so waiver text cannot scan itself or enter regenerated fingerprints. baseline_path: str | None + # Explicit root opt-in for a local ZIP, applied after safe materialization. + use_shipped_baseline: bool show_suppressed: bool suppressed_findings: list[object] # Kept findings as baseline suppression saw them: one per occurrence, before diff --git a/tests/nodes/test_resolve_input.py b/tests/nodes/test_resolve_input.py index b6103a833..0e2758c31 100644 --- a/tests/nodes/test_resolve_input.py +++ b/tests/nodes/test_resolve_input.py @@ -49,6 +49,35 @@ def test_resolve_input_with_skill_path_only(tmp_path: Path) -> None: assert update.get("selected_source_identity") == tmp_path.name +@pytest.mark.parametrize("interrupted", [False, True]) +def test_zip_baseline_loading_failure_cleans_extraction( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, interrupted: bool +) -> None: + """Loading fails inside the materializer's cleanup boundary, including interruption.""" + archive = tmp_path / "skill.zip" + with zipfile.ZipFile(archive, "w") as bundle: + bundle.writestr("SKILL.md", "# Skill") + bundle.writestr(".skillspector-baseline.yaml", "version: 2\nrules: [{}]\n") + created: list[Path] = [] + original = tempfile.mkdtemp + + def mkdtemp(**kwargs: object) -> str: + path = original(dir=tmp_path, **kwargs) + created.append(Path(path)) + return path + + monkeypatch.setattr("skillspector.input_handler.tempfile.mkdtemp", mkdtemp) + if interrupted: + + def interrupt(_path: Path) -> None: + raise KeyboardInterrupt + + monkeypatch.setattr("skillspector.nodes.resolve_input.load_baseline", interrupt) + with pytest.raises(KeyboardInterrupt if interrupted else ValueError): + resolve_input({"input_path": str(archive), "use_shipped_baseline": True}) + assert created and all(not path.exists() for path in created) + + 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" diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index df03c596f..86ed840d2 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -4128,6 +4128,95 @@ def test_cli_shipped_baseline_opt_in(tmp_path: Path, extra: list[str]) -> None: assert "Applying author-shipped baseline" in result.stderr +@pytest.mark.parametrize("wrapped", [False, True]) +@pytest.mark.parametrize("format", ["json", "sarif"]) +@pytest.mark.parametrize("extra", [[], ["--transitive", "--transitive-depth", "0"]]) +def test_cli_zip_shipped_baseline_opt_in( + tmp_path: Path, wrapped: bool, format: str, extra: list[str] +) -> None: + """Load the extracted root baseline before analysis and preserve report audit trails.""" + skill_dir = _make_skill_dir(tmp_path) + archive = tmp_path / "skill.zip" + with zipfile.ZipFile(archive, "w") as bundle: + for path in skill_dir.iterdir(): + bundle.write(path, f"skill/{path.name}" if wrapped else path.name) + result = runner.invoke( + app, + ["scan", str(archive), "--no-llm", "--format", format, "--use-shipped-baseline", *extra], + ) + assert result.exit_code == 0, result.output + data = json.loads(result.stdout) + assert "Applying author-shipped baseline" in result.stderr + if format == "json": + assert data["issues"] == [] + assert data["suppressed_count"] >= 1 + assert data["analysis_completeness"]["is_complete"] is True + exclusions = data["analysis_completeness"]["scope_exclusions"] + assert any(event["reason_code"] == "baseline_file" for event in exclusions) + else: + validate_sarif_report(data) + results = data["runs"][0]["results"] + assert results and all(finding.get("suppressions") for finding in results) + + +@pytest.mark.parametrize("mode", ["default", "opt-in", "explicit", "missing-explicit"]) +def test_cli_zip_malformed_shipped_baseline_trust(tmp_path: Path, mode: str) -> None: + """Only author opt-in reads the shipped file; an explicit baseline takes precedence.""" + archive = tmp_path / "skill.zip" + with zipfile.ZipFile(archive, "w") as bundle: + bundle.writestr("SKILL.md", _SKILL_MD) + bundle.writestr(".skillspector-baseline.yaml", "version: 2\nrules: [{}]\n") + args = [] if mode == "default" else ["--use-shipped-baseline"] + if "explicit" in mode: + explicit = tmp_path / "explicit.yaml" + if mode == "explicit": + explicit.write_text(_SHIPPED_BASELINE_YAML) + args += ["--baseline", str(explicit)] + result = runner.invoke(app, ["scan", str(archive), "--no-llm", "--format", "json", *args]) + assert ( + result.exit_code == {"default": 1, "opt-in": 2, "explicit": 0, "missing-explicit": 2}[mode] + ) + if mode == "default": + assert json.loads(result.stdout)["suppressed_count"] == 0 + elif mode == "explicit": + assert json.loads(result.stdout)["suppressed_count"] >= 1 + assert "Applying author-shipped baseline" not in result.stderr + + +@pytest.mark.parametrize("failure", [OSError, KeyboardInterrupt]) +def test_cli_zip_baseline_notice_failure_cleans_extraction( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: type[BaseException] +) -> None: + """A failed post-graph notice must not strand a temporary tree before result handoff.""" + from skillspector.input_handler import InputHandler + + archive = tmp_path / "skill.zip" + with zipfile.ZipFile(archive, "w") as bundle: + bundle.writestr("SKILL.md", _SKILL_MD) + bundle.writestr(".skillspector-baseline.yaml", _SHIPPED_BASELINE_YAML) + created: list[Path] = [] + original = InputHandler._get_temp_dir + noticed: list[bool] = [] + + def track(handler: InputHandler) -> Path: + path = original(handler) + created.append(path) + return path + + def fail_notice(*_args: object) -> None: + noticed.append(True) + raise failure("notice failed") + + monkeypatch.setattr(InputHandler, "_get_temp_dir", track) + monkeypatch.setattr(cli, "_shipped_baseline_notice", fail_notice) + result = runner.invoke( + app, ["scan", str(archive), "--no-llm", "--format", "json", "--use-shipped-baseline"] + ) + assert noticed == [True] + assert result.exit_code != 0 + assert created and all(not path.exists() for path in created) + + def test_cli_shipped_baseline_discovered_equals_explicit(tmp_path: Path) -> None: """A discovered baseline yields the same result as the same file passed explicitly (R10/P5).""" skill_dir = _make_skill_dir(tmp_path)