We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
CAPA rule path does not exist
# Expected Behavior

CAPA is able to load its rules.

# Current Behavior

CAPA is unable to load rules. Splits the file path into a list and loops through each character. Enumerates the / directory then moves to the next character o at which point it errors as the path does not exist:
OSError: rule path o does not exist or cannot be accessed

# Failure Information (for bugs)

The argument capa.main.RULES_PATH_DEFAULT_STRING passed to get_rules is a string, but should be a list of file paths:
https://github.com/kevoreilly/CAPEv2/blob/d9124712cabe5bf9a7a3a98da93cbdbd37a53da3/lib/cuckoo/common/integrations/capa.py#L44

## Steps to Reproduce

Enable CAPA in the config and run web server.

## Context

commit 1bd0bf62055fc3741ea19a85d510d54052dbf431
Ubuntu 20.04.04 LTS

## Failure Logs

  File "/opt/CAPEv2/web/../lib/cuckoo/common/integrations/parse_pe.py", line 74, in <module>
    from lib.cuckoo.common.integrations.capa import HAVE_FLARE_CAPA, flare_capa_details
  File "/opt/CAPEv2/web/../lib/cuckoo/common/integrations/capa.py", line 48, in <module>
    rules = capa.main.get_rules(capa.main.RULES_PATH_DEFAULT_STRING, disable_progress=True)
  File "/usr/local/lib/python3.8/dist-packages/capa/main.py", line 580, in get_rules
    raise IOError("rule path %s does not exist or cannot be accessed" % rule_path)
OSError: rule path o does not exist or cannot be accessed

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `lib/cuckoo/common/integrations/capa.py`
Content:
```
1 # Copyright (C) 2010-2015 Cuckoo Foundation.
2 # This file is part of Cuckoo Sandbox - http://www.cuckoosandbox.org
3 # See the file 'docs/LICENSE' for copying permission.
4 
5 import collections
6 import logging
7 import os
8 from typing import Any, Dict, List
9 
10 from lib.cuckoo.common.config import Config
11 from lib.cuckoo.common.constants import CUCKOO_ROOT
12 
13 log = logging.getLogger(__name__)
14 
15 processing_conf = Config("processing")
16 
17 """
18 from lib.cuckoo.common.integrations.capa import flare_capa_details, HAVE_FLARE_CAPA
19 path = "/opt/CAPEv2/storage/binaries/da034c11f0c396f6cd11d22f833f9501dc75a33047ba3bd5870ff79e479bc004"
20 details = flare_capa_details(path, "static", on_demand=True)
21 """
22 
23 HAVE_FLARE_CAPA = False
24 if processing_conf.flare_capa.enabled:
25     try:
26         from capa.version import __version__ as capa_version
27 
28         if capa_version[0] != "3":
29             print("FLARE-CAPA missed, pip3 install -U flare-capa")
30         else:
31             import capa.main
32             import capa.render.utils as rutils
33             import capa.rules
34             from capa.main import UnsupportedRuntimeError
35             from capa.render.result_document import (
36                 convert_capabilities_to_result_document as capa_convert_capabilities_to_result_document,
37             )
38             from capa.rules import InvalidRuleSet, InvalidRuleWithPath
39 
40             rules_path = os.path.join(CUCKOO_ROOT, "data", "capa-rules")
41             if os.path.exists(rules_path):
42                 capa.main.RULES_PATH_DEFAULT_STRING = os.path.join(CUCKOO_ROOT, "data", "capa-rules")
43                 try:
44                     rules = capa.main.get_rules(capa.main.RULES_PATH_DEFAULT_STRING, disable_progress=True)
45                     rules = capa.rules.RuleSet(rules)
46                     HAVE_FLARE_CAPA = True
47                 except InvalidRuleWithPath:
48                     print("FLARE_CAPA InvalidRuleWithPath")
49                     HAVE_FLARE_CAPA = False
50                 except InvalidRuleSet:
51                     print("FLARE_CAPA InvalidRuleSet")
52                     HAVE_FLARE_CAPA = False
53             else:
54                 print("FLARE CAPA rules missed! You can download them using python3 community.py -cr")
55                 HAVE_FLARE_CAPA = False
56 
57             signatures_path = os.path.join(CUCKOO_ROOT, "data", "capa-signatures")
58             if os.path.exists(signatures_path):
59                 capa.main.SIGNATURES_PATH_DEFAULT_STRING = os.path.join(CUCKOO_ROOT, "data", "capa-signatures")
60                 try:
61                     signatures = capa.main.get_signatures(capa.main.SIGNATURES_PATH_DEFAULT_STRING)
62                     HAVE_FLARE_CAPA = True
63                 except IOError:
64                     print("FLARE_CAPA InvalidSignatures")
65             else:
66                 print("FLARE CAPA rules missed! You can download them using python3 community.py -cr")
67                 HAVE_FLARE_CAPA = False
68     except ImportError as e:
69         HAVE_FLARE_CAPA = False
70         print(e)
71         print("FLARE-CAPA missed, pip3 install -U flare-capa")
72 
73 
74 def render_meta(doc: Dict[str, Any]) -> Dict[str, Any]:
75     return {
76         "md5": doc["meta"]["sample"]["md5"],
77         "sha1": doc["meta"]["sample"]["sha1"],
78         "sha256": doc["meta"]["sample"]["sha256"],
79         "path": doc["meta"]["sample"]["path"],
80     }
81 
82 
83 def find_subrule_matches(doc: Dict[str, Any]) -> set:
84     """
85     collect the rule names that have been matched as a subrule match.
86     this way we can avoid displaying entries for things that are too specific.
87     """
88 
89     def rec(node: dict) -> set:
90         matches = set()
91         if not node["success"]:
92             # there's probably a bug here for rules that do `not: match: ...`
93             # but we don't have any examples of this yet
94             return
95 
96         elif node["node"]["type"] == "statement":
97             for child in node["children"]:
98                 rec(child)
99 
100         elif node["node"]["type"] == "feature":
101             if node["node"]["feature"]["type"] == "match":
102                 matches.add(node["node"]["feature"]["match"])
103         return matches
104 
105     matches = set()
106 
107     for rule in rutils.capability_rules(doc):
108         for node in rule["matches"].values():
109             matches = matches.union(rec(node))
110 
111     return matches
112 
113 
114 def render_capabilities(doc: Dict[str, Any]) -> Dict[str, List[str]]:
115     """
116     example::
117         {'accept command line arguments': ['host-interaction/cli'],
118             'allocate thread local storage (2 matches)': ['host-interaction/process'],
119             'check for time delay via GetTickCount': ['anti-analysis/anti-debugging/debugger-detection'],
120             'check if process is running under wine': ['anti-analysis/anti-emulation/wine'],
121             'contain a resource (.rsrc) section': ['executable/pe/section/rsrc'],
122             'write file (3 matches)': ['host-interaction/file-system/write']
123         }
124     """
125     subrule_matches = find_subrule_matches(doc)
126 
127     capability_dict = {}
128     for rule in rutils.capability_rules(doc):
129         if rule["meta"]["name"] in subrule_matches:
130             # rules that are also matched by other rules should not get rendered by default.
131             # this cuts down on the amount of output while giving approx the same detail.
132             # see #224
133             continue
134 
135         count = len(rule["matches"])
136         if count == 1:
137             capability = rule["meta"]["name"]
138         else:
139             capability = f"{rule['meta']['name']} ({count} matches)"
140 
141         capability_dict.setdefault(rule["meta"]["namespace"], []).append(capability)
142     return capability_dict
143 
144 
145 def render_attack(doc: Dict[str, Any]) -> Dict[str, List[str]]:
146     """
147     example::
148         {'COLLECTION': ['Input Capture::Keylogging [T1056.001]'],
149             'DEFENSE EVASION': ['Obfuscated Files or Information [T1027]',
150                                 'Virtualization/Sandbox Evasion::System Checks '
151                                 '[T1497.001]'],
152             'DISCOVERY': ['File and Directory Discovery [T1083]',
153                           'Query Registry [T1012]',
154                           'System Information Discovery [T1082]'],
155             'EXECUTION': ['Shared Modules [T1129]']
156         }
157     """
158     attck_dict = {}
159     tactics = collections.defaultdict(set)
160     for rule in rutils.capability_rules(doc):
161         for attack in rule["meta"].get("att&ck", {}):
162             tactics[attack["tactic"]].add((attack["technique"], attack.get("subtechnique"), attack["id"]))
163 
164     for tactic, techniques in sorted(tactics.items()):
165         inner_rows = []
166         for technique, subtechnique, id in sorted(techniques):
167             if subtechnique is None:
168                 inner_rows.append(f"{technique} {id}")
169             else:
170                 inner_rows.append(f"{technique}::{subtechnique} {id}")
171         attck_dict.setdefault(tactic.upper(), inner_rows)
172     return attck_dict
173 
174 
175 def render_mbc(doc: Dict[str, Any]) -> Dict[str, List[str]]:
176     """
177     example::
178         {'ANTI-BEHAVIORAL ANALYSIS': ['Debugger Detection::Timing/Delay Check '
179                                       'GetTickCount [B0001.032]',
180                                       'Emulator Detection [B0004]',
181                                       'Virtual Machine Detection::Instruction '
182                                       'Testing [B0009.029]',
183                                       'Virtual Machine Detection [B0009]'],
184          'COLLECTION': ['Keylogging::Polling [F0002.002]'],
185          'CRYPTOGRAPHY': ['Encrypt Data::RC4 [C0027.009]',
186                           'Generate Pseudo-random Sequence::RC4 PRGA '
187                           '[C0021.004]']
188         }
189     """
190     mbc_dict = {}
191     objectives = collections.defaultdict(set)
192     for rule in rutils.capability_rules(doc):
193         for mbc in rule["meta"].get("mbc", {}):
194             objectives[mbc["objective"]].add((mbc["behavior"], mbc.get("method"), mbc["id"]))
195 
196     for objective, behaviors in sorted(objectives.items()):
197         inner_rows = []
198         for behavior, method, id in sorted(behaviors):
199             if method is None:
200                 inner_rows.append(f"{behavior} [{id}]")
201             else:
202                 inner_rows.append(f"{behavior}::{method} [{id}]")
203         mbc_dict.setdefault(objective.upper(), inner_rows)
204     return mbc_dict
205 
206 
207 def render_dictionary(doc: Dict[str, Any]) -> Dict[str, Any]:
208     ostream = render_meta(doc)
209     ostream["ATTCK"] = render_attack(doc)
210     ostream["MBC"] = render_mbc(doc)
211     ostream["CAPABILITY"] = render_capabilities(doc)
212 
213     return ostream
214 
215 
216 # ===== CAPA END
217 def flare_capa_details(file_path: str, category: str = False, on_demand=False, disable_progress=True) -> Dict[str, Any]:
218     capa_dictionary = {}
219     if (
220         HAVE_FLARE_CAPA
221         and processing_conf.flare_capa.enabled
222         and processing_conf.flare_capa.get(category, False)
223         and not processing_conf.flare_capa.on_demand
224         or on_demand
225     ):
226         try:
227             extractor = capa.main.get_extractor(
228                 file_path, "auto", capa.main.BACKEND_VIV, signatures, disable_progress=disable_progress
229             )
230             meta = capa.main.collect_metadata("", file_path, capa.main.RULES_PATH_DEFAULT_STRING, extractor)
231             capabilities, counts = capa.main.find_capabilities(rules, extractor, disable_progress=True)
232             meta["analysis"].update(counts)
233             doc = capa_convert_capabilities_to_result_document(meta, rules, capabilities)
234             capa_dictionary = render_dictionary(doc)
235         except MemoryError:
236             log.warning("FLARE CAPA -> MemoryError")
237         except UnsupportedRuntimeError:
238             log.error("FLARE CAPA -> UnsupportedRuntimeError")
239         except Exception as e:
240             log.error(e, exc_info=True)
241     return capa_dictionary
242
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/lib/cuckoo/common/integrations/capa.py b/lib/cuckoo/common/integrations/capa.py
--- a/lib/cuckoo/common/integrations/capa.py
+++ b/lib/cuckoo/common/integrations/capa.py
@@ -40,8 +40,10 @@
             rules_path = os.path.join(CUCKOO_ROOT, "data", "capa-rules")
             if os.path.exists(rules_path):
                 capa.main.RULES_PATH_DEFAULT_STRING = os.path.join(CUCKOO_ROOT, "data", "capa-rules")
+                rules_list = []
+                rules_list.append(rules_path)
                 try:
-                    rules = capa.main.get_rules(capa.main.RULES_PATH_DEFAULT_STRING, disable_progress=True)
+                    rules = capa.main.get_rules(rules_list, disable_progress=True)
                     rules = capa.rules.RuleSet(rules)
                     HAVE_FLARE_CAPA = True
                 except InvalidRuleWithPath:

{"golden_diff": "diff --git a/lib/cuckoo/common/integrations/capa.py b/lib/cuckoo/common/integrations/capa.py\n--- a/lib/cuckoo/common/integrations/capa.py\n+++ b/lib/cuckoo/common/integrations/capa.py\n@@ -40,8 +40,10 @@\n             rules_path = os.path.join(CUCKOO_ROOT, \"data\", \"capa-rules\")\n             if os.path.exists(rules_path):\n                 capa.main.RULES_PATH_DEFAULT_STRING = os.path.join(CUCKOO_ROOT, \"data\", \"capa-rules\")\n+                rules_list = []\n+                rules_list.append(rules_path)\n                 try:\n-                    rules = capa.main.get_rules(capa.main.RULES_PATH_DEFAULT_STRING, disable_progress=True)\n+                    rules = capa.main.get_rules(rules_list, disable_progress=True)\n                     rules = capa.rules.RuleSet(rules)\n                     HAVE_FLARE_CAPA = True\n                 except InvalidRuleWithPath:\n", "issue": "CAPA rule path does not exist\n# Expected Behavior\r\n\r\nCAPA is able to load its rules.\r\n\r\n# Current Behavior\r\n\r\nCAPA is unable to load rules. Splits the file path into a list and loops through each character. Enumerates the / directory then moves to the next character o at which point it errors as the path does not exist:\r\nOSError: rule path o does not exist or cannot be accessed\r\n\r\n# Failure Information (for bugs)\r\n\r\nThe argument capa.main.RULES_PATH_DEFAULT_STRING passed to get_rules is a string, but should be a list of file paths:\r\nhttps://github.com/kevoreilly/CAPEv2/blob/d9124712cabe5bf9a7a3a98da93cbdbd37a53da3/lib/cuckoo/common/integrations/capa.py#L44\r\n\r\n## Steps to Reproduce\r\n\r\nEnable CAPA in the config and run web server.\r\n\r\n## Context\r\n\r\ncommit 1bd0bf62055fc3741ea19a85d510d54052dbf431\r\nUbuntu 20.04.04 LTS\r\n\r\n## Failure Logs\r\n\r\n  File \"/opt/CAPEv2/web/../lib/cuckoo/common/integrations/parse_pe.py\", line 74, in <module>\r\n    from lib.cuckoo.common.integrations.capa import HAVE_FLARE_CAPA, flare_capa_details\r\n  File \"/opt/CAPEv2/web/../lib/cuckoo/common/integrations/capa.py\", line 48, in <module>\r\n    rules = capa.main.get_rules(capa.main.RULES_PATH_DEFAULT_STRING, disable_progress=True)\r\n  File \"/usr/local/lib/python3.8/dist-packages/capa/main.py\", line 580, in get_rules\r\n    raise IOError(\"rule path %s does not exist or cannot be accessed\" % rule_path)\r\nOSError: rule path o does not exist or cannot be accessed\n", "before_files": [{"content": "# Copyright (C) 2010-2015 Cuckoo Foundation.\n# This file is part of Cuckoo Sandbox - http://www.cuckoosandbox.org\n# See the file 'docs/LICENSE' for copying permission.\n\nimport collections\nimport logging\nimport os\nfrom typing import Any, Dict, List\n\nfrom lib.cuckoo.common.config import Config\nfrom lib.cuckoo.common.constants import CUCKOO_ROOT\n\nlog = logging.getLogger(__name__)\n\nprocessing_conf = Config(\"processing\")\n\n\"\"\"\nfrom lib.cuckoo.common.integrations.capa import flare_capa_details, HAVE_FLARE_CAPA\npath = \"/opt/CAPEv2/storage/binaries/da034c11f0c396f6cd11d22f833f9501dc75a33047ba3bd5870ff79e479bc004\"\ndetails = flare_capa_details(path, \"static\", on_demand=True)\n\"\"\"\n\nHAVE_FLARE_CAPA = False\nif processing_conf.flare_capa.enabled:\n    try:\n        from capa.version import __version__ as capa_version\n\n        if capa_version[0] != \"3\":\n            print(\"FLARE-CAPA missed, pip3 install -U flare-capa\")\n        else:\n            import capa.main\n            import capa.render.utils as rutils\n            import capa.rules\n            from capa.main import UnsupportedRuntimeError\n            from capa.render.result_document import (\n                convert_capabilities_to_result_document as capa_convert_capabilities_to_result_document,\n            )\n            from capa.rules import InvalidRuleSet, InvalidRuleWithPath\n\n            rules_path = os.path.join(CUCKOO_ROOT, \"data\", \"capa-rules\")\n            if os.path.exists(rules_path):\n                capa.main.RULES_PATH_DEFAULT_STRING = os.path.join(CUCKOO_ROOT, \"data\", \"capa-rules\")\n                try:\n                    rules = capa.main.get_rules(capa.main.RULES_PATH_DEFAULT_STRING, disable_progress=True)\n                    rules = capa.rules.RuleSet(rules)\n                    HAVE_FLARE_CAPA = True\n                except InvalidRuleWithPath:\n                    print(\"FLARE_CAPA InvalidRuleWithPath\")\n                    HAVE_FLARE_CAPA = False\n                except InvalidRuleSet:\n                    print(\"FLARE_CAPA InvalidRuleSet\")\n                    HAVE_FLARE_CAPA = False\n            else:\n                print(\"FLARE CAPA rules missed! You can download them using python3 community.py -cr\")\n                HAVE_FLARE_CAPA = False\n\n            signatures_path = os.path.join(CUCKOO_ROOT, \"data\", \"capa-signatures\")\n            if os.path.exists(signatures_path):\n                capa.main.SIGNATURES_PATH_DEFAULT_STRING = os.path.join(CUCKOO_ROOT, \"data\", \"capa-signatures\")\n                try:\n                    signatures = capa.main.get_signatures(capa.main.SIGNATURES_PATH_DEFAULT_STRING)\n                    HAVE_FLARE_CAPA = True\n                except IOError:\n                    print(\"FLARE_CAPA InvalidSignatures\")\n            else:\n                print(\"FLARE CAPA rules missed! You can download them using python3 community.py -cr\")\n                HAVE_FLARE_CAPA = False\n    except ImportError as e:\n        HAVE_FLARE_CAPA = False\n        print(e)\n        print(\"FLARE-CAPA missed, pip3 install -U flare-capa\")\n\n\ndef render_meta(doc: Dict[str, Any]) -> Dict[str, Any]:\n    return {\n        \"md5\": doc[\"meta\"][\"sample\"][\"md5\"],\n        \"sha1\": doc[\"meta\"][\"sample\"][\"sha1\"],\n        \"sha256\": doc[\"meta\"][\"sample\"][\"sha256\"],\n        \"path\": doc[\"meta\"][\"sample\"][\"path\"],\n    }\n\n\ndef find_subrule_matches(doc: Dict[str, Any]) -> set:\n    \"\"\"\n    collect the rule names that have been matched as a subrule match.\n    this way we can avoid displaying entries for things that are too specific.\n    \"\"\"\n\n    def rec(node: dict) -> set:\n        matches = set()\n        if not node[\"success\"]:\n            # there's probably a bug here for rules that do `not: match: ...`\n            # but we don't have any examples of this yet\n            return\n\n        elif node[\"node\"][\"type\"] == \"statement\":\n            for child in node[\"children\"]:\n                rec(child)\n\n        elif node[\"node\"][\"type\"] == \"feature\":\n            if node[\"node\"][\"feature\"][\"type\"] == \"match\":\n                matches.add(node[\"node\"][\"feature\"][\"match\"])\n        return matches\n\n    matches = set()\n\n    for rule in rutils.capability_rules(doc):\n        for node in rule[\"matches\"].values():\n            matches = matches.union(rec(node))\n\n    return matches\n\n\ndef render_capabilities(doc: Dict[str, Any]) -> Dict[str, List[str]]:\n    \"\"\"\n    example::\n        {'accept command line arguments': ['host-interaction/cli'],\n            'allocate thread local storage (2 matches)': ['host-interaction/process'],\n            'check for time delay via GetTickCount': ['anti-analysis/anti-debugging/debugger-detection'],\n            'check if process is running under wine': ['anti-analysis/anti-emulation/wine'],\n            'contain a resource (.rsrc) section': ['executable/pe/section/rsrc'],\n            'write file (3 matches)': ['host-interaction/file-system/write']\n        }\n    \"\"\"\n    subrule_matches = find_subrule_matches(doc)\n\n    capability_dict = {}\n    for rule in rutils.capability_rules(doc):\n        if rule[\"meta\"][\"name\"] in subrule_matches:\n            # rules that are also matched by other rules should not get rendered by default.\n            # this cuts down on the amount of output while giving approx the same detail.\n            # see #224\n            continue\n\n        count = len(rule[\"matches\"])\n        if count == 1:\n            capability = rule[\"meta\"][\"name\"]\n        else:\n            capability = f\"{rule['meta']['name']} ({count} matches)\"\n\n        capability_dict.setdefault(rule[\"meta\"][\"namespace\"], []).append(capability)\n    return capability_dict\n\n\ndef render_attack(doc: Dict[str, Any]) -> Dict[str, List[str]]:\n    \"\"\"\n    example::\n        {'COLLECTION': ['Input Capture::Keylogging [T1056.001]'],\n            'DEFENSE EVASION': ['Obfuscated Files or Information [T1027]',\n                                'Virtualization/Sandbox Evasion::System Checks '\n                                '[T1497.001]'],\n            'DISCOVERY': ['File and Directory Discovery [T1083]',\n                          'Query Registry [T1012]',\n                          'System Information Discovery [T1082]'],\n            'EXECUTION': ['Shared Modules [T1129]']\n        }\n    \"\"\"\n    attck_dict = {}\n    tactics = collections.defaultdict(set)\n    for rule in rutils.capability_rules(doc):\n        for attack in rule[\"meta\"].get(\"att&ck\", {}):\n            tactics[attack[\"tactic\"]].add((attack[\"technique\"], attack.get(\"subtechnique\"), attack[\"id\"]))\n\n    for tactic, techniques in sorted(tactics.items()):\n        inner_rows = []\n        for technique, subtechnique, id in sorted(techniques):\n            if subtechnique is None:\n                inner_rows.append(f\"{technique} {id}\")\n            else:\n                inner_rows.append(f\"{technique}::{subtechnique} {id}\")\n        attck_dict.setdefault(tactic.upper(), inner_rows)\n    return attck_dict\n\n\ndef render_mbc(doc: Dict[str, Any]) -> Dict[str, List[str]]:\n    \"\"\"\n    example::\n        {'ANTI-BEHAVIORAL ANALYSIS': ['Debugger Detection::Timing/Delay Check '\n                                      'GetTickCount [B0001.032]',\n                                      'Emulator Detection [B0004]',\n                                      'Virtual Machine Detection::Instruction '\n                                      'Testing [B0009.029]',\n                                      'Virtual Machine Detection [B0009]'],\n         'COLLECTION': ['Keylogging::Polling [F0002.002]'],\n         'CRYPTOGRAPHY': ['Encrypt Data::RC4 [C0027.009]',\n                          'Generate Pseudo-random Sequence::RC4 PRGA '\n                          '[C0021.004]']\n        }\n    \"\"\"\n    mbc_dict = {}\n    objectives = collections.defaultdict(set)\n    for rule in rutils.capability_rules(doc):\n        for mbc in rule[\"meta\"].get(\"mbc\", {}):\n            objectives[mbc[\"objective\"]].add((mbc[\"behavior\"], mbc.get(\"method\"), mbc[\"id\"]))\n\n    for objective, behaviors in sorted(objectives.items()):\n        inner_rows = []\n        for behavior, method, id in sorted(behaviors):\n            if method is None:\n                inner_rows.append(f\"{behavior} [{id}]\")\n            else:\n                inner_rows.append(f\"{behavior}::{method} [{id}]\")\n        mbc_dict.setdefault(objective.upper(), inner_rows)\n    return mbc_dict\n\n\ndef render_dictionary(doc: Dict[str, Any]) -> Dict[str, Any]:\n    ostream = render_meta(doc)\n    ostream[\"ATTCK\"] = render_attack(doc)\n    ostream[\"MBC\"] = render_mbc(doc)\n    ostream[\"CAPABILITY\"] = render_capabilities(doc)\n\n    return ostream\n\n\n# ===== CAPA END\ndef flare_capa_details(file_path: str, category: str = False, on_demand=False, disable_progress=True) -> Dict[str, Any]:\n    capa_dictionary = {}\n    if (\n        HAVE_FLARE_CAPA\n        and processing_conf.flare_capa.enabled\n        and processing_conf.flare_capa.get(category, False)\n        and not processing_conf.flare_capa.on_demand\n        or on_demand\n    ):\n        try:\n            extractor = capa.main.get_extractor(\n                file_path, \"auto\", capa.main.BACKEND_VIV, signatures, disable_progress=disable_progress\n            )\n            meta = capa.main.collect_metadata(\"\", file_path, capa.main.RULES_PATH_DEFAULT_STRING, extractor)\n            capabilities, counts = capa.main.find_capabilities(rules, extractor, disable_progress=True)\n            meta[\"analysis\"].update(counts)\n            doc = capa_convert_capabilities_to_result_document(meta, rules, capabilities)\n            capa_dictionary = render_dictionary(doc)\n        except MemoryError:\n            log.warning(\"FLARE CAPA -> MemoryError\")\n        except UnsupportedRuntimeError:\n            log.error(\"FLARE CAPA -> UnsupportedRuntimeError\")\n        except Exception as e:\n            log.error(e, exc_info=True)\n    return capa_dictionary\n", "path": "lib/cuckoo/common/integrations/capa.py"}], "after_files": [{"content": "# Copyright (C) 2010-2015 Cuckoo Foundation.\n# This file is part of Cuckoo Sandbox - http://www.cuckoosandbox.org\n# See the file 'docs/LICENSE' for copying permission.\n\nimport collections\nimport logging\nimport os\nfrom typing import Any, Dict, List\n\nfrom lib.cuckoo.common.config import Config\nfrom lib.cuckoo.common.constants import CUCKOO_ROOT\n\nlog = logging.getLogger(__name__)\n\nprocessing_conf = Config(\"processing\")\n\n\"\"\"\nfrom lib.cuckoo.common.integrations.capa import flare_capa_details, HAVE_FLARE_CAPA\npath = \"/opt/CAPEv2/storage/binaries/da034c11f0c396f6cd11d22f833f9501dc75a33047ba3bd5870ff79e479bc004\"\ndetails = flare_capa_details(path, \"static\", on_demand=True)\n\"\"\"\n\nHAVE_FLARE_CAPA = False\nif processing_conf.flare_capa.enabled:\n    try:\n        from capa.version import __version__ as capa_version\n\n        if capa_version[0] != \"3\":\n            print(\"FLARE-CAPA missed, pip3 install -U flare-capa\")\n        else:\n            import capa.main\n            import capa.render.utils as rutils\n            import capa.rules\n            from capa.main import UnsupportedRuntimeError\n            from capa.render.result_document import (\n                convert_capabilities_to_result_document as capa_convert_capabilities_to_result_document,\n            )\n            from capa.rules import InvalidRuleSet, InvalidRuleWithPath\n\n            rules_path = os.path.join(CUCKOO_ROOT, \"data\", \"capa-rules\")\n            if os.path.exists(rules_path):\n                capa.main.RULES_PATH_DEFAULT_STRING = os.path.join(CUCKOO_ROOT, \"data\", \"capa-rules\")\n                rules_list = []\n                rules_list.append(rules_path)\n                try:\n                    rules = capa.main.get_rules(rules_list, disable_progress=True)\n                    rules = capa.rules.RuleSet(rules)\n                    HAVE_FLARE_CAPA = True\n                except InvalidRuleWithPath:\n                    print(\"FLARE_CAPA InvalidRuleWithPath\")\n                    HAVE_FLARE_CAPA = False\n                except InvalidRuleSet:\n                    print(\"FLARE_CAPA InvalidRuleSet\")\n                    HAVE_FLARE_CAPA = False\n            else:\n                print(\"FLARE CAPA rules missed! You can download them using python3 community.py -cr\")\n                HAVE_FLARE_CAPA = False\n\n            signatures_path = os.path.join(CUCKOO_ROOT, \"data\", \"capa-signatures\")\n            if os.path.exists(signatures_path):\n                capa.main.SIGNATURES_PATH_DEFAULT_STRING = os.path.join(CUCKOO_ROOT, \"data\", \"capa-signatures\")\n                try:\n                    signatures = capa.main.get_signatures(capa.main.SIGNATURES_PATH_DEFAULT_STRING)\n                    HAVE_FLARE_CAPA = True\n                except IOError:\n                    print(\"FLARE_CAPA InvalidSignatures\")\n            else:\n                print(\"FLARE CAPA rules missed! You can download them using python3 community.py -cr\")\n                HAVE_FLARE_CAPA = False\n    except ImportError as e:\n        HAVE_FLARE_CAPA = False\n        print(e)\n        print(\"FLARE-CAPA missed, pip3 install -U flare-capa\")\n\n\ndef render_meta(doc: Dict[str, Any]) -> Dict[str, Any]:\n    return {\n        \"md5\": doc[\"meta\"][\"sample\"][\"md5\"],\n        \"sha1\": doc[\"meta\"][\"sample\"][\"sha1\"],\n        \"sha256\": doc[\"meta\"][\"sample\"][\"sha256\"],\n        \"path\": doc[\"meta\"][\"sample\"][\"path\"],\n    }\n\n\ndef find_subrule_matches(doc: Dict[str, Any]) -> set:\n    \"\"\"\n    collect the rule names that have been matched as a subrule match.\n    this way we can avoid displaying entries for things that are too specific.\n    \"\"\"\n\n    def rec(node: dict) -> set:\n        matches = set()\n        if not node[\"success\"]:\n            # there's probably a bug here for rules that do `not: match: ...`\n            # but we don't have any examples of this yet\n            return\n\n        elif node[\"node\"][\"type\"] == \"statement\":\n            for child in node[\"children\"]:\n                rec(child)\n\n        elif node[\"node\"][\"type\"] == \"feature\":\n            if node[\"node\"][\"feature\"][\"type\"] == \"match\":\n                matches.add(node[\"node\"][\"feature\"][\"match\"])\n        return matches\n\n    matches = set()\n\n    for rule in rutils.capability_rules(doc):\n        for node in rule[\"matches\"].values():\n            matches = matches.union(rec(node))\n\n    return matches\n\n\ndef render_capabilities(doc: Dict[str, Any]) -> Dict[str, List[str]]:\n    \"\"\"\n    example::\n        {'accept command line arguments': ['host-interaction/cli'],\n            'allocate thread local storage (2 matches)': ['host-interaction/process'],\n            'check for time delay via GetTickCount': ['anti-analysis/anti-debugging/debugger-detection'],\n            'check if process is running under wine': ['anti-analysis/anti-emulation/wine'],\n            'contain a resource (.rsrc) section': ['executable/pe/section/rsrc'],\n            'write file (3 matches)': ['host-interaction/file-system/write']\n        }\n    \"\"\"\n    subrule_matches = find_subrule_matches(doc)\n\n    capability_dict = {}\n    for rule in rutils.capability_rules(doc):\n        if rule[\"meta\"][\"name\"] in subrule_matches:\n            # rules that are also matched by other rules should not get rendered by default.\n            # this cuts down on the amount of output while giving approx the same detail.\n            # see #224\n            continue\n\n        count = len(rule[\"matches\"])\n        if count == 1:\n            capability = rule[\"meta\"][\"name\"]\n        else:\n            capability = f\"{rule['meta']['name']} ({count} matches)\"\n\n        capability_dict.setdefault(rule[\"meta\"][\"namespace\"], []).append(capability)\n    return capability_dict\n\n\ndef render_attack(doc: Dict[str, Any]) -> Dict[str, List[str]]:\n    \"\"\"\n    example::\n        {'COLLECTION': ['Input Capture::Keylogging [T1056.001]'],\n            'DEFENSE EVASION': ['Obfuscated Files or Information [T1027]',\n                                'Virtualization/Sandbox Evasion::System Checks '\n                                '[T1497.001]'],\n            'DISCOVERY': ['File and Directory Discovery [T1083]',\n                          'Query Registry [T1012]',\n                          'System Information Discovery [T1082]'],\n            'EXECUTION': ['Shared Modules [T1129]']\n        }\n    \"\"\"\n    attck_dict = {}\n    tactics = collections.defaultdict(set)\n    for rule in rutils.capability_rules(doc):\n        for attack in rule[\"meta\"].get(\"att&ck\", {}):\n            tactics[attack[\"tactic\"]].add((attack[\"technique\"], attack.get(\"subtechnique\"), attack[\"id\"]))\n\n    for tactic, techniques in sorted(tactics.items()):\n        inner_rows = []\n        for technique, subtechnique, id in sorted(techniques):\n            if subtechnique is None:\n                inner_rows.append(f\"{technique} {id}\")\n            else:\n                inner_rows.append(f\"{technique}::{subtechnique} {id}\")\n        attck_dict.setdefault(tactic.upper(), inner_rows)\n    return attck_dict\n\n\ndef render_mbc(doc: Dict[str, Any]) -> Dict[str, List[str]]:\n    \"\"\"\n    example::\n        {'ANTI-BEHAVIORAL ANALYSIS': ['Debugger Detection::Timing/Delay Check '\n                                      'GetTickCount [B0001.032]',\n                                      'Emulator Detection [B0004]',\n                                      'Virtual Machine Detection::Instruction '\n                                      'Testing [B0009.029]',\n                                      'Virtual Machine Detection [B0009]'],\n         'COLLECTION': ['Keylogging::Polling [F0002.002]'],\n         'CRYPTOGRAPHY': ['Encrypt Data::RC4 [C0027.009]',\n                          'Generate Pseudo-random Sequence::RC4 PRGA '\n                          '[C0021.004]']\n        }\n    \"\"\"\n    mbc_dict = {}\n    objectives = collections.defaultdict(set)\n    for rule in rutils.capability_rules(doc):\n        for mbc in rule[\"meta\"].get(\"mbc\", {}):\n            objectives[mbc[\"objective\"]].add((mbc[\"behavior\"], mbc.get(\"method\"), mbc[\"id\"]))\n\n    for objective, behaviors in sorted(objectives.items()):\n        inner_rows = []\n        for behavior, method, id in sorted(behaviors):\n            if method is None:\n                inner_rows.append(f\"{behavior} [{id}]\")\n            else:\n                inner_rows.append(f\"{behavior}::{method} [{id}]\")\n        mbc_dict.setdefault(objective.upper(), inner_rows)\n    return mbc_dict\n\n\ndef render_dictionary(doc: Dict[str, Any]) -> Dict[str, Any]:\n    ostream = render_meta(doc)\n    ostream[\"ATTCK\"] = render_attack(doc)\n    ostream[\"MBC\"] = render_mbc(doc)\n    ostream[\"CAPABILITY\"] = render_capabilities(doc)\n\n    return ostream\n\n\n# ===== CAPA END\ndef flare_capa_details(file_path: str, category: str = False, on_demand=False, disable_progress=True) -> Dict[str, Any]:\n    capa_dictionary = {}\n    if (\n        HAVE_FLARE_CAPA\n        and processing_conf.flare_capa.enabled\n        and processing_conf.flare_capa.get(category, False)\n        and not processing_conf.flare_capa.on_demand\n        or on_demand\n    ):\n        try:\n            extractor = capa.main.get_extractor(\n                file_path, \"auto\", capa.main.BACKEND_VIV, signatures, disable_progress=disable_progress\n            )\n            meta = capa.main.collect_metadata(\"\", file_path, capa.main.RULES_PATH_DEFAULT_STRING, extractor)\n            capabilities, counts = capa.main.find_capabilities(rules, extractor, disable_progress=True)\n            meta[\"analysis\"].update(counts)\n            doc = capa_convert_capabilities_to_result_document(meta, rules, capabilities)\n            capa_dictionary = render_dictionary(doc)\n        except MemoryError:\n            log.warning(\"FLARE CAPA -> MemoryError\")\n        except UnsupportedRuntimeError:\n            log.error(\"FLARE CAPA -> UnsupportedRuntimeError\")\n        except Exception as e:\n            log.error(e, exc_info=True)\n    return capa_dictionary\n", "path": "lib/cuckoo/common/integrations/capa.py"}]}