adding flag to omit ID from policy

src/confcom/azext_confcom/_help.py

@@ -89,6 +89,10 @@
           type: boolean
           short-summary: 'When enabled, the hashing algorithm used to generate the policy is faster but less memory efficient'
 
+        - name: --omit-id
+          type: boolean
+          short-summary: 'When enabled, the generated policy will not contain the ID field. This will keep the policy from being tied to a specific image name and tag.'
+
     examples:
         - name: Input an ARM Template file to inject a base64 encoded Confidential Container Security Policy into the ARM Template
           text: az confcom acipolicygen --template-file "./template.json"

src/confcom/azext_confcom/_params.py

@@ -153,6 +153,12 @@ def load_arguments(self, _):
             help="Use buffered image reader for dmverity hashing. This will speed up the hashing process but use much more memory.",
             validator=validate_faster_hashing,
         )
+        c.argument(
+            "omit_id",
+            options_list=("--omit-id"),
+            required=False,
+            help="Omit the id field in the policy. This is helpful if the image being used will be present in multiple registries and used interchangeably.",
+        )
 
     with self.argument_context("confcom katapolicygen") as c:
         c.argument(

src/confcom/azext_confcom/container.py

@@ -466,7 +466,6 @@ def extract_allow_privilege_escalation(container_json: Any) -> bool:
     allow_privilege_escalation = True
     # assumes that securityContext field is optional
     if security_context:
-
         # get the field for allow privilege escalation, default to true
         temp_privilege_escalation = case_insensitive_dict_get(
             security_context,
@@ -595,18 +594,13 @@ def __init__(
         self._user = user or {}
         self._capabilities = capabilities
         self._allow_privilege_escalation = allowPrivilegeEscalation
-        self._policy_json = None
-        self._policy_json_str = None
-        self._policy_json_str_pp = None
         self._identifier = id_val
         self._exec_processes = execProcesses or []
         self._signals = signals or []
         self._extraEnvironmentRules = extraEnvironmentRules
 
-    def get_policy_json(self) -> str:
-        if not self._policy_json:
-            self._policy_json_serialization()
-        return self._policy_json
+    def get_policy_json(self, omit_id: bool = False) -> str:
+        return self._populate_policy_json_elements(omit_id=omit_id)
 
     def get_id(self) -> str:
         return self._identifier
@@ -707,7 +701,6 @@ def _get_mounts_json(self) -> Dict[str, Any]:
             return []
 
         mounts = []
-
         for m in self._mounts:
             mount = copy.deepcopy(config.DEFAULT_MOUNT_POLICY)
             mount[
@@ -735,10 +728,8 @@ def _get_mounts_json(self) -> Dict[str, Any]:
 
         return mounts
 
-    def _populate_policy_json_elements(self) -> Dict[str, Any]:
-
+    def _populate_policy_json_elements(self, omit_id: bool = False) -> Dict[str, Any]:
         elements = {
-            config.POLICY_FIELD_CONTAINERS_ID: self._identifier,
             config.POLICY_FIELD_CONTAINERS_NAME: self.get_name(),
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_LAYERS: self._layers,
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_COMMANDS: self._command,
@@ -754,17 +745,11 @@ def _populate_policy_json_elements(self) -> Dict[str, Any]:
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_ALLOW_STDIO_ACCESS: self._allow_stdio_access,
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_NO_NEW_PRIVILEGES: not self._allow_privilege_escalation
         }
-        self._policy_json = elements
 
-        return self._policy_json
+        if not omit_id:
+            elements[config.POLICY_FIELD_CONTAINERS_ID] = self._identifier
 
-    def _policy_json_serialization(self):
-        policy = self._populate_policy_json_elements()
-        # serialize json policy to object, compact string and pretty print string
-        self._policy_json_str, self._policy_json_str_pp = (
-            json.dumps(policy, separators=(",", ":"), sort_keys=True),
-            json.dumps(policy, indent=2, sort_keys=True),
-        )
+        return elements
 
 
 class UserContainerImage(ContainerImage):
@@ -784,17 +769,12 @@ def from_json(
 
         # Start with the customer environment rules
         env_rules = _INJECTED_CUSTOMER_ENV_RULES
-
         # If is_vn2, add the VN2 environment rules
         if is_vn2:
             env_rules += _INJECTED_SERVICE_VN2_ENV_RULES
 
         image.set_extra_environment_rules(env_rules)
-
         return image
 
-    def _populate_policy_json_elements(self) -> Dict[str, Any]:
-        elements = super()._populate_policy_json_elements()
-        self._policy_json = elements
-
-        return self._policy_json
+    def _populate_policy_json_elements(self, omit_id: bool = False) -> Dict[str, Any]:
+        return super()._populate_policy_json_elements(omit_id=omit_id)

src/confcom/azext_confcom/custom.py

@@ -55,12 +55,13 @@ def acipolicygen_confcom(
     disable_stdio: bool = False,
     print_existing_policy: bool = False,
     faster_hashing: bool = False,
+    omit_id: bool = False,
 ):
 
     if print_existing_policy or outraw or outraw_pretty_print:
         logger.warning(
             "%s %s %s %s %s",
-            "Secrets that are included in the provided arm template or configuration files ",
+            "Secrets that are included in the provided arm template or configuration files",
             "in the container env or cmd sections will be printed out with this flag.",
             "These are outputed secrets that you must protect. Be sure that you do not include these secrets in your",
             "source control. Also verify that no secrets are present in the logs of your command or script.",
@@ -152,7 +153,7 @@ def acipolicygen_confcom(
             needed_metadata = {
                 VIRTUAL_NODE_YAML_METADATA: {
                     VIRTUAL_NODE_YAML_ANNOTATIONS: {
-                        VIRTUAL_NODE_YAML_POLICY: policy.get_serialized_output(),
+                        VIRTUAL_NODE_YAML_POLICY: policy.get_serialized_output(omit_id=omit_id),
                     }
                 }
             }
@@ -163,18 +164,21 @@ def acipolicygen_confcom(
             virtual_node_yaml[count_in_file] = current_yaml
 
             os_util.write_multiple_yaml_to_file(virtual_node_yaml_path, virtual_node_yaml)
+            print(str_to_sha256(policy.get_serialized_output(OutputType.RAW, omit_id=omit_id)))
+            logger.info("CCE Policy successfully injected into YAML file")
         elif diff:
             exit_code = get_diff_outputs(policy, output_type == security_policy.OutputType.PRETTY_PRINT)
         elif arm_template and not (print_policy_to_terminal or outraw or outraw_pretty_print):
             result = inject_policy_into_template(arm_template, arm_template_parameters,
-                                                 policy.get_serialized_output(), count)
+                                                 policy.get_serialized_output(omit_id=omit_id), count)
             if result:
                 # this is always going to be the unencoded policy
-                print(str_to_sha256(policy.get_serialized_output(OutputType.RAW)))
+                print(str_to_sha256(policy.get_serialized_output(OutputType.RAW, omit_id=omit_id)))
                 logger.info("CCE Policy successfully injected into ARM Template")
         else:
             # output to terminal
-            print(f"{policy.get_serialized_output(output_type)}\n\n")
+            print(f"{policy.get_serialized_output(output_type, omit_id=omit_id)}\n\n")
+            print(str_to_sha256(policy.get_serialized_output(OutputType.RAW, omit_id=omit_id)))
             # output to file
             if save_to_file:
                 logger.warning(

src/confcom/azext_confcom/security_policy.py

@@ -8,7 +8,6 @@
 import copy
 from typing import Any, List, Dict, Tuple
 from enum import Enum, auto
-import docker
 import deepdiff
 from knack.log import get_logger
 from tqdm import tqdm
@@ -64,7 +63,6 @@ def __init__(
         disable_stdio: bool = False,
         is_vn2: bool = False
     ) -> None:
-        self._docker_client = None
         self._rootfs_proxy = None
         self._policy_str = None
         self._policy_str_pp = None
@@ -135,42 +133,30 @@ def __init__(
 
         self._images = container_results
 
-    def __enter__(self) -> None:
+    def __enter__(self) -> Any:
         return self
 
     def __exit__(self, exception_type, exception_value, exception_traceback) -> None:
-        self.close()
-
-    def _get_docker_client(self) -> docker.client.DockerClient:
-        if not self._docker_client:
-            self._docker_client = docker.from_env()
-
-        return self._docker_client
+        return None
 
     def _get_rootfs_proxy(self) -> SecurityPolicyProxy:
         if not self._rootfs_proxy:
             self._rootfs_proxy = SecurityPolicyProxy()
 
         return self._rootfs_proxy
 
-    def _close_docker_client(self) -> None:
-        if self._docker_client:
-            self._get_docker_client().close()
-
-    def close(self) -> None:
-        self._close_docker_client()
-
     def get_serialized_output(
         self,
         output_type: OutputType = OutputType.DEFAULT,
         rego_boilerplate=True,
+        omit_id: bool = False,
     ) -> str:
         # error check the output type
         if not isinstance(output_type, Enum) or output_type.value not in [item.value for item in OutputType]:
             eprint("Unknown output type for serialization.")
 
         policy_str = self._policy_serialization(
-            output_type == OutputType.PRETTY_PRINT
+            output_type == OutputType.PRETTY_PRINT, omit_id=omit_id
         )
 
         if rego_boilerplate:
@@ -364,14 +350,14 @@ def save_to_file(
         output = self.get_serialized_output(output_type)
         os_util.write_str_to_file(file_path, output)
 
-    def _policy_serialization(self, pretty_print=False) -> str:
+    def _policy_serialization(self, pretty_print=False, omit_id: bool = False) -> str:
         policy = []
         regular_container_images = self.get_images()
 
         is_sidecars = True
         for image in regular_container_images:
             is_sidecars = is_sidecars and is_sidecar(image.containerImage)
-            image_dict = image.get_policy_json()
+            image_dict = image.get_policy_json(omit_id=omit_id)
             policy.append(image_dict)
 
         if not is_sidecars:
@@ -511,7 +497,6 @@ def populate_policy_content_for_all_images(
 
                 progress.update()
             progress.close()
-            self.close()
 
             # unload the message queue
             for message in message_queue:
@@ -520,10 +505,6 @@ def populate_policy_content_for_all_images(
     def get_images(self) -> List[ContainerImage]:
         return self._images
 
-    def pull_image(self, image: ContainerImage) -> Any:
-        client = self._get_docker_client()
-        return client.images.pull(image.base, image.tag)
-
 
 # pylint: disable=R0914,
 def load_policy_from_arm_template_str(

src/confcom/azext_confcom/template_util.py

@@ -27,8 +27,7 @@
 
 
 class DockerClient:
-    def __init__(self) -> None:
-        self._client = None
+    _client = None
 
     def __enter__(self) -> docker.DockerClient:
         return self.get_client()

src/confcom/azext_confcom/tests/latest/README.md

@@ -37,6 +37,7 @@ test_update_infrastructure_svn | mcr.microsoft.com/cbl-mariner/distroless/python
 test_multiple_policies | mcr.microsoft.com/cbl-mariner/distroless/python:3.9-nonroot & mcr.microsoft.com/cbl-mariner/distroless/minimal:2.0 | See if two unique policies are generated from a single ARM Template container multiple container groups. Also have an extra resource that is untouched. Also has a secureValue for an environment variable.
 test_arm_template_with_init_container | mcr.microsoft.com/cbl-mariner/distroless/python:3.9-nonroot & mcr.microsoft.com/cbl-mariner/distroless/minimal:2.0 | See if having an initContainer is picked up and added to the list of valid containers
 test_arm_template_without_stdio_access | mcr.microsoft.com/cbl-mariner/distroless/minimal:2.0 | See if disabling container stdio access gets passed down to individual containers
+test_arm_template_omit_id | mcr.microsoft.com/cbl-mariner/distroless/python:3.9-nonroot | Check that the id field is omitted from the policy
 test_arm_template_allow_elevated_false | mcr.microsoft.com/cbl-mariner/distroless/minimal:2.0 | Disabling allow_elevated via securityContext
 test_arm_template_policy_regex | mcr.microsoft.com/cbl-mariner/distroless/python:3.9-nonroot | Make sure the regex generated from the ARM Template workflow matches that of the policy.json workflow
 test_wildcard_env_var | mcr.microsoft.com/cbl-mariner/distroless/python:3.9-nonroot | Check that an "allow all" regex is created when a value for env var is not provided via a parameter value
@@ -78,6 +79,7 @@ test_docker_pull | mcr.microsoft.com/cbl-mariner/distroless/minimal:2.0 | Test p
 test_infrastructure_svn | mcr.microsoft.com/cbl-mariner/distroless/minimal:2.0 | make sure the correct infrastructure_svn is present in the policy
 test_stdio_access_default | mcr.microsoft.com/cbl-mariner/distroless/python:3.9-nonroot | Checking the default value for std I/O access
 test_stdio_access_updated | mcr.microsoft.com/cbl-mariner/distroless/python:3.9-nonroot | Checking the value for std I/O when it's set
+test_omit_id | mcr.microsoft.com/cbl-mariner/distroless/python:3.9-nonroot | Check that the id field is omitted from the policy
 test_environment_variables_parsing | mcr.microsoft.com/azuredocs/aci-dataprocessing-cc:v1 | Make sure env vars are output in the right format
 test_get_layers_from_not_exists_image | notexists:1.0.0 | Fail out grabbing layers if image doesn't exist
 test_incorrect_allow_elevated_data_type | mcr.microsoft.com/cbl-mariner/distroless/minimal:2.0 | Making allow_elevated fail out if it's not a boolean