feat(heuristics): add SimilarProjectAnalyzer to detect structural similarity across packages from same maintainer (#1089)

AmineRaouane · web-flow · commit 4daa7845af43 · 2025-08-08T11:55:15.000+10:00
This PR adds a new heuristic analyzer called SimilarProjectAnalyzer. It checks whether a PyPI package has a similar file/folder structure to other packages maintained by the same user. This helps in identifying potentially malicious packages that replicate existing structures.

Signed-off-by: Amine &lt;amine.raouane@enim.ac.ma&gt;
diff --git a/src/macaron/malware_analyzer/pypi_heuristics/heuristics.py b/src/macaron/malware_analyzer/pypi_heuristics/heuristics.py
@@ -46,6 +46,9 @@ class Heuristics(str, Enum):
     #: Indicates that the package maintainer's email address is suspicious or invalid.
     FAKE_EMAIL = "fake_email"
 
+    #: Indicates that the package has a similar structure to other packages maintained by the same user.
+    SIMILAR_PROJECTS = "similar_projects"
+
 
 class HeuristicResult(str, Enum):
     """Result type indicating the outcome of a heuristic."""
diff --git a/src/macaron/malware_analyzer/pypi_heuristics/metadata/similar_projects.py b/src/macaron/malware_analyzer/pypi_heuristics/metadata/similar_projects.py
@@ -0,0 +1,147 @@
+# Copyright (c) 2024 - 2025, Oracle and/or its affiliates. All rights reserved.
+# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/.
+
+"""This analyzer checks if the package has a similar structure to other packages maintained by the same user."""
+
+import hashlib
+import io
+import logging
+import tarfile
+
+from macaron.json_tools import JsonType
+from macaron.malware_analyzer.pypi_heuristics.base_analyzer import BaseHeuristicAnalyzer
+from macaron.malware_analyzer.pypi_heuristics.heuristics import HeuristicResult, Heuristics
+from macaron.slsa_analyzer.package_registry.pypi_registry import PyPIPackageJsonAsset
+from macaron.util import send_get_http, send_get_http_raw
+
+logger: logging.Logger = logging.getLogger(__name__)
+
+
+class SimilarProjectAnalyzer(BaseHeuristicAnalyzer):
+    """Check whether the package has a similar structure to other packages maintained by the same user."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            name="similar_project_analyzer",
+            heuristic=Heuristics.SIMILAR_PROJECTS,
+            depends_on=None,
+        )
+
+    def analyze(self, pypi_package_json: PyPIPackageJsonAsset) -> tuple[HeuristicResult, dict[str, JsonType]]:
+        """Analyze the package.
+
+        Parameters
+        ----------
+        pypi_package_json: PyPIPackageJsonAsset
+            The PyPI package JSON asset object.
+
+        Returns
+        -------
+        tuple[HeuristicResult, dict[str, JsonType]]:
+            The result and related information collected during the analysis.
+
+        Raises
+        ------
+        HeuristicAnalyzerValueError
+            if the analysis fails.
+        """
+        package_name = pypi_package_json.component_name
+        target_hash = self.get_structure_hash(package_name)
+        if not target_hash:
+            return HeuristicResult.SKIP, {}
+
+        maintainers = pypi_package_json.pypi_registry.get_maintainers_of_package(package_name)
+        if maintainers:
+            for maintainer in maintainers:
+                maintainer_packages = pypi_package_json.pypi_registry.get_packages_by_username(maintainer)
+                if not maintainer_packages:
+                    continue
+                for package in maintainer_packages:
+                    if package == package_name:
+                        continue
+
+                    hash_value = self.get_structure_hash(package)
+                    if target_hash == hash_value:
+                        return HeuristicResult.FAIL, {
+                            "message": f"The package {package_name} has a similar structure to {package}.",
+                            "similar_package": package,
+                        }
+
+        return HeuristicResult.PASS, {}
+
+    def get_url(self, package_name: str, package_type: str = "sdist") -> str | None:
+        """Get the URL of the package's sdist.
+
+        Parameters
+        ----------
+        package_name : str
+            The name of the package.
+        package_type: str
+            The package type to retrieve the URL of.
+
+        Returns
+        -------
+        str | None:
+            The URL of the package's sdist or None if not found.
+        """
+        json_url = f"https://pypi.org/pypi/{package_name}/json"
+        data = send_get_http(json_url, headers={})
+        if not data:
+            logger.debug("Failed to fetch package data for %s.", package_name)
+            return None
+
+        sdist = next((url for url in data["urls"] if url["packagetype"] == package_type and url.get("url")), None)
+        return sdist["url"] if sdist else None
+
+    def get_structure(self, package_name: str) -> list[str]:
+        """Get the file structure of the package's sdist.
+
+        Parameters
+        ----------
+        package_name : str
+            The name of the package.
+
+        Returns
+        -------
+        list[str]:
+            The list of files in the package's sdist.
+        """
+        sdist_url = self.get_url(package_name)
+        if not sdist_url:
+            logger.debug("Package %s does not have a sdist.", package_name)
+            return []
+
+        response = send_get_http_raw(sdist_url)
+        if not response:
+            logger.debug("Failed to download sdist for package %s.", package_name)
+            return []
+
+        buffer = io.BytesIO(response.content)
+        with tarfile.open(fileobj=buffer, mode="r:gz") as tf:
+            members = [
+                member.name for member in tf.getmembers() if member.name and not member.name.startswith("PAXHeaders/")
+            ]
+
+        return members
+
+    def get_structure_hash(self, package_name: str) -> str:
+        """Get the hash of the package's file structure.
+
+        Parameters
+        ----------
+        package_name : str
+            The name of the package.
+
+        Returns
+        -------
+        str:
+            The hash of the package's file structure.
+        """
+        structure = self.get_structure(package_name)
+        if not structure:
+            return ""
+
+        normalized = sorted([p.replace(package_name, "<ROOT>") for p in structure])
+
+        joined = "\n".join(normalized).encode("utf-8")
+        return hashlib.sha256(joined).hexdigest()
diff --git a/src/macaron/slsa_analyzer/checks/detect_malicious_metadata_check.py b/src/macaron/slsa_analyzer/checks/detect_malicious_metadata_check.py
@@ -23,6 +23,7 @@
 from macaron.malware_analyzer.pypi_heuristics.metadata.fake_email import FakeEmailAnalyzer
 from macaron.malware_analyzer.pypi_heuristics.metadata.high_release_frequency import HighReleaseFrequencyAnalyzer
 from macaron.malware_analyzer.pypi_heuristics.metadata.one_release import OneReleaseAnalyzer
+from macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects import SimilarProjectAnalyzer
 from macaron.malware_analyzer.pypi_heuristics.metadata.source_code_repo import SourceCodeRepoAnalyzer
 from macaron.malware_analyzer.pypi_heuristics.metadata.typosquatting_presence import TyposquattingPresenceAnalyzer
 from macaron.malware_analyzer.pypi_heuristics.metadata.unchanged_release import UnchangedReleaseAnalyzer
@@ -364,6 +365,7 @@ def run_check(self, ctx: AnalyzeContext) -> CheckResultData:
         AnomalousVersionAnalyzer,
         TyposquattingPresenceAnalyzer,
         FakeEmailAnalyzer,
+        SimilarProjectAnalyzer,
     ]
 
     # name used to query the result of all problog rules, so it can be accessed outside the model.
@@ -434,7 +436,9 @@ def run_check(self, ctx: AnalyzeContext) -> CheckResultData:
     % Package released recently with the a maintainer email address that is not valid.
     {Confidence.MEDIUM.value}::trigger(malware_medium_confidence_3) :-
         quickUndetailed,
-        failed({Heuristics.FAKE_EMAIL.value}).
+        failed({Heuristics.FAKE_EMAIL.value}),
+        failed({Heuristics.SIMILAR_PROJECTS.value}).
+
     % ----- Evaluation -----
 
     % Aggregate result
diff --git a/src/macaron/slsa_analyzer/package_registry/pypi_registry.py b/src/macaron/slsa_analyzer/package_registry/pypi_registry.py
@@ -369,6 +369,27 @@ def get_maintainer_profile_page(self, username: str) -> str | None:
             return html_snippets
         return None
 
+    def get_packages_by_username(self, username: str) -> list[str] | None:
+        """Implement custom API to get the maintainer's packages.
+
+        Parameters
+        ----------
+        username: str
+            The maintainer's username.
+
+        Returns
+        -------
+            list[str]: A list of package names.
+        """
+        user_page: str | None = self.get_maintainer_profile_page(username)
+        if user_page is None:
+            return None
+
+        soup = BeautifulSoup(user_page, "html.parser")
+        headers = soup.find_all("h3", class_="package-snippet__title")
+        packages = list({header.get_text(strip=True) for header in headers})
+        return packages
+
     def get_maintainer_join_date(self, username: str) -> datetime | None:
         """Implement custom API to get the maintainer's join date.
 
diff --git a/tests/malware_analyzer/pypi/test_similar_projects.py b/tests/malware_analyzer/pypi/test_similar_projects.py
@@ -0,0 +1,134 @@
+# Copyright (c) 2024 - 2025, Oracle and/or its affiliates. All rights reserved.
+# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/.
+
+"""Tests for the SimilarProjectAnalyzer heuristic."""
+# pylint: disable=redefined-outer-name
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from macaron.malware_analyzer.pypi_heuristics.heuristics import HeuristicResult
+from macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects import SimilarProjectAnalyzer
+
+
+@pytest.fixture()
+def analyzer() -> SimilarProjectAnalyzer:
+    """Pytest fixture to create a SimilarProjectAnalyzer instance."""
+    analyzer_instance = SimilarProjectAnalyzer()
+    return analyzer_instance
+
+
+def test_analyze_skip_no_target_hash(analyzer: SimilarProjectAnalyzer, pypi_package_json: MagicMock) -> None:
+    """Test the analyzer skips when the target package has no structure hash."""
+    pypi_package_json.component_name = "test_package"
+    with patch(
+        "macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects.SimilarProjectAnalyzer.get_structure_hash"
+    ) as mock_get_structure_hash:
+        mock_get_structure_hash.return_value = ""
+        result, info = analyzer.analyze(pypi_package_json)
+    assert result == HeuristicResult.SKIP
+    assert info == {}
+
+
+def test_analyze_fail_similar_project_found(analyzer: SimilarProjectAnalyzer, pypi_package_json: MagicMock) -> None:
+    """Test the analyzer fails when a similar project with the same structure hash is found."""
+    pypi_package_json.component_name = "test_package"
+    pypi_package_json.pypi_registry.get_maintainers_of_package.return_value = ["user1"]
+    pypi_package_json.pypi_registry.get_packages_by_username.return_value = ["similar_package"]
+
+    with patch(
+        "macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects.SimilarProjectAnalyzer.get_structure_hash"
+    ) as mock_get_structure_hash:
+        mock_get_structure_hash.return_value = "same_hash"
+        result, info = analyzer.analyze(pypi_package_json)
+
+    assert result == HeuristicResult.FAIL
+    assert info["similar_package"] == "similar_package"
+
+
+def test_analyze_fail_when_package_compares_to_itself(
+    analyzer: SimilarProjectAnalyzer, pypi_package_json: MagicMock
+) -> None:
+    """Test the analyzer passes when a package is compared against itself."""
+    pypi_package_json.component_name = "test_package"
+    pypi_package_json.pypi_registry.get_maintainers_of_package.return_value = ["user1"]
+    pypi_package_json.pypi_registry.get_packages_by_username.return_value = ["test_package", "other_package"]
+
+    with patch(
+        "macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects.SimilarProjectAnalyzer.get_structure_hash"
+    ) as mock_get_structure_hash:
+        mock_get_structure_hash.side_effect = ["hash1", "hash2"]  # test_package, other_package
+        result, _ = analyzer.analyze(pypi_package_json)
+
+    assert result == HeuristicResult.PASS
+
+
+def test_analyze_pass_no_similar_hash(analyzer: SimilarProjectAnalyzer, pypi_package_json: MagicMock) -> None:
+    """Test the analyzer passes when no similar project has the same structure hash."""
+    pypi_package_json.component_name = "test_package"
+    pypi_package_json.pypi_registry.get_maintainers_of_package.return_value = ["user1"]
+    pypi_package_json.pypi_registry.get_packages_by_username.return_value = ["other_package"]
+
+    with patch(
+        "macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects.SimilarProjectAnalyzer.get_structure_hash"
+    ) as mock_get_structure_hash:
+        mock_get_structure_hash.side_effect = ["hash1", "hash2"]
+        result, _ = analyzer.analyze(pypi_package_json)
+
+    assert result == HeuristicResult.PASS
+
+
+def test_get_url_success(analyzer: SimilarProjectAnalyzer) -> None:
+    """Test get_url method with a successful response."""
+    mock_data = {
+        "urls": [
+            {"packagetype": "bdist_wheel", "url": "http://example.com/wheel.whl"},
+            {"packagetype": "sdist", "url": "http://example.com/sdist.tar.gz"},
+        ]
+    }
+    with patch("macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects.send_get_http") as mock_send_get:
+        mock_send_get.return_value = mock_data
+        url = analyzer.get_url("test_package")
+        assert url == "http://example.com/sdist.tar.gz"
+
+
+def test_get_url_no_sdist(analyzer: SimilarProjectAnalyzer) -> None:
+    """Test get_url method when no sdist is found."""
+    mock_data = {"urls": [{"packagetype": "bdist_wheel", "url": "http://example.com/wheel.whl"}]}
+    with patch("macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects.send_get_http") as mock_send_get:
+        mock_send_get.return_value = mock_data
+        url = analyzer.get_url("test_package")
+        assert url is None
+
+
+def test_get_url_request_fails(analyzer: SimilarProjectAnalyzer) -> None:
+    """Test get_url method when the HTTP request fails."""
+    with patch("macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects.send_get_http") as mock_send_get:
+        mock_send_get.return_value = None
+        url = analyzer.get_url("test_package")
+        assert url is None
+
+
+def test_get_structure_hash(analyzer: SimilarProjectAnalyzer) -> None:
+    """Test get_structure_hash method."""
+    with patch(
+        "macaron.malware_analyzer.pypi_heuristics.metadata.similar_projects.SimilarProjectAnalyzer.get_structure"
+    ) as mock_get_structure:
+        mock_get_structure.return_value = [
+            "test_package-1.0/setup.py",
+            "test_package-1.0/test_package/__init__.py",
+        ]
+
+        structure_hash = analyzer.get_structure_hash("test_package")
+
+        assert isinstance(structure_hash, str)
+        assert len(structure_hash) == 64
+
+        # Verify normalization
+        mock_get_structure.return_value = [
+            "other_package-1.0/other_package/__init__.py",
+            "other_package-1.0/setup.py",
+        ]
+        structure_hash2 = analyzer.get_structure_hash("other_package")
+        assert structure_hash == structure_hash2
