(params, team_uids)
| 362 | |
| 363 | |
| 364 | def load_team_keys(params, team_uids): # type: (KeeperParams, List[str]) -> None |
| 365 | s = set(team_uids) |
| 366 | s.difference_update(params.key_cache.keys()) |
| 367 | if not s: |
| 368 | return |
| 369 | |
| 370 | if params.enterprise: |
| 371 | logging.debug('Resolve team keys shared with enterprise') |
| 372 | tree_key = params.enterprise['unencrypted_tree_key'] |
| 373 | for t in params.enterprise.get('teams', []): |
| 374 | team_uid = t.get('team_uid') |
| 375 | if team_uid in s: |
| 376 | try: |
| 377 | encrypted_team_key = t.get('encrypted_team_key') |
| 378 | if encrypted_team_key: |
| 379 | team_key = crypto.decrypt_aes_v2(utils.base64_url_decode(encrypted_team_key), tree_key) |
| 380 | params.key_cache[team_uid] = PublicKeys(aes=team_key) |
| 381 | s.remove(team_uid) |
| 382 | except Exception as e: |
| 383 | logging.debug('Team UID \"%s\": Decrypt key error: %s', team_uid, str(e)) |
| 384 | |
| 385 | if not s: |
| 386 | return |
| 387 | |
| 388 | logging.debug('Loading %d team keys', len(s)) |
| 389 | uids_to_load = list(s) |
| 390 | |
| 391 | while len(uids_to_load) > 0: |
| 392 | uids = uids_to_load[:90] |
| 393 | uids_to_load = uids_to_load[90:] |
| 394 | rq = { |
| 395 | 'command': 'team_get_keys', |
| 396 | 'teams': uids |
| 397 | } |
| 398 | rs = communicate(params, rq) |
| 399 | if 'keys' in rs: |
| 400 | for tk in rs['keys']: |
| 401 | if 'key' in tk: |
| 402 | team_uid = tk['team_uid'] |
| 403 | try: |
| 404 | aes = b'' |
| 405 | rsa = b'' |
| 406 | ec = b'' |
| 407 | encrypted_key = utils.base64_url_decode(tk['key']) |
| 408 | key_type = tk['type'] |
| 409 | if key_type == 1: |
| 410 | aes = crypto.decrypt_aes_v1(encrypted_key, params.data_key) |
| 411 | elif key_type == 2: |
| 412 | aes = crypto.decrypt_rsa(encrypted_key, params.rsa_key2) |
| 413 | elif key_type == 3: |
| 414 | aes = crypto.decrypt_aes_v2(encrypted_key, params.data_key) |
| 415 | elif key_type == 4: |
| 416 | aes = crypto.decrypt_ec(encrypted_key, params.ecc_key) |
| 417 | elif key_type == -1: |
| 418 | ec = encrypted_key |
| 419 | elif key_type == -3: |
| 420 | rsa = encrypted_key |
| 421 | params.key_cache[team_uid] = PublicKeys(rsa=rsa, aes=aes, ec=ec) |
nothing calls this directly
no test coverage detected