-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathparam_store.py
117 lines (93 loc) · 3.57 KB
/
param_store.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
"""
Utilities for interactibg with Parameter Store. Implemented for AWS SSM Parameter Store.
"""
import json
import time
import webexteamssdk
import boto3
def getSmartsheetId():
"""Returns the saved Smartsheet ID from Parameter Store
Args:
None
Returns:
ssSheetId: Smartsheet ID from Parameter Store
"""
# load parameters from parameter store
ssm_client = boto3.client("ssm")
ssmStoredParameter = ssm_client.get_parameter(
Name="/smartsheet-webex/smartsheetSheetId",
WithDecryption=True
)
ssSheetId = ssmStoredParameter['Parameter']['Value']
ssm_client.close()
return ssSheetId
def saveSmartsheetId(sheetId):
"""Saves Smartsheet ID to Parameter Store
Args:
sheetId (str)
Returns:
None
"""
ssm_client = boto3.client("ssm")
ssmStoredParameter = ssm_client.put_parameter(
Name="/smartsheet-webex/smartsheetSheetId",
Value=sheetId,
Type="String",
Overwrite=True
)
return ssmStoredParameter
def getWebexIntegrationToken(webex_integration_client_id, webex_integration_client_secret):
"""Returns a fresh, usable Webex Integration access token.
Webex Integration access tokens are acquired through OAuth and must be refreshed regularly.
OAuth-provided access token and refresh token have limited lifetimes. As of now,
access_token lifetime is 14 days since creation
refresh_token lifetime is 90 days since last use
This function reads tokens from Parameter Store, refreshes the access_token if it's more than halftime old,
and returns the access_token.
Args:
webex_integration_client_id - used if access token refresh is needed
webex_integration_client_secret - used if access token refresh is needed
Returns:
accessToken: fresh, usable Webex Integration access token
"""
# read access tokens from Parameter Store
ssm_client = boto3.client("ssm")
ssmStoredParameter = ssm_client.get_parameter(
Name="/smartsheet-webex/webexTokens",
WithDecryption=True
)
currentTokens = json.loads(ssmStoredParameter['Parameter']['Value'])
accessToken = currentTokens['access_token']
createdTime = currentTokens['created']
lifetime = 14*24*60*60 # 14 days
if createdTime + lifetime/2 < time.time():
# refresh token
refreshToken = currentTokens['refresh_token']
webexApi = webexteamssdk.WebexTeamsAPI(access_token=accessToken) # passing expired access_token should still work, the API object can be initiated with any string
newTokens = webexApi.access_tokens.refresh(
client_id=webex_integration_client_id,
client_secret=webex_integration_client_secret,
refresh_token=refreshToken
)
# save the new access token to the Parameter Store
saveWebexIntegrationTokens(dict(newTokens.json_data))
accessToken = newTokens.access_token
ssm_client.close()
return accessToken
def saveWebexIntegrationTokens(tokens):
"""Saves Webex Integration tokens to Parameter Store. Adds `created` timestamp for token lifetime tracking.
Args:
tokens: dict of Webex Integration tokens data, as it comes from the API call
Returns:
None
"""
tokens['created'] = time.time()
ssm_client = boto3.client("ssm")
ssmStoredParameter = ssm_client.put_parameter(
Name="/smartsheet-webex/webexTokens",
Value=json.dumps(tokens),
Type="SecureString",
Overwrite=True
)
ssm_client.close()
return ssmStoredParameter