add gardenlinux.oci.registry.push_additional_tags_manifest

enables pushing additional tags to OCI manifest

Author: yeoldegrove

poetry.lock

@@ -8,7 +8,7 @@ readme = "README.md"
 packages = [{include = "gardenlinux", from="src"}, {include = "python_gardenlinux_lib", from="src"}]
 
 [tool.poetry.dependencies]
-python = "^3.10"
+python = "^3.13"
 networkx = "^3.3"
 PyYAML = "^6.0.2"
 pytest = "^8.3.2"
@@ -19,6 +19,9 @@ oras = { git  = "https://github.com/oras-project/oras-py.git", rev="caf8db5b2793
 python-dotenv = "^1.0.1"
 cryptography = "^44.0.0"
 boto3 = "*"
+click = "^8.2.0"
+pygments = "^2.19.1"
+opencontainers = "^0.0.14"
 
 [tool.poetry.group.dev.dependencies]
 bandit = "^1.8.3"

pyproject.toml

@@ -26,13 +26,6 @@ def cli():
     type=click.Path(),
     help="Version of image",
 )
-@click.option(
-    "--commit",
-    required=False,
-    type=click.Path(),
-    default=None,
-    help="Commit of image",
-)
 @click.option(
     "--arch",
     required=True,
@@ -58,16 +51,22 @@ def cli():
     default=False,
     help="Use HTTP to communicate with the registry",
 )
+@click.option(
+    "--additional_tag",
+    required=False,
+    multiple=True,
+    help="Additional tag to push the manifest with",
+)
 def push_manifest(
     container,
     version,
-    commit,
     arch,
     cname,
     directory,
     cosign_file,
     manifest_file,
     insecure,
+    additional_tag,
 ):
     """push artifacts from a dir to a registry, get the index-entry for the manifest in return"""
     container_name = f"{container}:{version}"
@@ -77,7 +76,7 @@ def push_manifest(
         insecure=insecure,
     )
     digest = registry.push_from_dir(
-        arch, version, cname, directory, manifest_file, commit=commit
+        arch, version, cname, directory, manifest_file, additional_tag
     )
     if cosign_file:
         print(digest, file=open(cosign_file, "w"))

src/gardenlinux/oci/__main__.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 
 import base64
+import configparser
 import copy
 import hashlib
 import json
@@ -28,9 +29,11 @@
 from oras.provider import Registry
 from oras.schemas import manifest as oras_manifest_schema
 
-from ..constants import OCI_ANNOTATION_SIGNATURE_KEY, OCI_ANNOTATION_SIGNED_STRING_KEY
+from ..constants import (
+    OCI_ANNOTATION_SIGNATURE_KEY,
+    OCI_ANNOTATION_SIGNED_STRING_KEY
+)
 from ..features import CName
-
 from .checksum import (
     calculate_sha256,
     verify_sha256,
@@ -673,38 +676,73 @@ def push_from_dir(
         cname: str,
         directory: str,
         manifest_file: str,
-        commit: Optional[str] = None,
+        additional_tags: list = None,
     ):
-        # Step 1 scan and extract nested artifacts:
-        for file in os.listdir(directory):
-            try:
-                if file.endswith(".pxe.tar.gz"):
-                    logger.info(f"Found nested artifact {file}")
-                    nested_tar_obj = tarfile.open(f"{directory}/{file}")
-                    nested_tar_obj.extractall(filter="data", path=directory)
-                    nested_tar_obj.close()
-            except (OSError, tarfile.FilterError, tarfile.TarError) as e:
-                print(f"Failed to extract nested artifact {file}", e)
-                exit(1)
+        """
+        Push artifacts from a directory to a registry
+
+        Args:
+            architecture: Target architecture of the image
+            version: Version tag for the image
+            cname: Canonical name of the image
+            directory: Directory containing the artifacts
+            manifest_file: File to write the manifest index entry to
+            additional_tags: Additional tags to push the manifest with
+
+        Returns:
+            The digest of the pushed manifest
+        """
+        if additional_tags is None:
+            additional_tags = []
 
         try:
+            # scan and extract nested artifacts
+            for file in os.listdir(directory):
+                try:
+                    if file.endswith(".pxe.tar.gz"):
+                        logger.info(f"Found nested artifact {file}")
+                        nested_tar_obj = tarfile.open(f"{directory}/{file}")
+                        nested_tar_obj.extractall(filter="data", path=directory)
+                        nested_tar_obj.close()
+                except (OSError, tarfile.FilterError, tarfile.TarError) as e:
+                    print(f"Failed to extract nested artifact {file}", e)
+                    exit(1)
+
+            # Get metadata from files
             oci_metadata = get_oci_metadata_from_fileset(
                 os.listdir(directory), architecture
             )
 
             features = ""
+            commit = ""
             for artifact in oci_metadata:
                 if artifact["media_type"] == "application/io.gardenlinux.release":
-                    file = open(f"{directory}/{artifact["file_name"]}", "r")
-                    lines = file.readlines()
-                    for line in lines:
-                        if line.strip().startswith("GARDENLINUX_FEATURES="):
-                            features = line.strip().removeprefix(
-                                "GARDENLINUX_FEATURES="
+                    try:
+                        file_path = f"{directory}/{artifact['file_name']}"
+
+                        config = configparser.ConfigParser(allow_unnamed_section=True)
+                        config.read(file_path)
+
+                        if config.has_option(
+                            configparser.UNNAMED_SECTION, "GARDENLINUX_FEATURES"
+                        ):
+                            features = config.get(
+                                configparser.UNNAMED_SECTION, "GARDENLINUX_FEATURES"
+                            )
+                        if config.has_option(
+                            configparser.UNNAMED_SECTION, "GARDENLINUX_COMMIT_ID"
+                        ):
+                            commit = config.get(
+                                configparser.UNNAMED_SECTION, "GARDENLINUX_COMMIT_ID"
                             )
-                            break
-                    file.close()
 
+                    except (configparser.Error, IOError) as e:
+                        logger.error(
+                            f"Error reading config file {artifact['file_name']}: {e}"
+                        )
+                    break
+
+            # Push the image manifest
             digest = self.push_image_manifest(
                 architecture,
                 cname,
@@ -715,7 +753,103 @@ def push_from_dir(
                 manifest_file,
                 commit=commit,
             )
+
+            # Process additional tags if provided
+            if additional_tags and len(additional_tags) > 0:
+                print(f"DEBUG: Processing {len(additional_tags)} additional tags")
+                logger.info(f"Processing {len(additional_tags)} additional tags")
+
+                self.push_additional_tags_manifest(
+                    architecture,
+                    cname,
+                    version,
+                    additional_tags,
+                    container=self.container,
+                )
+
+            return digest
         except Exception as e:
             print("Error: ", e)
             exit(1)
-        return digest
+
+    def push_additional_tags_manifest(
+        self, architecture, cname, version, additional_tags, container
+    ):
+        """
+        Push additional tags for an existing manifest using ORAS Registry methods
+
+        Args:
+            architecture: Target architecture of the image
+            cname: Canonical name of the image
+            version: Version tag for the image
+            additional_tags: List of additional tags to push
+            container: Container object
+        """
+        try:
+            # Source tag is the tag containing the version-cname-architecture combination
+            source_tag = f"{version}-{cname}-{architecture}"
+            source_container = copy.deepcopy(container)
+            source_container.tag = source_tag
+
+            # Authentication credentials from environment
+            token = os.getenv("GL_CLI_REGISTRY_TOKEN")
+            username = os.getenv("GL_CLI_REGISTRY_USERNAME")
+            password = os.getenv("GL_CLI_REGISTRY_PASSWORD")
+
+            # Login to registry if credentials are provided
+            if username and password:
+                logger.debug(f"Logging in with username/password")
+                try:
+                    self.login(username, password)
+                except Exception as login_error:
+                    logger.error(f"Login error: {str(login_error)}")
+            elif token:
+                # If token is provided, set it directly on the Registry instance
+                logger.debug(f"Using token authentication")
+                self.token = base64.b64encode(token.encode("utf-8")).decode("utf-8")
+                self.auth.set_token_auth(self.token)
+
+            # Get the manifest from the source container
+            try:
+                logger.debug(f"Getting manifest from {source_container}")
+                manifest = self.get_manifest(source_container)
+                if not manifest:
+                    logger.error(f"Failed to get manifest for {source_container}")
+                    return
+                logger.info(
+                    f"Successfully retrieved manifest: {manifest['mediaType'] if 'mediaType' in manifest else 'unknown'}"
+                )
+            except Exception as get_error:
+                logger.error(f"Error getting manifest: {str(get_error)}")
+                return
+
+            # For each additional tag, push the manifest using Registry.upload_manifest
+            for tag in additional_tags:
+                try:
+                    logger.debug(f"Pushing additional tag: {tag}")
+
+                    # Create a new container for this tag
+                    tag_container = copy.deepcopy(container)
+                    tag_container.tag = tag
+
+                    logger.debug(f"Pushing to container: {tag_container}")
+
+                    # Upload the manifest to the new tag
+                    response = self.upload_manifest(manifest, tag_container)
+
+                    if response and response.status_code in [200, 201]:
+                        logger.info(f"Successfully pushed tag {tag} for manifest")
+                    else:
+                        status_code = getattr(response, "status_code", "unknown")
+                        response_text = getattr(response, "text", "No response text")
+                        logger.error(
+                            f"Failed to push tag {tag} for manifest: {status_code}"
+                        )
+
+                except Exception as tag_error:
+                    logger.error(
+                        f"Error pushing tag {tag} for manifest: {str(tag_error)}"
+                    )
+
+        except Exception as e:
+            logger.error(f"Error in push_additional_tags_manifest: {str(e)}")