Add REST support for dependencies and vulnerabilities

* add REST support for dependencies and vulnerabilities

* updated scanoss-py to v1.32.0

* fix issue with api key as evn

Author: eeisegn

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Upcoming changes...
 
+## [1.32.0] - 2025-09-01
+### Added
+- Switched vulnerability and dependency APIs to use REST by default
+
 ## [1.31.5] - 2025-08-27
 ### Added
 - Added jira markdown option for DT
@@ -655,4 +659,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 [1.31.2]: https://github.com/scanoss/scanoss.py/compare/v1.31.1...v1.31.2
 [1.31.3]: https://github.com/scanoss/scanoss.py/compare/v1.31.2...v1.31.3
 [1.31.4]: https://github.com/scanoss/scanoss.py/compare/v1.31.3...v1.31.4
-[1.31.5]: https://github.com/scanoss/scanoss.py/compare/v1.31.4...v1.31.5
+[1.31.5]: https://github.com/scanoss/scanoss.py/compare/v1.31.4...v1.31.5
+[1.31.5]: https://github.com/scanoss/scanoss.py/compare/v1.31.5...v1.32.0

src/scanoss/__init__.py

@@ -22,4 +22,4 @@
   THE SOFTWARE.
 """
 
-__version__ = '1.31.5'
+__version__ = '1.32.0'

src/scanoss/cli.py

@@ -308,6 +308,7 @@ def setup_args() -> None:  # noqa: PLR0912, PLR0915
         help='Retrieve vulnerabilities for the given components',
     )
     c_vulns.set_defaults(func=comp_vulns)
+    c_vulns.add_argument('--grpc', action='store_true', help='Enable gRPC support')
 
     # Component Sub-command: component semgrep
     c_semgrep = comp_sub.add_parser(
@@ -964,7 +965,7 @@ def setup_args() -> None:  # noqa: PLR0912, PLR0915
         p.add_argument(
             '--apiurl', type=str, help='SCANOSS API URL (optional - default: https://api.osskb.org/scan/direct)'
         )
-        p.add_argument('--ignore-cert-errors', action='store_true', help='Ignore certificate errors')
+        p.add_argument('--grpc', action='store_true', help='Enable gRPC support')
 
     # Global Scan/Fingerprint filter options
     for p in [p_scan, p_wfp]:
@@ -1055,6 +1056,7 @@ def setup_args() -> None:  # noqa: PLR0912, PLR0915
             type=str,
             help='Headers to be sent on request (e.g., -hdr "Name: Value") - can be used multiple times',
         )
+        p.add_argument('--ignore-cert-errors', action='store_true', help='Ignore certificate errors')
 
     # Syft options
     for p in [p_cs, p_dep]:
@@ -1418,6 +1420,7 @@ def scan(parser, args):  # noqa: PLR0912, PLR0915
         strip_snippet_ids=args.strip_snippet,
         scan_settings=scan_settings,
         req_headers=process_req_headers(args.header),
+        use_grpc=args.grpc
     )
     if args.wfp:
         if not scanner.is_file_or_snippet_scan():
@@ -2144,6 +2147,8 @@ def comp_vulns(parser, args):
         pac=pac_file,
         timeout=args.timeout,
         req_headers=process_req_headers(args.header),
+        ignore_cert_errors=args.ignore_cert_errors,
+        use_grpc=args.grpc,
     )
     if not comps.get_vulnerabilities(args.input, args.purl, args.output):
         sys.exit(1)

src/scanoss/components.py

@@ -52,6 +52,8 @@ def __init__(  # noqa: PLR0913, PLR0915
         ca_cert: str = None,
         pac: PACFile = None,
         req_headers: dict = None,
+        ignore_cert_errors: bool = False,
+        use_grpc: bool = False,
     ):
         """
         Handle all component style requests
@@ -66,6 +68,9 @@ def __init__(  # noqa: PLR0913, PLR0915
         :param grpc_proxy: Specific gRPC proxy (optional)
         :param ca_cert: TLS client certificate (optional)
         :param pac: Proxy Auto-Config file (optional)
+        :param req_headers: Additional headers to send with requests (optional)
+        :param ignore_cert_errors: Ignore TLS certificate errors (optional)
+        :param use_grpc: Use gRPC instead of HTTP (optional)
         """
         super().__init__(debug, trace, quiet)
         ver_details = Scanner.version_details()
@@ -82,14 +87,28 @@ def __init__(  # noqa: PLR0913, PLR0915
             grpc_proxy=grpc_proxy,
             timeout=timeout,
             req_headers=req_headers,
+            ignore_cert_errors=ignore_cert_errors,
+            use_grpc=use_grpc,
         )
 
-    def load_purls(self, json_file: Optional[str] = None, purls: Optional[List[str]] = None) -> Optional[dict]:
+    def load_comps(self, json_file: Optional[str] = None, purls: Optional[List[str]] = None)-> Optional[dict]:
+        """
+        Load the specified components and return a dictionary
+
+        :param json_file: JSON Components file (optional)
+        :param purls: list pf PURLs (optional)
+        :return: Components Request dictionary or None
+        """
+        return self.load_purls(json_file, purls, 'components')
+
+    def load_purls(self, json_file: Optional[str] = None, purls: Optional[List[str]] = None, field:str = 'purls'
+                   ) -> Optional[dict]:
         """
         Load the specified purls and return a dictionary
 
         :param json_file: JSON PURL file (optional)
         :param purls: list of PURLs (optional)
+        :param field: Name of the dictionary field to store the purls in (default: 'purls')
         :return: PURL Request dictionary or None
         """
         if json_file:
@@ -109,14 +128,14 @@ def load_purls(self, json_file: Optional[str] = None, purls: Optional[List[str]]
             parsed_purls = []
             for p in purls:
                 parsed_purls.append({'purl': p})
-            purl_request = {'purls': parsed_purls}
+            purl_request = {field: parsed_purls}
         else:
             self.print_stderr('ERROR: No purls specified to process.')
             return None
-        purl_count = len(purl_request.get('purls', []))
-        self.print_debug(f'Parsed Purls ({purl_count}): {purl_request}')
+        purl_count = len(purl_request.get(field, []))
+        self.print_debug(f'Parsed {field} ({purl_count}): {purl_request}')
         if purl_count == 0:
-            self.print_stderr('ERROR: No PURLs parsed from request.')
+            self.print_stderr(f'ERROR: No {field} parsed from request.')
             return None
         return purl_request
 
@@ -142,8 +161,8 @@ def _open_file_or_sdtout(self, filename):
         """
         Open the given filename if requested, otherwise return STDOUT
 
-        :param filename:
-        :return:
+        :param filename: filename to open or None to return STDOUT
+        :return: file descriptor or None
         """
         file = sys.stdout
         if filename:
@@ -202,7 +221,7 @@ def get_vulnerabilities(self, json_file: str = None, purls: [] = None, output_fi
         :return: True on success, False otherwise
         """
         success = False
-        purls_request = self.load_purls(json_file, purls)
+        purls_request = self.load_comps(json_file, purls)
         if purls_request is None or len(purls_request) == 0:
             return False
         file = self._open_file_or_sdtout(output_file)

src/scanoss/scanner.py

@@ -107,6 +107,7 @@ def __init__( # noqa: PLR0913, PLR0915
         skip_md5_ids=None,
         scan_settings: 'ScanossSettings | None' = None,
         req_headers: dict = None,
+        use_grpc: bool = False,
     ):
         """
         Initialise scanning class, including Winnowing, ScanossApi, ThreadedScanning
@@ -173,6 +174,8 @@ def __init__( # noqa: PLR0913, PLR0915
             pac=pac,
             grpc_proxy=grpc_proxy,
             req_headers=self.req_headers,
+            ignore_cert_errors=ignore_cert_errors,
+            use_grpc=use_grpc
         )
         self.threaded_deps = ThreadedDependencies(sc_deps, grpc_api, debug=debug, quiet=quiet, trace=trace)
         self.nb_threads = nb_threads

src/scanoss/scanossapi.py

@@ -22,23 +22,23 @@
   THE SOFTWARE.
 """
 
+import http.client as http_client
 import logging
 import os
 import sys
 import time
+import uuid
 from json.decoder import JSONDecodeError
+
 import requests
-import uuid
-import http.client as http_client
 import urllib3
-
 from pypac import PACSession
 from pypac.parser import PACFile
 from urllib3.exceptions import InsecureRequestWarning
 
-from .scanossbase import ScanossBase
 from . import __version__
-
+from .constants import DEFAULT_TIMEOUT, MIN_TIMEOUT
+from .scanossbase import ScanossBase
 
 DEFAULT_URL = 'https://api.osskb.org/scan/direct'  # default free service URL
 DEFAULT_URL2 = 'https://api.scanoss.com/scan/direct'  # default premium service URL
@@ -52,7 +52,7 @@ class ScanossApi(ScanossBase):
     Currently support posting scan requests to the SCANOSS streaming API
     """
 
-    def __init__(  # noqa: PLR0913, PLR0915
+    def __init__(  # noqa: PLR0912, PLR0913, PLR0915
         self,
         scan_format: str = None,
         flags: str = None,
@@ -61,7 +61,7 @@ def __init__(  # noqa: PLR0913, PLR0915
         debug: bool = False,
         trace: bool = False,
         quiet: bool = False,
-        timeout: int = 180,
+        timeout: int = DEFAULT_TIMEOUT,
         ver_details: str = None,
         ignore_cert_errors: bool = False,
         proxy: str = None,
@@ -87,30 +87,28 @@ def __init__(  # noqa: PLR0913, PLR0915
             HTTPS_PROXY='http://<ip>:<port>'
         """
         super().__init__(debug, trace, quiet)
-        self.url = url
-        self.api_key = api_key
         self.sbom = None
         self.scan_format = scan_format if scan_format else 'plain'
         self.flags = flags
-        self.timeout = timeout if timeout > 5 else 180
+        self.timeout = timeout if timeout > MIN_TIMEOUT else DEFAULT_TIMEOUT
         self.retry_limit = retry if retry >= 0 else 5
         self.ignore_cert_errors = ignore_cert_errors
         self.req_headers = req_headers if req_headers else {}
         self.headers = {}
-
+        # Set the correct URL/API key combination
+        self.url = url if url else SCANOSS_SCAN_URL
+        self.api_key = api_key if api_key else SCANOSS_API_KEY
+        if self.api_key and not url and not os.environ.get('SCANOSS_SCAN_URL'):
+            self.url = DEFAULT_URL2  # API key specific and no alternative URL, so use the default premium
         if ver_details:
             self.headers['x-scanoss-client'] = ver_details
         if self.api_key:
             self.headers['X-Session'] = self.api_key
             self.headers['x-api-key'] = self.api_key
-        self.headers['User-Agent'] = f'scanoss-py/{__version__}'
-        self.headers['user-agent'] = f'scanoss-py/{__version__}'
-        self.load_generic_headers()
-
-        self.url = url if url else SCANOSS_SCAN_URL
-        self.api_key = api_key if api_key else SCANOSS_API_KEY
-        if self.api_key and not url and not os.environ.get('SCANOSS_SCAN_URL'):
-            self.url = DEFAULT_URL2  # API key specific and no alternative URL, so use the default premium
+        user_agent = f'scanoss-py/{__version__}'
+        self.headers['User-Agent'] = user_agent
+        self.headers['user-agent'] = user_agent
+        self.load_generic_headers(url)
 
         if self.trace:
             logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
@@ -133,7 +131,7 @@ def __init__(  # noqa: PLR0913, PLR0915
         if self.proxies:
             self.session.proxies = self.proxies
 
-    def scan(self, wfp: str, context: str = None, scan_id: int = None):
+    def scan(self, wfp: str, context: str = None, scan_id: int = None):  # noqa: PLR0912, PLR0915
         """
         Scan the specified WFP and return the JSON object
         :param wfp: WFP to scan
@@ -192,7 +190,7 @@ def scan(self, wfp: str, context: str = None, scan_id: int = None):
                     else:
                         self.print_stderr(f'Warning: No response received from {self.url}. Retrying...')
                         time.sleep(5)
-                elif r.status_code == 503:  # Service limits have most likely been reached
+                elif r.status_code == requests.codes.service_unavailable:  # Service limits most likely reached
                     self.print_stderr(
                         f'ERROR: SCANOSS API rejected the scan request ({request_id}) due to '
                         f'service limits being exceeded'
@@ -202,7 +200,7 @@ def scan(self, wfp: str, context: str = None, scan_id: int = None):
                         f'ERROR: {r.status_code} - The SCANOSS API request ({request_id}) rejected '
                         f'for {self.url} due to service limits being exceeded.'
                     )
-                elif r.status_code >= 400:
+                elif r.status_code >= requests.codes.bad_request:
                     if retry > self.retry_limit:  # No response retry_limit or more times, fail
                         self.save_bad_req_wfp(scan_files, request_id, scan_id)
                         raise Exception(
@@ -269,7 +267,7 @@ def set_sbom(self, sbom):
         self.sbom = sbom
         return self
 
-    def load_generic_headers(self):
+    def load_generic_headers(self, url):
         """
          Adds custom headers from req_headers to the headers collection.
 
@@ -279,7 +277,7 @@ def load_generic_headers(self):
         if self.req_headers:  # Load generic headers
             for key, value in self.req_headers.items():
                 if key == 'x-api-key': # Set premium URL if x-api-key header is set
-                    if not self.url and not os.environ.get('SCANOSS_SCAN_URL'):
+                    if not url and not os.environ.get('SCANOSS_SCAN_URL'):
                         self.url = DEFAULT_URL2  # API key specific and no alternative URL, so use the default premium
                     self.api_key = value
                 self.headers[key] = value