Return the cached ``PublicKeys`` for a team, loading them if needed.
(params, team_uid_b64: str)
| 174 | |
| 175 | |
| 176 | def get_team_keys(params, team_uid_b64: str): |
| 177 | """Return the cached ``PublicKeys`` for a team, loading them if needed. |
| 178 | """ |
| 179 | from ..params import PublicKeys |
| 180 | |
| 181 | cached = params.key_cache.get(team_uid_b64) |
| 182 | has_asym = bool(cached and (getattr(cached, 'rsa', None) or getattr(cached, 'ec', None))) |
| 183 | if cached and has_asym: |
| 184 | return cached |
| 185 | |
| 186 | api.load_team_keys(params, [team_uid_b64]) |
| 187 | keys = params.key_cache.get(team_uid_b64) |
| 188 | |
| 189 | has_asym = bool(keys and (getattr(keys, 'rsa', None) or getattr(keys, 'ec', None))) |
| 190 | if not has_asym: |
| 191 | try: |
| 192 | rq = {'command': 'team_get_keys', 'teams': [team_uid_b64]} |
| 193 | rs = api.communicate(params, rq) |
| 194 | merged = { |
| 195 | 'aes': getattr(keys, 'aes', b'') or b'' if keys else b'', |
| 196 | 'rsa': getattr(keys, 'rsa', b'') or b'' if keys else b'', |
| 197 | 'ec': getattr(keys, 'ec', b'') or b'' if keys else b'', |
| 198 | } |
| 199 | for tk in (rs or {}).get('keys', []): |
| 200 | if tk.get('team_uid') != team_uid_b64: |
| 201 | continue |
| 202 | # Symmetric/wrapped key in 'key' field |
| 203 | if 'key' in tk: |
| 204 | try: |
| 205 | key_type = tk.get('type') |
| 206 | encrypted_key = utils.base64_url_decode(tk['key']) |
| 207 | if key_type == 1: |
| 208 | merged['aes'] = crypto.decrypt_aes_v1(encrypted_key, params.data_key) |
| 209 | elif key_type == 2: |
| 210 | merged['aes'] = crypto.decrypt_rsa(encrypted_key, params.rsa_key2) |
| 211 | elif key_type == 3: |
| 212 | merged['aes'] = crypto.decrypt_aes_v2(encrypted_key, params.data_key) |
| 213 | elif key_type == 4: |
| 214 | merged['aes'] = crypto.decrypt_ec(encrypted_key, params.ecc_key) |
| 215 | elif key_type == -1: |
| 216 | merged['ec'] = encrypted_key |
| 217 | elif key_type == -3: |
| 218 | merged['rsa'] = encrypted_key |
| 219 | except Exception as e: |
| 220 | logger.debug('team_get_keys key decode failed: %s', e) |
| 221 | # Raw public key in 'team_public_key' field (separate from 'key') |
| 222 | if 'team_public_key' in tk: |
| 223 | try: |
| 224 | pub_key_bytes = utils.base64_url_decode(tk['team_public_key']) |
| 225 | pub_key_type = tk.get('team_public_key_type') |
| 226 | if pub_key_type == -3: |
| 227 | merged['rsa'] = pub_key_bytes |
| 228 | elif pub_key_type == -1: |
| 229 | merged['ec'] = pub_key_bytes |
| 230 | except Exception as e: |
| 231 | logger.debug('team_get_keys public key decode failed: %s', e) |
| 232 | if any(merged.values()): |
| 233 | params.key_cache[team_uid_b64] = PublicKeys( |
no test coverage detected