[confcom] supporting OCI images and adding tests (#8570)

* adding fragment signing tests

* fixing where params and vars are filled in

Co-authored-by: Khalil Sayid <khalilsayid@microsoft.com>

* updating docs to explain types of fragments

Co-authored-by: Khalil Sayid <khalilsayid@microsoft.com>

* fixing error that came up when using diff mode with non-default fragments

* adding logging to some functions

* adding support for OCI formatted images

* adding exec process section to config file

* adding support for custom mount options

---------

Co-authored-by: Khalil Sayid <khalilsayid@microsoft.com>

Author: SethHollandsworth

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

@@ -30,6 +30,8 @@
     - [allow_unencrypted_scratch](#allow_unencrypted_scratch)
     - [allow_capabilities_dropping](#allow_capabilities_dropping)
 - [Microsoft Azure CLI 'confcom acifragmentgen' Extension Examples](#microsoft-azure-cli-confcom-acifragmentgen-extension-examples)
+  - [Types of Policy Fragments](#types-of-policy-fragments)
+  - [Examples](#examples)
 - [Microsoft Azure CLI 'confcom katapolicygen' Extension Examples](#microsoft-azure-cli-confcom-katapolicygen-extension-examples)
 
 ## Microsoft Azure CLI 'confcom acipolicygen' Extension Examples
@@ -192,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
@@ -665,6 +669,15 @@ Run `az confcom acifragmentgen --help` to see a list of supported arguments alon
 
 For information on what a policy fragment is, see [policy fragments](#policy-fragments). For a full walkthrough on how to generate a policy fragment and use it in a policy, see [Create a Key and Cert for Signing](../samples/certs/README.md).
 
+### Types of Policy Fragments
+
+There are two types of policy fragments:
+
+1. Image-attached fragments: These are fragments that are attached to an image in an ORAS-compliant registry. They are used to provide additional security information about the image and are to be used for a single image. Image-attached fragments are currently in development. Note that nested image-attached fragments are *not* supported.
+2. Standalone fragments: These are fragments that are uploaded to an ORAS-compliant registry independent of a specific image and can be used for multiple images. Standalone fragments are currently not supported. Once implemented, nested standalone fragments will be supported.
+
+### Examples
+
 **Examples:**
 
 Example 1: The following command creates a security fragment and prints it to stdout as well as saving it to a file `contoso.rego`:

src/confcom/azext_confcom/cose_proxy.py

@@ -9,6 +9,7 @@
 import platform
 from typing import List
 import requests
+from knack.log import get_logger
 from azext_confcom.errors import eprint
 from azext_confcom.config import (
     REGO_CONTAINER_START,
@@ -21,7 +22,7 @@
     ACI_FIELD_CONTAINERS_REGO_FRAGMENTS_INCLUDES,
 )
 
-
+logger = get_logger(__name__)
 host_os = platform.system()
 machine = platform.machine()
 
@@ -128,7 +129,7 @@ def cose_sign(
 
         if iss:
             arg_list.extend(["-issuer", iss])
-
+        logger.info("Signing the policy fragment: %s", out_path)
         call_cose_sign_tool(arg_list, "Error signing the policy fragment")
         return True
 
@@ -150,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("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")
@@ -182,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("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")
@@ -194,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("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/custom.py

@@ -97,7 +97,8 @@ def acipolicygen_confcom(
     # gather information about the fragments being used in the new policy
     if include_fragments:
         fragments_list = os_util.load_json_from_file(fragments_json or input_path)
-        fragments_list = fragments_list.get("fragments", []) or fragments_list
+        if isinstance(fragments_list, dict):
+            fragments_list = fragments_list.get("fragments", [])
 
         # convert to list if it's just a dict
         if not isinstance(fragments_list, list):
@@ -158,18 +159,21 @@ def acipolicygen_confcom(
     # and associate them with each container group
 
     if include_fragments:
+        logger.info("Including fragments in the policy")
         fragment_policy_list = []
         container_names = []
         fragment_imports = []
         for policy in container_group_policies:
             fragment_imports.extend(policy.get_fragments())
             for container in policy.get_images():
                 container_names.append(container.get_container_image())
+        # get all the fragments that are being used in the policy
         fragment_policy_list = get_all_fragment_contents(container_names, fragment_imports)
         for policy in container_group_policies:
             policy.set_fragment_contents(fragment_policy_list)
 
     for count, policy in enumerate(container_group_policies):
+        # this is where parameters and variables are populated
         policy.populate_policy_content_for_all_images(
             individual_image=bool(image_name), tar_mapping=tar_mapping, faster_hashing=faster_hashing
         )

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,6 +17,37 @@
 host_os = platform.system()
 machine = platform.machine()
 
+logger = get_logger(__name__)
+
+
+def prepend_docker_registry(image_name: str) -> str:
+    """
+    Normalize a Docker image reference by adding `docker.io/library` if necessary.
+
+    Args:
+        image (str): The Docker image reference (e.g., `nginx:latest` or `myrepo/myimage`).
+
+    Returns:
+        str: The normalized Docker image reference.
+    """
+    # Split the image into name and tag
+    if ":" in image_name:
+        name, _ = image_name.rsplit(":", 1)
+    else:
+        name, _ = image_name, "latest"
+
+    registry = ""
+    # Check if the image name contains a registry (e.g., docker.io, custom registry)
+    if "/" not in name or "." not in name.split("/")[0]:
+        # If no registry is specified, assume docker.io/library
+        if "/" not in name:
+            # Add the `library` namespace for official images
+            registry = "library/"
+        # Add the default `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)
@@ -26,10 +58,14 @@ def call_oras_cli(args, check=False):
 def discover(
     image: str,
 ) -> List[str]:
+    # normalize the name in case the docker registry is implied
+    image = prepend_docker_registry(image)
+
     arg_list = ["oras", "discover", image, "-o", "json", "--artifact-type", ARTIFACT_TYPE]
     item = call_oras_cli(arg_list, check=False)
     hashes = []
 
+    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", [])
@@ -55,6 +91,7 @@ def pull(
     if "@sha256:" in image:
         image = image.split("@")[0]
     arg_list = ["oras", "pull", f"{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/rootfs_proxy.py

@@ -103,6 +103,7 @@ def get_policy_image_layers(
 
         # decide if we're reading from a tarball or not
         if tar_location:
+            logger.info("Calculating layer hashes from tarball")
             arg_list += ["--tarball", tar_location]
         else:
             arg_list += ["-d"]