-
Notifications
You must be signed in to change notification settings - Fork 198
/
azkms.py
179 lines (148 loc) · 5.65 KB
/
azkms.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
"azkms secret module"
import os
import logging
import base64
from azure.keyvault.keys.crypto import CryptographyClient, EncryptionAlgorithm
from azure.keyvault.keys import KeyClient
from azure.identity import DefaultAzureCredential
from kapitan.refs.base64 import Base64Ref, Base64RefBackend
from kapitan.refs.base import RefError
from kapitan import cached
from kapitan.errors import KapitanError
logger = logging.getLogger(__name__)
class AzureKMSError(KapitanError):
"""
Generic Azure Key Vault error
"""
pass
def azkms_obj(key_id):
"""
Return Azure Key Vault Object
"""
# e.g of key_id https://kapitanbackend.vault.azure.net/keys/myKey/deadbeef
if not cached.azkms_obj:
attrs = key_id.split("/")
if key_id.startswith("http"):
key_vault_uri = attrs[2]
key_name = attrs[4]
key_version = attrs[5]
else:
key_vault_uri = attrs[0]
key_name = attrs[2]
key_version = attrs[3]
# If --verbose is set, show requests from azure
if logger.getEffectiveLevel() > logging.DEBUG:
logging.getLogger("azure").setLevel(logging.ERROR)
credential = DefaultAzureCredential()
key_client = KeyClient(vault_url=f"https://{key_vault_uri}", credential=credential)
key = key_client.get_key(key_name, key_version)
cached.azkms_obj = CryptographyClient(key, credential)
return cached.azkms_obj
class AzureKMSSecret(Base64Ref):
def __init__(self, data, key, encrypt=True, encode_base64=False, **kwargs):
"""
encrypts data with key
set encode_base64 to True to base64 encode data before encrypting and writing
set encrypt to False if loading data that is already encrypted and base64
"""
if encrypt:
self._encrypt(data, key, encode_base64)
if encode_base64:
kwargs["encoding"] = "base64"
else:
self.data = data
self.key = key
super().__init__(self.data, **kwargs)
self.type_name = "azkms"
@classmethod
def from_params(cls, data, ref_params):
"""
Return new AzureKMSSecret from data and ref_params: target_name
key will be grabbed from the inventory via target_name
"""
try:
target_name = ref_params.kwargs["target_name"]
if target_name is None:
raise ValueError("target_name not set")
target_inv = cached.inv["nodes"].get(target_name, None)
if target_inv is None:
raise ValueError("target_inv not set")
key = target_inv["parameters"]["kapitan"]["secrets"]["azkms"]["key"]
return cls(data, key, **ref_params.kwargs)
except KeyError:
raise RefError("Could not create AzureKMSSecret: target_name missing")
@classmethod
def from_path(cls, ref_full_path, **kwargs):
return super().from_path(ref_full_path, encrypt=False, **kwargs)
def reveal(self):
"""
returns decrypted data
raises AzureKMSError if decryption fails
"""
# can't use super().reveal() as we want bytes
ref_data = base64.b64decode(self.data)
return self._decrypt(ref_data, self.key)
def update_key(self, key):
"""
re-encrypts data with new key, respects original encoding
returns True if key is different and secret is updated, False otherwise
"""
if key == self.key:
return False
data_dec = self.reveal()
encode_base64 = self.encoding == "base64"
if encode_base64:
data_dec = base64.b64decode(data_dec).decode()
self._encrypt(data_dec, key, encode_base64)
self.data = base64.b64encode(self.data).decode()
return True
def _encrypt(self, data, key, encode_base64):
"""
encrypts data
set encode_base64 to True to base64 encode data before writing
"""
assert isinstance(key, str)
_data = data
self.encoding = "original"
if encode_base64:
_data = base64.b64encode(data.encode())
self.encoding = "base64"
else:
# To guarantee _data is bytes
if isinstance(data, str):
_data = data.encode()
try:
ciphertext = ""
# Mocking encrypted response for tests
if key == "mock":
ciphertext = base64.b64encode("mock".encode())
else:
request = azkms_obj(key).encrypt(EncryptionAlgorithm.rsa_oaep, _data)
ciphertext = request.ciphertext
self.data = ciphertext
self.key = key
except Exception as e:
raise AzureKMSError(e)
def _decrypt(self, data, key):
"""decrypt data"""
try:
plaintext = ""
# Mocking decrypted response for tests
if self.key == "mock":
plaintext = "mock".encode()
else:
request = azkms_obj(key).decrypt(EncryptionAlgorithm.rsa_oaep, data)
plaintext = request.plaintext
return plaintext.decode()
except Exception as e:
raise AzureKMSError(e)
def dump(self):
"""
Returns dict with keys/values to be serialised.
"""
return {"data": self.data, "encoding": self.encoding, "key": self.key, "type": self.type_name}
class AzureKMSBackend(Base64RefBackend):
def __init__(self, path, ref_type=AzureKMSSecret, **ref_kwargs):
"init AzureKMSBackend ref backend type"
super().__init__(path, ref_type, **ref_kwargs)
self.type_name = "azkms"