diff --git a/src/middlewared/middlewared/plugins/account.py b/src/middlewared/middlewared/plugins/account.py index 8ae30b68a1687..1b24263c5edea 100644 --- a/src/middlewared/middlewared/plugins/account.py +++ b/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']: diff --git a/src/middlewared/middlewared/plugins/filesystem_/perm_check.py b/src/middlewared/middlewared/plugins/filesystem_/perm_check.py index a3675a3b334f7..3b20137b6f966 100644 --- a/src/middlewared/middlewared/plugins/filesystem_/perm_check.py +++ b/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): diff --git a/tests/api2/test_account_ssh_key.py b/tests/api2/test_account_ssh_key.py index c55698e58331a..c628a21a5f0f9 100644 --- a/tests/api2/test_account_ssh_key.py +++ b/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()