[confcom] Add commands to support radius policy generation

linter_exclusions.yml

@@ -3504,3 +3504,15 @@ neon postgres organization:
 neon postgres project:
   rule_exclusions:
   - require_wait_command_if_no_wait
+
+confcom containers from_radius:
+  parameters:
+    template:
+      rule_exclusions:
+        - no_positional_parameters
+
+confcom radius policy insert:
+  parameters:
+    policy_file:
+      rule_exclusions:
+        - no_positional_parameters

src/confcom/azext_confcom/_help.py

@@ -278,3 +278,83 @@
         - 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_radius"
+] = """
+    type: command
+    short-summary: Create a Security Policy Container Definition based on a Radius app 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'
+
+        - name: --platform
+          type: str
+          short-summary: 'The name of the platform the container definition will run on'
+
+
+    examples:
+        - name: Input a Bicep Template and generate container definitions
+          text: az confcom containers from_radius app.bicep
+        - name: Input a Bicep Template with a bicepparam file and generate container definitions
+          text: az confcom containers from_radius app.bicep --parameters app.bicepparam
+        - name: Input a Bicep Template with inline parameter and generate container definitions
+          text: az confcom containers from_radius app.bicep --parameters image=my.azurecr.io/myimage:tag
+        - name: Input a Bicep Template and generate container definitions for the second container resource
+          text: az confcom containers from_radius app.bicep --idx 1
+"""
+
+
+helps[
+    "confcom radius"
+] = """
+    type: group
+    short-summary: Commands related to Radius.
+"""
+
+
+helps[
+    "confcom radius policy"
+] = """
+    type: group
+    short-summary: Commands related to Radius policies.
+"""
+
+
+helps[
+    "confcom radius policy insert"
+] = """
+    type: command
+    short-summary: Inserts a Security Policy into a Radius app template.
+
+    parameters:
+        - name: --template-file -f
+          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: Insert a Security Policy into a Radius app template
+          text: az confcom radius policy insert policy.rego -f app.bicep
+        - name: Insert a Security Policy into a Radius app template for the second container resource
+          text: az confcom radius policy insert policy.rego -f app.bicep --idx 1
+"""

src/confcom/azext_confcom/_params.py

@@ -4,8 +4,11 @@
 # --------------------------------------------------------------------------------------------
 # pylint: disable=line-too-long
 
+import argparse
 import json
+import sys
 from knack.arguments import CLIArgumentType
+from argcomplete.completers import FilesCompleter
 from azext_confcom._validators import (
     validate_params_file,
     validate_diff,
@@ -434,3 +437,59 @@ 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_radius") 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 radius template'
+        )
+        c.argument(
+            "container_index",
+            options_list=['--idx'],
+            required=False,
+            default=0,
+            type=int,
+            help='The index of the container definition in the template to use'
+        )
+        c.argument(
+            "platform",
+            options_list=["--platform"],
+            required=False,
+            default="aci",
+            type=str,
+            help="Platform to create container definition for",
+        )
+
+    with self.argument_context("confcom radius policy insert") as c:
+        c.positional(
+            "policy_file",
+            nargs='?',
+            type=argparse.FileType('rb'),
+            default=sys.stdin.buffer,
+            help="Policy to add to the template",
+        )
+        c.argument(
+            "template_path",
+            options_list=("--template-file", '-f'),
+            required=False,
+            help="Path to the template file"
+        )
+        c.argument(
+            "container_index",
+            options_list=['--idx'],
+            required=False,
+            default=0,
+            type=int,
+            help='The index of the container definition in the template to use'
+        )

src/confcom/azext_confcom/command/containers_from_radius.py

@@ -0,0 +1,105 @@
+# --------------------------------------------------------------------------------------------
+# 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 os
+import tempfile
+
+from azext_confcom.lib.images import get_image_config, get_image_layers
+from azext_confcom.lib.deployments import parse_deployment_template
+from azext_confcom.lib.platform import ACI_MOUNTS
+import re
+
+
+def containers_from_radius(
+    az_cli_command,
+    template: str,
+    parameters: list,
+    container_index: int,
+    platform: str,
+) -> None:
+
+    # Remove the radius extension inclusion to avoid parsing errors
+    # For the purpose of extracting the container info we don't care about it
+    with tempfile.NamedTemporaryFile('w+', delete=True, suffix=".bicep") as temp_template_file:
+        with open(template, 'r') as f:
+            temp_template_file.write(f.read().replace("extension radius", ""))
+        temp_template_file.flush()
+
+        # Handle parameters file if it's a path
+        if len(parameters) > 0 and isinstance(parameters[0][0], str) and os.path.isfile(parameters[0][0]):
+            parameters_path = parameters[0][0]
+            with open(parameters_path, 'r') as params_file:
+                params_content = params_file.read()
+
+            # Replace any references to the original template file with the temporary one
+            params_content = re.sub(
+                r"using\s+'.*\.bicep'",
+                f"using '{os.path.basename(temp_template_file.name)}'",
+                params_content
+            )
+
+            with tempfile.NamedTemporaryFile('w+', delete=False, suffix=".bicepparam") as temp_params_file:
+                temp_params_file.write(params_content)
+                temp_params_file.flush()
+                parameters = [[temp_params_file.name]]
+
+        template = parse_deployment_template(
+            az_cli_command,
+            temp_template_file.name,
+            parameters,
+        )
+
+    supported_resources = [r for r in template.get("resources", []) if r.get("type") in {
+        "Applications.Core/containers",
+    }]
+
+    resource = supported_resources[container_index].get("properties", {})
+    container = resource.get("container", {})
+    image = container.get("image")
+
+    mounts = {
+        "aci": ACI_MOUNTS,
+    }.get(platform, None)
+
+    mounts += [
+        {
+            "destination": mount_info["mountPath"],
+            "options": ["rbind", "rshared", "ro"],
+            "source": mount_info["source"],
+            "type": "bind"
+        }
+        for mount_info in container.get("volumes", {}).values()
+    ]
+
+    image_config = get_image_config(image)
+
+    env_rules = image_config.pop("env_rules", [])
+    env_rules += [
+        {
+            "pattern": f'{k}={v["value"]}',
+            "strategy": "string",
+            "required": False,
+        }
+        for k, v in container.get("env", {}).items()
+    ]
+    env_rules += [
+        {
+            "name": f"CONNECTIONS_{k.upper()}_.+",
+            "value": ".+",
+            "strategy": "re2",
+            "required": True,
+        }
+        for k in resource.get("connections", {}).keys()
+    ]
+
+    return json.dumps({
+        "id": image,
+        "name": image,
+        "layers": get_image_layers(image),
+        **({"mounts": mounts} if mounts else {}),
+        "env_rules": env_rules,
+        **image_config,
+    })

src/confcom/azext_confcom/command/radius_policy_insert.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 tempfile
+from typing import BinaryIO
+from azext_confcom.lib.serialization import policy_deserialize, policy_serialize
+from azext_confcom.lib.policy import Policy, Container
+import re
+import base64
+
+
+def radius_policy_insert(
+    policy_file: BinaryIO,
+    template_path: str,
+    container_index: int,
+) -> str:
+
+    policy = None
+    if policy_file.name == "<stdin>":
+        with tempfile.NamedTemporaryFile(delete=True) as temp_policy_file:
+            temp_policy_file.write(policy_file.read())
+            temp_policy_file.flush()
+            policy = policy_deserialize(temp_policy_file.name)
+    else:
+        policy = policy_deserialize(policy_file.name)
+
+    serialized_policy = policy_serialize(policy)
+
+    # Read the template file
+    with open(template_path, 'r') as template_file:
+        template_content = template_file.read()
+
+    # Base64 encode the serialized policy
+    encoded_policy = base64.b64encode(serialized_policy.encode()).decode()
+
+    # Replace the nth ccePolicy value with the encoded policy, preserving original quote types
+    def replace_cce_policy(match):
+        full_match = match.group(0)
+        # Extract the key part (before the colon)
+        key_part = re.match(r'(["\']?)ccePolicy\1\s*:', full_match).group(0)
+        # Extract the quote type used for the value
+        value_quote_match = re.search(r':\s*(["\'])', full_match)
+        value_quote = value_quote_match.group(1) if value_quote_match else '"'
+        return f'{key_part} {value_quote}{encoded_policy}{value_quote}'
+
+    # Find all matches and replace only the nth instance
+    pattern = r'(["\']?)ccePolicy\1\s*:\s*["\'][^"\']*["\']'
+    matches = list(re.finditer(pattern, template_content))
+
+    if len(matches) >= container_index:
+        # Replace only the nth match (convert to 0-based index)
+        target_match = matches[container_index]
+        start, end = target_match.span()
+        replacement = replace_cce_policy(target_match)
+        updated_content = template_content[:start] + replacement + template_content[end:]
+    else:
+        updated_content = template_content
+
+    # Write the updated content back to the template file
+    with open(template_path, 'w') as template_file:
+        template_file.write(updated_content)

src/confcom/azext_confcom/commands.py

@@ -13,3 +13,9 @@ def load_command_table(self, _):
 
     with self.command_group("confcom"):
         pass
+
+    with self.command_group("confcom containers") as g:
+        g.custom_command("from_radius", "containers_from_radius")
+
+    with self.command_group("confcom radius policy") as g:
+        g.custom_command("insert", "radius_policy_insert")