Skip to content

Add prefix directories in CRI-O mounts #146

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
FabienGhd wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Add prefix directories in CRI-O mounts #146

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2d7b668
Add prefix directories in CRI-O mounts
FabienGhd Jul 12, 2024
41b19d3
Clean up after adding prefix directories
vmojzis May 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions tests/test_main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,31 @@ def test_basic_specified_engine_cri(self):
)
self.assert_policy(test_file("test_basic.cri.cil"))

def test_mount_prefix_podman(self):
"""podman run with --mount-prefix /mnt to verify parameter works and paths are prefixed correctly"""
# This test verifies that:
# 1. The --mount-prefix parameter is accepted
# 2. Paths are properly joined using os.path.join (no double slashes)
# 3. The prefix is applied before SELinux context lookup
# Note: With /mnt prefix, paths like /home become /mnt/home. The mock SELinux
# context database still matches these (as would a real system with proper
# file context rules), so the generated policy is functionally equivalent.
# Meaning that this just tests that the parameter does not break anything.
output = self.run_udica(
[
"udica",
"-j",
"tests/test_basic.podman.json",
"--mount-prefix",
"/mnt",
"my_container",
]
)
self.assert_templates(
output, ["base_container", "net_container", "home_container"]
)
self.assert_policy(test_file("test_basic.podman.cil"))

def test_default_podman(self):
"""podman run fedora"""
output = self.run_udica(
Expand Down
8 changes: 8 additions & 0 deletions udica/__main__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,14 @@ def get_args():
required=False,
default="-",
)
parser.add_argument(
"--mount-prefix",
type=str,
help="Prefix to add to host paths of mounts (useful when running udica inside a container)",
dest="MountPrefix",
required=False,
default="",
)

args = parser.parse_args()
return vars(args)
Expand Down
4 changes: 4 additions & 0 deletions udica/man/man8/udica.8
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,10 @@ Allow container to stream connect with given SELinux domain.
.I \-e, \-\-container-engine ENGINE
Specify which container engine is used for the inspected container (supports: CRI-O, docker, podman)

.TP
.I \-\-mount\-prefix PREFIX
Prefix to add to host paths of mounts (useful when running udica inside a container)

.TP
.I \-\-full\-network\-access
Allow a container full network access
Expand Down
92 changes: 56 additions & 36 deletions udica/policy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@

from shutil import copy
from os import chdir, getcwd, remove
from os.path import exists
from os.path import exists, join
import tarfile

import selinux
Expand Down Expand Up @@ -106,7 +106,13 @@ def list_ports(port_number, port_proto):


def create_policy(
opts, capabilities, devices, mounts, ports, append_rules, inspect_format
opts,
capabilities,
devices,
mounts,
ports,
append_rules,
inspect_format,
):
policy = open(opts["ContainerName"] + ".cil", "w")
policy.write("(block " + opts["ContainerName"] + "\n")
Expand Down Expand Up @@ -177,12 +183,13 @@ def create_policy(
write_policy_for_podman_devices(devices, policy)

# mounts
mount_prefix = opts.get("MountPrefix", "")
if inspect_format == "CRI-O":
write_policy_for_crio_mounts(mounts, policy)
write_policy_for_crio_mounts(mounts, policy, mount_prefix)
elif inspect_format == "containerd":
write_policy_for_containerd_mounts(mounts, policy)
write_policy_for_containerd_mounts(mounts, policy, mount_prefix)
else:
write_policy_for_podman_mounts(mounts, policy)
write_policy_for_podman_mounts(mounts, policy, mount_prefix)

if append_rules != None:
for rule in append_rules:
Expand All @@ -207,52 +214,55 @@ def create_policy(
policy.close()


def write_policy_for_crio_mounts(mounts, policy):
def write_policy_for_crio_mounts(mounts, policy, mount_prefix=""):
contexts = []
contexts_readonly = []

for item in mounts:
if item["hostPath"].startswith("/var/lib/kubelet"):
# Include mount_prefix in the path for Kubernetes container calls.
host_path = (
join(mount_prefix, item["hostPath"]) if mount_prefix else item["hostPath"]
)

if host_path.startswith("/var/lib/kubelet"):
# These should already have the right context
continue
if item["hostPath"] == LOG_CONTAINER:
if host_path == LOG_CONTAINER:
if item["readonly"]:
policy.write(" (blockinherit log_container)\n")
else:
policy.write(" (blockinherit log_rw_container)\n")
add_template("log_container")
continue

if item["hostPath"] == HOME_CONTAINER:
if host_path == HOME_CONTAINER:
if item["readonly"]:
policy.write(" (blockinherit home_container)\n")
else:
policy.write(" (blockinherit home_rw_container)\n")
add_template("home_container")
continue

if item["hostPath"] == TMP_CONTAINER:
if host_path == TMP_CONTAINER:
if item["readonly"]:
policy.write(" (blockinherit tmp_container)\n")
else:
policy.write(" (blockinherit tmp_rw_container)\n")
add_template("tmp_container")
continue

if item["hostPath"] == CONFIG_CONTAINER:
if host_path == CONFIG_CONTAINER:
if item["readonly"]:
policy.write(" (blockinherit config_container)\n")
else:
policy.write(" (blockinherit config_rw_container)\n")
add_template("config_container")
continue

# TODO(jaosorior): Add prefix-dir to path. This way we could call this
# from a container in kubernetes
if item["readonly"] is False:
contexts.extend(list_contexts(item["hostPath"]))
contexts.extend(list_contexts(host_path))
else:
contexts_readonly.extend(list_contexts(item["hostPath"]))
contexts_readonly.extend(list_contexts(host_path))

for context in sorted(set(contexts)):
policy.write(
Expand Down Expand Up @@ -338,56 +348,61 @@ def write_policy_for_podman_devices(devices, policy):
)


def write_policy_for_podman_mounts(mounts, policy):
def write_policy_for_podman_mounts(mounts, policy, mount_prefix=""):
contexts = []
contexts_rw = []

for item in mounts:
if not item["Source"].find("/"):
if item["Source"] == LOG_CONTAINER and item["RW"] is False:
# Include mount_prefix in the path for container calls.
source_path = (
join(mount_prefix, item["Source"]) if mount_prefix else item["Source"]
)

if source_path == LOG_CONTAINER and item["RW"] is False:
policy.write(" (blockinherit log_container)\n")
add_template("log_container")
continue

if item["Source"] == LOG_CONTAINER and item["RW"] is True:
if source_path == LOG_CONTAINER and item["RW"] is True:
policy.write(" (blockinherit log_rw_container)\n")
add_template("log_container")
continue

if item["Source"] == HOME_CONTAINER and item["RW"] is False:
if source_path == HOME_CONTAINER and item["RW"] is False:
policy.write(" (blockinherit home_container)\n")
add_template("home_container")
continue

if item["Source"] == HOME_CONTAINER and item["RW"] is True:
if source_path == HOME_CONTAINER and item["RW"] is True:
policy.write(" (blockinherit home_rw_container)\n")
add_template("home_container")
continue

if item["Source"] == TMP_CONTAINER and item["RW"] is False:
if source_path == TMP_CONTAINER and item["RW"] is False:
policy.write(" (blockinherit tmp_container)\n")
add_template("tmp_container")
continue

if item["Source"] == TMP_CONTAINER and item["RW"] is True:
if source_path == TMP_CONTAINER and item["RW"] is True:
policy.write(" (blockinherit tmp_rw_container)\n")
add_template("tmp_container")
continue

if item["Source"] == CONFIG_CONTAINER and item["RW"] is False:
if source_path == CONFIG_CONTAINER and item["RW"] is False:
policy.write(" (blockinherit config_container)\n")
add_template("config_container")
continue

if item["Source"] == CONFIG_CONTAINER and item["RW"] is True:
if source_path == CONFIG_CONTAINER and item["RW"] is True:
policy.write(" (blockinherit config_rw_container)\n")
add_template("config_container")
continue

if item["RW"] is True:
contexts_rw.extend(list_contexts(item["Source"]))
contexts_rw.extend(list_contexts(source_path))
else:
contexts.extend(list_contexts(item["Source"]))
contexts.extend(list_contexts(source_path))

for context in sorted(set(contexts_rw)):
policy.write(
Expand Down Expand Up @@ -450,7 +465,7 @@ def write_policy_for_podman_mounts(mounts, policy):
)


def write_policy_for_containerd_mounts(mounts, policy):
def write_policy_for_containerd_mounts(mounts, policy, mount_prefix=""):
# mount JSON example:
# {
# "destination": "/sys/fs/cgroup",
Expand All @@ -465,47 +480,52 @@ def write_policy_for_containerd_mounts(mounts, policy):
# }
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"]:
# Include mount_prefix in the path for container calls.
source_path = (
join(mount_prefix, item["source"]) if mount_prefix else item["source"]
)

if source_path == 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"]:
if source_path == 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"]:
if source_path == 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"]:
if source_path == 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"]:
if source_path == 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"]:
if source_path == 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"]:
if source_path == 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"]:
if source_path == 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"])
contexts = list_contexts(source_path)
for context in contexts:
if "ro" not in item["options"]:
policy.write(
Expand Down