Merge pull request #1 from mandiant/binja-ci

Author: xusheng6

.github/workflows/tests.yml

@@ -90,3 +90,37 @@ jobs:
       run: pip install -e .[dev]
     - name: Run tests
       run: pytest -v tests/
+
+  binja-tests:
+    name: Binary Ninja tests for ${{ matrix.python-version }} on ${{ matrix.os }}
+    runs-on: ubuntu-20.04
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.7", "3.11"]
+    steps:
+    - name: Checkout capa with submodules
+      uses: actions/checkout@ac593985615ec2ede58e132d2e21d2b1cbd6127c # v3.3.0
+      with:
+        submodules: recursive
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@d27e3f3d7c64b4bbf8e4abfb9b63b83e846e0435 # v4.5.0
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install pyyaml
+      run: sudo apt-get install -y libyaml-dev
+    - name: Install capa
+      run: pip install -e .[dev]
+    - name: install Binary Ninja
+      env:
+        BN_SERIAL: ${{ secrets.BN_SERIAL }}
+      run: |
+        mkdir ./.github/binja
+        curl "https://raw.githubusercontent.com/Vector35/binaryninja-api/6812c97/scripts/download_headless.py" -o ./.github/binja/download_headless.py
+        python ./.github/binja/download_headless.py --serial $BN_SERIAL --output .github/binja/BinaryNinja-headless.zip
+        unzip .github/binja/BinaryNinja-headless.zip -d .github/binja/
+        python .github/binja/binaryninja/scripts/install_api.py --install-on-root --silent
+    - name: Run tests
+      env:
+        BN_LICENSE: ${{ secrets.BN_LICENSE }}
+      run: pytest -v tests/test_binja_features.py  # explicitly refer to the binja tests for performance. other tests run above.

.gitignore

@@ -125,3 +125,4 @@ scripts/perf/*.zip
 Pipfile
 Pipfile.lock
 /cache/
+.github/binja/binaryninja

CHANGELOG.md

@@ -8,7 +8,7 @@
 
 ### Breaking Changes
 
-### New Rules (12)
+### New Rules (20)
 
 - persistence/scheduled-tasks/schedule-task-via-at joren485
 - data-manipulation/prng/generate-random-numbers-via-rtlgenrandom william.ballenthin@mandiant.com
@@ -22,6 +22,14 @@
 - nursery/get-http-request-uri william.ballenthin@mandiant.com
 - nursery/create-zip-archive-in-dotnet michael.hunhoff@mandiant.com
 - nursery/extract-zip-archive-in-dotnet anushka.virgaonkar@mandiant.com michael.hunhoff@mandiant.com
+- data-manipulation/encryption/tea/decrypt-data-using-tea william.ballenthin@mandiant.com raymond.leong@mandiant.com
+- data-manipulation/encryption/tea/encrypt-data-using-tea william.ballenthin@mandiant.com raymond.leong@mandiant.com
+- data-manipulation/encryption/xtea/encrypt-data-using-xtea raymond.leong@mandiant.com
+- data-manipulation/encryption/xxtea/encrypt-data-using-xxtea raymond.leong@mandiant.com
+- nursery/hash-data-using-ripemd128 raymond.leong@mandiant.com
+- nursery/hash-data-using-ripemd256 raymond.leong@mandiant.com
+- nursery/hash-data-using-ripemd320 raymond.leong@mandiant.com
+- nursery/set-web-proxy-in-dotnet michael.hunhoff@mandiant.com
 -
 
 ### Bug Fixes
@@ -30,6 +38,8 @@
 - extractor: fix IDA and vivisect string and bytes features overlap and tests #1327 #1336 @xusheng6 
 
 ### capa explorer IDA Pro plugin
+- fix exception when plugin loaded in IDA hosted under idat #1341 @mike-hunhoff
+- improve embedded PE detection performance and reduce FP potential #1344 @mike-hunhoff
 
 ### Development
 

README.md

@@ -2,7 +2,7 @@
 
 [![PyPI - Python Version](https://img.shields.io/pypi/pyversions/flare-capa)](https://pypi.org/project/flare-capa)
 [![Last release](https://img.shields.io/github/v/release/mandiant/capa)](https://github.com/mandiant/capa/releases)
-[![Number of rules](https://img.shields.io/badge/rules-781-blue.svg)](https://github.com/mandiant/capa-rules)
+[![Number of rules](https://img.shields.io/badge/rules-787-blue.svg)](https://github.com/mandiant/capa-rules)
 [![CI status](https://github.com/mandiant/capa/workflows/CI/badge.svg)](https://github.com/mandiant/capa/actions?query=workflow%3ACI+event%3Apush+branch%3Amaster)
 [![Downloads](https://img.shields.io/github/downloads/mandiant/capa/total)](https://github.com/mandiant/capa/releases)
 [![License](https://img.shields.io/badge/license-Apache--2.0-green.svg)](LICENSE.txt)

capa/engine.py

@@ -43,10 +43,12 @@ def __init__(self, description=None):
         self.description = description
 
     def __str__(self):
+        name = self.name.lower()
+        children = ",".join(map(str, self.get_children()))
         if self.description:
-            return "%s(%s = %s)" % (self.name.lower(), ",".join(map(str, self.get_children())), self.description)
+            return f"{name}({children} = {self.description})"
         else:
-            return "%s(%s)" % (self.name.lower(), ",".join(map(str, self.get_children())))
+            return f"{name}({children})"
 
     def __repr__(self):
         return str(self)
@@ -232,9 +234,9 @@ def evaluate(self, ctx, **kwargs):
 
     def __str__(self):
         if self.max == (1 << 64 - 1):
-            return "range(%s, min=%d, max=infinity)" % (str(self.child), self.min)
+            return f"range({str(self.child)}, min={self.min}, max=infinity)"
         else:
-            return "range(%s, min=%d, max=%d)" % (str(self.child), self.min, self.max)
+            return f"range({str(self.child)}, min={self.min}, max={self.max})"
 
 
 class Subscope(Statement):

capa/features/common.py

@@ -149,11 +149,11 @@ def get_value_str(self) -> str:
     def __str__(self):
         if self.value is not None:
             if self.description:
-                return "%s(%s = %s)" % (self.get_name_str(), self.get_value_str(), self.description)
+                return f"{self.get_name_str()}({self.get_value_str()} = {self.description})"
             else:
-                return "%s(%s)" % (self.get_name_str(), self.get_value_str())
+                return f"{self.get_name_str()}({self.get_value_str()})"
         else:
-            return "%s" % self.get_name_str()
+            return f"{self.get_name_str()}"
 
     def __repr__(self):
         return str(self)
@@ -242,7 +242,7 @@ def get_value_str(self) -> str:
 
     def __str__(self):
         assert isinstance(self.value, str)
-        return "substring(%s)" % escape_string(self.value)
+        return f"substring({escape_string(self.value)})"
 
 
 class _MatchedSubstring(Substring):
@@ -267,11 +267,9 @@ def __init__(self, substring: Substring, matches: Dict[str, Set[Address]]):
         self.matches = matches
 
     def __str__(self):
+        matches = ", ".join(map(lambda s: '"' + s + '"', (self.matches or {}).keys()))
         assert isinstance(self.value, str)
-        return 'substring("%s", matches = %s)' % (
-            self.value,
-            ", ".join(map(lambda s: '"' + s + '"', (self.matches or {}).keys())),
-        )
+        return f'substring("{self.value}", matches = {matches})'
 
 
 class Regex(String):
@@ -290,7 +288,7 @@ def __init__(self, value: str, description=None):
             if value.endswith("/i"):
                 value = value[: -len("i")]
             raise ValueError(
-                "invalid regular expression: %s it should use Python syntax, try it at https://pythex.org" % value
+                f"invalid regular expression: {value} it should use Python syntax, try it at https://pythex.org"
             ) from exc
 
     def evaluate(self, ctx, short_circuit=True):
@@ -336,7 +334,7 @@ def evaluate(self, ctx, short_circuit=True):
 
     def __str__(self):
         assert isinstance(self.value, str)
-        return "regex(string =~ %s)" % self.value
+        return f"regex(string =~ {self.value})"
 
 
 class _MatchedRegex(Regex):
@@ -361,11 +359,9 @@ def __init__(self, regex: Regex, matches: Dict[str, Set[Address]]):
         self.matches = matches
 
     def __str__(self):
+        matches = ", ".join(map(lambda s: '"' + s + '"', (self.matches or {}).keys()))
         assert isinstance(self.value, str)
-        return "regex(string =~ %s, matches = %s)" % (
-            self.value,
-            ", ".join(map(lambda s: '"' + s + '"', (self.matches or {}).keys())),
-        )
+        return f"regex(string =~ {self.value}, matches = {matches})"
 
 
 class StringFactory:

capa/features/extractors/elf.py

@@ -121,14 +121,14 @@ def _parse(self):
         elif ei_class == 2:
             self.bitness = 64
         else:
-            raise CorruptElfFile("invalid ei_class: 0x%02x" % ei_class)
+            raise CorruptElfFile(f"invalid ei_class: 0x{ei_class:02x}")
 
         if ei_data == 1:
             self.endian = "<"
         elif ei_data == 2:
             self.endian = ">"
         else:
-            raise CorruptElfFile("not an ELF file: invalid ei_data: 0x%02x" % ei_data)
+            raise CorruptElfFile(f"not an ELF file: invalid ei_data: 0x{ei_data:02x}")
 
         if self.bitness == 32:
             e_phoff, e_shoff = struct.unpack_from(self.endian + "II", self.file_header, 0x1C)

capa/features/extractors/helpers.py

@@ -55,15 +55,15 @@ def generate_symbols(dll: str, symbol: str) -> Iterator[str]:
     dll = dll.lower()
 
     # kernel32.CreateFileA
-    yield "%s.%s" % (dll, symbol)
+    yield f"{dll}.{symbol}"
 
     if not is_ordinal(symbol):
         # CreateFileA
         yield symbol
 
     if is_aw_function(symbol):
         # kernel32.CreateFile
-        yield "%s.%s" % (dll, symbol[:-1])
+        yield f"{dll}.{symbol[:-1]}"
 
         if not is_ordinal(symbol):
             # CreateFile

capa/features/extractors/ida/basicblock.py

@@ -34,7 +34,7 @@ def get_printable_len(op: idaapi.op_t) -> int:
     elif op.dtype == idaapi.dt_qword:
         chars = struct.pack("<Q", op_val)
     else:
-        raise ValueError("Unhandled operand data type 0x%x." % op.dtype)
+        raise ValueError(f"Unhandled operand data type 0x{op.dtype:x}.")
 
     def is_printable_ascii(chars_: bytes):
         return all(c < 127 and chr(c) in string.printable for c in chars_)

capa/features/extractors/ida/file.py

@@ -21,12 +21,14 @@
 from capa.features.common import FORMAT_PE, FORMAT_ELF, Format, String, Feature, Characteristic
 from capa.features.address import NO_ADDRESS, Address, FileOffsetAddress, AbsoluteVirtualAddress
 
+MAX_OFFSET_PE_AFTER_MZ = 0x200
+
 
 def check_segment_for_pe(seg: idaapi.segment_t) -> Iterator[Tuple[int, int]]:
     """check segment for embedded PE
 
     adapted for IDA from:
-    https://github.com/vivisect/vivisect/blob/7be4037b1cecc4551b397f840405a1fc606f9b53/PE/carve.py#L19
+    https://github.com/vivisect/vivisect/blob/91e8419a861f49779f18316f155311967e696836/PE/carve.py#L25
     """
     seg_max = seg.end_ea
     mz_xor = [
@@ -40,30 +42,32 @@ def check_segment_for_pe(seg: idaapi.segment_t) -> Iterator[Tuple[int, int]]:
 
     todo = []
     for mzx, pex, i in mz_xor:
+        # find all segment offsets containing XOR'd "MZ" bytes
         for off in capa.features.extractors.ida.helpers.find_byte_sequence(seg.start_ea, seg.end_ea, mzx):
             todo.append((off, mzx, pex, i))
 
     while len(todo):
         off, mzx, pex, i = todo.pop()
 
-        # The MZ header has one field we will check e_lfanew is at 0x3c
+        # MZ header has one field we will check e_lfanew is at 0x3c
         e_lfanew = off + 0x3C
 
         if seg_max < (e_lfanew + 4):
             continue
 
         newoff = struct.unpack("<I", capa.features.extractors.helpers.xor_static(idc.get_bytes(e_lfanew, 4), i))[0]
 
+        # assume XOR'd "PE" bytes exist within threshold
+        if newoff > MAX_OFFSET_PE_AFTER_MZ:
+            continue
+
         peoff = off + newoff
         if seg_max < (peoff + 2):
             continue
 
         if idc.get_bytes(peoff, 2) == pex:
             yield off, i
 
-        for nextres in capa.features.extractors.ida.helpers.find_byte_sequence(off + 1, seg.end_ea, mzx):
-            todo.append((nextres, mzx, pex, i))
-
 
 def extract_file_embedded_pe() -> Iterator[Tuple[Feature, Address]]:
     """extract embedded PE features
@@ -102,13 +106,13 @@ def extract_file_import_names() -> Iterator[Tuple[Feature, Address]]:
             for name in capa.features.extractors.helpers.generate_symbols(info[0], info[1]):
                 yield Import(name), addr
             dll = info[0]
-            symbol = "#%d" % (info[2])
+            symbol = f"#{info[2]}"
         elif info[1]:
             dll = info[0]
             symbol = info[1]
         elif info[2]:
             dll = info[0]
-            symbol = "#%d" % (info[2])
+            symbol = f"#{info[2]}"
         else:
             continue
 
@@ -176,7 +180,7 @@ def extract_file_format() -> Iterator[Tuple[Feature, Address]]:
         # no file type to return when processing a binary file, but we want to continue processing
         return
     else:
-        raise NotImplementedError("unexpected file format: %d" % file_info.filetype)
+        raise NotImplementedError(f"unexpected file format: {file_info.filetype}")
 
 
 def extract_features() -> Iterator[Tuple[Feature, Address]]:

capa/features/extractors/ida/helpers.py

@@ -25,7 +25,7 @@ def find_byte_sequence(start: int, end: int, seq: bytes) -> Iterator[int]:
         end: max virtual address
         seq: bytes to search e.g. b"\x01\x03"
     """
-    seqstr = " ".join(["%02x" % b for b in seq])
+    seqstr = " ".join([f"{b:02x}" for b in seq])
     while True:
         # TODO find_binary: Deprecated. Please use ida_bytes.bin_search() instead.
         ea = idaapi.find_binary(start, end, seqstr, 0, idaapi.SEARCH_DOWN)

capa/features/extractors/pefile.py

@@ -64,7 +64,7 @@ def extract_file_import_names(pe, **kwargs):
 
             for imp in dll.imports:
                 if imp.import_by_ordinal:
-                    impname = "#%s" % imp.ordinal
+                    impname = f"#{imp.ordinal}"
                 else:
                     try:
                         impname = imp.name.partition(b"\x00")[0].decode("ascii")

capa/features/extractors/viv/basicblock.py

@@ -121,7 +121,7 @@ def get_printable_len(oper: envi.archs.i386.disasm.i386ImmOper) -> int:
     elif oper.tsize == 8:
         chars = struct.pack("<Q", oper.imm)
     else:
-        raise ValueError("unexpected oper.tsize: %d" % (oper.tsize))
+        raise ValueError(f"unexpected oper.tsize: {oper.tsize}")
 
     if is_printable_ascii(chars):
         return oper.tsize

capa/features/extractors/viv/file.py

@@ -44,7 +44,7 @@ def extract_file_import_names(vw, **kwargs) -> Iterator[Tuple[Feature, Address]]
         modname, impname = tinfo.split(".", 1)
         if is_viv_ord_impname(impname):
             # replace ord prefix with #
-            impname = "#%s" % impname[len("ord") :]
+            impname = "#" + impname[len("ord") :]
 
         addr = AbsoluteVirtualAddress(va)
         for name in capa.features.extractors.helpers.generate_symbols(modname, impname):

capa/features/freeze/__init__.py

@@ -329,7 +329,7 @@ def loads(s: str) -> capa.features.extractors.base_extractor.FeatureExtractor:
 
     freeze = Freeze.parse_raw(s)
     if freeze.version != 2:
-        raise ValueError("unsupported freeze format version: %d", freeze.version)
+        raise ValueError(f"unsupported freeze format version: {freeze.version}")
 
     return null.NullFeatureExtractor(
         base_address=freeze.base_address.to_capa(),

capa/features/insn.py

@@ -15,9 +15,9 @@
 def hex(n: int) -> str:
     """render the given number using upper case hex, like: 0x123ABC"""
     if n < 0:
-        return "-0x%X" % (-n)
+        return f"-0x{(-n):X}"
     else:
-        return "0x%X" % n
+        return f"0x{(n):X}"
 
 
 class API(Feature):
@@ -31,7 +31,7 @@ def __init__(self, value: str, access: Optional[str] = None, description: Option
         super().__init__(value, description=description)
         if access is not None:
             if access not in VALID_FEATURE_ACCESS:
-                raise ValueError("%s access type %s not valid" % (self.name, access))
+                raise ValueError(f"{self.name} access type {access} not valid")
         self.access = access
 
     def __hash__(self):
@@ -105,7 +105,7 @@ def __eq__(self, other):
 
 class OperandNumber(_Operand):
     # cached names so we don't do extra string formatting every ctor
-    NAMES = ["operand[%d].number" % i for i in range(MAX_OPERAND_COUNT)]
+    NAMES = [f"operand[{i}].number" for i in range(MAX_OPERAND_COUNT)]
 
     # operand[i].number: 0x12
     def __init__(self, index: int, value: int, description=None):
@@ -119,7 +119,7 @@ def get_value_str(self) -> str:
 
 class OperandOffset(_Operand):
     # cached names so we don't do extra string formatting every ctor
-    NAMES = ["operand[%d].offset" % i for i in range(MAX_OPERAND_COUNT)]
+    NAMES = [f"operand[{i}].offset" for i in range(MAX_OPERAND_COUNT)]
 
     # operand[i].offset: 0x12
     def __init__(self, index: int, value: int, description=None):