-
Notifications
You must be signed in to change notification settings - Fork 0
/
red_october.py
387 lines (335 loc) · 13.2 KB
/
red_october.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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
# -*- coding: utf-8 -*-
# Copyright 2016 LasLabs Inc.
# License MIT (https://opensource.org/licenses/MIT).
import requests
from datetime import timedelta
from .models.enum_user_role import EnumUserRole
from .models.enum_user_type import EnumUserType
from .exceptions import RedOctoberDecryptException
from .exceptions import RedOctoberRemoteException
class RedOctober(object):
""" It provides Python bindings to a remote RedOctober server via HTTP(S).
Additional documentation regarding the API endpoints is available at
https://github.com/cloudflare/redoctober
"""
def __init__(self, host, port, name, password, ssl=True, verify=True):
""" It initializes the RedOctober API using the provided credentials.
Args:
host (str): Host name/IP of Red October server.
port (int): Port number of server.
name (str): Account name for use as login.
password (str): Password for account.
ssl (bool): Is server SSL encrypted?
verify (bool or str): File path of CA cert for verification,
`True` to use system certs, or `False` to disable certificate
verification.
"""
ssl = 'https' if ssl else 'http'
self.uri_base = '%s://%s:%d' % (ssl, host, port)
self.name = name
self.password = password
self.verify = verify
def create_vault(self):
""" It creates a new vault.
Create is the necessary first call to a new vault.
It creates an admin account.
Returns:
bool: Status of vault creation
"""
return self.call('create')
def delegate(self, time=None, uses=None):
""" It allows for the delegation of decryption rights.
Delegate allows a user to delegate their decryption password to the
server for a fixed period of time and for a fixed number of
decryptions. If the user's account is not created, it creates it.
Any new delegation overrides the previous delegation.
Args:
time (datetime.timedelta): Period of time that delegation is valid
for.
uses (int): Number of times that delegation can be used.
Returns:
bool: Status of delegation creation.
"""
data = self._clean_mapping({
'Time': time and '%ds' % time.total_seconds() or None,
'Uses': uses,
})
return self.call('delegate', data=data)
def create_user(self, user_type='rsa'):
""" It creates a new user account.
Allows an optional ``UserType`` to be specified which controls how the
record is encrypted. This can have a value of either ``rsa`` or ``ecc``
and if none is provided will default to ``rsa``.
Args:
user_type (str): Controls how the record is encrypted. This can have
a value of either ``ecc`` or ``rsa``.
Returns:
bool: Status of user creation.
"""
data = self._clean_mapping({
'UserType': EnumUserType[user_type].name.upper(),
})
return self.call('create-user', data=data)
def get_summary(self):
""" It provides a list of keys and delegations for the server.
Returns:
dict: A mapping containing keys on the system, and users who have
currently delegated their key to the server. Example:
.. code-block:: python
{
"Live":{
"Bill":{"Admin":false,
"Type":"rsa",
"Expiry":"2013-11-26T08:42:29.65501032-08:00",
"Uses":3},
"Cat":{"Admin":false,
"Type":"rsa",
"Expiry":"2013-11-26T08:42:42.016311595-08:00",
"Uses":3},
"Dodo":{"Admin":false,
"Type":"rsa",
"Expiry":"2013-11-26T08:43:06.651429104-08:00",
"Uses":3}
},
"All":{
"Alice":{"Admin":true, "Type":"rsa"},
"Bill":{"Admin":false, "Type":"rsa"},
"Cat":{"Admin":false, "Type":"rsa"},
"Dodo":{"Admin":false, "Type":"rsa"}
}
"""
return self.call('summary')
def encrypt(self, minimum, owners, data):
""" It allows a user to encrypt a piece of data.
Args:
minimum (int): Minimum number of users from ``owners`` set that
must have delegated their keys to the server.
owners (iter): Iterator of strings indicating users that may
decrypt the document.
data (str): Data to encrypt.
Returns:
str: Base64 encoded string representing the encrypted string.
"""
data = self._clean_mapping({
'Minimum': minimum,
'Owners': owners,
'Data': data.encode('base64'),
})
return self.call('encrypt', data=data)
def decrypt(self, data):
""" It allows a user to decrypt a piece of data.
Args:
data (str): Base64 encoded string representing the encrypted
string.
Raises:
RedOctoberDecryptException: If not enough ``minimum`` users from
the set of ``owners`` have delegated their keys to the server,
or if the decryption credentials are incorrect.
Returns:
str: Decrypted string
"""
data = self._clean_mapping({
'Data': data,
})
try:
return self.call('decrypt', data=data)
except RedOctoberRemoteException as e:
raise RedOctoberDecryptException(e.message)
def get_owners(self, data):
""" It provides the delegates required to decrypt a piece of data.
Args:
data (str): Base64 encoded string representing the encrypted
string.
Raises:
RedOctoberDecryptException: If incorrect decryption credentials
are provided.
Returns:
list: List of strings representing users that are able to decrypt
the data.
"""
data = self._clean_mapping({
'Data': data,
})
try:
return self.call('owners', data=data)
except RedOctoberRemoteException as e:
raise RedOctoberDecryptException(e.message)
def change_password(self, new_password):
""" It allows users to change their password.
Args:
name (str): Name of account.
password (str): Password for account.
new_password (str): New password for account.
Returns:
bool: Password change status.
"""
data = self._clean_mapping({
'NewPassword': new_password,
})
return self.call('password', data=data)
def modify_user_role(self, modify_name, command='revoke'):
""" It allows for administration of user roles.
Args:
modify_name (str): Name of account to modify.
command (str): Command to apply to user:
``admin``: Promote user to administrator.
``revoke``: Revoke administrator rights.
``delete``: Delete user.
Returns:
bool: Role modfication status.
"""
data = self._clean_mapping({
'ToModify': modify_name,
'Command': EnumUserRole[command].name,
})
return self.call('modify', data=data)
def purge_delegates(self):
""" It deletes all delegates for an encryption key.
Returns:
bool: Purge status.
"""
return self.call('purge')
def create_order(self, labels, duration, uses, data):
""" It creates lets others users know delegations are needed.
Args:
labels (iter): Iterator of strings to label order with.
duration (datetime.timedelta): Proposed duration of delegation.
uses (int): Proposed delegation use amounts.
data (str): Base64 encoded string representing the encrypted
string.
Returns:
dict: Mapping representing the newly created order. Example:
.. code-block:: python
{
"Admins": [
"Bob",
"Eve"
],
"AdminsDelegated": null,
"Delegated": 0,
"DurationRequested": 3.6e+12,
"Labels": [
"blue",
"red"
],
"Name": "Alice",
"Num": "77da1cfd8962fb9685c15c84",
"TimeRequested": "2016-01-25T15:58:41.961906679-08:00",
}
"""
data = self._clean_mapping({
'Labels': labels,
'Duration': '%ds' % duration.total_seconds(),
'Uses': uses,
'Data': data,
})
return self.call('order', data=data)
def get_orders_outstanding(self):
""" It returns a mapping of current orders.
Returns:
dict: Mapping representing the currently open orders. Example:
.. code-block:: python
{
"77da1cfd8962fb9685c15c84":{
"Name":"Alice",
"Num":"77da1cfd8962fb9685c15c84",
"TimeRequested":"2016-01-25T15:58:41.961906679-08:00",
"DurationRequested":3600000000000,
"Delegated":0,"
AdminsDelegated":null,
"Admins":["Bob, Eve"],
"Labels":["Blue","Red"]
}
}
"""
return self.call('orderout')
def get_order_information(self, order_num):
""" It gets information for a specified order.
Args:
order_num (str): Order number to get.
Returns:
dict: Mapping representing the order information. Example:
.. code-block:: python
{
"Admins": [
"Bob",
"Eve"
],
"AdminsDelegated": null,
"Delegated": 0,
"DurationRequested": 3.6e+12,
"Labels": [
"blue",
"red"
],
"Name": "Alice",
"Num": "77da1cfd8962fb9685c15c84",
"TimeRequested": "2016-01-25T15:58:41.961906679-08:00"
}
"""
data = self._clean_mapping({
'OrderNum': order_num,
})
return self.call('orderinfo', data=data)
def cancel_order(self, order_num):
""" It cancels an order by number.
Args:
order_num (str): Order number to get.
Returns:
bool: Status of order cancellation.
"""
data = self._clean_mapping({
'OrderNum': order_num,
})
return self.call('ordercancel', data=data)
def call(self, endpoint, method='POST', params=None, data=None):
""" It calls the remote endpoint and returns the result, if success.
Args:
endpoint (str): RedOctober endpoint to call (e.g. ``newcert``).
method (str): HTTP method to utilize for the Request.
params: (dict or bytes) Data to be sent in the query string
for the Request.
data: (dict or bytes or file) Data to send in the body of the
Request.
Raises:
RedOctoberRemoteException: In the event of a ``False`` in the
``success`` key of the API response.
Returns:
mixed: Data contained in ``result`` key of the API response, or
``True`` if there was no response data, but the call was a
success.
"""
endpoint = '%s/%s' % (self.uri_base, endpoint)
if data is None:
data = {}
data.update({
'Name': self.name,
'Password': self.password,
})
response = requests.request(
method=method,
url=endpoint,
params=params,
json=data,
verify=self.verify,
)
response = response.json()
if response['Status'] != 'ok':
raise RedOctoberRemoteException(
'\n'.join([
'Response:',
'\n'.join(response.get('Response', [])),
])
)
try:
return response['Response']
except KeyError:
return True
def _clean_mapping(self, mapping):
""" It removes false entries from mapping.
Args:
mapping (dict): Mapping to remove values from.
Returns:
dict: Mapping with no values evaluating to False.
"""
return {k:v for k, v in mapping.iteritems() if v}