adding support for OCI formatted images

src/confcom/HISTORY.rst

@@ -2,6 +2,14 @@
 
 Release History
 ===============
+1.2.2
+++++++
+* support for pure OCI v1 schema 2 formatted images
+* adding debug logging
+* changing where parameters and variables are filled in for arm templates
+* updating documentation about fragments
+* bugfix for exec processes in fragment generation
+* bugfix for custom mount options in fragment generation
 
 1.2.1
 ++++++

src/confcom/azext_confcom/README.md

@@ -194,6 +194,8 @@ Use the following command to generate CCE policy for the image.
 az confcom acipolicygen -a .\sample-template-input.json --tar .\file.tar
 ```
 
+Note that multiple images saved to the tar file is only available using the docker-archive format for tar files. OCI does not support multi-image tar files at this time.
+
 Example 12: If it is necessary to put images in their own tarballs, an external file can be used that maps images to their respective tarball paths. See the following example:
 
 ```bash

src/confcom/azext_confcom/cose_proxy.py

@@ -129,7 +129,7 @@ def cose_sign(
 
         if iss:
             arg_list.extend(["-issuer", iss])
-        logger.info(f"Signing the policy fragment: {out_path}")
+        logger.info("Signing the policy fragment: %s", out_path)
         call_cose_sign_tool(arg_list, "Error signing the policy fragment")
         return True
 
@@ -151,7 +151,7 @@ def generate_import_from_path(self, fragment_path: str, minimum_svn: int) -> str
         policy_bin_str = str(self.policy_bin)
 
         arg_list_chain = [policy_bin_str, "check", "--in", fragment_path, "--verbose"]
-        logger.info(f"Extracting import statement from signed fragment: {fragment_path}")
+        logger.info("Extracting import statement from signed fragment: %s", fragment_path)
         item = call_cose_sign_tool(arg_list_chain, "Error getting information from signed fragment file")
 
         stdout = item.stdout.decode("utf-8")
@@ -183,7 +183,7 @@ def extract_payload_from_path(self, fragment_path: str) -> str:
             eprint(f"The fragment file at {fragment_path} does not exist")
 
         arg_list_chain = [policy_bin_str, "check", "--in", fragment_path, "--verbose"]
-        logger.info(f"Extracting payload from signed fragment: {fragment_path}")
+        logger.info("Extracting payload from signed fragment: %s", fragment_path)
         item = call_cose_sign_tool(arg_list_chain, "Error getting information from signed fragment file")
 
         stdout = item.stdout.decode("utf-8")
@@ -195,7 +195,7 @@ def extract_feed_from_path(self, fragment_path: str) -> str:
             eprint(f"The fragment file at {fragment_path} does not exist")
 
         arg_list_chain = [policy_bin_str, "check", "--in", fragment_path, "--verbose"]
-        logger.info(f"Extracting feed from signed fragment: {fragment_path}")
+        logger.info("Extracting feed from signed fragment: %s", fragment_path)
         item = call_cose_sign_tool(arg_list_chain, "Error getting information from signed fragment file")
 
         stdout = item.stdout.decode("utf-8")

src/confcom/azext_confcom/data/internal_config.json

@@ -1,5 +1,5 @@
 {
-    "version": "1.2.1",
+    "version": "1.2.2",
     "hcsshim_config": {
         "maxVersion": "1.0.0",
         "minVersion": "0.0.1"

src/confcom/azext_confcom/oras_proxy.py

@@ -7,6 +7,7 @@
 import json
 import platform
 import re
+from knack.log import get_logger
 from typing import List
 from azext_confcom.errors import eprint
 from azext_confcom.config import ARTIFACT_TYPE
@@ -16,8 +17,6 @@
 host_os = platform.system()
 machine = platform.machine()
 
-from knack.log import get_logger
-
 logger = get_logger(__name__)
 
 
@@ -43,12 +42,13 @@ def prepend_docker_registry(image_name: str) -> str:
         # If no registry is specified, assume docker.io/library
         if "/" not in name:
             # Add the `library` namespace for official images
-            registry = f"library/"
+            registry = "library/"
         # Add the default `docker.io` registry
-        registry = f"docker.io/" + registry
+        registry = f"docker.io/{registry}"
 
     return f"{registry}{image_name}"
 
+
 def call_oras_cli(args, check=False):
     return subprocess.run(args, check=check, capture_output=True, timeout=120)
 
@@ -65,7 +65,7 @@ def discover(
     item = call_oras_cli(arg_list, check=False)
     hashes = []
 
-    logger.info(f"Discovering fragments for {image}: {item.stdout.decode('utf-8')}")
+    logger.info("Discovering fragments for %s: %s", image, item.stdout.decode('utf-8'))
     if item.returncode == 0:
         json_output = json.loads(item.stdout.decode("utf-8"))
         manifests = json_output.get("manifests", [])
@@ -91,7 +91,7 @@ def pull(
     if "@sha256:" in image:
         image = image.split("@")[0]
     arg_list = ["oras", "pull", f"{image}@{image_hash}"]
-    logger.info(f"Pulling fragment: {image}@{image_hash}")
+    logger.info("Pulling fragment: %s@%s", image, image_hash)
     item = call_oras_cli(arg_list, check=False)
 
     # get the exit code from the subprocess

src/confcom/azext_confcom/os_util.py

@@ -11,6 +11,7 @@
 import shutil
 import json
 import os
+import stat
 from tarfile import TarFile
 from azext_confcom.errors import (
     eprint,
@@ -170,7 +171,7 @@ def map_image_from_tar_backwards_compatibility(image_name: str, tar: TarFile, ta
                 ][0]
                 break
         # remove the extracted manifest file to clean up
-        os.remove(manifest_path)
+        force_delete_silently(manifest_path)
     else:
         eprint(f"Tarball at {tar_location} contains no images")
 
@@ -182,14 +183,64 @@ def map_image_from_tar_backwards_compatibility(image_name: str, tar: TarFile, ta
     image_info_file_path = os.path.join(tar_dir, info_file.name)
     image_info_raw = load_json_from_file(image_info_file_path)
     # delete the extracted json file to clean up
-    os.remove(image_info_file_path)
+    force_delete_silently(image_info_file_path)
     image_info = image_info_raw.get("config")
     # importing the constant from config.py gives a circular dependency error
     image_info["Architecture"] = image_info_raw.get("architecture")
 
+    shutil.rmtree("blobs", ignore_errors=True)
     return image_info
 
 
+def get_oci_image_name(image_name: str) -> str:
+    if "/" not in image_name:
+        return f"docker.io/library/{image_name}"
+    return image_name
+
+
+def read_file_from_tar(tar: TarFile, filename: str) -> str:
+    try:
+        return tar.extractfile(filename).read()
+    except KeyError:
+        eprint(f"'{filename}' not found in tar file")
+
+
+def map_image_from_tar_oci_layout_v1(image_name: str, tar: TarFile, tar_location: str):
+    # since this uses containerd naming, we need to append the docker.io path
+    oci_image_name = get_oci_image_name(image_name)
+
+    index_bytes = read_file_from_tar(tar, "index.json")
+    index = load_json_from_str(index_bytes)
+
+    manifests = index.get("manifests") or []
+    for manifest in manifests:
+        image_annotations = manifest.get("annotations")
+        image_name_annotation = ""
+        if image_annotations:
+            image_name_annotation = image_annotations.get("io.containerd.image.name")
+            if image_name_annotation and image_name_annotation != oci_image_name:
+                continue
+        if (
+            manifest.get("mediaType") in
+            ["application/vnd.docker.distribution.manifest.v2+json", "application/vnd.oci.image.manifest.v1+json"]
+        ):
+            hashing_algo, manifest_name = manifest.get("digest").split(":")
+            manifest_location = f"blobs/{hashing_algo}/{manifest_name}"
+
+            nested_manifest_bytes = tar.extractfile(manifest_location).read()
+            nested_manifest = load_json_from_str(nested_manifest_bytes)
+            config = nested_manifest.get("config")
+
+            config_hashing_algo, config_digest = config.get("digest").split(":")
+            config_location = f"blobs/{config_hashing_algo}/{config_digest}"
+            image_info_raw_bytes = tar.extractfile(config_location).read()
+            image_info_raw = load_json_from_str(image_info_raw_bytes)
+            image_info = image_info_raw.get("config")
+            image_info["Architecture"] = image_info_raw.get("architecture")
+            return image_info
+    eprint(f"Image '{image_name}' is not found in '{tar_location}'")
+
+
 def map_image_from_tar(image_name: str, tar: TarFile, tar_location: str):
     tar_dir = os.path.dirname(tar_location)
     info_file = None
@@ -209,7 +260,7 @@ def map_image_from_tar(image_name: str, tar: TarFile, tar_location: str):
                 break
     finally:
         # remove the extracted manifest file to clean up
-        os.remove(manifest_path)
+        force_delete_silently(manifest_path)
 
     if not info_file:
         return None
@@ -219,14 +270,27 @@ def map_image_from_tar(image_name: str, tar: TarFile, tar_location: str):
     image_info_file_path = os.path.join(tar_dir, info_file)
     image_info_raw = load_json_from_file(image_info_file_path)
     # delete the extracted json file to clean up
-    os.remove(image_info_file_path)
+    force_delete_silently(image_info_file_path)
     image_info = image_info_raw.get("config")
     # importing the constant from config.py gives a circular dependency error
     image_info["Architecture"] = image_info_raw.get("architecture")
 
     return image_info
 
 
+# sometimes image tarfiles have readonly members. this will try to change their permissions and delete them
+def force_delete_silently(filename: str) -> None:
+    try:
+        os.chmod(filename, stat.S_IWRITE)
+    except FileNotFoundError:
+        pass
+    except PermissionError:
+        eprint(f"Permission denied to edit file: {filename}")
+    except OSError as e:
+        eprint(f"Error editing file: {filename}, {e}")
+    delete_silently(filename)
+
+
 # helper function to delete a file that may or may not exist
 def delete_silently(filename: str) -> None:
     try:

src/confcom/azext_confcom/template_util.py

@@ -104,16 +104,22 @@ def get_image_info(progress, message_queue, tar_mapping, image):
         tar_location = get_tar_location_from_mapping(tar_mapping, image_name)
         # if we have a tar location, we can try to get the image info
         if tar_location:
-            with tarfile.open(tar_location) as tar:
+            with tarfile.open(tar_location) as tar_file:
                 # get all the info out of the tarfile
                 try:
+                    logger.info("using backwards compatibility tar file")
                     image_info = os_util.map_image_from_tar_backwards_compatibility(
-                        image_name, tar, tar_location
+                        image_name, tar_file, tar_location
                     )
                 except IndexError:
+                    logger.info("using docker formatted tar file")
                     image_info = os_util.map_image_from_tar(
-                        image_name, tar, tar_location
+                        image_name, tar_file, tar_location
                     )
+                except (KeyError, AttributeError):
+                    # manifest.json not found
+                    logger.info("using OCI tar file")
+                    image_info = os_util.map_image_from_tar_oci_layout_v1(image_name, tar_file, tar_location)
                 if image_info is not None:
                     tar = True
                     message_queue.append(f"{image_name} read from local tar file")
@@ -177,6 +183,7 @@ def get_image_info(progress, message_queue, tar_mapping, image):
             f"{raw_image.attrs.get(config.ACI_FIELD_CONTAINERS_ARCHITECTURE_KEY)}. "
             f"Only {config.ACI_FIELD_CONTAINERS_ARCHITECTURE_VALUE} is supported by Confidential ACI"
         ))
+
     return image_info, tar
 
 
@@ -231,7 +238,7 @@ def process_env_vars_from_template(params: dict,
                     response = approve_wildcards or input(
                         f'Create a wildcard policy for the environment variable {name} (y/n): ')
                     if approve_wildcards or response.lower() == 'y':
-                        logger.info(f'Creating a wildcard policy for the environment variable {name}')
+                        logger.info('Creating a wildcard policy for the environment variable %s', name)
                         env_vars.append({
                             config.ACI_FIELD_CONTAINERS_ENVS_NAME: name,
                             config.ACI_FIELD_CONTAINERS_ENVS_VALUE: ".*",

src/confcom/azext_confcom/tests/latest/test_confcom_scenario.py

@@ -577,8 +577,8 @@ def test_image_layers_python(self):
             aci_policy.populate_policy_content_for_all_images()
             layers = aci_policy.get_images()[0]._layers
             expected_layers = [
-                "7fdd8009ff6abf7a9d48808a507baecd5b815c8947fc03faa921b68a0ac260e8",
-                "280e8f98eb3a45ba7647d3df812236494188bfedb2c1fafa0ac2fb2f47930c0c"
+                "6ee0f2697647b69975229db7445260a1c9ae5b039f9a52a911e72b6c3302d391",
+                "3ba5dd39bf58f28b7e57a52d10edf1813fe3b1fd1d0ef57573b1b912d6b9d1f6"
             ]
             self.assertEqual(len(layers), len(expected_layers))
             for i in range(len(expected_layers)):