| 376 | return share_folder_parser |
| 377 | |
| 378 | def execute(self, params, **kwargs): |
| 379 | from ..enforcement import MasterPasswordReentryEnforcer |
| 380 | if not MasterPasswordReentryEnforcer.check_and_enforce(params, "record_level"): |
| 381 | raise CommandError('share-folder', 'Operation cancelled: Re-authentication failed') |
| 382 | |
| 383 | def get_share_admin_obj_uids(obj_names, obj_type): |
| 384 | # type: (List[Optional[str], int], int) -> Optional[Set[str]] |
| 385 | if not obj_names: |
| 386 | return None |
| 387 | try: |
| 388 | rq = record_pb2.AmIShareAdmin() |
| 389 | for name in obj_names: |
| 390 | try: |
| 391 | uid = utils.base64_url_decode(name) |
| 392 | if isinstance(uid, bytes) and len(uid) == 16: |
| 393 | osa = record_pb2.IsObjectShareAdmin() |
| 394 | osa.uid = uid |
| 395 | osa.objectType = obj_type |
| 396 | rq.isObjectShareAdmin.append(osa) |
| 397 | except: |
| 398 | pass |
| 399 | if len(rq.isObjectShareAdmin) > 0: |
| 400 | rs = api.communicate_rest(params, rq, 'vault/am_i_share_admin', rs_type=record_pb2.AmIShareAdmin) |
| 401 | sa_obj_uids = {sa_obj.uid for sa_obj in rs.isObjectShareAdmin if sa_obj.isAdmin} |
| 402 | sa_obj_uids = {utils.base64_url_encode(uid) for uid in sa_obj_uids} |
| 403 | return sa_obj_uids |
| 404 | except Error as e: |
| 405 | print(f'get_share_admin: msg = {e.message}') |
| 406 | return None |
| 407 | |
| 408 | names = kwargs.get('folder') |
| 409 | if not isinstance(names, list): |
| 410 | names = [names] |
| 411 | |
| 412 | all_folders = any(True for x in names if x == '*') |
| 413 | if all_folders: |
| 414 | names = [x for x in names if x != '*'] |
| 415 | |
| 416 | shared_folder_uids = set() # type: Set[str] |
| 417 | if all_folders: |
| 418 | shared_folder_uids.update(params.shared_folder_cache.keys()) |
| 419 | else: |
| 420 | get_folder_by_uid = lambda uid: params.folder_cache.get(uid) |
| 421 | folder_uids = {uid for name in names for uid in get_folder_uids(params, name)} |
| 422 | folders = {get_folder_by_uid(uid) for uid in folder_uids} |
| 423 | shared_folder_uids.update([uid for uid in folder_uids if uid in params.shared_folder_cache]) |
| 424 | sf_subfolders = {f for f in folders if f.type == 'shared_folder_folder'} |
| 425 | shared_folder_uids.update({subfolder.shared_folder_uid for subfolder in sf_subfolders}) |
| 426 | unresolved_names = [name for name in names if not get_folder_uids(params, name)] |
| 427 | share_admin_folder_uids = get_share_admin_obj_uids(unresolved_names, record_pb2.CHECK_SA_ON_SF) |
| 428 | shared_folder_uids.update(share_admin_folder_uids or []) |
| 429 | |
| 430 | if not shared_folder_uids: |
| 431 | raise CommandError('share-folder', 'Enter name of at least one existing folder') |
| 432 | |
| 433 | action = kwargs.get('action') or 'grant' |
| 434 | |
| 435 | share_expiration = None |