tests: add snapshot tests for feature extraction

Introduces data-driven snapshot tests that regenerate capa freeze files
for a curated set of samples in the tests/data submodule and compare the
bytes against committed fixtures under tests/fixtures/freezes/. Any change
that perturbs feature extraction surfaces as a test failure with a feature-
count delta and a truncated unified diff.

Author: claude

CHANGELOG.md

@@ -3,6 +3,9 @@
 ## master (unreleased)
 
 ### New Features
+- ci: add support for arm64 binary releases
+- tests: add feature snapshot tests under `tests/fixtures/snapshots/features/`
+- freeze: add `--reproducible` flag that zeros dynamic header metadata
 
 ### Breaking Changes
 
@@ -113,7 +116,7 @@
 - tests: update binja version to 5.3 @mr-tz #3011
 - ci: use explicit and per job permissions @mike-hunhoff #3002
 - replace black/isort/flake8 with ruff @mike-hunhoff #2992
-
+- tests: add snapshot tests for feature extraction. @williballenthin #3054
 - ci: update GitHub Actions to support Node.js 24 (deprecate Node.js 20) @mr-tz #2984
 
 ### Raw diffs
@@ -258,6 +261,7 @@ Additionally a Binary Ninja bug has been fixed. Released binaries now include AR
 - nursery/get-dotnet-assembly-entry-point mehunhoff@google.com
 
 ### Bug Fixes
+- dotnetfile: sort namespace features so freeze output is deterministic across runs
 
 - binja: fix a crash during feature extraction when the MLIL is unavailable @xusheng6 #2714 
 

capa/features/extractors/dotnetfile.py

@@ -98,7 +98,7 @@ def extract_file_namespace_features(pe: dnfile.dnPE, **kwargs) -> Iterator[tuple
     # namespaces may be empty, discard
     namespaces.discard("")
 
-    for namespace in namespaces:
+    for namespace in sorted(namespaces):
         # namespace do not have an associated token, so we yield 0x0
         yield Namespace(namespace), NO_ADDRESS
 

capa/features/freeze/__init__.py

@@ -92,10 +92,7 @@ def from_capa(cls, a: capa.features.address.Address) -> "Address":
             return cls(type=AddressType.THREAD, value=(a.process.ppid, a.process.pid, a.tid))
 
         elif isinstance(a, capa.features.address.DynamicCallAddress):
-            return cls(
-                type=AddressType.CALL,
-                value=(a.thread.process.ppid, a.thread.process.pid, a.thread.tid, a.id),
-            )
+            return cls(type=AddressType.CALL, value=(a.thread.process.ppid, a.thread.process.pid, a.thread.tid, a.id))
 
         elif a == capa.features.address.NO_ADDRESS or isinstance(a, capa.features.address._NoAddress):
             return cls(type=AddressType.NO_ADDRESS, value=None)
@@ -144,17 +141,15 @@ def to_capa(self) -> capa.features.address.Address:
             assert isinstance(pid, int)
             assert isinstance(tid, int)
             return capa.features.address.ThreadAddress(
-                process=capa.features.address.ProcessAddress(ppid=ppid, pid=pid),
-                tid=tid,
+                process=capa.features.address.ProcessAddress(ppid=ppid, pid=pid), tid=tid
             )
 
         elif self.type is AddressType.CALL:
             assert isinstance(self.value, tuple)
             ppid, pid, tid, id_ = self.value
             return capa.features.address.DynamicCallAddress(
                 thread=capa.features.address.ThreadAddress(
-                    process=capa.features.address.ProcessAddress(ppid=ppid, pid=pid),
-                    tid=tid,
+                    process=capa.features.address.ProcessAddress(ppid=ppid, pid=pid), tid=tid
                 ),
                 id=id_,
             )
@@ -180,6 +175,20 @@ def __lt__(self, other: "Address") -> bool:
             return self.value < other.value  # type: ignore
 
 
+def _addr_sort_key(a: Address) -> tuple:
+    """
+    Canonical, comparable sort key for an Address.
+
+    We don't rely on Address.__lt__ here because it returns True for
+    NO_ADDRESS < NO_ADDRESS, which breaks strict weak ordering for sort.
+    """
+    if a.value is None:
+        return (a.type.value, ())
+    if isinstance(a.value, int):
+        return (a.type.value, (a.value,))
+    return (a.type.value, tuple(a.value))
+
+
 class GlobalFeature(HashableModel):
     feature: Feature
 
@@ -346,9 +355,14 @@ class Freeze(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
 
 
-def dumps_static(extractor: StaticFeatureExtractor) -> str:
+def dumps_static(extractor: StaticFeatureExtractor, *, reproducible: bool = False) -> str:
     """
     serialize the given extractor to a string
+
+    When `reproducible` is true, the freeze's dynamic header metadata (e.g. the
+    embedded capa version) is zeroed out so that output is identical across
+    capa versions for a given extractor. This is used by the feature snapshot
+    tests to keep fixtures stable across version bumps.
     """
     global_features: list[GlobalFeature] = []
     for feature, _ in extractor.extract_global_features():
@@ -357,6 +371,7 @@ def dumps_static(extractor: StaticFeatureExtractor) -> str:
                 feature=feature_from_capa(feature),
             )
         )
+    global_features.sort(key=lambda gf: gf.feature.model_dump_json())
 
     file_features: list[FileFeature] = []
     for feature, address in extractor.extract_file_features():
@@ -366,6 +381,7 @@ def dumps_static(extractor: StaticFeatureExtractor) -> str:
                 address=Address.from_capa(address),
             )
         )
+    file_features.sort(key=lambda ff: (_addr_sort_key(ff.address), ff.feature.model_dump_json()))
 
     function_features: list[FunctionFeatures] = []
     for f in extractor.get_functions():
@@ -378,6 +394,7 @@ def dumps_static(extractor: StaticFeatureExtractor) -> str:
             )
             for feature, addr in extractor.extract_function_features(f)
         ]
+        ffeatures.sort(key=lambda ff: (_addr_sort_key(ff.address), ff.feature.model_dump_json()))
 
         basic_blocks = []
         for bb in extractor.get_basic_blocks(f):
@@ -390,6 +407,7 @@ def dumps_static(extractor: StaticFeatureExtractor) -> str:
                 )
                 for feature, addr in extractor.extract_basic_block_features(f, bb)
             ]
+            bbfeatures.sort(key=lambda bf: (_addr_sort_key(bf.address), bf.feature.model_dump_json()))
 
             instructions = []
             for insn in extractor.get_instructions(f, bb):
@@ -402,6 +420,7 @@ def dumps_static(extractor: StaticFeatureExtractor) -> str:
                     )
                     for feature, addr in extractor.extract_insn_features(f, bb, insn)
                 ]
+                ifeatures.sort(key=lambda i: (_addr_sort_key(i.address), i.feature.model_dump_json()))
 
                 instructions.append(
                     InstructionFeatures(
@@ -410,6 +429,9 @@ def dumps_static(extractor: StaticFeatureExtractor) -> str:
                     )
                 )
 
+            # sort by address so regeneration is obviously idempotent regardless of
+            # any per-extractor iteration quirks.
+            instructions.sort(key=lambda i: _addr_sort_key(i.address))
             basic_blocks.append(
                 BasicBlockFeatures(
                     address=bbaddr,
@@ -418,6 +440,7 @@ def dumps_static(extractor: StaticFeatureExtractor) -> str:
                 )
             )
 
+        basic_blocks.sort(key=lambda bb: _addr_sort_key(bb.address))
         function_features.append(
             FunctionFeatures(
                 address=faddr,
@@ -426,28 +449,33 @@ def dumps_static(extractor: StaticFeatureExtractor) -> str:
             )
         )
 
+    function_features.sort(key=lambda ff: _addr_sort_key(ff.address))
+
     features = StaticFeatures(
         global_=global_features,  # type: ignore[call-arg]  # pydantic alias "global" not recognized by type checkers
         file=tuple(file_features),
         functions=tuple(function_features),
     )
 
+    extractor_version = "" if reproducible else capa.version.__version__
     freeze = Freeze(
         version=CURRENT_VERSION,
         base_address=Address.from_capa(extractor.get_base_address()),  # type: ignore[call-arg]  # pydantic alias "base address" not recognized by type checkers
         sample_hashes=extractor.get_sample_hashes(),
         flavor="static",
-        extractor=Extractor(name=extractor.__class__.__name__),
+        extractor=Extractor(name=extractor.__class__.__name__, version=extractor_version),
         features=features,
     )
     # type checkers are unable to recognise `base_address` as an argument due to alias
 
     return freeze.model_dump_json()
 
 
-def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
+def dumps_dynamic(extractor: DynamicFeatureExtractor, *, reproducible: bool = False) -> str:
     """
     serialize the given extractor to a string
+
+    See `dumps_static` for `reproducible`.
     """
     global_features: list[GlobalFeature] = []
     for feature, _ in extractor.extract_global_features():
@@ -456,6 +484,7 @@ def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
                 feature=feature_from_capa(feature),
             )
         )
+    global_features.sort(key=lambda gf: gf.feature.model_dump_json())
 
     file_features: list[FileFeature] = []
     for feature, address in extractor.extract_file_features():
@@ -465,6 +494,7 @@ def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
                 address=Address.from_capa(address),
             )
         )
+    file_features.sort(key=lambda ff: (_addr_sort_key(ff.address), ff.feature.model_dump_json()))
 
     process_features: list[ProcessFeatures] = []
     for p in extractor.get_processes():
@@ -478,6 +508,7 @@ def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
             )
             for feature, addr in extractor.extract_process_features(p)
         ]
+        pfeatures.sort(key=lambda pf: (_addr_sort_key(pf.address), pf.feature.model_dump_json()))
 
         threads = []
         for t in extractor.get_threads(p):
@@ -490,6 +521,7 @@ def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
                 )
                 for feature, addr in extractor.extract_thread_features(p, t)
             ]
+            tfeatures.sort(key=lambda tf: (_addr_sort_key(tf.address), tf.feature.model_dump_json()))
 
             calls = []
             for call in extractor.get_calls(p, t):
@@ -503,6 +535,7 @@ def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
                     )
                     for feature, addr in extractor.extract_call_features(p, t, call)
                 ]
+                cfeatures.sort(key=lambda cf: (_addr_sort_key(cf.address), cf.feature.model_dump_json()))
 
                 calls.append(
                     CallFeatures(
@@ -512,6 +545,9 @@ def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
                     )
                 )
 
+            # sort by address so regeneration is obviously idempotent regardless of
+            # any per-extractor iteration quirks.
+            calls.sort(key=lambda c: _addr_sort_key(c.address))
             threads.append(
                 ThreadFeatures(
                     address=taddr,
@@ -520,6 +556,7 @@ def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
                 )
             )
 
+        threads.sort(key=lambda t: _addr_sort_key(t.address))
         process_features.append(
             ProcessFeatures(
                 address=paddr,
@@ -529,6 +566,8 @@ def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
             )
         )
 
+    process_features.sort(key=lambda pf: _addr_sort_key(pf.address))
+
     features = DynamicFeatures(
         global_=global_features,  # type: ignore[call-arg]  # pydantic alias "global" not recognized by type checkers
         file=tuple(file_features),
@@ -539,12 +578,13 @@ def dumps_dynamic(extractor: DynamicFeatureExtractor) -> str:
     get_base_addr = getattr(extractor, "get_base_address", None)
     base_addr = get_base_addr() if get_base_addr else capa.features.address.NO_ADDRESS
 
+    extractor_version = "" if reproducible else capa.version.__version__
     freeze = Freeze(
         version=CURRENT_VERSION,
         base_address=Address.from_capa(base_addr),  # type: ignore[call-arg]  # pydantic alias "base address" not recognized by type checkers
         sample_hashes=extractor.get_sample_hashes(),
         flavor="dynamic",
-        extractor=Extractor(name=extractor.__class__.__name__),
+        extractor=Extractor(name=extractor.__class__.__name__, version=extractor_version),
         features=features,
     )
     # type checkers are unable to recognise `base_address` as an argument due to alias
@@ -627,21 +667,21 @@ def loads_dynamic(s: str) -> DynamicFeatureExtractor:
 MAGIC = "capa0000".encode("ascii")
 
 
-def dumps(extractor: FeatureExtractor) -> str:
+def dumps(extractor: FeatureExtractor, *, reproducible: bool = False) -> str:
     """serialize the given extractor to a string."""
     if isinstance(extractor, StaticFeatureExtractor):
-        doc = dumps_static(extractor)
+        doc = dumps_static(extractor, reproducible=reproducible)
     elif isinstance(extractor, DynamicFeatureExtractor):
-        doc = dumps_dynamic(extractor)
+        doc = dumps_dynamic(extractor, reproducible=reproducible)
     else:
         raise ValueError("Invalid feature extractor")
 
     return doc
 
 
-def dump(extractor: FeatureExtractor) -> bytes:
+def dump(extractor: FeatureExtractor, *, reproducible: bool = False) -> bytes:
     """serialize the given extractor to a byte array."""
-    return MAGIC + zlib.compress(dumps(extractor).encode("utf-8"))
+    return MAGIC + zlib.compress(dumps(extractor, reproducible=reproducible).encode("utf-8"))
 
 
 def is_freeze(buf: bytes) -> bool:
@@ -685,6 +725,11 @@ def main(argv=None):
     parser = argparse.ArgumentParser(description="save capa features to a file")
     capa.main.install_common_args(parser, {"input_file", "format", "backend", "os", "signatures"})
     parser.add_argument("output", type=str, help="Path to output file")
+    parser.add_argument(
+        "--reproducible",
+        action="store_true",
+        help="zero out dynamic header metadata (e.g. capa version) so output is stable across capa versions",
+    )
     args = parser.parse_args(args=argv)
 
     try:
@@ -696,11 +741,43 @@ def main(argv=None):
     except capa.main.ShouldExitError as e:
         return e.status_code
 
-    Path(args.output).write_bytes(dump(extractor))
+    output_path = Path(args.output)
+    output_path.write_bytes(dump(extractor, reproducible=args.reproducible))
+
+    # Log a manifest entry for the feature snapshot tests at INFO level. This
+    # makes it easy to copy/paste into
+    # `tests/fixtures/snapshots/features/manifest.json` when adding a new
+    # fixture or refreshing an existing one.
+    entry: dict[str, str] = {
+        "name": output_path.stem,
+        "sample": str(args.input_file),
+        "freeze": output_path.name,
+    }
+    if args.format and args.format != "auto":
+        entry["format"] = args.format
+    if args.backend and args.backend != "auto":
+        entry["backend"] = args.backend
+    if args.os and args.os != "auto":
+        entry["os"] = args.os
+    commit = _git_head_commit()
+    if commit:
+        entry["generated_at_commit"] = commit
+    logger.info("manifest entry: %s", json.dumps(entry))
 
     return 0
 
 
+def _git_head_commit() -> str:
+    """Return the HEAD commit, or empty string if this isn't a git checkout."""
+    import subprocess
+
+    try:
+        out = subprocess.check_output(["git", "rev-parse", "HEAD"], stderr=subprocess.DEVNULL)
+    except (subprocess.CalledProcessError, FileNotFoundError, OSError):
+        return ""
+    return out.decode("ascii", errors="replace").strip()
+
+
 if __name__ == "__main__":
     import sys
 

capa/features/freeze/__main__.py

@@ -0,0 +1,19 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+from capa.features.freeze import main
+
+sys.exit(main())