diff --git a/capa/rules/cache.py b/capa/rules/cache.py index a461997610..59457f6327 100644 --- a/capa/rules/cache.py +++ b/capa/rules/cache.py @@ -78,7 +78,18 @@ def get_cache_path(cache_dir: Path, id: CacheIdentifier) -> Path: MAGIC = b"capa" -VERSION = b"\x00\x00\x00\x01" +VERSION = b"\x00\x00\x00\x02" + + +def _is_valid_cached_ruleset(ruleset: capa.rules.RuleSet) -> bool: + """Return False when a cached ruleset uses an outdated internal schema.""" + feature_indexes = getattr(ruleset, "_feature_indexes_by_scopes", None) + if feature_indexes is None: + return False + for feature_index in feature_indexes.values(): + if not hasattr(feature_index, "bytes_prefix_index"): + return False + return True @dataclass @@ -158,12 +169,17 @@ def load_cached_ruleset(cache_dir: Path, rule_contents: list[bytes]) -> Optional try: cache = RuleCache.load(buf) - except AssertionError: + except (AssertionError, AttributeError, EOFError, pickle.UnpicklingError, ValueError, TypeError): logger.debug("rule set cache is invalid: %s", path) # delete the cache that seems to be invalid. path.unlink() return None + if not _is_valid_cached_ruleset(cache.ruleset): + logger.debug("rule set cache uses an outdated schema: %s", path) + path.unlink() + return None + return cache.ruleset diff --git a/tests/test_rule_cache.py b/tests/test_rule_cache.py index 3aa0c97df9..568d7375a0 100644 --- a/tests/test_rule_cache.py +++ b/tests/test_rule_cache.py @@ -63,6 +63,12 @@ ) +class StaleFeatureIndex: + def __init__(self, index): + self.rules_by_feature = index.rules_by_feature + self.string_rules = index.string_rules + + def test_ruleset_cache_ids(): rs = capa.rules.RuleSet([R1]) content = capa.rules.cache.get_ruleset_content(rs) @@ -94,6 +100,43 @@ def test_ruleset_cache_save_load(): assert capa.rules.cache.load_cached_ruleset(cache_dir, content) is not None +def test_ruleset_cache_rejects_outdated_schema(): + rs = capa.rules.RuleSet([R1]) + content = capa.rules.cache.get_ruleset_content(rs) + id = capa.rules.cache.compute_cache_identifier(content) + cache_dir = capa.rules.cache.get_default_cache_directory() + path = capa.rules.cache.get_cache_path(cache_dir, id) + with contextlib.suppress(OSError): + path.unlink() + + for scope, index in list(rs._feature_indexes_by_scopes.items()): + rs._feature_indexes_by_scopes[scope] = StaleFeatureIndex(index) + + cache = capa.rules.cache.RuleCache(id, rs) + path.write_bytes(cache.dump()) + assert path.exists() + + assert capa.rules.cache.load_cached_ruleset(cache_dir, content) is None + assert not path.exists() + + +def test_ruleset_cache_rejects_missing_feature_indexes(): + rs = capa.rules.RuleSet([R1]) + content = capa.rules.cache.get_ruleset_content(rs) + id = capa.rules.cache.compute_cache_identifier(content) + cache_dir = capa.rules.cache.get_default_cache_directory() + path = capa.rules.cache.get_cache_path(cache_dir, id) + with contextlib.suppress(OSError): + path.unlink() + + cache = capa.rules.cache.RuleCache(id, object()) + path.write_bytes(cache.dump()) + assert path.exists() + + assert capa.rules.cache.load_cached_ruleset(cache_dir, content) is None + assert not path.exists() + + def test_ruleset_cache_invalid(): rs = capa.rules.RuleSet([R1]) content = capa.rules.cache.get_ruleset_content(rs)