Merge branch 'master' into dependabot/pip/pydantic-2.9.2

.github/mypy/mypy.ini

@@ -1,8 +1,5 @@
 [mypy]
 
-[mypy-tqdm.*]
-ignore_missing_imports = True
-
 [mypy-ruamel.*]
 ignore_missing_imports = True
 

.github/pyinstaller/pyinstaller.spec

@@ -2,7 +2,6 @@
 # Copyright (C) 2020 Mandiant, Inc. All Rights Reserved.
 import sys
 
-import wcwidth
 import capa.rules.cache
 
 from pathlib import Path
@@ -29,13 +28,6 @@ a = Analysis(
         ("../../rules", "rules"),
         ("../../sigs", "sigs"),
         ("../../cache", "cache"),
-        # capa.render.default uses tabulate that depends on wcwidth.
-        # it seems wcwidth uses a json file `version.json`
-        # and this doesn't get picked up by pyinstaller automatically.
-        # so we manually embed the wcwidth resources here.
-        #
-        # ref: https://stackoverflow.com/a/62278462/87207
-        (Path(wcwidth.__file__).parent, "wcwidth"),
     ],
     # when invoking pyinstaller from the project root,
     # this gets run from the project root.
@@ -48,11 +40,6 @@ a = Analysis(
         "tkinter",
         "_tkinter",
         "Tkinter",
-        # tqdm provides renderers for ipython,
-        # however, this drags in a lot of dependencies.
-        # since we don't spawn a notebook, we can safely remove these.
-        "IPython",
-        "ipywidgets",
         # these are pulled in by networkx
         # but we don't need to compute the strongly connected components.
         "numpy",
@@ -70,7 +57,10 @@ a = Analysis(
         "qt5",
         "pyqtwebengine",
         "pyasn1",
+        # don't pull in Binary Ninja/IDA bindings that should
+        # only be installed locally.
         "binaryninja",
+        "ida",
     ],
 )
 

.github/workflows/build.yml

@@ -30,8 +30,8 @@ jobs:
             python_version: 3.8
           - os: ubuntu-20.04
             artifact_name: capa
-            asset_name: linux-py311
-            python_version: 3.11
+            asset_name: linux-py312
+            python_version: 3.12
           - os: windows-2019
             artifact_name: capa.exe
             asset_name: windows
@@ -88,7 +88,7 @@ jobs:
             asset_name: linux
           - os: ubuntu-22.04
             artifact_name: capa
-            asset_name: linux-py311
+            asset_name: linux-py312
           - os: windows-2022
             artifact_name: capa.exe
             asset_name: windows

.gitignore

@@ -127,3 +127,4 @@ Pipfile.lock
 .github/binja/download_headless.py
 .github/binja/BinaryNinja-headless.zip
 justfile
+data/

capa/capabilities/dynamic.py

@@ -6,20 +6,16 @@
 # 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
 import logging
 import itertools
 import collections
-from typing import Any, Tuple
-
-import tqdm
+from typing import Any, List, Tuple
 
 import capa.perf
 import capa.features.freeze as frz
 import capa.render.result_document as rdoc
 from capa.rules import Scope, RuleSet
 from capa.engine import FeatureSet, MatchResults
-from capa.helpers import redirecting_print_to_tqdm
 from capa.capabilities.common import find_file_capabilities
 from capa.features.extractors.base_extractor import CallHandle, ThreadHandle, ProcessHandle, DynamicFeatureExtractor
 
@@ -139,38 +135,30 @@ def find_dynamic_capabilities(
     feature_counts = rdoc.DynamicFeatureCounts(file=0, processes=())
 
     assert isinstance(extractor, DynamicFeatureExtractor)
-    with redirecting_print_to_tqdm(disable_progress):
-        with tqdm.contrib.logging.logging_redirect_tqdm():
-            pbar = tqdm.tqdm
-            if disable_progress:
-                # do not use tqdm to avoid unnecessary side effects when caller intends
-                # to disable progress completely
-                def pbar(s, *args, **kwargs):
-                    return s
-
-            elif not sys.stderr.isatty():
-                # don't display progress bar when stderr is redirected to a file
-                def pbar(s, *args, **kwargs):
-                    return s
-
-            processes = list(extractor.get_processes())
-
-            pb = pbar(processes, desc="matching", unit=" processes", leave=False)
-            for p in pb:
-                process_matches, thread_matches, call_matches, feature_count = find_process_capabilities(
-                    ruleset, extractor, p
-                )
-                feature_counts.processes += (
-                    rdoc.ProcessFeatureCount(address=frz.Address.from_capa(p.address), count=feature_count),
-                )
-                logger.debug("analyzed %s and extracted %d features", p.address, feature_count)
-
-                for rule_name, res in process_matches.items():
-                    all_process_matches[rule_name].extend(res)
-                for rule_name, res in thread_matches.items():
-                    all_thread_matches[rule_name].extend(res)
-                for rule_name, res in call_matches.items():
-                    all_call_matches[rule_name].extend(res)
+    processes: List[ProcessHandle] = list(extractor.get_processes())
+    n_processes: int = len(processes)
+
+    with capa.helpers.CapaProgressBar(
+        console=capa.helpers.log_console, transient=True, disable=disable_progress
+    ) as pbar:
+        task = pbar.add_task("matching", total=n_processes, unit="processes")
+        for p in processes:
+            process_matches, thread_matches, call_matches, feature_count = find_process_capabilities(
+                ruleset, extractor, p
+            )
+            feature_counts.processes += (
+                rdoc.ProcessFeatureCount(address=frz.Address.from_capa(p.address), count=feature_count),
+            )
+            logger.debug("analyzed %s and extracted %d features", p.address, feature_count)
+
+            for rule_name, res in process_matches.items():
+                all_process_matches[rule_name].extend(res)
+            for rule_name, res in thread_matches.items():
+                all_thread_matches[rule_name].extend(res)
+            for rule_name, res in call_matches.items():
+                all_call_matches[rule_name].extend(res)
+
+            pbar.advance(task)
 
     # collection of features that captures the rule matches within process and thread scopes.
     # mapping from feature (matched rule) to set of addresses at which it matched.

capa/capabilities/static.py

@@ -6,21 +6,18 @@
 # 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
 import time
 import logging
 import itertools
 import collections
-from typing import Any, Tuple
-
-import tqdm.contrib.logging
+from typing import Any, List, Tuple
 
 import capa.perf
+import capa.helpers
 import capa.features.freeze as frz
 import capa.render.result_document as rdoc
 from capa.rules import Scope, RuleSet
 from capa.engine import FeatureSet, MatchResults
-from capa.helpers import redirecting_print_to_tqdm
 from capa.capabilities.common import find_file_capabilities
 from capa.features.extractors.base_extractor import BBHandle, InsnHandle, FunctionHandle, StaticFeatureExtractor
 
@@ -143,75 +140,58 @@ def find_static_capabilities(
     library_functions: Tuple[rdoc.LibraryFunction, ...] = ()
 
     assert isinstance(extractor, StaticFeatureExtractor)
-    with redirecting_print_to_tqdm(disable_progress):
-        with tqdm.contrib.logging.logging_redirect_tqdm():
-            pbar = tqdm.tqdm
-            if capa.helpers.is_runtime_ghidra():
-                # Ghidrathon interpreter cannot properly handle
-                # the TMonitor thread that is created via a monitor_interval
-                # > 0
-                pbar.monitor_interval = 0
-            if disable_progress:
-                # do not use tqdm to avoid unnecessary side effects when caller intends
-                # to disable progress completely
-                def pbar(s, *args, **kwargs):
-                    return s
-
-            elif not sys.stderr.isatty():
-                # don't display progress bar when stderr is redirected to a file
-                def pbar(s, *args, **kwargs):
-                    return s
-
-            functions = list(extractor.get_functions())
-            n_funcs = len(functions)
-
-            pb = pbar(functions, desc="matching", unit=" functions", postfix="skipped 0 library functions", leave=False)
-            for f in pb:
-                t0 = time.time()
-                if extractor.is_library_function(f.address):
-                    function_name = extractor.get_function_name(f.address)
-                    logger.debug("skipping library function 0x%x (%s)", f.address, function_name)
-                    library_functions += (
-                        rdoc.LibraryFunction(address=frz.Address.from_capa(f.address), name=function_name),
-                    )
-                    n_libs = len(library_functions)
-                    percentage = round(100 * (n_libs / n_funcs))
-                    if isinstance(pb, tqdm.tqdm):
-                        pb.set_postfix_str(f"skipped {n_libs} library functions ({percentage}%)")
-                    continue
-
-                function_matches, bb_matches, insn_matches, feature_count = find_code_capabilities(
-                    ruleset, extractor, f
-                )
-                feature_counts.functions += (
-                    rdoc.FunctionFeatureCount(address=frz.Address.from_capa(f.address), count=feature_count),
-                )
-                t1 = time.time()
-
-                match_count = 0
-                for name, matches_ in itertools.chain(
-                    function_matches.items(), bb_matches.items(), insn_matches.items()
-                ):
-                    # in practice, most matches are derived rules,
-                    # like "check OS version/5bf4c7f39fd4492cbed0f6dc7d596d49"
-                    # but when we log to the human, they really care about "real" rules.
-                    if not ruleset.rules[name].is_subscope_rule():
-                        match_count += len(matches_)
-
-                logger.debug(
-                    "analyzed function 0x%x and extracted %d features, %d matches in %0.02fs",
-                    f.address,
-                    feature_count,
-                    match_count,
-                    t1 - t0,
+    functions: List[FunctionHandle] = list(extractor.get_functions())
+    n_funcs: int = len(functions)
+    n_libs: int = 0
+    percentage: float = 0
+
+    with capa.helpers.CapaProgressBar(
+        console=capa.helpers.log_console, transient=True, disable=disable_progress
+    ) as pbar:
+        task = pbar.add_task(
+            "matching", total=n_funcs, unit="functions", postfix=f"skipped {n_libs} library functions, {percentage}%"
+        )
+        for f in functions:
+            t0 = time.time()
+            if extractor.is_library_function(f.address):
+                function_name = extractor.get_function_name(f.address)
+                logger.debug("skipping library function 0x%x (%s)", f.address, function_name)
+                library_functions += (
+                    rdoc.LibraryFunction(address=frz.Address.from_capa(f.address), name=function_name),
                 )
-
-                for rule_name, res in function_matches.items():
-                    all_function_matches[rule_name].extend(res)
-                for rule_name, res in bb_matches.items():
-                    all_bb_matches[rule_name].extend(res)
-                for rule_name, res in insn_matches.items():
-                    all_insn_matches[rule_name].extend(res)
+                n_libs = len(library_functions)
+                percentage = round(100 * (n_libs / n_funcs))
+                pbar.update(task, postfix=f"skipped {n_libs} library functions, {percentage}%")
+                pbar.advance(task)
+                continue
+
+            function_matches, bb_matches, insn_matches, feature_count = find_code_capabilities(ruleset, extractor, f)
+            feature_counts.functions += (
+                rdoc.FunctionFeatureCount(address=frz.Address.from_capa(f.address), count=feature_count),
+            )
+            t1 = time.time()
+
+            match_count = 0
+            for name, matches_ in itertools.chain(function_matches.items(), bb_matches.items(), insn_matches.items()):
+                if not ruleset.rules[name].is_subscope_rule():
+                    match_count += len(matches_)
+
+            logger.debug(
+                "analyzed function 0x%x and extracted %d features, %d matches in %0.02fs",
+                f.address,
+                feature_count,
+                match_count,
+                t1 - t0,
+            )
+
+            for rule_name, res in function_matches.items():
+                all_function_matches[rule_name].extend(res)
+            for rule_name, res in bb_matches.items():
+                all_bb_matches[rule_name].extend(res)
+            for rule_name, res in insn_matches.items():
+                all_insn_matches[rule_name].extend(res)
+
+            pbar.advance(task)
 
     # collection of features that captures the rule matches within function, BB, and instruction scopes.
     # mapping from feature (matched rule) to set of addresses at which it matched.

capa/features/extractors/binja/file.py

@@ -5,8 +5,6 @@
 # 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 struct
 from typing import Tuple, Iterator
 
 from binaryninja import Segment, BinaryView, SymbolType, SymbolBinding
@@ -20,56 +18,24 @@
 from capa.features.extractors.binja.helpers import read_c_string, unmangle_c_name
 
 
-def check_segment_for_pe(bv: BinaryView, seg: Segment) -> Iterator[Tuple[int, int]]:
-    """check segment for embedded PE
-
-    adapted for binja from:
-    https://github.com/vivisect/vivisect/blob/7be4037b1cecc4551b397f840405a1fc606f9b53/PE/carve.py#L19
-    """
-    mz_xor = [
-        (
-            capa.features.extractors.helpers.xor_static(b"MZ", i),
-            capa.features.extractors.helpers.xor_static(b"PE", i),
-            i,
-        )
-        for i in range(256)
-    ]
-
-    todo = []
-    # If this is the first segment of the binary, skip the first bytes. Otherwise, there will always be a matched
-    # PE at the start of the binaryview.
-    start = seg.start
-    if bv.view_type == "PE" and start == bv.start:
+def check_segment_for_pe(bv: BinaryView, seg: Segment) -> Iterator[Tuple[Feature, Address]]:
+    """check segment for embedded PE"""
+    start = 0
+    if bv.view_type == "PE" and seg.start == bv.start:
+        # If this is the first segment of the binary, skip the first bytes.
+        # Otherwise, there will always be a matched PE at the start of the binaryview.
         start += 1
 
-    for mzx, pex, i in mz_xor:
-        for off, _ in bv.find_all_data(start, seg.end, 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
-        e_lfanew = off + 0x3C
-
-        if seg.end < (e_lfanew + 4):
-            continue
-
-        newoff = struct.unpack("<I", capa.features.extractors.helpers.xor_static(bv.read(e_lfanew, 4), i))[0]
-
-        peoff = off + newoff
-        if seg.end < (peoff + 2):
-            continue
+    buf = bv.read(seg.start, seg.length)
 
-        if bv.read(peoff, 2) == pex:
-            yield off, i
+    for offset, _ in capa.features.extractors.helpers.carve_pe(buf, start):
+        yield Characteristic("embedded pe"), FileOffsetAddress(seg.start + offset)
 
 
 def extract_file_embedded_pe(bv: BinaryView) -> Iterator[Tuple[Feature, Address]]:
     """extract embedded PE features"""
     for seg in bv.segments:
-        for ea, _ in check_segment_for_pe(bv, seg):
-            yield Characteristic("embedded pe"), FileOffsetAddress(ea)
+        yield from check_segment_for_pe(bv, seg)
 
 
 def extract_file_export_names(bv: BinaryView) -> Iterator[Tuple[Feature, Address]]: