(params, records)
| 128 | return score_changed_on_passkey or creds_removed or needs_alignment |
| 129 | |
| 130 | def update_security_audit_data(params, records): # type: (KeeperParams, List[KeeperRecord]) -> int |
| 131 | if not params.enterprise_ec_key: |
| 132 | return 0 |
| 133 | |
| 134 | from . import api |
| 135 | update_limit = 1000 |
| 136 | total_updates = len(records) |
| 137 | failed_updates = [] |
| 138 | while records: |
| 139 | chunk = records[:update_limit] |
| 140 | records = records[update_limit:] |
| 141 | rq = APIRequest_pb2.SecurityDataRequest() |
| 142 | rq.encryptionType = get_security_data_key_type(params) |
| 143 | try: |
| 144 | sec_data_objs = (prep_security_data_update(params, rec) for rec in chunk) |
| 145 | score_data_objs = (prep_score_data_update(params, rec) for rec in chunk) |
| 146 | rq.recordSecurityData.extend(sd for sd in sec_data_objs if sd) |
| 147 | rq.recordSecurityScoreData.extend(sd for sd in score_data_objs if sd) |
| 148 | rs = api.communicate_rest(params, rq, 'enterprise/update_security_data') |
| 149 | except: |
| 150 | failed_updates.extend(chunk) |
| 151 | |
| 152 | if failed_updates: |
| 153 | logging.error(f'Could not update security data for {len(failed_updates)} records') |
| 154 | |
| 155 | return total_updates - len(failed_updates) |
| 156 | |
| 157 | def attach_security_data(params, record, rq_param): |
| 158 | # type: (KeeperParams, Union[str, Dict[str, any], KeeperRecord], Union[record_pb2.RecordUpdate, record_pb2.RecordAdd]) -> Union[record_pb2.RecordUpdate, record_pb2.RecordAdd] |
no test coverage detected