refactor: update analyzer to use existing utility methods

Signed-off-by: Amine <amine.raouane@enim.ac.ma>

Author: Amine

src/macaron/malware_analyzer/pypi_heuristics/metadata/similar_projects.py

@@ -4,18 +4,15 @@
 """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
-import typing
 
-import requests
-from bs4 import BeautifulSoup
-
-from macaron.errors import HeuristicAnalyzerValueError
 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__)
 
@@ -50,184 +47,101 @@ def analyze(self, pypi_package_json: PyPIPackageJsonAsset) -> tuple[HeuristicRes
         """
         package_name = pypi_package_json.component_name
         target_hash = self.get_structure_hash(package_name)
-        if target_hash is None:
-            return HeuristicResult.SKIP, {
-                "message": f"the package {package_name} does not have a sdist.",
-            }
-
-        similar_packages = self.get_packages(package_name)
-        if not similar_packages:
-            return HeuristicResult.SKIP, {
-                "message": f"the maintainers of {package_name} do not maintain any other packages.",
-            }
-
-        for package in similar_packages:
-            package_hash = self.get_structure_hash(package)
-            if package_hash is None:
-                logger.info("Package does not have a sdist.")
-                continue
-            if package_hash == target_hash:
-                return HeuristicResult.FAIL, {
-                    "similar_package": package,
-                }
+        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_maintainers(self, package_name: str) -> list[str]:
-        """Get all maintainers of a package.
+    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_name : str
+            The name of the package.
+        package_type: str
+            The package type to retrieve the URL of.
 
         Returns
         -------
-            list[str]: A list of maintainers.
+        str | None:
+            The URL of the package's sdist or None if not found.
         """
-        url = f"https://pypi.org/project/{package_name}/"
-        response = requests.get(url, timeout=10)
-        if response.status_code != 200:
-            return []
-
-        soup = BeautifulSoup(response.text, "html.parser")
-        gravatar_spans = soup.find_all("span", class_="sidebar-section__user-gravatar-text")
-        maintainers = [span.get_text().strip() for span in gravatar_spans]
+        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
 
-        return maintainers
+        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_packages_by_user(self, username: str) -> list[str]:
-        """Get all packages by a user.
+    def get_structure(self, package_name: str) -> list[str]:
+        """Get the file structure of the package's sdist.
 
         Parameters
         ----------
-            username (str): The username of the user.
+        package_name : str
+            The name of the package.
 
         Returns
         -------
-            list[str]: A list of package names.
+        list[str]:
+            The list of files in the package's sdist.
         """
-        url = f"https://pypi.org/user/{username}/"
-        response = requests.get(url, timeout=10)
-        if response.status_code != 200:
+        sdist_url = self.get_url(package_name)
+        if not sdist_url:
+            logger.debug("Package %s does not have a sdist.", package_name)
             return []
 
-        soup = BeautifulSoup(response.text, "html.parser")
-        headers = soup.find_all("h3", class_="package-snippet__title")
-        packages = [header.get_text().strip() for header in headers]
-        return packages
+        response = send_get_http_raw(sdist_url)
+        if not response:
+            logger.debug("Failed to download sdist for package %s.", package_name)
+            return []
 
-    def get_packages(self, package_name: str) -> list[str]:
-        """Get packages that are maintained by this package's maintainers.
+        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/")
+            ]
 
-        Parameters
-        ----------
-            package_name (str): The name of the package.
+        return members
 
-        Returns
-        -------
-            list[str]: A list of similar projects.
-        """
-        similar_projects = []
-        maintainers = self.get_maintainers(package_name)
-        for user in maintainers:
-            user_packages = self.get_packages_by_user(user)
-            similar_projects.extend(user_packages)
-        # Remove the target package from the list of similar projects.
-        similar_projects_set = set(similar_projects)
-        similar_projects_set.discard(package_name)
-        return list(similar_projects_set)
-
-    def fetch_sdist_url(self, package_name: str, version: str | None = None) -> str:
-        """Fetch the sdist URL for a package.
+    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.
-            version (str): The version of the package. If None, the latest version will be used.
+        package_name : str
+            The name of the package.
 
         Returns
         -------
-            str: The sdist URL, or an empty string if not found.
+        str:
+            The hash of the package's file structure.
         """
-        url = f"https://pypi.org/pypi/{package_name}/json"
-        try:
-            response = requests.get(url, timeout=10)
-            response.raise_for_status()
-            data = response.json()
-        except requests.exceptions.RequestException as err:
-            err_message = f"Failed to fetch PyPI JSON for {package_name}: {err}"
-            raise HeuristicAnalyzerValueError(err_message) from err
-        except ValueError as err:
-            err_message = f"Failed to decode PyPI JSON for {package_name}: {err}"
-            raise HeuristicAnalyzerValueError(err_message) from err
-
-        actual_version: str
-        if version is None:
-            try:
-                actual_version = typing.cast(str, data["info"]["version"])
-            except (KeyError, TypeError) as err:
-                err_message = f"Failed to get version for {package_name}: {err}"
-                raise HeuristicAnalyzerValueError(err_message) from err
-        else:
-            actual_version = version
-
-        try:
-            for release_file in data.get("releases", {}).get(actual_version, []):
-                if isinstance(release_file, dict) and release_file.get("packagetype") == "sdist":
-                    sdist_url = release_file.get("url")
-                    if isinstance(sdist_url, str):
-                        return sdist_url
-        except Exception as err:
-            err_message = f"Failed to parse releases for {package_name} version {actual_version}: {err}"
-            raise HeuristicAnalyzerValueError(err_message) from err
-
-        return ""
-
-    def get_structure_hash(self, package_name: str) -> str | None:
-        """Calculate a hash based on the project's file structure.
+        structure = self.get_structure(package_name)
+        if not structure:
+            return ""
 
-        Parameters
-        ----------
-            package_name (str): The name of the package.
-
-        Returns
-        -------
-            str: The structure hash.
-
-        Raises
-        ------
-            ValueError: If the sdist URL cannot be fetched or the package structure cannot be hashed.
-        """
-        sdist_url = self.fetch_sdist_url(package_name)
-        if not sdist_url:
-            return None
+        normalized = sorted([p.replace(package_name, "<ROOT>") for p in structure])
 
-        try:
-            response = requests.get(sdist_url, stream=True, timeout=10)
-            response.raise_for_status()
-            raw_file_obj: typing.IO[bytes] = typing.cast(typing.IO[bytes], response.raw)
-
-            with tarfile.open(fileobj=raw_file_obj, mode="r:gz") as file_archive:
-                paths = []
-                for member in file_archive:
-                    if not member.isdir():
-                        # remove top‑level dir.
-                        parts = member.name.split("/", 1)
-                        normalized = parts[1] if len(parts) > 1 else parts[0]
-                        # replace the pkg name.
-                        normalized = normalized.replace(package_name, "<PKG>")
-                        paths.append(normalized)
-                paths.sort()
-                structure_hash_calculator = hashlib.sha256()
-                for path in paths:
-                    structure_hash_calculator.update(path.encode("utf-8"))
-                    structure_hash_calculator.update(b"\n")
-                return structure_hash_calculator.hexdigest()
-        except requests.exceptions.RequestException as err:
-            err_message = f"Failed to download sdist for {package_name} from {sdist_url}: {err}"
-            raise HeuristicAnalyzerValueError(err_message) from err
-        except tarfile.TarError as err:
-            err_message = f"Failed to process tarfile for {package_name} from {sdist_url}: {err}"
-            raise HeuristicAnalyzerValueError(err_message) from err
-        except Exception as err:
-            err_message = f"Failed to get structure hash for {package_name}: {err}"
-            raise HeuristicAnalyzerValueError(err_message) from err
+        joined = "\n".join(normalized).encode("utf-8")
+        return hashlib.sha256(joined).hexdigest()

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.