rules: discard rules with provably-incompatible OS/arch/format constraints before analysis begins

CHANGELOG.md

@@ -32,6 +32,7 @@
 
 ### Development
 
+- rules: prune rules with incompatible OS/arch/format requirements before analysis to skip them across all scopes #2127
 - ci: deprecate macos-13 runner and use Python v3.13 for testing @mike-hunhoff #2777
 
 ### Raw diffs

capa/capabilities/dynamic.py

@@ -279,6 +279,16 @@ def find_dynamic_capabilities(
 
     feature_counts = rdoc.DynamicFeatureCounts(file=0, processes=())
 
+    # Prune rules that cannot match this binary's global features (OS, arch, format)
+    # once, before the per-process matching loop.  This eliminates rules that could
+    # never match — e.g. Windows-specific rules when analysing a Linux trace — from all
+    # per-process, per-thread, and per-call evaluations.
+    # See: https://github.com/mandiant/capa/issues/2127
+    global_features: FeatureSet = collections.defaultdict(set)
+    for feature, addr in extractor.extract_global_features():
+        global_features[feature].add(addr)
+    ruleset = ruleset.filter_rules_by_meta_features(global_features)
+
     assert isinstance(extractor, DynamicFeatureExtractor)
     processes: list[ProcessHandle] = list(extractor.get_processes())
     n_processes: int = len(processes)

capa/capabilities/static.py

@@ -159,6 +159,16 @@ def find_static_capabilities(
     feature_counts = rdoc.StaticFeatureCounts(file=0, functions=())
     library_functions: tuple[rdoc.LibraryFunction, ...] = ()
 
+    # Prune rules that cannot match this binary's global features (OS, arch, format)
+    # once, before the per-function matching loop.  This eliminates rules that could
+    # never match — e.g. Windows-specific rules when analysing a Linux ELF — from all
+    # per-function, per-basic-block, and per-instruction evaluations.
+    # See: https://github.com/mandiant/capa/issues/2127
+    global_features: FeatureSet = collections.defaultdict(set)
+    for feature, addr in extractor.extract_global_features():
+        global_features[feature].add(addr)
+    ruleset = ruleset.filter_rules_by_meta_features(global_features)
+
     assert isinstance(extractor, StaticFeatureExtractor)
     functions: list[FunctionHandle] = list(extractor.get_functions())
     n_funcs: int = len(functions)

capa/rules/__init__.py

@@ -1909,6 +1909,94 @@ def filter_rules_by_meta(self, tag: str) -> "RuleSet":
                             break
         return RuleSet(list(rules_filtered))
 
+    def filter_rules_by_meta_features(self, features: FeatureSet) -> "RuleSet":
+        """
+        Return a new RuleSet with rules removed whose global-feature requirements
+        cannot be satisfied by the binary under analysis.
+
+        Global features — OS, architecture, and format — are determined once at the
+        start of analysis from the binary's headers.  Any rule that requires, for
+        example, ``os: windows`` while we are analyzing a Linux ELF can never match
+        and is safe to discard before the per-function matching loop begins.
+
+        The filtering is conservative: a rule is only removed when its global-feature
+        constraints are *provably* unsatisfiable.  Rules with no global-feature
+        constraints, or with ``os: any``-style wildcards, are always kept.
+
+        Rules that are kept as transitive dependencies of other kept rules are also
+        retained, so the returned RuleSet always satisfies internal dependency
+        invariants.
+
+        Args:
+            features: the global FeatureSet for the binary (typically the output of
+                ``extractor.extract_global_features()``).
+
+        Returns:
+            A new :class:`RuleSet` with incompatible rules removed, or *self* if
+            no rules were pruned.
+        """
+        global_features: FeatureSet = {
+            feature: locations
+            for feature, locations in features.items()
+            if capa.features.common.is_global_feature(feature)
+        }
+
+        if not global_features:
+            return self
+
+        def can_match(node) -> bool:
+            """
+            Return True if *node* might be satisfiable given the known global features.
+            Returns False only when provably unsatisfiable.
+            """
+            if isinstance(node, capa.features.common.Feature):
+                if capa.features.common.is_global_feature(node):
+                    return bool(node.evaluate(global_features))
+                return True
+
+            if isinstance(node, ceng.Not):
+                return True
+
+            if isinstance(node, ceng.And):
+                return all(can_match(child) for child in node.children)
+
+            if isinstance(node, (ceng.Or, ceng.Some)):
+                if isinstance(node, ceng.Some) and node.count == 0:
+                    return True
+                return any(can_match(child) for child in node.children)
+
+            if isinstance(node, ceng.Range):
+                if node.min == 0:
+                    return True
+                return can_match(node.child)
+
+            return True
+
+        compatible_rule_names = {rule.name for rule in self.rules.values() if can_match(rule.statement)}
+
+        if len(compatible_rule_names) == len(self.rules):
+            return self
+
+        # Collect the surviving rules plus all of their transitive dependencies
+        # to ensure RuleSet dependency invariants are maintained.
+        all_rules = list(self.rules.values())
+        rules_to_keep: set[str] = set()
+        for rule_name in compatible_rule_names:
+            rules_to_keep.update(r.name for r in get_rules_and_dependencies(all_rules, rule_name))
+
+        pruned_count = len(self.rules) - len(rules_to_keep)
+        if pruned_count == 0:
+            return self
+
+        logger.debug(
+            "pruned %d rules incompatible with global features (%s)",
+            pruned_count,
+            ", ".join(f"{f.name}: {f.value}" for f in global_features),
+        )
+
+        surviving_rules = [self.rules[name] for name in rules_to_keep]
+        return RuleSet(surviving_rules)
+
     # this routine is unstable and may change before the next major release.
     @staticmethod
     def _sort_rules_by_index(rule_index_by_rule_name: dict[str, int], rules: list[Rule]):

tests/test_match.py

@@ -21,7 +21,7 @@
 import capa.features.insn
 import capa.features.common
 from capa.rules import Scope
-from capa.features.common import OS, OS_ANY, OS_WINDOWS, String, MatchedRule
+from capa.features.common import OS, OS_ANY, OS_LINUX, OS_WINDOWS, String, MatchedRule
 
 
 def match(rules, features, va, scope=Scope.FUNCTION):
@@ -887,3 +887,75 @@ def test_index_features_nested_unstable():
 
     assert not index.string_rules
     assert not index.bytes_rules
+
+
+def test_filter_rules_by_meta_features_prunes_incompatible_os():
+    """Rules requiring a different OS than the binary's are removed from the RuleSet."""
+    windows_rule = textwrap.dedent(
+        """
+        rule:
+            meta:
+                name: windows only rule
+                scopes:
+                    static: function
+                    dynamic: process
+            features:
+                - and:
+                    - os: windows
+                    - api: CreateFile
+        """
+    )
+    linux_rule = textwrap.dedent(
+        """
+        rule:
+            meta:
+                name: linux only rule
+                scopes:
+                    static: function
+                    dynamic: process
+            features:
+                - and:
+                    - os: linux
+                    - api: open
+        """
+    )
+    rr = capa.rules.RuleSet(
+        [
+            capa.rules.Rule.from_yaml(windows_rule),
+            capa.rules.Rule.from_yaml(linux_rule),
+        ]
+    )
+    assert len(rr.rules) == 2
+
+    # When analyzing a Linux binary, windows-only rules are pruned
+    linux_features = {OS(OS_LINUX): {0x0}}
+    filtered = rr.filter_rules_by_meta_features(linux_features)
+    assert "linux only rule" in filtered.rules
+    assert "windows only rule" not in filtered.rules
+
+    # When analyzing a Windows binary, linux-only rules are pruned
+    windows_features = {OS(OS_WINDOWS): {0x0}}
+    filtered = rr.filter_rules_by_meta_features(windows_features)
+    assert "windows only rule" in filtered.rules
+    assert "linux only rule" not in filtered.rules
+
+
+def test_filter_rules_by_meta_features_keeps_any_os():
+    """Rules with os: any or no OS requirement are kept regardless of binary OS."""
+    any_os_rule = textwrap.dedent(
+        """
+        rule:
+            meta:
+                name: cross-platform rule
+                scopes:
+                    static: function
+                    dynamic: process
+            features:
+                - api: malloc
+        """
+    )
+    rr = capa.rules.RuleSet([capa.rules.Rule.from_yaml(any_os_rule)])
+
+    windows_features = {OS(OS_WINDOWS): {0x0}}
+    filtered = rr.filter_rules_by_meta_features(windows_features)
+    assert "cross-platform rule" in filtered.rules