Merge pull request #76 from scanoss/feature/mdaloia/SP-1801-Scanoss-Py-Add-replace-action-to-scanoss.json-post-processing

feat: SP-1801 Add support for replace when specifying settings file

Author: matiasdaloia

.github/workflows/python-local-test.yml

@@ -5,10 +5,10 @@ on:
   workflow_dispatch:
   push:
     branches:
-      - 'main'
+      - "main"
   pull_request:
     branches:
-      - 'main'
+      - "main"
 
 permissions:
   contents: read
@@ -22,7 +22,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v5
         with:
-          python-version: '3.10.x'
+          python-version: "3.10.x"
 
       - name: Install Dependencies
         run: |
@@ -78,3 +78,8 @@ jobs:
             echo "Error: WFP test did not produce any results. Failing"
             exit 1
           fi
+
+      - name: Run Unit Tests
+        run: |
+          python -m unittest
+

.gitignore

@@ -26,3 +26,6 @@ local-*.txt
 docs/build
 .devcontainer/devcontainer.json
 !.devcontainer/*.example.json
+
+
+!tests/data/*.json

CHANGELOG.md

@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Upcoming changes...
 
+## [1.18.0] - 2024-11-11
+### Fixed
+- Fixed post processor being accesed if not set
+### Added
+- Add support for replace action when specifying a settings file
+- Add replaced files as context to scan request
+
 ## [1.17.5] - 2024-11-12
 ### Fixed
 - Fix dependencies scan result structure

requirements.txt

@@ -8,4 +8,5 @@ pypac
 urllib3
 pyOpenSSL
 google-api-core
-importlib_resources
+importlib_resources
+packageurl-python

setup.cfg

@@ -35,6 +35,8 @@ install_requires =
     pyOpenSSL
     google-api-core
     importlib_resources
+    packageurl-python
+
 
 [options.extras_require]
 fast_winnowing =

src/scanoss/__init__.py

@@ -22,4 +22,4 @@
    THE SOFTWARE.
 """
 
-__version__ = '1.17.5'
+__version__ = "1.18.0"

src/scanoss/scanner.py

@@ -161,12 +161,13 @@ def __init__(self, wfp: str = None, scan_output: str = None, output_format: str
         if skip_extensions:  # Append extra file extensions to skip
             self.skip_extensions.extend(skip_extensions)
 
-        if scan_settings:
-            self.scan_settings = scan_settings
-            self.post_processor = ScanPostProcessor(scan_settings, debug=debug, trace=trace, quiet=quiet)
-            self._maybe_set_api_sbom()
+        self.scan_settings = scan_settings
+        self.post_processor = ScanPostProcessor(scan_settings, debug=debug, trace=trace, quiet=quiet) if scan_settings else None
+        self._maybe_set_api_sbom()
 
     def _maybe_set_api_sbom(self):
+        if not self.scan_settings:
+            return
         sbom = self.scan_settings.get_sbom()
         if sbom:
             self.scanoss_api.set_sbom(sbom)
@@ -521,11 +522,12 @@ def __finish_scan_threaded(self, file_map: Optional[Dict[Any, Any]] = None) -> b
                 success = False
             dep_responses = self.threaded_deps.responses
 
-        raw_scan_results = self._merge_scan_results(
-            scan_responses, dep_responses, file_map
-        )
+        raw_scan_results = self._merge_scan_results(scan_responses, dep_responses, file_map)
 
-        results = self.post_processor.load_results(raw_scan_results).post_process()
+        if self.post_processor:
+            results = self.post_processor.load_results(raw_scan_results).post_process()
+        else:
+            results = raw_scan_results
 
         if self.output_format == 'plain':
             self.__log_result(json.dumps(results, indent=2, sort_keys=True))

src/scanoss/scanoss_settings.py

@@ -69,15 +69,15 @@ def load_json_file(self, filepath: str):
         json_file = Path(filepath).resolve()
 
         if not json_file.exists():
-            self.print_stderr(f"Scan settings file not found: {filepath}")
+            self.print_stderr(f'Scan settings file not found: {filepath}')
             self.data = {}
 
-        with open(json_file, "r") as jsonfile:
-            self.print_debug(f"Loading scan settings from: {filepath}")
+        with open(json_file, 'r') as jsonfile:
+            self.print_debug(f'Loading scan settings from: {filepath}')
             try:
                 self.data = json.load(jsonfile)
             except Exception as e:
-                self.print_stderr(f"ERROR: Problem parsing input JSON: {e}")
+                self.print_stderr(f'ERROR: Problem parsing input JSON: {e}')
         return self
 
     def set_file_type(self, file_type: str):
@@ -91,9 +91,7 @@ def set_file_type(self, file_type: str):
         """
         self.settings_file_type = file_type
         if not self._is_valid_sbom_file:
-            raise Exception(
-                'Invalid scan settings file, missing "components" or "bom")'
-            )
+            raise Exception('Invalid scan settings file, missing "components" or "bom")')
         return self
 
     def set_scan_type(self, scan_type: str):
@@ -111,7 +109,7 @@ def _is_valid_sbom_file(self):
         Returns:
             bool: True if the file is valid, False otherwise
         """
-        if not self.data.get("components") or not self.data.get("bom"):
+        if not self.data.get('components') or not self.data.get('bom'):
             return False
         return True
 
@@ -122,46 +120,56 @@ def _get_bom(self):
             dict: If using scanoss.json
             list: If using SBOM.json
         """
-        if self.settings_file_type == "legacy":
+        if self.settings_file_type == 'legacy':
             if isinstance(self.data, list):
                 return self.data
-            elif isinstance(self.data, dict) and self.data.get("components"):
-                return self.data.get("components")
+            elif isinstance(self.data, dict) and self.data.get('components'):
+                return self.data.get('components')
             else:
                 return []
-        return self.data.get("bom", {})
+        return self.data.get('bom', {})
 
     def get_bom_include(self) -> List[BomEntry]:
         """Get the list of components to include in the scan
 
         Returns:
             list: List of components to include in the scan
         """
-        if self.settings_file_type == "legacy":
+        if self.settings_file_type == 'legacy':
             return self._get_bom()
-        return self._get_bom().get("include", [])
+        return self._get_bom().get('include', [])
 
     def get_bom_remove(self) -> List[BomEntry]:
         """Get the list of components to remove from the scan
 
         Returns:
             list: List of components to remove from the scan
         """
-        if self.settings_file_type == "legacy":
+        if self.settings_file_type == 'legacy':
             return self._get_bom()
-        return self._get_bom().get("remove", [])
+        return self._get_bom().get('remove', [])
+
+    def get_bom_replace(self) -> List[BomEntry]:
+        """Get the list of components to replace in the scan
+
+        Returns:
+            list: List of components to replace in the scan
+        """
+        if self.settings_file_type == 'legacy':
+            return []
+        return self._get_bom().get('replace', [])
 
     def get_sbom(self):
         """Get the SBOM to be sent to the SCANOSS API
 
         Returns:
-            dict: SBOM
+            dict: SBOM request payload
         """
         if not self.data:
             return None
         return {
-            "scan_type": self.scan_type,
-            "assets": json.dumps(self._get_sbom_assets()),
+            'scan_type': self.scan_type,
+            'assets': json.dumps(self._get_sbom_assets()),
         }
 
     def _get_sbom_assets(self):
@@ -170,8 +178,15 @@ def _get_sbom_assets(self):
         Returns:
             List: List of SBOM assets
         """
-        if self.scan_type == "identify":
-            return self.normalize_bom_entries(self.get_bom_include())
+        if self.scan_type == 'identify':
+            include_bom_entries = self._remove_duplicates(self.normalize_bom_entries(self.get_bom_include()))
+            replace_bom_entries = self._remove_duplicates(self.normalize_bom_entries(self.get_bom_replace()))
+            self.print_debug(
+                f"Scan type set to 'identify'. Adding {len(include_bom_entries) + len(replace_bom_entries)} components as context to the scan. \n"
+                f"From Include list: {[entry['purl'] for entry in include_bom_entries]} \n"
+                f"From Replace list: {[entry['purl'] for entry in replace_bom_entries]} \n"
+            )
+            return include_bom_entries + replace_bom_entries
         return self.normalize_bom_entries(self.get_bom_remove())
 
     @staticmethod
@@ -188,7 +203,26 @@ def normalize_bom_entries(bom_entries) -> List[BomEntry]:
         for entry in bom_entries:
             normalized_bom_entries.append(
                 {
-                    "purl": entry.get("purl", ""),
+                    'purl': entry.get('purl', ''),
                 }
             )
         return normalized_bom_entries
+
+    @staticmethod
+    def _remove_duplicates(bom_entries: List[BomEntry]) -> List[BomEntry]:
+        """Remove duplicate BOM entries
+
+        Args:
+            bom_entries (List[Dict]): List of BOM entries
+
+        Returns:
+            List: List of unique BOM entries
+        """
+        already_added = set()
+        unique_entries = []
+        for entry in bom_entries:
+            entry_tuple = tuple(entry.items())
+            if entry_tuple not in already_added:
+                already_added.add(entry_tuple)
+                unique_entries.append(entry)
+        return unique_entries