(params, record_uids, is_share_admin=False)
| 1435 | |
| 1436 | |
| 1437 | def get_record_shares(params, record_uids, is_share_admin=False): |
| 1438 | # type: (KeeperParams, Iterable[str], bool) -> Optional[List[Dict[str, Any]]] |
| 1439 | def need_share_info(uid): |
| 1440 | if uid in params.record_cache: |
| 1441 | r = params.record_cache[uid] |
| 1442 | return 'shares' not in r |
| 1443 | return is_share_admin |
| 1444 | |
| 1445 | result = [] |
| 1446 | unique = set(record_uids) |
| 1447 | uids = [x for x in unique if need_share_info(x)] |
| 1448 | try: |
| 1449 | while len(uids) > 0: |
| 1450 | chunk = uids[:999] |
| 1451 | uids = uids[999:] |
| 1452 | rq = record_pb2.GetRecordDataWithAccessInfoRequest() |
| 1453 | rq.clientTime = utils.current_milli_time() |
| 1454 | rq.recordUid.extend([utils.base64_url_decode(x) for x in chunk]) |
| 1455 | rq.recordDetailsInclude = record_pb2.SHARE_ONLY |
| 1456 | rs = communicate_rest(params, rq, 'vault/get_records_details', |
| 1457 | rs_type=record_pb2.GetRecordDataWithAccessInfoResponse) |
| 1458 | for info in rs.recordDataWithAccessInfo: |
| 1459 | record_uid = utils.base64_url_encode(info.recordUid) |
| 1460 | rec = params.record_cache[record_uid] if record_uid in params.record_cache else {'record_uid': record_uid} # type: dict |
| 1461 | if 'shares' not in rec: |
| 1462 | rec['shares'] = {} |
| 1463 | rec['shares']['user_permissions'] = [] |
| 1464 | rec['shares']['shared_folder_permissions'] = [] |
| 1465 | for up in info.userPermission: |
| 1466 | oup = { |
| 1467 | 'username': up.username, |
| 1468 | 'owner': up.owner, |
| 1469 | 'share_admin': up.shareAdmin, |
| 1470 | 'shareable': up.sharable, |
| 1471 | 'editable': up.editable, |
| 1472 | } |
| 1473 | if up.awaitingApproval: |
| 1474 | oup['awaiting_approval'] = up.awaitingApproval |
| 1475 | if up.expiration > 0: |
| 1476 | oup['expiration'] = up.expiration |
| 1477 | rec['shares']['user_permissions'].append(oup) |
| 1478 | for sp in info.sharedFolderPermission: |
| 1479 | osp = { |
| 1480 | 'shared_folder_uid': utils.base64_url_encode(sp.sharedFolderUid), |
| 1481 | 'reshareable': sp.resharable, |
| 1482 | 'editable': sp.editable, |
| 1483 | 'revision': sp.revision, |
| 1484 | } |
| 1485 | if sp.expiration > 0: |
| 1486 | osp['expiration'] = sp.expiration |
| 1487 | rec['shares']['shared_folder_permissions'].append(osp) |
| 1488 | |
| 1489 | if record_uid not in params.record_cache: |
| 1490 | result.append(rec) |
| 1491 | except Exception as e: |
| 1492 | logging.error(e) |
| 1493 | |
| 1494 | if len(result) > 0: |
no test coverage detected