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)
| 1146 | |
| 1147 | |
| 1148 | def add_record_v3(params, record, **kwargs): # type: (KeeperParams, dict, ...) -> Optional[bool] |
| 1149 | """ Push a record update to the cloud. |
| 1150 | Takes a Record() object, converts to record JSON |
| 1151 | and pushes to the Keeper cloud API |
| 1152 | """ |
| 1153 | |
| 1154 | record_rq = record |
| 1155 | if record_rq is None: |
| 1156 | return |
| 1157 | |
| 1158 | record_links = [] |
| 1159 | links = kwargs.get('record_links') or {} |
| 1160 | links = links.get('record_links') or [] |
| 1161 | |
| 1162 | for link in links: |
| 1163 | rl = record_pb2.RecordLink() |
| 1164 | rl.record_uid = link.get('record_uid') |
| 1165 | rl.record_key = link.get('record_key') |
| 1166 | if rl.record_uid and rl.record_key: |
| 1167 | record_links.append(rl) |
| 1168 | |
| 1169 | uid = loginv3.CommonHelperMethods.url_safe_str_to_bytes(record_rq['record_uid']) |
| 1170 | |
| 1171 | unencrypted_key = record_rq['record_key_unencrypted'] |
| 1172 | key = crypto.encrypt_aes_v2(unencrypted_key, params.data_key) |
| 1173 | #key = base64.urlsafe_b64encode(key) |
| 1174 | |
| 1175 | data = record_rq['data_unencrypted'].decode('utf-8') if isinstance(record_rq['data_unencrypted'], bytes) else record_rq['data_unencrypted'] |
| 1176 | try: |
| 1177 | d = json.loads(data) |
| 1178 | rt_name = d.get('type') or '' |
| 1179 | rt_def = RecordV3.resolve_record_type_by_name(params, rt_name) |
| 1180 | res = RecordV3.is_valid_record_type(data, rt_def) |
| 1181 | if not res.get('is_valid'): |
| 1182 | logging.info(res.get('error')) |
| 1183 | except Exception as e: |
| 1184 | logging.error(bcolors.FAIL + 'Invalid record type! Error: ' + str(e) + bcolors.ENDC) |
| 1185 | return None |
| 1186 | |
| 1187 | # Audit - if the title or url has changed and the user is an enterprise user |
| 1188 | audit = None |
| 1189 | if params.enterprise_ec_key: |
| 1190 | fields = itertools.chain(d.get('fields') or [], (d.get('custom') or [])) |
| 1191 | url_field = next((u.get('value') for u in fields if u.get('type') == 'url' and u.get('value')), None) |
| 1192 | url = '' |
| 1193 | if url_field and isinstance(url_field, list): |
| 1194 | if len(url_field) > 0: |
| 1195 | url = utils.url_strip(str(url_field[0])) |
| 1196 | |
| 1197 | adata = { |
| 1198 | 'title': d.get('title', ''), |
| 1199 | 'record_type': rt_name, |
| 1200 | } |
| 1201 | if url: adata['url'] = url # url will only be supplied if there is one on the record |
| 1202 | audit = record_pb2.RecordAudit() |
| 1203 | audit.version = 0 |
| 1204 | audit.data = crypto.encrypt_ec(json.dumps(adata).encode('utf-8'), params.enterprise_ec_key) |
| 1205 |
nothing calls this directly
no test coverage detected