diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c008f93..5fe6ae2b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/README.md b/README.md index a700f894..16d561bc 100644 --- a/README.md +++ b/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-797-blue.svg)](https://github.com/mandiant/capa-rules) +[![Number of rules](https://img.shields.io/badge/rules-799-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) diff --git a/capa/helpers.py b/capa/helpers.py index 25d9efe0..c03e0553 100644 --- a/capa/helpers.py +++ b/capa/helpers.py @@ -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 + # + 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.") diff --git a/capa/main.py b/capa/main.py index 6d270fc1..ee9676cd 100644 --- a/capa/main.py +++ b/capa/main.py @@ -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. diff --git a/rules b/rules index 312d4cad..188e6552 160000 --- a/rules +++ b/rules @@ -1 +1 @@ -Subproject commit 312d4cad891498e1d360dffcc98f669b63869c94 +Subproject commit 188e65528ec496eaaa792c3470cb4ab680a1b156 diff --git a/scripts/detect_duplicate_features.py b/scripts/detect_duplicate_features.py new file mode 100644 index 00000000..ff21fd0b --- /dev/null +++ b/scripts/detect_duplicate_features.py @@ -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()) diff --git a/scripts/lint.py b/scripts/lint.py index 92c7fbcf..a80d3e12 100644 --- a/scripts/lint.py +++ b/scripts/lint.py @@ -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 - # - 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)) diff --git a/setup.py b/setup.py index 6e36bb21..2f8c5348 100644 --- a/setup.py +++ b/setup.py @@ -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", diff --git a/tests/data b/tests/data index 91e2796c..a37873c8 160000 --- a/tests/data +++ b/tests/data @@ -1 +1 @@ -Subproject commit 91e2796c9950eb1942b6b910bc42714110d07f8b +Subproject commit a37873c8a571b515f2baaf19bfcfaff5c7ef5342 diff --git a/tests/test_scripts.py b/tests/test_scripts.py index 3b299a36..2d8fefac 100644 --- a/tests/test_scripts.py +++ b/tests/test_scripts.py @@ -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