[confcom] Add parse aci arm command

src/confcom/azext_confcom/_params.py

@@ -43,6 +43,65 @@ def load_arguments(self, _):
         c.argument("tags", tags_type)
         c.argument("confcom_name", confcom_name_type, options_list=["--name", "-n"])
 
+    with self.argument_context("confcom parse aci arm") as c:
+        c.positional(
+            "arm_template_path",
+            help="Path to the ARM template file to parse.",
+        )
+        c.argument(
+            "arm_template_parameters_path",
+            options_list=("--parameters", "-p"),
+            required=False,
+            help="ARM template parameters",
+            # validator=validate_params_file
+        )
+        c.argument(
+            "exclude_default_fragments",
+            options_list=("--exclude-default-fragments", "-e"),
+            default=False,
+            action="store_true",
+            required=False,
+            help="Exclude default fragments in the generated policy",
+        )
+        c.argument(
+            "infrastructure_svn",
+            options_list=("--infrastructure-svn",),
+            required=False,
+            help="Minimum Allowed Software Version Number for Infrastructure Fragment",
+            validator=validate_infrastructure_svn,
+        )
+        c.argument(
+            "debug_mode",
+            options_list=("--debug-mode",),
+            default=False,
+            action="store_true",
+            required=False,
+            help="Debug mode will enable processes in a container group that are helpful for debugging",
+        )
+        c.argument(
+            "disable_stdio",
+            options_list=("--disable-stdio",),
+            default=False,
+            action="store_true",
+            required=False,
+            help="Disabling container stdio will disable the ability to see the output of the container in the terminal for Confidential ACI",
+        )
+        c.argument(
+            "approve_wildcards",
+            options_list=("--approve-wildcards", "-y"),
+            default=False,
+            action="store_true",
+            required=False,
+            help="Approving wildcards by default will get rid of the prompts during the wildcard environment variable use case and auto-approve the use of wildcards",
+        )
+        c.argument(
+            "policy_format",
+            options_list=("--format",),
+            default="explicit",
+            required=False,
+            help="The format for the policy output, explicit includes all objects which can be implicitly added, minimal includes only the essential fields",
+        )
+
     with self.argument_context("confcom acipolicygen") as c:
         c.argument(
             "input_path",

src/confcom/azext_confcom/command/parse_aci_arm.py

@@ -0,0 +1,63 @@
+
+from dataclasses import asdict, fields
+from typing import Optional
+from azext_confcom import os_util
+from azext_confcom.lib.aci_infrastructure import omit_implicit_features
+from azext_confcom.lib.aci_policy_spec import POLICY_SPEC_CLASSES
+from azext_confcom.lib.arm_to_aci_policy_spec import arm_to_aci_policy_spec
+
+
+def omit_defaults_dict_factory(fields_dict) -> dict:
+
+    result = {}
+
+    for potential_class in POLICY_SPEC_CLASSES:
+        try:
+            instance = potential_class(**dict(fields_dict))
+            for field in fields(instance):
+                value = getattr(instance, field.name)
+                if value not in (None, field.default, []):
+                    result[field.name] = value
+            break
+        except TypeError:
+            continue
+
+    return result
+
+
+def parse_aci_arm(
+    arm_template_path: str,
+    arm_template_parameters_path: Optional[str],
+    debug_mode: bool,
+    exclude_default_fragments: bool,
+    infrastructure_svn: Optional[str],
+    disable_stdio: bool,
+    approve_wildcards: bool,
+    policy_format: str,
+) -> list[dict[str, str]]:
+
+    with open(arm_template_path, 'r') as f:
+        arm_template = os_util.load_json_from_str(f.read())
+
+    arm_template_parameters = {}
+    if arm_template_parameters_path is not None:
+        with open(arm_template_parameters_path, 'r') as f:
+            arm_template_parameters = os_util.load_json_from_str(f.read())
+
+    aci_policy_specs = list(arm_to_aci_policy_spec(
+        arm_template=arm_template,
+        arm_template_parameters=arm_template_parameters,
+        include_infrastructure_fragment=not exclude_default_fragments,
+        infrastructure_fragment_min_svn=infrastructure_svn,
+        debug_mode=debug_mode,
+        allow_stdio_access=not disable_stdio,
+        approve_wildcards=approve_wildcards,
+    ))
+
+    specs = []
+    for spec in aci_policy_specs:
+        if policy_format == "minimal":
+            spec = omit_implicit_features(spec)
+        specs.append(asdict(spec, dict_factory=omit_defaults_dict_factory))
+
+    return specs

src/confcom/azext_confcom/commands.py

@@ -11,5 +11,8 @@ def load_command_table(self, _):
         g.custom_command("acifragmentgen", "acifragmentgen_confcom")
         g.custom_command("katapolicygen", "katapolicygen_confcom")
 
+    with self.command_group("confcom parse aci") as g:
+        g.custom_command("arm", "parse_aci_arm", is_preview=True)
+
     with self.command_group("confcom"):
         pass

src/confcom/azext_confcom/container.py

@@ -701,25 +701,31 @@ def parse_all_parameters_and_variables(self, params, vars_dict) -> None:
 
     def _get_environment_rules(self) -> List[Dict[str, Any]]:
         out_rules = copy.deepcopy(self._environmentRules)
-        env_var_names = [
-            var[config.POLICY_FIELD_CONTAINERS_ELEMENTS_ENVS_RULE].split("=")[0]
-            for var in out_rules
+
+        # Remove variables from out_rules which appear in extraEnvironmentRules
+        # so they're always at the end of the list. This is to maintain the
+        # arbitrary order that already exists to avoid breaking customers
+        out_rules = [
+            rule for rule in out_rules
+            if rule["pattern"] not in {
+                f"{r["name"]}={r["value"]}" for r in self._extraEnvironmentRules
+            }
         ]
+
         for rule in self._extraEnvironmentRules:
-            if rule[config.ACI_FIELD_CONTAINERS_ENVS_NAME] not in env_var_names:
-                out_rules.append(
-                    {
-                        config.POLICY_FIELD_CONTAINERS_ELEMENTS_ENVS_RULE:
-                        f"{rule[config.ACI_FIELD_CONTAINERS_ENVS_NAME]}="
-                        + f"{rule[config.ACI_FIELD_CONTAINERS_ENVS_VALUE]}",
-                        config.POLICY_FIELD_CONTAINERS_ELEMENTS_ENVS_STRATEGY: rule[
-                            config.ACI_FIELD_CONTAINERS_ENVS_STRATEGY
-                        ],
-                        config.POLICY_FIELD_CONTAINERS_ELEMENTS_REQUIRED: rule[
-                            config.ACI_FIELD_CONTAINERS_ENVS_REQUIRED
-                        ],
-                    }
-                )
+            out_rules.append(
+                {
+                    config.POLICY_FIELD_CONTAINERS_ELEMENTS_ENVS_RULE:
+                    f"{rule[config.ACI_FIELD_CONTAINERS_ENVS_NAME]}="
+                    + f"{rule[config.ACI_FIELD_CONTAINERS_ENVS_VALUE]}",
+                    config.POLICY_FIELD_CONTAINERS_ELEMENTS_ENVS_STRATEGY: rule[
+                        config.ACI_FIELD_CONTAINERS_ENVS_STRATEGY
+                    ],
+                    config.POLICY_FIELD_CONTAINERS_ELEMENTS_REQUIRED: rule.get(
+                        config.ACI_FIELD_CONTAINERS_ENVS_REQUIRED, False
+                    ),
+                }
+            )
 
         return out_rules
 

src/confcom/azext_confcom/custom.py

@@ -23,6 +23,7 @@
     extract_confidential_properties, get_image_name, inject_policy_into_template, inject_policy_into_yaml,
     pretty_print_func, print_existing_policy_from_arm_template,
     print_existing_policy_from_yaml, print_func, str_to_sha256)
+from azext_confcom.command.parse_aci_arm import parse_aci_arm as _parse_aci_arm
 from knack.log import get_logger
 from pkg_resources import parse_version
 
@@ -497,3 +498,27 @@ def get_fragment_output_type(outraw):
     if outraw:
         output_type = security_policy.OutputType.RAW
     return output_type
+
+
+# This should be *args, **kwargs to avoid having to touch this, however the az
+# extension frameworks then expects literal args and kwargs parameters.
+def parse_aci_arm(
+    arm_template_path: str,
+    arm_template_parameters_path: Optional[str],
+    debug_mode: bool,
+    exclude_default_fragments: bool,
+    infrastructure_svn: Optional[str],
+    disable_stdio: bool,
+    approve_wildcards: bool,
+    policy_format: str,
+) -> str:
+    return _parse_aci_arm(
+        arm_template_path,
+        arm_template_parameters_path,
+        debug_mode,
+        exclude_default_fragments,
+        infrastructure_svn,
+        disable_stdio,
+        approve_wildcards,
+        policy_format,
+    )

src/confcom/azext_confcom/lib/aci_infrastructure.py

@@ -0,0 +1,37 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+
+from dataclasses import fields, is_dataclass, replace
+from azext_confcom import config
+from azext_confcom.lib.aci_policy_spec import AciContainerPropertyEnvVariable, AciContainerPropertyVolumeMounts, AciFragmentSpec
+
+
+INFRASTRUCTURE_FRAGMENTS = [AciFragmentSpec(**frag) for frag in config.DEFAULT_REGO_FRAGMENTS]
+OPENGCS_ENV_RULES = [AciContainerPropertyEnvVariable(**env_var) for env_var in config.OPENGCS_ENV_RULES]
+FABRIC_ENV_RULES = [AciContainerPropertyEnvVariable(**env_var) for env_var in config.FABRIC_ENV_RULES]
+MANAGED_IDENTITY_ENV_RULES = [AciContainerPropertyEnvVariable(**env_var) for env_var in config.MANAGED_IDENTITY_ENV_RULES]
+ENABLE_RESTART_ENV_RULE = [AciContainerPropertyEnvVariable(**env_var) for env_var in config.ENABLE_RESTART_ENV_RULE]
+DEFAULT_MOUNTS_USER = [AciContainerPropertyVolumeMounts(**mount) for mount in config.DEFAULT_MOUNTS_USER]
+
+
+implicit_features = [
+    *INFRASTRUCTURE_FRAGMENTS,
+    *OPENGCS_ENV_RULES,
+    *FABRIC_ENV_RULES,
+    *MANAGED_IDENTITY_ENV_RULES,
+    *ENABLE_RESTART_ENV_RULE,
+    *DEFAULT_MOUNTS_USER,
+]
+
+
+def omit_implicit_features(obj):
+    if obj is None:
+        return None
+    if is_dataclass(obj):
+        return replace(obj, **{f.name: omit_implicit_features(getattr(obj, f.name)) for f in fields(obj)})
+    if isinstance(obj, list):
+        return [omit_implicit_features(x) for x in obj if x not in implicit_features]
+    return obj

src/confcom/azext_confcom/lib/aci_policy_spec.py

@@ -3,11 +3,16 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-from dataclasses import dataclass
+from dataclasses import dataclass, is_dataclass
+import inspect
+import sys
 from typing import Optional
 from typing_extensions import Literal
 
 
+AciProfile = Literal["strict", "debug"]
+
+
 @dataclass
 class AciContainerPropertyEnvVariable:
     name: str
@@ -50,7 +55,7 @@ class AciContainerPropertySecurityContext:
 
 
 @dataclass
-class AciContainerProperties():
+class AciContainerProperties:
     image: str
     allowStdioAccess: bool = True
     environmentVariables: Optional[list[AciContainerPropertyEnvVariable]] = None
@@ -82,3 +87,14 @@ class AciContainerSpec:
 class AciPolicySpec:
     fragments: Optional[list[AciFragmentSpec]]
     containers: Optional[list[AciContainerSpec]]
+    profile: AciProfile = "strict"
+    include_infrastructure_fragment: bool = True
+    infrastructure_fragment_min_svn: Optional[str] = None
+    allow_stdio_access: bool = True
+
+
+POLICY_SPEC_CLASSES = [
+    cls
+    for _, cls in inspect.getmembers(sys.modules[__name__], inspect.isclass)
+    if is_dataclass(cls) and cls.__module__ == sys.modules[__name__].__name__
+]