(params, record_uids, is_share_admin=False)
| 1410 | |
| 1411 | |
| 1412 | def get_record_shares(params, record_uids, is_share_admin=False): |
| 1413 | # type: (KeeperParams, Iterable[str], bool) -> Optional[List[Dict[str, Any]]] |
| 1414 | def need_share_info(uid): |
| 1415 | if uid in params.record_cache: |
| 1416 | r = params.record_cache[uid] |
| 1417 | return 'shares' not in r |
| 1418 | return is_share_admin |
| 1419 | |
| 1420 | result = [] |
| 1421 | unique = set(record_uids) |
| 1422 | uids = [x for x in unique if need_share_info(x)] |
| 1423 | try: |
| 1424 | while len(uids) > 0: |
| 1425 | chunk = uids[:999] |
| 1426 | uids = uids[999:] |
| 1427 | rq = record_pb2.GetRecordDataWithAccessInfoRequest() |
| 1428 | rq.clientTime = utils.current_milli_time() |
| 1429 | rq.recordUid.extend([utils.base64_url_decode(x) for x in chunk]) |
| 1430 | rq.recordDetailsInclude = record_pb2.SHARE_ONLY |
| 1431 | rs = communicate_rest(params, rq, 'vault/get_records_details', |
| 1432 | rs_type=record_pb2.GetRecordDataWithAccessInfoResponse) |
| 1433 | for info in rs.recordDataWithAccessInfo: |
| 1434 | record_uid = utils.base64_url_encode(info.recordUid) |
| 1435 | rec = params.record_cache[record_uid] if record_uid in params.record_cache else {'record_uid': record_uid} # type: dict |
| 1436 | if 'shares' not in rec: |
| 1437 | rec['shares'] = {} |
| 1438 | rec['shares']['user_permissions'] = [] |
| 1439 | rec['shares']['shared_folder_permissions'] = [] |
| 1440 | for up in info.userPermission: |
| 1441 | oup = { |
| 1442 | 'username': up.username, |
| 1443 | 'owner': up.owner, |
| 1444 | 'share_admin': up.shareAdmin, |
| 1445 | 'shareable': up.sharable, |
| 1446 | 'editable': up.editable, |
| 1447 | } |
| 1448 | if up.awaitingApproval: |
| 1449 | oup['awaiting_approval'] = up.awaitingApproval |
| 1450 | if up.expiration > 0: |
| 1451 | oup['expiration'] = up.expiration |
| 1452 | rec['shares']['user_permissions'].append(oup) |
| 1453 | for sp in info.sharedFolderPermission: |
| 1454 | osp = { |
| 1455 | 'shared_folder_uid': utils.base64_url_encode(sp.sharedFolderUid), |
| 1456 | 'reshareable': sp.resharable, |
| 1457 | 'editable': sp.editable, |
| 1458 | 'revision': sp.revision, |
| 1459 | } |
| 1460 | if sp.expiration > 0: |
| 1461 | osp['expiration'] = sp.expiration |
| 1462 | rec['shares']['shared_folder_permissions'].append(osp) |
| 1463 | |
| 1464 | if record_uid not in params.record_cache: |
| 1465 | result.append(rec) |
| 1466 | except Exception as e: |
| 1467 | logging.error(e) |
| 1468 | |
| 1469 | if len(result) > 0: |
no test coverage detected