mirror of
https://github.com/mandiant/capa.git
synced 2026-07-08 05:17:31 -07:00
Merge branch 'mandiant:master' into ghidra_backend
This commit is contained in:
Colton Gabertan
+6
-1
@@ -3,15 +3,19 @@
|
||||
## master (unreleased)
|
||||
|
||||
### New Features
|
||||
- Utility script to detect feature overlap between new and existing CAPA rules [#1451](https://github.com/mandiant/capa/issues/1451) [@Aayush-Goel-04](https://github.com/aayush-goel-04)
|
||||
|
||||
### Breaking Changes
|
||||
|
||||
### New Rules (4)
|
||||
### New Rules (7)
|
||||
|
||||
- load-code/shellcode/execute-shellcode-via-windows-callback-function ervin.ocampo@mandiant.com jakub.jozwiak@mandiant.com
|
||||
- nursery/execute-shellcode-via-indirect-call ronnie.salomonsen@mandiant.com
|
||||
- data-manipulation/encryption/aes/encrypt-data-using-aes-mixcolumns-step @mr-tz
|
||||
- linking/static/aplib/linked-against-aplib still@teamt5.org
|
||||
- communication/mailslot/read-from-mailslot nick.simonian@mandiant.com
|
||||
- nursery/hash-data-using-sha512managed-in-dotnet jonathanlepore@google.com
|
||||
- nursery/compiled-with-exescript jonathanlepore@google.com
|
||||
-
|
||||
|
||||
### Bug Fixes
|
||||
@@ -22,6 +26,7 @@
|
||||
- improve ELF strtab and needed parsing @mr-tz
|
||||
- better handle exceptional cases when parsing ELF files [#1458](https://github.com/mandiant/capa/issues/1458) [@Aayush-Goel-04](https://github.com/aayush-goel-04)
|
||||
- Improved testing coverage for Binary Ninja Backend [#1446](https://github.com/mandiant/capa/issues/1446) [@Aayush-Goel-04](https://github.com/aayush-goel-04)
|
||||
- Add logging and print redirect to tqdm for capa main [#749](https://github.com/mandiant/capa/issues/749) [@Aayush-Goel-04](https://github.com/aayush-goel-04)
|
||||
- extractor: fix binja installation path detection does not work with Python 3.11
|
||||
|
||||
### capa explorer IDA Pro plugin
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
[](https://pypi.org/project/flare-capa)
|
||||
[](https://github.com/mandiant/capa/releases)
|
||||
[](https://github.com/mandiant/capa-rules)
|
||||
[](https://github.com/mandiant/capa-rules)
|
||||
[](https://github.com/mandiant/capa/actions?query=workflow%3ACI+event%3Apush+branch%3Amaster)
|
||||
[](https://github.com/mandiant/capa/releases)
|
||||
[](LICENSE.txt)
|
||||
|
||||
@@ -6,9 +6,13 @@
|
||||
# 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 os
|
||||
import inspect
|
||||
import logging
|
||||
import contextlib
|
||||
from typing import NoReturn
|
||||
|
||||
import tqdm
|
||||
|
||||
from capa.exceptions import UnsupportedFormatError
|
||||
from capa.features.common import FORMAT_PE, FORMAT_SC32, FORMAT_SC64, FORMAT_DOTNET, FORMAT_UNKNOWN, Format
|
||||
|
||||
@@ -85,6 +89,39 @@ def get_format(sample: str) -> str:
|
||||
return FORMAT_UNKNOWN
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def redirecting_print_to_tqdm(disable_progress):
|
||||
"""
|
||||
tqdm (progress bar) expects to have fairly tight control over console output.
|
||||
so calls to `print()` will break the progress bar and make things look bad.
|
||||
so, this context manager temporarily replaces the `print` implementation
|
||||
with one that is compatible with tqdm.
|
||||
via: https://stackoverflow.com/a/42424890/87207
|
||||
"""
|
||||
old_print = print
|
||||
|
||||
def new_print(*args, **kwargs):
|
||||
# If tqdm.tqdm.write raises error, use builtin print
|
||||
if disable_progress:
|
||||
old_print(*args, **kwargs)
|
||||
else:
|
||||
try:
|
||||
tqdm.tqdm.write(*args, **kwargs)
|
||||
except:
|
||||
old_print(*args, **kwargs)
|
||||
|
||||
try:
|
||||
# Globally replace print with new_print.
|
||||
# Verified this works manually on Python 3.11:
|
||||
# >>> import inspect
|
||||
# >>> inspect.builtins
|
||||
# <module 'builtins' (built-in)>
|
||||
inspect.builtins.print = new_print # type: ignore
|
||||
yield
|
||||
finally:
|
||||
inspect.builtins.print = old_print # type: ignore
|
||||
|
||||
|
||||
def log_unsupported_format_error():
|
||||
logger.error("-" * 80)
|
||||
logger.error(" Input file does not appear to be a PE or ELF file.")
|
||||
|
||||
+34
-28
@@ -25,6 +25,7 @@ from typing import Any, Dict, List, Tuple, Callable
|
||||
import halo
|
||||
import tqdm
|
||||
import colorama
|
||||
import tqdm.contrib.logging
|
||||
from pefile import PEFormatError
|
||||
from elftools.common.exceptions import ELFError
|
||||
|
||||
@@ -53,6 +54,7 @@ from capa.helpers import (
|
||||
get_file_taste,
|
||||
get_auto_format,
|
||||
log_unsupported_os_error,
|
||||
redirecting_print_to_tqdm,
|
||||
log_unsupported_arch_error,
|
||||
log_unsupported_format_error,
|
||||
)
|
||||
@@ -251,38 +253,42 @@ def find_capabilities(ruleset: RuleSet, extractor: FeatureExtractor, disable_pro
|
||||
"library_functions": {},
|
||||
} # type: Dict[str, Any]
|
||||
|
||||
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
|
||||
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
|
||||
|
||||
functions = list(extractor.get_functions())
|
||||
n_funcs = len(functions)
|
||||
functions = list(extractor.get_functions())
|
||||
n_funcs = len(functions)
|
||||
|
||||
pb = pbar(functions, desc="matching", unit=" functions", postfix="skipped 0 library functions")
|
||||
for f in pb:
|
||||
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)
|
||||
meta["library_functions"][f.address] = function_name
|
||||
n_libs = len(meta["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
|
||||
pb = pbar(functions, desc="matching", unit=" functions", postfix="skipped 0 library functions")
|
||||
for f in pb:
|
||||
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)
|
||||
meta["library_functions"][f.address] = function_name
|
||||
n_libs = len(meta["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)
|
||||
meta["feature_counts"]["functions"][f.address] = feature_count
|
||||
logger.debug("analyzed function 0x%x and extracted %d features", f.address, feature_count)
|
||||
function_matches, bb_matches, insn_matches, feature_count = find_code_capabilities(
|
||||
ruleset, extractor, f
|
||||
)
|
||||
meta["feature_counts"]["functions"][f.address] = feature_count
|
||||
logger.debug("analyzed function 0x%x and extracted %d features", f.address, feature_count)
|
||||
|
||||
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)
|
||||
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)
|
||||
|
||||
# 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.
|
||||
|
||||
+1
-1
Submodule rules updated: 312d4cad89...188e65528e
@@ -0,0 +1,111 @@
|
||||
import sys
|
||||
import logging
|
||||
import argparse
|
||||
|
||||
import capa.main
|
||||
import capa.rules
|
||||
import capa.engine as ceng
|
||||
|
||||
logger = logging.getLogger("detect_duplicate_features")
|
||||
|
||||
|
||||
def get_child_features(feature: ceng.Statement) -> list:
|
||||
"""
|
||||
Recursively extracts all feature statements from a given rule statement.
|
||||
|
||||
Args:
|
||||
feature (capa.engine.Statement): The feature statement to extract features from.
|
||||
|
||||
Returns:
|
||||
list: A list of all feature statements contained within the given feature statement.
|
||||
"""
|
||||
children = []
|
||||
|
||||
if isinstance(feature, (ceng.And, ceng.Or, ceng.Some)):
|
||||
for child in feature.children:
|
||||
children.extend(get_child_features(child))
|
||||
elif isinstance(feature, (ceng.Subscope, ceng.Range, ceng.Not)):
|
||||
children.extend(get_child_features(feature.child))
|
||||
else:
|
||||
children.append(feature)
|
||||
return children
|
||||
|
||||
|
||||
def get_features(rule_path: str) -> list:
|
||||
"""
|
||||
Extracts all features from a given rule file.
|
||||
|
||||
Args:
|
||||
rule_path (str): The path to the rule file to extract features from.
|
||||
|
||||
Returns:
|
||||
list: A list of all feature statements contained within the rule file.
|
||||
"""
|
||||
feature_list = []
|
||||
with open(rule_path, "r") as f:
|
||||
try:
|
||||
new_rule = capa.rules.Rule.from_yaml(f.read())
|
||||
feature_list = get_child_features(new_rule.statement)
|
||||
except Exception as e:
|
||||
logger.error("Error: New rule " + rule_path + " " + str(type(e)) + " " + str(e))
|
||||
sys.exit(-1)
|
||||
return feature_list
|
||||
|
||||
|
||||
def find_overlapping_rules(new_rule_path, rules_path):
|
||||
if not new_rule_path.endswith(".yml"):
|
||||
logger.error("FileNotFoundError ! New rule file name doesn't end with .yml")
|
||||
sys.exit(-1)
|
||||
|
||||
# Loads features of new rule in a list.
|
||||
new_rule_features = get_features(new_rule_path)
|
||||
|
||||
count = 0
|
||||
overlapping_rules = []
|
||||
|
||||
# capa.rules.RuleSet stores all rules in given paths
|
||||
ruleset = capa.main.get_rules(rules_path)
|
||||
|
||||
for rule_name, rule in ruleset.rules.items():
|
||||
rule_features = get_child_features(rule.statement)
|
||||
|
||||
if not len(rule_features):
|
||||
continue
|
||||
count += 1
|
||||
# Checks if any features match between existing and new rule.
|
||||
if any([feature in rule_features for feature in new_rule_features]):
|
||||
overlapping_rules.append(rule_name)
|
||||
|
||||
result = {"overlapping_rules": overlapping_rules, "count": count}
|
||||
return result
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Find overlapping features in Capa rules.")
|
||||
|
||||
parser.add_argument("rules", type=str, action="append", help="Path to rules")
|
||||
parser.add_argument("new_rule", type=str, help="Path to new rule")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
new_rule_path = args.new_rule
|
||||
rules_path = args.rules
|
||||
|
||||
result = find_overlapping_rules(new_rule_path, rules_path)
|
||||
|
||||
print("\nNew rule path : %s" % new_rule_path)
|
||||
print("Number of rules checked : %s " % result["count"])
|
||||
if result["overlapping_rules"]:
|
||||
print("Paths to overlapping rules : ")
|
||||
for r in result["overlapping_rules"]:
|
||||
print("- %s" % r)
|
||||
else:
|
||||
print("Paths to overlapping rules : None")
|
||||
print("Number of rules containing same features : %s" % len(result["overlapping_rules"]))
|
||||
print("\n")
|
||||
|
||||
return len(result["overlapping_rules"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
+1
-34
@@ -22,13 +22,11 @@ import time
|
||||
import string
|
||||
import difflib
|
||||
import hashlib
|
||||
import inspect
|
||||
import logging
|
||||
import pathlib
|
||||
import argparse
|
||||
import itertools
|
||||
import posixpath
|
||||
import contextlib
|
||||
from typing import Set, Dict, List
|
||||
from pathlib import Path
|
||||
from dataclasses import field, dataclass
|
||||
@@ -866,37 +864,6 @@ def width(s, count):
|
||||
return s.ljust(count)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def redirecting_print_to_tqdm():
|
||||
"""
|
||||
tqdm (progress bar) expects to have fairly tight control over console output.
|
||||
so calls to `print()` will break the progress bar and make things look bad.
|
||||
so, this context manager temporarily replaces the `print` implementation
|
||||
with one that is compatible with tqdm.
|
||||
|
||||
via: https://stackoverflow.com/a/42424890/87207
|
||||
"""
|
||||
old_print = print
|
||||
|
||||
def new_print(*args, **kwargs):
|
||||
# If tqdm.tqdm.write raises error, use builtin print
|
||||
try:
|
||||
tqdm.tqdm.write(*args, **kwargs)
|
||||
except:
|
||||
old_print(*args, **kwargs)
|
||||
|
||||
try:
|
||||
# Globally replace print with new_print.
|
||||
# Verified this works manually on Python 3.11:
|
||||
# >>> import inspect
|
||||
# >>> inspect.builtins
|
||||
# <module 'builtins' (built-in)>
|
||||
inspect.builtins.print = new_print # type: ignore
|
||||
yield
|
||||
finally:
|
||||
inspect.builtins.print = old_print # type: ignore
|
||||
|
||||
|
||||
def lint(ctx: Context):
|
||||
"""
|
||||
Returns: Dict[string, Tuple(int, int)]
|
||||
@@ -907,7 +874,7 @@ def lint(ctx: Context):
|
||||
|
||||
source_rules = [rule for rule in ctx.rules.rules.values() if not rule.is_subscope_rule()]
|
||||
with tqdm.contrib.logging.tqdm_logging_redirect(source_rules, unit="rule") as pbar:
|
||||
with redirecting_print_to_tqdm():
|
||||
with capa.helpers.redirecting_print_to_tqdm(False):
|
||||
for rule in pbar:
|
||||
name = rule.name
|
||||
pbar.set_description(width(f"linting rule: {name}", 48))
|
||||
|
||||
@@ -21,14 +21,14 @@ requirements = [
|
||||
"viv-utils[flirt]==0.7.9",
|
||||
"halo==0.0.31",
|
||||
"networkx==2.5.1", # newer versions no longer support py3.7.
|
||||
"ruamel.yaml==0.17.21",
|
||||
"ruamel.yaml==0.17.28",
|
||||
"vivisect==1.1.1",
|
||||
"pefile==2023.2.7",
|
||||
"pyelftools==0.29",
|
||||
"dnfile==0.13.0",
|
||||
"dncil==1.0.2",
|
||||
"pydantic==1.10.7",
|
||||
"protobuf==4.22.3",
|
||||
"protobuf==4.23.2",
|
||||
]
|
||||
|
||||
# this sets __version__
|
||||
@@ -74,7 +74,7 @@ setuptools.setup(
|
||||
"pytest-instafail==0.5.0",
|
||||
"pytest-cov==4.0.0",
|
||||
"pycodestyle==2.10.0",
|
||||
"ruff==0.0.265",
|
||||
"ruff==0.0.270",
|
||||
"black==23.3.0",
|
||||
"isort==5.11.4",
|
||||
"mypy==1.3.0",
|
||||
@@ -90,7 +90,7 @@ setuptools.setup(
|
||||
"types-termcolor==1.1.4",
|
||||
"types-psutil==5.8.23",
|
||||
"types_requests==2.28.1",
|
||||
"types-protobuf==4.22.0.2",
|
||||
"types-protobuf==4.23.0.1",
|
||||
],
|
||||
"build": [
|
||||
"pyinstaller==5.10.1",
|
||||
|
||||
+1
-1
Submodule tests/data updated: 91e2796c99...a37873c8a5
@@ -8,9 +8,11 @@
|
||||
|
||||
import os
|
||||
import sys
|
||||
import textwrap
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
from fixtures import *
|
||||
|
||||
CD = os.path.dirname(__file__)
|
||||
|
||||
@@ -82,3 +84,112 @@ def test_proto_conversion(tmpdir):
|
||||
assert p.returncode == 0
|
||||
|
||||
assert p.stdout.startswith(b'{\n "meta": ') or p.stdout.startswith(b'{\r\n "meta": ')
|
||||
|
||||
|
||||
def test_detect_duplicate_features(tmpdir):
|
||||
TEST_RULE_0 = textwrap.dedent(
|
||||
"""
|
||||
rule:
|
||||
meta:
|
||||
name: Test Rule 0
|
||||
scope: function
|
||||
features:
|
||||
- and:
|
||||
- number: 1
|
||||
- not:
|
||||
- string: process
|
||||
"""
|
||||
)
|
||||
|
||||
TEST_RULESET = {
|
||||
"rule_1": textwrap.dedent(
|
||||
"""
|
||||
rule:
|
||||
meta:
|
||||
name: Test Rule 1
|
||||
features:
|
||||
- or:
|
||||
- string: unique
|
||||
- number: 2
|
||||
- and:
|
||||
- or:
|
||||
- arch: i386
|
||||
- number: 4
|
||||
- not:
|
||||
- count(mnemonic(xor)): 5
|
||||
- not:
|
||||
- os: linux
|
||||
"""
|
||||
),
|
||||
"rule_2": textwrap.dedent(
|
||||
"""
|
||||
rule:
|
||||
meta:
|
||||
name: Test Rule 2
|
||||
features:
|
||||
- and:
|
||||
- string: "sites.ini"
|
||||
- basic block:
|
||||
- and:
|
||||
- api: CreateFile
|
||||
- mnemonic: xor
|
||||
"""
|
||||
),
|
||||
"rule_3": textwrap.dedent(
|
||||
"""
|
||||
rule:
|
||||
meta:
|
||||
name: Test Rule 3
|
||||
features:
|
||||
- or:
|
||||
- not:
|
||||
- number: 4
|
||||
- basic block:
|
||||
- and:
|
||||
- api: bind
|
||||
- number: 2
|
||||
"""
|
||||
),
|
||||
"rule_4": textwrap.dedent(
|
||||
"""
|
||||
rule:
|
||||
meta:
|
||||
name: Test Rule 4
|
||||
features:
|
||||
- not:
|
||||
- string: "expa"
|
||||
"""
|
||||
),
|
||||
}
|
||||
|
||||
"""
|
||||
The rule_overlaps list represents the number of overlaps between each rule in the RULESET.
|
||||
An overlap includes a rule overlap with itself.
|
||||
The scripts
|
||||
The overlaps are like:
|
||||
- Rule 0 has zero overlaps in RULESET
|
||||
- Rule 1 overlaps with 3 other rules in RULESET
|
||||
- Rule 4 overlaps with itself in RULESET
|
||||
These overlap values indicate the number of rules with which
|
||||
each rule in RULESET has overlapping features.
|
||||
"""
|
||||
rule_overlaps = [0, 4, 3, 3, 1]
|
||||
|
||||
rule_dir = tmpdir.mkdir("capa_rule_overlap_test")
|
||||
rule_paths = []
|
||||
|
||||
rule_file = tmpdir.join("rule_0.yml")
|
||||
rule_file.write(TEST_RULE_0)
|
||||
rule_paths.append(rule_file.strpath)
|
||||
|
||||
for rule_name, RULE_CONTENT in TEST_RULESET.items():
|
||||
rule_file = rule_dir.join("%s.yml" % rule_name)
|
||||
rule_file.write(RULE_CONTENT)
|
||||
rule_paths.append(rule_file.strpath)
|
||||
|
||||
# tests if number of overlaps for rules in RULESET found are correct.
|
||||
script_path = get_script_path("detect_duplicate_features.py")
|
||||
for expected_overlaps, rule_path in zip(rule_overlaps, rule_paths):
|
||||
args = [rule_dir.strpath, rule_path]
|
||||
overlaps_found = run_program(script_path, args)
|
||||
assert overlaps_found.returncode == expected_overlaps
|
||||
|
||||
Reference in New Issue
Block a user