Refactor yara_detected for better matching and self-extraction (#2759)

* Refactor yara_detected for better matching and self-extraction

Refactor yara_detected method to improve matching logic and encapsulate self-extraction processing.

* Apply suggestions from code review

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* Update lib/cuckoo/common/abstracts.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: doomedraven

lib/cuckoo/common/abstracts.py

@@ -887,82 +887,89 @@ def set_path(self, analysis_path):
             CuckooReportError(e)
 
     def yara_detected(self, name):
-        target = self.results.get("target", {})
-        if target.get("category") in ("file", "static") and target.get("file"):
+        name_pattern = re.compile(name, re.I)
+
+        def _check_matches(data_block, path, label_override=None):
+            if not isinstance(data_block, dict):
+                return
+
             for keyword in ("cape_yara", "yara"):
-                for yara_block in self.results["target"]["file"].get(keyword, []):
-                    if re.findall(name, yara_block["name"], re.I):
-                        yield "sample", self.results["target"]["file"]["path"], yara_block, self.results["target"]["file"]
-
-            if target["file"].get("selfextract"):
-                for _, toolsblock in target["file"]["selfextract"].items():
-                    for block in toolsblock.get("extracted_files", []):
-                        for keyword in ("cape_yara", "yara"):
-                            for yara_block in block[keyword]:
-                                if re.findall(name, yara_block["name"], re.I):
-                                    # we can't use here values from set_path
-                                    yield "sample", block["path"], yara_block, block
-
-        for block in self.results.get("CAPE", {}).get("payloads", []) or []:
-            for sub_keyword in ("cape_yara", "yara"):
-                for yara_block in block.get(sub_keyword, []):
-                    if re.findall(name, yara_block["name"], re.I):
-                        yield sub_keyword, block["path"], yara_block, block
-
-            if block.get("selfextract", {}):
-                for _, toolsblock in block["selfextract"].items():
-                    for subblock in toolsblock.get("extracted_files", []):
-                        for keyword in ("cape_yara", "yara"):
-                            for yara_block in subblock[keyword]:
-                                if re.findall(name, yara_block["name"], re.I):
-                                    yield "sample", subblock["path"], yara_block, block
-
-        for keyword in ("procdump", "procmemory", "extracted", "dropped"):
-            if self.results.get(keyword) is not None:
-                for block in self.results.get(keyword, []):
-                    if not isinstance(block, dict):
-                        continue
-                    for sub_keyword in ("cape_yara", "yara"):
-                        for yara_block in block.get(sub_keyword, []):
-                            if re.findall(name, yara_block["name"], re.I):
-                                path = block["path"] if block.get("path", False) else ""
-                                yield keyword, path, yara_block, block
-
-                    if keyword == "procmemory":
-                        for pe in block.get("extracted_pe", []) or []:
-                            for sub_keyword in ("cape_yara", "yara"):
-                                for yara_block in pe.get(sub_keyword, []) or []:
-                                    if re.findall(name, yara_block["name"], re.I):
-                                        yield "extracted_pe", pe["path"], yara_block, block
-
-                    if block.get("selfextract", {}):
-                        for _, toolsblock in block["selfextract"].items():
-                            for subblock in toolsblock.get("extracted_files", []):
-                                for keyword in ("cape_yara", "yara"):
-                                    for yara_block in subblock[keyword]:
-                                        if re.findall(name, yara_block["name"], re.I):
-                                            yield "sample", subblock["path"], yara_block, block
-
-        macro_path = os.path.join(CUCKOO_ROOT, "storage", "analyses", str(self.results["info"]["id"]), "macros")
-        for macroname in self.results.get("static", {}).get("office", {}).get("Macro", {}).get("info", []) or []:
-            for yara_block in self.results["static"]["office"]["Macro"]["info"].get("macroname", []) or []:
-                for sub_block in self.results["static"]["office"]["Macro"]["info"]["macroname"].get(yara_block, []) or []:
-                    if re.findall(name, sub_block["name"], re.I):
-                        yield (
-                            "macro",
-                            os.path.join(macro_path, macroname),
-                            sub_block,
-                            self.results["static"]["office"]["Macro"]["info"],
-                        )
-
-        if self.results.get("static", {}).get("office", {}).get("XLMMacroDeobfuscator", False):
-            for yara_block in self.results["static"]["office"]["XLMMacroDeobfuscator"].get("info", []).get("yara_macro", []) or []:
-                if re.findall(name, yara_block["name"], re.I):
+                for yara_block in data_block.get(keyword, []):
+                    if name_pattern.search(yara_block.get("name", "")):
+                        label = label_override if label_override else keyword
+                        yield label, path, yara_block, data_block
+
+        def _process_selfextract(parent_block):
+            selfextract = parent_block.get("selfextract")
+            if not selfextract:
+                return
+
+            tools_iter = selfextract.values() if isinstance(selfextract, dict) else []
+
+            for toolsblock in tools_iter:
+                for extracted_file in toolsblock.get("extracted_files", []) or []:
+                    yield from _check_matches(
+                        extracted_file,
+                        path=extracted_file.get("path"),
+                        label_override="sample"
+                    )
+
+        results = self.results
+        target = results.get("target", {})
+
+        # 1. Procesar Target
+        if target.get("category") in ("file", "static") and target.get("file"):
+            file_info = target["file"]
+            yield from _check_matches(file_info, file_info.get("path"), label_override="sample")
+            yield from _process_selfextract(file_info)
+
+        cape_payloads = results.get("CAPE", {}).get("payloads", []) or []
+        for block in cape_payloads:
+            yield from _check_matches(block, block.get("path"))
+            yield from _process_selfextract(block)
+
+        search_keys = ("procdump", "procmemory", "extracted", "dropped")
+        for keyword in search_keys:
+            blocks = results.get(keyword, []) or []
+            if not blocks:
+                continue
+
+            for block in blocks:
+                if not isinstance(block, dict):
+                    continue
+
+                path = block.get("path", "")
+                yield from _check_matches(block, path, label_override=keyword)
+
+                if keyword == "procmemory":
+                    for pe in block.get("extracted_pe", []) or []:
+                        yield from _check_matches(pe, pe.get("path"), label_override="extracted_pe")
+
+                yield from _process_selfextract(block)
+
+        # ToDo not sure if static still exist
+        office_info = results.get("static", {}).get("office", {})
+        macro_info = office_info.get("Macro", {}).get("info", [])
+        analysis_id = str(results.get("info", {}).get("id", "unknown"))
+        macro_base_path = os.path.join(CUCKOO_ROOT, "storage", "analyses", analysis_id, "macros")
+
+        if macro_info:
+            if isinstance(macro_info, list):
+                for item in macro_info:
+                    yield from _check_matches(item, os.path.join(macro_base_path, item.get("name", "macro")), label_override="macro")
+            elif isinstance(macro_info, dict):
+                for macroname, macro_data in macro_info.items():
+                    yield from _check_matches(macro_data, os.path.join(macro_base_path, macroname), label_override="macro")
+
+        xlm_info = office_info.get("XLMMacroDeobfuscator", {}).get("info", {})
+        if xlm_info:
+            for yara_block in xlm_info.get("yara_macro", []) or []:
+                if name_pattern.search(yara_block.get("name", "")):
                     yield (
                         "macro",
-                        os.path.join(macro_path, "xlm_macro"),
+                        os.path.join(macro_base_path, "xlm_macro"),
                         yara_block,
-                        self.results["static"]["office"]["XLMMacroDeobfuscator"]["info"],
+                        xlm_info
                     )
 
     def signature_matched(self, signame: str) -> bool: