extract dynamic capabilities

.github/mypy/mypy.ini

@@ -1,5 +1,11 @@
 [mypy]
 
+exclude = (?x)(
+    ^capa/render/proto/__init__.py$
+    | ^tests/_test_proto.py$
+    | ^capa/ida/helpers.py$
+  )
+
 [mypy-halo.*]
 ignore_missing_imports = True
 

CHANGELOG.md

@@ -16,6 +16,7 @@
 - publish via PyPI trusted publishing #1491 @williballenthin
 - migrate to pyproject.toml #1301 @williballenthin
 - Add ProcessesAddress and ThreadAddress #1612 @yelhamer
+- Add dynamic capability extraction @yelhamer
 
 ### Breaking Changes
 - Update Metadata type in capa main [#1411](https://github.com/mandiant/capa/issues/1411) [@Aayush-Goel-04](https://github.com/aayush-goel-04) @manasghandat

capa/features/extractors/base_extractor.py

@@ -7,6 +7,7 @@
 # See the License for the specific language governing permissions and limitations under the License.
 
 import abc
+import hashlib
 import dataclasses
 from typing import Any, Dict, Tuple, Union, Iterator
 from dataclasses import dataclass
@@ -24,6 +25,29 @@
 # the feature extractor from which they were created.
 
 
+@dataclass
+class SampleHashes:
+    md5: str
+    sha1: str
+    sha256: str
+
+    def __iter__(self) -> Iterator[str]:
+        yield self.md5
+        yield self.sha1
+        yield self.sha256
+
+    @classmethod
+    def from_sample(cls, buf) -> "SampleHashes":
+        md5 = hashlib.md5()
+        sha1 = hashlib.sha1()
+        sha256 = hashlib.sha256()
+        md5.update(buf)
+        sha1.update(buf)
+        sha256.update(buf)
+
+        return cls(md5=md5.hexdigest(), sha1=sha1.hexdigest(), sha256=sha256.hexdigest())
+
+
 @dataclass
 class FunctionHandle:
     """reference to a function recognized by a feature extractor.
@@ -104,6 +128,14 @@ def get_base_address(self) -> Union[AbsoluteVirtualAddress, capa.features.addres
         """
         raise NotImplementedError()
 
+    def get_sample_hashes(self) -> Tuple[str, str, str]:
+        """
+        fetch the hashes for the sample contained within the extractor.
+
+        the order of the hashes is: md5, sha1, sha256
+        """
+        raise NotImplementedError()
+
     @abc.abstractmethod
     def extract_global_features(self) -> Iterator[Tuple[Feature, Address]]:
         """
@@ -309,6 +341,14 @@ class DynamicFeatureExtractor:
     This class is not instantiated directly; it is the base class for other implementations.
     """
 
+    def get_sample_hashes(self) -> Tuple[str, str, str]:
+        """
+        fetch the hashes for the sample contained within the extractor.
+
+        the order of the hashes is: md5, sha1, sha256
+        """
+        raise NotImplementedError()
+
     @abc.abstractmethod
     def extract_global_features(self) -> Iterator[Tuple[Feature, Address]]:
         """

capa/features/extractors/binja/extractor.py

@@ -17,7 +17,13 @@
 import capa.features.extractors.binja.basicblock
 from capa.features.common import Feature
 from capa.features.address import Address, AbsoluteVirtualAddress
-from capa.features.extractors.base_extractor import BBHandle, InsnHandle, FunctionHandle, StaticFeatureExtractor
+from capa.features.extractors.base_extractor import (
+    BBHandle,
+    InsnHandle,
+    SampleHashes,
+    FunctionHandle,
+    StaticFeatureExtractor,
+)
 
 
 class BinjaFeatureExtractor(StaticFeatureExtractor):
@@ -28,10 +34,15 @@ def __init__(self, bv: binja.BinaryView):
         self.global_features.extend(capa.features.extractors.binja.file.extract_file_format(self.bv))
         self.global_features.extend(capa.features.extractors.binja.global_.extract_os(self.bv))
         self.global_features.extend(capa.features.extractors.binja.global_.extract_arch(self.bv))
+        with open(self.bv, "rb") as f:
+            self.sample_hashes = SampleHashes.from_sample(f.read())
 
     def get_base_address(self):
         return AbsoluteVirtualAddress(self.bv.start)
 
+    def get_sample_hashes(self):
+        return tuple(self.sample_hashes)
+
     def extract_global_features(self):
         yield from self.global_features
 

capa/features/extractors/cape/extractor.py

@@ -14,7 +14,7 @@
 import capa.features.extractors.cape.process
 from capa.features.common import Feature
 from capa.features.address import Address, AbsoluteVirtualAddress, _NoAddress
-from capa.features.extractors.base_extractor import ThreadHandle, ProcessHandle, DynamicFeatureExtractor
+from capa.features.extractors.base_extractor import SampleHashes, ThreadHandle, ProcessHandle, DynamicFeatureExtractor
 
 logger = logging.getLogger(__name__)
 
@@ -27,13 +27,21 @@ def __init__(self, cape_version: str, static: Dict, behavior: Dict):
         self.cape_version = cape_version
         self.static = static
         self.behavior = behavior
+        self.hashes = SampleHashes(
+            md5=static["file"]["md5"],
+            sha1=static["file"]["sha1"],
+            sha256=static["file"]["sha256"],
+        )
 
         self.global_features = capa.features.extractors.cape.global_.extract_features(self.static)
 
     def get_base_address(self) -> Union[AbsoluteVirtualAddress, _NoAddress, None]:
         # value according to the PE header, the actual trace may use a different imagebase
         return AbsoluteVirtualAddress(self.static["pe"]["imagebase"])
 
+    def get_sample_hashes(self):
+        return tuple(self.hashes)
+
     def extract_global_features(self) -> Iterator[Tuple[Feature, Address]]:
         yield from self.global_features
 

capa/features/extractors/dnfile/extractor.py

@@ -21,7 +21,13 @@
 from capa.features.common import Feature
 from capa.features.address import NO_ADDRESS, Address, DNTokenAddress, DNTokenOffsetAddress
 from capa.features.extractors.dnfile.types import DnType, DnUnmanagedMethod
-from capa.features.extractors.base_extractor import BBHandle, InsnHandle, FunctionHandle, StaticFeatureExtractor
+from capa.features.extractors.base_extractor import (
+    BBHandle,
+    InsnHandle,
+    SampleHashes,
+    FunctionHandle,
+    StaticFeatureExtractor,
+)
 from capa.features.extractors.dnfile.helpers import (
     get_dotnet_types,
     get_dotnet_fields,
@@ -71,6 +77,8 @@ class DnfileFeatureExtractor(StaticFeatureExtractor):
     def __init__(self, path: str):
         super().__init__()
         self.pe: dnfile.dnPE = dnfile.dnPE(path)
+        with open(path, "rb") as f:
+            self.sample_hashes = SampleHashes.from_sample(f.read())
 
         # pre-compute .NET token lookup tables; each .NET method has access to this cache for feature extraction
         # most relevant at instruction scope
@@ -85,6 +93,9 @@ def __init__(self, path: str):
     def get_base_address(self):
         return NO_ADDRESS
 
+    def get_sample_hashes(self):
+        return tuple(self.sample_hashes)
+
     def extract_global_features(self):
         yield from self.global_features
 

capa/features/extractors/ida/extractor.py

@@ -18,7 +18,13 @@
 import capa.features.extractors.ida.basicblock
 from capa.features.common import Feature
 from capa.features.address import Address, AbsoluteVirtualAddress
-from capa.features.extractors.base_extractor import BBHandle, InsnHandle, FunctionHandle, StaticFeatureExtractor
+from capa.features.extractors.base_extractor import (
+    BBHandle,
+    InsnHandle,
+    SampleHashes,
+    FunctionHandle,
+    StaticFeatureExtractor,
+)
 
 
 class IdaFeatureExtractor(StaticFeatureExtractor):
@@ -28,10 +34,15 @@ def __init__(self):
         self.global_features.extend(capa.features.extractors.ida.file.extract_file_format())
         self.global_features.extend(capa.features.extractors.ida.global_.extract_os())
         self.global_features.extend(capa.features.extractors.ida.global_.extract_arch())
+        with open(idaapi.get_input_file_path, "rb") as f:
+            self.sample_hashes = SampleHashes(f.read())
 
     def get_base_address(self):
         return AbsoluteVirtualAddress(idaapi.get_imagebase())
 
+    def get_sample_hashes(self):
+        return self.sample_hashes
+
     def extract_global_features(self):
         yield from self.global_features
 

capa/features/extractors/viv/extractor.py

@@ -19,7 +19,13 @@
 import capa.features.extractors.viv.basicblock
 from capa.features.common import Feature
 from capa.features.address import Address, AbsoluteVirtualAddress
-from capa.features.extractors.base_extractor import BBHandle, InsnHandle, FunctionHandle, StaticFeatureExtractor
+from capa.features.extractors.base_extractor import (
+    BBHandle,
+    InsnHandle,
+    SampleHashes,
+    FunctionHandle,
+    StaticFeatureExtractor,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -29,8 +35,9 @@ def __init__(self, vw, path, os):
         super().__init__()
         self.vw = vw
         self.path = path
-        with open(self.path, "rb") as f:
+        with open(path, "rb") as f:
             self.buf = f.read()
+            self.sample_hashes = SampleHashes.from_sample(self.buf)
 
         # pre-compute these because we'll yield them at *every* scope.
         self.global_features: List[Tuple[Feature, Address]] = []
@@ -42,6 +49,9 @@ def get_base_address(self):
         # assume there is only one file loaded into the vw
         return AbsoluteVirtualAddress(list(self.vw.filemeta.values())[0]["imagebase"])
 
+    def get_sample_hashes(self):
+        return tuple(self.sample_hashes)
+
     def extract_global_features(self):
         yield from self.global_features
 

capa/ida/plugin/model.py

@@ -500,16 +500,16 @@ def render_capa_doc_by_program(self, doc: rd.ResultDocument):
                 location = location_.to_capa()
 
                 parent2: CapaExplorerDataItem
-                if rule.meta.scope == capa.rules.FILE_SCOPE:
+                if capa.rules.FILE_SCOPE in rule.meta.scopes:
                     parent2 = parent
-                elif rule.meta.scope == capa.rules.FUNCTION_SCOPE:
+                elif capa.rules.FUNCTION_SCOPE in rule.meta.scopes:
                     parent2 = CapaExplorerFunctionItem(parent, location)
-                elif rule.meta.scope == capa.rules.BASIC_BLOCK_SCOPE:
+                elif capa.rules.BASIC_BLOCK_SCOPE in rule.meta.scopes:
                     parent2 = CapaExplorerBlockItem(parent, location)
-                elif rule.meta.scope == capa.rules.INSTRUCTION_SCOPE:
+                elif capa.rules.INSTRUCTION_SCOPE in rule.meta.scopes:
                     parent2 = CapaExplorerInstructionItem(parent, location)
                 else:
-                    raise RuntimeError("unexpected rule scope: " + str(rule.meta.scope))
+                    raise RuntimeError("unexpected rule scope: " + str(rule.meta.scopes.static))
 
                 self.render_capa_doc_match(parent2, match, doc)