Push a record update to the cloud. Takes a Record() object, converts to record JSON and pushes to the Keeper cloud API
(params, record, **kwargs)
| 1171 | |
| 1172 | |
| 1173 | def add_record_v3(params, record, **kwargs): # type: (KeeperParams, dict, ...) -> Optional[bool] |
| 1174 | """ Push a record update to the cloud. |
| 1175 | Takes a Record() object, converts to record JSON |
| 1176 | and pushes to the Keeper cloud API |
| 1177 | """ |
| 1178 | |
| 1179 | record_rq = record |
| 1180 | if record_rq is None: |
| 1181 | return |
| 1182 | |
| 1183 | record_links = [] |
| 1184 | links = kwargs.get('record_links') or {} |
| 1185 | links = links.get('record_links') or [] |
| 1186 | |
| 1187 | for link in links: |
| 1188 | rl = record_pb2.RecordLink() |
| 1189 | rl.record_uid = link.get('record_uid') |
| 1190 | rl.record_key = link.get('record_key') |
| 1191 | if rl.record_uid and rl.record_key: |
| 1192 | record_links.append(rl) |
| 1193 | |
| 1194 | uid = loginv3.CommonHelperMethods.url_safe_str_to_bytes(record_rq['record_uid']) |
| 1195 | |
| 1196 | unencrypted_key = record_rq['record_key_unencrypted'] |
| 1197 | key = crypto.encrypt_aes_v2(unencrypted_key, params.data_key) |
| 1198 | #key = base64.urlsafe_b64encode(key) |
| 1199 | |
| 1200 | data = record_rq['data_unencrypted'].decode('utf-8') if isinstance(record_rq['data_unencrypted'], bytes) else record_rq['data_unencrypted'] |
| 1201 | try: |
| 1202 | d = json.loads(data) |
| 1203 | rt_name = d.get('type') or '' |
| 1204 | rt_def = RecordV3.resolve_record_type_by_name(params, rt_name) |
| 1205 | res = RecordV3.is_valid_record_type(data, rt_def) |
| 1206 | if not res.get('is_valid'): |
| 1207 | logging.info(res.get('error')) |
| 1208 | except Exception as e: |
| 1209 | logging.error(bcolors.FAIL + 'Invalid record type! Error: ' + str(e) + bcolors.ENDC) |
| 1210 | return None |
| 1211 | |
| 1212 | # Audit - if the title or url has changed and the user is an enterprise user |
| 1213 | audit = None |
| 1214 | if params.enterprise_ec_key: |
| 1215 | fields = itertools.chain(d.get('fields') or [], (d.get('custom') or [])) |
| 1216 | url_field = next((u.get('value') for u in fields if u.get('type') == 'url' and u.get('value')), None) |
| 1217 | url = '' |
| 1218 | if url_field and isinstance(url_field, list): |
| 1219 | if len(url_field) > 0: |
| 1220 | url = utils.url_strip(str(url_field[0])) |
| 1221 | |
| 1222 | adata = { |
| 1223 | 'title': d.get('title', ''), |
| 1224 | 'record_type': rt_name, |
| 1225 | } |
| 1226 | if url: adata['url'] = url # url will only be supplied if there is one on the record |
| 1227 | audit = record_pb2.RecordAudit() |
| 1228 | audit.version = 0 |
| 1229 | audit.data = crypto.encrypt_ec(json.dumps(adata).encode('utf-8'), params.enterprise_ec_key) |
| 1230 |
nothing calls this directly
no test coverage detected