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
|
import datetime
import unittest
import pytest
from authlib.jose import JsonWebKey
from authlib.jose import JsonWebToken
from authlib.jose import JWTClaims
from authlib.jose import errors
from authlib.jose import jwt
from authlib.jose.errors import UnsupportedAlgorithmError
from tests.util import read_file_path
class JWTTest(unittest.TestCase):
def test_init_algorithms(self):
_jwt = JsonWebToken(["RS256"])
with pytest.raises(UnsupportedAlgorithmError):
_jwt.encode({"alg": "HS256"}, {}, "k")
_jwt = JsonWebToken("RS256")
with pytest.raises(UnsupportedAlgorithmError):
_jwt.encode({"alg": "HS256"}, {}, "k")
def test_encode_sensitive_data(self):
# check=False won't raise error
jwt.encode({"alg": "HS256"}, {"password": ""}, "k", check=False)
with pytest.raises(errors.InsecureClaimError):
jwt.encode(
{"alg": "HS256"},
{"password": ""},
"k",
)
with pytest.raises(errors.InsecureClaimError):
jwt.encode(
{"alg": "HS256"},
{"text": "4242424242424242"},
"k",
)
def test_encode_datetime(self):
now = datetime.datetime.now(tz=datetime.timezone.utc)
id_token = jwt.encode({"alg": "HS256"}, {"exp": now}, "k")
claims = jwt.decode(id_token, "k")
assert isinstance(claims.exp, int)
def test_validate_essential_claims(self):
id_token = jwt.encode({"alg": "HS256"}, {"iss": "foo"}, "k")
claims_options = {"iss": {"essential": True, "values": ["foo"]}}
claims = jwt.decode(id_token, "k", claims_options=claims_options)
claims.validate()
claims.options = {"sub": {"essential": True}}
with pytest.raises(errors.MissingClaimError):
claims.validate()
def test_attribute_error(self):
claims = JWTClaims({"iss": "foo"}, {"alg": "HS256"})
with pytest.raises(AttributeError):
claims.invalid # noqa: B018
def test_invalid_values(self):
id_token = jwt.encode({"alg": "HS256"}, {"iss": "foo"}, "k")
claims_options = {"iss": {"values": ["bar"]}}
claims = jwt.decode(id_token, "k", claims_options=claims_options)
with pytest.raises(errors.InvalidClaimError):
claims.validate()
claims.options = {"iss": {"value": "bar"}}
with pytest.raises(errors.InvalidClaimError):
claims.validate()
def test_validate_expected_issuer_received_None(self):
id_token = jwt.encode({"alg": "HS256"}, {"iss": None, "sub": None}, "k")
claims_options = {"iss": {"essential": True, "values": ["foo"]}}
claims = jwt.decode(id_token, "k", claims_options=claims_options)
with pytest.raises(errors.InvalidClaimError):
claims.validate()
def test_validate_aud(self):
id_token = jwt.encode({"alg": "HS256"}, {"aud": "foo"}, "k")
claims_options = {"aud": {"essential": True, "value": "foo"}}
claims = jwt.decode(id_token, "k", claims_options=claims_options)
claims.validate()
claims.options = {"aud": {"values": ["bar"]}}
with pytest.raises(errors.InvalidClaimError):
claims.validate()
id_token = jwt.encode({"alg": "HS256"}, {"aud": ["foo", "bar"]}, "k")
claims = jwt.decode(id_token, "k", claims_options=claims_options)
claims.validate()
# no validate
claims.options = {"aud": {"values": []}}
claims.validate()
def test_validate_exp(self):
id_token = jwt.encode({"alg": "HS256"}, {"exp": "invalid"}, "k")
claims = jwt.decode(id_token, "k")
with pytest.raises(errors.InvalidClaimError):
claims.validate()
id_token = jwt.encode({"alg": "HS256"}, {"exp": 1234}, "k")
claims = jwt.decode(id_token, "k")
with pytest.raises(errors.ExpiredTokenError):
claims.validate()
def test_validate_nbf(self):
id_token = jwt.encode({"alg": "HS256"}, {"nbf": "invalid"}, "k")
claims = jwt.decode(id_token, "k")
with pytest.raises(errors.InvalidClaimError):
claims.validate()
id_token = jwt.encode({"alg": "HS256"}, {"nbf": 1234}, "k")
claims = jwt.decode(id_token, "k")
claims.validate()
id_token = jwt.encode({"alg": "HS256"}, {"nbf": 1234}, "k")
claims = jwt.decode(id_token, "k")
with pytest.raises(errors.InvalidTokenError):
claims.validate(123)
def test_validate_iat_issued_in_future(self):
in_future = datetime.datetime.now(
tz=datetime.timezone.utc
) + datetime.timedelta(seconds=10)
id_token = jwt.encode({"alg": "HS256"}, {"iat": in_future}, "k")
claims = jwt.decode(id_token, "k")
with pytest.raises(
errors.InvalidTokenError,
match="The token is not valid as it was issued in the future",
):
claims.validate()
def test_validate_iat_issued_in_future_with_insufficient_leeway(self):
in_future = datetime.datetime.now(
tz=datetime.timezone.utc
) + datetime.timedelta(seconds=10)
id_token = jwt.encode({"alg": "HS256"}, {"iat": in_future}, "k")
claims = jwt.decode(id_token, "k")
with pytest.raises(
errors.InvalidTokenError,
match="The token is not valid as it was issued in the future",
):
claims.validate(leeway=5)
def test_validate_iat_issued_in_future_with_sufficient_leeway(self):
in_future = datetime.datetime.now(
tz=datetime.timezone.utc
) + datetime.timedelta(seconds=10)
id_token = jwt.encode({"alg": "HS256"}, {"iat": in_future}, "k")
claims = jwt.decode(id_token, "k")
claims.validate(leeway=20)
def test_validate_iat_issued_in_past(self):
in_future = datetime.datetime.now(
tz=datetime.timezone.utc
) - datetime.timedelta(seconds=10)
id_token = jwt.encode({"alg": "HS256"}, {"iat": in_future}, "k")
claims = jwt.decode(id_token, "k")
claims.validate()
def test_validate_iat(self):
id_token = jwt.encode({"alg": "HS256"}, {"iat": "invalid"}, "k")
claims = jwt.decode(id_token, "k")
with pytest.raises(errors.InvalidClaimError):
claims.validate()
def test_validate_jti(self):
id_token = jwt.encode({"alg": "HS256"}, {"jti": "bar"}, "k")
claims_options = {"jti": {"validate": lambda c, o: o == "foo"}}
claims = jwt.decode(id_token, "k", claims_options=claims_options)
with pytest.raises(errors.InvalidClaimError):
claims.validate()
def test_validate_custom(self):
id_token = jwt.encode({"alg": "HS256"}, {"custom": "foo"}, "k")
claims_options = {"custom": {"validate": lambda c, o: o == "bar"}}
claims = jwt.decode(id_token, "k", claims_options=claims_options)
with pytest.raises(errors.InvalidClaimError):
claims.validate()
def test_use_jws(self):
payload = {"name": "hi"}
private_key = read_file_path("rsa_private.pem")
pub_key = read_file_path("rsa_public.pem")
data = jwt.encode({"alg": "RS256"}, payload, private_key)
assert data.count(b".") == 2
claims = jwt.decode(data, pub_key)
assert claims["name"] == "hi"
def test_use_jwe(self):
payload = {"name": "hi"}
private_key = read_file_path("rsa_private.pem")
pub_key = read_file_path("rsa_public.pem")
_jwt = JsonWebToken(["RSA-OAEP", "A256GCM"])
data = _jwt.encode({"alg": "RSA-OAEP", "enc": "A256GCM"}, payload, pub_key)
assert data.count(b".") == 4
claims = _jwt.decode(data, private_key)
assert claims["name"] == "hi"
def test_use_jwks(self):
header = {"alg": "RS256", "kid": "abc"}
payload = {"name": "hi"}
private_key = read_file_path("jwks_private.json")
pub_key = read_file_path("jwks_public.json")
data = jwt.encode(header, payload, private_key)
assert data.count(b".") == 2
claims = jwt.decode(data, pub_key)
assert claims["name"] == "hi"
def test_use_jwks_single_kid(self):
"""Test that jwks can be decoded if a kid for decoding is given and encoded data has no kid and only one key is set."""
header = {"alg": "RS256"}
payload = {"name": "hi"}
private_key = read_file_path("jwks_single_private.json")
pub_key = read_file_path("jwks_single_public.json")
data = jwt.encode(header, payload, private_key)
assert data.count(b".") == 2
claims = jwt.decode(data, pub_key)
assert claims["name"] == "hi"
# Added a unit test to showcase my problem.
# This calls jwt.decode similarly as is done in parse_id_token method of the AsyncOpenIDMixin class when the id token does not contain a kid in the alg header.
def test_use_jwks_single_kid_keyset(self):
"""Test that jwks can be decoded if a kid for decoding is given and encoded data has no kid and a keyset with one key."""
header = {"alg": "RS256"}
payload = {"name": "hi"}
private_key = read_file_path("jwks_single_private.json")
pub_key = read_file_path("jwks_single_public.json")
data = jwt.encode(header, payload, private_key)
assert data.count(b".") == 2
claims = jwt.decode(data, JsonWebKey.import_key_set(pub_key))
assert claims["name"] == "hi"
def test_with_ec(self):
payload = {"name": "hi"}
private_key = read_file_path("secp521r1-private.json")
pub_key = read_file_path("secp521r1-public.json")
data = jwt.encode({"alg": "ES512"}, payload, private_key)
assert data.count(b".") == 2
claims = jwt.decode(data, pub_key)
assert claims["name"] == "hi"
|
