(self, params, **kwargs)
| 2524 | return trash_unshare_parser |
| 2525 | |
| 2526 | def execute(self, params, **kwargs): |
| 2527 | orphaned_records = self.get_orphaned_records(params) |
| 2528 | if len(orphaned_records) == 0: |
| 2529 | logging.info('Trash is empty') |
| 2530 | return |
| 2531 | |
| 2532 | records = kwargs.get('records') |
| 2533 | if not isinstance(records, (tuple, list)): |
| 2534 | records = None |
| 2535 | if not records: |
| 2536 | logging.info('records parameter is empty.') |
| 2537 | return |
| 2538 | |
| 2539 | to_restore = set() |
| 2540 | for rec in records: |
| 2541 | if rec in orphaned_records: |
| 2542 | to_restore.add(rec) |
| 2543 | else: |
| 2544 | title_pattern = re.compile(fnmatch.translate(rec), re.IGNORECASE) |
| 2545 | for record_uid, del_rec in orphaned_records.items(): |
| 2546 | if record_uid in to_restore: |
| 2547 | continue |
| 2548 | record = vault.KeeperRecord.load(params, del_rec) |
| 2549 | if title_pattern.match(record.title): |
| 2550 | to_restore.add(record_uid) |
| 2551 | |
| 2552 | if len(to_restore) == 0: |
| 2553 | logging.info('There are no records to unshare') |
| 2554 | return |
| 2555 | |
| 2556 | if not kwargs.get('force'): |
| 2557 | answer = base.user_choice(f'Do you want to remove shares from {len(to_restore)} record(s)?', 'yn', default='n') |
| 2558 | if answer.lower() == 'y': |
| 2559 | answer = 'yes' |
| 2560 | if answer.lower() != 'yes': |
| 2561 | return |
| 2562 | |
| 2563 | record_shares = api.get_record_shares(params, to_restore, True) |
| 2564 | if record_shares: |
| 2565 | remove_shares = [] # type: List[record_pb2.SharedRecord] |
| 2566 | for record_share in record_shares: |
| 2567 | if 'shares' in record_share: |
| 2568 | shares = record_share['shares'] |
| 2569 | if 'user_permissions' in shares: |
| 2570 | for user_permission in shares['user_permissions']: |
| 2571 | if user_permission.get('owner') is False: |
| 2572 | rs_rq = record_pb2.SharedRecord() |
| 2573 | rs_rq.toUsername = user_permission['username'] |
| 2574 | rs_rq.recordUid = utils.base64_url_decode(record_share['record_uid']) |
| 2575 | remove_shares.append(rs_rq) |
| 2576 | |
| 2577 | while len(remove_shares) > 0: |
| 2578 | chunk = remove_shares[:900] |
| 2579 | remove_shares = remove_shares[900:] |
| 2580 | rsu_rq = record_pb2.RecordShareUpdateRequest() |
| 2581 | rsu_rq.removeSharedRecord.extend(chunk) |
| 2582 | |
| 2583 | rsu_rs = api.communicate_rest(params, rsu_rq, 'vault/records_share_update', |
nothing calls this directly
no test coverage detected