[confcom] Add a --with-containers flag to policy and fragment gen

src/confcom/.gitignore

@@ -36,3 +36,5 @@ azext_confcom/bin/*
 **/.coverage
 
 **/htmlcov
+
+!lib/

src/confcom/HISTORY.rst

@@ -3,6 +3,10 @@
 Release History
 ===============
 
+1.4.0
+++++++
+* Add --with-containers flag to acipolicygen and acifragmentgen to allow passing container policy definitions directly
+
 1.3.1
 ++++++
 * bugfix for --exclude-default-fragments flag not working as intended

src/confcom/azext_confcom/_params.py

@@ -4,6 +4,7 @@
 # --------------------------------------------------------------------------------------------
 # pylint: disable=line-too-long
 
+import json
 from knack.arguments import CLIArgumentType
 from azext_confcom._validators import (
     validate_params_file,
@@ -198,6 +199,14 @@ def load_arguments(self, _):
             required=False,
             help="Exclude default fragments in the generated policy",
         )
+        c.argument(
+            "container_definitions",
+            options_list=['--with-containers'],
+            action='append',
+            type=json.loads,
+            required=False,
+            help='Container definitions to include in the policy'
+        )
 
     with self.argument_context("confcom acifragmentgen") as c:
         c.argument(
@@ -345,6 +354,14 @@ def load_arguments(self, _):
             help="Path to JSON file to write fragment import information. This is used with --generate-import. If not specified, the import statement will print to the console",
             validator=validate_fragment_json,
         )
+        c.argument(
+            "container_definitions",
+            options_list=['--with-containers'],
+            action='append',
+            required=False,
+            type=json.loads,
+            help='Container definitions to include in the policy'
+        )
 
     with self.argument_context("confcom katapolicygen") as c:
         c.argument(

src/confcom/azext_confcom/_validators.py

@@ -38,7 +38,8 @@ def validate_aci_source(namespace):
         namespace.input_path,
         namespace.arm_template,
         namespace.image_name,
-        namespace.virtual_node_yaml_path
+        namespace.virtual_node_yaml_path,
+        namespace.container_definitions is not None,
     ])) != 1:
         raise CLIError("Can only generate CCE policy from one source at a time")
 
@@ -71,7 +72,11 @@ def validate_fragment_key_and_chain(namespace):
 
 
 def validate_fragment_source(namespace):
-    if not namespace.generate_import and sum(map(bool, [namespace.image_name, namespace.input_path])) != 1:
+    if not namespace.generate_import and sum(map(bool, [
+        namespace.image_name,
+        namespace.input_path,
+        namespace.container_definitions is not None,
+    ])) != 1:
         raise CLIError("Must provide either an image name or an input file to generate a fragment")
 
 

src/confcom/azext_confcom/custom.py

@@ -11,13 +11,13 @@
 from azext_confcom._validators import resolve_stdio
 from azext_confcom.config import (
     DEFAULT_REGO_FRAGMENTS, POLICY_FIELD_CONTAINERS_ELEMENTS_REGO_FRAGMENTS,
-    REGO_IMPORT_FILE_STRUCTURE)
+    REGO_IMPORT_FILE_STRUCTURE, ACI_FIELD_VERSION, ACI_FIELD_CONTAINERS)
 from azext_confcom.cose_proxy import CoseSignToolProxy
 from azext_confcom.errors import eprint
 from azext_confcom.fragment_util import get_all_fragment_contents
 from azext_confcom.init_checks import run_initial_docker_checks
 from azext_confcom.kata_proxy import KataPolicyGenProxy
-from azext_confcom.security_policy import OutputType
+from azext_confcom.security_policy import AciPolicy, OutputType
 from azext_confcom.template_util import (
     get_image_name, inject_policy_into_template, inject_policy_into_yaml,
     pretty_print_func, print_existing_policy_from_arm_template,
@@ -37,6 +37,7 @@ def acipolicygen_confcom(
     virtual_node_yaml_path: str,
     infrastructure_svn: str,
     tar_mapping_location: str,
+    container_definitions: Optional[list] = None,
     approve_wildcards: str = False,
     outraw: bool = False,
     outraw_pretty_print: bool = False,
@@ -64,6 +65,9 @@ def acipolicygen_confcom(
             "For additional information, see http://aka.ms/clisecrets. \n",
         )
 
+    if container_definitions is None:
+        container_definitions = []
+
     stdio_enabled = resolve_stdio(enable_stdio, disable_stdio)
 
     if print_existing_policy and arm_template:
@@ -147,6 +151,16 @@ def acipolicygen_confcom(
             exclude_default_fragments=exclude_default_fragments,
             infrastructure_svn=infrastructure_svn,
         )
+    elif container_definitions:
+        container_group_policies = AciPolicy(
+            {
+                ACI_FIELD_VERSION: "1.0",
+                ACI_FIELD_CONTAINERS: [],
+            },
+            debug_mode=debug_mode,
+            disable_stdio=disable_stdio,
+            container_definitions=container_definitions,
+        )
 
     exit_code = 0
 
@@ -227,6 +241,7 @@ def acifragmentgen_confcom(
     key: str,
     chain: str,
     minimum_svn: str,
+    container_definitions: Optional[list] = None,
     image_target: str = "",
     algo: str = "ES384",
     fragment_path: str = None,
@@ -241,6 +256,8 @@ def acifragmentgen_confcom(
     no_print: bool = False,
     fragments_json: str = "",
 ):
+    if container_definitions is None:
+        container_definitions = []
 
     stdio_enabled = resolve_stdio(enable_stdio, disable_stdio)
 
@@ -299,13 +316,27 @@ def acifragmentgen_confcom(
         policy = security_policy.load_policy_from_image_name(
             image_name, debug_mode=debug_mode, disable_stdio=(not stdio_enabled)
         )
-    else:
+    elif input_path:
         # this is using --input
         if not tar_mapping:
             tar_mapping = os_util.load_tar_mapping_from_config_file(input_path)
         policy = security_policy.load_policy_from_json_file(
             input_path, debug_mode=debug_mode, disable_stdio=(not stdio_enabled)
         )
+    elif container_definitions:
+        policy = AciPolicy(
+            {
+                ACI_FIELD_VERSION: "1.0",
+                ACI_FIELD_CONTAINERS: [],
+            },
+            debug_mode=debug_mode,
+            disable_stdio=disable_stdio,
+            container_definitions=container_definitions,
+        )
+    else:
+        eprint("Either --image-name, --input, or --container-definitions must be provided", exit_code=2)
+        return
+
     # get all of the fragments that are being used in the policy
     # and associate them with each container group
     fragment_policy_list = []
@@ -321,7 +352,7 @@ def acifragmentgen_confcom(
 
     # make sure we have images to generate a fragment
     policy_images = policy.get_images()
-    if not policy_images:
+    if not policy_images and not container_definitions:
         eprint("No images found in the policy or all images are covered by fragments")
 
     if not feed:

src/confcom/azext_confcom/lib/policy.py

@@ -0,0 +1,118 @@
+# --------------------------------------------------------------------------------------------
+# 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 dataclass, field, is_dataclass
+import inspect
+import sys
+from typing import Literal, Optional
+
+
+def get_default_capabilities():
+    return [
+        "CAP_AUDIT_WRITE",
+        "CAP_CHOWN",
+        "CAP_DAC_OVERRIDE",
+        "CAP_FOWNER",
+        "CAP_FSETID",
+        "CAP_KILL",
+        "CAP_MKNOD",
+        "CAP_NET_BIND_SERVICE",
+        "CAP_NET_RAW",
+        "CAP_SETFCAP",
+        "CAP_SETGID",
+        "CAP_SETPCAP",
+        "CAP_SETUID",
+        "CAP_SYS_CHROOT"
+    ]
+
+
+@dataclass
+class ContainerCapabilities:
+    ambient: list[str] = field(default_factory=list)
+    bounding: list[str] = field(default_factory=get_default_capabilities)
+    effective: list[str] = field(default_factory=get_default_capabilities)
+    inheritable: list[str] = field(default_factory=list)
+    permitted: list[str] = field(default_factory=get_default_capabilities)
+
+
+@dataclass
+class ContainerRule:
+    pattern: str
+    strategy: str
+    required: Optional[bool] = False
+
+
+@dataclass
+class ContainerExecProcesses:
+    command: list[str]
+    signals: Optional[list[str]] = None
+    allow_stdio_access: bool = True
+
+
+@dataclass
+class ContainerMount:
+    destination: str
+    source: str
+    type: str
+    options: list[str] = field(default_factory=list)
+
+
+@dataclass
+class ContainerUser:
+    group_idnames: list[ContainerRule] = field(default_factory=lambda: [ContainerRule(pattern="", strategy="any")])
+    umask: str = "0022"
+    user_idname: ContainerRule = field(default_factory=lambda: ContainerRule(pattern="", strategy="any"))
+
+
+@dataclass
+class FragmentReference:
+    feed: str
+    issuer: str
+    minimum_svn: str
+    includes: list[Literal["containers", "fragments", "namespace", "external_processes"]]
+    path: Optional[str] = None
+
+
+@dataclass
+class Container:
+    allow_elevated: bool = False
+    allow_stdio_access: bool = True
+    capabilities: ContainerCapabilities = field(default_factory=ContainerCapabilities)
+    command: Optional[list[str]] = None
+    env_rules: list[ContainerRule] = field(default_factory=list)
+    exec_processes: list[ContainerExecProcesses] = field(default_factory=list)
+    id: Optional[str] = None
+    layers: list[str] = field(default_factory=list)
+    mounts: list[ContainerMount] = field(default_factory=list)
+    name: Optional[str] = None
+    no_new_privileges: bool = False
+    seccomp_profile_sha256: str = ""
+    signals: list[str] = field(default_factory=list)
+    user: ContainerUser = field(default_factory=ContainerUser)
+    working_dir: str = "/"
+
+
+@dataclass
+class Policy:
+    package: str = "policy"
+    api_version: str = "0.10.0"
+    framework_version: str = "0.2.3"
+    fragments: list[FragmentReference] = field(default_factory=list)
+    containers: list[Container] = field(default_factory=list)
+    allow_properties_access: bool = True
+    allow_dump_stacks: bool = False
+    allow_runtime_logging: bool = False
+    allow_environment_variable_dropping: bool = True
+    allow_unencrypted_scratch: bool = False
+    allow_capability_dropping: bool = True
+
+
+@dataclass
+class Fragment:
+    package: str = "fragment"
+    svn: str = "0"
+    framework_version: str = "0.2.3"
+    fragments: list[FragmentReference] = field(default_factory=list)
+    containers: list[Container] = field(default_factory=list)

src/confcom/azext_confcom/security_policy.py

@@ -4,13 +4,15 @@
 # --------------------------------------------------------------------------------------------
 
 import copy
+from dataclasses import asdict
 import json
 import warnings
 from enum import Enum, auto
-from typing import Any, Dict, List, Tuple, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 import deepdiff
 from azext_confcom import config, os_util
+from azext_confcom.lib.policy import Container
 from azext_confcom.container import ContainerImage, UserContainerImage
 from azext_confcom.errors import eprint
 from azext_confcom.fragment_util import sanitize_fragment_fields
@@ -65,6 +67,7 @@ def __init__(
         disable_stdio: bool = False,
         is_vn2: bool = False,
         fragment_contents: Any = None,
+        container_definitions: Optional[list] = None,
     ) -> None:
         self._rootfs_proxy = None
         self._policy_str = None
@@ -74,6 +77,15 @@ def __init__(
         self._existing_fragments = existing_rego_fragments
         self._api_version = config.API_VERSION
         self._fragment_contents = fragment_contents
+        self._container_definitions = container_definitions or []
+
+        self._container_definitions = []
+        if container_definitions:
+            for container_definition in container_definitions:
+                if isinstance(container_definition, list):
+                    self._container_definitions.extend(container_definition)
+                else:
+                    self._container_definitions.append(container_definition)
 
         if debug_mode:
             self._allow_properties_access = config.DEBUG_MODE_SETTINGS.get(
@@ -399,6 +411,8 @@ def _policy_serialization(self, pretty_print=False, include_sidecars: bool = Tru
                 for container in policy:
                     container[config.POLICY_FIELD_CONTAINERS_ELEMENTS_ALLOW_STDIO_ACCESS] = False
 
+        policy += [asdict(Container(**c)) for c in self._container_definitions]
+
         if pretty_print:
             return pretty_print_func(policy)
         return print_func(policy)

src/confcom/azext_confcom/tests/latest/test_confcom_acipolicygen_arm.py

@@ -252,4 +252,45 @@ def test_acipolicygen_arm_diff_with_allow_all():
         }
 
 
+@pytest.mark.parametrize(
+    "container_definitions",
+    [
+        ["{}"], # Single empty container definition (use all default values)
+        ["{}", "{}"], # Two empty container definitions
+        ["[{}]", "{}"], # Two empty container definitions, one in subarray
+        ["[{}, {}]", "{}"], # Three empty container definitions, two in subarray
+        ['{"id": "test"}'], # Single container definition a field changed
+    ]
+)
+def test_acipolicygen_with_containers(container_definitions):
+
+    acipolicygen_confcom(
+        input_path=None,
+        arm_template=None,
+        arm_template_parameters=None,
+        image_name=None,
+        virtual_node_yaml_path=None,
+        infrastructure_svn=None,
+        tar_mapping_location=None,
+        outraw=True,
+        container_definitions=[json.loads(c) for c in container_definitions]
+    )
 
+
+def test_acipolicygen_with_containers_field_changed():
+
+    buffer = io.StringIO()
+    with contextlib.redirect_stdout(buffer):
+        acipolicygen_confcom(
+            input_path=None,
+            arm_template=None,
+            arm_template_parameters=None,
+            image_name=None,
+            virtual_node_yaml_path=None,
+            infrastructure_svn=None,
+            tar_mapping_location=None,
+            outraw=True,
+            container_definitions=[json.loads('{"id": "test"}')]
+        )
+    actual_policy = buffer.getvalue()
+    assert '"id":"test"' in actual_policy