Example code still used simple authtoken

Problem:
Common usage is now to use client credentials from an
application registration.

Solution:
Changed all example code to use client credentials.
Allow auth to be None in Archivist creation
Added functest that uses client credentials to publish SBOMS.
Added confirmation of publish/withdraw of SBOMS.

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

Author: eccles

archivist/archivist.py

@@ -113,7 +113,7 @@ class Archivist:  # pylint: disable=too-many-instance-attributes
     def __init__(
         self,
         url: str,
-        auth: Union[str, MachineAuth],
+        auth: Union[None, str, MachineAuth],
         *,
         fixtures: Optional[Dict] = None,
         verify: bool = True,
@@ -189,6 +189,7 @@ def auth(self) -> str:
             apptoken = self.appidp.token(self._client_id, self._client_secret)  # type: ignore
             self._auth = apptoken["access_token"]
             self._expires_at = time() + apptoken["expires_in"] - 10  # fudge factor
+            LOGGER.debug("Refresh token")
 
         return self._auth  # type: ignore
 
@@ -218,7 +219,10 @@ def __add_headers(self, headers: Optional[Dict]) -> Dict:
             newheaders = self.headers
 
         auth = self.auth  # this may trigger a refetch so only do it once here
-        newheaders["authorization"] = "Bearer " + auth.strip()
+        # for appidp endpoint there may not be an authtoken
+        if auth is not None:
+            newheaders["authorization"] = "Bearer " + auth.strip()
+
         return newheaders
 
     @retry_429
@@ -336,7 +340,6 @@ def post(
             headers = self.__add_headers(headers)
             data = json.dumps(request) if request else None
 
-        LOGGER.debug("POST data %s", data)
         response = self._session.post(
             url,
             data=data,

archivist/errors.py

@@ -23,6 +23,14 @@ class ArchivistUnconfirmedError(ArchivistError):
     """asset or event failed to confirm after fixed timeout"""
 
 
+class ArchivistUnpublishedError(ArchivistError):
+    """Sbom failed to publish after fixed timeout"""
+
+
+class ArchivistUnwithdrawnError(ArchivistError):
+    """Sbom failed to be withdrawn after fixed timeout"""
+
+
 class ArchivistIllegalArgumentError(ArchivistError):
     """Optional keyword arguments are inconsistent"""
 

archivist/publisher.py

@@ -0,0 +1,73 @@
+"""publisher interface
+
+   Wrap base methods with constants for assets (path, etc...
+"""
+
+import logging
+
+from typing import overload
+
+import backoff
+
+from .errors import ArchivistUnpublishedError
+
+
+# pylint:disable=unused-import      # To prevent cyclical import errors forward referencing is used
+# pylint:disable=cyclic-import      # but pylint doesn't understand this feature
+from . import sboms
+
+MAX_TIME = 1200
+
+LOGGER = logging.getLogger(__name__)
+
+
+def __lookup_max_time():
+    return MAX_TIME
+
+
+# pylint: disable=consider-using-f-string
+def __backoff_handler(details):
+    LOGGER.debug("MAX_TIME %s", MAX_TIME)
+    LOGGER.debug(
+        "Backing off {wait:0.1f} seconds afters {tries} tries "
+        "calling function {target} with args {args} and kwargs "
+        "{kwargs}".format(**details)
+    )
+
+
+def __on_giveup_publication(details):
+    identity = details["args"][1]
+    elapsed = details["elapsed"]
+    raise ArchivistUnpublishedError(
+        f"publication for {identity} timed out after {elapsed} seconds"
+    )
+
+
+# These overloads are used for type hinting, if self is sboms client then
+# an SBOM metadata will be returned.
+# Overloads are evaluated at startup but not at runtime, therefore
+# no test coverage be done directly.
+
+
+@overload
+def _wait_for_publication(
+    self: "sboms._SbomsClient", identity: str
+) -> "sbommetadata.SBOM":
+    ...  # pragma: no cover
+
+
+@backoff.on_predicate(
+    backoff.expo,
+    logger=LOGGER,
+    max_time=__lookup_max_time,
+    on_backoff=__backoff_handler,
+    on_giveup=__on_giveup_publication,
+)
+def _wait_for_publication(self, identity):
+    """docstring"""
+    entity = self.read(identity)
+
+    if entity.published_date:
+        return entity
+
+    return None

archivist/sboms.py

@@ -43,6 +43,7 @@
     SBOMS_WITHDRAW,
     SBOMS_PUBLISH,
 )
+from . import publisher, withdrawer
 from .dictmerge import _deepmerge
 from .sbommetadata import SBOM
 
@@ -171,44 +172,86 @@ def list(
             )
         )
 
-    def publish(self, identity: str) -> SBOM:
+    def publish(self, identity: str, confirm: bool = False) -> SBOM:
         """Publish SBOMt
 
         Makes an SBOM public.
 
         Args:
             identity (str): identity of SBOM
+            confirm (bool): if True wait for sbom to be published.
 
         Returns:
             :class:`SBOM` instance
 
         """
         LOGGER.debug("Publish SBOM %s", identity)
-        return SBOM(
+        sbom = SBOM(
             **self._archivist.post(
                 f"{SBOMS_SUBPATH}/{identity}",
                 None,
                 verb=SBOMS_PUBLISH,
             )
         )
+        if not confirm:
+            return sbom
+
+        return self.wait_for_publication(sbom.identity)
+
+    def wait_for_publication(self, identity: str) -> SBOM:
+        """Wait for sbom to be published.
+
+        Waits for sbom to be published.
+
+        Args:
+            identity (str): identity of sbom
+
+        Returns:
+            True if sbom is confirmed.
+
+        """
+        publisher.MAX_TIME = self._archivist.max_time
+        # pylint: disable=protected-access
+        return publisher._wait_for_publication(self, identity)
 
-    def withdraw(self, identity: str) -> SBOM:
+    def withdraw(self, identity: str, confirm: bool = False) -> SBOM:
         """Withdraw SBOM
 
         Withdraws an SBOM.
 
         Args:
             identity (str): identity of SBOM
+            confirm (bool): if True wait for sbom to be withdrawn.
 
         Returns:
             :class:`SBOM` instance
 
         """
         LOGGER.debug("Withdraw SBOM %s", identity)
-        return SBOM(
+        sbom = SBOM(
             **self._archivist.post(
                 f"{SBOMS_SUBPATH}/{identity}",
                 None,
                 verb=SBOMS_WITHDRAW,
             )
         )
+        if not confirm:
+            return sbom
+
+        return self.wait_for_withdrawn(sbom.identity)
+
+    def wait_for_withdrawn(self, identity: str) -> SBOM:
+        """Wait for sbom to be withdrawn.
+
+        Waits for sbom to be withdrawn.
+
+        Args:
+            identity (str): identity of sbom
+
+        Returns:
+            True if sbom is confirmed.
+
+        """
+        withdrawer.MAX_TIME = self._archivist.max_time
+        # pylint: disable=protected-access
+        return withdrawer._wait_for_withdrawn(self, identity)

archivist/withdrawer.py

@@ -0,0 +1,73 @@
+"""withdrawer interface
+
+   Wrap base methods with constants for assets (path, etc...
+"""
+
+import logging
+
+from typing import overload
+
+import backoff
+
+from .errors import ArchivistUnwithdrawnError
+
+
+# pylint:disable=unused-import      # To prevent cyclical import errors forward referencing is used
+# pylint:disable=cyclic-import      # but pylint doesn't understand this feature
+from . import sboms
+
+MAX_TIME = 1200
+
+LOGGER = logging.getLogger(__name__)
+
+
+def __lookup_max_time():
+    return MAX_TIME
+
+
+# pylint: disable=consider-using-f-string
+def __backoff_handler(details):
+    LOGGER.debug("MAX_TIME %s", MAX_TIME)
+    LOGGER.debug(
+        "Backing off {wait:0.1f} seconds afters {tries} tries "
+        "calling function {target} with args {args} and kwargs "
+        "{kwargs}".format(**details)
+    )
+
+
+def __on_giveup_withdrawn(details):
+    identity = details["args"][1]
+    elapsed = details["elapsed"]
+    raise ArchivistUnwithdrawnError(
+        f"withdrawn for {identity} timed out after {elapsed} seconds"
+    )
+
+
+# These overloads are used for type hinting, if self is sboms client then
+# an SBOM metadata will be returned.
+# Overloads are evaluated at startup but not at runtime, therefore
+# no test coverage be done directly.
+
+
+@overload
+def _wait_for_publication(
+    self: "sboms._SbomsClient", identity: str
+) -> "sbommetadata.SBOM":
+    ...  # pragma: no cover
+
+
+@backoff.on_predicate(
+    backoff.expo,
+    logger=LOGGER,
+    max_time=__lookup_max_time,
+    on_backoff=__backoff_handler,
+    on_giveup=__on_giveup_withdrawn,
+)
+def _wait_for_withdrawn(self, identity):
+    """docstring"""
+    entity = self.read(identity)
+
+    if entity.withdrawn_date:
+        return entity
+
+    return None

examples/access_policies_filter.py

@@ -1,24 +1,35 @@
-"""Filter IAM access_policies of a archivist connection given url to Archivist and user Token.
+"""Filter IAM access_policies of a archivist connection given url to Archivist
+and user Token.
 
-Main function parses in a url to the Archivist and a token, which is a user authorization.
-The main function would initialize an archivist connection using the url and
-the token, called "arch", then call arch.access_policies.list() with suitable properties and
-attributes.
+Main function parses in a url to the Archivist and client credentials , which is
+a user authorization. The main function would initialize an archivist connection
+using the url and the credentials, called "arch", then call arch.access_policies.list()
+with suitable properties and attributes.
 
 """
+from os import getenv
 
 from archivist.archivist import Archivist
 
 
 def main():
     """Main function of filtering access_policies."""
-    with open(".auth_token", mode="r", encoding="utf-8") as tokenfile:
-        authtoken = tokenfile.read().strip()
+
+    # client id and client secret is obtained from the appidp endpoint - see the
+    # application registrations example code in examples/applications_registration.py
+    #
+    # client id is an environment variable. client_scret is stored in a file in a
+    # directory that has 0700 permissions. The location of this file is set in
+    # the client_secret_file environment variable.
+    client_id = getenv("ARCHIVIST_CLIENT_ID")
+    client_secret_file = getenv("ARCHIVIST_CLIENT_SECRET_FILE")
+    with open(client_secret_file, mode="r", encoding="utf-8") as tokenfile:
+        client_secret = tokenfile.read().strip()
 
     # Initialize connection to Archivist
     arch = Archivist(
         "https://app.rkvst.io",
-        authtoken,
+        (client_id, client_secret),
     )
 
     # count access_policies...

examples/access_policy_create.py

@@ -1,12 +1,14 @@
 """Create a IAM access_policy given url to Archivist and user Token.
 
 Main function parses in
-a url to the Archivist and a token, which is a user authorization.
+a url to the Archivist and client credentials, which is a user authorization.
 The main function would initialize an archivist connection using the url and
-the token, called "arch", then call arch.access_policys.create() and the access_policy
+the credentials, called "arch", then call arch.access_policies.create() and the access_policy
 will be created.
 """
 
+from os import getenv
+
 from archivist.archivist import Archivist
 
 
@@ -17,12 +19,20 @@ def main():
     create an example archivist connection and create an asset.
 
     """
-    with open(".auth_token", mode="r", encoding="utf-8") as tokenfile:
-        authtoken = tokenfile.read().strip()
+    # client id and client secret is obtained from the appidp endpoint - see the
+    # application registrations example code in examples/applications_registration.py
+    #
+    # client id is an environment variable. client_scret is stored in a file in a
+    # directory that has 0700 permissions. The location of this file is set in
+    # the client_secret_file environment variable.
+    client_id = getenv("ARCHIVIST_CLIENT_ID")
+    client_secret_file = getenv("ARCHIVIST_CLIENT_SECRET_FILE")
+    with open(client_secret_file, mode="r", encoding="utf-8") as tokenfile:
+        client_secret = tokenfile.read().strip()
 
     arch = Archivist(
         "https://app.rkvst.io",
-        authtoken,
+        (client_id, client_secret),
     )
 
     props = {