Add containers from_vn2 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_vn2:
+  parameters:
+    template:
+      rule_exclusions:
+        - no_positional_parameters

src/confcom/HISTORY.rst

@@ -3,6 +3,10 @@
 Release History
 ===============
 
+1.5.0
+++++++
+* Add containers from_vn2 command to generate container definitions from a VN2 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,29 @@
         - 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_vn2"
+] = """
+    type: command
+    short-summary: Create a Security Policy Container Definition based on a VN2 template.
+
+    parameters:
+        - name: --name -n
+          type: string
+          short-summary: 'The name of the container to generate the policy for'
+
+
+    examples:
+        - name: Input a VN2 Template and generate container definitions
+          text: az confcom containers from_vn2 vn2.yaml --name mycontainer
+"""

src/confcom/azext_confcom/_params.py

@@ -434,3 +434,17 @@ 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_vn2") as c:
+        c.positional(
+            "template",
+            type=str,
+            help="Template to create container definitions from",
+        )
+        c.argument(
+            "container_name",
+            options_list=['--name', "-n"],
+            required=True,
+            type=str,
+            help='The name of the container in the template to use'
+        )

src/confcom/azext_confcom/command/containers_from_vn2.py

@@ -0,0 +1,85 @@
+# --------------------------------------------------------------------------------------------
+# 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 pathlib import Path
+import yaml
+
+# 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 VN2_MOUNTS
+# from azext_confcom.lib.platform import ACI_MOUNTS
+
+
+def find_vn2_containers(vn2_template):
+    for key, value in vn2_template.items():
+        if key == "containers":
+            yield from value
+        elif isinstance(value, dict):
+            result = find_vn2_containers(value)
+            if result is not None:
+                yield from result
+        elif isinstance(value, list):
+            for item in value:
+                if isinstance(item, dict):
+                    result = find_vn2_containers(item)
+                    if result is not None:
+                        yield from result
+
+
+def containers_from_vn2(
+    template: str,
+    container_name: str
+) -> None:
+
+    with Path(template).open("r") as f:
+        template_yaml = yaml.safe_load(f)
+
+    # Find containers matching the specified name (and check there's exactly one)
+    template_containers = [
+        container
+        for container in find_vn2_containers(template_yaml)
+        if container.get("name") == container_name
+    ]
+    assert len(template_containers) > 0, f"No containers with name {container_name} found."
+    assert len(template_containers) <= 1, f"Multiple containers with name {container_name} found."
+
+    template_container = template_containers[0]
+
+    image = template_container.get("image")
+
+    image_config = get_image_config(image)
+
+    env_rules = image_config.pop("env_rules", [])
+    for env_var in template_container.get("env", []):
+        env_rules.append({
+            "pattern": f"{env_var.get('name')}={env_var.get('value')}",
+            "strategy": "string",
+            "required": False,
+        })
+
+    mounts = image_config.pop("mounts", [])
+    mounts += [
+        {
+            "destination": m.get("mountPath"),
+            "options": [
+                "rbind",
+                "rshared",
+                "ro" if m.get("readOnly") else "rw"
+            ],
+            "source": "sandbox:///tmp/atlas/emptydir/.+",
+            "type": "bind",
+        }
+        for m in template_container.get("volumeMounts", [])
+    ]
+    mounts += VN2_MOUNTS
+
+    return json.dumps({
+        "name": template_container.get("name"),
+        "layers": get_image_layers(image),
+        "env_rules": env_rules,
+        "mounts": mounts,
+        **image_config,
+    })

src/confcom/azext_confcom/commands.py

@@ -13,3 +13,6 @@ def load_command_table(self, _):
 
     with self.command_group("confcom"):
         pass
+
+    with self.command_group("confcom containers") as g:
+        g.custom_command("from_vn2", "containers_from_vn2")

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_vn2 import containers_from_vn2 as _containers_from_vn2
 from knack.log import get_logger
 from pkg_resources import parse_version
 
@@ -512,3 +513,13 @@ def get_fragment_output_type(outraw):
     if outraw:
         output_type = security_policy.OutputType.RAW
     return output_type
+
+
+def containers_from_vn2(
+    template: str,
+    container_name: str,
+) -> None:
+    print(_containers_from_vn2(
+        template=template,
+        container_name=container_name,
+    ))

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,57 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+VN2_MOUNTS = [
+    {
+        "destination": "/etc/resolv.conf",
+        "options": [
+            "rbind",
+            "rshared",
+            "rw"
+        ],
+        "source": "sandbox:///tmp/atlas/emptydir/.+",
+        "type": "bind"
+    },
+    {
+        "destination": "/var/run/secrets/kubernetes.io/serviceaccount",
+        "options": [
+            "rbind",
+            "rshared",
+            "ro"
+        ],
+        "source": "sandbox:///tmp/atlas/emptydir/.+",
+        "type": "bind"
+    },
+    {
+        "destination": "/etc/hosts",
+        "options": [
+            "rbind",
+            "rshared",
+            "rw"
+        ],
+        "source": "sandbox:///tmp/atlas/emptydir/.+",
+        "type": "bind"
+    },
+    {
+        "destination": "/dev/termination-log",
+        "options": [
+            "rbind",
+            "rshared",
+            "rw"
+        ],
+        "source": "sandbox:///tmp/atlas/emptydir/.+",
+        "type": "bind"
+    },
+    {
+        "destination": "/etc/hostname",
+        "options": [
+            "rbind",
+            "rshared",
+            "rw"
+        ],
+        "source": "sandbox:///tmp/atlas/emptydir/.+",
+        "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