refactor: similar projects looks at pypi inspector links

Signed-off-by: Carl Flottmann <[email protected]>

Author: art1f1c3R

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

@@ -4,15 +4,12 @@
 """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
+from macaron.slsa_analyzer.package_registry.pypi_registry import PyPIInspectorAsset, PyPIPackageJsonAsset
 
 logger: logging.Logger = logging.getLogger(__name__)
 
@@ -24,20 +21,7 @@ def __init__(self) -> None:
         super().__init__(
             name="similar_project_analyzer",
             heuristic=Heuristics.SIMILAR_PROJECTS,
-            # TODO: these dependencies are used as this heuristic currently downloads many package sourcecode
-            # tarballs. Refactoring this heuristic to run more efficiently means this should have depends_on=None.
-            depends_on=[
-                (Heuristics.EMPTY_PROJECT_LINK, HeuristicResult.FAIL),
-                (Heuristics.ONE_RELEASE, HeuristicResult.FAIL),
-                (Heuristics.HIGH_RELEASE_FREQUENCY, HeuristicResult.FAIL),
-                (Heuristics.UNCHANGED_RELEASE, HeuristicResult.FAIL),
-                (Heuristics.CLOSER_RELEASE_JOIN_DATE, HeuristicResult.FAIL),
-                (Heuristics.SUSPICIOUS_SETUP, HeuristicResult.FAIL),
-                (Heuristics.WHEEL_ABSENCE, HeuristicResult.FAIL),
-                (Heuristics.ANOMALOUS_VERSION, HeuristicResult.FAIL),
-                (Heuristics.TYPOSQUATTING_PRESENCE, HeuristicResult.FAIL),
-                (Heuristics.FAKE_EMAIL, HeuristicResult.FAIL),
-            ],
+            depends_on=None,
         )
 
     def analyze(self, pypi_package_json: PyPIPackageJsonAsset) -> tuple[HeuristicResult, dict[str, JsonType]]:
@@ -58,112 +42,127 @@ def analyze(self, pypi_package_json: PyPIPackageJsonAsset) -> tuple[HeuristicRes
         HeuristicAnalyzerValueError
             if the analysis fails.
         """
-        package_name = pypi_package_json.component_name
-        target_hash = self.get_structure_hash(package_name)
-        if not target_hash:
+        target_structure = self.get_normalized_structure(pypi_package_json)
+        if not target_structure:
             return HeuristicResult.SKIP, {}
+        target_hash = hashlib.sha256("\n".join(target_structure).encode("utf-8")).hexdigest()
+        detail_info: dict = {}
+        similar_projects: list[str] = []
+        result: HeuristicResult = HeuristicResult.PASS
+
+        maintainers = pypi_package_json.pypi_registry.get_maintainers_of_package(pypi_package_json.component_name)
+        if not maintainers:
+            # NOTE: This would ideally raise an error, identifying malformed package information, but issues with
+            # obtaining maintainer information from the HTML page means this will remains as a SKIP for now.
+            return HeuristicResult.SKIP, {}
+
+        analyzed: set[str] = {pypi_package_json.component_name}
 
-        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:
+        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:
+                # skip if it is a package we have already analyzed
+                if package in analyzed:
                     continue
-                for package in maintainer_packages:
-                    if package == package_name:
-                        continue
+                analyzed.add(package)
 
-                    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,
-                        }
+                adjacent_pypi_json = PyPIPackageJsonAsset(
+                    package, None, False, pypi_package_json.pypi_registry, {}, "", PyPIInspectorAsset("", [], {})
+                )
+                if not adjacent_pypi_json.download(""):
+                    continue
+                structure = self.get_normalized_structure(adjacent_pypi_json)
+                if not structure:
+                    continue
 
-        return HeuristicResult.PASS, {}
+                hash_value = hashlib.sha256("\n".join(structure).encode("utf-8")).hexdigest()
+                if target_hash == hash_value:
+                    similar_projects.append(package)
 
-    def get_url(self, package_name: str, package_type: str = "sdist") -> str | None:
-        """Get the URL of the package's sdist.
+        detail_info["similar_projects"] = similar_projects
+        if similar_projects:
+            result = HeuristicResult.FAIL
 
-        Parameters
-        ----------
-        package_name : str
-            The name of the package.
-        package_type: str
-            The package type to retrieve the URL of.
+        return result, detail_info
 
-        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_normalized_structure(self, pypi_package_json: PyPIPackageJsonAsset) -> set[str] | None:
+        """Extract a normalized structure for a package.
 
-    def get_structure(self, package_name: str) -> list[str]:
-        """Get the file structure of the package's sdist.
+        The normalized structure is the file tree structure of all python file in the package, with the package's
+        name removed, so it is comparable.
 
         Parameters
         ----------
-        package_name : str
-            The name of the package.
+        pypi_package_json: PyPIPackageJsonAsset
+            The PyPI package JSON asset object.
 
         Returns
         -------
-        list[str]:
-            The list of files in the package's sdist.
+        set[str] | None:
+            The normalized structure of file paths in a set, or None if a problem was encountered.
         """
-        # TODO: We should not download the source distributions for every package.
-        # This is very inefficient. We should find a different way to extract the package
-        # structure, e.g., the inspector service?
-        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)
-        try:
-            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/")
-                ]
-        except (tarfile.TarError, OSError) as error:
-            logger.debug("Error reading source code tar file: %s", error)
-            return []
-
-        return members
-
-    def get_structure_hash(self, package_name: str) -> str:
-        """Get the hash of the package's file structure.
+        if not pypi_package_json.get_inspector_links():
+            return None
 
-        Parameters
-        ----------
-        package_name : str
-            The name of the package.
+        # for normalizing the structure
+        version = pypi_package_json.component_version
+        if version is None:
+            version = pypi_package_json.get_latest_version()
+        if version is None:
+            return None
 
-        Returns
-        -------
-        str:
-            The hash of the package's file structure.
-        """
-        structure = self.get_structure(package_name)
-        if not structure:
-            return ""
+        prefix = "./" + pypi_package_json.component_name + "-" + version
+        normalized_structure = set()
+
+        # try using the tarball first
+        tarball_link = pypi_package_json.inspector_asset.package_sdist_link
+        if tarball_link and pypi_package_json.inspector_asset.package_link_reachability[tarball_link]:
+            # all files are always prefixed with ./<package_name>-<version>/<...> in tarballs
+            # non-metadaata files then have <package_name>/
+            # prefix += "/" + pypi_package_json.component_name + "/"
+            structure = PyPIInspectorAsset.get_structure(tarball_link)
+            if structure:
+                for file_path in structure:
+                    # we only consider python files. This avoids considering always package-specific files like PKG_INFO, licenses,
+                    # build metadata, etc.
+                    if file_path[-3:] != ".py":
+                        continue
+
+                    # remove the "/package_name" from the prefix as well, that way the structure between two packages with different
+                    # names will be the same
+                    normalized_structure.add(
+                        file_path.removeprefix(prefix).removeprefix("/" + pypi_package_json.component_name)
+                    )
+
+                # We can't compare against wheel structures if we keep setup.py in there
+                normalized_structure.discard("/setup.py")
+                return normalized_structure
+
+        wheel_links = pypi_package_json.inspector_asset.package_whl_links
+        if len(wheel_links) > 0:
+            # wheels have this extra field for package metadata
+            prefix += ".dist-info/"
+            # structure is generally going to be the same, platform-specific details may vary for pacakges
+            # which have platform-specific wheels
+            structure = PyPIInspectorAsset.get_structure(wheel_links[0])
+            if structure:
+                for file_path in structure:
+                    # the .dist-info stuff is usually metadata
+                    if file_path.startswith(prefix) or file_path[-3:] != ".py":
+                        continue
+
+                    # remove the "./package_name" from the prefix as well, that way the structure between
+                    # two packages with different names will be the same
+                    normalized_structure.add(
+                        file_path.removeprefix(pypi_package_json.component_name + "/").removeprefix(
+                            "./" + pypi_package_json.component_name
+                        )
+                    )
 
-        normalized = sorted([p.replace(package_name, "<ROOT>") for p in structure])
+                return normalized_structure
 
-        joined = "\n".join(normalized).encode("utf-8")
-        return hashlib.sha256(joined).hexdigest()
+        # doesn't have wheel or tarball links even made, so shouldn't get here if the first line of this
+        # function worked.
+        return None

src/macaron/slsa_analyzer/package_registry/pypi_registry.py

@@ -499,6 +499,40 @@ def __bool__(self) -> bool:
             return True
         return False
 
+    @staticmethod
+    def get_structure(pypi_inspector_url: str) -> list[str] | None:
+        """Get the folder structure of a package from the inspector HTML.
+
+        Parameters
+        ----------
+        pypi_inspector_url: str
+            The URL to a pypi inspector package page.
+
+        Returns
+        -------
+        list[str] | None
+            A list containing the folder structure, or None if it could not be extracted.
+        """
+        # TODO: may have to change this in the asset. Got a client challenge without the "/" appended.
+        response = send_get_http_raw(pypi_inspector_url)
+        if not response:
+            return None
+
+        html = response.content.decode("utf-8")
+        soup = BeautifulSoup(html, "html.parser")
+        # The package structure is present on an inspector.pypi.io page inside an unordered list (<ul>). This
+        # is the only unordered list on the page.
+        if soup.ul is None:
+            return None
+
+        # All the file names sit inside <li> elements in our unordered list, under <a> tags with the 'href' class.
+        files_list = []
+        for element in soup.ul.find_all("li"):
+            if element.a and element.a["href"]:
+                files_list.append(element.a["href"])
+
+        return files_list
+
 
 @dataclass
 class PyPIPackageJsonAsset: