Add containers from_aci command

linter_exclusions.yml

@@ -3504,3 +3504,9 @@ neon postgres organization:
 neon postgres project:
   rule_exclusions:
   - require_wait_command_if_no_wait
+
+confcom containers from_aci:
+  parameters:
+    template:
+      rule_exclusions:
+        - no_positional_parameters

src/confcom/HISTORY.rst

@@ -3,6 +3,10 @@
 Release History
 ===============
 
+1.5.0
+++++++
+* Add containers from_aci command to generate container definitions from an ARM or bicep ACI template.
+
 1.4.0
 ++++++
 * Add --with-containers flag to acipolicygen and acifragmentgen to allow passing container policy definitions directly

src/confcom/azext_confcom/_help.py

@@ -278,3 +278,41 @@
         - name: Input a Kubernetes YAML file with a custom containerd socket path
           text: az confcom katapolicygen --yaml "./pod.json" --containerd-pull --containerd-socket-path "/my/custom/containerd.sock"
 """
+
+
+helps[
+    "confcom containers"
+] = """
+    type: group
+    short-summary: Commands which generate Security Policy Container Definitions.
+"""
+
+
+helps[
+    "confcom containers from_aci"
+] = """
+    type: command
+    short-summary: Create a Security Policy Container Definition based on an ACI template.
+
+    parameters:
+        - name: --parameters -p
+          type: string
+          short-summary: 'Input parameters file to optionally accompany a Bicep Template'
+
+        - name: --idx
+          type: int
+          short-summary: 'The index of the container resource in the template to generate the policy for. Default is 0'
+
+
+    examples:
+        - name: Input an ACI Template and generate container definitions
+          text: az confcom containers from_aci arm_template.json
+        - name: Input an ACI Template with a bicepparam file and generate container definitions
+          text: az confcom containers from_aci arm_template.json --parameters parameters.json
+        - name: Input an ACI Template with inline parameter and generate container definitions
+          text: az confcom containers from_aci arm_template.json --parameters image=my.azurecr.io/myimage:tag
+        - name: Input an ACI Template as Bicep
+          text: az confcom containers from_aci my_app.bicep --parameters my_app.bicepparam
+        - name: Input an ACI Template and generate container definitions for the second container resource
+          text: az confcom containers from_aci arm_template.json --idx 1
+"""

src/confcom/azext_confcom/_params.py

@@ -6,6 +6,7 @@
 
 import json
 from knack.arguments import CLIArgumentType
+from argcomplete.completers import FilesCompleter
 from azext_confcom._validators import (
     validate_params_file,
     validate_diff,
@@ -434,3 +435,28 @@ def load_arguments(self, _):
             help="Path to containerd socket if not using the default",
             validator=validate_katapolicygen_input,
         )
+
+    with self.argument_context("confcom containers from_aci") as c:
+        c.positional(
+            "template",
+            type=str,
+            help="Template to create container definitions from",
+        )
+        c.argument(
+            "parameters",
+            options_list=['--parameters', '-p'],
+            action='append',
+            nargs='+',
+            completer=FilesCompleter(),
+            required=False,
+            default=[],
+            help='The parameters for the ARM template'
+        )
+        c.argument(
+            "group_index",
+            options_list=['--idx'],
+            required=False,
+            default=0,
+            type=int,
+            help='The index of the container group in the template to use'
+        )

src/confcom/azext_confcom/command/containers_from_aci.py

@@ -0,0 +1,63 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import json
+
+from azext_confcom.lib.deployments import parse_deployment_template
+from azext_confcom.lib.images import get_image_config, get_image_layers
+from azext_confcom.lib.platform import ACI_MOUNTS
+
+
+def aci_container_to_policy(
+    arm_container: dict,
+):
+    properties = arm_container.get("properties", {})
+    image = properties.get("image")
+    image_config = get_image_config(image)
+
+    return {
+        "name": arm_container.get("name"),
+        "id": image,
+        "layers": get_image_layers(image),
+        "command": (
+            properties.get("command") or
+            image_config.get("command")
+        ),
+        "env_rules": (
+            image_config.get("env_rules") +
+            [{
+                "pattern": f"{env.get('name')}={env.get('value')}",
+                "strategy": "string",
+                "required": False,
+            } for env in properties.get("environmentVariables", [])]
+        ),
+        "mounts": ACI_MOUNTS,
+    }
+
+
+def containers_from_aci(
+    az_cli_command,
+    template: str,
+    parameters: dict,
+    group_index: int
+) -> None:
+
+    template = parse_deployment_template(
+        az_cli_command,
+        template,
+        parameters,
+    )
+
+    supported_resources = [r for r in template.get("resources", []) if r.get("type") in {
+        "Microsoft.ContainerInstance/containerGroups",
+        "Microsoft.ContainerInstance/containerGroupProfiles",
+    }]
+
+    container_group = supported_resources[group_index]
+
+    return json.dumps([
+        aci_container_to_policy(container)
+        for container in container_group.get("properties", {}).get("containers", [])
+    ])

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 containers") as g:
+        g.custom_command("from_aci", "containers_from_aci")
+
     with self.command_group("confcom"):
         pass

src/confcom/azext_confcom/custom.py

@@ -22,6 +22,7 @@
     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.containers_from_aci import containers_from_aci as _containers_from_aci
 from knack.log import get_logger
 from pkg_resources import parse_version
 
@@ -512,3 +513,17 @@ def get_fragment_output_type(outraw):
     if outraw:
         output_type = security_policy.OutputType.RAW
     return output_type
+
+
+def containers_from_aci(
+    cmd,
+    template: str,
+    parameters: dict,
+    group_index: int,
+) -> None:
+    print(_containers_from_aci(
+        az_cli_command=cmd,
+        template=template,
+        parameters=parameters,
+        group_index=group_index,
+    ))

src/confcom/azext_confcom/lib/deployments.py

@@ -0,0 +1,93 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import json
+import re
+
+from azure.cli.command_modules.resource.custom import (
+    _prepare_deployment_properties_unmodified,
+)
+from azure.cli.core.profiles import ResourceType
+
+
+class _ResourceDeploymentCommandAdapter:
+    """Ensure required resource type defaults are present when reusing resource module helpers."""
+
+    def __init__(self, cmd):
+        self._cmd = cmd
+        self.cli_ctx = cmd.cli_ctx
+
+    def get_models(self, *attr_args, **kwargs):
+        kwargs.setdefault('resource_type', ResourceType.MGMT_RESOURCE_DEPLOYMENTS)
+        return self._cmd.get_models(*attr_args, **kwargs)
+
+    def __getattr__(self, name):
+        return getattr(self._cmd, name)
+
+
+def get_parameters(
+    arm_template: dict,
+    arm_template_parameters: dict,
+) -> dict:
+
+    return {
+        parameter_key: (
+            arm_template_parameters.get(parameter_key, {}).get("value")
+            or arm_template.get("parameters", {}).get(parameter_key, {}).get("value")
+            or arm_template.get("parameters", {}).get(parameter_key, {}).get("defaultValue")
+        )
+        for parameter_key in arm_template.get("parameters", {}).keys()
+    }
+
+
+def eval_parameters(
+    arm_template: dict,
+    arm_template_parameters: dict,
+) -> dict:
+
+    parameters = get_parameters(arm_template, arm_template_parameters)
+    return json.loads(re.compile(r"\[parameters\(\s*'([^']+)'\s*\)\]").sub(
+        lambda match: json.dumps(parameters.get(match.group(1)) or match.group(0))[1:-1],
+        json.dumps(arm_template),
+    ))
+
+
+def eval_variables(
+    arm_template: dict,
+    _arm_template_parameters: dict,
+) -> dict:
+
+    variables = arm_template.get("variables", {})
+    return json.loads(re.compile(r"\[variables\(\s*'([^']+)'\s*\)\]").sub(
+        lambda match: json.dumps(variables.get(match.group(1), match.group(0)))[1:-1],
+        json.dumps(arm_template),
+    ))
+
+
+EVAL_FUNCS = [
+    eval_parameters,
+    eval_variables,
+]
+
+
+def parse_deployment_template(
+    az_cli_command,
+    template: str,
+    parameters: dict,
+) -> dict:
+    properties = _prepare_deployment_properties_unmodified(
+        cmd=_ResourceDeploymentCommandAdapter(az_cli_command),
+        deployment_scope='resourceGroup',
+        template_file=template,
+        parameters=parameters,
+        no_prompt=True,
+    )
+    template = json.loads(properties.template)
+    parameters = properties.parameters or {}
+
+    for eval_func in EVAL_FUNCS:
+        template = eval_func(template, parameters)
+
+    return template

src/confcom/azext_confcom/lib/images.py

@@ -0,0 +1,64 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import functools
+import os
+import subprocess
+import docker
+
+
+@functools.lru_cache()
+def get_image(image_ref: str) -> docker.models.images.Image:
+
+    client = docker.from_env()
+
+    try:
+        image = client.images.get(image_ref)
+    except docker.errors.ImageNotFound:
+        client.images.pull(image_ref)
+
+    image = client.images.get(image_ref)
+    return image
+
+
+def get_image_layers(image: str) -> list[str]:
+
+    binary_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "..", "bin", "dmverity-vhd")
+
+    get_image(image)
+    result = subprocess.run(
+        [binary_path, "-d", "roothash", "-i", image],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=True,
+        text=True,
+    )
+
+    return [line.split("hash: ")[-1] for line in result.stdout.splitlines()]
+
+
+def get_image_config(image: str) -> dict:
+
+    image_config = get_image(image).attrs.get("Config")
+
+    config = {}
+
+    if image_config.get("Cmd") or image_config.get("Entrypoint"):
+        config["command"] = (
+            image_config.get("Entrypoint") or [] +
+            image_config.get("Cmd") or []
+        )
+
+    if image_config.get("Env"):
+        config["env_rules"] = [{
+            "pattern": p,
+            "strategy": "string",
+            "required": False,
+        } for p in image_config.get("Env")]
+
+    if image_config.get("WorkingDir"):
+        config["working_dir"] = image_config.get("WorkingDir")
+
+    return config

src/confcom/azext_confcom/lib/platform.py

@@ -0,0 +1,17 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+ACI_MOUNTS = [
+    {
+        "destination": "/etc/resolv.conf",
+        "options": [
+            "rbind",
+            "rshared",
+            "rw"
+        ],
+        "source": "sandbox:///tmp/atlas/resolvconf/.+",
+        "type": "bind"
+    }
+]

src/confcom/setup.py

@@ -19,7 +19,7 @@
 
     logger.warn("Wheel is not available, disabling bdist_wheel hook")
 
-VERSION = "1.4.1"
+VERSION = "1.5.0"
 
 # The full list of classifiers is available at
 # https://pypi.python.org/pypi?%3Aaction=list_classifiers