@@ -76,13 +76,19 @@ def _merge_jwt_claims(defaults, overrides):
76
76
77
77
def verify_custom_token (custom_token , expected_claims , tenant_id = None ):
78
78
assert isinstance (custom_token , bytes )
79
- token = google .oauth2 .id_token .verify_token (
80
- custom_token ,
81
- testutils .MockRequest (200 , MOCK_PUBLIC_CERTS ),
82
- _token_gen .FIREBASE_AUDIENCE )
79
+ if _is_emulated ():
80
+ token = jwt .decode (custom_token , verify = False )
81
+ else :
82
+ token = google .oauth2 .id_token .verify_token (
83
+ custom_token ,
84
+ testutils .MockRequest (200 , MOCK_PUBLIC_CERTS ),
85
+ _token_gen .FIREBASE_AUDIENCE )
83
86
assert token ['uid' ] == MOCK_UID
84
- assert token ['iss' ] == MOCK_SERVICE_ACCOUNT_EMAIL
85
- assert token ['sub' ] == MOCK_SERVICE_ACCOUNT_EMAIL
87
+ expected_email = MOCK_SERVICE_ACCOUNT_EMAIL
88
+ if _is_emulated ():
89
+ expected_email = _token_gen .AUTH_EMULATOR_EMAIL
90
+ assert token ['iss' ] == expected_email
91
+ assert token ['sub' ] == expected_email
86
92
if tenant_id is None :
87
93
assert 'tenant_id' not in token
88
94
else :
@@ -141,7 +147,13 @@ def _overwrite_iam_request(app, request):
141
147
client = auth ._get_client (app )
142
148
client ._token_generator .request = request
143
149
144
- @pytest .fixture (scope = 'module' , params = [{'emulated' : False }, {'emulated' : True }])
150
+
151
+ def _is_emulated ():
152
+ emulator_host = os .getenv (EMULATOR_HOST_ENV_VAR , '' )
153
+ return emulator_host and '//' not in emulator_host
154
+
155
+
156
+ @pytest .fixture (scope = 'function' , params = [{'emulated' : False }, {'emulated' : True }])
145
157
def auth_app (request ):
146
158
"""Returns an App initialized with a mock service account credential.
147
159
@@ -217,6 +229,12 @@ class TestCreateCustomToken:
217
229
'MultipleReservedClaims' : (MOCK_UID , {'sub' :'1234' , 'aud' :'foo' }, ValueError ),
218
230
}
219
231
232
+ def _check_emulated_token (self , uid , app ):
233
+ # Should work fine with the emulator, so do a condensed version of
234
+ # test_sign_with_iam above.
235
+ custom_token = auth .create_custom_token (uid , app = app ).decode ()
236
+ self ._verify_signer (custom_token , _token_gen .AUTH_EMULATOR_EMAIL )
237
+
220
238
@pytest .mark .parametrize ('values' , valid_args .values (), ids = list (valid_args ))
221
239
def test_valid_params (self , auth_app , values ):
222
240
user , claims = values
@@ -230,20 +248,27 @@ def test_invalid_params(self, auth_app, values):
230
248
auth .create_custom_token (user , claims , app = auth_app )
231
249
232
250
def test_noncert_credential (self , user_mgt_app ):
251
+ if _is_emulated ():
252
+ self ._check_emulated_token (MOCK_UID , user_mgt_app )
253
+ return
233
254
with pytest .raises (ValueError ):
234
255
auth .create_custom_token (MOCK_UID , app = user_mgt_app )
235
256
236
257
def test_sign_with_iam (self ):
237
- options = {'serviceAccountId' : 'test-service-account' , 'projectId' : 'mock-project-id' }
258
+ signer = _token_gen .AUTH_EMULATOR_EMAIL if _is_emulated () else 'test-service-account'
259
+ options = {'serviceAccountId' : signer , 'projectId' : 'mock-project-id' }
238
260
app = firebase_admin .initialize_app (
239
261
testutils .MockCredential (), name = 'iam-signer-app' , options = options )
240
262
try :
241
263
signature = base64 .b64encode (b'test' ).decode ()
242
264
iam_resp = '{{"signature": "{0}"}}' .format (signature )
243
265
_overwrite_iam_request (app , testutils .MockRequest (200 , iam_resp ))
244
266
custom_token = auth .create_custom_token (MOCK_UID , app = app ).decode ()
245
- assert custom_token .endswith ('.' + signature .rstrip ('=' ))
246
- self ._verify_signer (custom_token , 'test-service-account' )
267
+ if _is_emulated ():
268
+ assert custom_token .endswith ('.' )
269
+ else :
270
+ assert custom_token .endswith ('.' + signature .rstrip ('=' ))
271
+ self ._verify_signer (custom_token , signer )
247
272
finally :
248
273
firebase_admin .delete_app (app )
249
274
@@ -254,7 +279,10 @@ def test_sign_with_iam_error(self):
254
279
try :
255
280
iam_resp = '{"error": {"code": 403, "message": "test error"}}'
256
281
_overwrite_iam_request (app , testutils .MockRequest (403 , iam_resp ))
257
- with pytest .raises (auth .TokenSignError ) as excinfo :
282
+ if _is_emulated ():
283
+ self ._check_emulated_token (MOCK_UID , app )
284
+ return
285
+ with pytest .raises ((ValueError , auth .TokenSignError )) as excinfo :
258
286
auth .create_custom_token (MOCK_UID , app = app )
259
287
error = excinfo .value
260
288
assert error .code == exceptions .UNKNOWN
@@ -264,7 +292,8 @@ def test_sign_with_iam_error(self):
264
292
firebase_admin .delete_app (app )
265
293
266
294
def test_sign_with_discovered_service_account (self ):
267
- request = testutils .MockRequest (200 , 'discovered-service-account' )
295
+ signer = _token_gen .AUTH_EMULATOR_EMAIL if _is_emulated () else 'discovered-service-account'
296
+ request = testutils .MockRequest (200 , signer )
268
297
options = {'projectId' : 'mock-project-id' }
269
298
app = firebase_admin .initialize_app (testutils .MockCredential (), name = 'iam-signer-app' ,
270
299
options = options )
@@ -279,10 +308,17 @@ def test_sign_with_discovered_service_account(self):
279
308
request .response = testutils .MockResponse (
280
309
200 , '{{"signature": "{0}"}}' .format (signature ))
281
310
custom_token = auth .create_custom_token (MOCK_UID , app = app ).decode ()
282
- assert custom_token .endswith ('.' + signature .rstrip ('=' ))
283
- self ._verify_signer (custom_token , 'discovered-service-account' )
284
- assert len (request .log ) == 2
285
- assert request .log [0 ][1 ]['headers' ] == {'Metadata-Flavor' : 'Google' }
311
+ if _is_emulated ():
312
+ # No signature from the emulator
313
+ assert custom_token .endswith ('.' )
314
+ else :
315
+ assert custom_token .endswith ('.' + signature .rstrip ('=' ))
316
+ if _is_emulated ():
317
+ # No requests will be made with the emulator
318
+ assert len (request .log ) == 0
319
+ else :
320
+ assert len (request .log ) == 2
321
+ assert request .log [0 ][1 ]['headers' ] == {'Metadata-Flavor' : 'Google' }
286
322
finally :
287
323
firebase_admin .delete_app (app )
288
324
@@ -293,6 +329,9 @@ def test_sign_with_discovery_failure(self):
293
329
options = options )
294
330
try :
295
331
_overwrite_iam_request (app , request )
332
+ if _is_emulated ():
333
+ self ._check_emulated_token (MOCK_UID , app )
334
+ return
296
335
with pytest .raises (ValueError ) as excinfo :
297
336
auth .create_custom_token (MOCK_UID , app = app )
298
337
assert str (excinfo .value ).startswith ('Failed to determine service account: test error' )
@@ -304,6 +343,14 @@ def test_sign_with_discovery_failure(self):
304
343
def _verify_signer (self , token , signer ):
305
344
segments = token .split ('.' )
306
345
assert len (segments ) == 3
346
+ # See https://github.com/googleapis/google-auth-library-python/pull/324
347
+ # and also RFC 7515 which is the basis of that PR:
348
+ # JWT segments don't have padding and base64 decoding might fail.
349
+ #
350
+ # Workaround from https://stackoverflow.com/a/9807138/2072269
351
+ missing_padding = len (segments [1 ]) % 4
352
+ if missing_padding :
353
+ segments [1 ] += '=' * (4 - missing_padding )
307
354
body = json .loads (base64 .b64decode (segments [1 ]).decode ())
308
355
assert body ['iss' ] == signer
309
356
assert body ['sub' ] == signer
@@ -406,6 +453,12 @@ class TestVerifyIdToken:
406
453
'BadFormatToken' : 'foobar'
407
454
}
408
455
456
+ tokens_not_invalid_in_emulator = [
457
+ 'WrongKid' ,
458
+ 'FutureToken' ,
459
+ 'ExpiredToken'
460
+ ]
461
+
409
462
@pytest .mark .parametrize ('id_token' , valid_tokens .values (), ids = list (valid_tokens ))
410
463
def test_valid_token (self , user_mgt_app , id_token ):
411
464
_overwrite_cert_request (user_mgt_app , MOCK_REQUEST )
@@ -458,15 +511,25 @@ def test_invalid_arg(self, user_mgt_app, id_token):
458
511
auth .verify_id_token (id_token , app = user_mgt_app )
459
512
assert 'Illegal ID token provided' in str (excinfo .value )
460
513
461
- @pytest .mark .parametrize ('id_token' , invalid_tokens .values (), ids = list (invalid_tokens ))
462
- def test_invalid_token (self , user_mgt_app , id_token ):
514
+ @pytest .mark .parametrize ('id_token_key' , list (invalid_tokens ))
515
+ def test_invalid_token (self , user_mgt_app , id_token_key ):
516
+ id_token = self .invalid_tokens [id_token_key ]
517
+ if _is_emulated () and id_token_key in self .tokens_not_invalid_in_emulator :
518
+ # These tokens won't be invalid when using the emulator
519
+ claims = auth .verify_id_token (id_token , app = user_mgt_app )
520
+ assert claims ['admin' ] is True
521
+ assert claims ['uid' ] == claims ['sub' ]
522
+ return
463
523
_overwrite_cert_request (user_mgt_app , MOCK_REQUEST )
464
524
with pytest .raises (auth .InvalidIdTokenError ) as excinfo :
465
525
auth .verify_id_token (id_token , app = user_mgt_app )
466
526
assert isinstance (excinfo .value , exceptions .InvalidArgumentError )
467
527
assert excinfo .value .http_response is None
468
528
469
529
def test_expired_token (self , user_mgt_app ):
530
+ if _is_emulated ():
531
+ pytest .skip ('Not supported in emulator mode: Token expiration' )
532
+ _overwrite_cert_request (user_mgt_app , MOCK_REQUEST )
470
533
_overwrite_cert_request (user_mgt_app , MOCK_REQUEST )
471
534
id_token = self .invalid_tokens ['ExpiredToken' ]
472
535
with pytest .raises (auth .ExpiredIdTokenError ) as excinfo :
@@ -506,6 +569,10 @@ def test_custom_token(self, auth_app):
506
569
507
570
def test_certificate_request_failure (self , user_mgt_app ):
508
571
_overwrite_cert_request (user_mgt_app , testutils .MockRequest (404 , 'not found' ))
572
+ if _is_emulated ():
573
+ # Shouldn't fetch certificates in emulator mode.
574
+ auth .verify_id_token (TEST_ID_TOKEN , app = user_mgt_app )
575
+ return
509
576
with pytest .raises (auth .CertificateFetchError ) as excinfo :
510
577
auth .verify_id_token (TEST_ID_TOKEN , app = user_mgt_app )
511
578
assert 'Could not fetch certificates' in str (excinfo .value )
@@ -540,6 +607,12 @@ class TestVerifySessionCookie:
540
607
'IDToken' : TEST_ID_TOKEN ,
541
608
}
542
609
610
+ cookies_not_invalid_in_emulator = [
611
+ 'WrongKid' ,
612
+ 'FutureCookie' ,
613
+ 'ExpiredCookie'
614
+ ]
615
+
543
616
@pytest .mark .parametrize ('cookie' , valid_cookies .values (), ids = list (valid_cookies ))
544
617
def test_valid_cookie (self , user_mgt_app , cookie ):
545
618
_overwrite_cert_request (user_mgt_app , MOCK_REQUEST )
@@ -578,15 +651,24 @@ def test_invalid_args(self, user_mgt_app, cookie):
578
651
auth .verify_session_cookie (cookie , app = user_mgt_app )
579
652
assert 'Illegal session cookie provided' in str (excinfo .value )
580
653
581
- @pytest .mark .parametrize ('cookie' , invalid_cookies .values (), ids = list (invalid_cookies ))
582
- def test_invalid_cookie (self , user_mgt_app , cookie ):
654
+ @pytest .mark .parametrize ('cookie_key' , list (invalid_cookies ))
655
+ def test_invalid_cookie (self , user_mgt_app , cookie_key ):
656
+ cookie = self .invalid_cookies [cookie_key ]
657
+ if _is_emulated () and cookie_key in self .cookies_not_invalid_in_emulator :
658
+ # These cookies won't be invalid when using the emulator
659
+ claims = auth .verify_session_cookie (cookie , app = user_mgt_app )
660
+ assert claims ['admin' ] is True
661
+ assert claims ['uid' ] == claims ['sub' ]
662
+ return
583
663
_overwrite_cert_request (user_mgt_app , MOCK_REQUEST )
584
664
with pytest .raises (auth .InvalidSessionCookieError ) as excinfo :
585
665
auth .verify_session_cookie (cookie , app = user_mgt_app )
586
666
assert isinstance (excinfo .value , exceptions .InvalidArgumentError )
587
667
assert excinfo .value .http_response is None
588
668
589
669
def test_expired_cookie (self , user_mgt_app ):
670
+ if _is_emulated ():
671
+ pytest .skip ('Not supported in emulator mode: Cookie expiration' )
590
672
_overwrite_cert_request (user_mgt_app , MOCK_REQUEST )
591
673
cookie = self .invalid_cookies ['ExpiredCookie' ]
592
674
with pytest .raises (auth .ExpiredSessionCookieError ) as excinfo :
@@ -621,6 +703,10 @@ def test_custom_token(self, auth_app):
621
703
622
704
def test_certificate_request_failure (self , user_mgt_app ):
623
705
_overwrite_cert_request (user_mgt_app , testutils .MockRequest (404 , 'not found' ))
706
+ if _is_emulated ():
707
+ # Shouldn't fetch certificates in emulator mode.
708
+ auth .verify_session_cookie (TEST_SESSION_COOKIE , app = user_mgt_app )
709
+ return
624
710
with pytest .raises (auth .CertificateFetchError ) as excinfo :
625
711
auth .verify_session_cookie (TEST_SESSION_COOKIE , app = user_mgt_app )
626
712
assert 'Could not fetch certificates' in str (excinfo .value )
@@ -637,9 +723,11 @@ def test_certificate_caching(self, user_mgt_app, httpserver):
637
723
verifier .cookie_verifier .cert_url = httpserver .url
638
724
verifier .id_token_verifier .cert_url = httpserver .url
639
725
verifier .verify_session_cookie (TEST_SESSION_COOKIE )
640
- assert len (httpserver .requests ) == 1
726
+ # No requests should be made in emulated mode
727
+ request_count = 0 if _is_emulated () else 1
728
+ assert len (httpserver .requests ) == request_count
641
729
# Subsequent requests should not fetch certs from the server
642
730
verifier .verify_session_cookie (TEST_SESSION_COOKIE )
643
- assert len (httpserver .requests ) == 1
731
+ assert len (httpserver .requests ) == request_count
644
732
verifier .verify_id_token (TEST_ID_TOKEN )
645
- assert len (httpserver .requests ) == 1
733
+ assert len (httpserver .requests ) == request_count
0 commit comments