Add a --with-containers flag to policy and fragment gen

Author: DomAyre

src/confcom/.gitignore

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

src/confcom/azext_confcom/_params.py

@@ -198,6 +198,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',
+            required=False,
+            default=[],
+            help='Container definitions to include in the policy'
+        )
 
     with self.argument_context("confcom acifragmentgen") as c:
         c.argument(
@@ -345,6 +353,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,
+            default=[],
+            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 != [],
     ])) != 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 != [],
+    ])) != 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: list,
     approve_wildcards: str = False,
     outraw: bool = False,
     outraw_pretty_print: bool = False,
@@ -147,6 +148,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 +238,7 @@ def acifragmentgen_confcom(
     key: str,
     chain: str,
     minimum_svn: str,
+    container_definitions: list,
     image_target: str = "",
     algo: str = "ES384",
     fragment_path: str = None,
@@ -299,13 +311,24 @@ 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,
+        )
+
     # get all of the fragments that are being used in the policy
     # and associate them with each container group
     fragment_policy_list = []
@@ -321,7 +344,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,124 @@
+# --------------------------------------------------------------------------------------------
+# 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)
+
+
+POLICY_CLASSES = [
+    cls
+    for _, cls in inspect.getmembers(sys.modules[__name__], inspect.isclass)
+    if is_dataclass(cls) and cls.__module__ == sys.modules[__name__].__name__
+]

src/confcom/azext_confcom/security_policy.py

@@ -4,13 +4,16 @@
 # --------------------------------------------------------------------------------------------
 
 import copy
+from dataclasses import asdict
 import json
+import tempfile
 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 +68,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 +78,7 @@ 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 []
 
         if debug_mode:
             self._allow_properties_access = config.DEBUG_MODE_SETTINGS.get(
@@ -399,6 +404,12 @@ 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
 
+        if self._container_definitions:
+            policy += [
+                asdict(Container(**json.loads(c)))
+                for c in self._container_definitions
+            ]
+
         if pretty_print:
             return pretty_print_func(policy)
         return print_func(policy)