chore: check file size limit before downloading source and skip to prevent loss of result

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

Author: art1f1c3R

src/macaron/slsa_analyzer/checks/detect_malicious_metadata_check.py

@@ -148,6 +148,10 @@ def analyze_source(
         if not force and analyzer.depends_on and self._should_skip(results, analyzer.depends_on):
             return {analyzer.heuristic: HeuristicResult.SKIP}, {}
 
+        if not pypi_package_json.can_download_sourcecode():
+            logger.debug("Source code will exceed download limits. Please increase the download size limit to analyze.")
+            return {analyzer.heuristic: HeuristicResult.SKIP}, {}
+
         try:
             with pypi_package_json.sourcecode():
                 result, detail_info = analyzer.analyze(pypi_package_json)

src/macaron/slsa_analyzer/package_registry/pypi_registry.py

@@ -26,7 +26,12 @@
 from macaron.json_tools import json_extract
 from macaron.malware_analyzer.datetime_parser import parse_datetime
 from macaron.slsa_analyzer.package_registry.package_registry import PackageRegistry
-from macaron.util import download_file_with_size_limit, send_get_http_raw, stream_file_with_size_limit
+from macaron.util import (
+    can_download_file,
+    download_file_with_size_limit,
+    send_get_http_raw,
+    stream_file_with_size_limit,
+)
 
 if TYPE_CHECKING:
     from macaron.slsa_analyzer.specs.package_registry_spec import PackageRegistryInfo
@@ -209,6 +214,23 @@ def cleanup_sourcecode_directory(
             raise InvalidHTTPResponseError(error_message) from error
         raise InvalidHTTPResponseError(error_message)
 
+    def can_download_package_sourcecode(self, url: str) -> bool:
+        """Check if the package source code can be downloaded within the default file limits.
+
+        Parameters
+        ----------
+        url: str
+            The package source code url.
+
+        Returns
+        -------
+        bool
+            True if it can be downloaded within the size limits, otherwise False.
+        """
+        size_limit = defaults.getint("slsa.verifier", "max_download_size", fallback=10000000)
+        timeout = defaults.getint("downloads", "timeout", fallback=120)
+        return can_download_file(url, size_limit, timeout=timeout)
+
     def download_package_sourcecode(self, url: str) -> str:
         """Download the package source code from pypi registry.
 
@@ -624,6 +646,19 @@ def download_sourcecode(self) -> bool:
                 logger.debug(error)
         return False
 
+    def can_download_sourcecode(self) -> bool:
+        """Return whether the package source code can be downloaded within the download file size limits.
+
+        Returns
+        -------
+        bool
+            ``True`` if the source code can be downloaded; ``False`` if not.
+        """
+        url = self.get_sourcecode_url()
+        if url:
+            return self.pypi_registry.can_download_package_sourcecode(url)
+        return False
+
     def get_sourcecode_file_contents(self, path: str) -> bytes:
         """
         Get the contents of a single source code file specified by the path.

src/macaron/util.py

@@ -286,6 +286,35 @@ def chunk_function(self, chunk: bytes) -> None:
         self.file.write(chunk)
 
 
+def can_download_file(url: str, size_limit: int, timeout: int | None = None) -> bool:
+    """Send a head request to check if the file provided at url can be downloaded within the size limit.
+
+    It expects a URL to a file, and checks the "Content-Length" field of the response.
+
+    Parameters
+    ----------
+    url: str
+        The target of the request.
+    size_limit: int
+        The size limit in bytes of the file.
+    timeout: int | None
+        The request timeout (optional).
+
+    Returns
+    -------
+    bool
+        True if the file can be downloaded within the size limit, False otherwise.
+    """
+    response = send_head_http_raw(url, timeout=timeout, allow_redirects=True)
+    if not response:
+        return False
+
+    size = response.headers.get("Content-Length")
+    if size and int(size) <= size_limit:
+        return True
+    return False
+
+
 def download_file_with_size_limit(
     url: str, headers: dict, file_path: str, timeout: int = 40, size_limit: int = 0
 ) -> bool: