Return the cached ``PublicKeys`` for a team, loading them if needed.
(params, team_uid_b64: str)
| 196 | |
| 197 | |
| 198 | def get_team_keys(params, team_uid_b64: str): |
| 199 | """Return the cached ``PublicKeys`` for a team, loading them if needed. |
| 200 | """ |
| 201 | from ..params import PublicKeys |
| 202 | |
| 203 | if not is_keeper_uid(team_uid_b64): |
| 204 | raise ValueError(f'Invalid team UID: {team_uid_b64}') |
| 205 | |
| 206 | cached = params.key_cache.get(team_uid_b64) |
| 207 | has_asym = bool(cached and (getattr(cached, 'rsa', None) or getattr(cached, 'ec', None))) |
| 208 | if cached and has_asym: |
| 209 | return cached |
| 210 | |
| 211 | api.load_team_keys(params, [team_uid_b64]) |
| 212 | keys = params.key_cache.get(team_uid_b64) |
| 213 | |
| 214 | has_asym = bool(keys and (getattr(keys, 'rsa', None) or getattr(keys, 'ec', None))) |
| 215 | if not has_asym: |
| 216 | try: |
| 217 | rq = {'command': 'team_get_keys', 'teams': [team_uid_b64]} |
| 218 | rs = api.communicate(params, rq) |
| 219 | merged = { |
| 220 | 'aes': getattr(keys, 'aes', b'') or b'' if keys else b'', |
| 221 | 'rsa': getattr(keys, 'rsa', b'') or b'' if keys else b'', |
| 222 | 'ec': getattr(keys, 'ec', b'') or b'' if keys else b'', |
| 223 | } |
| 224 | for tk in (rs or {}).get('keys', []): |
| 225 | if tk.get('team_uid') != team_uid_b64: |
| 226 | continue |
| 227 | # Symmetric/wrapped key in 'key' field |
| 228 | if 'key' in tk: |
| 229 | try: |
| 230 | key_type = tk.get('type') |
| 231 | encrypted_key = utils.base64_url_decode(tk['key']) |
| 232 | if key_type == 1: |
| 233 | merged['aes'] = crypto.decrypt_aes_v1(encrypted_key, params.data_key) |
| 234 | elif key_type == 2: |
| 235 | merged['aes'] = crypto.decrypt_rsa(encrypted_key, params.rsa_key2) |
| 236 | elif key_type == 3: |
| 237 | merged['aes'] = crypto.decrypt_aes_v2(encrypted_key, params.data_key) |
| 238 | elif key_type == 4: |
| 239 | merged['aes'] = crypto.decrypt_ec(encrypted_key, params.ecc_key) |
| 240 | elif key_type == -1: |
| 241 | merged['ec'] = encrypted_key |
| 242 | elif key_type == -3: |
| 243 | merged['rsa'] = encrypted_key |
| 244 | except Exception as e: |
| 245 | logger.debug('team_get_keys key decode failed: %s', e) |
| 246 | # Raw public key in 'team_public_key' field (separate from 'key') |
| 247 | if 'team_public_key' in tk: |
| 248 | try: |
| 249 | pub_key_bytes = utils.base64_url_decode(tk['team_public_key']) |
| 250 | pub_key_type = tk.get('team_public_key_type') |
| 251 | if pub_key_type == -3: |
| 252 | merged['rsa'] = pub_key_bytes |
| 253 | elif pub_key_type == -1: |
| 254 | merged['ec'] = pub_key_bytes |
| 255 | except Exception as e: |