-
Notifications
You must be signed in to change notification settings - Fork 1
/
jumpcloud.py
50 lines (39 loc) · 1.57 KB
/
jumpcloud.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
import requests, json
from getTokens import get_secret
import logging
JUMPCLOUD_USERGROUP_URI = 'https://console.jumpcloud.com/api/v2/usergroups/<groupnumber>/members'
logger = logging.getLogger('cassandra-broker')
def get_data_from_jumpcloud():
jumpcloud_user_ids_collection = []
jumpcloud_user_emails_collection = []
jumpcloud_creds = get_secret("jc_credentials")
result = json.loads(jc_credentials)
jumpcloudApi = result['jumpcloudApi']
logger.info('Querying secrets - success')
logger.info('Invoke jumpcloud user group API')
s = requests.session()
s.headers = {'x-api-key': jumpcloudApi}
try:
r = s.get(JUMPCLOUD_USERGROUP_URI)
except Exception:
logger.exception('Error while invoking jumpcloud API - usergroups endpoint')
else:
if r.status_code != 200:
e = r.json()
logger.error(e['error'])
else:
json_response = json.loads(r.text)
for user in json_response:
userid = user['to']['id']
jumpcloud_user_ids_collection.append(userid)
try:
for user in jumpcloud_user_ids_collection:
r = s.get(f'https://console.jumpcloud.com/api/systemusers/{user}')
if r.status_code != 200:
e = r.json()
logger.error(e['error'])
else:
jumpcloud_user_emails_collection.append(json.loads(r.text)['email'])
return jumpcloud_user_emails_collection
except Exception:
logger.exception('Error while invoking jumpcloud API - systemusers endpoint')