-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathutils.py
More file actions
100 lines (85 loc) · 2.93 KB
/
Copy pathutils.py
File metadata and controls
100 lines (85 loc) · 2.93 KB
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
"""
utils
~~~~~~~~~~
Various utilities for nextcode-sdk functionality.
"""
import jwt
import logging
from urllib.parse import urlsplit
import requests
from requests import codes
from .exceptions import ServerError, InvalidToken
log = logging.getLogger(__name__)
def decode_token(token):
try:
decoded_token = jwt.decode(token, algorithms=["RS256"], verify=False)
return decoded_token
except (KeyError, jwt.InvalidTokenError):
raise InvalidToken("Token could not be decoded")
def check_resp_error(resp):
response_json = None
try:
resp.raise_for_status()
except Exception:
desc = resp.text
try:
response_json = resp.json()
desc = response_json["error"]["description"]
log.info(response_json)
if "errors" in response_json["error"]:
desc += " (%s)" % (response_json["error"]["errors"])
except Exception:
pass
if not desc:
desc = "Status code %s received (%s)" % (resp.status_code, resp.text)
else:
desc += " (code %s)" % resp.status_code
if resp.status_code >= 500:
desc = "Server error in call to %s" % resp.url
desc += " - Response headers: %s" % resp.headers
desc += " - Response body: %s" % resp.text
log.error(desc)
else:
log.info("Server error in call to %s", resp.url)
error = ServerError(desc, url=resp.url, response=response_json)
raise error from None
def root_url_from_api_key(api_key):
payload = decode_token(api_key)
parts = urlsplit(payload["iss"])
root_url = "{scheme}://{netloc}".format(scheme=parts.scheme, netloc=parts.netloc)
return root_url
def get_access_token(api_key):
"""
"""
payload = decode_token(api_key)
client_id = payload["azp"]
body = {
"grant_type": "refresh_token",
"client_id": client_id,
"refresh_token": api_key,
"username": "dummy_user",
}
headers = {"Content-Type": "application/x-www-form-urlencoded"}
token_endpoint = "{}/protocol/openid-connect/token".format(payload["iss"])
# Call the auth server
log.info("Authenticating with %s", token_endpoint)
response = requests.post(token_endpoint, data=body, headers=headers)
if (
response.status_code == codes.bad_request
and "Refresh token expired" in response.text
):
raise InvalidToken("Refresh token has expired")
elif response.status_code >= codes.bad_request:
try:
if response.json():
raise InvalidToken(response.json().get("error_description"))
except Exception:
pass
try:
response.raise_for_status()
except Exception:
log.error("Body: %s" % body)
raise InvalidToken(
"Error authenticating with %s: %s" % (token_endpoint, response.text)
)
return response.json()["access_token"]