diff --git a/conftest.py b/conftest.py new file mode 100644 index 000000000..f3487f2bc --- /dev/null +++ b/conftest.py @@ -0,0 +1,71 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Fork-owned pytest hook: xfail the upstream tests the exaforce patches invert. + +``exaforce/_schema_patches`` prunes keys from the LLM structured-output schema, +so the upstream tests that assert the un-pruned shape fail. + +Fork policy is to keep upstream test files byte-identical — no deselect +markers, no ``xfail`` decorators, no edits — so an upstream sync never +conflicts in them. This file satisfies that: it lives at the repo root, is not +upstream-tracked, and attaches the marker at collection time. + +``strict=True`` is the point. A doc that merely lists the expected failures +cannot tell a genuine regression from the expected noise, and cannot notice +when an upstream sync makes one of these pass again. Under strict xfail both +show up: an unexpected failure elsewhere in the file still fails the run, and +an XPASS here fails too — meaning the fork patch no longer changes this +behavior, so re-check the patch and drop the entry. + +``exaforce/_filter_patches`` needs no entry in its default ``semantic`` mode. +It lifts the CRITICAL/HIGH floor only for a finding that both carries a +``category`` (i.e. came from a static rule) and was actually adjudicated by the +meta-analyzer; every fixture in upstream's ``TestApplyFilterSeverityFloor`` +fails one of those two conditions, so those tests still pass. Under the opt-in +``SKILLSPECTOR_META_SEVERITY_FLOOR=none`` the floor is empty for everything, +which does invert one of them — marked conditionally below, so the run stays +green and strictly checked in that mode too. Fork-side coverage lives in +``tests/exaforce/test_patches.py``. + +Keep this list in sync with ``docs/superpowers/EXPECTED_TEST_FAILURES.md``. +""" + +from __future__ import annotations + +import pytest + +_SCHEMA_PRUNING_REASON = ( + "exaforce _schema_patches prunes explanation/remediation/intent from the " + "LLM structured-output schema; upstream asserts the un-pruned shape" +) + +_NO_FLOOR_REASON = ( + "SKILLSPECTOR_META_SEVERITY_FLOOR=none empties the floor for every finding, " + "including LLM-backed ones; upstream asserts the floor" +) + +_SCHEMA_PRUNING = ( + "tests/nodes/test_llm_analyzer_base.py::TestLLMAnalysisResult::test_to_finding", + "tests/nodes/test_llm_analyzer_base.py::TestLLMAnalysisResult::test_model_dump", + "tests/nodes/test_llm_analyzer_base.py::TestMetaAnalyzerResult::test_intent_validation", + "tests/nodes/test_semantic_quality_policy.py::TestFixtureMaliciousSkill" + "::test_malicious_skill_findings_preserve_metadata", +) + +# Only inverted by the fully-empty floor; passes under "semantic" and "upstream". +_NO_FLOOR_ONLY = ( + "tests/nodes/test_llm_analyzer_base.py::TestApplyFilterSeverityFloor" + "::test_critical_unconfirmed_kept_with_llm_unconfirmed_tag", +) + + +def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None: + """Attach ``xfail(strict=True)`` to the upstream tests the fork patches invert.""" + from skillspector.exaforce._filter_patches import resolve_mode + + expected = dict.fromkeys(_SCHEMA_PRUNING, _SCHEMA_PRUNING_REASON) + if resolve_mode() == "none": + expected.update(dict.fromkeys(_NO_FLOOR_ONLY, _NO_FLOOR_REASON)) + for item in items: + reason = expected.get(item.nodeid) + if reason is not None: + item.add_marker(pytest.mark.xfail(reason=reason, strict=True)) diff --git a/docs/superpowers/EXPECTED_TEST_FAILURES.md b/docs/superpowers/EXPECTED_TEST_FAILURES.md index 2ddc9cc00..233c1e70d 100644 --- a/docs/superpowers/EXPECTED_TEST_FAILURES.md +++ b/docs/superpowers/EXPECTED_TEST_FAILURES.md @@ -1,22 +1,65 @@ -# Expected test failures (fork: exaforce schema pruning) +# Expected test failures (fork: exaforce runtime patches) -These upstream tests are kept at upstream parity on purpose and therefore -assert the *un-pruned* schema, which the exaforce runtime patch removes. They -are expected to FAIL. A failure here is only a problem if the failure is NOT an -assertion about a pruned key (e.g. an import/collection error). +`exaforce/_schema_patches` prunes keys from the LLM structured-output schema, so +the four upstream tests that assert the un-pruned shape fail. Those tests are +kept at upstream parity on purpose — no deselect markers, no `xfail` +decorators, no edits — so an upstream sync never conflicts in them. -Captured from: -`uv run pytest tests/nodes/test_llm_analyzer_base.py tests/nodes/test_semantic_quality_policy.py -q` +They are marked `xfail(strict=True)` at collection time by the fork-owned +`conftest.py` at the repo root, which is not upstream-tracked. CI +(`.github/workflows/ci.yml` → `make test-ci`) therefore stays green, and the +expectation is machine-checked in both directions: -- tests/nodes/test_llm_analyzer_base.py::TestLLMAnalysisResult::test_to_finding -- tests/nodes/test_llm_analyzer_base.py::TestLLMAnalysisResult::test_model_dump -- tests/nodes/test_llm_analyzer_base.py::TestMetaAnalyzerResult::test_intent_validation -- tests/nodes/test_semantic_quality_policy.py::TestFixtureMaliciousSkill::test_malicious_skill_findings_preserve_metadata +- a genuine new failure in one of these files still fails the run, instead of + hiding inside a documented block of expected noise; +- if an upstream sync ever makes one of these pass again, `strict` turns the + XPASS into a failure — so the stale entry gets noticed rather than quietly + masking the fact that the fork patch no longer changes anything. -All four fail with an `AssertionError` (or `KeyError`) about a pruned key -(`explanation`, `intent`) being absent — not an import/collection error. -Confirmed bounded to these two files via `uv run pytest -q -rf`: +Keep this list in sync with `conftest.py`. + +## Schema pruning (`exaforce/_schema_patches.py`) + +Each fails with an `AssertionError` (or `KeyError`) about a pruned key +(`explanation`, `intent`) being absent — never an import/collection error. + +- `tests/nodes/test_llm_analyzer_base.py::TestLLMAnalysisResult::test_to_finding` +- `tests/nodes/test_llm_analyzer_base.py::TestLLMAnalysisResult::test_model_dump` +- `tests/nodes/test_llm_analyzer_base.py::TestMetaAnalyzerResult::test_intent_validation` +- `tests/nodes/test_semantic_quality_policy.py::TestFixtureMaliciousSkill::test_malicious_skill_findings_preserve_metadata` + +## Severity floor (`exaforce/_filter_patches.py`) — none in the default mode + +`_filter_patches` lifts the CRITICAL/HIGH floor only for a finding that both +carries a `category` — i.e. came from a static rule, since `LLMFinding.to_finding` +sets none — and was actually adjudicated by the meta-analyzer. Every fixture in +upstream's `TestApplyFilterSeverityFloor` fails one of those two conditions: +the fixtures build a bare `Finding` with no `category`, and the omission cases +pass an empty verdict list, which keeps the upstream floor by design (see the +module docstring). So under the default `SKILLSPECTOR_META_SEVERITY_FLOOR=semantic`, +and under `=upstream`, upstream's assertions all still hold — and they still +exercise the floored path rather than passing vacuously. + +The opt-in `=none` empties the floor for every finding regardless of source, +which inverts exactly one of them: + +- `tests/nodes/test_llm_analyzer_base.py::TestApplyFilterSeverityFloor::test_critical_unconfirmed_kept_with_llm_unconfirmed_tag` + +`conftest.py` marks that one only when the resolved mode is `none`, so the run +is green and strictly checked in all three modes. + +Fork-side coverage of the three modes, the empty-verdict case, and the +`category` invariant that makes this work lives in +`tests/exaforce/test_patches.py`. + +## Captured + +`uv run pytest -m "not integration and not provider" tests/ -q` (2026-09-08): ``` -4 failed, 1261 passed, 13 skipped, 34 deselected, 6 xfailed +1924 passed, 13 skipped, 38 deselected, 8 xfailed ``` + +The 8 xfailed are the 4 schema-pruning entries above plus 4 pre-existing +upstream xfails. With `SKILLSPECTOR_META_SEVERITY_FLOOR=none` it is +1923 passed / 9 xfailed. Nothing fails in any mode. diff --git a/src/skillspector/__init__.py b/src/skillspector/__init__.py index 27d7c776c..05b46e85d 100644 --- a/src/skillspector/__init__.py +++ b/src/skillspector/__init__.py @@ -38,4 +38,5 @@ # ExaForce fork: apply runtime schema/prompt patches (kept out of upstream files). from skillspector import exaforce as _exaforce # noqa: E402 + _exaforce.apply_patches() diff --git a/src/skillspector/exaforce/__init__.py b/src/skillspector/exaforce/__init__.py index 3583a9373..55f3ce017 100644 --- a/src/skillspector/exaforce/__init__.py +++ b/src/skillspector/exaforce/__init__.py @@ -2,14 +2,15 @@ """ExaForce fork-local runtime patches. Keeps fork behavior — pruning unused LLM structured-output keys and prompt text -to shrink requests and reduce LLM timeouts — out of upstream-tracked source -files. All mutations are guarded: an upstream rename/rewrite raises -``PatchDriftError`` at import time rather than silently going stale. +to shrink requests and reduce LLM timeouts, and trusting the meta-analyzer LLM +verdict over static severity — out of upstream-tracked source files. All +mutations are guarded: an upstream rename/rewrite raises ``PatchDriftError`` +at import time rather than silently going stale. """ from __future__ import annotations -from . import _prompt_patches, _sampling_patches, _schema_patches +from . import _filter_patches, _prompt_patches, _sampling_patches, _schema_patches _PATCHED = False @@ -22,4 +23,5 @@ def apply_patches() -> None: _schema_patches.apply() _prompt_patches.apply() _sampling_patches.apply() + _filter_patches.apply() _PATCHED = True diff --git a/src/skillspector/exaforce/_filter_patches.py b/src/skillspector/exaforce/_filter_patches.py new file mode 100644 index 000000000..dd0aaa487 --- /dev/null +++ b/src/skillspector/exaforce/_filter_patches.py @@ -0,0 +1,220 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Control how far the meta-analyzer LLM verdict overrides a finding's severity (fork behavior). + +Upstream ``LLMMetaAnalyzer.apply_filter`` keeps CRITICAL/HIGH findings even when +the meta-analyzer LLM denies or omits them, tagging them ``llm-unconfirmed``. +That floor is a prompt-injection defense: a skill's content could talk the LLM +into dropping a real finding. It also applies to *every* upstream finding, +including the LLM-backed analyzers (semantic ``SQP-*``/``SDI-*``/``SSD-*`` and +the ``TP4`` description-behavior check), so in practice it shields first-pass +LLM findings from a second LLM's re-verification as much as it shields static +regex hits. + +``SKILLSPECTOR_META_SEVERITY_FLOOR`` selects the policy, read on every +``apply_filter`` call so it can be flipped at runtime: + +``none`` + Empty the floor. Every finding, any severity, is dropped unless the + meta-analyzer confirms it. Fewest false positives, clearly worse recall. +``semantic`` (default) + Keep the upstream floor for LLM-backed findings only; static findings of + any severity follow the meta-analyzer verdict. This is the literal "trust + the LLM over static": the meta-analyzer may overrule a regex, but one LLM + does not silently overrule another. +``upstream`` + Leave the floor untouched. + +The floor policy applies only to findings the LLM actually adjudicated. +Batches that raise or never return never reach ``apply_filter`` at all — +upstream routes them through ``_fallback_filtered`` itself. A batch that +*returns* an empty verdict list, however, counts to upstream as a successful +"nothing confirmed" response, so lifting the floor there would drop every one +of its findings: a truncated or degenerate decode that still validates against +the schema would silently clear a file. Those findings therefore keep the +upstream floor — CRITICAL/HIGH retained and tagged ``llm-unconfirmed``, +MEDIUM/LOW dropped, exactly as ``upstream`` mode — and the event is logged. +Deciding it that way rather than diverting them to ``_fallback_filtered`` +keeps the fork's change confined to the floor: MEDIUM/LOW handling and the +``llm-unconfirmed`` tag stay bit-for-bit upstream in the no-verdict case. + +Measured 2026-09-02 on nvidia.nemotron-super-3-120b, same-day, two replicates +each, re-scanning the 94 borderline units (87 malicious / 7 benign) that a +900-unit ``none`` run had got wrong (188 unit-scans per mode): + + mode TP FP correct + upstream 68 10 72 + semantic 65 7 72 <- default: upstream accuracy, 30 % fewer FPs + none 54 3 65 + +The ``none`` losses are not "correct": spot-checked drops included obfuscated +PowerShell download-and-execute and Fernet-decrypted ``exec()`` in ``setup.py`` +that the semantic analyzers had flagged CRITICAL at confidence >= 0.9 and the +meta-analyzer then rejected. +""" + +from __future__ import annotations + +import functools +import inspect +import os +from typing import Any + +import skillspector.nodes.meta_analyzer as meta +from skillspector.logging_config import get_logger +from skillspector.models import Finding + +from ._patchlib import PatchDriftError + +logger = get_logger(__name__) + +ENV_VAR = "SKILLSPECTOR_META_SEVERITY_FLOOR" +MODES = ("none", "semantic", "upstream") +DEFAULT_MODE = "semantic" + +_UPSTREAM_FLOOR = frozenset({"CRITICAL", "HIGH"}) +_MISSING = object() + +# ``Finding.category`` is the discriminator: every static finding acquires one +# (``static_runner.analyzer_finding_to_finding`` falls back to +# ``get_category``, which returns "Security" for an unmapped rule id, and the +# MCP analyzers pass ``category=`` explicitly), while ``LLMFinding.to_finding`` +# — and the fork's pruned replacement in ``_schema_patches`` — set none. Rule +# ids are not used as the primary signal because they are free-form LLM output, +# never validated or normalized: a semantic analyzer that emitted ``SSD_1`` +# instead of ``SSD-2`` would lose the floor, which is the recall loss this +# module exists to avoid. ``TP4`` is the one LLM-backed finding built by hand +# with a category, so it is matched by id; the prefixes are kept as a +# belt-and-braces signal in case a future upstream starts populating +# ``category`` on LLM findings. Both extra checks can only *add* the floor, so +# the failure direction is a retained false positive, never a silent drop. +LLM_RULE_PREFIXES = ("SQP-", "SDI-", "SSD-") +LLM_RULE_IDS = frozenset({"TP4"}) + + +def resolve_mode() -> str: + raw = os.environ.get(ENV_VAR, "").strip().lower() + if not raw: + return DEFAULT_MODE + if raw not in MODES: + logger.warning("%s=%r is not one of %s — using %r.", ENV_VAR, raw, MODES, DEFAULT_MODE) + return DEFAULT_MODE + return raw + + +def is_llm_finding(finding: Finding) -> bool: + """Whether *finding* came from an LLM-backed analyzer rather than a static rule.""" + if getattr(finding, "category", None) is None: + return True + rule_id = (finding.rule_id or "").strip().upper() + return rule_id in LLM_RULE_IDS or rule_id.startswith(LLM_RULE_PREFIXES) + + +def _no_verdict_finding_ids(batch_results: list[Any], mode: str) -> set[str]: + """Ids of findings whose batch returned successfully but adjudicated nothing. + + Upstream treats an empty verdict list as "nothing confirmed", so with the + floor lifted every finding in such a batch would be dropped. The LLM never + actually ruled on them, so they keep the upstream floor instead. + """ + ids: set[str] = set() + for batch, llm_items in batch_results: + if batch.findings and not llm_items: + logger.warning( + "Meta-analyzer returned no verdicts for %s (%d findings); under " + "%s=%s they keep the upstream severity floor rather than being dropped.", + batch.file_path, + len(batch.findings), + ENV_VAR, + mode, + ) + ids.update(f.finding_id for f in batch.findings) + return ids + + +def _mode_dispatching_apply_filter(original: Any) -> Any: + """Wrap upstream ``apply_filter`` to apply the env-selected floor policy per call. + + Upstream reads the floor via ``self._HIGH_SEVERITY_FLOOR``; an instance + attribute shadows the class-level frozenset for the duration of one call and + any prior instance value is restored in ``finally``, so a raise inside + upstream code cannot leave the analyzer mis-configured. + """ + + @functools.wraps(original) + def apply_filter( + self: Any, + findings: list[Finding], + batch_results: list[tuple[Any, list[dict[str, Any]]]], + ) -> list[Finding]: + mode = resolve_mode() + if mode == "upstream": + return list(original(self, findings, batch_results)) + # Consumed more than once below, so materialize before the first pass. + batch_results = list(batch_results) + no_verdict_ids = _no_verdict_finding_ids(batch_results, mode) + + def keeps_floor(finding: Finding) -> bool: + if finding.finding_id in no_verdict_ids: + return True # never adjudicated; fail closed + return mode == "semantic" and is_llm_finding(finding) + + floored = [f for f in findings if keeps_floor(f)] + unfloored = [f for f in findings if not keeps_floor(f)] + kept: list[Finding] = [] + previous = self.__dict__.get("_HIGH_SEVERITY_FLOOR", _MISSING) + try: + if floored: + self._HIGH_SEVERITY_FLOOR = _UPSTREAM_FLOOR + kept.extend(original(self, floored, batch_results)) + self._HIGH_SEVERITY_FLOOR = frozenset() + kept.extend(original(self, unfloored, batch_results)) + finally: + if previous is _MISSING: + self.__dict__.pop("_HIGH_SEVERITY_FLOOR", None) + else: + self._HIGH_SEVERITY_FLOOR = previous + # Upstream forwards ``finding_id`` unchanged, so restore the caller's + # ordering by it — keeps the contract identical to upstream's single pass. + order = {f.finding_id: i for i, f in enumerate(findings)} + kept.sort(key=lambda f: order.get(f.finding_id, len(order))) + return kept + + apply_filter._exaforce_wrapped = True # type: ignore[attr-defined] + return apply_filter + + +def apply() -> None: + cls = meta.LLMMetaAnalyzer + qual = f"{cls.__module__}.{cls.__qualname__}" + if cls.__dict__.get("_HIGH_SEVERITY_FLOOR") != _UPSTREAM_FLOOR: + raise PatchDriftError( + f"{qual}._HIGH_SEVERITY_FLOOR is not {sorted(_UPSTREAM_FLOOR)}; " + "upstream changed — update the exaforce patch." + ) + current = cls.__dict__.get("apply_filter") + if current is None: + raise PatchDriftError( + f"{qual}.apply_filter is missing; upstream changed — update the exaforce patch." + ) + if getattr(current, "_exaforce_wrapped", False): + return # already applied + # The per-call shadowing above only works if upstream reads the floor + # through the instance. Fail at import time if that access path changes. + try: + source = inspect.getsource(current) + except OSError: + # No source on disk: zipimport, a frozen build, a .pyc-only deploy. + # The guard is unverifiable there, but refusing to import would take + # the whole CLI down, so proceed and say so. + logger.warning( + "Cannot read the source of %s.apply_filter to verify it still reads " + "self._HIGH_SEVERITY_FLOOR; applying the exaforce floor patch unverified.", + qual, + ) + else: + if "self._HIGH_SEVERITY_FLOOR" not in source: + raise PatchDriftError( + f"{qual}.apply_filter no longer reads self._HIGH_SEVERITY_FLOOR; " + "upstream changed — update the exaforce patch." + ) + setattr(cls, "apply_filter", _mode_dispatching_apply_filter(current)) # noqa: B010 diff --git a/src/skillspector/exaforce/_schema_patches.py b/src/skillspector/exaforce/_schema_patches.py index 5d8c6c45c..0c842351f 100644 --- a/src/skillspector/exaforce/_schema_patches.py +++ b/src/skillspector/exaforce/_schema_patches.py @@ -15,7 +15,7 @@ from ._patchlib import pop_field_validator, remove_model_fields -def _pruned_to_finding(self: "llm_base.LLMFinding", file: str) -> Finding: +def _pruned_to_finding(self: llm_base.LLMFinding, file: str) -> Finding: """``LLMFinding.to_finding`` without the removed explanation/remediation.""" return Finding( rule_id=self.rule_id, diff --git a/tests/exaforce/test_patches.py b/tests/exaforce/test_patches.py index e089e84ec..de5d93ba0 100644 --- a/tests/exaforce/test_patches.py +++ b/tests/exaforce/test_patches.py @@ -78,3 +78,242 @@ def test_apply_patches_is_idempotent(run_in_subprocess): """ ) assert "OK" in out + + +_FLOOR_HARNESS = """ + import os + os.environ["SKILLSPECTOR_META_SEVERITY_FLOOR"] = {mode!r} + from unittest.mock import MagicMock, patch + import skillspector + from skillspector.exaforce import apply_patches + apply_patches() + from skillspector.llm_analyzer_base import Batch + from skillspector.models import Finding + from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer + + # ``category`` is what separates the two sources, so the fixture mirrors how + # findings are really built: static analyzers always end up with one + # (``analyzer_finding_to_finding`` falls back to "Security"), the semantic + # analyzers' ``LLMFinding.to_finding`` sets none, and ``TP4`` is hand-built + # with one. See test_llm_findings_have_no_category for the invariant. + def _finding(rule_id, severity, line, category="Security"): + return Finding( + rule_id=rule_id, message="msg", severity=severity, + confidence=0.8, file="skill.md", start_line=line, category=category, + ) + + with patch("skillspector.llm_analyzer_base.get_chat_model", lambda **kw: MagicMock()): + analyzer = LLMMetaAnalyzer(model="test-model") + + # Static rules (regex/YARA style ids) and semantic rules (SQP/SDI/SSD), each + # with a CRITICAL the LLM explicitly denies, a HIGH the LLM omits, and a + # MEDIUM the LLM confirms. + findings = [ + _finding("E2", "CRITICAL", 10), _finding("PE3", "HIGH", 11), _finding("P1", "MEDIUM", 12), + _finding("SDI-1", "CRITICAL", 20, None), _finding("SSD-1", "HIGH", 21, None), + _finding("SQP-2", "MEDIUM", 22, None), + _finding("TP4", "HIGH", 30), # LLM-backed, but carries a category; LLM omits it + ] + batch = Batch(file_path="skill.md", content="code", findings=findings) + llm_items = [ + {{"pattern_id": "E2", "start_line": 10, "is_vulnerability": False, "confidence": 0.2, "_file": "skill.md"}}, + {{"pattern_id": "P1", "start_line": 12, "is_vulnerability": True, "confidence": 0.9, "_file": "skill.md"}}, + {{"pattern_id": "SDI-1", "start_line": 20, "is_vulnerability": False, "confidence": 0.2, "_file": "skill.md"}}, + {{"pattern_id": "SQP-2", "start_line": 22, "is_vulnerability": True, "confidence": 0.9, "_file": "skill.md"}}, + ] + result = analyzer.apply_filter(findings, [(batch, llm_items)]) + kept = [f.rule_id for f in result] + unconfirmed = sorted(f.rule_id for f in result if "llm-unconfirmed" in f.tags) + assert kept == {expected_kept!r}, kept + assert unconfirmed == {expected_unconfirmed!r}, unconfirmed + print("OK") +""" + + +def test_floor_mode_none_drops_every_unconfirmed_finding(run_in_subprocess): + out = run_in_subprocess( + _FLOOR_HARNESS.format( + mode="none", + expected_kept=["P1", "SQP-2"], + expected_unconfirmed=[], + ) + ) + assert "OK" in out + + +def test_floor_mode_semantic_keeps_only_semantic_high_severity(run_in_subprocess): + """Static CRITICAL/HIGH follow the LLM verdict; semantic CRITICAL/HIGH keep the + upstream floor. Output order matches input order despite the two-pass filter.""" + out = run_in_subprocess( + _FLOOR_HARNESS.format( + mode="semantic", + expected_kept=["P1", "SDI-1", "SSD-1", "SQP-2", "TP4"], + expected_unconfirmed=["SDI-1", "SSD-1", "TP4"], + ) + ) + assert "OK" in out + + +def test_floor_mode_upstream_is_untouched(run_in_subprocess): + out = run_in_subprocess( + _FLOOR_HARNESS.format( + mode="upstream", + expected_kept=["E2", "PE3", "P1", "SDI-1", "SSD-1", "SQP-2", "TP4"], + expected_unconfirmed=["E2", "PE3", "SDI-1", "SSD-1", "TP4"], + ) + ) + assert "OK" in out + + +def test_floor_mode_default_is_semantic_and_switchable_at_runtime(run_in_subprocess): + """Mode is read per apply_filter call, so an in-process A/B can flip it after import.""" + out = run_in_subprocess( + """ + import os + os.environ.pop("SKILLSPECTOR_META_SEVERITY_FLOOR", None) + from unittest.mock import MagicMock, patch + import skillspector + from skillspector.exaforce import _filter_patches + from skillspector.llm_analyzer_base import Batch + from skillspector.models import Finding + from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer + assert _filter_patches.resolve_mode() == "semantic" + assert getattr(LLMMetaAnalyzer.apply_filter, "_exaforce_wrapped", False) + with patch("skillspector.llm_analyzer_base.get_chat_model", lambda **kw: MagicMock()): + analyzer = LLMMetaAnalyzer(model="test-model") + f = Finding(rule_id="E2", message="m", severity="CRITICAL", confidence=0.8, + file="skill.md", start_line=1, category="Security") + batch = Batch(file_path="skill.md", content="c", findings=[f]) + denied = [{"pattern_id": "E2", "start_line": 1, "is_vulnerability": False, + "confidence": 0.1, "_file": "skill.md"}] + assert analyzer.apply_filter([f], [(batch, denied)]) == [] # semantic: static dropped + os.environ["SKILLSPECTOR_META_SEVERITY_FLOOR"] = "upstream" + assert len(analyzer.apply_filter([f], [(batch, denied)])) == 1 # upstream: floor kept + os.environ["SKILLSPECTOR_META_SEVERITY_FLOOR"] = "none" + assert analyzer.apply_filter([f], [(batch, denied)]) == [] + assert "_HIGH_SEVERITY_FLOOR" not in analyzer.__dict__ # instance state restored + print("OK") + """ + ) + assert "OK" in out + + +def test_empty_verdict_batch_keeps_the_upstream_floor(run_in_subprocess): + """A batch that returns successfully but adjudicates nothing must not clear the file. + + Upstream counts an empty verdict list as "nothing confirmed", so lifting the + floor would drop every finding in that batch — a truncated or degenerate + decode that still validates against the schema would silently report a + malicious file clean. Those findings keep the upstream floor instead: + CRITICAL/HIGH retained and tagged, MEDIUM/LOW dropped, exactly as upstream. + """ + out = run_in_subprocess( + """ + import os + os.environ["SKILLSPECTOR_META_SEVERITY_FLOOR"] = "none" # the most aggressive mode + from unittest.mock import MagicMock, patch + import skillspector + from skillspector.llm_analyzer_base import Batch + from skillspector.models import Finding + from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer + + with patch("skillspector.llm_analyzer_base.get_chat_model", lambda **kw: MagicMock()): + analyzer = LLMMetaAnalyzer(model="test-model") + + def _finding(rule_id, severity, file, line): + return Finding(rule_id=rule_id, message="m", severity=severity, + confidence=0.8, file=file, start_line=line, + category="Security") + + # a.py: the LLM returned an empty verdict list. b.py: a real denial. + empty = [_finding("E2", "CRITICAL", "a.py", 1), _finding("P1", "MEDIUM", "a.py", 2)] + adjudicated = [_finding("PE3", "HIGH", "b.py", 5)] + batch_a = Batch(file_path="a.py", content="c", findings=empty) + batch_b = Batch(file_path="b.py", content="c", findings=adjudicated) + denied_b = [{"pattern_id": "PE3", "start_line": 5, "is_vulnerability": False, + "confidence": 0.1, "_file": "b.py"}] + + result = analyzer.apply_filter( + empty + adjudicated, [(batch_a, []), (batch_b, denied_b)] + ) + kept = [f.rule_id for f in result] + # E2 survives the empty verdict on the floor and is tagged; P1 is MEDIUM so + # the floor does not cover it; PE3 was genuinely adjudicated and follows the + # verdict even though it is HIGH. + assert kept == ["E2"], kept + assert "llm-unconfirmed" in result[0].tags, result[0].tags + print("OK") + """ + ) + assert "OK" in out + + +def test_llm_findings_have_no_category_and_static_findings_do(run_in_subprocess): + """The invariant ``is_llm_finding`` relies on, asserted against real constructors. + + If an upstream sync starts setting ``category`` on LLM findings, or stops + setting it on static ones, the semantic mode silently mis-routes findings. + This fails instead. + """ + out = run_in_subprocess( + """ + import skillspector # applies the fork patches, incl. pruned to_finding + from skillspector.exaforce._filter_patches import is_llm_finding + from skillspector.llm_analyzer_base import LLMFinding + from skillspector.nodes.analyzers.static_runner import analyzer_finding_to_finding + from skillspector.models import AnalyzerFinding, Location, Severity + + llm = LLMFinding(rule_id="SSD-2", message="m", severity="CRITICAL", + confidence=0.9, start_line=1).to_finding("skill.md") + assert llm.category is None, llm.category + assert is_llm_finding(llm) + + # A rule id no LLM prefix matches still routes as LLM-backed on category + # alone — the fail-safe direction (floor retained, never silently lost). + odd = LLMFinding(rule_id="Semantic-Prompt-Injection", message="m", + severity="CRITICAL", confidence=0.9, start_line=1).to_finding("s.md") + assert is_llm_finding(odd) + + static = analyzer_finding_to_finding( + AnalyzerFinding(rule_id="YR1", message="m", severity=Severity.HIGH, + confidence=0.9, location=Location(file="SKILL.md", start_line=65)) + ) + assert static.category is not None, static.category + assert not is_llm_finding(static) + + # An unmapped static rule id must still get a category, so it does not + # fall through to the LLM branch by accident. + unmapped = analyzer_finding_to_finding( + AnalyzerFinding(rule_id="ZZ-999", message="m", severity=Severity.HIGH, + confidence=0.9, location=Location(file="SKILL.md", start_line=1)) + ) + assert unmapped.category is not None, unmapped.category + print("OK") + """ + ) + assert "OK" in out + + +def test_import_survives_unreadable_apply_filter_source(run_in_subprocess): + """The drift guard reads ``inspect.getsource``; no source must not kill the CLI. + + Under zipimport, a frozen build, or a ``.pyc``-only deploy, ``getsource`` + raises ``OSError``. That must not propagate out of ``import skillspector``. + """ + out = run_in_subprocess( + """ + import inspect + _real = inspect.getsource + def _boom(obj): + raise OSError("could not get source code") + inspect.getsource = _boom + try: + import skillspector # must not raise + from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer + assert getattr(LLMMetaAnalyzer.apply_filter, "_exaforce_wrapped", False) + finally: + inspect.getsource = _real + print("OK") + """ + ) + assert "OK" in out