diff --git a/keepercommander/commands/nested_share_folder/folder_commands.py b/keepercommander/commands/nested_share_folder/folder_commands.py index 52eb39dbb..28f3a677f 100644 --- a/keepercommander/commands/nested_share_folder/folder_commands.py +++ b/keepercommander/commands/nested_share_folder/folder_commands.py @@ -30,6 +30,7 @@ check_folder_edit_permission, check_folder_share_permission, check_folder_delete_permission, classify_share_recipient, ensure_nested_share_folder, is_nested_share_folder, + is_nested_share_folder_owner_email, owner_share_target_message, ) from .parsers import ( nested_share_folder_mkdir_parser, @@ -465,6 +466,11 @@ def _expand_existing(params, folder_uid, folder_arg): elif accessor.get('access_type') == 'AT_USER': username = accessor.get('username') if username and username != params.user: + if is_nested_share_folder_owner_email(params, folder_uid, username): + logging.info( + "nsf-share-folder: skipping owner '%s' for " + "folder '%s'", username, folder_arg) + continue result.append(('user', username)) except Exception as exc: logging.debug( @@ -479,6 +485,11 @@ def _expand_existing(params, folder_uid, folder_arg): if access_type == at_user: username = a.get('username') if username and username != params.user: + if is_nested_share_folder_owner_email(params, folder_uid, username): + logging.info( + "nsf-share-folder: skipping owner '%s' for " + "folder '%s'", username, folder_arg) + continue result.append(('user', username)) elif access_type == at_team: team_uid = a.get('access_type_uid') @@ -493,6 +504,10 @@ def _expand_existing(params, folder_uid, folder_arg): @classmethod def _apply(cls, params, action, folder_uid, recipient, role, expiration, as_team=False, rotate_on_expiration=False): + if not as_team and is_nested_share_folder_owner_email(params, folder_uid, recipient): + raise CommandError( + 'nsf-share-folder', + owner_share_target_message(recipient, entity='folder')) api_name, verb = cls._ACTIONS[action] api_func = getattr(_nsf, api_name) kw = dict(params=params, folder_uid=folder_uid, user_uid=recipient, diff --git a/keepercommander/commands/nested_share_folder/helpers.py b/keepercommander/commands/nested_share_folder/helpers.py index e5dd397ec..855c6c700 100644 --- a/keepercommander/commands/nested_share_folder/helpers.py +++ b/keepercommander/commands/nested_share_folder/helpers.py @@ -174,6 +174,57 @@ def resolve_folder_uid(params, identifier): "only on Nested Share Folders." ) +_OWNER_SHARE_TARGET_MSG = ( + "'{email}' is the owner of this {entity} and already has full access. " + "Share permissions cannot be granted, changed, or revoked for the owner." +) + + +def owner_share_target_message(email, entity='record'): + """User-facing message when a share command targets the owner.""" + return _OWNER_SHARE_TARGET_MSG.format(email=email, entity=entity) + + +def is_nested_share_folder_owner_email(params, folder_uid, email): + """True when *email* matches the NSF folder owner from sync-down.""" + if not folder_uid or not email: + return False + fobj = getattr(params, 'nested_share_folders', {}).get(folder_uid) or {} + owner_username = fobj.get('owner_username') or '' + return bool(owner_username) and owner_username.casefold() == email.casefold() + + +def raise_if_record_share_target_is_owner(params, record_uid, email, cmd_name, *, + is_ownership_transfer=False): + """Raise CommandError when *email* is the owner of *record_uid*. + + When *is_ownership_transfer* is True (``-a owner``), raising means the + target already owns the record so the transfer would be a no-op. When False, + grant/revoke/update against the owner is rejected. + + If owner lookup via ``get_record_accesses_v3`` fails, we log a warning and + return without raising. Blocking the share on a transient access-API error + would reject legitimate grants to non-owners; the server still rejects + invalid owner-targeted shares if we miss the check client-side. + """ + from ...nested_share_folder.record_api import ( + get_record_accesses_v3, find_record_owner_username) + try: + access_result = get_record_accesses_v3(params, [record_uid]) + except Exception as exc: + logging.getLogger(__name__).warning( + "Could not resolve owner for record '%s'; proceeding without " + "client-side owner check: %s", record_uid, exc) + return + owner_username = find_record_owner_username(access_result, record_uid) + if not owner_username or owner_username.casefold() != email.casefold(): + return + if is_ownership_transfer: + raise CommandError( + cmd_name, + f"'{email}' already owns this record. Ownership transfer is a no-op.") + raise CommandError(cmd_name, owner_share_target_message(email, entity='record')) + def is_nested_share_record(params, record_uid): """Return True when *record_uid* is a Nested Share Folder (v3) record.""" diff --git a/keepercommander/commands/nested_share_folder/sharing_commands.py b/keepercommander/commands/nested_share_folder/sharing_commands.py index 36128473c..8872ba78f 100644 --- a/keepercommander/commands/nested_share_folder/sharing_commands.py +++ b/keepercommander/commands/nested_share_folder/sharing_commands.py @@ -34,6 +34,7 @@ check_record_share_permission, collect_records_in_folder, ensure_nested_share_folder, ensure_nested_share_record, + raise_if_record_share_target_is_owner, ) from .parsers import ( nested_share_record_share_parser, @@ -102,6 +103,9 @@ def execute(self, params, **kwargs): with command_error_handler('nsf-share-record'): for email in emails: for record_uid in record_uids: + raise_if_record_share_target_is_owner( + params, record_uid, email, 'nsf-share-record', + is_ownership_transfer=(action == 'owner')) result, effective_action = self._dispatch( params, action, record_uid, email, access_role_type, expiration, rotate_on_expiration) diff --git a/keepercommander/commands/register.py b/keepercommander/commands/register.py index b8de28b37..4c3222e6b 100644 --- a/keepercommander/commands/register.py +++ b/keepercommander/commands/register.py @@ -302,6 +302,65 @@ def _folder_user_lookup(shared_folder, email): return None +def _owner_share_target_message(email, entity='record'): + # type: (str, str) -> str + """User-facing message when a share command targets the owner.""" + return ( + f"'{email}' is the owner of this {entity} and already has full access. " + f"Share permissions cannot be granted, changed, or revoked for the owner." + ) + + +def _is_shared_folder_owner_email(shared_folder, email, params=None): + # type: (dict, str, Optional[KeeperParams]) -> bool + """True when *email* matches the classic shared folder owner from sync-down. + + Resolution order: + 1. ``owner_username`` from sync-down (primary; present for most folders) + 2. ``users[].owner`` flag when sync includes it on the user row + 3. ``owner_account_uid`` matched to the current session user — only used + when username/flag are absent. If sync omits ``owner_account_uid`` or + the session has no ``account_uid_bytes``, this path is skipped and we + return False rather than guessing. + """ + if not shared_folder or not email: + return False + email_cf = email.casefold() + owner_username = shared_folder.get('owner_username') or '' + if owner_username and owner_username.casefold() == email_cf: + return True + # Some sync payloads also put an owner flag on the user row. + users = shared_folder.get('users', []) + if any( + user.get('owner') is True + and (user.get('username') or '').casefold() == email_cf + for user in users + ): + return True + # Last resort: only when the target is the logged-in user and sync exposes + # owner_account_uid. Skipped when either field is missing. + owner_uid = shared_folder.get('owner_account_uid') + account_uid_bytes = getattr(params, 'account_uid_bytes', None) if params else None + if (params and owner_uid and account_uid_bytes + and params.user and params.user.casefold() == email_cf): + if owner_uid == utils.base64_url_encode(account_uid_bytes): + return True + return False + + +def _find_record_owner_username(existing_shares, params=None, record_uid=None): + # type: (dict, Optional[KeeperParams], Optional[str]) -> Optional[str] + """Return the record owner username from share rows or local owner cache.""" + for username, perm in (existing_shares or {}).items(): + if perm and perm.get('owner'): + return username + if params is not None and record_uid and params.user: + owner_info = getattr(params, 'record_owner_cache', {}).get(record_uid) + if owner_info and owner_info.owner: + return params.user + return None + + def format_share_expiration_ms(expiration_ms): # type: (int) -> str """Format a share expiration timestamp (milliseconds) for log output.""" @@ -545,7 +604,18 @@ def prep_rq(recs, users, curr_sf): if all_users or all_records: if all_users: if 'users' in sh_fol: - sf_users.update((x['username'] for x in sh_fol['users'] if x['username'] != params.user)) + for x in sh_fol['users']: + username = x.get('username') + if not username or username == params.user: + continue + if _is_shared_folder_owner_email( + sh_fol, username, params=params): + logging.debug( + "share-folder: skipping owner '%s' in " + "bulk user update for folder '%s'", + username, sf_uid) + continue + sf_users.add(username) if 'teams' in sh_fol: sf_teams.update((x['team_uid'] for x in sh_fol['teams'])) if all_records: @@ -691,6 +761,13 @@ def apply_share_expiration(target): current_user = _folder_user_lookup(curr_sf, email) if current_user: email = current_user['username'] + if _is_shared_folder_owner_email(curr_sf, email, params=params): + # Allow the existing owner self-removal / succession flow in + # _confirm_folder_user_removals; reject grant/update of owner perms. + if action == 'grant': + raise CommandError( + 'share-folder', + _owner_share_target_message(email, entity='shared folder')) uo = folder_pb2.SharedFolderUpdateUser() uo.username = email apply_share_expiration(uo) @@ -1104,7 +1181,24 @@ def apply_share_expiration(ro): # type: (record_pb2.SharedRecord) -> None pass record_path = api.resolve_record_share_path(params, record_uid) + owner_username = _find_record_owner_username( + existing_shares, params=params, record_uid=record_uid) for email in all_users: + # Precedence: owner rejection runs before existing-share add/update + # logic below. Granting/revoking the owner is never valid, even when + # they also appear in existing_shares; "already shared" handling only + # applies to non-owners. + if owner_username and email.casefold() == owner_username.casefold(): + # Ownership transfer (-a owner) is allowed only when the target + # is not already the owner. Grant/revoke must never target owner. + if action == 'owner': + raise CommandError( + 'share-record', + f"'{email}' already owns this record. Ownership transfer is a no-op.") + raise CommandError( + 'share-record', + _owner_share_target_message(email, entity='record')) + ro = record_pb2.SharedRecord() ro.toUsername = email ro.recordUid = utils.base64_url_decode(record_uid) diff --git a/keepercommander/nested_share_folder/__init__.py b/keepercommander/nested_share_folder/__init__.py index 1feda6a61..b4942b52a 100644 --- a/keepercommander/nested_share_folder/__init__.py +++ b/keepercommander/nested_share_folder/__init__.py @@ -43,7 +43,8 @@ 'create_record_data_v3', 'record_add_v3', 'record_add_pam_configuration_v3', 'record_update_v3', 'create_record_v3', 'update_record_v3', 'create_records_batch_v3', 'get_record_details_v3', 'get_record_accesses_v3', - 'find_direct_user_share_access', 'is_record_share_update_noop', + 'find_direct_user_share_access', 'find_record_owner_username', + 'is_record_share_update_noop', 'share_record_v3', 'update_record_share_v3', 'unshare_record_v3', 'share_record_to_application_v3', 'update_record_share_to_application_v3', 'unshare_record_from_application_v3', diff --git a/keepercommander/nested_share_folder/record_api.py b/keepercommander/nested_share_folder/record_api.py index f3d13979a..075e75089 100644 --- a/keepercommander/nested_share_folder/record_api.py +++ b/keepercommander/nested_share_folder/record_api.py @@ -380,6 +380,17 @@ def find_direct_user_share_access(access_result, record_uid, email): return None +def find_record_owner_username(access_result, record_uid): + """Return the owner username for *record_uid* from a get_record_accesses_v3 result.""" + return next( + ( + access.get('accessor_name') or '' + for access in access_result.get('record_accesses', []) + if access.get('record_uid') == record_uid and access.get('owner') + ), + '', + ) + _SHARE_EXPIRATION_NOOP_TOLERANCE_MS = 60_000 diff --git a/unit-tests/test_command_register.py b/unit-tests/test_command_register.py index 2f35ea315..c5a8b9d88 100644 --- a/unit-tests/test_command_register.py +++ b/unit-tests/test_command_register.py @@ -68,6 +68,65 @@ def shared(params, record_uids, is_share_admin): cmd.execute(params, email=['user2@keepersecurity.com'], action='revoke', record=record_uid) self.assertEqual(len(TestRegister.expected_commands), 0) + def _shared_with_owner_side_effect(self, owner_username=None, extra_users=None): + """Build a get_record_shares side_effect that marks *owner_username* as owner. + + Defaults to the session user. *extra_users* is an optional list of + non-owner emails included in user_permissions. + """ + extras = list(extra_users or []) + + def shared_with_owner(params_, record_uids, is_share_admin): + owner = owner_username if owner_username is not None else params_.user + for uid in record_uids: + if uid not in params_.record_cache: + continue + perms = [{'username': owner, 'owner': True}] + for email in extras: + perms.append({ + 'username': email, 'owner': False, + 'shareable': False, 'editable': False, + }) + params_.record_cache[uid]['shares'] = {'user_permissions': perms} + + return shared_with_owner + + def test_share_record_rejects_grant_to_owner(self): + """Granting share permissions to the record owner must fail client-side.""" + params = get_synced_params() + record_uid = next(iter([x['record_uid'] for x in params.meta_data_cache.values() if x['can_share']])) + cmd = register.ShareRecordCommand() + + self.record_share_mock = mock.patch('keepercommander.api.get_record_shares').start() + self.record_share_mock.side_effect = self._shared_with_owner_side_effect( + extra_users=['user2@keepersecurity.com']) + + with self.assertRaises(CommandError) as ctx: + cmd.prep_request(params, dict( + email=[params.user], + action='grant', + record=record_uid, + )) + self.assertIn('is the owner', str(ctx.exception)) + self.assertIn('already has full access', str(ctx.exception)) + + def test_share_record_rejects_owner_transfer_to_self(self): + """Ownership transfer to the current owner is a no-op and must be rejected.""" + params = get_synced_params() + record_uid = next(iter([x['record_uid'] for x in params.meta_data_cache.values() if x['can_share']])) + cmd = register.ShareRecordCommand() + + self.record_share_mock = mock.patch('keepercommander.api.get_record_shares').start() + self.record_share_mock.side_effect = self._shared_with_owner_side_effect() + + with self.assertRaises(CommandError) as ctx: + cmd.prep_request(params, dict( + email=[params.user], + action='owner', + record=record_uid, + )) + self.assertIn('already owns', str(ctx.exception)) + @contextmanager def _make_record_rotation_eligible(self, params, target_uid): """Present the record as a pamUser with rotation configured (ROE-eligible).""" @@ -308,6 +367,30 @@ def test_share_folder(self): cmd.execute(params, action='remove', user=['user2@keepersecurity.com'], folder=shared_folder_uid) self.assertEqual(len(TestRegister.expected_commands), 0) + def test_share_folder_rejects_grant_to_owner(self): + """Granting folder permissions to the folder owner must fail client-side.""" + params = get_synced_params() + shared_folder_uid = next(iter(params.shared_folder_cache.keys())) + owner = params.user + curr_sf = dict(params.shared_folder_cache[shared_folder_uid]) + curr_sf['owner_username'] = owner + curr_sf['users'] = [ + {'username': owner, 'manage_records': True, 'manage_users': True}, + {'username': 'user2@keepersecurity.com', 'manage_records': True, 'manage_users': True}, + ] + + with self.assertRaises(CommandError) as ctx: + register.ShareFolderCommand.prepare_request( + params, + kwargs={'action': 'grant', 'manage_records': 'on', 'manage_users': 'off'}, + curr_sf=curr_sf, + users=[owner], + teams=[], + rec_uids=[], + ) + self.assertIn('is the owner', str(ctx.exception)) + self.assertIn('shared folder', str(ctx.exception)) + def test_share_folder_prepare_request_sets_rotate_on_expiration(self): """Folder-wide expiration/ROE applies to user/team protos, not record protos.""" params = get_synced_params() diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py index db4b6a0a3..aa72a4438 100644 --- a/unit-tests/test_nested_share_folder.py +++ b/unit-tests/test_nested_share_folder.py @@ -881,7 +881,7 @@ def test_share_record(self, mock_share): cmd = NestedShareRecordShareCommand() with mock.patch('builtins.print'): cmd.execute(_make_params(nested_share_records={ruid: robj}), - record=ruid, email='user@example.com', + record=ruid, email=['user@example.com'], action='grant', role='viewer') @patch('keepercommander.nested_share_folder.record_api.unshare_record_v3') @@ -895,9 +895,45 @@ def test_share_record_revoke(self, mock_unshare): cmd = NestedShareRecordShareCommand() with mock.patch('builtins.print'): cmd.execute(_make_params(nested_share_records={ruid: robj}), - record=ruid, email='user@example.com', + record=ruid, email=['user@example.com'], action='revoke') + @patch('keepercommander.nested_share_folder.record_api.get_record_accesses_v3') + @patch('keepercommander.nested_share_folder.record_api.share_record_v3') + def test_share_record_rejects_grant_to_owner(self, mock_share, mock_accesses): + from keepercommander.commands.nested_share_folder import NestedShareRecordShareCommand + ruid, robj = _make_record() + owner = 'owner@example.com' + mock_accesses.return_value = { + 'record_accesses': [{ + 'record_uid': ruid, + 'accessor_name': owner, + 'owner': True, + 'access_type': 'AT_USER', + }], + 'forbidden_records': [], + } + cmd = NestedShareRecordShareCommand() + with self.assertRaises(CommandError) as ctx: + cmd.execute(_make_params(nested_share_records={ruid: robj}), + record=ruid, email=[owner], + action='grant', role='viewer') + self.assertIn('is the owner', str(ctx.exception)) + mock_share.assert_not_called() + + @patch('keepercommander.nested_share_folder.folder_api.grant_folder_access_v3') + def test_share_folder_rejects_grant_to_owner(self, mock_grant): + from keepercommander.commands.nested_share_folder import NestedShareFolderShareCommand + fuid, fobj = _make_folder() + owner = 'owner@example.com' + fobj['owner_username'] = owner + cmd = NestedShareFolderShareCommand() + with self.assertRaises(CommandError) as ctx: + cmd.execute(_make_params(nested_share_folders={fuid: fobj}), + folder=[fuid], user=[owner], action='grant', role='viewer') + self.assertIn('is the owner', str(ctx.exception)) + mock_grant.assert_not_called() + @patch('keepercommander.nested_share_folder.folder_api.grant_folder_access_v3') def test_share_folder_invite_message_uses_command_prefix(self, mock_grant): from keepercommander.commands.nested_share_folder import NestedShareFolderShareCommand