| 435 | return share_folder_parser |
| 436 | |
| 437 | def execute(self, params, **kwargs): |
| 438 | from ..enforcement import MasterPasswordReentryEnforcer |
| 439 | if not MasterPasswordReentryEnforcer.check_and_enforce(params, "record_level"): |
| 440 | raise CommandError('share-folder', 'Operation cancelled: Re-authentication failed') |
| 441 | |
| 442 | def get_share_admin_obj_uids(obj_names, obj_type): |
| 443 | # type: (List[Optional[str], int], int) -> Optional[Set[str]] |
| 444 | if not obj_names: |
| 445 | return None |
| 446 | try: |
| 447 | rq = record_pb2.AmIShareAdmin() |
| 448 | for name in obj_names: |
| 449 | try: |
| 450 | uid = utils.base64_url_decode(name) |
| 451 | if isinstance(uid, bytes) and len(uid) == 16: |
| 452 | osa = record_pb2.IsObjectShareAdmin() |
| 453 | osa.uid = uid |
| 454 | osa.objectType = obj_type |
| 455 | rq.isObjectShareAdmin.append(osa) |
| 456 | except: |
| 457 | pass |
| 458 | if len(rq.isObjectShareAdmin) > 0: |
| 459 | rs = api.communicate_rest(params, rq, 'vault/am_i_share_admin', rs_type=record_pb2.AmIShareAdmin) |
| 460 | sa_obj_uids = {sa_obj.uid for sa_obj in rs.isObjectShareAdmin if sa_obj.isAdmin} |
| 461 | sa_obj_uids = {utils.base64_url_encode(uid) for uid in sa_obj_uids} |
| 462 | return sa_obj_uids |
| 463 | except Error as e: |
| 464 | print(f'get_share_admin: msg = {e.message}') |
| 465 | return None |
| 466 | |
| 467 | names = kwargs.get('folder') |
| 468 | if not isinstance(names, list): |
| 469 | names = [names] |
| 470 | |
| 471 | all_folders = any(True for x in names if x == '*') |
| 472 | if all_folders: |
| 473 | names = [x for x in names if x != '*'] |
| 474 | |
| 475 | shared_folder_uids = set() # type: Set[str] |
| 476 | if all_folders: |
| 477 | shared_folder_uids.update(params.shared_folder_cache.keys()) |
| 478 | else: |
| 479 | get_folder_by_uid = lambda uid: params.folder_cache.get(uid) |
| 480 | folder_uids = {uid for name in names for uid in get_folder_uids(params, name)} |
| 481 | folders = {get_folder_by_uid(uid) for uid in folder_uids} |
| 482 | shared_folder_uids.update([uid for uid in folder_uids if uid in params.shared_folder_cache]) |
| 483 | sf_subfolders = {f for f in folders if f.type == 'shared_folder_folder'} |
| 484 | shared_folder_uids.update({subfolder.shared_folder_uid for subfolder in sf_subfolders}) |
| 485 | unresolved_names = [name for name in names if not get_folder_uids(params, name)] |
| 486 | share_admin_folder_uids = get_share_admin_obj_uids(unresolved_names, record_pb2.CHECK_SA_ON_SF) |
| 487 | shared_folder_uids.update(share_admin_folder_uids or []) |
| 488 | |
| 489 | if not shared_folder_uids: |
| 490 | raise CommandError('share-folder', 'Enter name of at least one existing folder') |
| 491 | |
| 492 | action = kwargs.get('action') or 'grant' |
| 493 | |
| 494 | share_expiration = None |