Fix ScanCapa, Add Tests, Add Elf

build/python/backend/Dockerfile

@@ -43,6 +43,7 @@ RUN apt-get -q update && \
     redis-server \
     tesseract-ocr \
     unrar \
+    unzip \
     upx \
     jq && \
 # Download and compile Archive library, needed for exiftool to work best
@@ -61,15 +62,16 @@ RUN apt-get -q update && \
     perl Makefile.PL && \
     make && \
     make install && \
-# Install FireEye CAPA
-#   - Binary installation, not supported as Python 3 plugin
-#   - Requires binary to be executable
-#   - Vivisect dependency requires available /.viv/ folder.
-    cd /tmp/ && \
-    curl -OL https://github.com/fireeye/capa/releases/download/v$CAPA_VERSION/capa-linux && \
-    chmod +x /tmp/capa-linux && \
-    mkdir /.viv/ && \
-    chmod -R a+rw /.viv && \
+# Install FireEye CAPA rules and signatures
+    mkdir -p /etc/capa/rules/ && \
+    curl -OL https://github.com/mandiant/capa-rules/archive/refs/tags/v$CAPA_VERSION.zip && \
+    unzip v$CAPA_VERSION.zip -d /etc/capa/rules/ && \
+    rm v$CAPA_VERSION.zip && \
+    mkdir -p /etc/capa/signatures/ && \
+    cd /etc/capa/signatures/ && \
+    curl -OL https://github.com/mandiant/capa/raw/master/sigs/1_flare_msvc_rtf_32_64.sig && \
+    curl -OL https://github.com/mandiant/capa/raw/master/sigs/2_flare_msvc_atlmfc_32_64.sig && \
+    curl -OL https://github.com/mandiant/capa/raw/master/sigs/3_flare_common_libs.sig && \
 # Install FireEye FLOSS
 #   - Binary installation, not supported as Python 3 plugin
 #   - Requires binary to be executable

build/python/backend/requirements.txt

@@ -8,6 +8,7 @@ cryptography==3.4.7
 docker==5.0.0
 eml-parser>=1.17
 esprima==4.0.1
+flare-capa==4.0.1
 formulas==1.2.2
 git+https://github.com/jshlbrd/python-entropy.git   # v0.11 as of this freeze (package installed as 'entropy')
 grpcio-tools==1.42.0
@@ -25,11 +26,11 @@ oletools==0.56.1
 opencv-contrib-python==4.6.0.66
 opencv-python==4.6.0.66
 openpyxl==3.0.9
-pefile==2019.4.18
+pefile
 pgpdump3==1.5.2
 py-tlsh==4.7.2
 pycdlib==1.13.0
-pyelftools==0.27
+pyelftools==0.28
 pygments==2.9.0
 pylzma==0.5.0
 pytesseract==0.3.7

configs/python/backend/backend.yaml

@@ -56,10 +56,13 @@ scanners:
 #        flavors:
 #          - 'application/x-dosexec'
 #          - 'mz_file'
+#          - 'application/x-sharedlib'
+#          - 'elf_file'
 #      priority: 5
 #      options:
 #        tmp_directory: '/dev/shm/'
-#        location: '/etc/capa/'
+#        location_rules: '/etc/capa/rules/'
+#        location_signatures: '/etc/capa/signatures/'
 #  'ScanCcn':
 #    - positive:
 #        flavors:

misc/kubernetes/backend-configmap.yaml

@@ -42,15 +42,18 @@ data:
               - 'application/x-bzip2'
               - 'bzip2_file'
           priority: 5
-      #  'ScanCapa':
-      #    - positive:
-      #        flavors:
-      #          - 'application/x-dosexec'
-      #          - 'mz_file'
-      #      priority: 5
-      #      options:
-      #        tmp_directory: '/dev/shm/'
-      #        location: '/etc/capa/'
+#        'ScanCapa':
+#          - positive:
+#              flavors:
+#                - 'application/x-dosexec'
+#                - 'mz_file'
+#                - 'application/x-sharedlib'
+#                - 'elf_file'
+#            priority: 5
+#            options:
+#              tmp_directory: '/dev/shm/'
+#              location_rules: '/etc/capa/rules/'
+#              location_signatures: '/etc/capa/signatures/'
       'ScanDocx':
         - positive:
             flavors:

src/python/requirements.txt

@@ -8,6 +8,7 @@ cryptography==3.4.7
 docker==5.0.0
 eml-parser>=1.17
 esprima==4.0.1
+flare-capa==4.0.1
 formulas==1.2.2
 git+https://github.com/jshlbrd/python-entropy.git   # v0.11 as of this freeze (package installed as 'entropy')
 grpcio-tools==1.42.0
@@ -25,11 +26,11 @@ oletools==0.56.1
 opencv-contrib-python==4.6.0.66
 opencv-python==4.6.0.66
 openpyxl==3.0.9
-pefile==2019.4.18
+pefile
 pgpdump3==1.5.2
 py-tlsh==4.7.2
 pycdlib==1.13.0
-pyelftools==0.27
+pyelftools==0.28
 pygments==2.9.0
 pylzma==0.5.0
 pytesseract==0.3.7

src/python/strelka/scanners/scan_capa.py

@@ -1,4 +1,3 @@
-import re
 import os
 import json
 import subprocess
@@ -19,58 +18,67 @@ class ScanCapa(strelka.Scanner):
 
     def scan(self, data, file, options, expire_at):
         tmp_directory = options.get('tmp_directory', '/tmp/')
-        location = options.get('location', '/etc/capa/')
+        location_rules = options.get('location_rules', '/etc/capa/rules/')
+        location_signatures = options.get('location_signatures', '/etc/capa/signatures/')
 
-        # Only run if rules file exists, otherwise return no rules error
-        if len(os.listdir(location)) != 0:
-            try:
-                with tempfile.NamedTemporaryFile(dir=tmp_directory) as tmp_data:
-                    tmp_data.write(data)
-                    tmp_data.flush()
+        # Check rules and signatures locationss
+        if os.path.isdir(location_rules):
+            if not os.listdir(location_rules):
+                self.flags.append('error_norules')
+                return
+        else:
+            self.flags.append('error_norules')
+            return
 
-                    try:
-                        (stdout, stderr) = subprocess.Popen(
-                            ['/tmp/capa-linux', tmp_data.name, '-r', location, '-j'],
+        if os.path.isdir(location_signatures):
+            if not os.listdir(location_signatures):
+                self.flags.append('error_nosignatures')
+                return
+        else:
+            self.flags.append('error_nosignatures')
+            return
+
+        try:
+            with tempfile.NamedTemporaryFile(dir=tmp_directory) as tmp_data:
+                tmp_data.write(data)
+                tmp_data.flush()
+
+                try:
+                    (stdout, stderr) = subprocess.Popen(
+                            ['capa', '-j', '-r', location_rules, '-s', location_signatures, tmp_data.name],
                             stdout=subprocess.PIPE,
                             stderr=subprocess.DEVNULL
                         ).communicate()
+                except:
+                    self.flags.append('error_processing')
+                    return
+
+                if stdout:
+                    try:
+                        capa_json = json.loads(stdout.rstrip())
                     except:
-                        self.flags.append('error_processing')
+                        self.flags.append('error_parsing')
                         return
 
-                    if stdout:
-                        # Observed extraneous data in stdout requiring string trimming. Parse out JSON response.
-                        # This can be fixed when CAPA is aviailable as a Python 3 library.
-                        try:
-                            stdout = stdout[stdout.find(b'{'):]
-                            stdout = stdout[:stdout.rfind(b'}')]
-                            stdout += b'}'
-                            capa_json = json.loads(stdout)
-                        except:
-                            self.flags.append('error_parsing')
-                            return
-
-                        try:
-                            # Sets are used to remove duplicative values
-                            self.event['matches'] = set()
-                            self.event['mitre_techniques'] = set()
-                            self.event['mitre_ids'] = set()
-
-                            for k, v in capa_json['rules'].items():
-                                self.event['matches'].add(k)
-                                if 'att&ck' in v['meta']:
-                                    result = re.search(r'^([^:]+)::([^\[)]+)\s\[([^\]]+)\]', v['meta']['att&ck'][0])
-                                    self.event['mitre_techniques'].add(result.group(2))
-                                    self.event['mitre_ids'].add(result.group(3))
-                            # For consistency, convert sets to list
-                            self.event['matches'] = list(self.event['matches'])
-                            self.event['mitre_techniques'] = list(self.event['mitre_techniques'])
-                            self.event['mitre_ids'] = list(self.event['mitre_ids'])
-                        except:
-                            self.flags.append('error_collection')
-            except:
-                self.flags.append('error_execution')
-        else:
-            self.flags.append('error_norules')
-
+                    try:
+                        # Sets are used to remove duplicative values
+                        self.event['matches'] = []
+                        self.event['mitre_techniques'] = []
+                        self.event['mitre_ids'] = []
 
+                        for rule_key, rule_value in capa_json['rules'].items():
+                            self.event['matches'].append(rule_key)
+                            if 'attack' in rule_value.get('meta', []):
+                                if attacks := rule_value.get('meta', []).get('attack', []):
+                                    for attack in attacks:
+                                        self.event['mitre_techniques'].append(
+                                            "::".join(attack.get("parts", [])))
+                                        self.event['mitre_ids'].append(attack.get("id", ""))
+                        # For consistency, convert sets to list
+                        self.event['matches'] = list(set(self.event['matches']))
+                        self.event['mitre_techniques'] = list(set(self.event['mitre_techniques']))
+                        self.event['mitre_ids'] = list(set(self.event['mitre_ids']))
+                    except:
+                        self.flags.append('error_collection')
+        except:
+            self.flags.append('error_execution')

src/python/strelka/scanners/scan_dmg.py

@@ -79,7 +79,6 @@ def get_all_items(root, exclude=None):
                         if not name.is_file():
                             continue
 
-                        print(name)
                         # Skip duplicate files created with these extended attributes
                         if str(name).endswith(":com.apple.quarantine") or str(name).endswith(":com.apple.FinderInfo"):
                             continue

src/python/strelka/tests/test_scan_capa.py

@@ -0,0 +1,89 @@
+from pathlib import Path
+from pytest_unordered import unordered
+from unittest import TestCase, mock
+
+from strelka.scanners.scan_capa import ScanCapa as ScanUnderTest
+from strelka.tests import run_test_scan
+
+
+def test_scan_capa_dotnet(mocker):
+    """
+    Pass: Sample event matches output of scanner.
+    Failure: Unable to load file or sample event fails to match.
+    """
+
+    test_scan_event = {
+        "elapsed": mock.ANY,
+        "flags": [],
+        "matches": unordered(["contains PDB path", "compiled to the .NET platform"]),
+        "mitre_ids": [],
+        "mitre_techniques": [],
+    }
+
+    scanner_event = run_test_scan(
+        mocker=mocker,
+        scan_class=ScanUnderTest,
+        fixture_path=Path(__file__).parent / "fixtures/test.exe",
+    )
+
+    TestCase.maxDiff = None
+    TestCase().assertDictEqual(test_scan_event, scanner_event)
+
+
+def test_scan_capa_elf(mocker):
+    """
+    Pass: Sample event matches output of scanner.
+    Failure: Unable to load file or sample event fails to match.
+    """
+
+    test_scan_event = {
+        "elapsed": mock.ANY,
+        "flags": [],
+        "matches": [],
+        "mitre_ids": [],
+        "mitre_techniques": [],
+    }
+
+    scanner_event = run_test_scan(
+        mocker=mocker,
+        scan_class=ScanUnderTest,
+        fixture_path=Path(__file__).parent / "fixtures/test.elf",
+    )
+
+    TestCase.maxDiff = None
+    TestCase().assertDictEqual(test_scan_event, scanner_event)
+
+
+def test_scan_capa_pe_xor(mocker):
+    """
+    Pass: Sample event matches output of scanner.
+    Failure: Unable to load file or sample event fails to match.
+    """
+
+    test_scan_event = {
+        "elapsed": mock.ANY,
+        "flags": [],
+        "matches": unordered([
+            "encode data using XOR",
+            "contains PDB path",
+            "contain a resource (.rsrc) section",
+            "parse PE header",
+            "contain loop",
+        ]),
+        "mitre_ids": unordered(["T1129", "T1027"]),
+        "mitre_techniques": unordered(
+            [
+                "Execution::Shared Modules",
+                "Defense Evasion::Obfuscated Files or Information",
+            ]
+        ),
+    }
+
+    scanner_event = run_test_scan(
+        mocker=mocker,
+        scan_class=ScanUnderTest,
+        fixture_path=Path(__file__).parent / "fixtures/test_xor.exe",
+    )
+
+    TestCase.maxDiff = None
+    TestCase().assertDictEqual(test_scan_event, scanner_event)