adding capabilities array for ARM input

src/confcom/azext_confcom/config.py

@@ -66,6 +66,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
@@ -98,6 +102,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_SECCOMP_PROFILE_SHA256 = "seccomp_profile_sha256"
 POLICY_FIELD_CONTAINERS_ELEMENTS_ALLOW_STDIO_ACCESS = "allow_stdio_access"
@@ -158,3 +168,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

@@ -242,7 +242,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
@@ -278,12 +278,118 @@ 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"
             }
+
     return user
 
+
+def extract_capabilities(container_json):
+    security_context = case_insensitive_dict_get(
+        container_json, config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT
+    )
+    # 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:
+        non_added_fields = [
+                config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_BOUNDING,
+                config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_EFFECTIVE,
+                config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES_PERMITTED,
+            ]
+
+        # add the default capabilities to the output
+        for key in non_added_fields:
+            output_capabilities[key] = copy.deepcopy(config.DEFAULT_UNPRIVILEGED_CAPABILITIES)
+        # 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.'
+                )
+
+            # 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))
+
+    return output_capabilities
+
 def extract_seccomp_profile_sha256(container_json: Any) -> Dict:
     security_context = case_insensitive_dict_get(
         container_json, config.ACI_FIELD_CONTAINERS_SECURITY_CONTEXT
-    ) 
+    )
 
     seccomp_profile_sha256 = ""
     # assumes that securityContext field is optional
@@ -307,22 +413,27 @@ def extract_seccomp_profile_sha256(container_json: Any) -> Dict:
 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 = False
     # assumes that securityContext field is optional
     if security_context:
-        try: 
-            # get the field for allow privilege escalation, default to false
-            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 false
+        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
 
 
@@ -372,6 +483,7 @@ def from_json(
         )
         signals = extract_get_signals(container_json)
         user = extract_user(container_json)
+        capabilities = extract_capabilities(container_json)
         seccomp_profile_sha256 = extract_seccomp_profile_sha256(container_json)
         allow_stdio_access = extract_allow_stdio_access(container_json)
         allow_privilege_escalation = extract_allow_privilege_escalation(container_json)
@@ -386,6 +498,7 @@ def from_json(
             execProcesses=exec_processes,
             signals=signals,
             user=user,
+            capabilities=capabilities,
             seccomp_profile_sha256=seccomp_profile_sha256,
             allowStdioAccess=allow_stdio_access,
             allowPrivilegeEscalation=allow_privilege_escalation,
@@ -402,10 +515,11 @@ def __init__(
         allow_elevated: bool,
         id_val: str,
         extraEnvironmentRules: Dict,
+        capabilities: Dict,
         user: Dict = copy.deepcopy(_DEFAULT_USER),
         seccomp_profile_sha256: str = "",
         allowStdioAccess: bool = True,
-        allowPrivilegeEscalation: bool = True,
+        allowPrivilegeEscalation: bool = False,
         execProcesses: List = None,
         signals: List = None,
     ) -> None:
@@ -423,6 +537,7 @@ def __init__(
         self._allow_stdio_access = allowStdioAccess
         self._seccomp_profile_sha256 = seccomp_profile_sha256
         self._user = user or {}
+        self._capabilities = capabilities
         self._allow_privilege_escalation = allowPrivilegeEscalation
         self._policy_json = None
         self._policy_json_str = None
@@ -564,6 +679,7 @@ def _populate_policy_json_elements(self) -> Dict[str, Any]:
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_EXEC_PROCESSES: self._exec_processes,
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_SIGNAL_CONTAINER_PROCESSES: self._signals,
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_USER: self.get_user(),
+            config.POLICY_FIELD_CONTAINERS_ELEMENTS_CAPABILITIES: self._capabilities,
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_SECCOMP_PROFILE_SHA256: self._seccomp_profile_sha256,
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_ALLOW_STDIO_ACCESS: self._allow_stdio_access,
             config.POLICY_FIELD_CONTAINERS_ELEMENTS_NO_NEW_PRIVILEGES: not self._allow_privilege_escalation

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"
+    ]
 }