cache: handle invalid caches

capa/rules/cache.py

@@ -127,5 +127,13 @@ def load_cached_ruleset(cache_dir: str, rule_contents: List[bytes]) -> Optional[
     logger.debug("loading rule set from cache: %s", path)
     with open(path, "rb") as f:
         buf = f.read()
+
+    try:
         cache = RuleCache.load(buf)
+    except AssertionError:
+        logger.debug("rule set cache is invalid: %s", path)
+        # delete the cache that seems to be invalid.
+        os.remove(path)
+        return None
+    else:
         return cache.ruleset

tests/test_rule_cache.py

@@ -84,3 +84,32 @@ def test_ruleset_cache_save_load():
     assert os.path.exists(path)
 
     assert capa.rules.cache.load_cached_ruleset(cache_dir, content) is not None
+
+
+def test_ruleset_cache_invalid():
+    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)
+    try:
+        os.remove(path)
+    except OSError:
+        pass
+
+    capa.rules.cache.cache_ruleset(cache_dir, rs)
+    assert os.path.exists(path)
+
+    with open(path, "rb") as f:
+        buf = f.read()
+
+    # corrupt the magic header
+    buf = b"x" + buf[1:]
+
+    with open(path, "wb") as f:
+        f.write(buf)
+
+    assert os.path.exists(path)
+    assert capa.rules.cache.load_cached_ruleset(cache_dir, content) is None
+    # the invalid cache should be deleted
+    assert not os.path.exists(path)