Merge pull request #905 from mandiant/v320

release v3.2.0

CHANGELOG.md

@@ -12,15 +12,38 @@
 
 ### Bug Fixes
 
-- elf: fix OS detection for Linux kernel modules #867 @williballenthin
-
 ### capa explorer IDA Pro plugin
 
 ### Development
 
 ### Raw diffs
-- [capa v3.1.0...master](https://github.com/mandiant/capa/compare/v3.1.0...master)
-- [capa-rules v3.1.0...master](https://github.com/mandiant/capa-rules/compare/v3.1.0...master)
+- [capa v3.2.0...master](https://github.com/mandiant/capa/compare/v3.2.0...master)
+- [capa-rules v3.2.0...master](https://github.com/mandiant/capa-rules/compare/v3.2.0...master)
+
+## v3.2.0 (2022-03-03)
+This release adds a new characteristic `characteristic: call $+5` enabling users to create more explicit rules. The linter now also validates ATT&CK and MBC categories. Additionally, many dependencies, including the vivisect backend, have been updated.
+
+One rule has been added and many more have been improved.
+
+Thanks for all the support, especially to @kn0wl3dge and first time contributor @uckelman-sf!
+
+### New Features
+
+- linter: validate ATT&CK/MBC categories and IDs #103 @kn0wl3dge
+- extractor: add characteristic "call $+5" feature #366 @kn0wl3dge
+
+### New Rules (1)
+
+- anti-analysis/obfuscation/obfuscated-with-advobfuscator jakub.jozwiak@mandiant.com
+
+### Bug Fixes
+
+- remove typing package as a requirement for Python 3.7+ compatibility #901 @uckelman-sf
+- elf: fix OS detection for Linux kernel modules #867 @williballenthin
+
+### Raw diffs
+- [capa v3.1.0...v3.2.0](https://github.com/mandiant/capa/compare/v3.1.0...v3.2.0)
+- [capa-rules v3.1.0...v3.2.0](https://github.com/mandiant/capa-rules/compare/v3.1.0...v3.2.0)
 
 ## v3.1.0 (2022-01-10)
 This release improves the performance of capa while also adding 23 new rules and many code quality enhancements. We profiled capa's CPU usage and optimized the way that it matches rules, such as by short circuiting when appropriate. According to our testing, the matching phase is approximately 66% faster than v3.0.3! We also added support for Python 3.10, aarch64 builds, and additional MAEC metadata in the rule headers.

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-658-blue.svg)](https://github.com/mandiant/capa-rules)
+[![Number of rules](https://img.shields.io/badge/rules-661-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/features/extractors/ida/insn.py

@@ -340,6 +340,17 @@ def extract_insn_mnemonic_features(f, bb, insn):
     yield Mnemonic(idc.print_insn_mnem(insn.ea)), insn.ea
 
 
+def extract_insn_obfs_call_plus_5_characteristic_features(f, bb, insn):
+    """
+    parse call $+5 instruction from the given instruction.
+    """
+    if not idaapi.is_call_insn(insn):
+        return
+
+    if insn.ea + 5 == idc.get_operand_value(insn.ea, 0):
+        yield Characteristic("call $+5"), insn.ea
+
+
 def extract_insn_peb_access_characteristic_features(f, bb, insn):
     """parse instruction peb access
 
@@ -455,6 +466,7 @@ INSTRUCTION_HANDLERS = (
     extract_insn_offset_features,
     extract_insn_nzxor_characteristic_features,
     extract_insn_mnemonic_features,
+    extract_insn_obfs_call_plus_5_characteristic_features,
     extract_insn_peb_access_characteristic_features,
     extract_insn_cross_section_cflow,
     extract_insn_segment_access_features,

capa/features/extractors/smda/insn.py

@@ -280,6 +280,20 @@ def extract_insn_mnemonic_features(f, bb, insn):
     yield Mnemonic(insn.mnemonic), insn.offset
 
 
+def extract_insn_obfs_call_plus_5_characteristic_features(f, bb, insn):
+    """
+    parse call $+5 instruction from the given instruction.
+    """
+    if insn.mnemonic != "call":
+        return
+
+    if not insn.operands.startswith("0x"):
+        return
+
+    if int(insn.operands, 16) == insn.offset + 5:
+        yield Characteristic("call $+5"), insn.offset
+
+
 def extract_insn_peb_access_characteristic_features(f, bb, insn):
     """
     parse peb access from the given function. fs:[0x30] on x86, gs:[0x60] on x64
@@ -389,6 +403,7 @@ INSTRUCTION_HANDLERS = (
     extract_insn_offset_features,
     extract_insn_nzxor_characteristic_features,
     extract_insn_mnemonic_features,
+    extract_insn_obfs_call_plus_5_characteristic_features,
     extract_insn_peb_access_characteristic_features,
     extract_insn_cross_section_cflow,
     extract_insn_segment_access_features,

capa/features/extractors/viv/insn.py

@@ -453,6 +453,24 @@ def extract_insn_mnemonic_features(f, bb, insn):
     yield Mnemonic(insn.mnem), insn.va
 
 
+def extract_insn_obfs_call_plus_5_characteristic_features(f, bb, insn):
+    """
+    parse call $+5 instruction from the given instruction.
+    """
+    if insn.mnem != "call":
+        return
+
+    if isinstance(insn.opers[0], envi.archs.i386.disasm.i386PcRelOper):
+        if insn.va + 5 == insn.opers[0].getOperValue(insn):
+            yield Characteristic("call $+5"), insn.va
+
+    if isinstance(insn.opers[0], envi.archs.i386.disasm.i386ImmMemOper) or isinstance(
+        insn.opers[0], envi.archs.amd64.disasm.Amd64RipRelOper
+    ):
+        if insn.va + 5 == insn.opers[0].getOperAddr(insn):
+            yield Characteristic("call $+5"), insn.va
+
+
 def extract_insn_peb_access_characteristic_features(f, bb, insn):
     """
     parse peb access from the given function. fs:[0x30] on x86, gs:[0x60] on x64
@@ -626,6 +644,7 @@ INSTRUCTION_HANDLERS = (
     extract_insn_offset_features,
     extract_insn_nzxor_characteristic_features,
     extract_insn_mnemonic_features,
+    extract_insn_obfs_call_plus_5_characteristic_features,
     extract_insn_peb_access_characteristic_features,
     extract_insn_cross_section_cflow,
     extract_insn_segment_access_features,

capa/rules.py

@@ -120,6 +120,7 @@ SUPPORTED_FEATURES = {
         capa.features.common.Characteristic("tight loop"),
         capa.features.common.Characteristic("stack string"),
         capa.features.common.Characteristic("indirect call"),
+        capa.features.common.Characteristic("call $+5"),
         capa.features.common.OS,
         capa.features.common.Arch,
     },

capa/version.py

@@ -1 +1 @@
-__version__ = "3.1.0"
+__version__ = "3.2.0"

scripts/capa_as_library.py

@@ -163,13 +163,12 @@ def render_dictionary(doc):
 
 # ==== render dictionary helpers
 def capa_details(file_path, output_format="dictionary"):
-    # collect metadata (used only to make rendering more complete)
-    meta = capa.main.collect_metadata("", file_path, RULES_PATH, extractor)
-
     # extract features and find capabilities
     extractor = capa.main.get_extractor(file_path, "auto", capa.main.BACKEND_VIV, [], False, disable_progress=True)
     capabilities, counts = capa.main.find_capabilities(rules, extractor, disable_progress=True)
 
+    # collect metadata (used only to make rendering more complete)
+    meta = capa.main.collect_metadata("", file_path, RULES_PATH, extractor)
     meta["analysis"].update(counts)
     meta["analysis"]["layout"] = capa.main.compute_layout(rules, extractor, capabilities)
 

scripts/lint.py

@@ -15,14 +15,15 @@ See the License for the specific language governing permissions and limitations
 """
 import gc
 import os
+import re
 import sys
+import json
 import time
 import string
 import difflib
 import hashlib
 import inspect
 import logging
-import os.path
 import pathlib
 import argparse
 import itertools
@@ -221,6 +222,61 @@ class ExampleFileDNE(Lint):
         return not found
 
 
+class InvalidAttckOrMbcTechnique(Lint):
+    name = "att&ck/mbc entry is malformed or does not exist"
+    recommendation = """
+    The att&ck and mbc fields must respect the following format:
+    <Tactic/Objective>::<Technique/Behavior> [<ID>]
+    OR
+    <Tactic/Objective>::<Technique/Behavior>::<Subtechnique/Method> [<ID.SubID>]
+    """
+
+    def __init__(self):
+        super(InvalidAttckOrMbcTechnique, self).__init__()
+
+        try:
+            with open(f"{os.path.dirname(__file__)}/linter-data.json", "rb") as fd:
+                self.data = json.load(fd)
+            self.enabled_frameworks = self.data.keys()
+        except BaseException:
+            # If linter-data.json is not present, or if an error happen
+            # we log an error and lint nothing.
+            logger.warning(
+                "Could not load 'scripts/linter-data.json'. The att&ck and mbc information will not be linted."
+            )
+            self.enabled_frameworks = []
+
+        # This regex matches the format defined in the recommendation attribute
+        self.reg = re.compile("^([\w\s-]+)::(.+) \[([A-Za-z0-9.]+)\]$")
+
+    def _entry_check(self, framework, category, entry, eid):
+        if category not in self.data[framework].keys():
+            self.recommendation = f'Unknown category: "{category}"'
+            return True
+        if eid not in self.data[framework][category].keys():
+            self.recommendation = f"Unknown entry ID: {eid}"
+            return True
+        if self.data[framework][category][eid] != entry:
+            self.recommendation = (
+                f'{eid} should be associated to entry "{self.data[framework][category][eid]}" instead of "{entry}"'
+            )
+            return True
+        return False
+
+    def check_rule(self, ctx: Context, rule: Rule):
+        for framework in self.enabled_frameworks:
+            if framework in rule.meta.keys():
+                for r in rule.meta[framework]:
+                    m = self.reg.match(r)
+                    if m is None:
+                        return True
+
+                    args = m.group(1, 2, 3)
+                    if self._entry_check(framework, *args):
+                        return True
+        return False
+
+
 DEFAULT_SIGNATURES = capa.main.get_default_signatures()
 
 
@@ -647,6 +703,7 @@ META_LINTS = (
     UnusualMetaField(),
     LibRuleNotInLibDirectory(),
     LibRuleHasNamespace(),
+    InvalidAttckOrMbcTechnique(),
 )
 
 

scripts/setup-linter-dependencies.py

@@ -0,0 +1,190 @@
+"""
+Generate capa linter-data.json, used to validate Att&ck/MBC IDs and names.
+
+Use the --extractor option to extract data from Att&ck or MBC (or both) frameworks.
+Use the --output to choose the output json file.
+By default, the script will create a linter-data.json in the scripts/ directory for both frameworks.
+
+Note: The capa rules linter will try to load from its default location (scripts/linter-data.json).
+
+Usage:
+
+    usage: setup-linter-dependencies.py [-h] [--extractor {both,mbc,att&ck}] [--output OUTPUT]
+
+    Setup linter dependencies.
+
+    optional arguments:
+      -h, --help            show this help message and exit
+      --extractor {both,mbc,att&ck}
+                            Extractor that will be run
+      --output OUTPUT, -o OUTPUT
+                            Path to output file (lint.py will be looking for linter-data.json)
+
+
+Example:
+
+    $ python3 setup-linter-dependencies.py
+    2022-01-24 22:35:06,901 [INFO] Extracting Mitre Att&ck techniques...
+    2022-01-24 22:35:06,901 [INFO] Downloading STIX data at: https://raw.githubusercontent.com/mitre-attack/attack-stix-data/master/enterprise-attack/enterprise-attack.json
+    2022-01-24 22:35:13,001 [INFO] Starting extraction...
+    2022-01-24 22:35:39,395 [INFO] Extracting MBC behaviors...
+    2022-01-24 22:35:39,395 [INFO] Downloading STIX data at: https://raw.githubusercontent.com/MBCProject/mbc-stix2/master/mbc/mbc.json
+    2022-01-24 22:35:39,839 [INFO] Starting extraction...
+    2022-01-24 22:35:42,632 [INFO] Writing results to linter-data.json
+"""
+import json
+import logging
+import argparse
+from sys import argv
+from typing import Dict, List
+from os.path import dirname
+
+import requests
+from stix2 import Filter, MemoryStore, AttackPattern  # type: ignore
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+
+class MitreExtractor:
+    """
+    This class extract Mitre techniques and sub techniques that are represented as "attack-pattern" in STIX format.
+    The STIX data is collected in JSON format by requesting the specified URL.
+
+    url: must point to json stix location
+    kill_chain_name: mitre-attack, mitre-mbc...
+    """
+
+    url = ""
+    kill_chain_name = ""
+
+    def __init__(self):
+        """Download and store in memory the STIX data on instantiation."""
+        if self.kill_chain_name == "":
+            raise ValueError(f"Kill chain name not specified in class {self.__class__.__name__}")
+
+        if self.url == "":
+            raise ValueError(f"URL not specified in class {self.__class__.__name__}")
+
+        logging.info(f"Downloading STIX data at: {self.url}")
+        stix_json = requests.get(self.url).json()
+        self._memory_store = MemoryStore(stix_data=stix_json["objects"])
+
+    @staticmethod
+    def _remove_deprecated_objetcs(stix_objects) -> List[AttackPattern]:
+        """Remove any revoked or deprecated objects from queries made to the data source."""
+        return list(
+            filter(
+                lambda x: x.get("x_mitre_deprecated", False) is False and x.get("revoked", False) is False,
+                stix_objects,
+            )
+        )
+
+    def _get_tactics(self) -> List[Dict]:
+        """Get tactics IDs from Mitre matrix."""
+        # Only one matrix for enterprise att&ck framework
+        matrix = self._remove_deprecated_objetcs(
+            self._memory_store.query(
+                [
+                    Filter("type", "=", "x-mitre-matrix"),
+                ]
+            )
+        )[0]
+        return list(map(self._memory_store.get, matrix["tactic_refs"]))
+
+    def _get_techniques_from_tactic(self, tactic: str) -> List[AttackPattern]:
+        """Get techniques and sub techniques from a Mitre tactic (kill_chain_phases->phase_name)"""
+        techniques = self._remove_deprecated_objetcs(
+            self._memory_store.query(
+                [
+                    Filter("type", "=", "attack-pattern"),
+                    Filter("kill_chain_phases.phase_name", "=", tactic),
+                    Filter("kill_chain_phases.kill_chain_name", "=", self.kill_chain_name),
+                ]
+            )
+        )
+        return techniques
+
+    def _get_parent_technique_from_subtechnique(self, technique: AttackPattern) -> AttackPattern:
+        """Get parent technique of a sub technique using the technique ID TXXXX.YYY"""
+        sub_id = technique["external_references"][0]["external_id"].split(".")[0]
+        parent_technique = self._remove_deprecated_objetcs(
+            self._memory_store.query(
+                [
+                    Filter("type", "=", "attack-pattern"),
+                    Filter("external_references.external_id", "=", sub_id),
+                ]
+            )
+        )[0]
+        return parent_technique
+
+    def run(self) -> Dict[str, Dict[str, str]]:
+        """Iterate over every technique over every tactic. If the technique is a sub technique, then
+        we also search for the parent technique name.
+        """
+        logging.info("Starting extraction...")
+        data: Dict[str, Dict[str, str]] = {}
+        for tactic in self._get_tactics():
+            data[tactic["name"]] = {}
+            for technique in self._get_techniques_from_tactic(tactic["x_mitre_shortname"]):
+                tid = technique["external_references"][0]["external_id"]
+                technique_name = technique["name"].split("::")[0]
+                if technique["x_mitre_is_subtechnique"]:
+                    parent_technique = self._get_parent_technique_from_subtechnique(technique)
+                    data[tactic["name"]][tid] = f"{parent_technique['name']}::{technique_name}"
+                else:
+                    data[tactic["name"]][tid] = technique_name
+        return data
+
+
+class AttckExtractor(MitreExtractor):
+    """Extractor for the Mitre Enterprise Att&ck Framework."""
+
+    url = "https://raw.githubusercontent.com/mitre-attack/attack-stix-data/master/enterprise-attack/enterprise-attack.json"
+    kill_chain_name = "mitre-attack"
+
+
+class MbcExtractor(MitreExtractor):
+    """Extractor for the Mitre Malware Behavior Catalog."""
+
+    url = "https://raw.githubusercontent.com/MBCProject/mbc-stix2/master/mbc/mbc.json"
+    kill_chain_name = "mitre-mbc"
+
+    def _get_tactics(self) -> List[Dict]:
+        """Override _get_tactics to edit the tactic name for Micro-objective"""
+        tactics = super(MbcExtractor, self)._get_tactics()
+        # We don't want the Micro-objective string inside objective names
+        for tactic in tactics:
+            tactic["name"] = tactic["name"].replace(" Micro-objective", "")
+        return tactics
+
+
+def main(args: argparse.Namespace) -> None:
+    data = {}
+    if args.extractor == "att&ck" or args.extractor == "both":
+        logging.info("Extracting Mitre Att&ck techniques...")
+        data["att&ck"] = AttckExtractor().run()
+    if args.extractor == "mbc" or args.extractor == "both":
+        logging.info("Extracting MBC behaviors...")
+        data["mbc"] = MbcExtractor().run()
+
+    logging.info(f"Writing results to {args.output}")
+    try:
+        with open(args.output, "w") as jf:
+            json.dump(data, jf, indent=2)
+    except BaseException as e:
+        logging.error(f"Exception encountered when writing results: {e}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Setup linter dependencies.")
+    parser.add_argument(
+        "--extractor", type=str, choices=["both", "mbc", "att&ck"], default="both", help="Extractor that will be run"
+    )
+    parser.add_argument(
+        "--output",
+        "-o",
+        type=str,
+        default=f"{dirname(__file__)}/linter-data.json",
+        help="Path to output file (lint.py will be looking for linter-data.json)",
+    )
+    main(parser.parse_args(args=argv[1:]))

setup.py

@@ -11,22 +11,21 @@ import os
 import setuptools
 
 requirements = [
-    "tqdm==4.62.3",
+    "tqdm==4.63.0",
     "pyyaml==6.0",
     "tabulate==0.8.9",
     "colorama==0.4.4",
     "termcolor==1.1.0",
     "wcwidth==0.2.5",
     "ida-settings==2.1.0",
-    "viv-utils[flirt]==0.6.9",
+    "viv-utils[flirt]==0.6.11",
     "halo==0.0.31",
     "networkx==2.5.1",
-    "ruamel.yaml==0.17.20",
-    "vivisect==1.0.5",
-    "smda==1.6.2",
+    "ruamel.yaml==0.17.21",
+    "vivisect==1.0.7",
+    "smda==1.7.0",
     "pefile==2021.9.3",
-    "typing==3.7.4.3",
-    "pyelftools==0.27",
+    "pyelftools==0.28",
 ]
 
 # this sets __version__
@@ -67,22 +66,25 @@ setuptools.setup(
     install_requires=requirements,
     extras_require={
         "dev": [
-            "pytest==6.2.5",
+            "pytest==7.0.1",
             "pytest-sugar==0.9.4",
             "pytest-instafail==0.4.2",
             "pytest-cov==3.0.0",
             "pycodestyle==2.8.0",
-            "black==21.12b0",
+            "black==22.1.0",
             "isort==5.10.1",
             "mypy==0.931",
             "psutil==5.9.0",
+            "stix2==3.0.1",
+            "requests==2.27.1",
             # type stubs for mypy
             "types-backports==0.1.3",
-            "types-colorama==0.4.6",
-            "types-PyYAML==6.0.3",
+            "types-colorama==0.4.8",
+            "types-PyYAML==6.0.4",
             "types-tabulate==0.8.5",
             "types-termcolor==1.1.3",
-            "types-psutil==5.8.19",
+            "types-psutil==5.8.20",
+            "types_requests==2.27.11",
         ],
     },
     zip_safe=False,

tests/fixtures.py

@@ -220,6 +220,10 @@ def get_data_path_by_name(name):
         return os.path.join(CD, "data", "3b13b6f1d7cd14dc4a097a12e2e505c0a4cff495262261e2bfc991df238b9b04.dll_")
     elif name == "7351f.elf":
         return os.path.join(CD, "data", "7351f8a40c5450557b24622417fc478d.elf_")
+    elif name.startswith("79abd"):
+        return os.path.join(CD, "data", "79abd17391adc6251ecdc58d13d76baf.dll_")
+    elif name.startswith("946a9"):
+        return os.path.join(CD, "data", "946a99f36a46d335dec080d9a4371940.dll_")
     else:
         raise ValueError("unexpected sample fixture: %s" % name)
 
@@ -269,6 +273,10 @@ def get_sample_md5_by_name(name):
         return "56a6ffe6a02941028cc8235204eef31d"
     elif name == "7351f.elf":
         return "7351f8a40c5450557b24622417fc478d"
+    elif name.startswith("79abd"):
+        return "79abd17391adc6251ecdc58d13d76baf"
+    elif name.startswith("946a9"):
+        return "946a99f36a46d335dec080d9a4371940.dll_"
     else:
         raise ValueError("unexpected sample fixture: %s" % name)
 
@@ -561,6 +569,8 @@ FEATURE_PRESENCE_TESTS = sorted(
         ("7351f.elf", "file", Arch(ARCH_AMD64), True),
         ("7351f.elf", "function=0x408753", capa.features.common.String("/dev/null"), True),
         ("7351f.elf", "function=0x408753,bb=0x408781", capa.features.insn.API("open"), True),
+        ("79abd...", "function=0x10002385,bb=0x10002385", capa.features.common.Characteristic("call $+5"), True),
+        ("946a9...", "function=0x10001510,bb=0x100015c0", capa.features.common.Characteristic("call $+5"), True),
     ],
     # order tests by (file, item)
     # so that our LRU cache is most effective.