(params, record, folder_uid=None, pb_only:bool=False)
| 24 | |
| 25 | |
| 26 | def add_record_to_folder(params, record, folder_uid=None, pb_only:bool=False): |
| 27 | # type: (KeeperParams, vault.KeeperRecord, Optional[str], Optional[bool]) -> Optional[Union[record_pb2.RecordUpdate, record_pb2.RecordAdd]] |
| 28 | if not record.record_uid: |
| 29 | record.record_uid = utils.generate_uid() |
| 30 | if not record.record_key: |
| 31 | record.record_key = utils.generate_aes_key() |
| 32 | |
| 33 | folder = params.folder_cache.get(folder_uid) # type: Union[None, subfolder.BaseFolderNode] |
| 34 | folder_key = None # type: Optional[bytes] |
| 35 | if isinstance(folder, subfolder.SharedFolderFolderNode): |
| 36 | shared_folder_uid = folder.shared_folder_uid |
| 37 | elif isinstance(folder, subfolder.SharedFolderNode): |
| 38 | shared_folder_uid = folder.uid |
| 39 | else: |
| 40 | shared_folder_uid = None |
| 41 | if shared_folder_uid and shared_folder_uid in params.shared_folder_cache: |
| 42 | shared_folder = params.shared_folder_cache.get(shared_folder_uid) |
| 43 | folder_key = shared_folder.get('shared_folder_key_unencrypted') |
| 44 | |
| 45 | if isinstance(record, vault.PasswordRecord): |
| 46 | rq = { |
| 47 | 'command': 'record_add', |
| 48 | 'record_uid': record.record_uid, |
| 49 | 'record_key': utils.base64_url_encode( |
| 50 | crypto.encrypt_aes_v1(record.record_key, params.data_key)), |
| 51 | 'record_type': 'password', |
| 52 | 'folder_type': folder.type if folder else 'user_folder', |
| 53 | 'how_long_ago': 0, |
| 54 | } |
| 55 | if folder: |
| 56 | rq['folder_uid'] = folder.uid |
| 57 | if folder_key: |
| 58 | rq['folder_key'] = utils.base64_url_encode(crypto.encrypt_aes_v1(record.record_key, folder_key)) |
| 59 | data = vault_extensions.extract_password_record_data(record) |
| 60 | rq['data'] = utils.base64_url_encode(crypto.encrypt_aes_v1(json.dumps(data).encode(), record.record_key)) |
| 61 | extra = vault_extensions.extract_password_record_extras(record) |
| 62 | rq['extra'] = utils.base64_url_encode(crypto.encrypt_aes_v1(json.dumps(extra).encode(), record.record_key)) |
| 63 | file_ids = [] |
| 64 | if record.attachments: |
| 65 | for atta in record.attachments: |
| 66 | file_ids.append(atta.id) |
| 67 | if atta.thumbnails: |
| 68 | for thumb in atta.thumbnails: |
| 69 | file_ids.append(thumb.id) |
| 70 | rq['file_ids'] = file_ids |
| 71 | |
| 72 | rs = api.communicate(params, rq) |
| 73 | record.revision = rs.get('revision', 0) |
| 74 | add_record_audit_data(params, record) |
| 75 | if record.attachments: |
| 76 | for atta in record.attachments: |
| 77 | params.queue_audit_event( |
| 78 | 'file_attachment_uploaded', record_uid=record.record_uid, attachment_id=atta.id) |
| 79 | |
| 80 | elif isinstance(record, vault.TypedRecord): |
| 81 | add_record = record_pb2.RecordAdd() |
| 82 | add_record.record_uid = utils.base64_url_decode(record.record_uid) |
| 83 | add_record.record_key = crypto.encrypt_aes_v2(record.record_key, params.data_key) |
no test coverage detected