SBOM upload client_secret failure

Problem:
SBOM upload results in a 401 code and the error handler fails
to deal wth this correctly.

Solution:
Modified the error handling to properly handle when response body
has non-JSON format.
Added extra functional tests for appidp and sboms endpoints.

Signed-off-by: Paul Hewlett <phewlett76@gmail.com>

Author: eccles

archivist/errors.py

@@ -5,11 +5,14 @@
 """
 
 import json
+import logging
 from typing import Optional
 
 from .constants import HEADERS_RETRY_AFTER
 from .headers import _headers_get
 
+LOGGER = logging.getLogger(__name__)
+
 
 class ArchivistError(Exception):
     """Base exception for archivist package"""
@@ -90,21 +93,31 @@ class Archivist5xxError(ArchivistError):
 def __identity(response):
     identity = "unknown"
     if response.request:
+        LOGGER.debug("Request %s", response.request)
         req = response.request
         body = getattr(req, "body", None)
         if body:
             # when uploading a file the body attribute is a
             # MultiPartEncoder
             try:
                 body = json.loads(body)
-            except TypeError:
+            except (TypeError, json.decoder.JSONDecodeError):
                 pass
             else:
                 identity = body.get("identity", "unknown")
 
     return identity
 
 
+def __description(response):
+    status_code = response.status_code
+    if status_code == 404:
+        return f"{__identity(response)} not found ({status_code})"
+
+    text = response.text or ""
+    return f"{text} ({status_code})"
+
+
 def _parse_response(response):
     """Parse REST response
 
@@ -120,35 +133,33 @@ def _parse_response(response):
     """
 
     status_code = response.status_code
+    LOGGER.debug("Status %s", status_code)
     if status_code < 400:
         return None
 
-    text = response.text or ""
+    desc = __description(response)
 
     if status_code == 429:
         return ArchivistTooManyRequestsError(
             _headers_get(response.headers, HEADERS_RETRY_AFTER),
-            f"{text} ({status_code})",
+            desc,
         )
 
     if 400 <= status_code < 500:
         err, arg = {
-            400: (ArchivistBadRequestError, f"{text} ({status_code})"),
-            401: (ArchivistUnauthenticatedError, f"{text} ({status_code})"),
-            402: (ArchivistPaymentRequiredError, f"{text} ({status_code})"),
-            403: (ArchivistForbiddenError, f"{text} ({status_code})"),
-            404: (
-                ArchivistNotFoundError,
-                f"{__identity(response)} not found ({status_code})",
-            ),
-        }.get(status_code, (Archivist4xxError, f"{text} ({status_code})"))
+            400: (ArchivistBadRequestError, desc),
+            401: (ArchivistUnauthenticatedError, desc),
+            402: (ArchivistPaymentRequiredError, desc),
+            403: (ArchivistForbiddenError, desc),
+            404: (ArchivistNotFoundError, desc),
+        }.get(status_code, (Archivist4xxError, desc))
         return err(arg)
 
     if 500 <= status_code < 600:
         err, arg = {
-            501: (ArchivistNotImplementedError, f"{text} ({status_code})"),
-            503: (ArchivistUnavailableError, f"{text} ({status_code})"),
-        }.get(status_code, (Archivist5xxError, f"{text} ({status_code})"))
+            501: (ArchivistNotImplementedError, desc),
+            503: (ArchivistUnavailableError, desc),
+        }.get(status_code, (Archivist5xxError, desc))
         return err(arg)
 
-    return ArchivistError(f"{text} ({status_code})")
+    return ArchivistError(desc)

archivist/sboms.py

@@ -92,7 +92,7 @@ def upload(
 
         """
 
-        LOGGER.debug("Upload SBOM")
+        LOGGER.debug("Upload SBOM %s", params)
 
         sbom = SBOM(
             **self._archivist.post_file(

functests/execapplications.py

@@ -9,6 +9,7 @@
 from uuid import uuid4
 
 from archivist.archivist import Archivist
+from archivist.errors import ArchivistUnauthenticatedError
 from archivist.logger import set_logger
 from archivist.proof_mechanism import ProofMechanism
 
@@ -171,6 +172,22 @@ def test_appidp_token(self):
         )
         print("appidp", json_dumps(appidp, indent=4))
 
+    def test_appidp_token_404(self):
+        """
+        Test appidp token
+        """
+        application = self.arch.applications.create(
+            self.display_name,
+            CUSTOM_CLAIMS,
+        )
+        client_id = application["client_id"]
+        client_secret = "X" + application["credentials"][0]["secret"][1:]
+        with self.assertRaises(ArchivistUnauthenticatedError):
+            appidp = self.arch.appidp.token(
+                client_id,
+                client_secret,
+            )
+
     def test_archivist_token(self):
         """
         Test archivist with client id/secret

functests/execsboms.py

@@ -9,6 +9,7 @@
 from unittest import TestCase
 
 from archivist.archivist import Archivist
+from archivist.errors import ArchivistBadRequestError
 from archivist.logger import set_logger
 from archivist.timestamp import now_timestamp
 
@@ -25,6 +26,7 @@
 }
 
 TEST_SBOM_PATH = "functests/test_resources/bom.xml"
+TEST_SBOM_SPDX_PATH = "functests/test_resources/bom.spdx"
 TEST_SBOM_DOWNLOAD_PATH = "functests/test_resources/downloaded_bom.xml"
 
 if "TEST_DEBUG" in environ and environ["TEST_DEBUG"]:
@@ -57,15 +59,15 @@ def tearDown(cls) -> None:
         with suppress(FileNotFoundError):
             remove(TEST_SBOM_DOWNLOAD_PATH)
 
-    def test_sbom_upload_with_public_privacy(self):
+    def test_sbom_upload_with_private_privacy(self):
         """
         Test sbom upload with privacy
         """
         now = now_timestamp()
-        print("Title:", self.title, now)
+        print("Public Upload Title:", self.title, now)
         with open(TEST_SBOM_PATH, "rb") as fd:
             metadata = self.arch.sboms.upload(
-                fd, confirm=True, params={"privacy": "PUBLIC"}
+                fd, confirm=True, params={"privacy": "PRIVATE"}
             )
         print("first upload", json_dumps(metadata.dict(), indent=4))
         identity = metadata.identity
@@ -78,12 +80,57 @@ def test_sbom_upload_with_public_privacy(self):
             msg="Metadata not correct",
         )
 
+    def test_sbom_upload_with_illegal_privacy(self):
+        """
+        Test sbom upload with privacy
+        """
+        now = now_timestamp()
+        print("Illegal Upload Title:", self.title, now)
+        with open(TEST_SBOM_PATH, "rb") as fd:
+            with self.assertRaises(ArchivistBadRequestError):
+                metadata = self.arch.sboms.upload(
+                    fd, confirm=True, params={"privacy": "XXXXXX"}
+                )
+
+    def test_sbom_upload_with_spdx(self):
+        """
+        Test sbom upload with spdx
+        """
+        now = now_timestamp()
+        print("SPDX Upload Title:", self.title, now)
+        with open(TEST_SBOM_SPDX_PATH, "rb") as fd:
+            metadata = self.arch.sboms.upload(
+                fd, confirm=True, params={"sbomType": "spdx-tag"}
+            )
+        print("first upload", json_dumps(metadata.dict(), indent=4))
+        identity = metadata.identity
+
+        metadata1 = self.arch.sboms.read(identity)
+        print("read", json_dumps(metadata1.dict(), indent=4))
+        self.assertEqual(
+            metadata,
+            metadata1,
+            msg="Metadata not correct",
+        )
+
+    def test_sbom_upload_with_illegal_format(self):
+        """
+        Test sbom upload with illegal format
+        """
+        now = now_timestamp()
+        print("SPDX Upload Title:", self.title, now)
+        with open(TEST_SBOM_SPDX_PATH, "rb") as fd:
+            with self.assertRaises(ArchivistBadRequestError):
+                metadata = self.arch.sboms.upload(
+                    fd, confirm=True, params={"sbomType": "xxxxxxxx"}
+                )
+
     def test_sbom_upload_with_confirmation(self):
         """
         Test sbom upload with confirmation
         """
         now = now_timestamp()
-        print("Title:", self.title, now)
+        print("Confirmed Upload Title:", self.title, now)
         with open(TEST_SBOM_PATH, "rb") as fd:
             metadata = self.arch.sboms.upload(fd, confirm=True)
         print("first upload", json_dumps(metadata.dict(), indent=4))
@@ -105,6 +152,35 @@ def test_sbom_upload_with_confirmation(self):
             msg="No. of SBOMS should be 1",
         )
 
+    def test_sbom_upload_with_cyclonedx_xml(self):
+        """
+        Test sbom upload with cyclonedx-xml
+        """
+        now = now_timestamp()
+        print("CycloneDX-XML Upload Title:", self.title, now)
+        with open(TEST_SBOM_PATH, "rb") as fd:
+            metadata = self.arch.sboms.upload(
+                fd, params={"sbomType": "cyclonedx-xml"}, confirm=True
+            )
+        print("first upload", json_dumps(metadata.dict(), indent=4))
+        identity = metadata.identity
+
+        metadata1 = self.arch.sboms.read(identity)
+        print("read", json_dumps(metadata1.dict(), indent=4))
+        self.assertEqual(
+            metadata,
+            metadata1,
+            msg="Metadata not correct",
+        )
+
+        sleep(1)  # the data may have not reached cogsearch
+        metadatas = list(self.arch.sboms.list(metadata={"uploaded_since": now}))
+        self.assertEqual(
+            len(metadatas),
+            1,
+            msg="No. of SBOMS should be 1",
+        )
+
     def test_sbom_upload_and_download(self):
         """
         Test sbom upload and download through the SDK

functests/test_resources/bom.spdx

@@ -0,0 +1,58 @@
+SPDXVersion: SPDX-2.2
+DataLicense: CC0-1.0
+SPDXID: SPDXRef-DOCUMENT
+DocumentName: hello
+DocumentNamespace: https://swinslow.net/spdx-examples/example1/hello-v3
+Creator: Person: Steve Winslow (steve@swinslow.net)
+Creator: Tool: github.com/spdx/tools-golang/builder
+Creator: Tool: github.com/spdx/tools-golang/idsearcher
+Created: 2021-08-26T01:46:00Z
+
+##### Package: hello
+
+PackageName: hello
+SPDXID: SPDXRef-Package-hello
+PackageDownloadLocation: git+https://github.com/swinslow/spdx-examples.git#example1/content
+FilesAnalyzed: true
+PackageVerificationCode: 9d20237bb72087e87069f96afb41c6ca2fa2a342
+PackageVersion: v0.1.0
+PackageLicenseConcluded: GPL-3.0-or-later
+PackageLicenseInfoFromFiles: GPL-3.0-or-later
+PackageLicenseDeclared: GPL-3.0-or-later
+PackageCopyrightText: NOASSERTION
+
+Relationship: SPDXRef-DOCUMENT DESCRIBES SPDXRef-Package-hello
+
+FileName: /build/hello
+SPDXID: SPDXRef-hello-binary
+FileType: BINARY
+FileChecksum: SHA1: 20291a81ef065ff891b537b64d4fdccaf6f5ac02
+FileChecksum: SHA256: 83a33ff09648bb5fc5272baca88cf2b59fd81ac4cc6817b86998136af368708e
+FileChecksum: MD5: 08a12c966d776864cc1eb41fd03c3c3d
+LicenseConcluded: GPL-3.0-or-later
+LicenseInfoInFile: NOASSERTION
+FileCopyrightText: NOASSERTION
+
+FileName: /src/Makefile
+SPDXID: SPDXRef-Makefile
+FileType: SOURCE
+FileChecksum: SHA1: 69a2e85696fff1865c3f0686d6c3824b59915c80
+FileChecksum: SHA256: 5da19033ba058e322e21c90e6d6d859c90b1b544e7840859c12cae5da005e79c
+FileChecksum: MD5: 559424589a4f3f75fd542810473d8bc1
+LicenseConcluded: GPL-3.0-or-later
+LicenseInfoInFile: GPL-3.0-or-later
+FileCopyrightText: NOASSERTION
+
+FileName: /src/hello.c
+SPDXID: SPDXRef-hello-src
+FileType: SOURCE
+FileChecksum: SHA1: 20862a6d08391d07d09344029533ec644fac6b21
+FileChecksum: SHA256: b4e5ca56d1f9110ca94ed0bf4e6d9ac11c2186eb7cd95159c6fdb50e8db5a823
+FileChecksum: MD5: 935054fe899ca782e11003bbae5e166c
+LicenseConcluded: GPL-3.0-or-later
+LicenseInfoInFile: GPL-3.0-or-later
+FileCopyrightText: Copyright Contributors to the spdx-examples project.
+
+Relationship: SPDXRef-hello-binary GENERATED_FROM SPDXRef-hello-src
+Relationship: SPDXRef-hello-binary GENERATED_FROM SPDXRef-Makefile
+Relationship: SPDXRef-Makefile BUILD_TOOL_OF SPDXRef-Package-hello

unittests/testerrors.py

@@ -202,6 +202,60 @@ class Object:
             msg="incorrect error",
         )
 
+    def test_errors_404_response_body_is_string(self):
+        """
+        Test errors
+        """
+
+        class Object:
+            pass
+
+        request = Object()
+        request.body = "xyz"
+        response = MockResponse(
+            404,
+            request=request,
+        )
+        error = _parse_response(response)
+        self.assertIsNotNone(
+            error,
+            msg="error should not be None",
+        )
+        with self.assertRaises(ArchivistNotFoundError) as ex:
+            raise error
+        self.assertEqual(
+            str(ex.exception),
+            "unknown not found (404)",
+            msg="incorrect error",
+        )
+
+    def test_errors_404_response_body_is_None(self):
+        """
+        Test errors
+        """
+
+        class Object:
+            pass
+
+        request = Object()
+        request.body = None
+        response = MockResponse(
+            404,
+            request=request,
+        )
+        error = _parse_response(response)
+        self.assertIsNotNone(
+            error,
+            msg="error should not be None",
+        )
+        with self.assertRaises(ArchivistNotFoundError) as ex:
+            raise error
+        self.assertEqual(
+            str(ex.exception),
+            "unknown not found (404)",
+            msg="incorrect error",
+        )
+
     def test_errors_429(self):
         """
         Test errors