diff --git a/src/skillspector/nodes/analyzers/static_yara.py b/src/skillspector/nodes/analyzers/static_yara.py index 5de066b1a..b00b27e6a 100644 --- a/src/skillspector/nodes/analyzers/static_yara.py +++ b/src/skillspector/nodes/analyzers/static_yara.py @@ -28,6 +28,7 @@ import math import os import stat +import threading import time from collections.abc import Callable from contextvars import ContextVar @@ -46,6 +47,7 @@ InspectionLedgerEvent, LedgerOutcome, LedgerReason, + LedgerRecordType, analyzer_status_event, ledger_event, ) @@ -173,8 +175,18 @@ def _enforce_rule_load_deadline() -> None: # Module-level cache keyed by a content hash of all rule directories. +# +# These three are one logical value: the compiled rules, the hash they were +# compiled from, and how many files were dropped producing them. They must only +# ever be written or read as a set, under ``_RULES_LOCK`` -- see +# :func:`load_rules_with_skips` for why reading them separately is unsafe. _compiled_rules: yara.Rules | None = None _rules_hash: str | None = None +_rules_skipped_count: int = 0 + +# Reentrant so the load-and-read transaction in :func:`load_rules_with_skips` +# can hold it across its own call to :func:`_load_rules`. +_RULES_LOCK = threading.RLock() def _collect_rule_files(*dirs: Path) -> list[Path]: @@ -334,13 +346,36 @@ def _read_rule_source(rule_file: Path, data: bytes | None = None) -> str: return base64.b64decode("".join(encoded_source.split())).decode("utf-8") +#: Cap on how much of a decode/compile error is echoed into logs. Rule sources +#: are attacker-influenced when ``--yara-rules-dir`` points at untrusted content, +#: and YARA syntax errors can quote the offending source line, so the reason is +#: truncated rather than passed through whole. +MAX_RULE_REJECTION_REASON_CHARS = 200 + + +def _bounded_rejection_reason(exc: Exception) -> str: + """Return a single-line, length-capped description of a rule rejection.""" + reason = " ".join(str(exc).split()) + if len(reason) > MAX_RULE_REJECTION_REASON_CHARS: + reason = f"{reason[:MAX_RULE_REJECTION_REASON_CHARS]}..." + return reason or exc.__class__.__name__ + + def _build_namespace_map( rule_files: list[Path], temp_dir: Path | None = None, *, raw_cache: dict[Path, bytes] | None = None, + namespace_files: dict[str, str] | None = None, ) -> tuple[dict[str, str], int]: - """Build a {namespace: source} dict and count malformed rule files.""" + """Build a {namespace: source} dict and count malformed rule files. + + If ``namespace_files`` is given it is populated with ``{namespace: filename}`` + so a later compile failure can name the file the operator has to fix -- a + namespace has its extension stripped, so it is not a usable filename on its + own. Passed in rather than returned to keep this function's two-value + signature, which existing callers and tests unpack directly. + """ del temp_dir sources: dict[str, str] = {} skipped = 0 @@ -351,17 +386,35 @@ def _build_namespace_map( ns = _rule_namespace(rf) if ns in sources: ns = f"{rf.parent.name}/{ns}" + if namespace_files is not None: + namespace_files[ns] = rf.name try: sources[ns] = _read_rule_source(rf, raw_cache[rf]) except (binascii.Error, UnicodeDecodeError, ValueError) as exc: skipped += 1 - logger.debug("%s: skipping malformed encoded rule %s: %s", ANALYZER_ID, rf, exc) + # WARNING, not DEBUG: a dropped rule silently removes a detector, so + # the operator has to be able to identify and repair the file from a + # default-level run (#554). The filename is named explicitly because + # the ledger event is scoped to the rule set, not to one file. + logger.warning( + "%s: rejected rule file %s (could not decode): %s", + ANALYZER_ID, + rf.name, + _bounded_rejection_reason(exc), + ) return sources, skipped -def _compile_rules(sources: dict[str, str]) -> tuple[yara.Rules | None, int]: +def _compile_rules( + sources: dict[str, str], + *, + namespace_files: dict[str, str] | None = None, +) -> tuple[yara.Rules | None, int]: """Compile YARA rules from a namespace map. Falls back to per-source compilation on error. + ``namespace_files`` maps namespace to filename so a rejection can name the + file the operator has to fix rather than its extension-stripped namespace. + Returns (compiled_rules, skipped_count). """ _enforce_rule_load_deadline() @@ -382,7 +435,14 @@ def _compile_rules(sources: dict[str, str]) -> tuple[yara.Rules | None, int]: good[ns] = source except (yara.SyntaxError, yara.Error) as exc: skipped += 1 - logger.debug("%s: skipping %s: %s", ANALYZER_ID, ns, exc) + # WARNING for the same reason as the decode path above: without it a + # broken detector disappears with no default-level trace (#554). + logger.warning( + "%s: rejected rule file %s (could not compile): %s", + ANALYZER_ID, + (namespace_files or {}).get(ns, ns), + _bounded_rejection_reason(exc), + ) _enforce_rule_load_deadline() compiled = yara.compile(sources=good) if good else None @@ -394,38 +454,94 @@ def _load_rules(extra_dir: Path | None = None) -> yara.Rules | None: """Compile YARA rules from built-in and optional user-supplied directories. Results are cached at module level and reused if directory contents haven't changed. + + Rule files that fail to decode (malformed base64) or fail to compile (YARA + syntax errors) are dropped from the active rule set. The count is recorded + in the module-level ``_rules_skipped_count`` (read via + :func:`rules_skipped_count`) rather than returned here, so this keeps its + original single-value signature and every existing + ``monkeypatch.setattr(static_yara, "_load_rules", ...)`` test double stays + valid; callers that care about the skip count must surface it themselves + or a scan can report ``completed``/SAFE while some of its own detections + never ran (#554). + + Callers should prefer :func:`load_rules_with_skips`, which returns both + halves as one value; reading the count separately after this returns is + racy across concurrent scans. """ - global _compiled_rules, _rules_hash # noqa: PLW0603 + global _compiled_rules, _rules_hash, _rules_skipped_count # noqa: PLW0603 - dirs = [_BUILTIN_RULES_DIR] - if extra_dir and extra_dir.is_dir(): - dirs.append(extra_dir) - elif extra_dir: - logger.warning("%s: user rules directory %s does not exist", ANALYZER_ID, extra_dir) + with _RULES_LOCK: + dirs = [_BUILTIN_RULES_DIR] + if extra_dir and extra_dir.is_dir(): + dirs.append(extra_dir) + elif extra_dir: + logger.warning("%s: user rules directory %s does not exist", ANALYZER_ID, extra_dir) - rule_files = _collect_rule_files(*dirs) - if not rule_files: - logger.info("%s: no YARA rule files found", ANALYZER_ID) - return None + rule_files = _collect_rule_files(*dirs) + if not rule_files: + logger.info("%s: no YARA rule files found", ANALYZER_ID) + _rules_skipped_count = 0 + return None + + raw_cache = _read_rule_bytes_cache(rule_files) + current_hash = _content_hash(rule_files, raw_cache) + if _compiled_rules is not None and _rules_hash == current_hash: + # Cache hit: _rules_skipped_count already describes this exact file + # set, because it is only ever written together with _rules_hash. + return _compiled_rules + + namespace_files: dict[str, str] = {} + sources, materialize_skipped = _build_namespace_map( + rule_files, raw_cache=raw_cache, namespace_files=namespace_files + ) + compiled, compile_skipped = _compile_rules(sources, namespace_files=namespace_files) + skipped = materialize_skipped + compile_skipped + _rules_skipped_count = skipped + + if compiled is None: + logger.warning("%s: failed to compile any YARA rules", ANALYZER_ID) + return None + + _compiled_rules = compiled + _rules_hash = current_hash + loaded = len(sources) - compile_skipped + logger.info("%s: compiled %d YARA rule file(s) (%d skipped)", ANALYZER_ID, loaded, skipped) + return compiled + + +def load_rules_with_skips(extra_dir: Path | None = None) -> tuple[yara.Rules | None, int]: + """Load rules and return them with their own skip count, as one value. + + The two halves must be obtained in a single locked transaction. Reading the + count separately after :func:`_load_rules` returns lets two concurrent + MCP/graph scans interleave: scan A loads rule set A, scan B loads rule set B + and overwrites the module-level count, then scan A reads B's count. Scan A + would then run rules A while reporting B's skip total -- and if B skipped + nothing, A reports ``completed`` even though one of A's own rules was + dropped, which is exactly the false-clean result #554 is about. + + :func:`_load_rules` is called through the module global so existing + ``monkeypatch.setattr(static_yara, "_load_rules", ...)`` doubles still apply. + """ + with _RULES_LOCK: + rules = _load_rules(extra_dir) + return rules, _rules_skipped_count - raw_cache = _read_rule_bytes_cache(rule_files) - current_hash = _content_hash(rule_files, raw_cache) - if _compiled_rules is not None and _rules_hash == current_hash: - return _compiled_rules - sources, materialize_skipped = _build_namespace_map(rule_files, raw_cache=raw_cache) - compiled, compile_skipped = _compile_rules(sources) - skipped = materialize_skipped + compile_skipped +def rules_skipped_count() -> int: + """Return how many rule files the most recent :func:`_load_rules` call dropped. - if compiled is None: - logger.warning("%s: failed to compile any YARA rules", ANALYZER_ID) - return None + Zero both when nothing was skipped and when a cache hit meant no reload + ran; a cache hit implies the same file set was already validated by the + load that populated the cache, so nothing new could have been skipped. - _compiled_rules = compiled - _rules_hash = current_hash - loaded = len(sources) - compile_skipped - logger.info("%s: compiled %d YARA rule file(s) (%d skipped)", ANALYZER_ID, loaded, skipped) - return compiled + Retained for callers that already hold :data:`_RULES_LOCK` or run + single-threaded. Anything reading this straight after :func:`_load_rules` + should use :func:`load_rules_with_skips` instead. + """ + with _RULES_LOCK: + return _rules_skipped_count def _bounded_match_instances( @@ -916,7 +1032,9 @@ def _rule_limit_response( ) deadline_token = _RULE_LOAD_DEADLINE.set(load_budget) try: - rules = _load_rules(extra_dir) + # One transaction: the skip count must describe *these* rules, not + # whatever a concurrent scan loaded in between. + rules, rules_skipped = load_rules_with_skips(extra_dir) except _YaraRuleResourceLimitError as exc: return _rule_limit_response(exc.reason, dict(exc.metrics)) finally: @@ -1072,6 +1190,41 @@ def _rule_limit_response( ) logger.info("%s: %d findings", ANALYZER_ID, len(findings)) + if rules_skipped: + # A rule that fails to compile or decode is dropped from the active + # set with no per-file signal: every scanned component can still + # report COMPLETED, because the rule that would have flagged it + # simply never ran. Surface that as its own ledger event, scoped to + # the rule directory rather than a skill file, so it isn't silently + # absorbed into a clean-looking events list (#554). + events.append( + ledger_event( + # analyzer_id is deliberately omitted. ledger_event derives the + # work identity as ``analyzer_id or f"{record_type}:{phase}"``, + # so passing it would identify this event as + # ``static_yara`` + path -- identical to the planned work item + # for a *scanned component of the same name*. A skill file + # literally named ``yara_rules`` then collides with this event, + # both planned targets resolve to two matching events, and + # reconciliation raises a fatal ``unaccounted_work`` instead of + # the nonfatal partial scan this is meant to record. Falling + # back to ``system:static`` makes the identity disjoint from + # every analyzer work item by construction, so no choice of + # filename can collide -- renaming the synthetic path alone + # would only move the collision to the next unlucky name. + outcome=LedgerOutcome.PARTIAL, + record_type=LedgerRecordType.SYSTEM, + phase="static", + # Not a scanned skill file: a synthetic scope for the rule + # set itself. Ledger paths must be relative POSIX paths, and + # the real rules directory (builtin or --yara-rules-dir) is + # absolute, so it cannot be used here. + path="yara_rules/", + reason=LedgerReason.READ_ERROR, + observed_artifacts=rules_skipped, + limit_artifacts=0, + ) + ) if not events: status = analyzer_status_event( analyzer_id=ANALYZER_ID, diff --git a/tests/nodes/analyzers/test_static_yara.py b/tests/nodes/analyzers/test_static_yara.py index e3fd7cba0..b8601d3d6 100644 --- a/tests/nodes/analyzers/test_static_yara.py +++ b/tests/nodes/analyzers/test_static_yara.py @@ -23,6 +23,8 @@ import base64 import json +import logging +import threading from pathlib import Path from unittest.mock import MagicMock @@ -37,12 +39,19 @@ @pytest.fixture(autouse=True) def _clear_rule_cache(): - """Reset the module-level compiled rules cache between tests.""" + """Reset the module-level compiled rules cache between tests. + + The skip count is part of that cache: it is only meaningful alongside the + hash it was produced from, so leaving it set would leak a previous test's + dropped-rule total into the next one. + """ static_yara._compiled_rules = None static_yara._rules_hash = None + static_yara._rules_skipped_count = 0 yield static_yara._compiled_rules = None static_yara._rules_hash = None + static_yara._rules_skipped_count = 0 def _write_rule( @@ -953,6 +962,49 @@ def test_build_namespace_map_skips_malformed_encoded_rules(self, tmp_path): assert "invalid" not in ns_map assert skipped == 1 + def test_malformed_rule_is_reported_not_silently_dropped(self, tmp_path, monkeypatch): + """A custom rule that can't compile must not report a clean, SAFE scan (#554). + + Reproduces the issue's own scenario: a valid rule plus a rule with a + YARA syntax error in the same --yara-rules-dir. The good rule must + still fire, but the analyzer status must not be "completed" -- that + claim would be false, since the broken rule never ran against + anything. + """ + static_yara._compiled_rules = None + static_yara._rules_hash = None + static_yara._rules_skipped_count = 0 + monkeypatch.setattr(static_yara, "_BUILTIN_RULES_DIR", tmp_path / "empty_builtin") + (tmp_path / "empty_builtin").mkdir() + + rules_dir = tmp_path / "rules" + rules_dir.mkdir() + (rules_dir / "good.yar").write_text( + 'rule good_rule { meta: category = "malware" ' + 'strings: $a = "ACME_CANARY" condition: $a }' + ) + # Missing closing brace: a real YARA syntax error, not a decode failure. + (rules_dir / "bad.yar").write_text('rule bad_rule { strings: $a = "x" condition: $a') + + result = static_yara.node( + { + "components": ["skill.md"], + "file_cache": {"skill.md": "contains ACME_CANARY"}, + "yara_rules_dir": str(rules_dir), + } + ) + + assert any("good_rule" in f.message for f in result["findings"]), ( + "the valid rule must still fire" + ) + status = result["analyzer_status_events"][0] + assert status["status"] != "completed", "a dropped custom rule must not report a clean scan" + assert any( + event.get("reason_code") == LedgerReason.READ_ERROR + and event.get("observed_artifacts") == 1 + for event in result["inspection_ledger"] + ) + @pytest.mark.parametrize("payload", ["not base64", "not base64 é"]) def test_malformed_extra_encoded_rule_does_not_block_builtin_rules(self, tmp_path, payload): (tmp_path / "bad.yar.b64").write_text(payload) @@ -1322,3 +1374,238 @@ def test_yara_does_not_start_without_one_enforceable_engine_second(self) -> None match.assert_not_called() assert matched.reason == "runtime_limit" assert matched.metrics == {"observed_seconds": 0.0, "limit_seconds": 0.5} + + +class TestRuleSkipAccounting: + """Regressions for the three review findings on the #554 skip-count surface. + + All three share one root shape: the dropped-rule total was reported through + channels not tied to the scan that produced it -- a module global read after + the fact, a ledger work ID shared with component work, and a DEBUG log the + operator never sees at default verbosity. + """ + + @staticmethod + def _isolated_builtin(tmp_path: Path, monkeypatch) -> None: + """Point the built-in rule dir at an empty dir so counts are only ours.""" + builtin = tmp_path / "empty_builtin" + builtin.mkdir(exist_ok=True) + monkeypatch.setattr(static_yara, "_BUILTIN_RULES_DIR", builtin) + + @staticmethod + def _rule_dir(tmp_path: Path, name: str, *, broken: int) -> Path: + """Build a rule dir with one valid rule and ``broken`` uncompilable ones.""" + rules_dir = tmp_path / name + rules_dir.mkdir(parents=True, exist_ok=True) + marker = f"MARKER_{name.upper()}" + (rules_dir / "good.yar").write_text( + f'rule good_{name} {{ strings: $a = "{marker}" condition: $a }}' + ) + for index in range(broken): + # Missing closing brace: a real YARA syntax error, not a decode failure. + (rules_dir / f"bad{index}.yar").write_text( + f'rule bad_{name}_{index} {{ strings: $a = "x" condition: $a' + ) + return rules_dir + + def test_skip_count_travels_with_the_rules_it_describes(self, tmp_path, monkeypatch): + """Two loads in sequence must each report their own skip total. + + Deterministic form of the concurrency finding: reading the count as a + separate step after the load is what lets a later load answer for an + earlier one. ``load_rules_with_skips`` returns both halves together, so + the pairing cannot be broken by anything that happens afterwards. + """ + self._isolated_builtin(tmp_path, monkeypatch) + dir_a = self._rule_dir(tmp_path, "a", broken=1) + dir_b = self._rule_dir(tmp_path, "b", broken=0) + + rules_a, skipped_a = static_yara.load_rules_with_skips(dir_a) + rules_b, skipped_b = static_yara.load_rules_with_skips(dir_b) + + assert rules_a is not None + assert rules_b is not None + assert skipped_a == 1, "rule set A dropped one rule and must say so" + assert skipped_b == 0, "rule set B dropped nothing and must not inherit A's count" + + # The separate-read path is what made this unsafe: after B's load the + # module global describes B, so anyone still holding A's rules and + # reading the global now would report a clean scan for A. + assert static_yara.rules_skipped_count() == 0 + + def test_load_and_read_is_serialized_against_other_scans(self, tmp_path, monkeypatch): + """The load-and-read pair must be atomic, not merely adjacent. + + Proves the lock is genuinely held across the whole transaction rather + than racing threads and hoping, so the test cannot pass by luck of + timing: mid-transaction, another thread must not be able to acquire the + rules lock at all. + """ + self._isolated_builtin(tmp_path, monkeypatch) + dir_a = self._rule_dir(tmp_path, "a", broken=2) + + lock_was_held: list[bool] = [] + real_load = static_yara._load_rules + + def probing_load(extra_dir=None): + rules = real_load(extra_dir) + acquired_elsewhere: list[bool] = [] + + def try_acquire() -> None: + got = static_yara._RULES_LOCK.acquire(blocking=False) + acquired_elsewhere.append(got) + if got: + static_yara._RULES_LOCK.release() + + probe = threading.Thread(target=try_acquire) + probe.start() + probe.join() + lock_was_held.append(not acquired_elsewhere[0]) + return rules + + monkeypatch.setattr(static_yara, "_load_rules", probing_load) + _, skipped = static_yara.load_rules_with_skips(dir_a) + + assert skipped == 2 + assert lock_was_held == [True], ( + "another scan could enter the load-and-read transaction, so the rules " + "and their skip count are not obtained atomically" + ) + + def test_concurrent_scans_never_report_another_rule_sets_count(self, tmp_path, monkeypatch): + """Under real contention every scan must still see its own total.""" + self._isolated_builtin(tmp_path, monkeypatch) + dir_a = self._rule_dir(tmp_path, "a", broken=1) + dir_b = self._rule_dir(tmp_path, "b", broken=0) + + mismatches: list[tuple[str, int, int]] = [] + failures: list[BaseException] = [] + observations = 0 + + def scan(label: str, rules_dir: Path, expected: int) -> None: + nonlocal observations + try: + for _ in range(25): + _, skipped = static_yara.load_rules_with_skips(rules_dir) + observations += 1 + if skipped != expected: + mismatches.append((label, expected, skipped)) + except BaseException as exc: # noqa: BLE001 - re-raised in the main thread + failures.append(exc) + + threads = [ + threading.Thread(target=scan, args=("A", dir_a, 1)), + threading.Thread(target=scan, args=("B", dir_b, 0)), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + # An exception inside a worker thread does not fail the test on its own, + # so it is surfaced explicitly -- otherwise this test passes vacuously + # when the scans never actually ran. + assert failures == [], f"a scan thread raised: {failures!r}" + assert observations == 50, f"expected 50 observations, made {observations}" + assert mismatches == [], f"scans observed another rule set's skip count: {mismatches}" + + def test_rule_load_event_does_not_collide_with_a_component_of_the_same_name( + self, tmp_path, monkeypatch + ): + """A skill file named ``yara_rules`` must not collide with the rule-load event. + + The ledger derives a work ID from ``analyzer_id`` plus the normalized + path. The synthetic rule-set scope normalizes to ``yara_rules``, so + attributing the event to ``static_yara`` gave it the same work ID as a + scanned component of that name: both planned targets then resolved to two + matching events and reconciliation raised a fatal ``unaccounted_work`` + instead of recording a nonfatal partial scan. Renaming the synthetic path + alone would only move the collision to the next unlucky filename. + """ + self._isolated_builtin(tmp_path, monkeypatch) + rules_dir = self._rule_dir(tmp_path, "r", broken=1) + + result = static_yara.node( + { + "components": ["yara_rules"], + "file_cache": {"yara_rules": "contains MARKER_R"}, + "yara_rules_dir": str(rules_dir), + } + ) + + events = result["inspection_ledger"] + work_ids = [event["work_id"] for event in events] + assert len(work_ids) == len(set(work_ids)), ( + "the rule-load event shares a work ID with the scanned component" + ) + + # The planned work the status advertises must be equally distinct, since + # reconciliation requires exactly one event per planned target. + planned = result["analyzer_status_events"][0]["planned_work"] + planned_ids = [target["work_id"] for target in planned] + assert len(planned_ids) == len(set(planned_ids)) + + # The dropped rule is still surfaced, and the scan is partial not clean. + assert result["analyzer_status_events"][0]["status"] != "completed" + assert any( + event.get("reason_code") == LedgerReason.READ_ERROR + and event.get("observed_artifacts") == 1 + for event in events + ) + + @pytest.mark.parametrize( + ("filename", "content", "expected_fragment"), + [ + ("acme.yar", b'rule broken { strings: $a = "x" condition: $a', "could not compile"), + ( + "bom.yar", + b'\xef\xbb\xbfrule bomrule { strings: $a = "y" condition: $a }', + "could not compile", + ), + ( + "bad_utf8.yar", + b'rule u { strings: $a = "\xff\xfe" condition: $a }', + "could not decode", + ), + ], + ) + def test_rejected_rule_is_named_at_default_log_level( + self, tmp_path, monkeypatch, caplog, filename, content, expected_fragment + ): + """Each rejected rule must be reported at WARNING, naming the file (#554). + + A dropped rule removes a detector. At DEBUG the operator gets no signal + at default verbosity, and the ledger event is scoped to the rule set + rather than to one file, so without this the specific file that needs + repairing cannot be identified. + """ + self._isolated_builtin(tmp_path, monkeypatch) + rules_dir = tmp_path / "rejected" + rules_dir.mkdir() + (rules_dir / filename).write_bytes(content) + + with caplog.at_level(logging.WARNING, logger=static_yara.logger.name): + static_yara._load_rules(rules_dir) + + rejections = [ + record.getMessage() + for record in caplog.records + if record.levelno == logging.WARNING and "rejected rule file" in record.getMessage() + ] + assert len(rejections) == 1, f"expected one rejection warning, got {rejections}" + assert filename in rejections[0], f"the warning must name {filename}: {rejections[0]}" + assert expected_fragment in rejections[0] + + def test_rejection_reason_is_length_bounded(self): + """Rule sources can be untrusted, so the echoed reason must be capped.""" + reason = static_yara._bounded_rejection_reason(ValueError("x" * 5_000)) + + assert len(reason) <= static_yara.MAX_RULE_REJECTION_REASON_CHARS + 3 + assert reason.endswith("...") + + def test_rejection_reason_collapses_newlines(self): + """A multi-line YARA error must stay one log line.""" + reason = static_yara._bounded_rejection_reason(ValueError("line one\nline two\r\nthree")) + + assert "\n" not in reason + assert reason == "line one line two three"