(self, changes)
| 138 | return len(self._records) |
| 139 | |
| 140 | def rebuild_data(self, changes): # type: (RebuildTask) -> None |
| 141 | def decrypt(data): # type: (bytes or str) -> str |
| 142 | if isinstance(data, str): |
| 143 | return data |
| 144 | |
| 145 | decrypted = '' |
| 146 | try: |
| 147 | decrypted_bytes = crypto.decrypt_aes_v1(data, self.tree_key) if data else b'' |
| 148 | decrypted = decrypted_bytes.decode() |
| 149 | except Error as e: |
| 150 | logging.info(f'Error decrypting data: type = {type(data)}, value = {data}, message = {e.message}') |
| 151 | return decrypted |
| 152 | |
| 153 | def link_record_permissions(store, record_lookup): |
| 154 | links = store.get_record_permissions().get_all_links() |
| 155 | for link in links: |
| 156 | record = record_lookup.get(link.record_uid) |
| 157 | if record: |
| 158 | record.user_permissions.update({link.user_uid: link.permissions}) |
| 159 | else: |
| 160 | logging.info(f'record (uid = {link.record_uid}) not found') |
| 161 | |
| 162 | return record_lookup |
| 163 | |
| 164 | def load_records(store, changes): |
| 165 | # type: (sqlite_storage.SqliteSoxStorage, RebuildTask) -> Dict[str, sox_types.Record] |
| 166 | entities = [] # type: List[storage_types.StorageRecord] |
| 167 | if changes.records: |
| 168 | for uid in changes.records: |
| 169 | entity = store.records.get_entity(uid) |
| 170 | if entity: |
| 171 | entities.append(entity) |
| 172 | else: |
| 173 | entities.extend(store.records.get_all()) |
| 174 | |
| 175 | record_lookup = {} |
| 176 | for entity in entities: |
| 177 | record = self._records.get(entity.record_uid) or sox_types.Record() |
| 178 | record.update_properties(entity, self.ec_private_key) |
| 179 | record_lookup[record.record_uid] = record |
| 180 | |
| 181 | record_lookup = link_record_aging(store, record_lookup) |
| 182 | return link_record_permissions(store, record_lookup) if changes.load_compliance_data else record_lookup |
| 183 | |
| 184 | def link_record_aging(store, record_lookup): |
| 185 | for aging in store.record_aging.get_all(): |
| 186 | record = record_lookup.get(aging.record_uid) |
| 187 | if record: |
| 188 | record.created = aging.created |
| 189 | record.last_pw_change = aging.last_pw_change |
| 190 | record.last_modified = aging.last_modified |
| 191 | record.last_rotation = aging.last_rotation |
| 192 | return record_lookup |
| 193 | |
| 194 | def link_user_records(store, user_lookup): |
| 195 | links = store.get_user_record_links().get_all_links() |
| 196 | for link in links: |
| 197 | user = user_lookup.get(link.user_uid) |
no test coverage detected