adding capabilities array for ARM input

src/confcom/azext_confcom/config.py

@@ -65,6 +65,10 @@
 ACI_FIELD_TEMPLATE_MOUNTS_READONLY = "readOnly"
 ACI_FIELD_TEMPLATE_CONFCOM_PROPERTIES = "confidentialComputeProperties"
 ACI_FIELD_TEMPLATE_CCE_POLICY = "ccePolicy"
+ACI_FIELD_CONTAINERS_PRIVILEGED = "privileged"
+ACI_FIELD_CONTAINERS_CAPABILITIES = "capabilities"
+ACI_FIELD_CONTAINERS_CAPABILITIES_ADD = "add"
+ACI_FIELD_CONTAINERS_CAPABILITIES_DROP = "drop"
 
 
 # output json values
@@ -97,6 +101,12 @@
 POLICY_FIELD_CONTAINERS_ELEMENTS_USER_GROUP_IDNAMES = "group_idnames"
 POLICY_FIELD_CONTAINERS_ELEMENTS_USER_UMASK = "umask"
 POLICY_FIELD_CONTAINERS_ELEMENTS_USER_PATTERN = "pattern"
+POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES = "capabilities"
+POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_BOUNDING = "bounding"
+POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_EFFECTIVE = "effective"
+POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_INHERITABLE = "inheritable"
+POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_PERMITTED = "permitted"
+POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_AMBIENT = "ambient"
 POLICY_FIELD_CONTAINERS_ELEMENTS_USER_STRATEGY = "strategy"
 POLICY_FIELD_CONTAINERS_ELEMENTS_ALLOW_STDIO_ACCESS = "allow_stdio_access"
 POLICY_FIELD_CONTAINERS_ELEMENTS_REGO_FRAGMENTS = "fragments"
@@ -156,3 +166,7 @@
 DEFAULT_CONTAINERS = _config["default_containers"]
 # default container user config to be added for security context
 DEFAULT_USER = _config["default_user"]
+# default unpriviliged user capabilities to be added for security context
+DEFAULT_UNPRIVILEGED_CAPABILITIES = _config["default_unprivileged_capabilities"]
+# default priviliged user capabilities to be added for security context
+DEFAULT_PRIVILEGED_CAPABILITIES = _config["default_privileged_capabilities"]

src/confcom/azext_confcom/container.py

@@ -243,7 +243,7 @@ def extract_allow_stdio_access(container_json: Any) -> bool:
 def extract_user(container_json: Any) -> Dict:
     security_context = case_insensitive_dict_get(
         container_json, config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT
-    ) 
+    )
 
     user = copy.deepcopy(_DEFAULT_USER)
     # assumes that securityContext field is optional
@@ -279,28 +279,136 @@ def extract_user(container_json: Any) -> Dict:
                 config.POLICY_FIELD_CONTAINERS_ELEMENTS_USER_PATTERN: str(run_as_group_value),
                 config.POLICY_FIELD_CONTAINERS_ELEMENTS_USER_STRATEGY: "id"
             }
+
+        # get the field for privileged
+        privileged_value = case_insensitive_dict_get(
+            security_context, config.ACI_FIELD_CONTAINERS_PRIVILEGED
+        )
+        if not isinstance(privileged_value, bool) and not isinstance(privileged_value, str):
+            eprint(
+                f'Field ["{config.ACI_FIELD_CONTAINERS}"]["{config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT}"]'
+                + f'["{config.ACI_FIELD_CONTAINERS_PRIVILEGED}"] can only be a boolean or string value.'
+            )
+
+        # force the field into a bool
+        if isinstance(privileged_value, str):
+            privileged_value = privileged_value.lower() == "true"
+
+        output_capabilities = {
+            config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_BOUNDING: [],
+            config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_EFFECTIVE: [],
+            config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_INHERITABLE: [],
+            config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_PERMITTED: [],
+            config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_AMBIENT: [],
+        }
+
+
+        # if privileged is true, then set all capabilities to true
+        # else, get the capabilities field from the ARM Template
+        if privileged_value:
+            for key in output_capabilities.keys():
+                output_capabilities[key] = copy.deepcopy(config.DEFAULT_PRIVILEGED_CAPABILITIES)
+        else:
+            # get the capabilities field
+            capabilities = case_insensitive_dict_get(
+                security_context, config.ACI_FIELD_CONTAINERS_CAPABILITIES
+            )
+            if capabilities:
+                # error check if capabilities is not a dict
+                if not isinstance(capabilities, dict):
+                    eprint(
+                        f'Field ["{config.ACI_FIELD_CONTAINERS}"]["{config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT}"]'
+                        + f'["{config.ACI_FIELD_CONTAINERS_CAPABILITIES}"] can only be a dictionary.'
+                    )
+
+                non_added_fields = [
+                    config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_BOUNDING,
+                    config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_EFFECTIVE,
+                    config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_PERMITTED,
+                ]
+
+                # drop the capabilities from the output
+                for key in non_added_fields:
+                    output_capabilities[key] = copy.deepcopy(config.DEFAULT_UNPRIVILEGED_CAPABILITIES)
+
+                # get the add field
+                add = case_insensitive_dict_get(
+                    capabilities, config.ACI_FIELD_CONTAINERS_CAPABILITIES_ADD
+                )
+                if add:
+                    # error check if add is not a list
+                    if not isinstance(add, list):
+                        eprint(
+                            f'Field ["{config.ACI_FIELD_CONTAINERS}"]["{config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT}"]'
+                            + f'["{config.ACI_FIELD_CONTAINERS_CAPABILITIES_ADD}"] can only be a list.'
+                        )
+
+                    # add the capabilities to the output
+                    for value in output_capabilities.values():
+                        for capability in add:
+                            if not isinstance(capability, str):
+                                eprint(
+                                    f'Field ["{config.ACI_FIELD_CONTAINERS}"]["{config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT}"]'
+                                    + f'["{config.ACI_FIELD_CONTAINERS_CAPABILITIES_ADD}"] can only contain strings.'
+                                )
+                            value.append(capability)
+
+                # get the drop field
+                drop = case_insensitive_dict_get(
+                    capabilities, config.ACI_FIELD_CONTAINERS_CAPABILITIES_DROP
+                )
+                if drop:
+                    # error check if drop is not a list
+                    if not isinstance(drop, list):
+                        eprint(
+                            f'Field ["{config.ACI_FIELD_CONTAINERS}"]["{config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT}"]'
+                            + f'["{config.ACI_FIELD_CONTAINERS_CAPABILITIES_DROP}"] can only be a list.'
+                        )
+
+                    # drop the capabilities from the output
+                    for value in non_added_fields:
+                        for capability in drop:
+                            if not isinstance(capability, str):
+                                eprint(
+                                    f'Field ["{config.ACI_FIELD_CONTAINERS}"]["{config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT}"]'
+                                    + f'["{config.ACI_FIELD_CONTAINERS_CAPABILITIES_DROP}"] can only contain strings.'
+                                )
+                            output_capabilities[value].append(capability)
+
+        # de-duplicate the capabilities
+        for key, value in output_capabilities.items():
+            output_capabilities[key] = list(set(value))
+
+        user[config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES] = output_capabilities
+
+
     return user
 
 
 def extract_allow_privilege_escalation(container_json: Any) -> bool:
     security_context = case_insensitive_dict_get(
         container_json, config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT
-    ) 
-            
+    )
+
     allow_privilege_escalation = True
     # assumes that securityContext field is optional
     if security_context:
-        try: 
-            # get the field for allow privilege escalation, default to true
-            allow_privilege_escalation_value = bool(case_insensitive_dict_get(
-                security_context, config.ACI_FIELD_CONTAINERS_ALLOW_PRIVILEGE_ESCALATION
-            ))
-            allow_privilege_escalation = allow_privilege_escalation_value
-        except ValueError:
+
+        # get the field for allow privilege escalation, default to true
+        allow_privilege_escalation = case_insensitive_dict_get(
+            security_context, config.ACI_FIELD_CONTAINERS_ALLOW_PRIVILEGE_ESCALATION
+        )
+
+        if not isinstance(allow_privilege_escalation, bool) and not isinstance(allow_privilege_escalation, str):
             eprint(
                 f'Field ["{config.ACI_FIELD_CONTAINERS}"]["{config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT}"]'
-                + f'["{config.ACI_FIELD_CONTAINERS_ALLOW_PRIVILEGE_ESCALATION}"] can only be a boolean value.'
+                + f'["{config.ACI_FIELD_CONTAINERS_PRIVILEGED}"] can only be a boolean or string value.'
             )
+
+        # force the field into a bool
+        if isinstance(allow_privilege_escalation, str):
+            allow_privilege_escalation = allow_privilege_escalation.lower() == "true"
+
     return allow_privilege_escalation
 
 
@@ -396,7 +504,7 @@ def __init__(
         self._mounts = mounts
         self._allow_elevated = allow_elevated
         self._allow_stdio_access = allowStdioAccess
-        self._user = user or {},
+        self._user = user or {}
         self._allow_privilege_escalation = allowPrivilegeEscalation
         self._policy_json = None
         self._policy_json_str = None

src/confcom/azext_confcom/data/internal_config.json

@@ -219,5 +219,64 @@
             }
         ],
         "umask": "0022"
-    }
+    },
+    "default_unprivileged_capabilities": [
+        "CAP_CHOWN",
+        "CAP_DAC_OVERRIDE",
+        "CAP_FSETID",
+        "CAP_FOWNER",
+        "CAP_MKNOD",
+        "CAP_NET_RAW",
+        "CAP_SETGID",
+        "CAP_SETUID",
+        "CAP_SETFCAP",
+        "CAP_SETPCAP",
+        "CAP_NET_BIND_SERVICE",
+        "CAP_SYS_CHROOT",
+        "CAP_KILL",
+        "CAP_AUDIT_WRITE"
+    ],
+    "default_privileged_capabilities": [
+        "CAP_CHOWN",
+        "CAP_DAC_OVERRIDE",
+        "CAP_DAC_READ_SEARCH",
+        "CAP_FOWNER",
+        "CAP_FSETID",
+        "CAP_KILL",
+        "CAP_SETGID",
+        "CAP_SETUID",
+        "CAP_SETPCAP",
+        "CAP_LINUX_IMMUTABLE",
+        "CAP_NET_BIND_SERVICE",
+        "CAP_NET_BROADCAST",
+        "CAP_NET_ADMIN",
+        "CAP_NET_RAW",
+        "CAP_IPC_LOCK",
+        "CAP_IPC_OWNER",
+        "CAP_SYS_MODULE",
+        "CAP_SYS_RAWIO",
+        "CAP_SYS_CHROOT",
+        "CAP_SYS_PTRACE",
+        "CAP_SYS_PACCT",
+        "CAP_SYS_ADMIN",
+        "CAP_SYS_BOOT",
+        "CAP_SYS_NICE",
+        "CAP_SYS_RESOURCE",
+        "CAP_SYS_TIME",
+        "CAP_SYS_TTY_CONFIG",
+        "CAP_MKNOD",
+        "CAP_LEASE",
+        "CAP_AUDIT_WRITE",
+        "CAP_AUDIT_CONTROL",
+        "CAP_SETFCAP",
+        "CAP_MAC_OVERRIDE",
+        "CAP_MAC_ADMIN",
+        "CAP_SYSLOG",
+        "CAP_WAKE_ALARM",
+        "CAP_BLOCK_SUSPEND",
+        "CAP_AUDIT_READ",
+        "CAP_PERFMON",
+        "CAP_BPF",
+        "CAP_CHECKPOINT_RESTORE"
+    ]
 }