(params, record_uids, cache_only=False)
| 9 | |
| 10 | |
| 11 | def get_shared_records(params, record_uids, cache_only=False): |
| 12 | def fetch_team_members(t_uids): |
| 13 | members = {} |
| 14 | if params.enterprise_ec_key: |
| 15 | for team_uid in t_uids: |
| 16 | team_users = members.get(team_uid, set()) |
| 17 | rq = enterprise_pb2.GetTeamMemberRequest() |
| 18 | rq.teamUid = utils.base64_url_decode(team_uid) |
| 19 | endpoint = 'vault/get_team_members' |
| 20 | rs = api.communicate_rest(params, rq, endpoint, rs_type=enterprise_pb2.GetTeamMemberResponse) |
| 21 | if rs.enterpriseUser: |
| 22 | team_users.update({x.email for x in rs.enterpriseUser}) |
| 23 | members[team_uid] = team_users |
| 24 | return members |
| 25 | |
| 26 | def get_cached_team_members(t_uids, uname_lookup): |
| 27 | members = {} |
| 28 | if not params.enterprise: |
| 29 | return members |
| 30 | |
| 31 | team_users = params.enterprise.get('team_users') or [] |
| 32 | team_users = [tu for tu in team_users if tu.get('user_type') != 2 and tu.get('team_uid') in t_uids] |
| 33 | |
| 34 | for tu in team_users: |
| 35 | user_id = tu.get('enterprise_user_id') |
| 36 | username = uname_lookup.get(user_id) |
| 37 | team_uid = tu.get('team_uid') |
| 38 | t_members = members.get(team_uid, set()) |
| 39 | t_members.add(username) |
| 40 | members[team_uid] = t_members |
| 41 | |
| 42 | return members |
| 43 | |
| 44 | def fetch_sf_admins(): |
| 45 | sf_uids = [uid for uid in params.shared_folder_cache] |
| 46 | return {sf_uid: api.get_share_admins_for_shared_folder(params, sf_uid) for sf_uid in sf_uids} |
| 47 | |
| 48 | def get_restricted_role_members(uname_lookup): |
| 49 | # Get team_uids and usernames (assigned directly and indirectly) in share-restricted roles |
| 50 | members = set() |
| 51 | if not params.enterprise: |
| 52 | return members |
| 53 | |
| 54 | restrict_key = 'restrict_sharing_all' |
| 55 | enf_key = 'enforcements' |
| 56 | r_enforcements = params.enterprise.get('role_enforcements', []) |
| 57 | no_share_roles = {re.get('role_id') for re in r_enforcements if re.get(enf_key).get(restrict_key) == 'true'} |
| 58 | r_users = [u for u in params.enterprise.get('role_users', []) if u.get('role_id') in no_share_roles] |
| 59 | r_teams = [t for t in params.enterprise.get('role_teams', []) if t.get('role_id') in no_share_roles] |
| 60 | no_share_users = {uname_lookup.get(u.get('enterprise_user_id')) for u in r_users} |
| 61 | no_share_teams = {t.get('team_uid') for t in r_teams} |
| 62 | cached_team_members = get_cached_team_members(no_share_teams, uname_lookup) |
| 63 | no_share_team_members = {t for team_uid in no_share_teams for t in cached_team_members.get(team_uid) or []} |
| 64 | members = no_share_users.union(no_share_teams).union(no_share_team_members) |
| 65 | return members |
| 66 | |
| 67 | api.get_record_shares(params, record_uids) |
| 68 | sf_teams = [shared_folder.get('teams', []) for shared_folder in params.shared_folder_cache.values()] |
no test coverage detected