(record, new_password)
| 27 | |
| 28 | |
| 29 | def rotate(record, new_password): # type: (vault.KeeperRecord, str) -> bool |
| 30 | private_key = '' |
| 31 | if isinstance(record, vault.PasswordRecord): |
| 32 | cf = next((x for x in record.custom if x.name == 'cmdr:private_key'), None) |
| 33 | if cf: |
| 34 | private_key = cf.value |
| 35 | elif isinstance(record, vault.TypedRecord): |
| 36 | ssh_key_field = record.get_typed_field('keyPair') |
| 37 | if ssh_key_field: |
| 38 | key_value = ssh_key_field.get_default_value(dict) |
| 39 | if isinstance(key_value, dict): |
| 40 | private_key = key_value.get('privateKey', '') |
| 41 | else: |
| 42 | cf = next((x for x in record.custom if x.label == 'cmdr:private_key'), None) |
| 43 | if cf: |
| 44 | value = cf.get_default_value(str) |
| 45 | if value: |
| 46 | private_key = value |
| 47 | |
| 48 | old_password = RecordMixin.get_record_field(record, 'password') |
| 49 | |
| 50 | old_pk = None |
| 51 | backend = default_backend() |
| 52 | ssh_key_format = serialization.PrivateFormat.TraditionalOpenSSL |
| 53 | if private_key: |
| 54 | header, _, _ = private_key.partition('\n') |
| 55 | if 'BEGIN OPENSSH PRIVATE KEY' in header: |
| 56 | ssh_key_format = serialization.PrivateFormat.OpenSSH |
| 57 | try: |
| 58 | old_pk = load_private_key(private_key, old_password) |
| 59 | if isinstance(old_pk, rsa.RSAPrivateKey): |
| 60 | pub_key_exponent = old_pk.public_key().public_numbers().e |
| 61 | new_pk = rsa.generate_private_key( |
| 62 | key_size=old_pk.key_size, public_exponent=pub_key_exponent, backend=backend) |
| 63 | elif isinstance(old_pk, ec.EllipticCurvePrivateKey): |
| 64 | new_pk = ec.generate_private_key(old_pk.curve, backend=backend) |
| 65 | elif isinstance(old_pk, ed25519.Ed25519PrivateKey): |
| 66 | new_pk = ed25519.Ed25519PrivateKey.generate() |
| 67 | ssh_key_format = serialization.PrivateFormat.OpenSSH |
| 68 | else: |
| 69 | raise Exception('Unsupported private key type') |
| 70 | except Exception as e: |
| 71 | logging.info('SSH Key rotation plugin: load key error: %s', e) |
| 72 | return False |
| 73 | else: |
| 74 | new_pk, _ = crypto.generate_rsa_key() |
| 75 | |
| 76 | new_private_key = new_pk.private_bytes( |
| 77 | encoding=serialization.Encoding.PEM, format=ssh_key_format, |
| 78 | encryption_algorithm=serialization.BestAvailableEncryption(new_password.encode('utf-8'))).decode() |
| 79 | |
| 80 | new_public_ssh_key = new_pk.public_key().public_bytes( |
| 81 | encoding=serialization.Encoding.OpenSSH, format=serialization.PublicFormat.OpenSSH).decode() |
| 82 | |
| 83 | optional_port = RecordMixin.get_custom_field(record, 'cmdr:port') |
| 84 | if optional_port: |
| 85 | try: |
| 86 | port = int(optional_port) |
nothing calls this directly
no test coverage detected