feat: added support for authentication to OSS Index #1

chore: typing in tests

Signed-off-by: Paul Horton <phorton@sonatype.com>

ossindex/exception.py

@@ -0,0 +1,29 @@
+#
+# Copyright 2022-Present Sonatype Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+class OssIndexException(Exception):
+    """
+    Base exception which all exceptions raised by this library extend.
+    """
+    pass
+
+
+class AccessDeniedException(OssIndexException):
+    """
+    Raised if supplied credentials for Oss Index are invalid.
+    """
+    pass

ossindex/ossindex.py

@@ -16,37 +16,68 @@
 import json
 import logging
 import os
+import sys
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple
 
-import pkg_resources
 import requests
+import yaml
 # See https://github.com/package-url/packageurl-python/issues/65
 from packageurl import PackageURL  # type: ignore
 from tinydb import TinyDB, Query
 from tinydb.table import Document
 
+from .exception import AccessDeniedException
 from .model import OssIndexComponent
 from .serializer import json_decoder, OssIndexJsonEncoder
 
 logger = logging.getLogger('ossindex')
 
+if sys.version_info >= (3, 8):
+    from importlib.metadata import version as meta_version
+else:
+    from importlib_metadata import version as meta_version
+
+ossindex_lib_version: str = 'TBC'
+try:
+    ossindex_lib_version = str(meta_version('ossindex-lib'))  # type: ignore[no-untyped-call]
+except Exception:
+    ossindex_lib_version = 'DEVELOPMENT'
+
 
 class OssIndex:
+    DEFAULT_CONFIG_FILE = '.oss-index.config'
+
     _caching_enabled: bool = False
     _cache_directory: str = '.ossindex'
     _cache_ttl_in_hours: int = 12
 
     _oss_index_api_version: str = 'v3'
     _oss_index_host: str = 'https://ossindex.sonatype.org'
     _oss_max_coordinates_per_request: int = 128
+    _oss_index_authentication: Optional[requests.auth.HTTPBasicAuth] = None
 
     def __init__(self, *, enable_cache: bool = True, cache_location: Optional[str] = None) -> None:
         self._caching_enabled = enable_cache
         if self._caching_enabled:
             logger.info('OssIndex caching is ENABLED')
             self._setup_cache(cache_location=cache_location)
+        self._attempt_config_load()
+
+    def has_ossindex_authentication(self) -> bool:
+        return self._oss_index_authentication is not None
+
+    def _attempt_config_load(self) -> None:
+        try:
+            config_filename: str = os.path.join(os.path.expanduser('~'), OssIndex.DEFAULT_CONFIG_FILE)
+            with open(config_filename, 'r') as ossindex_confg_f:
+                ossindex_config = yaml.safe_load(ossindex_confg_f.read())
+                self._oss_index_authentication = requests.auth.HTTPBasicAuth(
+                    ossindex_config['username'], ossindex_config['password']
+                )
+        except FileNotFoundError:
+            pass
 
     def get_component_report(self, packages: List[PackageURL]) -> List[OssIndexComponent]:
         logger.debug('A total of {} Packages to be queried against OSS Index'.format(len(packages)))
@@ -117,9 +148,7 @@ def _get_headers() -> Dict[str, str]:
         return {
             'Accept': 'application/vnd.ossindex.component-report.v1+json',
             'Content-type': 'application/vnd.ossindex.component-report-request.v1+json',
-            'User-Agent': 'python-oss-index-lib@{}'.format(
-                pkg_resources.get_distribution('ossindex-lib').version
-            )
+            'User-Agent': f'python-oss-index-lib@{ossindex_lib_version}'
         }
 
     def _get_results(self, packages: List[PackageURL]) -> List[OssIndexComponent]:
@@ -150,8 +179,13 @@ def _make_oss_index_component_report_call(self, packages: List[PackageURL]) -> L
             headers=self._get_headers(),
             json={
                 'coordinates': list(map(lambda p: str(p.to_string()), packages))
-            }
+            },
+            auth=self._oss_index_authentication
         )
+
+        if not response.status_code == 200:
+            raise AccessDeniedException()
+
         results: List[OssIndexComponent] = []
         for oic in response.json(object_hook=json_decoder):
             results.append(oic)

pyproject.toml

@@ -40,7 +40,9 @@ keywords = [
 
 [tool.poetry.dependencies]
 python = "^3.6"
+importlib-metadata = { version = ">= 3.4", python = "< 3.8" }
 packageurl-python = "^0.9.0"
+PyYAML = "^5.4.1"
 requests = "^2.20.0"
 tinydb = "^4.5.0"
 # `types-requeests` should stay in sync with `requests`

requirements.lowest.txt

@@ -14,6 +14,8 @@
 # limitations under the License.
 #
 packageurl-python == 0.9.0
+importlib-metadata == 3.4.0 # ; python_version < '3.8'
+PyYAML == 5.4.1
 requests == 2.20.0
 tinydb == 4.5.0
 types-requests == 2.25.1

tests/fixtures/.oss-index.config

@@ -0,0 +1,2 @@
+username: test
+password: password

tests/mocks.py

@@ -14,14 +14,22 @@
 # limitations under the License.
 #
 import json
-from typing import Callable
+from typing import Callable, Optional
 
 
 class MockResponse:
-    def __init__(self, data, status_code):
-        self.text = data
+    def __init__(self, data: Optional[str], status_code: int) -> None:
+        self._text = data if data else ''
         self._status_code = status_code
 
+    @property
+    def status_code(self) -> int:
+        return self._status_code
+
+    @property
+    def text(self) -> str:
+        return self._text
+
     def json(self, object_hook: Callable) -> object:
         return json.loads(self.text, object_hook=object_hook)
 
@@ -32,6 +40,9 @@ def mock_oss_index_post(*args, **kwargs) -> MockResponse:
 
         if 'coordinates' in request_json.keys() and len(request_json['coordinates']) > 0:
             mock_response_data = []
+            if 'pkg:pypi/pip@0.0.7' in request_json['coordinates']:
+                return MockResponse(None, 401)
+
             if 'pkg:pypi/pip@21.2.3' in request_json['coordinates']:
                 mock_response_data.append({
                     "coordinates": "pkg:pypi/pip@21.2.3",