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
|
# Copyright (c) 2021 AtLarge Research
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import json
import time
import urllib3
from flask import request
from jose import jwt, JWTError
def get_token():
"""
Obtain the Access Token from the Authorization Header
"""
auth = request.headers.get("Authorization", None)
if not auth:
raise AuthError({
"code": "authorization_header_missing",
"description": "Authorization header is expected"
}, 401)
parts = auth.split()
if parts[0].lower() != "bearer":
raise AuthError({"code": "invalid_header", "description": "Authorization header must start with Bearer"}, 401)
if len(parts) == 1:
raise AuthError({"code": "invalid_header", "description": "Token not found"}, 401)
if len(parts) > 2:
raise AuthError({"code": "invalid_header", "description": "Authorization header must be" " Bearer token"}, 401)
token = parts[1]
return token
class AuthError(Exception):
"""
This error is thrown when the request failed to authorize.
"""
def __init__(self, error, status_code):
Exception.__init__(self, error)
self.error = error
self.status_code = status_code
class AuthContext:
"""
This class handles the authorization of requests.
"""
def __init__(self, alg, issuer, audience):
self._alg = alg
self._issuer = issuer
self._audience = audience
def validate(self, token):
"""
Validate the specified JWT token.
:param token: The authorization token specified by the user.
:return: The token payload on success, otherwise `AuthError`.
"""
try:
header = jwt.get_unverified_header(token)
except JWTError as e:
raise AuthError({"code": "invalid_token", "message": str(e)}, 401)
alg = header.get('alg', None)
if alg != self._alg.algorithm:
raise AuthError(
{
"code":
"invalid_header",
"message":
f"Signature algorithm of {alg} is not supported. Expected the ID token "
f"to be signed with {self._alg.algorithm}"
}, 401)
kid = header.get('kid', None)
try:
secret_or_certificate = self._alg.get_key(key_id=kid)
except TokenValidationError as e:
raise AuthError({"code": "invalid_header", "message": str(e)}, 401)
try:
payload = jwt.decode(token,
key=secret_or_certificate,
algorithms=[self._alg.algorithm],
audience=self._audience,
issuer=self._issuer)
return payload
except jwt.ExpiredSignatureError:
raise AuthError({"code": "token_expired", "message": "Token is expired"}, 401)
except jwt.JWTClaimsError:
raise AuthError(
{
"code": "invalid_claims",
"message": "Incorrect claims, please check the audience and issuer"
}, 401)
except Exception as e:
print(e)
raise AuthError({"code": "invalid_header", "message": "Unable to parse authentication token."}, 401)
class SymmetricJwtAlgorithm:
"""Verifier for HMAC signatures, which rely on shared secrets.
Args:
shared_secret (str): The shared secret used to decode the token.
algorithm (str, optional): The expected signing algorithm. Defaults to "HS256".
"""
def __init__(self, shared_secret, algorithm="HS256"):
self.algorithm = algorithm
self._shared_secret = shared_secret
# pylint: disable=W0613
def get_key(self, key_id=None):
"""
Obtain the key for this algorithm.
:param key_id: The identifier of the key.
:return: The JWK key.
"""
return self._shared_secret
class AsymmetricJwtAlgorithm:
"""Verifier for RSA signatures, which rely on public key certificates.
Args:
jwks_url (str): The url where the JWK set is located.
algorithm (str, optional): The expected signing algorithm. Defaults to "RS256".
"""
def __init__(self, jwks_url, algorithm="RS256"):
self.algorithm = algorithm
self._fetcher = JwksFetcher(jwks_url)
def get_key(self, key_id=None):
"""
Obtain the key for this algorithm.
:param key_id: The identifier of the key.
:return: The JWK key.
"""
return self._fetcher.get_key(key_id)
class TokenValidationError(Exception):
"""
Error thrown when the token cannot be validated
"""
class JwksFetcher:
"""Class that fetches and holds a JSON web key set.
This class makes use of an in-memory cache. For it to work properly, define this instance once and re-use it.
Args:
jwks_url (str): The url where the JWK set is located.
cache_ttl (str, optional): The lifetime of the JWK set cache in seconds. Defaults to 600 seconds.
"""
CACHE_TTL = 600 # 10 min cache lifetime
def __init__(self, jwks_url, cache_ttl=CACHE_TTL):
self._jwks_url = jwks_url
self._http = urllib3.PoolManager()
self._cache_value = {}
self._cache_date = 0
self._cache_ttl = cache_ttl
self._cache_is_fresh = False
def _fetch_jwks(self, force=False):
"""Attempts to obtain the JWK set from the cache, as long as it's still valid.
When not, it will perform a network request to the jwks_url to obtain a fresh result
and update the cache value with it.
Args:
force (bool, optional): whether to ignore the cache and force a network request or not. Defaults to False.
"""
has_expired = self._cache_date + self._cache_ttl < time.time()
if not force and not has_expired:
# Return from cache
self._cache_is_fresh = False
return self._cache_value
# Invalidate cache and fetch fresh data
self._cache_value = {}
response = self._http.request('GET', self._jwks_url)
if response.status == 200:
# Update cache
jwks = json.loads(response.data.decode('utf-8'))
self._cache_value = self._parse_jwks(jwks)
self._cache_is_fresh = True
self._cache_date = time.time()
return self._cache_value
@staticmethod
def _parse_jwks(jwks):
"""Converts a JWK string representation into a binary certificate in PEM format.
"""
keys = {}
for key in jwks['keys']:
keys[key["kid"]] = key
return keys
def get_key(self, key_id):
"""Obtains the JWK associated with the given key id.
Args:
key_id (str): The id of the key to fetch.
Returns:
the JWK associated with the given key id.
Raises:
TokenValidationError: when a key with that id cannot be found
"""
keys = self._fetch_jwks()
if keys and key_id in keys:
return keys[key_id]
if not self._cache_is_fresh:
keys = self._fetch_jwks(force=True)
if keys and key_id in keys:
return keys[key_id]
raise TokenValidationError(f"RSA Public Key with ID {key_id} was not found.")
|
