Get mediaType from the request's content-type

CHANGES/883.bugfix

@@ -0,0 +1 @@
+Fixed an error that was raised when an OCI manifest did not contain ``mediaType``.

pulp_container/app/tasks/sync_stages.py

@@ -30,7 +30,11 @@
     ManifestSignature,
     Tag,
 )
-from pulp_container.app.utils import extract_data_from_signature, urlpath_sanitize
+from pulp_container.app.utils import (
+    determine_media_type,
+    extract_data_from_signature,
+    urlpath_sanitize,
+)
 
 log = logging.getLogger(__name__)
 
@@ -160,10 +164,10 @@ async def get_signature_source(self):
                 tag_dc = DeclarativeContent(Tag(name=tag_name))
 
                 content_data = json.loads(raw_data)
-                media_type = content_data.get("mediaType")
+                media_type = determine_media_type(content_data, tag)
                 if media_type in (MEDIA_TYPE.MANIFEST_LIST, MEDIA_TYPE.INDEX_OCI):
                     list_dc = self.create_tagged_manifest_list(
-                        tag_name, saved_artifact, content_data
+                        tag_name, saved_artifact, content_data, media_type
                     )
                     for manifest_data in content_data.get("manifests"):
                         man_dc = self.create_manifest(list_dc, manifest_data)
@@ -199,7 +203,7 @@ async def get_signature_source(self):
 
                 else:
                     man_dc = self.create_tagged_manifest(
-                        tag_name, saved_artifact, content_data, raw_data
+                        tag_name, saved_artifact, content_data, raw_data, media_type
                     )
                     if signature_source is not None:
                         man_sig_dcs = await self.create_signatures(man_dc, signature_source)
@@ -286,28 +290,29 @@ async def handle_blobs(self, manifest_dc, content_data):
             blob_dc.extra_data["config_relation"] = manifest_dc
             await self.put(blob_dc)
 
-    def create_tagged_manifest_list(self, tag_name, saved_artifact, manifest_list_data):
+    def create_tagged_manifest_list(self, tag_name, saved_artifact, manifest_list_data, media_type):
         """
         Create a ManifestList.
 
         Args:
             tag_name (str): A name of a tag
             saved_artifact (pulpcore.plugin.models.Artifact): A saved manifest's Artifact
             manifest_list_data (dict): Data about a ManifestList
+            media_type (str): The type of a manifest
 
         """
         digest = f"sha256:{saved_artifact.sha256}"
         manifest_list = Manifest(
             digest=digest,
             schema_version=manifest_list_data["schemaVersion"],
-            media_type=manifest_list_data["mediaType"],
+            media_type=media_type,
         )
 
         return self._create_manifest_declarative_content(
             manifest_list, saved_artifact, tag_name, digest
         )
 
-    def create_tagged_manifest(self, tag_name, saved_artifact, manifest_data, raw_data):
+    def create_tagged_manifest(self, tag_name, saved_artifact, manifest_data, raw_data, media_type):
         """
         Create an Image Manifest.
 
@@ -316,9 +321,9 @@ def create_tagged_manifest(self, tag_name, saved_artifact, manifest_data, raw_da
             saved_artifact (pulpcore.plugin.models.Artifact): A saved manifest's Artifact
             manifest_data (dict): Data about a single new ImageManifest.
             raw_data: (str): The raw JSON representation of the ImageManifest.
+            media_type (str): The type of a manifest
 
         """
-        media_type = manifest_data.get("mediaType", MEDIA_TYPE.MANIFEST_V1)
         if media_type in (MEDIA_TYPE.MANIFEST_V2, MEDIA_TYPE.MANIFEST_OCI):
             digest = f"sha256:{saved_artifact.sha256}"
         else:

pulp_container/app/utils.py

@@ -9,7 +9,7 @@
 
 from pulpcore.plugin.models import Task
 
-from pulp_container.constants import SIGNATURE_TYPE
+from pulp_container.constants import MEDIA_TYPE, SIGNATURE_TYPE
 
 
 gpg = gnupg.GPG()
@@ -131,3 +131,27 @@ def has_task_completed(dispatched_task):
             task.delete()
             raise Exception(str(error))
     raise Throttled()
+
+
+def determine_media_type(content_data, response):
+    """Determine the media type of a manifest either from the JSON data or the response object."""
+    if media_type := content_data.get("mediaType"):
+        return media_type
+    elif media_type := response.headers.get("content-type"):
+        return media_type
+    elif manifests := content_data.get("manifests"):
+        if len(manifests):
+            if manifests[0].get("mediaType") in (MEDIA_TYPE.MANIFEST_V2, MEDIA_TYPE.MANIFEST_V1):
+                return MEDIA_TYPE.MANIFEST_LIST
+            elif manifests[0].get("mediaType") in (MEDIA_TYPE.MANIFEST_OCI, MEDIA_TYPE.INDEX_OCI):
+                return MEDIA_TYPE.INDEX_OCI
+        return MEDIA_TYPE.MANIFEST_LIST
+    else:
+        if config := content_data.get("config"):
+            config_media_type = config.get("mediaType")
+            if config_media_type == MEDIA_TYPE.CONFIG_BLOB_OCI:
+                return MEDIA_TYPE.MANIFEST_OCI
+            else:
+                return MEDIA_TYPE.MANIFEST_V2
+        else:
+            return MEDIA_TYPE.MANIFEST_V1