Add the type field to the Manifest model

closes: #1751

CHANGES/1751.deprecation

@@ -0,0 +1 @@
+Deprecated the Manifest `is_flatpak` and `is_bootable` fields in favor of the new `type` field.

CHANGES/1751.feature

@@ -0,0 +1,2 @@
+Introduced the `type` field on the Manifests endpoint to enable easier differentiation of image
+types.

pulp_container/app/management/commands/container-handle-image-data.py

@@ -39,10 +39,14 @@ class Command(BaseCommand):
     def handle(self, *args, **options):
         manifests_updated_count = 0
 
-        manifests_v1 = Manifest.objects.filter(data__isnull=True, media_type=MEDIA_TYPE.MANIFEST_V1)
+        manifests_v1 = Manifest.objects.filter(
+            Q(media_type=MEDIA_TYPE.MANIFEST_V1), Q(data__isnull=True) | Q(type__isnull=True)
+        )
         manifests_updated_count += self.update_manifests(manifests_v1)
 
-        manifests_v2 = Manifest.objects.filter(Q(data__isnull=True) | Q(annotations={}, labels={}))
+        manifests_v2 = Manifest.objects.filter(
+            Q(data__isnull=True) | Q(annotations={}, labels={}) | Q(type__isnull=True)
+        )
         manifests_v2 = manifests_v2.exclude(
             media_type__in=[MEDIA_TYPE.MANIFEST_LIST, MEDIA_TYPE.INDEX_OCI, MEDIA_TYPE.MANIFEST_V1]
         )
@@ -68,6 +72,15 @@ def handle(self, *args, **options):
     def update_manifests(self, manifests_qs):
         manifests_updated_count = 0
         manifests_to_update = []
+        fields_to_update = [
+            "annotations",
+            "labels",
+            "is_bootable",
+            "is_flatpak",
+            "data",
+            "type",
+        ]
+
         for manifest in manifests_qs.iterator():
             # suppress non-existing/already migrated artifacts and corrupted JSON files
             with suppress(ObjectDoesNotExist, JSONDecodeError):
@@ -76,7 +89,6 @@ def update_manifests(self, manifests_qs):
                     manifests_to_update.append(manifest)
 
             if len(manifests_to_update) > 1000:
-                fields_to_update = ["annotations", "labels", "is_bootable", "is_flatpak", "data"]
                 manifests_qs.model.objects.bulk_update(
                     manifests_to_update,
                     fields_to_update,
@@ -85,7 +97,6 @@ def update_manifests(self, manifests_qs):
                 manifests_to_update.clear()
 
         if manifests_to_update:
-            fields_to_update = ["annotations", "labels", "is_bootable", "is_flatpak", "data"]
             manifests_qs.model.objects.bulk_update(
                 manifests_to_update,
                 fields_to_update,
@@ -100,11 +111,12 @@ def init_manifest(self, manifest):
             manifest_data, raw_bytes_data = get_content_data(manifest_artifact)
             manifest.data = raw_bytes_data.decode("utf-8")
 
-            if not (manifest.annotations or manifest.labels):
+            if not (manifest.annotations or manifest.labels or manifest.type):
                 manifest.init_metadata(manifest_data)
 
             manifest._artifacts.clear()
-
             return True
 
+        elif not manifest.type:
+            return manifest.init_image_nature()
         return False

pulp_container/app/migrations/0042_add_manifest_nature_field.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.16 on 2024-10-21 19:14
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('container', '0041_add_pull_through_pull_permissions'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='manifest',
+            name='type',
+            field=models.CharField(null=True),
+        ),
+    ]

pulp_container/app/models.py

@@ -32,7 +32,13 @@
 
 from . import downloaders
 from pulp_container.app.utils import get_content_data
-from pulp_container.constants import MEDIA_TYPE, SIGNATURE_TYPE
+from pulp_container.constants import (
+    COSIGN_MEDIA_TYPES,
+    MANIFEST_MEDIA_TYPES,
+    MANIFEST_TYPE,
+    MEDIA_TYPE,
+    SIGNATURE_TYPE,
+)
 
 
 logger = getLogger(__name__)
@@ -72,6 +78,7 @@ class Manifest(Content):
         digest (models.TextField): The manifest digest.
         schema_version (models.IntegerField): The manifest schema version.
         media_type (models.TextField): The manifest media type.
+        type (models.TextField): The manifest's type (flatpak, bootable, signature, etc.).
         data (models.TextField): The manifest's data in text format.
         annotations (models.JSONField): Metadata stored inside the image manifest.
         labels (models.JSONField): Metadata stored inside the image configuration.
@@ -99,12 +106,15 @@ class Manifest(Content):
     digest = models.TextField(db_index=True)
     schema_version = models.IntegerField()
     media_type = models.TextField(choices=MANIFEST_CHOICES)
+    type = models.CharField(null=True)
     data = models.TextField(null=True)
 
     annotations = models.JSONField(default=dict)
     labels = models.JSONField(default=dict)
 
+    # DEPRECATED: this field is deprecated and will be removed in a future release.
     is_bootable = models.BooleanField(default=False)
+    # DEPRECATED: this field is deprecated and will be removed in a future release.
     is_flatpak = models.BooleanField(default=False)
 
     blobs = models.ManyToManyField(Blob, through="BlobManifest")
@@ -154,40 +164,93 @@ def init_image_nature(self):
             return self.init_manifest_nature()
 
     def init_manifest_list_nature(self):
+        updated_type = False
+        if not self.type:
+            self.type = MANIFEST_TYPE.UNKNOWN
+            updated_type = True
+
         for manifest in self.listed_manifests.all():
             # it suffices just to have a single manifest of a specific nature;
             # there is no case where the manifest is both bootable and flatpak-based
-            if manifest.is_bootable:
+            if manifest.type == MANIFEST_TYPE.BOOTABLE:
+                self.type = MANIFEST_TYPE.BOOTABLE
                 self.is_bootable = True
                 return True
-            elif manifest.is_flatpak:
+            elif manifest.type == MANIFEST_TYPE.FLATPAK:
+                self.type = MANIFEST_TYPE.FLATPAK
                 self.is_flatpak = True
                 return True
 
-        return False
+        return updated_type
 
     def init_manifest_nature(self):
         if self.is_bootable_image():
+            # DEPRECATED: is_bootable is deprecated and will be removed in a future release.
             self.is_bootable = True
+            self.type = MANIFEST_TYPE.BOOTABLE
             return True
         elif self.is_flatpak_image():
+            # DEPRECATED: is_flatpak is deprecated and will be removed in a future release.
             self.is_flatpak = True
+            self.type = MANIFEST_TYPE.FLATPAK
             return True
-        else:
-            return False
+        elif self.is_helm_chart():
+            self.type = MANIFEST_TYPE.HELM
+            return True
+        elif media_type := self.is_cosign():
+            self.type = self.get_cosign_type(media_type)
+            return True
+        elif self.is_manifest_image():
+            self.type = MANIFEST_TYPE.IMAGE
+            return True
+
+        return False
 
     def is_bootable_image(self):
-        if (
+        return (
             self.annotations.get("containers.bootc") == "1"
             or self.labels.get("containers.bootc") == "1"
-        ):
-            return True
-        else:
-            return False
+        )
 
     def is_flatpak_image(self):
         return True if self.labels.get("org.flatpak.ref") else False
 
+    def is_manifest_image(self):
+        return self.media_type in MANIFEST_MEDIA_TYPES.IMAGE
+
+    @property
+    def json_manifest(self):
+        return json.loads(self.data)
+
+    def is_cosign(self):
+        try:
+            # layers is not a mandatory field
+            layers = self.json_manifest["layers"]
+        except KeyError:
+            return False
+
+        for layer in layers:
+            if layer["mediaType"] in COSIGN_MEDIA_TYPES:
+                return layer["mediaType"]
+        return False
+
+    def get_cosign_type(self, media_type):
+        if media_type in MEDIA_TYPE.COSIGN_SBOM:
+            return MANIFEST_TYPE.COSIGN_SBOM
+
+        cosign_types_mapping = {
+            MEDIA_TYPE.COSIGN_BLOB: MANIFEST_TYPE.COSIGN_SIGNATURE,
+            MEDIA_TYPE.COSIGN_ATTESTATION: MANIFEST_TYPE.COSIGN_ATTESTATION,
+            MEDIA_TYPE.COSIGN_ATTESTATION_BUNDLE: MANIFEST_TYPE.COSIGN_ATTESTATION_BUNDLE,
+        }
+        return cosign_types_mapping.get(media_type)
+
+    def is_helm_chart(self):
+        try:
+            return self.json_manifest["config"]["mediaType"] == MEDIA_TYPE.CONFIG_BLOB_HELM
+        except KeyError:
+            return False
+
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"
         unique_together = ("digest",)

pulp_container/app/registry_api.py

@@ -90,6 +90,7 @@
 )
 from pulp_container.constants import (
     EMPTY_BLOB,
+    MANIFEST_TYPE,
     SIGNATURE_API_EXTENSION_VERSION,
     SIGNATURE_HEADER,
     SIGNATURE_PAYLOAD_MAX_SIZE,
@@ -1213,6 +1214,11 @@ def put(self, request, path, pk=None):
                 ManifestInvalid(digest=manifest_digest)
 
             manifest_list = self._init_manifest(manifest_digest, media_type, raw_text_data)
+            manifest_list.type = MANIFEST_TYPE.UNKNOWN
+            for manifest in found_manifests:
+                if manifest.type in [MANIFEST_TYPE.BOOTABLE, MANIFEST_TYPE.FLATPAK]:
+                    manifest_list.type = manifest.type
+                    break
             manifest_list = self._save_manifest(manifest_list)
 
             manifests_to_list = []
@@ -1235,6 +1241,9 @@ def put(self, request, path, pk=None):
             )
             manifest = manifest_list
 
+            # DEPRECATED: is_bootable and is_flatpak are deprecated and will be removed in a
+            #             future release. Keeping this block for now to avoid introducing a bug or
+            #             a regression.
             # once relations for listed manifests are established, it is
             # possible to initialize the nature of the manifest list
             if manifest.init_manifest_list_nature():

pulp_container/app/serializers.py

@@ -64,6 +64,11 @@ class ManifestSerializer(NoArtifactContentSerializer):
     digest = serializers.CharField(help_text="sha256 of the Manifest file")
     schema_version = serializers.IntegerField(help_text="Manifest schema version")
     media_type = serializers.CharField(help_text="Manifest media type of the file")
+    type = serializers.CharField(
+        help_text="Manifest type (flatpak, bootable, signature, etc.).",
+        required=False,
+        default=None,
+    )
     listed_manifests = DetailRelatedField(
         many=True,
         help_text="Manifests that are referenced by this Manifest List",
@@ -93,15 +98,23 @@ class ManifestSerializer(NoArtifactContentSerializer):
         help_text=_("Property describing metadata stored inside the image configuration"),
     )
 
+    # DEPRECATED: this field is deprecated and will be removed in a future release.
     is_bootable = serializers.BooleanField(
         required=False,
         default=False,
-        help_text=_("A boolean determining whether users can boot from an image or not."),
+        help_text=_(
+            "A boolean determining whether users can boot from an image or not."
+            "[deprecated] check type field instead"
+        ),
     )
+    # DEPRECATED: this field is deprecated and will be removed in a future release.
     is_flatpak = serializers.BooleanField(
         required=False,
         default=False,
-        help_text=_("A boolean determining whether the image bundles a Flatpak application"),
+        help_text=_(
+            "A boolean determining whether the image bundles a Flatpak application."
+            "[deprecated] check type field instead"
+        ),
     )
 
     class Meta:
@@ -116,6 +129,7 @@ class Meta:
             "labels",
             "is_bootable",
             "is_flatpak",
+            "type",
         )
         model = models.Manifest
 

pulp_container/app/tasks/builder.py

@@ -83,6 +83,7 @@ def add_image_from_directory_to_repository(path, repository, tag):
 
     with repository.new_version() as new_repo_version:
         manifest_json = json.loads(manifest_text_data)
+        manifest.init_metadata(manifest_json)
 
         config_blob = get_or_create_blob(manifest_json["config"], manifest, path)
         manifest.config_blob = config_blob

pulp_container/app/tasks/sync_stages.py

@@ -12,6 +12,7 @@
 from pulpcore.plugin.stages import DeclarativeArtifact, DeclarativeContent, Stage, ContentSaver
 
 from pulp_container.constants import (
+    MANIFEST_TYPE,
     MEDIA_TYPE,
     SIGNATURE_API_EXTENSION_VERSION,
     SIGNATURE_HEADER,
@@ -287,9 +288,21 @@ async def resolve_flush(self):
         self.manifest_dcs.clear()
 
         for manifest_list_dc in self.manifest_list_dcs:
+            manifest_list_type = MANIFEST_TYPE.UNKNOWN
             for listed_manifest in manifest_list_dc.extra_data["listed_manifests"]:
                 # Just await here. They will be associated in the post_save hook.
                 await listed_manifest["manifest_dc"].resolution()
+
+                # if manifestlist type is not defined as "unknown" we already checked its nature,
+                # in this case, there is no need to verify the other manifests from list
+                if manifest_list_type != MANIFEST_TYPE.UNKNOWN and listed_manifest[
+                    "manifest_dc"
+                ].content.type in [
+                    MANIFEST_TYPE.BOOTABLE,
+                    MANIFEST_TYPE.FLATPAK,
+                ]:
+                    manifest_list_type = listed_manifest["manifest_dc"].content.type
+            manifest_list_dc.content.type = manifest_list_type
             await self.put(manifest_list_dc)
         self.manifest_list_dcs.clear()
 
@@ -392,6 +405,7 @@ def create_manifest(self, manifest_data, raw_text_data, media_type, digest=None)
             annotations=manifest_data.get("annotations", {}),
         )
 
+        manifest.init_manifest_nature()
         manifest_dc = DeclarativeContent(content=manifest)
         return manifest_dc
 
@@ -644,6 +658,8 @@ def _post_save(self, batch):
         if manifest_list_manifests:
             ManifestListManifest.objects.bulk_create(manifest_list_manifests, ignore_conflicts=True)
 
+        # DEPRECATED: is_bootable/is_flatpak are deprecated and will be removed in a future release.
+        # keeping this block for now to avoid introducing a bug or a regression
         # after creating the relation between listed manifests and manifest lists,
         # it is possible to initialize the nature of the corresponding manifest lists
         for ml in manifest_lists: