feat: add ports and mounts support for containerd engine

Signed-off-by: Alessio Greggi <ale_grey_91@hotmail.it>

udica/__main__.py

@@ -18,7 +18,7 @@ import argparse
 
 # import udica
 from udica.parse import parse_avc_file
-from udica.parse import ENGINE_ALL, ENGINE_PODMAN, ENGINE_DOCKER
+from udica.parse import ENGINE_ALL, ENGINE_PODMAN, ENGINE_DOCKER, ENGINE_CONTAINERD
 from udica.version import version
 from udica import parse
 from udica.policy import create_policy, load_policy, generate_playbook

udica/parse.py

@@ -25,8 +25,11 @@ ENGINE_CRIO = "CRI-O"
 #: Constant for the docker engine
 ENGINE_DOCKER = "docker"
 
+#: Constant for the containerd engine
+ENGINE_CONTAINERD = "containerd"
+
 #: All supported engines
-ENGINE_ALL = [ENGINE_PODMAN, ENGINE_CRIO, ENGINE_DOCKER]
+ENGINE_ALL = [ENGINE_PODMAN, ENGINE_CRIO, ENGINE_DOCKER, ENGINE_CONTAINERD]
 
 
 # Decorator for verifying that getting value from "data" won't
@@ -79,6 +82,8 @@ def get_engine_helper(data, ContainerEngine):
             return PodmanHelper()
         elif engine == ENGINE_CRIO:
             return CrioHelper()
+        elif engine == ENGINE_CONTAINERD:
+            return ContainerdHelper()
         raise RuntimeError("Unkown engine")
 
 
@@ -204,6 +209,35 @@ class CrioHelper(EngineHelper):
         return []
 
 
+class ContainerdHelper(EngineHelper):
+    def __init__(self):
+        super().__init__(ENGINE_CONTAINERD)
+
+    @getter_decorator
+    def get_devices(self, data):
+        return []
+
+    @getter_decorator
+    def get_mounts(self, data):
+        return data[0]["Spec"]["mounts"]
+
+    @getter_decorator
+    def get_ports(self, data):
+        json_data = json.loads(data[0]["Labels"]["nerdctl/ports"])
+        ports = []
+        for port in json_data:
+            new_ports = {
+                "portNumber": port["HostPort"],
+                "protocol": port["Protocol"]
+            }
+            ports.append(new_ports)
+        return ports
+
+    @getter_decorator
+    def get_caps(self, data, opts):
+        return []
+
+
 def parse_cap(data):
     return data.decode().split("\n")[1].split(",")
 
@@ -254,6 +288,7 @@ def parse_avc_file(data):
 
 
 def validate_container_engine(ContainerEngine):
+    print(ContainerEngine)
     if ContainerEngine in ENGINE_ALL + ["CRIO", "-"]:
         # Fix CRIO reference to use ENGINE_CRIO
         if ContainerEngine == "CRIO":

udica/policy.py

@@ -178,6 +178,8 @@ def create_policy(
     # mounts
     if inspect_format == "CRI-O":
         write_policy_for_crio_mounts(mounts, policy)
+    elif inspect_format == "containerd":
+        write_policy_for_containerd_mounts(mounts, policy)
     else:
         write_policy_for_podman_mounts(mounts, policy)
 
@@ -429,6 +431,122 @@ def write_policy_for_podman_mounts(mounts, policy):
                         + " ))) \n"
                     )
 
+def write_policy_for_containerd_mounts(mounts, policy):
+    # mount JSON example:
+    # {
+    #   "destination": "/sys/fs/cgroup",
+    #   "type": "cgroup",
+    #   "source": "cgroup",
+    #   "options": [
+    #     "ro",
+    #     "nosuid",
+    #     "noexec",
+    #     "nodev"
+    #     ]
+    # }
+    for item in sorted(mounts, key=lambda x: str(x["source"])):
+        if not item["source"].find("/"):
+            if item["source"] == LOG_CONTAINER and "ro" in item["options"]:
+                policy.write("    (blockinherit log_container)\n")
+                add_template("log_container")
+                continue
+
+            if item["source"] == LOG_CONTAINER and "ro" not in item["options"]:
+                policy.write("    (blockinherit log_rw_container)\n")
+                add_template("log_container")
+                continue
+
+            if item["source"] == HOME_CONTAINER and "ro" in item["options"]:
+                policy.write("    (blockinherit home_container)\n")
+                add_template("home_container")
+                continue
+
+            if item["source"] == HOME_CONTAINER and "ro" not in item["options"]:
+                policy.write("    (blockinherit home_rw_container)\n")
+                add_template("home_container")
+                continue
+
+            if item["source"] == TMP_CONTAINER and "ro" in item["options"]:
+                policy.write("    (blockinherit tmp_container)\n")
+                add_template("tmp_container")
+                continue
+
+            if item["source"] == TMP_CONTAINER and "ro" not in item["options"]:
+                policy.write("    (blockinherit tmp_rw_container)\n")
+                add_template("tmp_container")
+                continue
+
+            if item["source"] == CONFIG_CONTAINER and "ro" in item["options"]:
+                policy.write("    (blockinherit config_container)\n")
+                add_template("config_container")
+                continue
+
+            if item["source"] == CONFIG_CONTAINER and "ro" not in item["options"]:
+                policy.write("    (blockinherit config_rw_container)\n")
+                add_template("config_container")
+                continue
+
+            contexts = list_contexts(item["source"])
+            for context in contexts:
+                if "ro" not in item["options"]:
+                    policy.write(
+                        "    (allow process "
+                        + context
+                        + " ( dir ( "
+                        + perms.perm["dir_rw"]
+                        + " ))) \n"
+                    )
+                    policy.write(
+                        "    (allow process "
+                        + context
+                        + " ( file ( "
+                        + perms.perm["file_rw"]
+                        + " ))) \n"
+                    )
+                    policy.write(
+                        "    (allow process "
+                        + context
+                        + " ( fifo_file ( "
+                        + perms.perm["fifo_rw"]
+                        + " ))) \n"
+                    )
+                    policy.write(
+                        "    (allow process "
+                        + context
+                        + " ( sock_file ( "
+                        + perms.perm["socket_rw"]
+                        + " ))) \n"
+                    )
+                if "ro" in item["options"]:
+                    policy.write(
+                        "    (allow process "
+                        + context
+                        + " ( dir ( "
+                        + perms.perm["dir_ro"]
+                        + " ))) \n"
+                    )
+                    policy.write(
+                        "    (allow process "
+                        + context
+                        + " ( file ( "
+                        + perms.perm["file_ro"]
+                        + " ))) \n"
+                    )
+                    policy.write(
+                        "    (allow process "
+                        + context
+                        + " ( fifo_file ( "
+                        + perms.perm["fifo_ro"]
+                        + " ))) \n"
+                    )
+                    policy.write(
+                        "    (allow process "
+                        + context
+                        + " ( sock_file ( "
+                        + perms.perm["socket_ro"]
+                        + " ))) \n"
+                    )
+
 
 def load_policy(opts):
     PWD = getcwd()