From 76343d20375e58de3f19d0003b1aed4d62640e42 Mon Sep 17 00:00:00 2001 From: sshrushanth-ks Date: Thu, 9 Jul 2026 10:54:02 +0530 Subject: [PATCH] KC-1339: Fix nsf-share-folder team/user remove on NSF subfolders (#2185) * Fix nsf-share-folder team/user remove on NSF subfolders Team recipients are now resolved via team_cache when absent from share_objects, @existing uses get_folder_access_v3 for subfolders, revoke verifies direct (non-inherited) accessors before calling access_update, and successful removes set sync_data and update the local access cache. * refactored code * Fix nsf-share-folder team grant/remove on NSF subfolders * addressed review comment --- keepercommander/api.py | 53 +++-- .../nested_share_folder/folder_commands.py | 55 ++++- .../commands/nested_share_folder/helpers.py | 11 +- keepercommander/nested_share_folder/common.py | 70 ++++-- .../nested_share_folder/folder_api.py | 108 ++++++++- unit-tests/test_command_record.py | 1 + unit-tests/test_nested_share_folder.py | 217 +++++++++++++++++- 7 files changed, 463 insertions(+), 52 deletions(-) diff --git a/keepercommander/api.py b/keepercommander/api.py index 1ada50da6..1da4a1e32 100644 --- a/keepercommander/api.py +++ b/keepercommander/api.py @@ -377,8 +377,12 @@ def load_team_keys(params, team_uids): # type: (KeeperParams, List[str] encrypted_team_key = t.get('encrypted_team_key') if encrypted_team_key: team_key = crypto.decrypt_aes_v2(utils.base64_url_decode(encrypted_team_key), tree_key) - params.key_cache[team_uid] = PublicKeys(aes=team_key) - s.remove(team_uid) + existing = params.key_cache.get(team_uid) + params.key_cache[team_uid] = PublicKeys( + aes=team_key, + rsa=getattr(existing, 'rsa', b'') or b'', + ec=getattr(existing, 'ec', b'') or b'') + # Still fetch asymmetric public keys via team_get_keys below. except Exception as e: logging.debug('Team UID \"%s\": Decrypt key error: %s', team_uid, str(e)) @@ -397,30 +401,51 @@ def load_team_keys(params, team_uids): # type: (KeeperParams, List[str] } rs = communicate(params, rq) if 'keys' in rs: + merged = {} # team_uid -> PublicKeys fields for tk in rs['keys']: + team_uid = tk.get('team_uid') + if not team_uid: + continue + if team_uid not in merged: + existing = params.key_cache.get(team_uid) + merged[team_uid] = { + 'aes': getattr(existing, 'aes', b'') or b'', + 'rsa': getattr(existing, 'rsa', b'') or b'', + 'ec': getattr(existing, 'ec', b'') or b'', + } + # Read the symmetric/wrapped team key from the 'key' field if 'key' in tk: - team_uid = tk['team_uid'] try: - aes = b'' - rsa = b'' - ec = b'' encrypted_key = utils.base64_url_decode(tk['key']) - key_type = tk['type'] + key_type = tk.get('type') if key_type == 1: - aes = crypto.decrypt_aes_v1(encrypted_key, params.data_key) + merged[team_uid]['aes'] = crypto.decrypt_aes_v1(encrypted_key, params.data_key) elif key_type == 2: - aes = crypto.decrypt_rsa(encrypted_key, params.rsa_key2) + merged[team_uid]['aes'] = crypto.decrypt_rsa(encrypted_key, params.rsa_key2) elif key_type == 3: - aes = crypto.decrypt_aes_v2(encrypted_key, params.data_key) + merged[team_uid]['aes'] = crypto.decrypt_aes_v2(encrypted_key, params.data_key) elif key_type == 4: - aes = crypto.decrypt_ec(encrypted_key, params.ecc_key) + merged[team_uid]['aes'] = crypto.decrypt_ec(encrypted_key, params.ecc_key) elif key_type == -1: - ec = encrypted_key + merged[team_uid]['ec'] = encrypted_key elif key_type == -3: - rsa = encrypted_key - params.key_cache[team_uid] = PublicKeys(rsa=rsa, aes=aes, ec=ec) + merged[team_uid]['rsa'] = encrypted_key + except Exception as e: + logging.debug(e) + # Read the raw team public key from 'team_public_key' field (separate from 'key') + if 'team_public_key' in tk: + try: + pub_key_bytes = utils.base64_url_decode(tk['team_public_key']) + pub_key_type = tk.get('team_public_key_type') + if pub_key_type == -3: + merged[team_uid]['rsa'] = pub_key_bytes + elif pub_key_type == -1: + merged[team_uid]['ec'] = pub_key_bytes except Exception as e: logging.debug(e) + for team_uid, kd in merged.items(): + params.key_cache[team_uid] = PublicKeys( + rsa=kd['rsa'], aes=kd['aes'], ec=kd['ec']) def load_available_teams(params): diff --git a/keepercommander/commands/nested_share_folder/folder_commands.py b/keepercommander/commands/nested_share_folder/folder_commands.py index 57fa7c348..eb8be8341 100644 --- a/keepercommander/commands/nested_share_folder/folder_commands.py +++ b/keepercommander/commands/nested_share_folder/folder_commands.py @@ -362,6 +362,10 @@ def execute(self, params, **kwargs): check_folder_share_permission(params, folder_uid, 'nsf-share-folder') targets = self._collect_targets(params, recipients, folder_uid, folder_arg) + if not targets: + raise CommandError( + 'nsf-share-folder', + f'No valid recipients resolved for folder {folder_arg!r}') for recipient, is_team in targets: self._apply(params, action, folder_uid, recipient, role, expiration, as_team=is_team) @@ -408,24 +412,50 @@ def _expand_existing(params, folder_uid, folder_arg): """Expand ``@existing`` / ``@current`` into all users and teams currently on the folder, excluding the caller. Mirrors legacy behaviour (``shared_folder_cache[...]['users']`` + ``['teams']`` union). + + Uses ``get_folder_access_v3`` so sub-folders (whose sync cache only + carries the current user's own row) still enumerate every accessor + that can be removed, including inherited access. """ from keepercommander.proto import folder_pb2 - accesses = (getattr(params, 'nested_share_folder_accesses', {}) - .get(folder_uid, [])) at_user = int(folder_pb2.AT_USER) at_team = int(folder_pb2.AT_TEAM) result = [] - for a in accesses: - access_type = int(a.get('access_type', 0) or 0) - if access_type == at_user: - username = a.get('username') - if username and username != params.user: - result.append(('user', username)) - elif access_type == at_team: - team_uid = a.get('access_type_uid') - if team_uid: - result.append(('team', team_uid)) + try: + info = _nsf.get_folder_access_v3(params, [folder_uid], resolve_usernames=True) + for fr in info.get('results', []): + if not fr.get('success'): + continue + for accessor in fr.get('accessors', []): + if accessor.get('access_type') == 'AT_OWNER': + continue + if accessor.get('access_type') == 'AT_TEAM': + team_uid = accessor.get('accessor_uid') + if team_uid: + result.append(('team', team_uid)) + elif accessor.get('access_type') == 'AT_USER': + username = accessor.get('username') + if username and username != params.user: + result.append(('user', username)) + except Exception as exc: + logging.debug( + 'nsf-share-folder: get_folder_access_v3 failed for %s: %s', + folder_arg, exc) + + if not result: + accesses = (getattr(params, 'nested_share_folder_accesses', {}) + .get(folder_uid, [])) + for a in accesses: + access_type = int(a.get('access_type', 0) or 0) + if access_type == at_user: + username = a.get('username') + if username and username != params.user: + result.append(('user', username)) + elif access_type == at_team: + team_uid = a.get('access_type_uid') + if team_uid: + result.append(('team', team_uid)) if not result: logging.info("No existing users or teams found in folder '%s'", folder_arg) @@ -447,6 +477,7 @@ def _apply(cls, params, action, folder_uid, recipient, role, expiration, try: result = api_func(**kw) if result['success']: + params.sync_data = True taken = result.get('action_taken', verb) if taken == 'already_had_access': logging.info("%s '%s' already has access", kind, recipient) diff --git a/keepercommander/commands/nested_share_folder/helpers.py b/keepercommander/commands/nested_share_folder/helpers.py index 57ab381e5..b4352a1dd 100644 --- a/keepercommander/commands/nested_share_folder/helpers.py +++ b/keepercommander/commands/nested_share_folder/helpers.py @@ -201,14 +201,16 @@ def classify_share_recipient(params, recipient): Mirrors the legacy ``share-folder`` resolution exactly: 1. If *recipient* matches ``EMAIL_PATTERN`` → ``('user', email_lower)``. 2. Otherwise look it up in ``api.get_share_objects(params)['teams']`` - (cap of 500 entries) and, if needed, ``params.available_team_cache``. - A match by team name *or* team UID returns ``('team', team_uid_b64)``. + (cap of 500 entries), ``params.available_team_cache``, and + ``resolve_team_identifier`` (``team_cache``). A match by team name + *or* team UID returns ``('team', team_uid_b64)``. 3. No match → logs the same warning as legacy and returns ``None``. 4. Multiple matches → logs the same warning and returns ``None``. Returns ``(kind, identifier)`` or ``None``. """ from ... import constants, api + from ...nested_share_folder.common import resolve_team_identifier if re.match(constants.EMAIL_PATTERN, recipient): return 'user', recipient.lower() @@ -228,12 +230,15 @@ def classify_share_recipient(params, recipient): pass matches = [uid for uid, name in teams_map.items() - if recipient in (name, uid)] + if recipient == uid or (name and name.lower() == recipient.lower())] if len(matches) == 1: return 'team', matches[0] if not matches: + resolved_team = resolve_team_identifier(params, recipient) + if resolved_team: + return 'team', resolved_team[0] logger.warning('User "%s" could not be resolved as email or team', recipient) else: diff --git a/keepercommander/nested_share_folder/common.py b/keepercommander/nested_share_folder/common.py index 0a3142a1b..261f47651 100644 --- a/keepercommander/nested_share_folder/common.py +++ b/keepercommander/nested_share_folder/common.py @@ -178,6 +178,11 @@ def get_team_keys(params, team_uid_b64: str): """ from ..params import PublicKeys + cached = params.key_cache.get(team_uid_b64) + has_asym = bool(cached and (getattr(cached, 'rsa', None) or getattr(cached, 'ec', None))) + if cached and has_asym: + return cached + api.load_team_keys(params, [team_uid_b64]) keys = params.key_cache.get(team_uid_b64) @@ -186,21 +191,47 @@ def get_team_keys(params, team_uid_b64: str): try: rq = {'command': 'team_get_keys', 'teams': [team_uid_b64]} rs = api.communicate(params, rq) - existing_aes = getattr(keys, 'aes', None) if keys else None - rsa_pub = b'' - ec_pub = b'' + merged = { + 'aes': getattr(keys, 'aes', b'') or b'' if keys else b'', + 'rsa': getattr(keys, 'rsa', b'') or b'' if keys else b'', + 'ec': getattr(keys, 'ec', b'') or b'' if keys else b'', + } for tk in (rs or {}).get('keys', []): - if tk.get('team_uid') != team_uid_b64 or 'key' not in tk: + if tk.get('team_uid') != team_uid_b64: continue - key_type = tk.get('type') - encrypted_key = utils.base64_url_decode(tk['key']) - if key_type == -1: - ec_pub = encrypted_key - elif key_type == -3: - rsa_pub = encrypted_key - if rsa_pub or ec_pub: + # Symmetric/wrapped key in 'key' field + if 'key' in tk: + try: + key_type = tk.get('type') + encrypted_key = utils.base64_url_decode(tk['key']) + if key_type == 1: + merged['aes'] = crypto.decrypt_aes_v1(encrypted_key, params.data_key) + elif key_type == 2: + merged['aes'] = crypto.decrypt_rsa(encrypted_key, params.rsa_key2) + elif key_type == 3: + merged['aes'] = crypto.decrypt_aes_v2(encrypted_key, params.data_key) + elif key_type == 4: + merged['aes'] = crypto.decrypt_ec(encrypted_key, params.ecc_key) + elif key_type == -1: + merged['ec'] = encrypted_key + elif key_type == -3: + merged['rsa'] = encrypted_key + except Exception as e: + logger.debug('team_get_keys key decode failed: %s', e) + # Raw public key in 'team_public_key' field (separate from 'key') + if 'team_public_key' in tk: + try: + pub_key_bytes = utils.base64_url_decode(tk['team_public_key']) + pub_key_type = tk.get('team_public_key_type') + if pub_key_type == -3: + merged['rsa'] = pub_key_bytes + elif pub_key_type == -1: + merged['ec'] = pub_key_bytes + except Exception as e: + logger.debug('team_get_keys public key decode failed: %s', e) + if any(merged.values()): params.key_cache[team_uid_b64] = PublicKeys( - aes=existing_aes, rsa=rsa_pub, ec=ec_pub) + aes=merged['aes'], rsa=merged['rsa'], ec=merged['ec']) keys = params.key_cache[team_uid_b64] except Exception as exc: logger.debug("team_get_keys fallback failed for %s: %s", @@ -215,6 +246,10 @@ def encrypt_for_team(plaintext_key: bytes, team_keys, prefer_aes: bool = False, forbid_rsa: bool = False) -> Tuple[bytes, int]: """Encrypt *plaintext_key* using the best available team key. + + Mirrors legacy ``share-folder``: prefer the team AES key when + *prefer_aes* is set, otherwise try asymmetric keys first, then fall + back to AES when no public key is available (common for enterprise teams). """ aes = getattr(team_keys, 'aes', None) ec_bytes = getattr(team_keys, 'ec', None) @@ -237,7 +272,9 @@ def encrypt_for_team(plaintext_key: bytes, team_keys, return (crypto.encrypt_ec(plaintext_key, ec_key), folder_pb2.encrypted_by_public_key_ecc) - raise ValueError("No public key found for team") + raise ValueError( + "No public key found for team; NSF folder sharing requires the " + "team's RSA or ECC public key (server requires key type 2)") def resolve_uid_email(params, user_identifier: str) -> Tuple[Optional[bytes], str]: @@ -451,13 +488,12 @@ def parse_folder_access_result(response, folder_uid, user_uid, default_message): if response.folderAccessResults: result = response.folderAccessResults[0] status_value = result.status - is_failure = (status_value != 0) or (result.message and len(result.message) > 0) - status_name = (folder_pb2.FolderModifyStatus.Name(status_value) - if status_value != 0 else 'SUCCESS') + is_failure = status_value != folder_pb2.SUCCESS + status_name = folder_pb2.FolderModifyStatus.Name(status_value) return { 'folder_uid': folder_uid, 'user_uid': user_uid, - 'status': 'ERROR' if is_failure and status_value == 0 else status_name, + 'status': status_name, 'message': result.message if result.message else default_message, 'success': not is_failure, } diff --git a/keepercommander/nested_share_folder/folder_api.py b/keepercommander/nested_share_folder/folder_api.py index 6656f324e..493986741 100644 --- a/keepercommander/nested_share_folder/folder_api.py +++ b/keepercommander/nested_share_folder/folder_api.py @@ -390,8 +390,6 @@ def grant_folder_access_v3(params, folder_uid, user_uid, role='viewer', fk = get_folder_key(params, folder_uid) ek = folder_pb2.EncryptedDataKey() if as_team: - # v3 folder team grants must use the team's *asymmetric* public - # key (server rejects AES with "Key type 2 required"). efk, key_type = encrypt_for_team( fk, team_keys, prefer_aes=False, forbid_rsa=getattr(params, 'forbid_rsa', False)) @@ -464,7 +462,73 @@ def update_folder_access_v3(params, folder_uid, user_uid, role=None, hidden=None return result +def _lookup_folder_accessor(params, folder_uid, accessor_uid_b64, access_type_label): + """Return a folder accessor row from ``get_folder_access_v3``, if present.""" + try: + info = get_folder_access_v3(params, [folder_uid], resolve_usernames=True) + for fr in info.get('results', []): + if not fr.get('success'): + continue + accessors = fr.get('accessors', []) + for accessor in accessors: + if (accessor.get('accessor_uid') == accessor_uid_b64 + and accessor.get('access_type') == access_type_label): + return accessor + for accessor in accessors: + if accessor.get('accessor_uid') == accessor_uid_b64: + return accessor + except Exception as exc: + logger.debug('Folder accessor lookup failed for %s: %s', folder_uid, exc) + return None + + +def _find_direct_access_ancestor(params, folder_uid, accessor_uid_b64, access_type_label): + """Walk up the NSF folder tree to find the ancestor where the accessor has direct (non-inherited) access. + + Returns the ancestor folder UID, or ``None`` if no such ancestor is found. + Used when a remove is requested on a subfolder but the access is inherited from a parent. + """ + nsf_folders = getattr(params, 'nested_share_folders', {}) + visited = set() + current_uid = folder_uid + + while current_uid: + if current_uid in visited: + break + visited.add(current_uid) + + parent_uid = nsf_folders.get(current_uid, {}).get('parent_uid') or '' + if not parent_uid or parent_uid not in nsf_folders: + break + + accessor = _lookup_folder_accessor( + params, parent_uid, accessor_uid_b64, access_type_label) + if accessor and not accessor.get('inherited'): + return parent_uid + + current_uid = parent_uid + + return None + + +def _evict_folder_accessor_cache(params, folder_uid, accessor_uid_b64): + """Drop a cached folder-access row after a successful revoke.""" + accesses = getattr(params, 'nested_share_folder_accesses', {}).get(folder_uid) + if not accesses: + return + params.nested_share_folder_accesses[folder_uid] = [ + fa for fa in accesses + if fa.get('access_type_uid') != accessor_uid_b64 + ] + + def revoke_folder_access_v3(params, folder_uid, user_uid, as_team=False): + """Revoke user or team access to an NSF folder. + + Looks up the accessor via ``get_folder_access_v3`` so sub-folders use the + server-reported UID and access type (including inherited accessors). + On success, evicts the local access cache and sets ``params.sync_data``. + """ resolved = resolve_folder_identifier(params, folder_uid) if not resolved: raise ValueError(f"Folder '{folder_uid}' not found") @@ -475,6 +539,41 @@ def revoke_folder_access_v3(params, folder_uid, user_uid, as_team=False): if not actual_uid_bytes: raise ValueError(f"{'Team' if as_team else 'User'} '{user_uid}' not found") + accessor_uid_b64 = utils.base64_url_encode(actual_uid_bytes) + access_type_label = folder_pb2.AccessType.Name(access_type_enum) + accessor = _lookup_folder_accessor( + params, folder_uid, accessor_uid_b64, access_type_label) + if accessor: + server_uid = accessor.get('accessor_uid') + if server_uid: + actual_uid_bytes = utils.base64_url_decode(server_uid) + accessor_uid_b64 = server_uid + server_type = accessor.get('access_type') + if server_type: + access_type_label = server_type + try: + access_type_enum = folder_pb2.AccessType.Value(server_type) + except ValueError: + raise ValueError( + f"Unrecognised access type '{server_type}' returned by server " + f"for folder '{folder_uid}'") + + if accessor.get('inherited'): + kind = 'Team' if as_team else 'User' + ancestor_uid = _find_direct_access_ancestor( + params, folder_uid, accessor_uid_b64, access_type_label) + nsf_folders = getattr(params, 'nested_share_folders', {}) + if ancestor_uid: + ancestor_name = nsf_folders.get(ancestor_uid, {}).get('name') or ancestor_uid + raise ValueError( + f"{kind} '{identifier_label}' has inherited access on this folder. " + f"To remove it, run: nsf-share-folder {ancestor_name!r} " + f"-e {identifier_label} -a remove") + else: + raise ValueError( + f"{kind} '{identifier_label}' has inherited access on this folder. " + f"Remove it from the parent folder where access was originally granted.") + ad = folder_pb2.FolderAccessData() ad.folderUid = utils.base64_url_decode(folder_uid) ad.accessTypeUid = actual_uid_bytes @@ -483,7 +582,10 @@ def revoke_folder_access_v3(params, folder_uid, user_uid, as_team=False): response = folder_access_update_v3(params, folder_access_removes=[ad]) result = parse_folder_access_result(response, folder_uid, identifier_label, 'Access revoked successfully') - result['access_type'] = 'AT_TEAM' if as_team else 'AT_USER' + result['access_type'] = access_type_label + if result.get('success'): + _evict_folder_accessor_cache(params, folder_uid, accessor_uid_b64) + params.sync_data = True return result diff --git a/unit-tests/test_command_record.py b/unit-tests/test_command_record.py index 03de045fc..fecaa1176 100644 --- a/unit-tests/test_command_record.py +++ b/unit-tests/test_command_record.py @@ -305,6 +305,7 @@ def test_get_shared_folder_uid(self): cmd.execute(params, format='json', uid=shared_folder_uid) def test_get_user_folder_json_consistent_by_name_and_uid(self): + """get --format json returns the same folder payload by name or UID.""" params = get_synced_params() cmd = record.RecordGetUidCommand() user_folder = next( diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py index 301afc3ed..e70494247 100644 --- a/unit-tests/test_nested_share_folder.py +++ b/unit-tests/test_nested_share_folder.py @@ -13,7 +13,15 @@ from unittest.mock import Mock, MagicMock, patch from keepercommander import utils, crypto +from keepercommander.commands.nested_share_folder import ( + NestedShareFolderMkdirCommand, + NestedShareFolderShareCommand, +) +from keepercommander.commands.nested_share_folder.helpers import classify_share_recipient from keepercommander.error import CommandError +from keepercommander.nested_share_folder.common import parse_folder_access_result +from keepercommander.nested_share_folder.folder_api import revoke_folder_access_v3 +from keepercommander.proto import folder_pb2 _DATA_KEY = utils.generate_aes_key() @@ -245,7 +253,7 @@ def test_mkdir(self, mock_create): @patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3') def test_mkdir_resolves_parent_uid_in_path(self, mock_create): - from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand + """Parent path segments that are NSF UIDs resolve to the folder, not a name.""" parent_uid = 'tY6D-RanxY252zzBY_xU4A' child_uid = utils.generate_uid() mock_create.return_value = { @@ -270,7 +278,7 @@ def test_mkdir_resolves_parent_uid_in_path(self, mock_create): @patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3') def test_mkdir_resolves_parent_name_in_path(self, mock_create): - from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand + """Parent path segments that match an existing folder name resolve by name.""" parent_fuid, parent_fobj = _make_folder(name='Engineering') child_uid = utils.generate_uid() mock_create.return_value = { @@ -293,7 +301,7 @@ def test_mkdir_resolves_parent_name_in_path(self, mock_create): @patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3') def test_mkdir_creates_intermediate_name_segments(self, mock_create): - from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand + """Multi-segment name paths create missing intermediate folders.""" eng_uid = utils.generate_uid() child_uid = utils.generate_uid() mock_create.side_effect = [ @@ -858,6 +866,19 @@ def test_share_folder_invite_message_uses_command_prefix(self, mock_grant): class TestNestedShareFolderFolderApi(TestCase): + def test_encrypt_for_team_uses_rsa_public_key(self): + """Team grants use the RSA public key when available (server requires key type 2).""" + from keepercommander.nested_share_folder.common import encrypt_for_team + from keepercommander.params import PublicKeys + + rsa_priv, rsa_pub = crypto.generate_rsa_key() + rsa_pub_bytes = crypto.unload_rsa_public_key(rsa_pub) + folder_key = utils.generate_aes_key() + team_keys = PublicKeys(aes=utils.generate_aes_key(), rsa=rsa_pub_bytes, ec=b'') + encrypted, key_type = encrypt_for_team(folder_key, team_keys) + self.assertEqual(key_type, folder_pb2.encrypted_by_public_key) + self.assertEqual(crypto.decrypt_rsa(encrypted, rsa_priv), folder_key) + @patch('keepercommander.nested_share_folder.folder_api.folder_access_update_v3') @patch('keepercommander.nested_share_folder.folder_api.handle_share_invite') @patch('keepercommander.nested_share_folder.folder_api.get_user_public_key') @@ -960,6 +981,196 @@ def test_grant_folder_access_same_role_updates_expiration( expiration_timestamp=expiration) + @patch('keepercommander.nested_share_folder.folder_api.folder_access_update_v3') + @patch('keepercommander.nested_share_folder.folder_api.get_folder_access_v3') + @patch('keepercommander.nested_share_folder.folder_api.resolve_folder_identifier') + @patch('keepercommander.nested_share_folder.folder_api._resolve_accessor') + def test_revoke_folder_access_subfolder_team( + self, mock_resolve_accessor, mock_resolve_folder, + mock_get_access, mock_access_update): + """Revoking team access on an NSF subfolder uses server accessor UIDs.""" + parent_uid, parent_obj = _make_folder(name='Parent') + child_uid, child_obj = _make_folder( + name='Child', parent_uid=parent_uid) + team_uid = utils.generate_uid() + team_uid_bytes = utils.base64_url_decode(team_uid) + params = _make_params(nested_share_folders={ + parent_uid: parent_obj, + child_uid: child_obj, + }) + mock_resolve_folder.return_value = child_uid + mock_resolve_accessor.return_value = ( + team_uid_bytes, team_uid, folder_pb2.AT_TEAM) + mock_get_access.return_value = { + 'results': [{ + 'folder_uid': child_uid, + 'success': True, + 'accessors': [{ + 'accessor_uid': team_uid, + 'access_type': 'AT_TEAM', + 'role': 'VIEWER', + 'inherited': False, + }], + }], + } + mock_response = Mock() + mock_result = Mock() + mock_result.status = folder_pb2.SUCCESS + mock_result.message = '' + mock_response.folderAccessResults = [mock_result] + mock_access_update.return_value = mock_response + + result = revoke_folder_access_v3( + params, child_uid, team_uid, as_team=True) + + self.assertTrue(result['success']) + self.assertTrue(params.sync_data) + mock_access_update.assert_called_once() + remove_call = mock_access_update.call_args + ad = remove_call.kwargs['folder_access_removes'][0] + self.assertEqual(ad.accessType, folder_pb2.AT_TEAM) + self.assertEqual( + utils.base64_url_encode(ad.accessTypeUid), team_uid) + self.assertEqual( + utils.base64_url_encode(ad.folderUid), child_uid) + + @patch('keepercommander.nested_share_folder.folder_api.folder_access_update_v3') + @patch('keepercommander.nested_share_folder.folder_api.get_folder_access_v3') + @patch('keepercommander.nested_share_folder.folder_api.resolve_folder_identifier') + @patch('keepercommander.nested_share_folder.folder_api._resolve_accessor') + def test_revoke_folder_access_subfolder_team_inherited( + self, mock_resolve_accessor, mock_resolve_folder, + mock_get_access, mock_access_update): + """Inherited access with no resolvable parent raises a generic error.""" + parent_uid, parent_obj = _make_folder(name='Parent') + child_uid, child_obj = _make_folder( + name='Child', parent_uid=parent_uid) + team_uid = utils.generate_uid() + team_uid_bytes = utils.base64_url_decode(team_uid) + # Only child is in nested_share_folders — parent lookup will find no direct access + params = _make_params(nested_share_folders={child_uid: child_obj}) + mock_resolve_folder.return_value = child_uid + mock_resolve_accessor.return_value = ( + team_uid_bytes, team_uid, folder_pb2.AT_TEAM) + mock_get_access.return_value = { + 'results': [{ + 'folder_uid': child_uid, + 'success': True, + 'accessors': [{ + 'accessor_uid': team_uid, + 'access_type': 'AT_TEAM', + 'role': 'VIEWER', + 'inherited': True, + }], + }], + } + + with self.assertRaises(ValueError) as ctx: + revoke_folder_access_v3(params, child_uid, team_uid, as_team=True) + + self.assertIn('inherited', str(ctx.exception)) + mock_access_update.assert_not_called() + + @patch('keepercommander.nested_share_folder.folder_api.folder_access_update_v3') + @patch('keepercommander.nested_share_folder.folder_api.get_folder_access_v3') + @patch('keepercommander.nested_share_folder.folder_api.resolve_folder_identifier') + @patch('keepercommander.nested_share_folder.folder_api._resolve_accessor') + def test_revoke_inherited_access_raises_with_parent_hint( + self, mock_resolve_accessor, mock_resolve_folder, + mock_get_access, mock_access_update): + """Revoking inherited team access raises an error that names the parent folder.""" + parent_uid, parent_obj = _make_folder(name='ParentFolder') + child_uid, child_obj = _make_folder(name='Child', parent_uid=parent_uid) + team_uid = utils.generate_uid() + team_uid_bytes = utils.base64_url_decode(team_uid) + params = _make_params(nested_share_folders={ + parent_uid: parent_obj, + child_uid: child_obj, + }) + mock_resolve_folder.return_value = child_uid + mock_resolve_accessor.return_value = ( + team_uid_bytes, team_uid, folder_pb2.AT_TEAM) + + def get_access_side_effect(p, folder_uids, **kw): + uid = folder_uids[0] + if uid == child_uid: + return {'results': [{'folder_uid': uid, 'success': True, 'accessors': [{ + 'accessor_uid': team_uid, 'access_type': 'AT_TEAM', + 'role': 'VIEWER', 'inherited': True, + }]}]} + if uid == parent_uid: + return {'results': [{'folder_uid': uid, 'success': True, 'accessors': [{ + 'accessor_uid': team_uid, 'access_type': 'AT_TEAM', + 'role': 'VIEWER', 'inherited': False, + }]}]} + return {'results': []} + + mock_get_access.side_effect = get_access_side_effect + + with self.assertRaises(ValueError) as ctx: + revoke_folder_access_v3(params, child_uid, team_uid, as_team=True) + + error = str(ctx.exception) + self.assertIn('inherited', error) + self.assertIn('ParentFolder', error) + mock_access_update.assert_not_called() + + @patch('keepercommander.nested_share_folder.folder_api.revoke_folder_access_v3') + @patch('keepercommander.api.get_share_objects') + def test_share_folder_remove_subfolder_team(self, mock_share_objects, mock_revoke): + """nsf-share-folder -a remove resolves team names on NSF subfolders.""" + mock_share_objects.return_value = {'teams': {}} + parent_uid, parent_obj = _make_folder(name='Parent') + child_uid, child_obj = _make_folder( + name='Child', parent_uid=parent_uid) + team_uid = utils.generate_uid() + mock_revoke.return_value = { + 'success': True, 'action_taken': 'removed', + 'status': 'SUCCESS', 'message': '', + } + params = _make_params( + nested_share_folders={parent_uid: parent_obj, child_uid: child_obj}, + team_cache={team_uid: {'name': 'Ops Team', 'team_uid': team_uid}}, + ) + cmd = NestedShareFolderShareCommand() + with mock.patch('builtins.print'): + cmd.execute( + params, + folder=[child_uid], + user=['Ops Team'], + action='remove', + ) + mock_revoke.assert_called_once_with( + params=params, + folder_uid=child_uid, + user_uid=team_uid, + as_team=True, + ) + + def test_classify_share_recipient_resolves_team_from_cache(self): + """Team names fall back to team_cache when share-objects is empty.""" + team_uid = utils.generate_uid() + params = _make_params(team_cache={ + team_uid: {'name': 'Ops Team', 'team_uid': team_uid}, + }) + with mock.patch('keepercommander.api.get_share_objects', return_value={'teams': {}}): + result = classify_share_recipient(params, 'Ops Team') + self.assertEqual(result, ('team', team_uid)) + + def test_parse_folder_access_result_treats_success_message_as_success(self): + """SUCCESS status with a non-empty message is still treated as success.""" + response = Mock() + result = Mock() + result.status = folder_pb2.SUCCESS + result.message = 'Access revoked successfully' + response.folderAccessResults = [result] + + parsed = parse_folder_access_result( + response, 'folder', 'team', 'Access revoked successfully') + self.assertTrue(parsed['success']) + self.assertEqual(parsed['status'], 'SUCCESS') + + class TestNestedShareFolderRecordApi(TestCase): def setUp(self):