YaraX Analyzer with Yara-Forge Rule Repository integration (#2980)

* added yarax analyzer

* dumped migration and minor changes to analyzer

* tracking rules latest version and update analyzer config

* created new mixin for helper methods for managing rules

* refactored yarax analyzer to helper methods from RulesUtilityMixin

---------

Co-authored-by: Akshit Maheshwary <[email protected]>

Author: spoiicy

api_app/analyzers_manager/file_analyzers/yarax.py

@@ -0,0 +1,160 @@
+import logging
+import os
+import pathlib
+
+import requests
+import yara_x
+from django.conf import settings
+
+from api_app.analyzers_manager.classes import FileAnalyzer
+from api_app.analyzers_manager.exceptions import AnalyzerRunException
+from api_app.mixins import RulesUtiliyMixin
+
+logger = logging.getLogger(__name__)
+
+
+RULES_URL = "https://api.github.com/repos/YARAHQ/yara-forge/releases/latest"
+BASE_RULES_LOCATION = f"{settings.MEDIA_ROOT}/yarax"
+
+
+class YaraX(FileAnalyzer, RulesUtiliyMixin):
+    rule_set: str = "core"
+
+    def get_rule_location(self):
+        logger.info(f"Searching for rules at {BASE_RULES_LOCATION}/{self.rule_set}")
+        try:
+            rule_set_dir = pathlib.Path(BASE_RULES_LOCATION) / self.rule_set
+            rule = rule_set_dir.rglob("*.yar")
+            return next(rule).__str__()
+
+        except StopIteration as e:
+            logger.error(f"{self.rule_set} not found, function exited with error {e}")
+            raise AnalyzerRunException(f"{self.rule_set} rules not present")
+
+    @classmethod
+    def update(cls, rule_set, analyzer_module) -> bool:
+        logger.info(f"Updating {rule_set} rule set")
+        rule_set_download_url = ""
+        filename = ""
+        try:
+            response = requests.get(RULES_URL)
+            assets = response.json()["assets"]
+            latest_version = response.json()["tag_name"]
+            for asset in assets:
+                if rule_set in asset["browser_download_url"]:
+                    rule_set_download_url = asset["browser_download_url"]
+                    filename = asset["name"]
+                    break
+
+            rule_set_directory = f"{BASE_RULES_LOCATION}/{rule_set}"
+            rule_file_path = f"{rule_set_directory}/{filename}"
+
+            cls._download_rules(
+                rule_set_download_url,
+                rule_set_directory,
+                rule_file_path,
+                latest_version,
+                analyzer_module,
+            )
+
+            rule_file_path = pathlib.Path(BASE_RULES_LOCATION) / rule_set / filename
+
+            cls._unzip(rule_file_path)
+
+            logger.info(f"Successfully updated {rule_set} rules")
+            return True
+
+        except Exception as e:
+            logger.exception(f"Failed to update yara-forge rules. Error: {e}")
+            raise AnalyzerRunException(
+                f"Failed to update yara-forge ruleset. Error: {e}"
+            )
+
+        return False
+
+    def run(self):
+
+        if self.rule_set not in ("core", "extended", "full"):
+            raise AnalyzerRunException(
+                "Please select the correct ruleset pack from available options."
+                " Available options are core, extended, full"
+            )
+
+        rule_dir = f"{BASE_RULES_LOCATION}/{self.rule_set}"
+
+        response = requests.get(RULES_URL)
+
+        latest_version = response.json()["tag_name"]
+
+        update_status = (
+            True
+            if self._check_if_latest_version(latest_version, self.python_module)
+            else self.update(self.rule_set, self.python_module)
+        )
+
+        if not os.path.isdir(rule_dir) and not update_status:
+            raise AnalyzerRunException(f"Couldn't update {self.rule_set} rules")
+
+        rules_file_path = self.get_rule_location()
+        logger.info(f"Found rules at {rules_file_path}")
+
+        with open(rules_file_path, mode="r") as f:
+            rules_source = f.read()
+
+        try:
+            logger.info(f"Compiling rules present at {self.rule_set}")
+            compiler = yara_x.Compiler()
+            compiler.add_source(rules_source, origin=rules_file_path)
+            rules = compiler.build()
+            logger.info("Successfully compiled and built rules")
+
+            logger.info(
+                f"Starting scanning file: {self.filename} having hash: {self.md5} with {self.rule_set} rules"
+            )
+            scanner = yara_x.Scanner(rules)
+
+            result = []
+            scan_results = scanner.scan_file(self.filepath)
+            for rule in scan_results.matching_rules:
+                logger.info(f"Rule Identifier: {rule.identifier}")
+                rule_metadata = dict(rule.metadata)
+                rule_details = {
+                    "rule_identifier": rule.identifier,
+                    "rule_metadata": rule_metadata,
+                    "pattern_details": [],
+                }
+                for pattern in rule.patterns:
+                    pattern_details = {
+                        "pattern_identifier": pattern.identifier,
+                        "match_details": [],
+                    }
+
+                    for match in pattern.matches:
+                        match_details = {
+                            "match_offset": match.offset,
+                            "match_length": match.length,
+                            "match_xor_key": match.xor_key,
+                        }
+                        pattern_details["match_details"].append(match_details)
+
+                    rule_details["pattern_details"].append(pattern_details)
+
+                result.append(rule_details)
+
+            logger.info(f"Successfully scanned {self.filename} with hash {self.md5}")
+
+            return {"results": "No Match"} if not result else {"results": result}
+
+        except yara_x.CompileError as e:
+            logger.error(
+                f"Failed to compile {self.rule_set} rules present at {rules_file_path} with error {e}"
+            )
+            raise AnalyzerRunException(f"Failed to compile {self.rule_set} rules")
+
+        except yara_x.ScanError as e:
+            logger.error(f"Failed to scan file {self.filename} with error {e}")
+            raise AnalyzerRunException(f"Failed to scan {self.filename}")
+
+        except yara_x.TimeoutError as e:
+            logger.error(f"Failed with timeout with error {e}")
+            raise AnalyzerRunException("Failed with timeout")

api_app/analyzers_manager/migrations/0166_analyzer_config_yarax.py

@@ -0,0 +1,160 @@
+from django.db import migrations
+from django.db.models.fields.related_descriptors import (
+    ForwardManyToOneDescriptor,
+    ForwardOneToOneDescriptor,
+    ManyToManyDescriptor,
+    ReverseManyToOneDescriptor,
+    ReverseOneToOneDescriptor,
+)
+
+plugin = {
+    "python_module": {
+        "health_check_schedule": None,
+        "update_schedule": None,
+        "module": "yarax.YaraX",
+        "base_path": "api_app.analyzers_manager.file_analyzers",
+    },
+    "name": "YaraX",
+    "description": "[YaraX](https://virustotal.github.io/yara-x/docs/intro/getting-started/) is a re-incarnation of YARA, a pattern matching tool designed with malware researchers in mind. This new incarnation intends to be faster, safer and more user-friendly than its predecessor.",
+    "disabled": False,
+    "soft_time_limit": 480,
+    "routing_key": "default",
+    "health_check_status": True,
+    "type": "file",
+    "docker_based": False,
+    "maximum_tlp": "RED",
+    "observable_supported": [],
+    "supported_filetypes": [],
+    "run_hash": False,
+    "run_hash_type": "",
+    "not_supported_filetypes": ["application/vnd.tcpdump.pcap"],
+    "mapping_data_model": {},
+    "model": "analyzers_manager.AnalyzerConfig",
+}
+
+params = [
+    {
+        "python_module": {
+            "module": "yarax.YaraX",
+            "base_path": "api_app.analyzers_manager.file_analyzers",
+        },
+        "name": "rule_set",
+        "type": "str",
+        "description": "Yara-Forge Ruleset pack to use. Available options are core, extended, full. By default, core ruleset pack is selected for analysis. Refer the docs for more information.",
+        "is_secret": False,
+        "required": False,
+    }
+]
+
+values = [
+    {
+        "parameter": {
+            "python_module": {
+                "module": "yarax.YaraX",
+                "base_path": "api_app.analyzers_manager.file_analyzers",
+            },
+            "name": "rule_set",
+            "type": "str",
+            "description": "Yara-Forge Ruleset pack to use. Available options are core, extended, full. By default, core ruleset pack is selected for analysis. Refer the docs for more information.",
+            "is_secret": False,
+            "required": False,
+        },
+        "analyzer_config": "YaraX",
+        "connector_config": None,
+        "visualizer_config": None,
+        "ingestor_config": None,
+        "pivot_config": None,
+        "for_organization": False,
+        "value": "core",
+        "updated_at": "2025-09-07T11:44:16.043896Z",
+        "owner": None,
+    }
+]
+
+
+def _get_real_obj(Model, field, value):
+    def _get_obj(Model, other_model, value):
+        if isinstance(value, dict):
+            real_vals = {}
+            for key, real_val in value.items():
+                real_vals[key] = _get_real_obj(other_model, key, real_val)
+            value = other_model.objects.get_or_create(**real_vals)[0]
+        # it is just the primary key serialized
+        else:
+            if isinstance(value, int):
+                if Model.__name__ == "PluginConfig":
+                    value = other_model.objects.get(name=plugin["name"])
+                else:
+                    value = other_model.objects.get(pk=value)
+            else:
+                value = other_model.objects.get(name=value)
+        return value
+
+    if (
+        type(getattr(Model, field))
+        in [
+            ForwardManyToOneDescriptor,
+            ReverseManyToOneDescriptor,
+            ReverseOneToOneDescriptor,
+            ForwardOneToOneDescriptor,
+        ]
+        and value
+    ):
+        other_model = getattr(Model, field).get_queryset().model
+        value = _get_obj(Model, other_model, value)
+    elif type(getattr(Model, field)) in [ManyToManyDescriptor] and value:
+        other_model = getattr(Model, field).rel.model
+        value = [_get_obj(Model, other_model, val) for val in value]
+    return value
+
+
+def _create_object(Model, data):
+    mtm, no_mtm = {}, {}
+    for field, value in data.items():
+        value = _get_real_obj(Model, field, value)
+        if type(getattr(Model, field)) is ManyToManyDescriptor:
+            mtm[field] = value
+        else:
+            no_mtm[field] = value
+    try:
+        o = Model.objects.get(**no_mtm)
+    except Model.DoesNotExist:
+        o = Model(**no_mtm)
+        o.full_clean()
+        o.save()
+        for field, value in mtm.items():
+            attribute = getattr(o, field)
+            if value is not None:
+                attribute.set(value)
+        return False
+    return True
+
+
+def migrate(apps, schema_editor):
+    Parameter = apps.get_model("api_app", "Parameter")
+    PluginConfig = apps.get_model("api_app", "PluginConfig")
+    python_path = plugin.pop("model")
+    Model = apps.get_model(*python_path.split("."))
+    if not Model.objects.filter(name=plugin["name"]).exists():
+        exists = _create_object(Model, plugin)
+        if not exists:
+            for param in params:
+                _create_object(Parameter, param)
+            for value in values:
+                _create_object(PluginConfig, value)
+
+
+def reverse_migrate(apps, schema_editor):
+    python_path = plugin.pop("model")
+    Model = apps.get_model(*python_path.split("."))
+    Model.objects.get(name=plugin["name"]).delete()
+
+
+class Migration(migrations.Migration):
+    atomic = False
+    dependencies = [
+        ("api_app", "0071_delete_last_elastic_report"),
+        ("analyzers_manager", "0165_analyzer_config_joesandboxurl"),
+    ]
+
+    operations = [migrations.RunPython(migrate, reverse_migrate)]

api_app/analyzers_manager/migrations/0167_analyzerrulesfileversion.py

@@ -0,0 +1,43 @@
+# Generated by Django 4.2.17 on 2025-09-07 12:15
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("api_app", "0071_delete_last_elastic_report"),
+        ("analyzers_manager", "0166_analyzer_config_yarax"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="AnalyzerRulesFileVersion",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "last_downloaded_version",
+                    models.CharField(blank=True, default="", max_length=50),
+                ),
+                ("download_url", models.URLField(blank=True, default="")),
+                ("downloaded_at", models.DateTimeField(auto_now_add=True)),
+                (
+                    "python_module",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.PROTECT,
+                        related_name="rules_version",
+                        to="api_app.pythonmodule",
+                    ),
+                ),
+            ],
+        ),
+    ]

api_app/analyzers_manager/models.py

@@ -350,3 +350,13 @@ def plugin_type(cls) -> str:
     @property
     def config_exception(cls):
         return AnalyzerConfigurationException
+
+
+class AnalyzerRulesFileVersion(models.Model):
+    last_downloaded_version = models.CharField(max_length=50, blank=True, default="")
+    download_url = models.URLField(max_length=200, blank=True, default="")
+    downloaded_at = models.DateTimeField(auto_now_add=True)
+
+    python_module = models.ForeignKey(
+        PythonModule, on_delete=models.PROTECT, related_name="rules_version"
+    )