-
Notifications
You must be signed in to change notification settings - Fork 16
/
auth.py
416 lines (358 loc) · 15.1 KB
/
auth.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
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
"""
Clients for working with the Forge Authentication service.
"""
from enum import Enum
from datetime import datetime, timedelta
from urllib.parse import quote, quote_plus
from typing import Dict, List
from .base import BaseClient
BASE_URL = "https://developer.api.autodesk.com/authentication/v1"
class Scope(Enum):
"""
Authentication scopes.
"""
USER_PROFILE_READ = "user-profile:read"
"""
The application will be able to read the end user’s profile data
(not including associated products and services).
"""
USER_READ = "user:read"
"""
The application will be able to read the end user’s profile data,
including associated products and services.
"""
USER_WRITE = "user:write"
"""
The application will be able to create, update, and delete the end user’s profile data,
including associated products and services.
"""
VIEWABLES_READ = "viewables:read"
"""
The application will only be able to read the end user’s viewable data
(e.g., PNG and SVF files) within the Autodesk ecosystem.
"""
DATA_READ = "data:read"
"""
The application will be able to read all the end user’s data (viewable and non-viewable)
within the Autodesk ecosystem.
"""
DATA_WRITE = "data:write"
"""
The application will be able to create, update,
and delete data on behalf of the end user within the Autodesk ecosystem.
"""
DATA_CREATE = "data:create"
"""
The application will be able to create data on behalf of the end user
within the Autodesk ecosystem.
"""
DATA_SEARCH = "data:search"
"""
The application will be able to search the end user’s data
within the Autodesk ecosystem.
"""
BUCKET_CREATE = "bucket:create"
"""The application will be able to create an OSS bucket it will own."""
BUCKET_READ = "bucket:read"
"""
The application will be able to read the metadata and list contents
for OSS buckets that it has access to.
"""
BUCKET_UPDATE = "bucket:update"
"""
The application will be able to set permissions and entitlements
for OSS buckets that it has permission to modify.
"""
BUCKET_DELETE = "bucket:delete"
"""The application will be able to delete a bucket that it has permission to delete."""
CODE_ALL = "code:all"
"""
The application will be able to author and execute code on behalf of the end user
(e.g., scripts processed by the Design Automation API).
"""
ACCOUNT_READ = "account:read"
"""
For Product APIs, the application will be able to read the account data
the end user has entitlements to.
"""
ACCOUNT_WRITE = "account:write"
"""
For Product APIs, the application will be able to update the account data
the end user has entitlements to.
"""
def get_authorization_url(
client_id: str, response_type: str, redirect_uri: str,
scopes: List[Scope], state: str = None
) -> str:
"""
Generate a URL to redirect an end user to in order to acquire the user’s consent
for your app to access the specified resources.
**Documentation**: https://forge.autodesk.com/en/docs/oauth/v2/reference/http/authorize-GET
Args:
client_id (str): Client ID of the app.
response_type (str): Must be either "code" for authorization code grant flow or "token"
for implicit grant flow.
redirect_uri (str): URL-encoded callback URL that the end user will be redirected to
after completing the authorization flow.
scopes (List[Scope]): List of required scopes.
state (str, optional): Payload containing arbitrary data that the authentication flow
will pass back verbatim in a state query parameter to the callback URL.
Returns:
str: Complete authorization URL.
Examples:
```
FORGE_CLIENT_ID = os.environ["FORGE_CLIENT_ID"]
client = AuthenticationClient()
url = client.get_authorization_url(
FORGE_CLIENT_ID, "code", "http://localhost:3000/callback", [Scope.VIEWABLES_READ])
print(url)
```
"""
url = "https://developer.api.autodesk.com/authentication/v1/authorize"
url += f"?response_type={quote(response_type)}"
url += f"&client_id={quote(client_id)}"
url += f"&redirect_uri={quote_plus(redirect_uri)}"
url += f"&scope={'%20'.join(map(lambda s: s.value, scopes))}"
if state:
url += f"&state={state}"
return url
class AuthenticationClient(BaseClient):
"""
Forge Authentication service client.
**Documentation**: https://forge.autodesk.com/en/docs/oauth/v2/reference/http
"""
def __init__(self, base_url: str = BASE_URL):
"""
Create new instance of the client.
Args:
base_url (str, optional): Base URL for API calls.
"""
BaseClient.__init__(self, base_url)
def authenticate(self, client_id: str, client_secret: str, scopes: List[Scope]) -> Dict:
"""
Generate a two-legged access token for specific set of scopes.
**Documentation**:
https://forge.autodesk.com/en/docs/oauth/v2/reference/http/authenticate-POST
Args:
client_id (str): Client ID of the app.
client_secret (str): Client secret of the app.
scopes (List[Scope]): List of required scopes.
Returns:
Dict: Parsed response object with properties `access_token`, `token_type`,
and `expires_in`.
Examples:
```
FORGE_CLIENT_ID = os.environ["FORGE_CLIENT_ID"]
FORGE_CLIENT_SECRET = os.environ["FORGE_CLIENT_SECRET"]
client = AuthenticationClient()
auth = client.authenticate(FORGE_CLIENT_ID, FORGE_CLIENT_SECRET, [Scope.VIEWABLES_READ])
print(auth["access_token"])
```
"""
form = {
'client_id': client_id,
'client_secret': client_secret,
'grant_type': 'client_credentials',
'scope': ' '.join(map(lambda s: s.value, scopes))
}
return self._post('/authenticate', form=form).json()
def get_token(self, client_id: str, client_secret: str, code: str, redirect_uri: str) -> Dict:
"""
Exchange an authorization code extracted from `get_authorization_url` callback
for a three-legged access token. This API will only be used when the 'Authorization Code'
grant type is being adopted.
**Documentation**: https://forge.autodesk.com/en/docs/oauth/v2/reference/http/gettoken-POST
Args:
client_id (str): Client ID of the app.
client_secret (str): Client secret of the app.
code (str): The authorization code captured from the code query parameter
when the user is redirected back to the callback URL.
redirect_uri (str): Must match the redirect_uri parameter
used in the `get_authorization_url`.
Returns:
Dict: Parsed response object with properties `token_type`, `access_token`,
`refresh_token`, and `expires_in`.
Examples:
```
FORGE_CLIENT_ID = os.environ["FORGE_CLIENT_ID"]
FORGE_CLIENT_SECRET = os.environ["FORGE_CLIENT_SECRET"]
client = AuthenticationClient()
url = client.get_authorization_url(
FORGE_CLIENT_ID, "code", "http://localhost:3000/callback", [Scope.VIEWABLES_READ])
# redirect the user to URL, and wait for callback to http://localhost:3000/callback
code = '...' # extract 'code' query parameter from the callback
auth = client.get_token(
FORGE_CLIENT_ID, FORGE_CLIENT_SECRET, code, "http://localhost:3000/callback")
print(auth["access_token"])
```
"""
form = {
'client_id': client_id,
'client_secret': client_secret,
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': redirect_uri
}
return self._post('/gettoken', form=form).json()
def refresh_token(
self, client_id: str, client_secret: str, refresh_token: str, scopes: List[Scope]) -> Dict:
"""
Acquire a new access token by using the refresh token provided by `get_token`.
**Documentation**:
https://forge.autodesk.com/en/docs/oauth/v2/reference/http/refreshtoken-POST
Args:
client_id (str): Client ID of the app.
client_secret (str): Client secret of the app.
refresh_token (str): Refresh token used to acquire a new access token.
scopes (List[str]): List of required scopes.
Returns:
Dict: Parsed response object with properties `token_type`, `access_token`,
`refresh_token`, and `expires_in`.
Examples:
```
FORGE_CLIENT_ID = os.environ["FORGE_CLIENT_ID"]
FORGE_CLIENT_SECRET = os.environ["FORGE_CLIENT_SECRET"]
refresh_token = '...' # retrieve the refresh token, for example, from cookies
client = AuthenticationClient()
auth = client.refresh_token(
FORGE_CLIENT_ID, FORGE_CLIENT_SECRET, refresh_token, [Scope.VIEWABLES_READ])
print(auth["access_token"])
```
"""
form = {
'client_id': client_id,
'client_secret': client_secret,
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
'scope': ' '.join(map(lambda s: s.value, scopes))
}
return self._post('/refreshtoken', form=form).json()
def get_user_profile(self, access_token: str) -> Dict:
"""
Get the profile information of an authorizing end user in a three-legged context.
**Documentation**: https://forge.autodesk.com/en/docs/oauth/v2/reference/http/users-@me-GET
Args:
access_token (str): Token obtained via a three-legged OAuth flow.
Returns:
Dict: Parsed response object with properties `userId`, `userName`, `emaillId`,
`firstName`, `lastName`, etc.
Examples:
```
access_token = '...' # get a three-legged access token, for example, from cookies
client = AuthenticationClient()
info = client.get_user_profile(access_token)
print(auth["userName"])
```
"""
headers = {'Authorization': 'Bearer {}'.format(access_token)}
return self._get('/users/@me', headers=headers).json()
class TokenProviderInterface:
"""
Interface for any class that can provide access tokens to API clients
based on a set of OAuth scopes.
"""
def get_token(self, scopes: List[Scope]) -> str:
"""
Generates access token for given set of scopes.
Args:
scopes (List[Scope]): List of scopes that the generated access token should support.
Returns:
str: Access token.
"""
raise NotImplementedError()
class SimpleTokenProvider(TokenProviderInterface):
"""
Simple implementation of `TokenProviderInterface` when you already have an access token
that you want to use. When using this approach, make sure that the hard-coded access token
supports all the scopes that may be needed.
"""
def __init__(self, access_token: str):
"""
Create new instance of the provider.
Args:
access_token (str): Token that will always be returned by
`SimpleTokenProvider.get_token`.
"""
self.access_token = access_token
def get_token(self, scopes: List[Scope]) -> str:
return self.access_token
class OAuthTokenProvider(TokenProviderInterface):
"""
Helper class that automatically generates (and caches) access tokens
using specific app credentials.
"""
def __init__(self, client_id: str, client_secret: str):
"""
Create new instance of the provider.
Args:
client_id (str): Application client ID.
client_secret (str): Application client secret.
"""
self.client_id = client_id
self.client_secret = client_secret
self.auth_client = AuthenticationClient()
self.cache = {}
def get_token(self, scopes: List[Scope]) -> str:
cache_key = "+".join(map(lambda s: s.value, scopes))
now = datetime.now()
if cache_key in self.cache:
auth = self.cache[cache_key]
if auth["expires_at"] > now:
return auth
auth = self.auth_client.authenticate(
self.client_id, self.client_secret, scopes)
auth["expires_at"] = now + timedelta(0, auth["expires_in"])
return auth
class BaseOAuthClient(BaseClient):
"""
Base class for API clients with authentication based on OAuth 2.0.
"""
def __init__(self, token_provider: TokenProviderInterface, base_url: str):
"""
Create new instance of the client.
Args:
token_provider (TokenProviderInterface): Provider that will be used
to generate access tokens for API calls.
base_url (str): Base URL for API calls.
"""
BaseClient.__init__(self, base_url)
self.token_provider = token_provider
def _head(self, url: str, **kwargs):
if "scopes" in kwargs:
if "headers" not in kwargs:
kwargs["headers"] = {}
self._set_auth_headers(kwargs["headers"], kwargs["scopes"])
del kwargs["scopes"]
return BaseClient._head(self, url, **kwargs)
def _get(self, url: str, **kwargs):
if "scopes" in kwargs:
if "headers" not in kwargs:
kwargs["headers"] = {}
self._set_auth_headers(kwargs["headers"], kwargs["scopes"])
del kwargs["scopes"]
return BaseClient._get(self, url, **kwargs)
def _post(self, url: str, form: Dict = None, json: Dict = None, buff=None, **kwargs):
if "scopes" in kwargs:
if "headers" not in kwargs:
kwargs["headers"] = {}
self._set_auth_headers(kwargs["headers"], kwargs["scopes"])
del kwargs["scopes"]
return BaseClient._post(self, url, form, json, buff, **kwargs)
def _put(self, url: str, form: Dict = None, json: Dict = None, buff=None, **kwargs):
if "scopes" in kwargs:
if "headers" not in kwargs:
kwargs["headers"] = {}
self._set_auth_headers(kwargs["headers"], kwargs["scopes"])
del kwargs["scopes"]
return BaseClient._put(self, url, form, json, buff, **kwargs)
def _delete(self, url: str, **kwargs):
if "scopes" in kwargs:
if "headers" not in kwargs:
kwargs["headers"] = {}
self._set_auth_headers(kwargs["headers"], kwargs["scopes"])
del kwargs["scopes"]
return BaseClient._delete(self, url, **kwargs)
def _set_auth_headers(self, headers: Dict, scopes: List[Scope]):
if "Authorization" not in headers:
auth = self.token_provider.get_token(scopes)
headers["Authorization"] = "Bearer {}".format(auth["access_token"])