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
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
|
# Copyright 2021 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import datetime
import http.client as http_client
import json
import urllib
import mock
import pytest # type: ignore
from google.auth import _helpers
from google.auth import credentials
from google.auth import downscoped
from google.auth import exceptions
from google.auth import transport
from google.auth.credentials import TokenState
EXPRESSION = (
"resource.name.startsWith('projects/_/buckets/example-bucket/objects/customer-a')"
)
TITLE = "customer-a-objects"
DESCRIPTION = (
"Condition to make permissions available for objects starting with customer-a"
)
AVAILABLE_RESOURCE = "//storage.googleapis.com/projects/_/buckets/example-bucket"
AVAILABLE_PERMISSIONS = ["inRole:roles/storage.objectViewer"]
OTHER_EXPRESSION = (
"resource.name.startsWith('projects/_/buckets/example-bucket/objects/customer-b')"
)
OTHER_TITLE = "customer-b-objects"
OTHER_DESCRIPTION = (
"Condition to make permissions available for objects starting with customer-b"
)
OTHER_AVAILABLE_RESOURCE = "//storage.googleapis.com/projects/_/buckets/other-bucket"
OTHER_AVAILABLE_PERMISSIONS = ["inRole:roles/storage.objectCreator"]
QUOTA_PROJECT_ID = "QUOTA_PROJECT_ID"
GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"
REQUESTED_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"
TOKEN_EXCHANGE_ENDPOINT = "https://sts.googleapis.com/v1/token"
SUBJECT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"
SUCCESS_RESPONSE = {
"access_token": "ACCESS_TOKEN",
"issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
"token_type": "Bearer",
"expires_in": 3600,
}
ERROR_RESPONSE = {
"error": "invalid_grant",
"error_description": "Subject token is invalid.",
"error_uri": "https://tools.ietf.org/html/rfc6749",
}
CREDENTIAL_ACCESS_BOUNDARY_JSON = {
"accessBoundary": {
"accessBoundaryRules": [
{
"availablePermissions": AVAILABLE_PERMISSIONS,
"availableResource": AVAILABLE_RESOURCE,
"availabilityCondition": {
"expression": EXPRESSION,
"title": TITLE,
"description": DESCRIPTION,
},
}
]
}
}
class SourceCredentials(credentials.Credentials):
def __init__(self, raise_error=False, expires_in=3600):
super(SourceCredentials, self).__init__()
self._counter = 0
self._raise_error = raise_error
self._expires_in = expires_in
def refresh(self, request):
if self._raise_error:
raise exceptions.RefreshError(
"Failed to refresh access token in source credentials."
)
now = _helpers.utcnow()
self._counter += 1
self.token = "ACCESS_TOKEN_{}".format(self._counter)
self.expiry = now + datetime.timedelta(seconds=self._expires_in)
def make_availability_condition(expression, title=None, description=None):
return downscoped.AvailabilityCondition(expression, title, description)
def make_access_boundary_rule(
available_resource, available_permissions, availability_condition=None
):
return downscoped.AccessBoundaryRule(
available_resource, available_permissions, availability_condition
)
def make_credential_access_boundary(rules):
return downscoped.CredentialAccessBoundary(rules)
class TestAvailabilityCondition(object):
def test_constructor(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
assert availability_condition.expression == EXPRESSION
assert availability_condition.title == TITLE
assert availability_condition.description == DESCRIPTION
def test_constructor_required_params_only(self):
availability_condition = make_availability_condition(EXPRESSION)
assert availability_condition.expression == EXPRESSION
assert availability_condition.title is None
assert availability_condition.description is None
def test_setters(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
availability_condition.expression = OTHER_EXPRESSION
availability_condition.title = OTHER_TITLE
availability_condition.description = OTHER_DESCRIPTION
assert availability_condition.expression == OTHER_EXPRESSION
assert availability_condition.title == OTHER_TITLE
assert availability_condition.description == OTHER_DESCRIPTION
def test_invalid_expression_type(self):
with pytest.raises(TypeError) as excinfo:
make_availability_condition([EXPRESSION], TITLE, DESCRIPTION)
assert excinfo.match("The provided expression is not a string.")
def test_invalid_title_type(self):
with pytest.raises(TypeError) as excinfo:
make_availability_condition(EXPRESSION, False, DESCRIPTION)
assert excinfo.match("The provided title is not a string or None.")
def test_invalid_description_type(self):
with pytest.raises(TypeError) as excinfo:
make_availability_condition(EXPRESSION, TITLE, False)
assert excinfo.match("The provided description is not a string or None.")
def test_to_json_required_params_only(self):
availability_condition = make_availability_condition(EXPRESSION)
assert availability_condition.to_json() == {"expression": EXPRESSION}
def test_to_json_(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
assert availability_condition.to_json() == {
"expression": EXPRESSION,
"title": TITLE,
"description": DESCRIPTION,
}
class TestAccessBoundaryRule(object):
def test_constructor(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
assert access_boundary_rule.available_resource == AVAILABLE_RESOURCE
assert access_boundary_rule.available_permissions == tuple(
AVAILABLE_PERMISSIONS
)
assert access_boundary_rule.availability_condition == availability_condition
def test_constructor_required_params_only(self):
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS
)
assert access_boundary_rule.available_resource == AVAILABLE_RESOURCE
assert access_boundary_rule.available_permissions == tuple(
AVAILABLE_PERMISSIONS
)
assert access_boundary_rule.availability_condition is None
def test_setters(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
other_availability_condition = make_availability_condition(
OTHER_EXPRESSION, OTHER_TITLE, OTHER_DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
access_boundary_rule.available_resource = OTHER_AVAILABLE_RESOURCE
access_boundary_rule.available_permissions = OTHER_AVAILABLE_PERMISSIONS
access_boundary_rule.availability_condition = other_availability_condition
assert access_boundary_rule.available_resource == OTHER_AVAILABLE_RESOURCE
assert access_boundary_rule.available_permissions == tuple(
OTHER_AVAILABLE_PERMISSIONS
)
assert (
access_boundary_rule.availability_condition == other_availability_condition
)
def test_invalid_available_resource_type(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
with pytest.raises(TypeError) as excinfo:
make_access_boundary_rule(
None, AVAILABLE_PERMISSIONS, availability_condition
)
assert excinfo.match("The provided available_resource is not a string.")
def test_invalid_available_permissions_type(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
with pytest.raises(TypeError) as excinfo:
make_access_boundary_rule(
AVAILABLE_RESOURCE, [0, 1, 2], availability_condition
)
assert excinfo.match(
"Provided available_permissions are not a list of strings."
)
def test_invalid_available_permissions_value(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
with pytest.raises(ValueError) as excinfo:
make_access_boundary_rule(
AVAILABLE_RESOURCE,
["roles/storage.objectViewer"],
availability_condition,
)
assert excinfo.match("available_permissions must be prefixed with 'inRole:'.")
def test_invalid_availability_condition_type(self):
with pytest.raises(TypeError) as excinfo:
make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, {"foo": "bar"}
)
assert excinfo.match(
"The provided availability_condition is not a 'google.auth.downscoped.AvailabilityCondition' or None."
)
def test_to_json(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
assert access_boundary_rule.to_json() == {
"availablePermissions": AVAILABLE_PERMISSIONS,
"availableResource": AVAILABLE_RESOURCE,
"availabilityCondition": {
"expression": EXPRESSION,
"title": TITLE,
"description": DESCRIPTION,
},
}
def test_to_json_required_params_only(self):
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS
)
assert access_boundary_rule.to_json() == {
"availablePermissions": AVAILABLE_PERMISSIONS,
"availableResource": AVAILABLE_RESOURCE,
}
class TestCredentialAccessBoundary(object):
def test_constructor(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
rules = [access_boundary_rule]
credential_access_boundary = make_credential_access_boundary(rules)
assert credential_access_boundary.rules == tuple(rules)
def test_setters(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
rules = [access_boundary_rule]
other_availability_condition = make_availability_condition(
OTHER_EXPRESSION, OTHER_TITLE, OTHER_DESCRIPTION
)
other_access_boundary_rule = make_access_boundary_rule(
OTHER_AVAILABLE_RESOURCE,
OTHER_AVAILABLE_PERMISSIONS,
other_availability_condition,
)
other_rules = [other_access_boundary_rule]
credential_access_boundary = make_credential_access_boundary(rules)
credential_access_boundary.rules = other_rules
assert credential_access_boundary.rules == tuple(other_rules)
def test_add_rule(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
rules = [access_boundary_rule] * 9
credential_access_boundary = make_credential_access_boundary(rules)
# Add one more rule. This should not raise an error.
additional_access_boundary_rule = make_access_boundary_rule(
OTHER_AVAILABLE_RESOURCE, OTHER_AVAILABLE_PERMISSIONS
)
credential_access_boundary.add_rule(additional_access_boundary_rule)
assert len(credential_access_boundary.rules) == 10
assert credential_access_boundary.rules[9] == additional_access_boundary_rule
def test_add_rule_invalid_value(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
rules = [access_boundary_rule] * 10
credential_access_boundary = make_credential_access_boundary(rules)
# Add one more rule to exceed maximum allowed rules.
with pytest.raises(ValueError) as excinfo:
credential_access_boundary.add_rule(access_boundary_rule)
assert excinfo.match(
"Credential access boundary rules can have a maximum of 10 rules."
)
assert len(credential_access_boundary.rules) == 10
def test_add_rule_invalid_type(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
rules = [access_boundary_rule]
credential_access_boundary = make_credential_access_boundary(rules)
# Add an invalid rule to exceed maximum allowed rules.
with pytest.raises(TypeError) as excinfo:
credential_access_boundary.add_rule("invalid")
assert excinfo.match(
"The provided rule does not contain a valid 'google.auth.downscoped.AccessBoundaryRule'."
)
assert len(credential_access_boundary.rules) == 1
assert credential_access_boundary.rules[0] == access_boundary_rule
def test_invalid_rules_type(self):
with pytest.raises(TypeError) as excinfo:
make_credential_access_boundary(["invalid"])
assert excinfo.match(
"List of rules provided do not contain a valid 'google.auth.downscoped.AccessBoundaryRule'."
)
def test_invalid_rules_value(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
too_many_rules = [access_boundary_rule] * 11
with pytest.raises(ValueError) as excinfo:
make_credential_access_boundary(too_many_rules)
assert excinfo.match(
"Credential access boundary rules can have a maximum of 10 rules."
)
def test_to_json(self):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
rules = [access_boundary_rule]
credential_access_boundary = make_credential_access_boundary(rules)
assert credential_access_boundary.to_json() == {
"accessBoundary": {
"accessBoundaryRules": [
{
"availablePermissions": AVAILABLE_PERMISSIONS,
"availableResource": AVAILABLE_RESOURCE,
"availabilityCondition": {
"expression": EXPRESSION,
"title": TITLE,
"description": DESCRIPTION,
},
}
]
}
}
class TestCredentials(object):
@staticmethod
def make_credentials(source_credentials=SourceCredentials(), quota_project_id=None):
availability_condition = make_availability_condition(
EXPRESSION, TITLE, DESCRIPTION
)
access_boundary_rule = make_access_boundary_rule(
AVAILABLE_RESOURCE, AVAILABLE_PERMISSIONS, availability_condition
)
rules = [access_boundary_rule]
credential_access_boundary = make_credential_access_boundary(rules)
return downscoped.Credentials(
source_credentials, credential_access_boundary, quota_project_id
)
@staticmethod
def make_mock_request(data, status=http_client.OK):
response = mock.create_autospec(transport.Response, instance=True)
response.status = status
response.data = json.dumps(data).encode("utf-8")
request = mock.create_autospec(transport.Request)
request.return_value = response
return request
@staticmethod
def assert_request_kwargs(request_kwargs, headers, request_data):
"""Asserts the request was called with the expected parameters.
"""
assert request_kwargs["url"] == TOKEN_EXCHANGE_ENDPOINT
assert request_kwargs["method"] == "POST"
assert request_kwargs["headers"] == headers
assert request_kwargs["body"] is not None
body_tuples = urllib.parse.parse_qsl(request_kwargs["body"])
for (k, v) in body_tuples:
assert v.decode("utf-8") == request_data[k.decode("utf-8")]
assert len(body_tuples) == len(request_data.keys())
def test_default_state(self):
credentials = self.make_credentials()
# No token acquired yet.
assert not credentials.token
assert not credentials.valid
# Expiration hasn't been set yet.
assert not credentials.expiry
assert not credentials.expired
# No quota project ID set.
assert not credentials.quota_project_id
def test_with_quota_project(self):
credentials = self.make_credentials()
assert not credentials.quota_project_id
quota_project_creds = credentials.with_quota_project("project-foo")
assert quota_project_creds.quota_project_id == "project-foo"
@mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min)
def test_refresh(self, unused_utcnow):
response = SUCCESS_RESPONSE.copy()
# Test custom expiration to confirm expiry is set correctly.
response["expires_in"] = 2800
expected_expiry = datetime.datetime.min + datetime.timedelta(
seconds=response["expires_in"]
)
headers = {"Content-Type": "application/x-www-form-urlencoded"}
request_data = {
"grant_type": GRANT_TYPE,
"subject_token": "ACCESS_TOKEN_1",
"subject_token_type": SUBJECT_TOKEN_TYPE,
"requested_token_type": REQUESTED_TOKEN_TYPE,
"options": urllib.parse.quote(json.dumps(CREDENTIAL_ACCESS_BOUNDARY_JSON)),
}
request = self.make_mock_request(status=http_client.OK, data=response)
source_credentials = SourceCredentials()
credentials = self.make_credentials(source_credentials=source_credentials)
# Spy on calls to source credentials refresh to confirm the expected request
# instance is used.
with mock.patch.object(
source_credentials, "refresh", wraps=source_credentials.refresh
) as wrapped_souce_cred_refresh:
credentials.refresh(request)
self.assert_request_kwargs(request.call_args[1], headers, request_data)
assert credentials.valid
assert credentials.expiry == expected_expiry
assert not credentials.expired
assert credentials.token == response["access_token"]
# Confirm source credentials called with the same request instance.
wrapped_souce_cred_refresh.assert_called_with(request)
@mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min)
def test_refresh_without_response_expires_in(self, unused_utcnow):
response = SUCCESS_RESPONSE.copy()
# Simulate the response is missing the expires_in field.
# The downscoped token expiration should match the source credentials
# expiration.
del response["expires_in"]
expected_expires_in = 1800
# Simulate the source credentials generates a token with 1800 second
# expiration time. The generated downscoped token should have the same
# expiration time.
source_credentials = SourceCredentials(expires_in=expected_expires_in)
expected_expiry = datetime.datetime.min + datetime.timedelta(
seconds=expected_expires_in
)
headers = {"Content-Type": "application/x-www-form-urlencoded"}
request_data = {
"grant_type": GRANT_TYPE,
"subject_token": "ACCESS_TOKEN_1",
"subject_token_type": SUBJECT_TOKEN_TYPE,
"requested_token_type": REQUESTED_TOKEN_TYPE,
"options": urllib.parse.quote(json.dumps(CREDENTIAL_ACCESS_BOUNDARY_JSON)),
}
request = self.make_mock_request(status=http_client.OK, data=response)
credentials = self.make_credentials(source_credentials=source_credentials)
# Spy on calls to source credentials refresh to confirm the expected request
# instance is used.
with mock.patch.object(
source_credentials, "refresh", wraps=source_credentials.refresh
) as wrapped_souce_cred_refresh:
credentials.refresh(request)
self.assert_request_kwargs(request.call_args[1], headers, request_data)
assert credentials.valid
assert credentials.expiry == expected_expiry
assert not credentials.expired
assert credentials.token == response["access_token"]
# Confirm source credentials called with the same request instance.
wrapped_souce_cred_refresh.assert_called_with(request)
def test_refresh_token_exchange_error(self):
request = self.make_mock_request(
status=http_client.BAD_REQUEST, data=ERROR_RESPONSE
)
credentials = self.make_credentials()
with pytest.raises(exceptions.OAuthError) as excinfo:
credentials.refresh(request)
assert excinfo.match(
r"Error code invalid_grant: Subject token is invalid. - https://tools.ietf.org/html/rfc6749"
)
assert not credentials.expired
assert credentials.token is None
def test_refresh_source_credentials_refresh_error(self):
# Initialize downscoped credentials with source credentials that raise
# an error on refresh.
credentials = self.make_credentials(
source_credentials=SourceCredentials(raise_error=True)
)
with pytest.raises(exceptions.RefreshError) as excinfo:
credentials.refresh(mock.sentinel.request)
assert excinfo.match(r"Failed to refresh access token in source credentials.")
assert not credentials.expired
assert credentials.token is None
def test_apply_without_quota_project_id(self):
headers = {}
request = self.make_mock_request(status=http_client.OK, data=SUCCESS_RESPONSE)
credentials = self.make_credentials()
credentials.refresh(request)
credentials.apply(headers)
assert headers == {
"authorization": "Bearer {}".format(SUCCESS_RESPONSE["access_token"])
}
def test_apply_with_quota_project_id(self):
headers = {"other": "header-value"}
request = self.make_mock_request(status=http_client.OK, data=SUCCESS_RESPONSE)
credentials = self.make_credentials(quota_project_id=QUOTA_PROJECT_ID)
credentials.refresh(request)
credentials.apply(headers)
assert headers == {
"other": "header-value",
"authorization": "Bearer {}".format(SUCCESS_RESPONSE["access_token"]),
"x-goog-user-project": QUOTA_PROJECT_ID,
}
def test_before_request(self):
headers = {"other": "header-value"}
request = self.make_mock_request(status=http_client.OK, data=SUCCESS_RESPONSE)
credentials = self.make_credentials()
# First call should call refresh, setting the token.
credentials.before_request(request, "POST", "https://example.com/api", headers)
assert headers == {
"other": "header-value",
"authorization": "Bearer {}".format(SUCCESS_RESPONSE["access_token"]),
}
# Second call shouldn't call refresh (request should be untouched).
credentials.before_request(
mock.sentinel.request, "POST", "https://example.com/api", headers
)
assert headers == {
"other": "header-value",
"authorization": "Bearer {}".format(SUCCESS_RESPONSE["access_token"]),
}
@mock.patch("google.auth._helpers.utcnow")
def test_before_request_expired(self, utcnow):
headers = {}
request = self.make_mock_request(status=http_client.OK, data=SUCCESS_RESPONSE)
credentials = self.make_credentials()
credentials.token = "token"
utcnow.return_value = datetime.datetime.min
# Set the expiration to one second more than now plus the clock skew
# accommodation. These credentials should be valid.
credentials.expiry = (
datetime.datetime.min
+ _helpers.REFRESH_THRESHOLD
+ datetime.timedelta(seconds=1)
)
assert credentials.valid
assert not credentials.expired
assert credentials.token_state == TokenState.FRESH
credentials.before_request(request, "POST", "https://example.com/api", headers)
# Cached token should be used.
assert headers == {"authorization": "Bearer token"}
# Next call should simulate 1 second passed.
utcnow.return_value = datetime.datetime.min + datetime.timedelta(seconds=1)
assert not credentials.valid
assert credentials.expired
assert credentials.token_state == TokenState.STALE
credentials.before_request(request, "POST", "https://example.com/api", headers)
assert credentials.token_state == TokenState.FRESH
# New token should be retrieved.
assert headers == {
"authorization": "Bearer {}".format(SUCCESS_RESPONSE["access_token"])
}
utcnow.return_value = datetime.datetime.min + datetime.timedelta(seconds=6000)
assert not credentials.valid
assert credentials.expired
assert credentials.token_state == TokenState.INVALID
credentials.before_request(request, "POST", "https://example.com/api", headers)
assert credentials.token_state == TokenState.FRESH
# New token should be retrieved.
assert headers == {
"authorization": "Bearer {}".format(SUCCESS_RESPONSE["access_token"])
}
|