Machofile analyzer (intelowlproject#3268)

* feature/analyzer : machofile analyzer

* fix/added unit tests, freeToUse, test_files.zip

* fix/cicd and copilot recommendations

* fix: precommit using ruff

* fix: precommit using ruff(again)

* fix: precommit using ruff(again)

* fix/cicd

* fix/cicd

* fix/cicd

* fix/migration-conflicts

* fix/migration-conflicts

* fix/cicd migrations

* fix/cicd migrations due to zip

* fix/add macho_sample

* fix/copilot and maintainer recommendations

* fix/cicd

* fix/zip-pass

* fix/cicd

* minor fixes : copilot recommendations

* fix/maintainer-feedback

* fix/cicd failure due to BlintAnalyzer

Author: IshaanXCoder

api_app/analyzers_manager/file_analyzers/blint_scan.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import shutil
+from unittest.mock import patch
 
 from blint.config import BlintOptions
 from blint.lib.runners import AnalysisRunner
@@ -28,10 +29,12 @@ def run(self) -> dict:
         set_permissions(reports_dir)
 
         analyzer = AnalysisRunner()
-        findings, reviews, fuzzables = analyzer.start(
-            blint_options=BlintOptions(reports_dir=reports_dir),
-            exe_files=[self.filepath],
-        )
+        # patch export_metadata to avoid RecursionError on Macho files and because we don't need the metadata report anyways
+        with patch("blint.lib.runners.export_metadata"):
+            findings, reviews, fuzzables = analyzer.start(
+                blint_options=BlintOptions(reports_dir=reports_dir),
+                exe_files=[self.filepath],
+            )
         response = {"findings": findings, "reviews": reviews, "fuzzables": fuzzables}
 
         shutil.rmtree(reports_dir)

api_app/analyzers_manager/file_analyzers/macho_info.py

@@ -0,0 +1,114 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+import logging
+from typing import Any, Dict
+
+import machofile
+
+from api_app.analyzers_manager.classes import FileAnalyzer
+from api_app.analyzers_manager.exceptions import AnalyzerRunException
+
+logger = logging.getLogger(__name__)
+
+
+def _safe_decode(value: Any) -> str:
+    """Helper to safely decode bytes to string."""
+    if isinstance(value, bytes):
+        return value.decode("utf-8", "ignore")
+    return str(value)
+
+
+class MachoInfo(FileAnalyzer):
+    """
+    Analyzer for Mach-O binary files (macOS/iOS executables).
+    Uses the machofile library to parse and extract information.
+    """
+
+    @classmethod
+    def update(cls) -> bool:
+        return False
+
+    def _parse_macho(self):
+        """Attempts to parse the file as Single or Universal Mach-O."""
+        try:
+            macho = machofile.MachO(self.filepath)
+            try:
+                macho.parse()
+            except AttributeError:
+                pass
+            return macho
+        except Exception as e:
+            try:
+                macho = machofile.UniversalMachO(self.filepath)
+                try:
+                    macho.parse()
+                except AttributeError:
+                    pass
+                return macho
+            except Exception as universal_error:
+                raise AnalyzerRunException(
+                    f"Failed to parse as both single and universal binary. "
+                    f"Single: {e}, Universal: {universal_error}"
+                )
+
+    def _extract_basic_info(self, macho, results: Dict[str, Any]):
+        """Extract basic information like headers and hashes."""
+        results["general_info"] = macho.get_general_info()
+        results["header"] = macho.get_macho_header()
+        results["hashes"] = macho.get_similarity_hashes()
+        results["code_signature"] = macho.code_signature_info
+        try:
+            results["architectures"] = macho.get_architectures()
+        except AttributeError:
+            results["architectures"] = []
+        results["uuid"] = macho.uuid
+        results["entrypoint"] = macho.entry_point
+        results["version_info"] = macho.version_info
+
+    def _extract_lists(self, macho, results: Dict[str, Any]):
+        """Extract list-based structures like load commands and segments."""
+        is_universal = isinstance(macho, machofile.UniversalMachO)
+
+        def get_macho_lists(m):
+            return {
+                "load_commands": [str(lc) for lc in m.load_commands],
+                "segments": [str(s) for s in m.segments],
+                "dylib_names": [_safe_decode(d) for d in m.dylib_names],
+            }
+
+        if is_universal:
+            for k in ["load_commands", "segments", "dylib_names"]:
+                results[k] = {}
+            for arch, m in macho.architectures.items():
+                sub_lists = get_macho_lists(m)
+                for k, v in sub_lists.items():
+                    results[k][arch] = v
+        else:
+            results.update(get_macho_lists(macho))
+
+    def _extract_symbols(self, macho, results: Dict[str, Any]):
+        """Extract imported and exported symbols."""
+        results["imports"] = macho.get_imported_functions()
+        results["exports"] = macho.get_exported_symbols()
+
+    def run(self) -> Dict[str, Any]:
+        results: Dict[str, Any] = {}
+
+        try:
+            macho = self._parse_macho()
+            self._extract_basic_info(macho, results)
+            self._extract_lists(macho, results)
+            self._extract_symbols(macho, results)
+
+        except Exception as e:
+            error_msg = (
+                f"job_id:{self.job_id} analyzer:{self.analyzer_name} "
+                f"md5:{self.md5} filename:{self.filename} "
+                f"MachoInfo parsing error: {e}"
+            )
+            logger.error(error_msg, exc_info=True)
+            self.report.errors.append(error_msg)
+            raise AnalyzerRunException(error_msg)
+
+        return results

api_app/analyzers_manager/migrations/0176_analyzer_config_macho_info.py

@@ -0,0 +1,130 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+from django.db import migrations
+from django.db.models.fields.related_descriptors import (
+    ForwardManyToOneDescriptor,
+    ForwardOneToOneDescriptor,
+    ManyToManyDescriptor,
+    ReverseManyToOneDescriptor,
+    ReverseOneToOneDescriptor,
+)
+
+plugin = {
+    "name": "MachoInfo",
+    "python_module": {
+        "module": "macho_info.MachoInfo",
+        "base_path": "api_app.analyzers_manager.file_analyzers",
+    },
+    "description": "Parse Mach-O binary files (macOS/iOS executables) using [machofile](https://github.com/pstirparo/machofile) library. Extracts headers, segments, dylibs, imports, exports, hashes and code signatures.",
+    "disabled": False,
+    "soft_time_limit": 60,
+    "routing_key": "local",
+    "health_check_status": True,
+    "type": "file",
+    "docker_based": False,
+    "maximum_tlp": "RED",
+    "observable_supported": [],
+    "supported_filetypes": [
+        "application/x-mach-binary",
+        "application/mac-binary",
+        "application/x-binary",
+        "application/x-executable",
+    ],
+    "run_hash": False,
+    "run_hash_type": "",
+    "not_supported_filetypes": [],
+    "health_check_task": None,
+    "model": "analyzers_manager.AnalyzerConfig",
+}
+
+params = []
+values = []
+
+
+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:
+        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):
+    dependencies = [
+        ("api_app", "0071_delete_last_elastic_report"),
+        ("analyzers_manager", "0175_analyzer_config_cleanbrowsing_malicious_detector"),
+    ]
+
+    operations = [migrations.RunPython(migrate, reverse_migrate)]
+    atomic = False

api_app/playbooks_manager/migrations/0063_add_machofile_to_free_to_use.py

@@ -0,0 +1,34 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+
+from django.db import migrations
+
+
+def migrate(apps, schema_editor):
+    playbook_config = apps.get_model("playbooks_manager", "PlaybookConfig")
+    AnalyzerConfig = apps.get_model("analyzers_manager", "AnalyzerConfig")
+    pc = playbook_config.objects.get(name="FREE_TO_USE_ANALYZERS")
+    pc.analyzers.add(AnalyzerConfig.objects.get(name="MachoInfo").id)
+    pc.full_clean()
+    pc.save()
+
+
+def reverse_migrate(apps, schema_editor):
+    playbook_config = apps.get_model("playbooks_manager", "PlaybookConfig")
+    AnalyzerConfig = apps.get_model("analyzers_manager", "AnalyzerConfig")
+    pc = playbook_config.objects.get(name="FREE_TO_USE_ANALYZERS")
+    pc.analyzers.remove(AnalyzerConfig.objects.get(name="MachoInfo").id)
+    pc.full_clean()
+    pc.save()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("playbooks_manager", "0062_add_cleanbrowsing_to_free_to_use"),
+        ("analyzers_manager", "0176_analyzer_config_macho_info"),
+    ]
+
+    operations = [
+        migrations.RunPython(migrate, reverse_migrate),
+    ]

api_app/playbooks_manager/migrations/0064_add_machoinfo_to_sample_static_analysis.py

@@ -0,0 +1,34 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+
+from django.db import migrations
+
+
+def migrate(apps, schema_editor):
+    playbook_config = apps.get_model("playbooks_manager", "PlaybookConfig")
+    AnalyzerConfig = apps.get_model("analyzers_manager", "AnalyzerConfig")
+    pc = playbook_config.objects.get(name="Sample_Static_Analysis")
+    pc.analyzers.add(AnalyzerConfig.objects.get(name="MachoInfo").id)
+    pc.full_clean()
+    pc.save()
+
+
+def reverse_migrate(apps, schema_editor):
+    playbook_config = apps.get_model("playbooks_manager", "PlaybookConfig")
+    AnalyzerConfig = apps.get_model("analyzers_manager", "AnalyzerConfig")
+    pc = playbook_config.objects.get(name="Sample_Static_Analysis")
+    pc.analyzers.remove(AnalyzerConfig.objects.get(name="MachoInfo").id)
+    pc.full_clean()
+    pc.save()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("playbooks_manager", "0063_add_machofile_to_free_to_use"),
+        ("analyzers_manager", "0176_analyzer_config_macho_info"),
+    ]
+
+    operations = [
+        migrations.RunPython(migrate, reverse_migrate),
+    ]

intel_owl/settings/_util.py

@@ -13,8 +13,13 @@
 
 get_secret = os.environ.get
 
-uid = pwd.getpwnam("www-data").pw_uid
-gid = grp.getgrnam("www-data").gr_gid
+try:
+    uid = pwd.getpwnam("www-data").pw_uid
+    gid = grp.getgrnam("www-data").gr_gid
+except (KeyError, AttributeError):
+    # fallback for environments without www-data (like local macOS or CI envs)
+    uid = os.getuid()
+    gid = os.getgid()
 
 
 def set_permissions(directory: Path, force_create: bool = False):

requirements/project-requirements.txt

@@ -96,6 +96,7 @@ jbxapi==3.23.0
 yara-x==1.10.0
 flare-floss==3.1.1
 flare-capa==9.3.1
+machofile==2026.2.4
 
 # httpx required for HTTP/2 support (Mullvad DNS rejects HTTP/1.1 with protocol errors)
 httpx[http2]==0.28.1