feat: dependency on empty project link, and context manager for sourcecode download

Author: art1f1c3R

src/macaron/malware_analyzer/pypi_heuristics/sourcecode/pypi_sourcecode_analyzer.py

@@ -23,13 +23,14 @@
 from macaron.config.global_config import global_config
 from macaron.errors import ConfigurationError, HeuristicAnalyzerValueError
 from macaron.json_tools import JsonType, json_extract
-from macaron.malware_analyzer.pypi_heuristics.heuristics import HeuristicResult
+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
 
 logger: logging.Logger = logging.getLogger(__name__)
 
 
-class PyPISourcecodeAnalyzer:
+class PyPISourcecodeAnalyzer(BaseHeuristicAnalyzer):
     """This class is used to analyze the source code of python PyPI packages. This analyzer is a work in progress.
 
     This analyzer works in two phases. In the first phase, it will perform a pattern-based scan of all python files
@@ -48,8 +49,14 @@ class PyPISourcecodeAnalyzer:
     of the package.
     """
 
-    def __init__(self, resources_path: str = global_config.resources_path) -> None:
-        """Collect required data for analysing the source code."""
+    def __init__(self, resources_path: str | None = None) -> None:
+        super().__init__(
+            name="anomalous_version_analyzer",
+            heuristic=Heuristics.SUSPICIOUS_PATTERNS,
+            depends_on=[(Heuristics.EMPTY_PROJECT_LINK, HeuristicResult.FAIL)],
+        )
+        if resources_path is None:
+            resources_path = global_config.resources_path
         self.default_rule_path, self.custom_rule_path = self._load_defaults(resources_path)
 
     def _load_defaults(self, resources_path: str) -> tuple[str, str | None]:

src/macaron/slsa_analyzer/checks/detect_malicious_metadata_check.py

@@ -11,7 +11,7 @@
 
 from macaron.database.db_custom_types import DBJsonDict
 from macaron.database.table_definitions import CheckFacts
-from macaron.errors import ConfigurationError, HeuristicAnalyzerValueError
+from macaron.errors import ConfigurationError, HeuristicAnalyzerValueError, SourceCodeError
 from macaron.json_tools import JsonType, json_extract
 from macaron.malware_analyzer.pypi_heuristics.base_analyzer import BaseHeuristicAnalyzer
 from macaron.malware_analyzer.pypi_heuristics.heuristics import HeuristicResult, Heuristics
@@ -282,27 +282,45 @@ def _should_skip(
                 return True
         return False
 
-    def analyze_source(self, pypi_package_json: PyPIPackageJsonAsset) -> tuple[HeuristicResult, dict[str, JsonType]]:
+    def analyze_source(
+        self, pypi_package_json: PyPIPackageJsonAsset, results: dict[Heuristics, HeuristicResult]
+    ) -> tuple[HeuristicResult, dict[str, JsonType]]:
         """Analyze the source code of the package with a textual scan, looking for malicious code patterns.
 
         Parameters
         ----------
         pypi_package_json: PyPIPackageJsonAsset
             The PyPI package JSON asset object.
+        results: dict[Heuristics, HeuristicResult]
+            Containing all heuristics' results (excluding this one), where the key is the heuristic and the value is the result
+            associated with that heuristic.
 
         Returns
         -------
         tuple[HeuristicResult, dict[str, JsonType]]
             Containing the analysis results and relevant patterns identified.
+
+        Raises
+        ------
+        HeuristicAnalyzerValueError
+            If the analyzer fails due to malformed package information.
+        ConfigurationError
+            If the configuration of the analyzer encountered a problem.
         """
         logger.debug("Instantiating %s", PyPISourcecodeAnalyzer.__name__)
-        try:
-            sourcecode_analyzer = PyPISourcecodeAnalyzer()
-            return sourcecode_analyzer.analyze(pypi_package_json)
-        except (ConfigurationError, HeuristicAnalyzerValueError) as source_code_error:
-            logger.debug("Unable to perform source code analysis: %s", source_code_error)
+        analyzer = PyPISourcecodeAnalyzer()
+
+        if analyzer.depends_on and self._should_skip(results, analyzer.depends_on):
             return HeuristicResult.SKIP, {}
 
+        try:
+            with pypi_package_json.sourcecode():
+                return analyzer.analyze(pypi_package_json)
+        except SourceCodeError as error:
+            error_msg = f"Unable to perform analysis, source code not available: {error}"
+            logger.debug(error_msg)
+            raise HeuristicAnalyzerValueError(error_msg) from error
+
     def run_heuristics(
         self, pypi_package_json: PyPIPackageJsonAsset
     ) -> tuple[dict[Heuristics, HeuristicResult], dict[str, JsonType]]:
@@ -428,9 +446,15 @@ def run_check(self, ctx: AnalyzeContext) -> CheckResultData:
                             confidence = Confidence.HIGH
                             result_type = CheckResultType.PASSED
 
-                        # experimental analyze sourcecode feature
-                        if ctx.dynamic_data["analyze_source"] and pypi_package_json.download_sourcecode():
-                            sourcecode_result, sourcecode_detail_info = self.analyze_source(pypi_package_json)
+                        # experimental sourcecode analysis feature
+                        if ctx.dynamic_data["analyze_source"]:
+                            try:
+                                sourcecode_result, sourcecode_detail_info = self.analyze_source(
+                                    pypi_package_json, heuristic_results
+                                )
+                            except (HeuristicAnalyzerValueError, ConfigurationError):
+                                return CheckResultData(result_tables=[], result_type=CheckResultType.UNKNOWN)
+
                             heuristic_results[Heuristics.SUSPICIOUS_PATTERNS] = sourcecode_result
                             heuristics_detail_info.update(sourcecode_detail_info)
 
@@ -440,8 +464,6 @@ def run_check(self, ctx: AnalyzeContext) -> CheckResultData:
                                     confidence = Confidence.LOW
                                 result_type = CheckResultType.FAILED
 
-                            pypi_package_json.cleanup_sourcecode()
-
                         result_tables.append(
                             MaliciousMetadataFacts(
                                 result=heuristic_results,

src/macaron/slsa_analyzer/package_registry/pypi_registry.py

@@ -10,7 +10,8 @@
 import tarfile
 import tempfile
 import urllib.parse
-from collections.abc import Callable, Iterator
+from collections.abc import Callable, Generator, Iterator
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime
 
@@ -538,6 +539,14 @@ def get_latest_release_upload_time(self) -> str | None:
             return upload_time
         return None
 
+    @contextmanager
+    def sourcecode(self) -> Generator[None]:
+        """Download and cleanup source code of the package with a context manager."""
+        if not self.download_sourcecode():
+            raise SourceCodeError("Unable to download package source code.")
+        yield
+        self.cleanup_sourcecode()
+
     def download_sourcecode(self) -> bool:
         """Get the source code of the package and store it in a temporary directory.