Add support for build metadata in version string

To be compatible with SemVer, we should be able to handle versions which
include optional build metadata
([see](https://semver.org/#spec-item-10)). In general, this is already
the case. However, (some) OCI registries don't allow `+` in their tags,
which is used as separator for these build metadata. That's why, the
version has to be sanitised prior to uploading to a registry or after
retrieving it from there.

ccc/oci.py

@@ -1,3 +1,4 @@
+import collections.abc
 import functools
 import logging
 import traceback
@@ -67,6 +68,8 @@ def oci_client(
     install_logging_handler: bool=True,
     cfg_factory=None,
     http_connection_pool_size:int=16,
+    tag_preprocessing_callback: collections.abc.Callable[[str], str]=None,
+    tag_postprocessing_callback: collections.abc.Callable[[str], str]=None,
 ) -> oc.Client:
     def base_api_lookup(image_reference):
         registry_cfg = model.container_registry.find_config(
@@ -107,6 +110,8 @@ def base_api_lookup(image_reference):
         credentials_lookup=credentials_lookup,
         routes=routes,
         session=session,
+        tag_preprocessing_callback=tag_preprocessing_callback,
+        tag_postprocessing_callback=tag_postprocessing_callback,
     )
 
 

cnudie/util.py

@@ -22,6 +22,8 @@
     | str
 )
 
+META_SEPARATOR = '.build-'
+
 
 def to_component_id(
     component: ComponentId, /
@@ -566,3 +568,24 @@ def enumerate_group_pairs(
             _add_if_not_duplicate(resource_diff.resource_refs_only_right, i)
 
     return resource_diff
+
+
+def sanitise_version(version: str) -> str:
+    '''
+    Additional build metadata as defined in SemVer can be added via `+` to the version. However,
+    OCI registries don't support `+` as tag character, which is why it has to be sanitised, for
+    example using `META_SEPARATOR`.
+    '''
+    sanitised_version = version.replace('+', META_SEPARATOR)
+
+    return sanitised_version
+
+
+def desanitise_version(version: str) -> str:
+    '''
+    This function reverts the sanitisation of the `sanitise_version` function, which allows
+    processing the version the same way as prior to using `sanitise_version`.
+    '''
+    desanitised_version = version.replace(META_SEPARATOR, '+')
+
+    return desanitised_version

oci/client.py

@@ -1,4 +1,5 @@
 import base64
+import collections.abc
 import dataclasses
 import datetime
 import enum
@@ -220,12 +221,19 @@ def blob_url(self, image_reference: typing.Union[str, om.OciImageReference], dig
             digest
         )
 
-    def manifest_url(self, image_reference: typing.Union[str, om.OciImageReference]) -> str:
+    def manifest_url(
+        self,
+        image_reference: typing.Union[str, om.OciImageReference],
+        tag_preprocessing_callback: collections.abc.Callable[[str], str]=None,
+    ) -> str:
         image_reference = om.OciImageReference.to_image_ref(image_reference)
 
         if not (tag := image_reference.tag):
             raise ValueError(f'{image_reference=} does not seem to contain a tag')
 
+        if tag_preprocessing_callback:
+            tag = tag_preprocessing_callback(tag)
+
         return urljoin(
             self.artifact_base_url(image_reference=image_reference),
             'manifests',
@@ -248,10 +256,25 @@ def __init__(
         self,
         credentials_lookup: typing.Callable,
         routes: OciRoutes=OciRoutes(),
-        disable_tls_validation=False,
+        disable_tls_validation: bool=False,
         timeout_seconds: int=None,
         session: requests.Session=None,
+        tag_preprocessing_callback: collections.abc.Callable[[str], str]=None,
+        tag_postprocessing_callback: collections.abc.Callable[[str], str]=None,
     ):
+        '''
+        @param credentials_lookup <Callable>
+        @param routes <OciRoutes>
+        @param disable_tls_validation <bool>
+        @param timeout_seconds <int>
+        @param session <Session>
+        @param tag_preprocessing_callback <Callable>
+            callback which is instrumented _prior_ to interacting with the OCI registry, i.e. useful
+            in case the tag has to be sanitised so it is accepted by the OCI registry
+        @param tag_postprocessing_callback <Callable>
+            callback which is instrumented _after_ interacting with the OCI registry, i.e. useful to
+            revert required sanitisation of `tag_preprocessing_callback`
+        '''
         self.credentials_lookup = credentials_lookup
         self.token_cache = OauthTokenCache()
         if not session:
@@ -260,6 +283,8 @@ def __init__(
             self.session = session
         self.routes = routes
         self.disable_tls_validation = disable_tls_validation
+        self.tag_preprocessing_callback = tag_preprocessing_callback
+        self.tag_postprocessing_callback = tag_postprocessing_callback
 
         if timeout_seconds:
             timeout_seconds = int(timeout_seconds)
@@ -510,7 +535,10 @@ def manifest_raw(
 
         try:
             res = self._request(
-                url=self.routes.manifest_url(image_reference=image_reference),
+                url=self.routes.manifest_url(
+                    image_reference=image_reference,
+                    tag_preprocessing_callback=self.tag_preprocessing_callback,
+                ),
                 image_reference=image_reference,
                 scope=scope,
                 warn_if_not_ok=not absent_ok,
@@ -647,7 +675,10 @@ def head_manifest(
             accept = om.MimeTypes.single_image
 
         res = self._request(
-            url=self.routes.manifest_url(image_reference=image_reference),
+            url=self.routes.manifest_url(
+                image_reference=image_reference,
+                tag_preprocessing_callback=self.tag_preprocessing_callback,
+            ),
             image_reference=image_reference,
             method='HEAD',
             headers={
@@ -701,7 +732,15 @@ def tags(self, image_reference: str):
             method='GET'
         )
 
-        return res.json()['tags']
+        tags = res.json()['tags']
+
+        if self.tag_postprocessing_callback:
+            tags = [
+                self.tag_postprocessing_callback(tag)
+                for tag in tags
+            ]
+
+        return tags
 
     def has_multiarch(self, image_reference: str) -> bool:
         res = self.head_manifest(
@@ -734,7 +773,10 @@ def put_manifest(
         logger.debug(f'manifest-mimetype: {content_type=}')
 
         res = self._request(
-            url=self.routes.manifest_url(image_reference=image_reference),
+            url=self.routes.manifest_url(
+                image_reference=image_reference,
+                tag_preprocessing_callback=self.tag_preprocessing_callback,
+            ),
             image_reference=image_reference,
             scope=scope,
             method='PUT',
@@ -782,7 +824,10 @@ def delete_manifest(
                 headers = {}
 
             return self._request(
-                url=self.routes.manifest_url(image_reference=image_reference),
+                url=self.routes.manifest_url(
+                    image_reference=image_reference,
+                    tag_preprocessing_callback=self.tag_preprocessing_callback,
+                ),
                 image_reference=image_reference,
                 scope=scope,
                 headers=headers,