(self, params, **kwargs)
| 2905 | return f'Can {" & ".join([p for p in privs if p])}' |
| 2906 | |
| 2907 | def execute(self, params, **kwargs): |
| 2908 | export_format = kwargs['format'] if 'format' in kwargs else None |
| 2909 | export_name = kwargs.get('output') |
| 2910 | |
| 2911 | all_records = kwargs.get('all_records') is True |
| 2912 | records = {} # type: Dict[str, vault.KeeperRecord] |
| 2913 | containers = kwargs.get('folder') |
| 2914 | filter_folders = None |
| 2915 | versions = (0, 1, 2, 3, 5, 6) if all_records else (2, 3) |
| 2916 | if isinstance(containers, list) and len(containers) > 0: |
| 2917 | filter_folders = set() |
| 2918 | log_folder_fn = lambda f_name: logging.info(f'Folder {f_name} could not be found.') |
| 2919 | def on_folder_fn(f): # type: (BaseFolderNode) -> None |
| 2920 | filter_folders.add(f.uid) |
| 2921 | for record_uid in params.subfolder_record_cache.get(f.uid or '', []): |
| 2922 | if record_uid not in records: |
| 2923 | record = vault.KeeperRecord.load(params, record_uid) |
| 2924 | if not record: |
| 2925 | continue |
| 2926 | if not record.shared: |
| 2927 | continue |
| 2928 | if not all_records: |
| 2929 | ro = params.record_owner_cache.get(record.record_uid) |
| 2930 | if not ro or not ro.owner: |
| 2931 | continue |
| 2932 | if record.version not in versions: |
| 2933 | continue |
| 2934 | records[record.record_uid] = record |
| 2935 | for name in containers: |
| 2936 | folder_uids = get_folder_uids(params, name) |
| 2937 | if not folder_uids: |
| 2938 | log_folder_fn(name) |
| 2939 | continue |
| 2940 | for uid in folder_uids: |
| 2941 | FolderMixin.traverse_folder_tree(params, uid, on_folder_fn) |
| 2942 | else: |
| 2943 | for record in vault_extensions.find_records(params, record_version=versions): |
| 2944 | if not record.shared: |
| 2945 | continue |
| 2946 | if not all_records: |
| 2947 | ro = params.record_owner_cache.get(record.record_uid) |
| 2948 | if not ro or not ro.owner: |
| 2949 | continue |
| 2950 | records[record.record_uid] = record |
| 2951 | |
| 2952 | expand_teams = kwargs.get('show_team_users') is True |
| 2953 | team_membership = None # type: Optional[Dict[str, List[str]]] |
| 2954 | if expand_teams: |
| 2955 | team_membership = {} |
| 2956 | if params.enterprise is not None: |
| 2957 | user_lookup = {x['enterprise_user_id']: x['username'] for x in params.enterprise['users'] if x.get('status') == 'active'} |
| 2958 | if 'team_users' in params.enterprise: |
| 2959 | for tu in params.enterprise['team_users']: |
| 2960 | team_uid = tu['team_uid'] |
| 2961 | enterprise_user_id = tu['enterprise_user_id'] |
| 2962 | if enterprise_user_id in user_lookup: |
| 2963 | members = team_membership.get(team_uid) |
| 2964 | if members is None: |
nothing calls this directly
no test coverage detected