diff --git a/examples/sdk_examples/nested_shared_folders/nsf_access_batch_execution.py b/examples/sdk_examples/nested_shared_folders/nsf_access_batch_execution.py index f73ae792..e83db5cd 100644 --- a/examples/sdk_examples/nested_shared_folders/nsf_access_batch_execution.py +++ b/examples/sdk_examples/nested_shared_folders/nsf_access_batch_execution.py @@ -514,11 +514,16 @@ def close_vault(vault: vault_online.VaultOnline, keeper_auth_context: keeper_aut def nsf_access_list( vault: vault_online.VaultOnline, folder_uids: List[str], - record_uids: List[str]) -> None: + record_uids: List[str], + show_inherited: bool = False, + show_denied: bool = False) -> None: """ Print access details for a caller-supplied list of NSF folder and record UIDs. + Inherited and denied entries are omitted unless ``show_inherited`` / + ``show_denied`` is True (the SDK default is to hide them). + Access details are not pre-populated in the NSF in-memory cache, so this fetches them for exactly the requested UIDs via ``get_nsf_folder_access()`` / ``get_nsf_record_accesses()``, caches @@ -535,7 +540,8 @@ def nsf_access_list( if not folder_uids: print("No NSF folder UIDs requested.") else: - access = nsf_management.get_nsf_folder_access(vault, folder_uids) + access = nsf_management.get_nsf_folder_access( + vault, folder_uids, show_inherited=show_inherited, show_denied=show_denied) for fr in access.get("results") or []: fuid = fr.get("folder_uid") if fuid: @@ -560,6 +566,7 @@ def nsf_access_list( f" | access_type={access.get('access_type')}" f" | role={access.get('role')}" f" | inherited={access.get('inherited')}" + f" | denied={access.get('denied_access')}" f" | hidden={access.get('hidden')}" f" | owner={access.get('owner', False)}" ) @@ -574,7 +581,8 @@ def nsf_access_list( if not record_uids: print("No NSF record UIDs requested.") else: - access = nsf_management.get_nsf_record_accesses(vault, record_uids) + access = nsf_management.get_nsf_record_accesses( + vault, record_uids, show_inherited=show_inherited, show_denied=show_denied) by_uid: Dict[str, List] = {} for ao in access.get("record_accesses") or []: by_uid.setdefault(ao["record_uid"], []).append(ao) @@ -619,9 +627,13 @@ def nsf_access_list_run(keeper_auth_context: keeper_auth.KeeperAuth) -> None: folder_uids = ["", ""] record_uids = ["", ""] + # Set to True to include inherited / denied entries (hidden by default). + show_inherited = False + show_denied = False + vault = open_vault(keeper_auth_context) try: - nsf_access_list(vault, folder_uids, record_uids) + nsf_access_list(vault, folder_uids, record_uids, show_inherited, show_denied) except Exception as e: print(f"Error: {e}") finally: diff --git a/examples/sdk_examples/nested_shared_folders/nsf_access_list.py b/examples/sdk_examples/nested_shared_folders/nsf_access_list.py index 62daa13d..e6f81ca0 100644 --- a/examples/sdk_examples/nested_shared_folders/nsf_access_list.py +++ b/examples/sdk_examples/nested_shared_folders/nsf_access_list.py @@ -515,8 +515,6 @@ def nsf_access_list( vault: vault_online.VaultOnline, folder_uids: List[str], record_uids: List[str], - show_inherited: bool = False, - show_denied: bool = False ) -> None: load_folder = bool(folder_uids) load_record = bool(record_uids) @@ -524,7 +522,7 @@ def nsf_access_list( if view is None: print("NSF storage is not available on this vault.") return - nsf_management.load_nsf_access_details(vault, load_folder=load_folder, load_record=load_record, show_inherited=show_inherited, show_denied=show_denied) + nsf_management.load_nsf_access_details(vault, load_folder=load_folder, load_record=load_record) for folder_uid in folder_uids: folder = view.get_folder(folder_uid) @@ -584,12 +582,9 @@ def nsf_access_list_run(keeper_auth_context: keeper_auth.KeeperAuth) -> None: folder_uids = ["", ""] record_uids = ["", ""] - show_inherited = True - show_denied = True - vault = open_vault(keeper_auth_context) try: - nsf_access_list(vault, folder_uids, record_uids, show_inherited, show_denied) + nsf_access_list(vault, folder_uids, record_uids) except Exception as e: print(f"Error: {e}") finally: diff --git a/examples/sdk_examples/nested_shared_folders/nsf_get.py b/examples/sdk_examples/nested_shared_folders/nsf_get.py index c56e1e91..15a90837 100644 --- a/examples/sdk_examples/nested_shared_folders/nsf_get.py +++ b/examples/sdk_examples/nested_shared_folders/nsf_get.py @@ -515,10 +515,14 @@ def nsf_get(vault: vault_online.VaultOnline) -> None: """Get NSF record or folder details by UID or title (nsf-get). For folders, the response includes parent_uid and parent_name (when available). + Inherited and denied access entries are omitted unless requested. """ ITEM_UID_OR_TITLE = "" # Record/folder UID or title + SHOW_INHERITED = False # include inherited permission entries + SHOW_DENIED = False # include denied-access permission entries - detail = nsf_management.get_nsf_item(vault, ITEM_UID_OR_TITLE) + detail = nsf_management.get_nsf_item( + vault, ITEM_UID_OR_TITLE, show_inherited=SHOW_INHERITED, show_denied=SHOW_DENIED) print(json.dumps(detail, indent=2, default=str)) diff --git a/keepercli-package/src/keepercli/__init__.py b/keepercli-package/src/keepercli/__init__.py index 37b8df1a..f25f01dc 100644 --- a/keepercli-package/src/keepercli/__init__.py +++ b/keepercli-package/src/keepercli/__init__.py @@ -9,5 +9,5 @@ # Contact: commander@keepersecurity.com # -__version__ = '1.2.8' +__version__ = '1.2.9' diff --git a/keepersdk-package/src/keepersdk/__init__.py b/keepersdk-package/src/keepersdk/__init__.py index 6b7ebdc9..e63d26ea 100644 --- a/keepersdk-package/src/keepersdk/__init__.py +++ b/keepersdk-package/src/keepersdk/__init__.py @@ -10,6 +10,6 @@ # from . import background -__version__ = '1.2.8' +__version__ = '1.2.9' background.init() diff --git a/keepersdk-package/src/keepersdk/vault/nsf_common.py b/keepersdk-package/src/keepersdk/vault/nsf_common.py index 1c8d0a0e..cd53d8a6 100644 --- a/keepersdk-package/src/keepersdk/vault/nsf_common.py +++ b/keepersdk-package/src/keepersdk/vault/nsf_common.py @@ -681,15 +681,6 @@ def folder_inherits_parent_permissions(vault: VaultOnline, folder_uid: str) -> b return row.inherit_user_permissions != int(folder_pb2.BOOLEAN_FALSE) -def ensure_folder_direct_permissions( - vault: VaultOnline, - folder_uid: str) -> bool: - """Check whether folder permission inheritance is enabled.""" - if not folder_inherits_parent_permissions(vault, folder_uid): - return False - return True - - def access_role_label(access: Dict[str, Any]) -> str: if access.get('owner'): return 'owner' @@ -759,8 +750,11 @@ def plan_folder_access_change(state: str, action: str) -> List[Dict[str, Any]]: ``action`` is ``'grant'`` (set/change a role), ``'deny'`` or ``'remove'``. Each step is ``{'request': , 'include_folder_key': bool, - 'denied_access': bool}``. Steps are sent sequentially and a later step is - sent only when the previous one succeeded. + 'denied_access': bool, 'rollback': Optional[str]}``. Steps are sent + sequentially and a later step is sent only when the previous one + succeeded. ``rollback`` names the compensating action to take for a + completed step when a later step fails (``'deny'`` re-applies a denial + that was removed, so a failed grant never leaves inherited access open). * inherited -> grant : folderAccessAdds + recipient-encrypted folderKey * denied -> grant : folderAccessRemoves (denial), then @@ -774,15 +768,19 @@ def plan_folder_access_change(state: str, action: str) -> List[Dict[str, Any]]: Raises ValueError for transitions that are not allowed. """ - add_with_key = {'request': FOLDER_ACCESS_ADD, 'include_folder_key': True, 'denied_access': False} - update = {'request': FOLDER_ACCESS_UPDATE, 'include_folder_key': False, 'denied_access': False} - remove = {'request': FOLDER_ACCESS_REMOVE, 'include_folder_key': False, 'denied_access': False} + add_with_key = {'request': FOLDER_ACCESS_ADD, 'include_folder_key': True, + 'denied_access': False, 'rollback': None} + update = {'request': FOLDER_ACCESS_UPDATE, 'include_folder_key': False, + 'denied_access': False, 'rollback': None} + remove = {'request': FOLDER_ACCESS_REMOVE, 'include_folder_key': False, + 'denied_access': False, 'rollback': None} if action == 'grant': if state == FOLDER_ACCESS_DIRECT: return [update] if state == FOLDER_ACCESS_DENIED: - return [remove, add_with_key] + # If the add fails, the removed denial must be restored. + return [dict(remove, rollback='deny'), add_with_key] return [add_with_key] if action == 'deny': @@ -803,8 +801,13 @@ def plan_folder_access_change(state: str, action: str) -> List[Dict[str, Any]]: def current_folder_role_name(accessors: List[Dict[str, Any]]) -> Optional[str]: - """AccessRoleType name of the most specific row (direct before inherited).""" + """AccessRoleType name of the most specific row (direct before inherited). + + Denied rows grant nothing, so they are ignored. + """ for a in sorted(accessors or [], key=lambda r: bool(r.get('inherited'))): + if a.get('denied_access'): + continue if a.get('role'): return a['role'] return None diff --git a/keepersdk-package/src/keepersdk/vault/nsf_management.py b/keepersdk-package/src/keepersdk/vault/nsf_management.py index b0c33936..9d418cad 100644 --- a/keepersdk-package/src/keepersdk/vault/nsf_management.py +++ b/keepersdk-package/src/keepersdk/vault/nsf_management.py @@ -394,7 +394,11 @@ def get_nsf_folder_detail( include_access: bool = True, show_inherited: bool = False, show_denied: bool = False) -> Dict[str, Any]: - """Folder detail payload for ``nsf-get`` (folder branch).""" + """Folder detail payload for ``nsf-get`` (folder branch). + + Inherited / denied accessors are omitted unless ``show_inherited`` / + ``show_denied`` is True (see :func:`get_nsf_folder_access`). + """ folder = _nsf_view(vault).get_folder(folder_uid) if folder is None: raise NsfError(f'NSF folder not found: {folder_uid}') @@ -445,7 +449,11 @@ def get_nsf_record_detail( include_access: bool = True, show_inherited: bool = False, show_denied: bool = False) -> Dict[str, Any]: - """Record detail payload for ``nsf-get`` (record branch).""" + """Record detail payload for ``nsf-get`` (record branch). + + Inherited / denied accessors are omitted unless ``show_inherited`` / + ``show_denied`` is True (see :func:`get_nsf_record_accesses`). + """ meta = load_nsf_record_metadata(vault, record_uid) entry = _nsf_view(vault).get_record(record_uid) result: Dict[str, Any] = { @@ -478,7 +486,11 @@ def get_nsf_item( include_access: bool = True, show_inherited: bool = False, show_denied: bool = False) -> Dict[str, Any]: - """Resolve and return folder or record detail.""" + """Resolve and return folder or record detail. + + Inherited / denied accessors are omitted unless ``show_inherited`` / + ``show_denied`` is True. + """ folder_uid = resolve_nsf_folder_uid(vault, uid_or_title) if folder_uid: return {'item_type': 'folder', **get_nsf_folder_detail( @@ -1047,7 +1059,14 @@ def get_nsf_record_accesses( record_uids: Iterable[str], show_inherited: bool = False, show_denied: bool = False) -> Dict[str, Any]: - """``vault/records/v3/details/access``, chunked at 100 UIDs per request.""" + """``vault/records/v3/details/access``, chunked at 100 UIDs per request. + + By default only direct, non-denied rows are returned. Pass + ``show_inherited=True`` and/or ``show_denied=True`` to include inherited / + denied rows. Code that reasons about effective access (permission checks, + sharing, revocation) must pass both as True, otherwise it sees an + incomplete picture. + """ uids = [resolve_nsf_record_uid(vault, u) or u for u in record_uids] uids = [u for u in uids if u] if not uids: @@ -1066,7 +1085,7 @@ def get_nsf_record_accesses( for ra in rs.recordAccesses: d = ra.data ai = ra.accessorInfo - if (not show_inherited and d.inherited) or not (show_denied and d.deniedAccess): + if not _include_access_row(bool(d.inherited), bool(d.deniedAccess), show_inherited, show_denied): continue ao = { 'record_uid': utils.base64_url_encode(d.recordUid), @@ -1113,15 +1132,29 @@ def _resolve_uid_to_username(vault: VaultOnline, uid_b64: str) -> Optional[str]: """Safety cap on continuation-token pages consumed per UID chunk.""" -def _parse_folder_accessor(vault: VaultOnline, a: Any, show_inherited: bool = False, show_denied: bool = False) -> Dict[str, Any]: +def _include_access_row(inherited: bool, denied: bool, show_inherited: bool, show_denied: bool) -> bool: + """Shared inherited/denied filter for folder and record access rows.""" + if inherited and not show_inherited: + return False + if denied and not show_denied: + return False + return True + + +def _parse_folder_accessor( + vault: VaultOnline, + a: Any, + show_inherited: bool = False, + show_denied: bool = False) -> Optional[Dict[str, Any]]: + """Parse one folder accessor row, or return None when it is filtered out.""" + if not _include_access_row(bool(a.inherited), bool(a.deniedAccess), show_inherited, show_denied): + return None auid = utils.base64_url_encode(a.accessTypeUid) at = folder_pb2.AccessType.Name(a.accessType) rt = folder_pb2.AccessRoleType.Name(a.accessRoleType) username = None if at == 'AT_USER': username = _resolve_uid_to_username(vault, auid) - if (not show_inherited and a.inherited) or (not show_denied and a.deniedAccess): - return {} ai = { 'accessor_uid': auid, 'access_type': at, 'role': rt, 'access_role_type': int(a.accessRoleType), @@ -1154,7 +1187,10 @@ def _parse_folder_accessor(vault: VaultOnline, a: Any, show_inherited: bool = Fa def _fetch_nsf_folder_access_chunk( - vault: VaultOnline, chunk: List[str], show_inherited: bool = False, show_denied: bool = False) -> Dict[str, Dict[str, Any]]: + vault: VaultOnline, + chunk: List[str], + show_inherited: bool = False, + show_denied: bool = False) -> Dict[str, Dict[str, Any]]: """Fetch access details for one <=100-UID chunk, fully draining continuation pages.""" accumulated: Dict[str, Dict[str, Any]] = {} continuation_token = None @@ -1179,7 +1215,11 @@ def _fetch_nsf_folder_access_chunk( 'message': err.message}, 'success': False} else: - accessors = [_parse_folder_accessor(vault, a, show_inherited, show_denied) for a in fr.accessors] + accessors = [] + for a in fr.accessors: + parsed = _parse_folder_accessor(vault, a, show_inherited, show_denied) + if parsed is not None: + accessors.append(parsed) entry = accumulated.setdefault( fuid, {'folder_uid': fuid, 'accessors': [], 'success': True}) entry['accessors'].extend(accessors) @@ -1197,6 +1237,12 @@ def get_nsf_folder_access( show_denied: bool = False) -> Dict[str, Any]: """``vault/folders/v3/access``, chunked at 100 UIDs per request. + By default only direct, non-denied rows are returned. Pass + ``show_inherited=True`` and/or ``show_denied=True`` to include inherited / + denied rows. Code that reasons about effective access (permission checks, + sharing, revocation) must pass both as True, otherwise it sees an + incomplete picture. + Fully drains continuation-token pagination within each chunk before moving to the next, so the returned ``results`` are always complete — there is no caller-facing ``has_more``/``continuation_token`` to track. @@ -1217,7 +1263,7 @@ def get_nsf_folder_access( return {'results': results} -def load_nsf_access_details(vault: VaultOnline, load_folder: bool = False, load_record: bool = False, show_inherited: bool = False, show_denied: bool = False) -> Dict[str, int]: +def load_nsf_access_details(vault: VaultOnline, load_folder: bool = False, load_record: bool = False) -> Dict[str, int]: """Fetch access details for every NSF folder/record and cache them on ``vault.nsf_data``. Scans all folder and record UIDs currently known to the NSF cache @@ -1226,6 +1272,9 @@ def load_nsf_access_details(vault: VaultOnline, load_folder: bool = False, load_ the results back onto ``vault.nsf_data`` — folders keyed by folder UID (the full result entry returned by :func:`get_nsf_folder_access`), records keyed by record UID (the list of accessor rows for that record). + + The cache always holds the complete access state, including inherited and + denied rows; filter at read time if only direct access is wanted. """ view = _nsf_view(vault) if not load_folder and not load_record: @@ -1238,7 +1287,7 @@ def load_nsf_access_details(vault: VaultOnline, load_folder: bool = False, load_ loaded = {'folders': 0, 'records': 0} if folder_uid_list: - access = get_nsf_folder_access(vault, folder_uid_list, show_inherited=show_inherited, show_denied=show_denied) + access = get_nsf_folder_access(vault, folder_uid_list, show_inherited=True, show_denied=True) for fr in access.get('results') or []: fuid = fr.get('folder_uid') if not fuid: @@ -1247,7 +1296,7 @@ def load_nsf_access_details(vault: VaultOnline, load_folder: bool = False, load_ loaded['folders'] += 1 if record_uid_list: - access = get_nsf_record_accesses(vault, record_uid_list, show_inherited=show_inherited, show_denied=show_denied) + access = get_nsf_record_accesses(vault, record_uid_list, show_inherited=True, show_denied=True) by_uid: Dict[str, List[Dict[str, Any]]] = {} for ao in access.get('record_accesses') or []: by_uid.setdefault(ao['record_uid'], []).append(ao) diff --git a/keepersdk-package/src/keepersdk/vault/nsf_sharing.py b/keepersdk-package/src/keepersdk/vault/nsf_sharing.py index 988240cf..7ba1ce55 100644 --- a/keepersdk-package/src/keepersdk/vault/nsf_sharing.py +++ b/keepersdk-package/src/keepersdk/vault/nsf_sharing.py @@ -48,11 +48,6 @@ def _ensure_record_ownership_permission(vault: VaultOnline, record_uid: str) -> raise NsfError(str(exc)) from exc -def _prepare_folder_for_access_change(vault: VaultOnline, folder_uid: str) -> None: - """Check folder permission inheritance before changing folder accessors.""" - nsf_common.ensure_folder_direct_permissions(vault, folder_uid) - - @dataclass class NsfShareResult: success: bool @@ -115,27 +110,6 @@ def _resolve_folder_accessor( return uid_bytes, recipient, folder_pb2.AT_USER -def _check_existing_nsf_folder_access( - vault: VaultOnline, - folder_uid: str, - uid_bytes: bytes, - access_type_label: str) -> Optional[str]: - """Return the existing AccessRoleType name for an accessor, or None.""" - try: - uid_encoded = utils.base64_url_encode(uid_bytes) - info = get_nsf_folder_access(vault, [folder_uid], show_inherited=True, show_denied=True) - for result in info.get('results', []): - if not result.get('success'): - continue - for accessor in result.get('accessors', []): - if (accessor.get('access_type') == access_type_label - and accessor.get('accessor_uid') == uid_encoded): - return accessor.get('role') - except Exception: - pass - return None - - def _lookup_nsf_folder_accessors( vault: VaultOnline, folder_uid: str, @@ -143,25 +117,41 @@ def _lookup_nsf_folder_accessors( access_type_label: Optional[str] = None) -> List[Dict[str, Any]]: """Every server accessor row (including inherited and denied) for one actor. - Rows matching *access_type_label* are preferred; otherwise all rows for the - UID are returned. Lookup failures return an empty list (state ``none``). + Rows matching *access_type_label* are preferred; otherwise the remaining + rows for the UID are returned. ``AT_OWNER`` rows are never used as a + stand-in for another access type: if the only match is the folder owner + the call raises, so owner access is never updated, denied or removed by + accident. + + Raises :class:`NsfError` when the current access cannot be read, instead + of guessing state ``none`` and sending the wrong kind of request. """ uid_encoded = utils.base64_url_encode(uid_bytes) - rows: List[Dict[str, Any]] = [] try: info = get_nsf_folder_access(vault, [folder_uid], show_inherited=True, show_denied=True) - except Exception: - return rows + except Exception as exc: + raise NsfError(f'Could not read current access for folder {folder_uid}: {exc}') from exc + + rows: List[Dict[str, Any]] = [] for result in info.get('results', []): - if not result.get('success'): + if result.get('folder_uid') not in (None, folder_uid): continue + if not result.get('success'): + err = result.get('error') or {} + raise NsfError( + f"Could not read current access for folder {folder_uid}: " + f"{err.get('status', 'error')} {err.get('message', '')}".strip()) rows.extend(a for a in result.get('accessors', []) if a and a.get('accessor_uid') == uid_encoded) + if access_type_label: typed = [a for a in rows if a.get('access_type') == access_type_label] if typed: return typed - return rows + others = [a for a in rows if a.get('access_type') != 'AT_OWNER'] + if rows and not others: + raise NsfError('Folder owner access cannot be changed with this operation') + return others def _encrypted_folder_key_for( @@ -253,30 +243,90 @@ def _execute_folder_access_plan( For the denied re-add, the add is only sent once the denial removal succeeded, so no add is attempted while the denied row still exists. + + Every request is fully built (including the encrypted folder key) before + the first one is sent, so client-side failures cannot leave a partially + applied plan. If a later step is rejected by the server, completed steps + that carry a ``rollback`` action are compensated (a removed denial is + re-applied) before the error is raised. """ msgs = dict(_FOLDER_STEP_MESSAGES, **(messages or {})) - result: Dict[str, Any] = {} - for idx, step in enumerate(plan): - folder_key = folder_key_factory() if step['include_folder_key'] else None - ad = _build_folder_access_step( - folder_uid, uid_bytes, access_type_enum, step, folder_key=folder_key, **fields) - request = step['request'] - response = _folder_access_update( + + folder_key: Optional[folder_pb2.EncryptedDataKey] = None + if any(step['include_folder_key'] for step in plan): + if folder_key_factory is None: + raise NsfError('Folder key is required to create direct folder access') + folder_key = folder_key_factory() + requests = [ + _build_folder_access_step( + folder_uid, uid_bytes, access_type_enum, step, + folder_key=folder_key if step['include_folder_key'] else None, **fields) + for step in plan + ] + + def _send(request: str, ad: folder_pb2.FolderAccessData) -> folder_pb2.FolderAccessResponse: + return _folder_access_update( vault, adds=[ad] if request == nsf_common.FOLDER_ACCESS_ADD else None, updates=[ad] if request == nsf_common.FOLDER_ACCESS_UPDATE else None, removes=[ad] if request == nsf_common.FOLDER_ACCESS_REMOVE else None) + + result: Dict[str, Any] = {} + completed: List[Dict[str, Any]] = [] + for idx, (step, ad) in enumerate(zip(plan, requests)): + request = step['request'] message = 'Access denied successfully' if step['denied_access'] else msgs[request] - result = nsf_common.parse_folder_access_result(response, folder_uid, label, message) + result = nsf_common.parse_folder_access_result(_send(request, ad), folder_uid, label, message) result['request'] = request - if not result['success']: - msg = result['message'] - if idx < len(plan) - 1: - msg = f'{msg} (step {idx + 1} of {len(plan)}; remaining steps skipped)' - raise KeeperApiError(result['status'], msg) + if result['success']: + completed.append(step) + continue + + msg = result['message'] + if idx < len(plan) - 1: + msg = f'{msg} (step {idx + 1} of {len(plan)}; remaining steps skipped)' + rollback_notes = _rollback_folder_access_steps( + vault, folder_uid, label, uid_bytes, access_type_enum, completed) + if rollback_notes: + msg = f"{msg}; {'; '.join(rollback_notes)}" + raise KeeperApiError(result['status'], msg) return result +def _rollback_folder_access_steps( + vault: VaultOnline, + folder_uid: str, + label: str, + uid_bytes: bytes, + access_type_enum: int, + completed: List[Dict[str, Any]]) -> List[str]: + """Compensate completed plan steps after a later step failed. + + Returns human-readable notes describing what was (or could not be) undone. + """ + notes: List[str] = [] + for step in reversed(completed): + if step.get('rollback') != 'deny': + continue + deny_step = {'request': nsf_common.FOLDER_ACCESS_UPDATE, 'include_folder_key': False, + 'denied_access': True, 'rollback': None} + ad = _build_folder_access_step(folder_uid, uid_bytes, access_type_enum, deny_step) + try: + response = _folder_access_update(vault, updates=[ad]) + restored = nsf_common.parse_folder_access_result( + response, folder_uid, label, 'Access denied successfully') + ok = bool(restored.get('success')) + except Exception: + ok = False + if ok: + notes.append('the previous denial was restored') + else: + notes.append( + 'WARNING: the previous denial was removed and could not be restored; ' + 'the accessor may now have inherited access to this folder') + return notes + + def collect_nsf_records_in_folder( vault: VaultOnline, folder_identifier: Optional[str], @@ -309,6 +359,113 @@ def walk(folder_uid: str, visited: Set[str]) -> None: return record_uids +def _already_had_access_result( + folder_uid: str, label: str, access_type_label: str, message: str, state: str) -> Dict[str, Any]: + return { + 'folder_uid': folder_uid, + 'accessor': label, + 'access_type': access_type_label, + 'status': 'SUCCESS', + 'message': message, + 'success': True, + 'action_taken': 'already_had_access', + 'previous_access_state': state, + } + + +def _update_folder_access_for_state( + vault: VaultOnline, + folder_uid: str, + label: str, + uid_bytes: bytes, + access_type_enum: int, + state: str, + *, + folder_key_factory, + role: Optional[str] = None, + hidden: Optional[bool] = None, + expiration_timestamp: Optional[int] = None, + messages: Optional[Dict[str, str]] = None) -> Dict[str, Any]: + """Apply a role / hidden / expiration change given the accessor's current *state*. + + Direct access is changed in place with ``folderAccessUpdates``. Inherited or + denied access can only be changed by turning it into a direct grant + (``folderAccessAdds`` + folder key), which then survives a later revoke on + the parent folder. Because that widens access, it requires an explicit + *role*: changing only ``hidden`` or the expiration never silently creates a + direct grant. + """ + if state in (nsf_common.FOLDER_ACCESS_DIRECT, nsf_common.FOLDER_ACCESS_NONE): + # NONE: nothing known to convert; let the server validate the update. + plan = [{'request': nsf_common.FOLDER_ACCESS_UPDATE, 'include_folder_key': False, + 'denied_access': False, 'rollback': None}] + else: + if role is None: + raise NsfError( + f'Access on this folder is {state}. Changing it creates a direct grant ' + 'that persists even if access to the parent folder is revoked; pass an ' + 'explicit role to confirm.') + plan = nsf_common.plan_folder_access_change(state, 'grant') + + result = _execute_folder_access_plan( + vault, folder_uid, label, uid_bytes, access_type_enum, plan, + folder_key_factory=folder_key_factory, + messages=messages, + role=role, hidden=hidden, expiration_timestamp=expiration_timestamp) + result['previous_access_state'] = state + return result + + +def _revoke_folder_access_for_rows( + vault: VaultOnline, + folder_uid: str, + label: str, + uid_bytes: bytes, + access_type_enum: int, + access_type_label: str, + accessors: List[Dict[str, Any]], + *, + messages: Optional[Dict[str, str]] = None) -> Dict[str, Any]: + """Remove direct access, deny inherited access, or no-op for denied access.""" + if accessors: + # Use the server-reported UID / access type (e.g. inherited team rows). + # _lookup_nsf_folder_accessors never returns AT_OWNER rows here. + server_uid = accessors[0].get('accessor_uid') + if server_uid: + uid_bytes = utils.base64_url_decode(server_uid) + server_type = accessors[0].get('access_type') + if server_type: + try: + access_type_enum = folder_pb2.AccessType.Value(server_type) + access_type_label = server_type + except ValueError as exc: + raise NsfError( + f"Unrecognised access type '{server_type}' returned by server " + f"for folder '{folder_uid}'") from exc + state = nsf_common.folder_access_state_from_accessors(accessors) + + action = 'deny' if state == nsf_common.FOLDER_ACCESS_INHERITED else 'remove' + plan = nsf_common.plan_folder_access_change(state, action) + if not plan: + return { + 'folder_uid': folder_uid, + 'accessor': label, + 'access_type': access_type_label, + 'status': 'SUCCESS', + 'message': 'Access is already denied on this folder', + 'success': True, + 'action_taken': 'already_denied', + 'previous_access_state': state, + } + + result = _execute_folder_access_plan( + vault, folder_uid, label, uid_bytes, access_type_enum, plan, messages=messages) + result['access_type'] = access_type_label + result['previous_access_state'] = state + result['action_taken'] = 'denied' if action == 'deny' else 'revoked' + return result + + def grant_nsf_folder_access( vault: VaultOnline, folder_identifier: str, @@ -326,14 +483,14 @@ def grant_nsf_folder_access( * inherited - becomes a direct grant via ``folderAccessAdds`` with the recipient-encrypted folder key (even when the role is unchanged) * denied - ``folderAccessRemoves`` for the denial, then - ``folderAccessAdds`` with the folder key + ``folderAccessAdds`` with the folder key (the denial is restored if the + add fails) * none - ``folderAccessAdds`` with the folder key """ folder_uid = resolve_nsf_folder_uid(vault, folder_identifier) or folder_identifier if not is_nsf_folder(vault, folder_uid): raise NsfError(f'NSF folder not found: {folder_identifier}') _ensure_folder_share_permission(vault, folder_uid) - _prepare_folder_for_access_change(vault, folder_uid) access_role = nsf_common.resolve_nsf_role(role) target_role_name = folder_pb2.AccessRoleType.Name(access_role) @@ -345,31 +502,28 @@ def grant_nsf_folder_access( accessors = _lookup_nsf_folder_accessors(vault, folder_uid, uid_bytes, access_type_label) state = nsf_common.folder_access_state_from_accessors(accessors) + def key_factory(): + return _encrypted_folder_key_for(vault, folder_uid, uid_bytes, recipient, as_team=as_team) + if state == nsf_common.FOLDER_ACCESS_DIRECT: if (nsf_common.current_folder_role_name(accessors) == target_role_name and expiration_timestamp is None): - return { - 'folder_uid': folder_uid, - 'accessor': label, - 'access_type': access_type_label, - 'status': 'SUCCESS', - 'message': f"{'Team' if as_team else 'User'} already has {role} access", - 'success': True, - 'action_taken': 'already_had_access', - 'previous_access_state': state, - } - result = update_nsf_folder_access( - vault, folder_uid, recipient, role=role, as_team=as_team, - expiration_timestamp=expiration_timestamp, request_sync=request_sync, - _known_state=state) + return _already_had_access_result( + folder_uid, label, access_type_label, + f"{'Team' if as_team else 'User'} already has {role} access", state) + result = _update_folder_access_for_state( + vault, folder_uid, label, uid_bytes, access_type_enum, state, + folder_key_factory=key_factory, + role=role, expiration_timestamp=expiration_timestamp) + result['access_type'] = access_type_label result['action_taken'] = 'updated' + _request_sync(vault, request_sync) return result plan = nsf_common.plan_folder_access_change(state, 'grant') result = _execute_folder_access_plan( vault, folder_uid, label, uid_bytes, access_type_enum, plan, - folder_key_factory=lambda: _encrypted_folder_key_for( - vault, folder_uid, uid_bytes, recipient, as_team=as_team), + folder_key_factory=key_factory, role=role, expiration_timestamp=expiration_timestamp) result['access_type'] = access_type_label result['previous_access_state'] = state @@ -383,6 +537,27 @@ def _nsf_app_role_for_editable(is_editable: bool) -> str: return 'content-manager' if is_editable else 'viewer' +def _app_folder_key_factory(vault: VaultOnline, folder_uid: str, app_uid: str): + """Return a callable producing the folder key encrypted with the KSM app's record key.""" + def factory() -> folder_pb2.EncryptedDataKey: + app_key = vault.vault_data.get_record_key(app_uid) + if not app_key: + raise NsfError(f'Could not resolve record key for application {app_uid}') + folder_key = _get_folder_key(vault, folder_uid) + ek = folder_pb2.EncryptedDataKey() + ek.encryptedKey = crypto.encrypt_aes_v2(folder_key, app_key) + ek.encryptedKeyType = folder_pb2.encrypted_by_data_key_gcm + return ek + return factory + + +_APP_STEP_MESSAGES = { + nsf_common.FOLDER_ACCESS_ADD: 'Application access granted successfully', + nsf_common.FOLDER_ACCESS_UPDATE: 'Application access updated successfully', + nsf_common.FOLDER_ACCESS_REMOVE: 'Application access revoked successfully', +} + + def grant_nsf_folder_to_application( vault: VaultOnline, folder_identifier: str, @@ -391,15 +566,17 @@ def grant_nsf_folder_to_application( is_editable: bool = False, request_sync: bool = True) -> Dict[str, Any]: """Share an NSF folder with a KSM application via AT_APPLICATION. + + Uses the same direct / inherited / denied / none transitions as + :func:`grant_nsf_folder_access`, with the folder key encrypted by the + application's record key. """ folder_uid = resolve_nsf_folder_uid(vault, folder_identifier) or folder_identifier if not is_nsf_folder(vault, folder_uid): raise NsfError(f'NSF folder not found: {folder_identifier}') _ensure_folder_share_permission(vault, folder_uid) - _prepare_folder_for_access_change(vault, folder_uid) - app_key = vault.vault_data.get_record_key(app_uid) - if not app_key: + if not vault.vault_data.get_record_key(app_uid): raise NsfError(f'Could not resolve record key for application {app_uid}') role = _nsf_app_role_for_editable(is_editable) @@ -407,43 +584,30 @@ def grant_nsf_folder_to_application( target_role_name = folder_pb2.AccessRoleType.Name(access_role) app_uid_bytes = utils.base64_url_decode(app_uid) - existing_role = _check_existing_nsf_folder_access( - vault, folder_uid, app_uid_bytes, 'AT_APPLICATION') - if existing_role is not None: - if existing_role == target_role_name: - return { - 'folder_uid': folder_uid, - 'accessor': app_uid, - 'access_type': 'AT_APPLICATION', - 'status': 'SUCCESS', - 'message': f'Application already has {role} access', - 'success': True, - 'action_taken': 'already_had_access', - } - return update_nsf_folder_application_access( - vault, folder_uid, app_uid, is_editable=is_editable, request_sync=request_sync) - - ad = folder_pb2.FolderAccessData() - ad.folderUid = utils.base64_url_decode(folder_uid) - ad.accessTypeUid = app_uid_bytes - ad.accessType = folder_pb2.AT_APPLICATION - ad.accessRoleType = access_role - ad.permissions.CopyFrom(nsf_common.get_folder_permissions_for_role(access_role)) + accessors = _lookup_nsf_folder_accessors(vault, folder_uid, app_uid_bytes, 'AT_APPLICATION') + state = nsf_common.folder_access_state_from_accessors(accessors) + key_factory = _app_folder_key_factory(vault, folder_uid, app_uid) - folder_key = _get_folder_key(vault, folder_uid) - ek = folder_pb2.EncryptedDataKey() - ek.encryptedKey = crypto.encrypt_aes_v2(folder_key, app_key) - ek.encryptedKeyType = folder_pb2.encrypted_by_data_key_gcm - ad.folderKey.CopyFrom(ek) + if state == nsf_common.FOLDER_ACCESS_DIRECT: + if nsf_common.current_folder_role_name(accessors) == target_role_name: + return _already_had_access_result( + folder_uid, app_uid, 'AT_APPLICATION', + f'Application already has {role} access', state) + result = _update_folder_access_for_state( + vault, folder_uid, app_uid, app_uid_bytes, folder_pb2.AT_APPLICATION, state, + folder_key_factory=key_factory, role=role, messages=_APP_STEP_MESSAGES) + result['access_type'] = 'AT_APPLICATION' + result['action_taken'] = 'updated' + _request_sync(vault, request_sync) + return result - response = _folder_access_update(vault, adds=[ad]) - result = nsf_common.parse_folder_access_result( - response, folder_uid, app_uid, 'Application access granted successfully') + plan = nsf_common.plan_folder_access_change(state, 'grant') + result = _execute_folder_access_plan( + vault, folder_uid, app_uid, app_uid_bytes, folder_pb2.AT_APPLICATION, plan, + folder_key_factory=key_factory, messages=_APP_STEP_MESSAGES, role=role) result['access_type'] = 'AT_APPLICATION' - result.setdefault( - 'action_taken', 'granted' if result['success'] else 'grant_failed') - if not result['success']: - raise KeeperApiError(result['status'], result['message']) + result['previous_access_state'] = state + result.setdefault('action_taken', 'granted') _request_sync(vault, request_sync) return result @@ -455,32 +619,28 @@ def update_nsf_folder_application_access( *, is_editable: bool = False, request_sync: bool = True) -> Dict[str, Any]: - """Update AT_APPLICATION role for an NSF folder already shared with a KSM app.""" + """Update the AT_APPLICATION role for a KSM app on an NSF folder. + + Inherited or denied application access becomes a direct grant carrying + the app-encrypted folder key (see :func:`update_nsf_folder_access`). + """ folder_uid = resolve_nsf_folder_uid(vault, folder_identifier) or folder_identifier if not is_nsf_folder(vault, folder_uid): raise NsfError(f'NSF folder not found: {folder_identifier}') _ensure_folder_share_permission(vault, folder_uid) - _prepare_folder_for_access_change(vault, folder_uid) role = _nsf_app_role_for_editable(is_editable) - access_role = nsf_common.resolve_nsf_role(role) app_uid_bytes = utils.base64_url_decode(app_uid) + accessors = _lookup_nsf_folder_accessors(vault, folder_uid, app_uid_bytes, 'AT_APPLICATION') + state = nsf_common.folder_access_state_from_accessors(accessors) - ad = folder_pb2.FolderAccessData() - ad.folderUid = utils.base64_url_decode(folder_uid) - ad.accessTypeUid = app_uid_bytes - ad.accessType = folder_pb2.AT_APPLICATION - ad.accessRoleType = access_role - ad.permissions.CopyFrom(nsf_common.get_folder_permissions_for_role(access_role)) - - response = _folder_access_update(vault, updates=[ad]) - result = nsf_common.parse_folder_access_result( - response, folder_uid, app_uid, 'Application access updated successfully') + result = _update_folder_access_for_state( + vault, folder_uid, app_uid, app_uid_bytes, folder_pb2.AT_APPLICATION, state, + folder_key_factory=_app_folder_key_factory(vault, folder_uid, app_uid), + role=role, messages=dict(_APP_STEP_MESSAGES, **{ + nsf_common.FOLDER_ACCESS_ADD: 'Application access updated successfully'})) result['access_type'] = 'AT_APPLICATION' - result.setdefault( - 'action_taken', 'updated' if result['success'] else 'update_failed') - if not result['success']: - raise KeeperApiError(result['status'], result['message']) + result.setdefault('action_taken', 'updated') _request_sync(vault, request_sync) return result @@ -491,25 +651,22 @@ def revoke_nsf_folder_from_application( app_uid: str, *, request_sync: bool = True) -> Dict[str, Any]: - """Revoke AT_APPLICATION access for a KSM app on an NSF folder.""" + """Revoke AT_APPLICATION access for a KSM app on an NSF folder. + + Direct access is removed; inherited access is denied on this folder. + """ folder_uid = resolve_nsf_folder_uid(vault, folder_identifier) or folder_identifier if not is_nsf_folder(vault, folder_uid): raise NsfError(f'NSF folder not found: {folder_identifier}') _ensure_folder_share_permission(vault, folder_uid) - _prepare_folder_for_access_change(vault, folder_uid) - ad = folder_pb2.FolderAccessData() - ad.folderUid = utils.base64_url_decode(folder_uid) - ad.accessTypeUid = utils.base64_url_decode(app_uid) - ad.accessType = folder_pb2.AT_APPLICATION - - response = _folder_access_update(vault, removes=[ad]) - result = nsf_common.parse_folder_access_result( - response, folder_uid, app_uid, 'Application access revoked successfully') - result['access_type'] = 'AT_APPLICATION' - if not result['success']: - raise KeeperApiError(result['status'], result['message']) - _request_sync(vault, request_sync) + app_uid_bytes = utils.base64_url_decode(app_uid) + accessors = _lookup_nsf_folder_accessors(vault, folder_uid, app_uid_bytes, 'AT_APPLICATION') + result = _revoke_folder_access_for_rows( + vault, folder_uid, app_uid, app_uid_bytes, folder_pb2.AT_APPLICATION, + 'AT_APPLICATION', accessors, messages=_APP_STEP_MESSAGES) + if result.get('action_taken') != 'already_denied': + _request_sync(vault, request_sync) return result @@ -522,14 +679,14 @@ def update_nsf_folder_access( hidden: Optional[bool] = None, expiration_timestamp: Optional[int] = None, as_team: bool = False, - request_sync: bool = True, - _known_state: Optional[str] = None) -> Dict[str, Any]: + request_sync: bool = True) -> Dict[str, Any]: """Update role, visibility, or expiration for an NSF folder accessor. Direct access is changed with ``folderAccessUpdates`` and keeps its folder key. Inherited (or denied) access is turned into a direct grant with ``folderAccessAdds`` plus the recipient-encrypted folder key, so the child - keeps its own key edge if access to the parent is later revoked. + keeps its own key edge if access to the parent is later revoked. That + conversion widens access, so it requires an explicit ``role``. """ if role is None and hidden is None and expiration_timestamp is None: raise NsfError('At least one field (role, hidden, or expiration) is required') @@ -538,41 +695,21 @@ def update_nsf_folder_access( if not is_nsf_folder(vault, folder_uid): raise NsfError(f'NSF folder not found: {folder_identifier}') _ensure_folder_share_permission(vault, folder_uid) - _prepare_folder_for_access_change(vault, folder_uid) uid_bytes, label, access_type_enum = _resolve_folder_accessor( vault, recipient, as_team=as_team) - accessors: List[Dict[str, Any]] = [] - state = _known_state - if state is None: - accessors = _lookup_nsf_folder_accessors( - vault, folder_uid, uid_bytes, folder_pb2.AccessType.Name(access_type_enum)) - state = nsf_common.folder_access_state_from_accessors(accessors) - - if state in (nsf_common.FOLDER_ACCESS_DIRECT, nsf_common.FOLDER_ACCESS_NONE): - # NONE: nothing known to convert; let the server validate the update. - plan = [{'request': nsf_common.FOLDER_ACCESS_UPDATE, - 'include_folder_key': False, 'denied_access': False}] - else: - plan = nsf_common.plan_folder_access_change(state, 'grant') - if role is None: - # An add needs a role; keep the one currently in effect. - current = (nsf_common.current_folder_role_name(accessors) or '').lower() - if current not in nsf_common.ROLE_NAME_MAP: - raise NsfError( - 'A role is required to change inherited or denied access on ' - 'this folder (it becomes a direct grant)') - role = current + accessors = _lookup_nsf_folder_accessors( + vault, folder_uid, uid_bytes, folder_pb2.AccessType.Name(access_type_enum)) + state = nsf_common.folder_access_state_from_accessors(accessors) - result = _execute_folder_access_plan( - vault, folder_uid, label, uid_bytes, access_type_enum, plan, + result = _update_folder_access_for_state( + vault, folder_uid, label, uid_bytes, access_type_enum, state, folder_key_factory=lambda: _encrypted_folder_key_for( vault, folder_uid, uid_bytes, recipient, as_team=as_team), messages={nsf_common.FOLDER_ACCESS_ADD: 'Access updated successfully'}, role=role, hidden=hidden, expiration_timestamp=expiration_timestamp) result['access_type'] = 'AT_TEAM' if as_team else 'AT_USER' - result['previous_access_state'] = state _request_sync(vault, request_sync) return result @@ -591,54 +728,22 @@ def revoke_nsf_folder_access( parent), so it is denied with ``folderAccessUpdates`` + ``deniedAccess=True`` and no folder key. * Already-denied access needs no request. + * Folder owner access is never touched (raises :class:`NsfError`). """ folder_uid = resolve_nsf_folder_uid(vault, folder_identifier) or folder_identifier if not is_nsf_folder(vault, folder_uid): raise NsfError(f'NSF folder not found: {folder_identifier}') _ensure_folder_share_permission(vault, folder_uid) - _prepare_folder_for_access_change(vault, folder_uid) uid_bytes, label, access_type_enum = _resolve_folder_accessor( vault, recipient, as_team=as_team) access_type_label = 'AT_TEAM' if as_team else 'AT_USER' accessors = _lookup_nsf_folder_accessors(vault, folder_uid, uid_bytes, access_type_label) - if accessors: - # Use the server-reported UID / access type (e.g. inherited team rows). - server_uid = accessors[0].get('accessor_uid') - if server_uid: - uid_bytes = utils.base64_url_decode(server_uid) - server_type = accessors[0].get('access_type') - if server_type: - try: - access_type_enum = folder_pb2.AccessType.Value(server_type) - access_type_label = server_type - except ValueError as exc: - raise NsfError( - f"Unrecognised access type '{server_type}' returned by server " - f"for folder '{folder_uid}'") from exc - state = nsf_common.folder_access_state_from_accessors(accessors) - - action = 'deny' if state == nsf_common.FOLDER_ACCESS_INHERITED else 'remove' - plan = nsf_common.plan_folder_access_change(state, action) - if not plan: - return { - 'folder_uid': folder_uid, - 'accessor': label, - 'access_type': access_type_label, - 'status': 'SUCCESS', - 'message': 'Access is already denied on this folder', - 'success': True, - 'action_taken': 'already_denied', - 'previous_access_state': state, - } - - result = _execute_folder_access_plan( - vault, folder_uid, label, uid_bytes, access_type_enum, plan) - result['access_type'] = access_type_label - result['previous_access_state'] = state - result['action_taken'] = 'denied' if action == 'deny' else 'revoked' - _request_sync(vault, request_sync) + result = _revoke_folder_access_for_rows( + vault, folder_uid, label, uid_bytes, access_type_enum, access_type_label, accessors) + if result.get('action_taken') != 'already_denied': + _request_sync(vault, request_sync) return result diff --git a/keepersdk-package/unit_tests/test_nsf_folder_access_transitions.py b/keepersdk-package/unit_tests/test_nsf_folder_access_transitions.py index d6bb1ee0..afa88ff3 100644 --- a/keepersdk-package/unit_tests/test_nsf_folder_access_transitions.py +++ b/keepersdk-package/unit_tests/test_nsf_folder_access_transitions.py @@ -6,7 +6,7 @@ from keepersdk import utils from keepersdk.errors import KeeperApiError from keepersdk.proto import folder_pb2 -from keepersdk.vault import nsf_common, nsf_sharing +from keepersdk.vault import nsf_common, nsf_management, nsf_sharing _S = 'keepersdk.vault.nsf_sharing.' @@ -71,7 +71,6 @@ def setUp(self): mock.patch(_S + 'resolve_nsf_folder_uid', return_value=self.folder_uid), mock.patch(_S + 'is_nsf_folder', return_value=True), mock.patch(_S + '_ensure_folder_share_permission'), - mock.patch(_S + '_prepare_folder_for_access_change'), mock.patch(_S + '_request_sync'), mock.patch(_S + '_resolve_folder_accessor', return_value=(self.uid_bytes, self.email, folder_pb2.AT_USER)), @@ -165,11 +164,20 @@ def test_grant_direct_new_role_updates_without_key(self): self.assertFalse(ad.HasField('folderKey')) self.enc.assert_not_called() - def test_update_inherited_keeps_current_role_and_adds_key(self): + def test_update_inherited_requires_explicit_role(self): + self._rows(_row(self.uid_b64, inherited=True)) + upd = self._access() + with self.assertRaises(nsf_management.NsfError): + nsf_sharing.update_nsf_folder_access( + self.vault, self.folder_uid, self.email, expiration_timestamp=1_900_000_000_000) + upd.assert_not_called() + + def test_update_inherited_with_role_adds_key(self): self._rows(_row(self.uid_b64, inherited=True)) upd = self._access(_response(0)) nsf_sharing.update_nsf_folder_access( - self.vault, self.folder_uid, self.email, expiration_timestamp=1_900_000_000_000) + self.vault, self.folder_uid, self.email, role='viewer', + expiration_timestamp=1_900_000_000_000) ad = upd.call_args.kwargs['adds'][0] self.assertEqual(ad.accessRoleType, folder_pb2.VIEWER) self.assertTrue(ad.HasField('folderKey')) @@ -213,6 +221,143 @@ def test_revoke_failure_raises(self): with self.assertRaises(KeeperApiError): nsf_sharing.revoke_nsf_folder_access(self.vault, self.folder_uid, self.email) + def test_grant_denied_restores_denial_when_add_fails(self): + self._rows(_row(self.uid_b64, inherited=True, denied=True)) + upd = self._access(_response(0), _response(_FAILURE), _response(0)) + with self.assertRaises(KeeperApiError) as ctx: + nsf_sharing.grant_nsf_folder_access( + self.vault, self.folder_uid, self.email, role='viewer') + self.assertEqual(upd.call_count, 3) + restore = upd.call_args_list[2].kwargs['updates'][0] + self.assertTrue(restore.deniedAccess) + self.assertFalse(restore.HasField('folderKey')) + self.assertIn('denial was restored', str(ctx.exception)) + + def test_grant_denied_warns_when_denial_cannot_be_restored(self): + self._rows(_row(self.uid_b64, denied=True)) + self._access(_response(0), _response(_FAILURE), _response(_FAILURE)) + with self.assertRaises(KeeperApiError) as ctx: + nsf_sharing.grant_nsf_folder_access( + self.vault, self.folder_uid, self.email, role='viewer') + self.assertIn('could not be restored', str(ctx.exception)) + + def test_grant_denied_key_failure_sends_nothing(self): + self._rows(_row(self.uid_b64, denied=True)) + upd = self._access() + self.enc.side_effect = nsf_management.NsfError('no public key') + with self.assertRaises(nsf_management.NsfError): + nsf_sharing.grant_nsf_folder_access( + self.vault, self.folder_uid, self.email, role='viewer') + upd.assert_not_called() + + def test_lookup_failure_raises_instead_of_assuming_none(self): + p = mock.patch(_S + 'get_nsf_folder_access', side_effect=RuntimeError('network')) + p.start() + self.addCleanup(p.stop) + upd = self._access() + with self.assertRaises(nsf_management.NsfError): + nsf_sharing.grant_nsf_folder_access(self.vault, self.folder_uid, self.email) + upd.assert_not_called() + + def test_lookup_error_result_raises(self): + info = {'results': [{'folder_uid': self.folder_uid, 'success': False, + 'error': {'status': 'ACCESS_DENIED', 'message': 'nope'}}]} + p = mock.patch(_S + 'get_nsf_folder_access', return_value=info) + p.start() + self.addCleanup(p.stop) + upd = self._access() + with self.assertRaises(nsf_management.NsfError): + nsf_sharing.revoke_nsf_folder_access(self.vault, self.folder_uid, self.email) + upd.assert_not_called() + + def test_revoke_owner_is_refused(self): + self._rows(_row(self.uid_b64, access_type='AT_OWNER', role='MANAGER')) + upd = self._access() + with self.assertRaises(nsf_management.NsfError): + nsf_sharing.revoke_nsf_folder_access(self.vault, self.folder_uid, self.email) + upd.assert_not_called() + + def test_grant_owner_is_refused(self): + self._rows(_row(self.uid_b64, access_type='AT_OWNER', role='MANAGER')) + upd = self._access() + with self.assertRaises(nsf_management.NsfError): + nsf_sharing.grant_nsf_folder_access(self.vault, self.folder_uid, self.email) + upd.assert_not_called() + + +class TestNsfApplicationFolderAccess(unittest.TestCase): + + def setUp(self): + self.folder_uid = utils.generate_uid() + self.app_uid = utils.generate_uid() + self.vault = mock.Mock() + self.vault.vault_data.get_record_key.return_value = b'\x01' * 32 + for target, kw in ( + ('resolve_nsf_folder_uid', {'return_value': self.folder_uid}), + ('is_nsf_folder', {'return_value': True}), + ('_ensure_folder_share_permission', {}), + ('_request_sync', {}), + ('_get_folder_key', {'return_value': b'\x02' * 32})): + p = mock.patch(_S + target, **kw) + p.start() + self.addCleanup(p.stop) + + def _rows(self, *rows): + info = {'results': [{'folder_uid': self.folder_uid, 'success': True, + 'accessors': list(rows)}]} + p = mock.patch(_S + 'get_nsf_folder_access', return_value=info) + p.start() + self.addCleanup(p.stop) + + def _access(self, *responses): + p = mock.patch(_S + '_folder_access_update', side_effect=list(responses)) + m = p.start() + self.addCleanup(p.stop) + return m + + def test_grant_inherited_app_adds_with_key(self): + self._rows(_row(self.app_uid, inherited=True, access_type='AT_APPLICATION')) + upd = self._access(_response(0)) + result = nsf_sharing.grant_nsf_folder_to_application( + self.vault, self.folder_uid, self.app_uid) + self.assertEqual(result['previous_access_state'], 'inherited') + ad = upd.call_args.kwargs['adds'][0] + self.assertTrue(ad.HasField('folderKey')) + self.assertEqual(ad.folderKey.encryptedKeyType, folder_pb2.encrypted_by_data_key_gcm) + self.assertEqual(ad.accessType, folder_pb2.AT_APPLICATION) + + def test_grant_direct_app_same_role_is_noop(self): + self._rows(_row(self.app_uid, access_type='AT_APPLICATION')) + upd = self._access() + result = nsf_sharing.grant_nsf_folder_to_application( + self.vault, self.folder_uid, self.app_uid) + self.assertEqual(result['action_taken'], 'already_had_access') + upd.assert_not_called() + + def test_grant_direct_app_new_role_updates(self): + self._rows(_row(self.app_uid, access_type='AT_APPLICATION')) + upd = self._access(_response(0)) + nsf_sharing.grant_nsf_folder_to_application( + self.vault, self.folder_uid, self.app_uid, is_editable=True) + ad = upd.call_args.kwargs['updates'][0] + self.assertEqual(ad.accessRoleType, folder_pb2.CONTENT_MANAGER) + self.assertFalse(ad.HasField('folderKey')) + + def test_update_inherited_app_becomes_direct_grant(self): + self._rows(_row(self.app_uid, inherited=True, access_type='AT_APPLICATION')) + upd = self._access(_response(0)) + nsf_sharing.update_nsf_folder_application_access( + self.vault, self.folder_uid, self.app_uid, is_editable=True) + self.assertTrue(upd.call_args.kwargs['adds'][0].HasField('folderKey')) + + def test_revoke_inherited_app_is_denied(self): + self._rows(_row(self.app_uid, inherited=True, access_type='AT_APPLICATION')) + upd = self._access(_response(0)) + result = nsf_sharing.revoke_nsf_folder_from_application( + self.vault, self.folder_uid, self.app_uid) + self.assertEqual(result['action_taken'], 'denied') + self.assertTrue(upd.call_args.kwargs['updates'][0].deniedAccess) + if __name__ == '__main__': unittest.main() diff --git a/keepersdk-package/unit_tests/test_nsf_permissions.py b/keepersdk-package/unit_tests/test_nsf_permissions.py index 59d8481a..9e7efc1b 100644 --- a/keepersdk-package/unit_tests/test_nsf_permissions.py +++ b/keepersdk-package/unit_tests/test_nsf_permissions.py @@ -2,8 +2,8 @@ from unittest.mock import MagicMock, patch from keepersdk import utils -from keepersdk.proto import folder_pb2 -from keepersdk.vault import memory_nsf_storage, nsf_common, nsf_storage_types as nsf +from keepersdk.proto import folder_access_pb2, folder_pb2, record_details_pb2 +from keepersdk.vault import memory_nsf_storage, nsf_common, nsf_management, nsf_storage_types as nsf class TestNsfPermissions(unittest.TestCase): @@ -127,211 +127,133 @@ def test_folder_inherit_detection(self): ]) self.assertFalse(nsf_common.folder_inherits_parent_permissions(vault, folder_uid)) - def test_record_access_filter_hides_inherited_by_default(self): - accesses = [ - { - 'accessor_name': 'direct@example.com', - 'inherited': False, - 'denied_access': False, - }, - { - 'accessor_name': 'inherited@example.com', - 'inherited': True, - 'denied_access': False, - }, - ] - - filtered = [ - access for access in accesses - if not access.get('inherited') and not access.get('denied_access') - ] - - self.assertEqual( - ['direct@example.com'], - [access['accessor_name'] for access in filtered], - ) - - def test_record_access_filter_shows_inherited_when_requested(self): - accesses = [ - { - 'accessor_name': 'direct@example.com', - 'inherited': False, - 'denied_access': False, - }, - { - 'accessor_name': 'inherited@example.com', - 'inherited': True, - 'denied_access': False, - }, - ] - - show_inherited = True - show_denied = False - - filtered = [ - access for access in accesses - if (show_inherited or not access.get('inherited')) - and (show_denied or not access.get('denied_access')) - ] - - self.assertEqual( - [ - 'direct@example.com', - 'inherited@example.com', - ], - [access['accessor_name'] for access in filtered], - ) - - def test_record_access_filter_hides_denied_by_default(self): - accesses = [ - { - 'accessor_name': 'direct@example.com', - 'inherited': False, - 'denied_access': False, - }, - { - 'accessor_name': 'denied@example.com', - 'inherited': False, - 'denied_access': True, - }, - ] - - filtered = [ - access for access in accesses - if not access.get('inherited') and not access.get('denied_access') - ] - - self.assertEqual( - ['direct@example.com'], - [access['accessor_name'] for access in filtered], - ) - - def test_record_access_filter_shows_denied_when_requested(self): - accesses = [ - { - 'accessor_name': 'direct@example.com', - 'inherited': False, - 'denied_access': False, - }, - { - 'accessor_name': 'denied@example.com', - 'inherited': False, - 'denied_access': True, - }, - ] - - show_inherited = False - show_denied = True - - filtered = [ - access for access in accesses - if (show_inherited or not access.get('inherited')) - and (show_denied or not access.get('denied_access')) - ] - - self.assertEqual( - [ - 'direct@example.com', - 'denied@example.com', - ], - [access['accessor_name'] for access in filtered], - ) - - def test_record_access_filter_shows_inherited_and_denied_when_requested(self): - accesses = [ - { - 'accessor_name': 'direct@example.com', - 'inherited': False, - 'denied_access': False, - }, - { - 'accessor_name': 'inherited@example.com', - 'inherited': True, - 'denied_access': False, - }, - { - 'accessor_name': 'denied@example.com', - 'inherited': False, - 'denied_access': True, - }, - { - 'accessor_name': 'inherited-denied@example.com', - 'inherited': True, - 'denied_access': True, - }, - ] - - show_inherited = True - show_denied = True - - filtered = [ - access for access in accesses - if (show_inherited or not access.get('inherited')) - and (show_denied or not access.get('denied_access')) - ] - - self.assertEqual( - [ - 'direct@example.com', - 'inherited@example.com', - 'denied@example.com', - 'inherited-denied@example.com', - ], - [access['accessor_name'] for access in filtered], - ) - - def test_record_access_filter_hides_inherited_denied_by_default(self): - accesses = [ - { - 'accessor_name': 'direct@example.com', - 'inherited': False, - 'denied_access': False, - }, - { - 'accessor_name': 'inherited@example.com', - 'inherited': True, - 'denied_access': False, - }, - { - 'accessor_name': 'denied@example.com', - 'inherited': False, - 'denied_access': True, - }, - { - 'accessor_name': 'inherited-denied@example.com', - 'inherited': True, - 'denied_access': True, - }, - ] - - filtered = [ - access for access in accesses - if not access.get('inherited') and not access.get('denied_access') - ] - - self.assertEqual( - ['direct@example.com'], - [access['accessor_name'] for access in filtered], - ) - - def test_permission_state_values_are_preserved(self): - access = { - 'accessor_name': 'user@example.com', - 'inherited': True, - 'denied_access': False, - } - - self.assertTrue(access['inherited']) - self.assertFalse(access['denied_access']) - access = { - 'accessor_name': 'user@example.com', - 'inherited': False, - 'denied_access': True, - } - - self.assertFalse(access['inherited']) - self.assertTrue(access['denied_access']) +class TestNsfAccessListingFilters(unittest.TestCase): + """Exercise the real SDK filters (not a copy of their logic).""" + + _ROWS = ( + ('direct@example.com', False, False), + ('inherited@example.com', True, False), + ('denied@example.com', False, True), + ('inherited-denied@example.com', True, True), + ) + + def setUp(self): + self.record_uid = utils.generate_uid() + self.folder_uid = utils.generate_uid() + self.vault = MagicMock() + for target in ('resolve_nsf_record_uid', 'resolve_nsf_folder_uid'): + p = patch('keepersdk.vault.nsf_management.' + target, side_effect=lambda _v, u: u) + p.start() + self.addCleanup(p.stop) + p = patch('keepersdk.vault.nsf_management._resolve_uid_to_username', + side_effect=lambda _v, uid: f'user-{uid}') + self.resolve_username = p.start() + self.addCleanup(p.stop) + + def _record_response(self): + rs = record_details_pb2.RecordAccessResponse() + for name, inherited, denied in self._ROWS: + ra = rs.recordAccesses.add() + ra.accessorInfo.name = name + ra.data.recordUid = utils.base64_url_decode(self.record_uid) + ra.data.accessTypeUid = utils.base64_url_decode(utils.generate_uid()) + ra.data.accessType = folder_pb2.AT_USER + ra.data.inherited = inherited + ra.data.deniedAccess = denied + return rs + + def _folder_response(self): + rs = folder_access_pb2.GetFolderAccessResponse() + fr = rs.folderAccessResults.add() + fr.folderUid = utils.base64_url_decode(self.folder_uid) + for _name, inherited, denied in self._ROWS: + a = fr.accessors.add() + a.folderUid = fr.folderUid + a.accessTypeUid = utils.base64_url_decode(utils.generate_uid()) + a.accessType = folder_pb2.AT_USER + a.accessRoleType = folder_pb2.VIEWER + a.inherited = inherited + a.deniedAccess = denied + return rs + + def _records(self, **kwargs): + self.vault.keeper_auth.execute_auth_rest.return_value = self._record_response() + rs = nsf_management.get_nsf_record_accesses(self.vault, [self.record_uid], **kwargs) + return [a['accessor_name'] for a in rs['record_accesses']] + + def _folders(self, **kwargs): + self.vault.keeper_auth.execute_auth_rest.return_value = self._folder_response() + rs = nsf_management.get_nsf_folder_access(self.vault, [self.folder_uid], **kwargs) + accessors = rs['results'][0]['accessors'] + self.assertTrue(all(accessors), 'filtered rows must be dropped, not left as {}') + return [(a['inherited'], a['denied_access']) for a in accessors] + + def test_record_default_hides_inherited_and_denied(self): + self.assertEqual(['direct@example.com'], self._records()) + + def test_record_show_all(self): + self.assertEqual([r[0] for r in self._ROWS], + self._records(show_inherited=True, show_denied=True)) + + def test_record_hide_inherited_and_denied(self): + self.assertEqual(['direct@example.com'], + self._records(show_inherited=False, show_denied=False)) + + def test_record_hide_inherited_only(self): + self.assertEqual(['direct@example.com', 'denied@example.com'], + self._records(show_inherited=False, show_denied=True)) + + def test_record_hide_denied_only(self): + self.assertEqual(['direct@example.com', 'inherited@example.com'], + self._records(show_inherited=True, show_denied=False)) + + def test_record_rows_keep_state_flags(self): + self.vault.keeper_auth.execute_auth_rest.return_value = self._record_response() + rows = nsf_management.get_nsf_record_accesses( + self.vault, [self.record_uid], show_inherited=True, show_denied=True)['record_accesses'] + self.assertEqual([(r['inherited'], r['denied_access']) for r in rows], + [(i, d) for _n, i, d in self._ROWS]) + + def test_folder_default_hides_inherited_and_denied(self): + self.assertEqual([(False, False)], self._folders()) + + def test_folder_show_all(self): + self.assertEqual([(i, d) for _n, i, d in self._ROWS], + self._folders(show_inherited=True, show_denied=True)) + + def test_folder_hide_inherited_and_denied(self): + self.assertEqual([(False, False)], self._folders(show_inherited=False, show_denied=False)) + + def test_folder_hide_inherited_only(self): + self.assertEqual([(False, False), (False, True)], + self._folders(show_inherited=False, show_denied=True)) + + def test_folder_hide_denied_only(self): + self.assertEqual([(False, False), (True, False)], + self._folders(show_inherited=True, show_denied=False)) + + def test_folder_filtered_rows_skip_username_lookup(self): + self._folders(show_inherited=False, show_denied=False) + self.assertEqual(1, self.resolve_username.call_count) + + +class TestNsfLoadAccessDetails(unittest.TestCase): + + def test_load_always_requests_inherited_and_denied(self): + view = MagicMock() + view.folders.return_value = [MagicMock(folder_uid='f1')] + view.records.return_value = [MagicMock(record_uid='r1')] + vault = MagicMock() + with patch('keepersdk.vault.nsf_management._nsf_view', return_value=view), \ + patch('keepersdk.vault.nsf_management.get_nsf_folder_access', + return_value={'results': [{'folder_uid': 'f1', 'success': True, 'accessors': []}]}) as gfa, \ + patch('keepersdk.vault.nsf_management.get_nsf_record_accesses', + return_value={'record_accesses': []}) as gra: + loaded = nsf_management.load_nsf_access_details(vault, load_folder=True, load_record=True) + self.assertEqual(loaded, {'folders': 1, 'records': 1}) + self.assertEqual(gfa.call_args.kwargs, {'show_inherited': True, 'show_denied': True}) + self.assertEqual(gra.call_args.kwargs, {'show_inherited': True, 'show_denied': True}) class TestEncryptForTeam(unittest.TestCase):