| 261 | return share_folder_parser |
| 262 | |
| 263 | def execute(self, params, **kwargs): |
| 264 | from ..enforcement import MasterPasswordReentryEnforcer |
| 265 | if not MasterPasswordReentryEnforcer.check_and_enforce(params, "record_level"): |
| 266 | raise CommandError('share-folder', 'Operation cancelled: Re-authentication failed') |
| 267 | |
| 268 | def get_share_admin_obj_uids(obj_names, obj_type): |
| 269 | # type: (List[Optional[str], int], int) -> Optional[Set[str]] |
| 270 | if not obj_names: |
| 271 | return None |
| 272 | try: |
| 273 | rq = record_pb2.AmIShareAdmin() |
| 274 | for name in obj_names: |
| 275 | try: |
| 276 | uid = utils.base64_url_decode(name) |
| 277 | if isinstance(uid, bytes) and len(uid) == 16: |
| 278 | osa = record_pb2.IsObjectShareAdmin() |
| 279 | osa.uid = uid |
| 280 | osa.objectType = obj_type |
| 281 | rq.isObjectShareAdmin.append(osa) |
| 282 | except: |
| 283 | pass |
| 284 | if len(rq.isObjectShareAdmin) > 0: |
| 285 | rs = api.communicate_rest(params, rq, 'vault/am_i_share_admin', rs_type=record_pb2.AmIShareAdmin) |
| 286 | sa_obj_uids = {sa_obj.uid for sa_obj in rs.isObjectShareAdmin if sa_obj.isAdmin} |
| 287 | sa_obj_uids = {utils.base64_url_encode(uid) for uid in sa_obj_uids} |
| 288 | return sa_obj_uids |
| 289 | except Error as e: |
| 290 | print(f'get_share_admin: msg = {e.message}') |
| 291 | return None |
| 292 | |
| 293 | names = kwargs.get('folder') |
| 294 | if not isinstance(names, list): |
| 295 | names = [names] |
| 296 | |
| 297 | all_folders = any(True for x in names if x == '*') |
| 298 | if all_folders: |
| 299 | names = [x for x in names if x != '*'] |
| 300 | |
| 301 | shared_folder_uids = set() # type: Set[str] |
| 302 | if all_folders: |
| 303 | shared_folder_uids.update(params.shared_folder_cache.keys()) |
| 304 | else: |
| 305 | get_folder_by_uid = lambda uid: params.folder_cache.get(uid) |
| 306 | folder_uids = {uid for name in names for uid in get_folder_uids(params, name)} |
| 307 | folders = {get_folder_by_uid(uid) for uid in folder_uids} |
| 308 | shared_folder_uids.update([uid for uid in folder_uids if uid in params.shared_folder_cache]) |
| 309 | sf_subfolders = {f for f in folders if f.type == 'shared_folder_folder'} |
| 310 | shared_folder_uids.update({subfolder.shared_folder_uid for subfolder in sf_subfolders}) |
| 311 | unresolved_names = [name for name in names if not get_folder_uids(params, name)] |
| 312 | share_admin_folder_uids = get_share_admin_obj_uids(unresolved_names, record_pb2.CHECK_SA_ON_SF) |
| 313 | shared_folder_uids.update(share_admin_folder_uids or []) |
| 314 | |
| 315 | if not shared_folder_uids: |
| 316 | raise CommandError('share-folder', 'Enter name of at least one existing folder') |
| 317 | |
| 318 | action = kwargs.get('action') or 'grant' |
| 319 | |
| 320 | share_expiration = None |