extract dynamic capabilities

.github/mypy/mypy.ini

@@ -1,5 +1,12 @@
 [mypy]
 
+# TODO(yelhamer): remove this once proto has been added
+# for the dynamic rendering
+exclude = (?x)(
+    ^capa/render/proto/__init__.py$
+    | ^tests/_test_proto.py$
+  )
+
 [mypy-halo.*]
 ignore_missing_imports = True
 

CHANGELOG.md

@@ -10,6 +10,7 @@
 - Add a new thread scope for the dynamic analysis flavor #1517 @yelhamer
 - Add support for flavor-based rule scopes @yelhamer
 - Add ProcessesAddress and ThreadAddress #1612 @yelhamer
+- Add dynamic capability extraction @yelhamer
 
 ### Breaking Changes
 

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_bytes(cls, buf: bytes) -> "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,12 @@ def get_base_address(self) -> Union[AbsoluteVirtualAddress, capa.features.addres
         """
         raise NotImplementedError()
 
+    def get_sample_hashes(self) -> SampleHashes:
+        """
+        fetch the hashes for the sample contained within the extractor.
+        """
+        raise NotImplementedError()
+
     @abc.abstractmethod
     def extract_global_features(self) -> Iterator[Tuple[Feature, Address]]:
         """
@@ -309,6 +339,22 @@ class DynamicFeatureExtractor:
     This class is not instantiated directly; it is the base class for other implementations.
     """
 
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self):
+        #
+        # note: a subclass should define ctor parameters for its own use.
+        #  for example, the Vivisect feature extract might require the vw and/or path.
+        # this base class doesn't know what to do with that info, though.
+        #
+        super().__init__()
+
+    def get_sample_hashes(self) -> SampleHashes:
+        """
+        fetch the hashes for the sample contained within the extractor.
+        """
+        raise NotImplementedError()
+
     @abc.abstractmethod
     def extract_global_features(self) -> Iterator[Tuple[Feature, Address]]:
         """

capa/features/extractors/binja/extractor.py

@@ -6,6 +6,7 @@
 #  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.
 from typing import List, Tuple, Iterator
+from pathlib import Path
 
 import binaryninja as binja
 
@@ -17,7 +18,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 +35,14 @@ 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))
+        self.sample_hashes = SampleHashes.from_bytes(Path(bv.file.original_filename).read_bytes())
 
     def get_base_address(self):
         return AbsoluteVirtualAddress(self.bv.start)
 
+    def get_sample_hashes(self) -> SampleHashes:
+        return self.sample_hashes
+
     def extract_global_features(self):
         yield from self.global_features
 

capa/features/extractors/cape/extractor.py

@@ -15,7 +15,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__)
 
@@ -28,13 +28,21 @@ def __init__(self, cape_version: str, static: Dict, behavior: Dict):
         self.cape_version = cape_version
         self.static = static
         self.behavior = behavior
+        self.sample_hashes = SampleHashes(
+            md5=static["file"]["md5"].lower(),
+            sha1=static["file"]["sha1"].lower(),
+            sha256=static["file"]["sha256"].lower(),
+        )
 
         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) -> SampleHashes:
+        return self.sample_hashes
+
     def extract_global_features(self) -> Iterator[Tuple[Feature, Address]]:
         yield from self.global_features
 

capa/features/extractors/dnfile/extractor.py

@@ -22,7 +22,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,
@@ -72,6 +78,7 @@ class DnfileFeatureExtractor(StaticFeatureExtractor):
     def __init__(self, path: Path):
         super().__init__()
         self.pe: dnfile.dnPE = dnfile.dnPE(str(path))
+        self.sample_hashes = SampleHashes.from_bytes(path.read_bytes())
 
         # pre-compute .NET token lookup tables; each .NET method has access to this cache for feature extraction
         # most relevant at instruction scope
@@ -86,6 +93,9 @@ def __init__(self, path: Path):
     def get_base_address(self):
         return NO_ADDRESS
 
+    def get_sample_hashes(self) -> SampleHashes:
+        return self.sample_hashes
+
     def extract_global_features(self):
         yield from self.global_features
 

capa/features/extractors/dnfile_.py

@@ -25,7 +25,7 @@
     Feature,
 )
 from capa.features.address import NO_ADDRESS, Address, AbsoluteVirtualAddress
-from capa.features.extractors.base_extractor import StaticFeatureExtractor
+from capa.features.extractors.base_extractor import SampleHashes, StaticFeatureExtractor
 
 logger = logging.getLogger(__name__)
 
@@ -86,10 +86,14 @@ def __init__(self, path: Path):
         super().__init__()
         self.path: Path = path
         self.pe: dnfile.dnPE = dnfile.dnPE(str(path))
+        self.sample_hashes = SampleHashes.from_bytes(self.path.read_bytes())
 
     def get_base_address(self) -> AbsoluteVirtualAddress:
         return AbsoluteVirtualAddress(0x0)
 
+    def get_sample_hashes(self) -> SampleHashes:
+        return self.sample_hashes
+
     def get_entry_point(self) -> int:
         # self.pe.net.Flags.CLT_NATIVE_ENTRYPOINT
         #  True: native EP: Token

capa/features/extractors/dotnetfile.py

@@ -31,7 +31,7 @@
     Characteristic,
 )
 from capa.features.address import NO_ADDRESS, Address, DNTokenAddress
-from capa.features.extractors.base_extractor import StaticFeatureExtractor
+from capa.features.extractors.base_extractor import SampleHashes, StaticFeatureExtractor
 from capa.features.extractors.dnfile.helpers import (
     DnType,
     iter_dotnet_table,
@@ -170,10 +170,14 @@ def __init__(self, path: Path):
         super().__init__()
         self.path: Path = path
         self.pe: dnfile.dnPE = dnfile.dnPE(str(path))
+        self.sample_hashes = SampleHashes.from_bytes(self.path.read_bytes())
 
     def get_base_address(self):
         return NO_ADDRESS
 
+    def get_sample_hashes(self) -> SampleHashes:
+        return self.sample_hashes
+
     def get_entry_point(self) -> int:
         # self.pe.net.Flags.CLT_NATIVE_ENTRYPOINT
         #  True: native EP: Token

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,16 @@ 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())
+        self.sample_hashes = SampleHashes(
+            md5=idaapi.get_input_file_md5(), sha1=idaapi.get_input_file_sha1(), sha256=idaapi.get_input_file_sha256()
+        )
 
     def get_base_address(self):
         return AbsoluteVirtualAddress(idaapi.get_imagebase())
 
+    def get_sample_hashes(self) -> SampleHashes:
+        return self.sample_hashes
+
     def extract_global_features(self):
         yield from self.global_features
 

capa/features/extractors/null.py

@@ -15,6 +15,7 @@
 from capa.features.extractors.base_extractor import (
     BBHandle,
     InsnHandle,
+    SampleHashes,
     ThreadHandle,
     ProcessHandle,
     FunctionHandle,
@@ -49,6 +50,7 @@ class NullStaticFeatureExtractor(StaticFeatureExtractor):
     """
 
     base_address: Address
+    sample_hashes: SampleHashes
     global_features: List[Feature]
     file_features: List[Tuple[Address, Feature]]
     functions: Dict[Address, FunctionFeatures]
@@ -60,6 +62,9 @@ def extract_global_features(self):
         for feature in self.global_features:
             yield feature, NO_ADDRESS
 
+    def get_sample_hashes(self) -> SampleHashes:
+        return self.sample_hashes
+
     def extract_file_features(self):
         for address, feature in self.file_features:
             yield feature, address
@@ -103,6 +108,7 @@ class ProcessFeatures:
 @dataclass
 class NullDynamicFeatureExtractor(DynamicFeatureExtractor):
     base_address: Address
+    sample_hashes: SampleHashes
     global_features: List[Feature]
     file_features: List[Tuple[Address, Feature]]
     processes: Dict[Address, ProcessFeatures]
@@ -111,6 +117,9 @@ def extract_global_features(self):
         for feature in self.global_features:
             yield feature, NO_ADDRESS
 
+    def get_sample_hashes(self) -> SampleHashes:
+        return self.sample_hashes
+
     def extract_file_features(self):
         for address, feature in self.file_features:
             yield feature, address

capa/features/extractors/pefile.py

@@ -19,7 +19,7 @@
 from capa.features.file import Export, Import, Section
 from capa.features.common import OS, ARCH_I386, FORMAT_PE, ARCH_AMD64, OS_WINDOWS, Arch, Format, Characteristic
 from capa.features.address import NO_ADDRESS, FileOffsetAddress, AbsoluteVirtualAddress
-from capa.features.extractors.base_extractor import StaticFeatureExtractor
+from capa.features.extractors.base_extractor import SampleHashes, StaticFeatureExtractor
 
 logger = logging.getLogger(__name__)
 
@@ -190,10 +190,14 @@ def __init__(self, path: Path):
         super().__init__()
         self.path: Path = path
         self.pe = pefile.PE(str(path))
+        self.sample_hashes = SampleHashes.from_bytes(self.path.read_bytes())
 
     def get_base_address(self):
         return AbsoluteVirtualAddress(self.pe.OPTIONAL_HEADER.ImageBase)
 
+    def get_sample_hashes(self) -> SampleHashes:
+        return self.sample_hashes
+
     def extract_global_features(self):
         buf = Path(self.path).read_bytes()
 

capa/features/extractors/viv/extractor.py

@@ -20,7 +20,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__)
 
@@ -31,6 +37,7 @@ def __init__(self, vw, path: Path, os):
         self.vw = vw
         self.path = path
         self.buf = path.read_bytes()
+        self.sample_hashes = SampleHashes.from_bytes(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) -> SampleHashes:
+        return self.sample_hashes
+
     def extract_global_features(self):
         yield from self.global_features