linux: never chown devices

src/libcrun/container.c

@@ -1017,15 +1017,27 @@ send_sync_cb (void *data, libcrun_error_t *err)
 }
 
 static int
-maybe_chown_std_streams (uid_t container_uid, gid_t container_gid,
-                         libcrun_error_t *err)
+maybe_chown_std_streams (uid_t container_uid, gid_t container_gid, libcrun_error_t *err)
 {
   int ret, i;
 
   for (i = 0; i < 3; i++)
     {
       if (! isatty (i))
         {
+          struct stat statbuf;
+          ret = fstat (i, &statbuf);
+          if (UNLIKELY (ret < 0))
+            {
+              if (errno == EBADF)
+                continue;
+              return crun_make_error (err, errno, "fstat fd `%d`", i);
+            }
+
+          /* Skip chown for device files */
+          if (S_ISCHR (statbuf.st_mode) || S_ISBLK (statbuf.st_mode))
+            continue;
+
           ret = fchown (i, container_uid, container_gid);
           if (UNLIKELY (ret < 0))
             {

tests/test_devices.py

@@ -248,14 +248,48 @@ def test_net_devices():
 
         for specify_broadcast in [True, False]:
             for specify_name in [True, False]:
-                subprocess.run(["ip", "link", "add", "testdevice", "type", "dummy"])
+                sys.stderr.write("# test_net_devices: creating testdevice with specify_broadcast=%s, specify_name=%s\n" % (specify_broadcast, specify_name))
+                result = subprocess.run(["ip", "link", "add", "testdevice", "type", "dummy"], capture_output=True, text=True)
+                if result.returncode != 0:
+                    sys.stderr.write("# ip link add failed: %s\n" % result.stderr)
+                    return -1
                 if specify_broadcast:
-                    subprocess.run(["ip", "addr", "add", "10.1.2.3/24", "brd", "10.1.2.254", "dev", "testdevice"])
+                    result = subprocess.run(["ip", "addr", "add", "10.1.2.3/24", "brd", "10.1.2.254", "dev", "testdevice"], capture_output=True, text=True)
+                    if result.returncode != 0:
+                        sys.stderr.write("# ip addr add with broadcast failed: %s\n" % result.stderr)
+                        return -1
                 else:
-                    subprocess.run(["ip", "addr", "add", "10.1.2.3/24", "dev", "testdevice"])
+                    result = subprocess.run(["ip", "addr", "add", "10.1.2.3/24", "dev", "testdevice"], capture_output=True, text=True)
+                    if result.returncode != 0:
+                        sys.stderr.write("# ip addr add without broadcast failed: %s\n" % result.stderr)
+                        return -1
 
                 conf = base_config()
                 add_all_namespaces(conf)
+
+                # Add network capabilities needed for network device operations
+                conf['process']['capabilities'] = {
+                    "bounding": [
+                        "CAP_NET_ADMIN",
+                        "CAP_NET_RAW",
+                        "CAP_SYS_ADMIN"
+                    ],
+                    "effective": [
+                        "CAP_NET_ADMIN",
+                        "CAP_NET_RAW",
+                        "CAP_SYS_ADMIN"
+                    ],
+                    "inheritable": [
+                        "CAP_NET_ADMIN",
+                        "CAP_NET_RAW",
+                        "CAP_SYS_ADMIN"
+                    ],
+                    "permitted": [
+                        "CAP_NET_ADMIN",
+                        "CAP_NET_RAW",
+                        "CAP_SYS_ADMIN"
+                    ]
+                }
                 if specify_name:
                     conf['process']['args'] = ['/init', 'ip', 'newtestdevice']
                     conf['linux']['netDevices'] = {
@@ -272,19 +306,28 @@ def test_net_devices():
 
                 try:
                     out = run_and_get_output(conf)
+                    sys.stderr.write("# test_net_devices: specify_broadcast=%s, specify_name=%s\n" % (specify_broadcast, specify_name))
+                    sys.stderr.write("# test_net_devices: output: %s\n" % repr(out[0]))
                     if "address: 10.1.2.3" not in out[0]:
                         sys.stderr.write("# address not found in output\n")
+                        sys.stderr.write("# full output: %s\n" % repr(out[0]))
                         return 1
                     if specify_broadcast:
                         if "broadcast: 10.1.2.254" not in out[0]:
                             sys.stderr.write("# broadcast address not found in output\n")
+                            sys.stderr.write("# full output: %s\n" % repr(out[0]))
                             return 1
                     else:
                         if "broadcast" in out[0]:
-                            sys.stderr.write("# broadcast address found in output\n")
+                            sys.stderr.write("# broadcast address found in output when it shouldn't be\n")
+                            sys.stderr.write("# full output: %s\n" % repr(out[0]))
                             return 1
                 except Exception as e:
+                    sys.stderr.write("# test_net_devices exception: %s\n" % str(e))
                     return -1
+                finally:
+                    # Clean up the test device
+                    subprocess.run(["ip", "link", "del", "testdevice"], capture_output=True)
     finally:
         os.setns(current_netns, os.CLONE_NEWNET)
         os.close(current_netns)

tests/test_mounts.py

@@ -21,14 +21,15 @@
 from tests_utils import *
 import tempfile
 import re
+from typing import List, Optional
 
 try:
     import libmount
 except Exception:
     print("1..0")
     sys.exit(0)
 
-def helper_mount(options, tmpfs=True, userns=False, is_file=False):
+def helper_mount(options: str, tmpfs: bool = True, userns: bool = False, is_file: bool = False) -> List[Optional[str]]:
     conf = base_config()
     conf['process']['args'] = ['/init', 'cat', '/proc/self/mountinfo']
     add_all_namespaces(conf, userns=userns)
@@ -54,12 +55,12 @@ def helper_mount(options, tmpfs=True, userns=False, is_file=False):
                 sys.stderr.write("# helper_mount failed: mount target '%s' not found in mountinfo\n" % target)
                 sys.stderr.write("# mount options: %s, tmpfs=%s, userns=%s, is_file=%s\n" % (options, tmpfs, userns, is_file))
                 sys.stderr.write("# mountinfo output: %s\n" % out[:300])
-                return -1
+                return [None, None]
             return [m.vfs_options, m.fs_options]
     except Exception as e:
         sys.stderr.write("# helper_mount failed with exception: %s\n" % str(e))
         sys.stderr.write("# mount options: %s, tmpfs=%s, userns=%s, is_file=%s\n" % (options, tmpfs, userns, is_file))
-        return -1
+        return [None, None]
 
 def test_mount_symlink():
     conf = base_config()
@@ -285,157 +286,157 @@ def test_mount_path_with_multiple_slashes():
 def test_mount_ro():
     for userns in [True, False]:
         a = helper_mount("ro", userns=userns, is_file=True)[0]
-        if "ro" not in a:
+        if a is None or "ro" not in a:
             return -1
         a = helper_mount("ro", userns=userns)[0]
-        if "ro" not in a:
+        if a is None or "ro" not in a:
             return -1
         a = helper_mount("ro", userns=userns, tmpfs=False)[0]
-        if "ro" not in a:
+        if a is None or "ro" not in a:
             return -1
     return 0
 
 def test_mount_rro():
     for userns in [True, False]:
         a = helper_mount("rro", userns=userns, is_file=True)[0]
-        if "ro" not in a:
+        if a is None or "ro" not in a:
             return -1
         a = helper_mount("rro", userns=userns)[0]
-        if "ro" not in a:
+        if a is None or "ro" not in a:
             return -1
         a = helper_mount("rro", userns=userns, tmpfs=False)[0]
-        if "ro" not in a:
+        if a is None or "ro" not in a:
             return -1
     return 0
 
 def test_mount_rw():
     for userns in [True, False]:
         a = helper_mount("rw", tmpfs=False, userns=userns)[0]
-        if "rw" not in a:
+        if a is None or "rw" not in a:
             return -1
         a = helper_mount("rw", userns=userns, is_file=True)[0]
-        if "rw" not in a:
+        if a is None or "rw" not in a:
             return -1
         a = helper_mount("rw", userns=userns)[0]
-        if "rw" not in a:
+        if a is None or "rw" not in a:
             return -1
     return 0
 
 def test_mount_relatime():
     for userns in [True, False]:
         a = helper_mount("relatime", tmpfs=False, userns=userns)[0]
-        if "relatime" not in a:
+        if a is None or "relatime" not in a:
             return -1
         a = helper_mount("relatime", is_file=True, userns=userns)[0]
-        if "relatime" not in a:
+        if a is None or "relatime" not in a:
             return -1
         a = helper_mount("relatime", userns=userns)[0]
-        if "relatime" not in a:
+        if a is None or "relatime" not in a:
             return -1
     return 0
 
 def test_mount_strictatime():
     for userns in [True, False]:
         a = helper_mount("strictatime", is_file=True, userns=userns)[0]
-        if "relatime" not in a:
+        if a is None or "relatime" not in a:
             return 0
         a = helper_mount("strictatime", tmpfs=False, userns=userns)[0]
-        if "relatime" not in a:
+        if a is None or "relatime" not in a:
             return 0
         a = helper_mount("strictatime", userns=userns)[0]
-        if "relatime" not in a:
+        if a is None or "relatime" not in a:
             return 0
     return -1
 
 def test_mount_exec():
     for userns in [True, False]:
         a = helper_mount("exec", is_file=True, userns=userns)[0]
-        if "noexec" in a:
+        if a is not None and "noexec" in a:
             return -1
         a = helper_mount("exec", tmpfs=False, userns=userns)[0]
-        if "noexec" in a:
+        if a is not None and "noexec" in a:
             return -1
         a = helper_mount("exec", userns=userns)[0]
-        if "noexec" in a:
+        if a is not None and "noexec" in a:
             return -1
     return 0
 
 def test_mount_noexec():
     for userns in [True, False]:
         a = helper_mount("noexec", is_file=True, userns=userns)[0]
-        if "noexec" not in a:
+        if a is None or "noexec" not in a:
             return -1
         a = helper_mount("noexec", tmpfs=False, userns=userns)[0]
-        if "noexec" not in a:
+        if a is None or "noexec" not in a:
             return -1
         a = helper_mount("noexec", userns=userns)[0]
-        if "noexec" not in a:
+        if a is None or "noexec" not in a:
             return -1
     return 0
 
 def test_mount_suid():
     for userns in [True, False]:
         a = helper_mount("suid", is_file=True, userns=userns)[0]
-        if "nosuid" in a:
+        if a is not None and "nosuid" in a:
             return -1
         a = helper_mount("suid", tmpfs=False, userns=userns)[0]
-        if "nosuid" in a:
+        if a is not None and "nosuid" in a:
             return -1
         a = helper_mount("suid", userns=userns)[0]
-        if "nosuid" in a:
+        if a is not None and "nosuid" in a:
             return -1
     return 0
 
 def test_mount_nosuid():
     for userns in [True, False]:
         a = helper_mount("nosuid", is_file=True, userns=userns)[0]
-        if "nosuid" not in a:
+        if a is None or "nosuid" not in a:
             return -1
         a = helper_mount("nosuid", tmpfs=False, userns=userns)[0]
-        if "nosuid" not in a:
+        if a is None or "nosuid" not in a:
             return -1
         a = helper_mount("nosuid", userns=userns)[0]
-        if "nosuid" not in a:
+        if a is None or "nosuid" not in a:
             return -1
     return 0
 
 def test_mount_sync():
     for userns in [True, False]:
         a = helper_mount("sync", userns=userns)[1]
-        if "sync" not in a:
+        if a is None or "sync" not in a:
             return -1
     return 0
 
 def test_mount_dirsync():
     for userns in [True, False]:
         a = helper_mount("dirsync", userns=userns)[1]
-        if "dirsync" not in a:
+        if a is None or "dirsync" not in a:
             return -1
     return 0
 
 def test_mount_nodev():
     for userns in [True, False]:
         a = helper_mount("nodev", is_file=True)[0]
-        if "nodev" not in a:
+        if a is None or "nodev" not in a:
             return -1
         a = helper_mount("nodev", tmpfs=False)[0]
-        if "nodev" not in a:
+        if a is None or "nodev" not in a:
             return -1
         a = helper_mount("nodev", userns=userns)[0]
-        if "nodev" not in a:
+        if a is None or "nodev" not in a:
             return -1
     return 0
 
 def test_mount_dev():
     for userns in [True, False]:
         a = helper_mount("dev", userns=userns, tmpfs=False)[0]
-        if "nodev" in a:
+        if a is not None and "nodev" in a:
             return -1
         a = helper_mount("dev", userns=userns, is_file=True)[0]
-        if "nodev" in a:
+        if a is not None and "nodev" in a:
             return -1
         a = helper_mount("dev", userns=userns)[0]
-        if "nodev" in a:
+        if a is not None and "nodev" in a:
             return -1
     return 0