fix(gstfsd): a key added to an SELinux guest gets the guest's labels

Files that libguestfs creates have no SELinux label. On an enforcing
guest (RHEL, Rocky, Alma, Fedora) sshd may not read the unlabeled
/root/.ssh/authorized_keys, so a key added from the panel did not let
anyone log in ("avc: denied { read } ... unlabeled_t").

After writing a key, gstfsd now relabels /root/.ssh with the guest's own
file contexts (/etc/selinux/<SELINUXTYPE>/contexts/files/file_contexts),
as virt-customize does. Guests without an SELinux policy or with SELinux
disabled are left as they are. If the guest needs labels but the host's
libguestfs cannot set them, the request fails before anything is written.
A root password needs no labelling: /etc/shadow keeps its file and label.

conf/daemon/gstfsd

@@ -69,13 +69,26 @@ def add_root_key(gfs, key):
     gfs.chmod(0o600, keys_path)
 
 
-# action: (function, the files it changes)
+# action: (function, the files it changes, where it may create files)
 ACTIONS = {
-    'password': (lambda gfs, data: set_root_password(gfs, data['passwd']), ['/etc/shadow']),
-    'publickey': (lambda gfs, data: add_root_key(gfs, data['key']), ['/root/.ssh/authorized_keys']),
+    'password': (lambda gfs, data: set_root_password(gfs, data['passwd']), ['/etc/shadow'], None),
+    'publickey': (lambda gfs, data: add_root_key(gfs, data['key']), ['/root/.ssh/authorized_keys'], '/root/.ssh'),
 }
 
 
+def selinux_specfile(gfs):
+    """The guest's SELinux file contexts, or None without an SELinux policy or
+    with SELinux disabled (enabling it again relabels the whole guest)."""
+    if not gfs.is_file('/etc/selinux/config'):
+        return None
+    config = gfs.cat('/etc/selinux/config')
+    if re.search(r"^\s*SELINUX\s*=\s*[\"']?disabled\b", config, re.MULTILINE):
+        return None
+    found = re.search(r"^\s*SELINUXTYPE\s*=\s*[\"']?([\w.-]+)", config, re.MULTILINE)
+    specfile = '/etc/selinux/%s/contexts/files/file_contexts' % (found.group(1) if found else 'targeted')
+    return specfile if gfs.is_file(specfile) else None
+
+
 def holds(mountpoint, path):
     return mountpoint == '/' or path == mountpoint or path.startswith(mountpoint.rstrip('/') + '/')
 
@@ -99,7 +112,7 @@ def mount_guest(gfs, paths):
 def change_guest(data):
     import guestfs
 
-    action, paths = ACTIONS[data['action']]
+    action, paths, creates = ACTIONS[data['action']]
     gfs = guestfs.GuestFS(python_return_dict=True)
     launched = False
     try:
@@ -109,7 +122,14 @@ def change_guest(data):
         mount_guest(gfs, paths)
         if '/etc/shadow' in paths and not gfs.is_file('/etc/shadow'):
             raise RuntimeError("the guest has no /etc/shadow")
+        # A file libguestfs creates has no SELinux label, and an enforcing
+        # guest's sshd does not read an unlabeled authorized_keys.
+        specfile = creates and selinux_specfile(gfs)
+        if specfile and not gfs.feature_available(['selinuxrelabel']):
+            raise RuntimeError("the guest uses SELinux, but libguestfs on this host cannot set SELinux labels")
         action(gfs, data)
+        if specfile:
+            gfs.selinux_relabel(specfile, creates, force=True)
     except Exception:
         # the first error is the one to report; cleanup still runs
         if launched:

instances/test_gstfsd.py

@@ -77,6 +77,13 @@ class GuestFS:
     def mkdir(self, path):
         self._guest(); state = _load(); state["dirs"].append(path); _save(state)
     def chmod(self, mode, path): self._guest()
+    def feature_available(self, names): return not _load().get("no_selinuxrelabel")
+    def selinux_relabel(self, specfile, path, force=None):
+        self._guest(); state = _load()
+        if specfile not in state["files"]:
+            raise RuntimeError("selinux_relabel: " + specfile + ": No such file or directory")
+        # with the number of writes before it: labels are set on written files
+        state.setdefault("relabels", []).append([specfile, path, force, state["writes"]]); _save(state)
 '''
 
 
@@ -329,6 +336,56 @@ class GstfsdTestCase(unittest.TestCase):
         reply = self._request("127.0.0.1", self._serve(), {"action": "publickey", "vname": "vm", "key": "ssh-ed25519 K k@host"})
         self.assertEqual(reply, {"return": "success"})
 
+    def _selinux_guest(self, selinuxtype="targeted", quote="", mode="enforcing"):
+        state = json.loads(self.state.read_text())
+        state["files"]["/etc/selinux/config"] = "SELINUX=%s\nSELINUXTYPE=%s%s%s\n" % (mode, quote, selinuxtype, quote)
+        state["files"]["/etc/selinux/%s/contexts/files/file_contexts" % selinuxtype] = ""
+        self.state.write_text(json.dumps(state))
+
+    def test_a_key_gets_the_selinux_labels_of_the_guest(self):
+        # a file libguestfs creates has no label, and sshd may not read an unlabeled authorized_keys
+        port = self._serve()
+        for i, quote in enumerate(("", '"', "'")):
+            with self.subTest(quote=quote):
+                self._selinux_guest("mls", quote)
+                reply = self._request("127.0.0.1", port, {"action": "publickey", "vname": "vm", "key": "ssh-ed25519 K%d k@host" % i})
+                self.assertEqual(reply, {"return": "success"})
+                state = json.loads(self.state.read_text())
+                # once, after the key is written
+                self.assertEqual(len(state.get("relabels", [])), i + 1)
+                self.assertEqual(
+                    state["relabels"][-1], ["/etc/selinux/mls/contexts/files/file_contexts", "/root/.ssh", True, state["writes"]]
+                )
+
+    def test_a_guest_without_selinux_is_not_relabeled(self):
+        reply = self._request("127.0.0.1", self._serve(), {"action": "publickey", "vname": "vm", "key": "ssh-ed25519 K k@host"})
+        self.assertEqual(reply, {"return": "success"})
+        self.assertNotIn("relabels", json.loads(self.state.read_text()))
+
+    def test_a_guest_with_selinux_disabled_is_not_relabeled(self):
+        # enabling SELinux again relabels the whole guest
+        self._selinux_guest(mode="disabled")
+        self._update_state(no_selinuxrelabel=True)
+        reply = self._request("127.0.0.1", self._serve(), {"action": "publickey", "vname": "vm", "key": "ssh-ed25519 K k@host"})
+        self.assertEqual(reply, {"return": "success"})
+        self.assertNotIn("relabels", json.loads(self.state.read_text()))
+
+    def test_a_key_for_a_selinux_guest_needs_a_host_that_can_label(self):
+        self._selinux_guest()
+        self._update_state(no_selinuxrelabel=True)
+        reply = self._request("127.0.0.1", self._serve(), {"action": "publickey", "vname": "vm", "key": "ssh-ed25519 K k@host"})
+        self.assertEqual(reply["return"], "error")
+        self.assertIn("SELinux", reply["message"])
+        self.assertEqual(json.loads(self.state.read_text())["writes"], 0)
+
+    def test_a_password_keeps_the_label_of_the_existing_shadow(self):
+        # rewriting /etc/shadow keeps its file and label: no labelling needed
+        self._selinux_guest()
+        self._update_state(no_selinuxrelabel=True)
+        reply = self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$x"})
+        self.assertEqual(reply, {"return": "success"})
+        self.assertNotIn("relabels", json.loads(self.state.read_text()))
+
     def test_a_client_that_leaves_does_not_break_the_service(self):
         port = free_port()
         proc = self._start("127.0.0.1", port, GSTFSD_DEADLINE="1")