NAS-143402 / 26.0.0 / Stricter user SSH public key validation to match OpenSSH behavior

src/middlewared/middlewared/plugins/account.py

@@ -617,6 +617,85 @@ def setup_homedir(self, path, username, mode, uid, gid, create=False):
 
         return target
 
+    @private
+    def validate_sshpubkey_home_access(
+        self, verrors, schema, username, uid, gid, groups, home, new_mode, new_owner_uid
+    ):
+        """Verify that sshd will be able to read `home`/.ssh/authorized_keys.
+
+        `username`, `uid`, `gid` and `groups` describe the account as it will exist once the operation completes.
+
+        `new_mode` and `new_owner_uid` are the mode and the owner that the home directory will be given by this
+        operation, or None when it is left as it is.
+        """
+
+        if new_mode is not None:
+            probe_path = os.path.dirname(home)
+        else:
+            probe_path = home
+
+        access_errors = self.middleware.call_sync(
+            'filesystem.can_access_as_cred_errors',
+            {'pw_name': username, 'pw_uid': uid, 'pw_gid': gid, 'grouplist': groups},
+            probe_path,
+            ['EXECUTE'],
+            False,
+            True,
+        )
+        if access_errors:
+            # access_errors[0] is the shallowest component of the path that the credential cannot use.
+            component = access_errors[0].failing_component.decode()
+            try:
+                st = os.stat(component)
+                details = f'it is mode {stat.S_IMODE(st.st_mode):03o}, owned by uid {st.st_uid} and gid {st.st_gid}'
+            except OSError:
+                details = os.strerror(access_errors[0].errnum)
+
+            verrors.add(
+                f'{schema}.sshpubkey',
+                f'User {username!r} is not allowed to access {component!r} ({details}). OpenSSH will reject the public '
+                f'key. Please set permissions that allow the user to traverse every directory leading to {home!r}.',
+            )
+            return
+
+        if new_mode is not None:
+            mode = int(new_mode, 8)
+        else:
+            mode = None
+        owner_uid = new_owner_uid
+        if mode is None or owner_uid is None:
+            try:
+                st = os.stat(home)
+            except FileNotFoundError:
+                # The home directory is created as part of this operation, so its `owner_uid` and `mode` are already
+                # correct.
+                pass
+            else:
+                if mode is None:
+                    mode = stat.S_IMODE(st.st_mode)
+                if owner_uid is None:
+                    owner_uid = st.st_uid
+
+        if owner_uid is not None and owner_uid not in (0, uid):
+            verrors.add(
+                f'{schema}.sshpubkey',
+                f'Home directory {home!r} is owned by uid {owner_uid}, which is neither the user nor root. OpenSSH '
+                'will reject the public key. Please set correct home directory ownership.'
+            )
+
+        if mode is not None and mode & 0o002:
+            verrors.add(
+                f'{schema}.sshpubkey',
+                f'Home directory {home!r} is world-writable (mode {mode:03o}). OpenSSH will reject the public key. '
+                'Please set correct home directory permissions.'
+            )
+
+    @private
+    def group_ids_to_gids(self, group_ids):
+        return [
+            grp['gid'] for grp in self.middleware.call_sync('group.query', [['id', 'in', list(group_ids)]])
+        ]
+
     @api_method(UserCreateArgs, UserCreateResult, audit='Create user', audit_extended=lambda data: data['username'])
     def do_create(self, data):
         """
@@ -704,6 +783,31 @@ def do_create(self, data):
                     self.middleware.call_sync('group.delete', data['group'])
                 idmap_verrors.check()
 
+        if data['sshpubkey'] and data['home'] and data['home'] != DEFAULT_HOME_PATH:
+            if data['home_create']:
+                home = os.path.join(data['home'], data['username'])
+            else:
+                home = data['home']
+
+            pubkey_verrors = ValidationErrors()
+            self.validate_sshpubkey_home_access(
+                pubkey_verrors,
+                'user_create',
+                data['username'],
+                data['uid'],
+                group['gid'],
+                self.group_ids_to_gids(data['groups']),
+                home,
+                data['home_mode'],
+                data['uid'],
+            )
+            if pubkey_verrors:
+                with SYNC_NEXT_UID_LOCK:
+                    self.ReservedUids.remove_entry(data['uid'])
+                if group_created:
+                    self.middleware.call_sync('group.delete', data['group'])
+                pubkey_verrors.check()
+
         new_homedir = False
         home_mode = data.pop('home_mode')
         if data['home'] and data['home'] != DEFAULT_HOME_PATH:
@@ -894,6 +998,22 @@ def do_update(self, app, audit_callback, pk, data):
                 self.middleware.call_sync('filesystem.is_dataset_path', home)
             ):
                 verrors.add('user_update.sshpubkey', 'Home directory is not writable, leave this blank"')
+            elif has_home:
+                new_home = home
+                if 'home' in data and data.get('home_create', False):
+                    new_home = os.path.join(home, data.get('username') or user['username'])
+
+                self.validate_sshpubkey_home_access(
+                    verrors,
+                    'user_update',
+                    user['username'],
+                    user['uid'],
+                    group['bsdgrp_gid'],
+                    self.group_ids_to_gids(group_ids),
+                    new_home,
+                    data.get('home_mode'),
+                    None,
+                )
 
         # Do not allow attributes to be changed for builtin user
         if user['immutable']:

src/middlewared/middlewared/plugins/filesystem_/perm_check.py

@@ -2,6 +2,7 @@
 import os
 import pathlib
 from types import MappingProxyType
+from typing import Any
 
 import truenas_os
 
@@ -111,9 +112,38 @@ def can_access_as_user(self, username: str, path: str, perms: list[str]) -> bool
         Check whether `username` is granted every permission in `perms` on `path`.
 
         `perms` is a list of ``"READ"`` / ``"WRITE"`` / ``"EXECUTE"`` tokens —
-        at least one must be specified.  Returns True iff every requested bit
+        at least one must be specified.  Returns True if every requested bit
         is granted by the filesystem (mode bits + native ACLs).
         """
+        return not self.can_access_as_user_errors(username, path, perms)
+
+    @private
+    def can_access_as_user_errors(
+        self, username: str, path: str, perms: list[str], path_must_exist: bool = True,
+        probe_ancestors: bool = False
+    ) -> list[truenas_os.AccessFailure]:
+        """
+        `can_access_as_user` but return a list of access errors.
+        """
+        try:
+            user_details = self.middleware.call_sync('user.get_user_obj', {'username': username, 'get_groups': True})
+        except KeyError:
+            raise CallError(f'{username!r} user does not exist', errno=errno.ENOENT)
+
+        return self.can_access_as_cred_errors(
+            user_details, path, perms, path_must_exist, probe_ancestors
+        )
+
+    @private
+    def can_access_as_cred_errors(
+        self, user_details: dict[str, Any], path: str, perms: list[str], path_must_exist: bool = True,
+        probe_ancestors: bool = False
+    ) -> list[truenas_os.AccessFailure]:
+        """
+        `can_access_as_user_errors` but for user that might not exist yet.
+
+        Setting `path_must_exist` to False skips missing components.
+        """
         if not perms:
             raise CallError('At least one of READ/WRITE/EXECUTE must be set', errno.EINVAL)
 
@@ -122,22 +152,27 @@ def can_access_as_user(self, username: str, path: str, perms: list[str]) -> bool
         path_obj = pathlib.Path(path)
         if not path_obj.is_absolute():
             raise CallError('A valid absolute path must be provided', errno.EINVAL)
-        elif not path_obj.exists():
+        elif path_must_exist and not path_obj.exists():
             raise CallError(f'{path!r} does not exist', errno.EINVAL)
 
-        try:
-            user_details = self.middleware.call_sync('user.get_user_obj', {'username': username, 'get_groups': True})
-        except KeyError:
-            raise CallError(f'{username!r} user does not exist', errno=errno.ENOENT)
+        cred_entry = _cred_from_user_details({**user_details, 'id_name': user_details['pw_name']})
 
-        user_details['id_name'] = user_details['pw_name']
-        failures = truenas_os.check_path_access(
-            creds=[_cred_from_user_details(user_details)],
+        if probe_ancestors:
+            if ancestors := _path_ancestor_components(path):
+                failures = truenas_os.check_path_access(
+                    creds=[cred_entry],
+                    components=ancestors,
+                    path_must_exist=path_must_exist,
+                )
+                if failures:
+                    return failures
+
+        return truenas_os.check_path_access(
+            creds=[cred_entry],
             components=[path.encode()],
             mode=mode,
-            path_must_exist=True,
+            path_must_exist=path_must_exist,
         )
-        return not failures
 
     @private
     def check_path_execute(self, path, id_type, xid, path_must_exist):

tests/api2/test_account_ssh_key.py

@@ -1,3 +1,8 @@
+import re
+
+import pytest
+
+from middlewared.service_exception import ValidationErrors
 from middlewared.test.integration.assets.account import user
 from middlewared.test.integration.assets.pool import dataset
 from middlewared.test.integration.utils import call, ssh
@@ -77,3 +82,152 @@ def test_account_delete_ssh_key_on_user_delete():
             call("user.delete", u["id"])
 
             assert ssh(f"cat {homedir}/test/.ssh/authorized_keys", check=False) == ""
+
+
+@pytest.fixture(scope="module")
+def keypair():
+    return call("keychaincredential.generate_ssh_key_pair")
+
+
+def test_account_create_ssh_key_rejected_when_home_parent_not_traversable(keypair):
+    """sshd reads authorized_keys as the user itself, so a home directory the user cannot reach
+    makes the key useless. Setting one has to be rejected instead of silently doing nothing."""
+    with dataset("home") as ds:
+        parent = f"/mnt/{ds}/parent"
+        ssh(f"mkdir -m 700 {parent}")
+
+        errmsg = (
+            f"User 'keyuser' is not allowed to access '{parent}' (it is mode 700, owned by uid 0 and gid 0). "
+            f"OpenSSH will reject the public key. Please set permissions that allow the user to traverse every "
+            f"directory leading to '{parent}/keyuser'."
+        )
+        with pytest.raises(ValidationErrors, match=re.escape(errmsg)):
+            with user({
+                "username": "keyuser",
+                "full_name": "Key User",
+                "group_create": True,
+                "password": "test1234",
+                "home": parent,
+                "sshpubkey": keypair["public_key"],
+            }):
+                pass
+
+        # the rejected user.create must not leave the group or the home directory behind
+        assert call("group.query", [["group", "=", "keyuser"], ["local", "=", True]]) == []
+        assert ssh(f"ls -A {parent}") == ""
+
+
+def test_account_update_ssh_key_rejected_when_home_parent_not_traversable(keypair):
+    with dataset("home") as ds:
+        parent = f"/mnt/{ds}/parent"
+        ssh(f"mkdir -m 700 {parent}")
+
+        with user({
+            "username": "keyuser",
+            "full_name": "Key User",
+            "group_create": True,
+            "password": "test1234",
+            "home": parent,
+        }) as u:
+            errmsg = (
+                f"User 'keyuser' is not allowed to access '{parent}' (it is mode 700, owned by uid 0 and gid 0). "
+                f"OpenSSH will reject the public key. Please set permissions that allow the user to traverse every "
+                f"directory leading to '{parent}/keyuser'."
+            )
+            with pytest.raises(ValidationErrors, match=re.escape(errmsg)):
+                call("user.update", u["id"], {"sshpubkey": keypair["public_key"]})
+
+            assert call("user.get_instance", u["id"])["sshpubkey"] is None
+
+            # opening up the parent directory makes the key acceptable
+            ssh(f"chmod 755 {parent}")
+            call("user.update", u["id"], {"sshpubkey": keypair["public_key"]})
+
+
+def test_account_create_ssh_key_rejected_for_world_writable_home(keypair):
+    """sshd `StrictModes` refuses an authorized_keys file that lives in a world-writable home
+    directory, so the mode the home directory is about to be given has to be checked too."""
+    with dataset("home") as ds:
+        errmsg = (
+            f"Home directory '/mnt/{ds}/keyuser' is world-writable (mode 777). OpenSSH will reject the public "
+            f"key. Please set correct home directory permissions."
+        )
+        with pytest.raises(ValidationErrors, match=re.escape(errmsg)):
+            with user({
+                "username": "keyuser",
+                "full_name": "Key User",
+                "group_create": True,
+                "password": "test1234",
+                "home": f"/mnt/{ds}",
+                "home_mode": "777",
+                "sshpubkey": keypair["public_key"],
+            }):
+                pass
+
+
+def test_account_update_ssh_key_rejected_for_world_writable_home(keypair):
+    """The home directory mode an update inherits has to be checked as well, not only the one an
+    update sets explicitly."""
+    with dataset("home") as ds:
+        with user({
+            "username": "keyuser",
+            "full_name": "Key User",
+            "group_create": True,
+            "password": "test1234",
+            "home": f"/mnt/{ds}",
+        }) as u:
+            home = call("user.get_instance", u["id"])["home"]
+            ssh(f"chmod 777 {home}")
+
+            errmsg = (
+                f"Home directory '{home}' is world-writable (mode 777). OpenSSH will reject the public key. "
+                f"Please set correct home directory permissions."
+            )
+            with pytest.raises(ValidationErrors, match=re.escape(errmsg)):
+                call("user.update", u["id"], {"sshpubkey": keypair["public_key"]})
+
+
+def test_account_update_ssh_key_rejected_when_home_owned_by_another_user(keypair):
+    """sshd `StrictModes` refuses an authorized_keys file out of a home directory that is owned
+    neither by the user nor by root, even when the user can read it."""
+    with dataset("home") as ds:
+        with user({
+            "username": "keyuser",
+            "full_name": "Key User",
+            "group_create": True,
+            "password": "test1234",
+            "home": f"/mnt/{ds}",
+        }) as u:
+            home = call("user.get_instance", u["id"])["home"]
+            ssh(f"chown 12345:12345 {home}; chmod 755 {home}")
+
+            errmsg = (
+                f"Home directory '{home}' is owned by uid 12345, which is neither the user nor root. OpenSSH "
+                f"will reject the public key. Please set correct home directory ownership."
+            )
+            with pytest.raises(ValidationErrors, match=re.escape(errmsg)):
+                call("user.update", u["id"], {"sshpubkey": keypair["public_key"]})
+
+            # root owning the home directory is fine for sshd
+            ssh(f"chown 0:0 {home}")
+            call("user.update", u["id"], {"sshpubkey": keypair["public_key"]})
+
+
+def test_account_update_ssh_key_and_home_mode_at_once(keypair):
+    """A call that repairs the home directory mode and sets a public key at the same time must
+    be judged on the mode it is setting, not on the one it is replacing."""
+    with dataset("home") as ds:
+        with user({
+            "username": "keyuser",
+            "full_name": "Key User",
+            "group_create": True,
+            "password": "test1234",
+            "home": f"/mnt/{ds}",
+            "home_mode": "777",
+        }) as u:
+            call("user.update", u["id"], {
+                "home_mode": "700",
+                "sshpubkey": keypair["public_key"],
+            })
+
+            assert call("user.get_instance", u["id"])["sshpubkey"] == keypair["public_key"].strip()